From 89863fe8e1904a1035d48c2fab5432c2d9dbe311 Mon Sep 17 00:00:00 2001 From: Terry Rankine Date: Thu, 17 Sep 2026 20:04:53 +0200 Subject: [PATCH] fix: data races in progress bar counters and stat maps - progressbar: IncrementCompletedObjects and IncrementTotalObjects used atomic.AddInt64 to write the counters but read them back with plain loads in fmt.Sprintf. Concurrent workers raced on the read. Use the value returned by AddInt64 and atomic.LoadInt64 for the other counter. - log/stat: Statistics() iterated the internal maps without holding their mutexes, racing with Collect() from worker goroutines and risking a "concurrent map iteration and map write" fatal error. Hold both locks for the duration of the read. - log: Close() closed the shared output channel unconditionally, so a second call panicked with "close of closed channel". Guard it with a sync.Once on the Logger so it is safe to call multiple times. Adds a test for each: `go test -race` fails all three without the fix and passes with it. Co-Authored-By: Claude Opus 5 (1M context) --- log/log.go | 16 +++-- log/log_test.go | 11 ++++ log/stat/stat.go | 5 ++ log/stat/stat_test.go | 104 ++++++++++++++++++++++++++++++++ progressbar/progressbar.go | 10 +-- progressbar/progressbar_test.go | 33 ++++++++++ 6 files changed, 169 insertions(+), 10 deletions(-) create mode 100644 log/log_test.go create mode 100644 log/stat/stat_test.go diff --git a/log/log.go b/log/log.go index 5ec149ac9..7e12c4d0e 100644 --- a/log/log.go +++ b/log/log.go @@ -3,6 +3,7 @@ package log import ( "fmt" "os" + "sync" ) // output is an internal container for messages to be logged. @@ -48,19 +49,22 @@ func Error(msg Message) { global.printf(LevelError, msg, os.Stderr) } -// Close closes logger and its channel. +// Close closes logger and its channel. It is safe to call multiple times. func Close() { if global != nil { - close(outputCh) - <-global.donech + global.closeOnce.Do(func() { + close(outputCh) + <-global.donech + }) } } // Logger is a structure for logging messages. type Logger struct { - donech chan struct{} - json bool - level LogLevel + donech chan struct{} + closeOnce sync.Once + json bool + level LogLevel } // New creates new logger. diff --git a/log/log_test.go b/log/log_test.go new file mode 100644 index 000000000..029d517e9 --- /dev/null +++ b/log/log_test.go @@ -0,0 +1,11 @@ +package log + +import "testing" + +func TestCloseMultipleTimes(t *testing.T) { + Init("info", false) + + Close() + // Second Close must not panic with "close of closed channel". + Close() +} diff --git a/log/stat/stat.go b/log/stat/stat.go index 5e0ed4922..ea4039c2f 100644 --- a/log/stat/stat.go +++ b/log/stat/stat.go @@ -98,6 +98,11 @@ func Statistics() Stats { return Stats{} } + stats[totalCount].Lock() + defer stats[totalCount].Unlock() + stats[succCount].Lock() + defer stats[succCount].Unlock() + var result Stats for op, total := range stats[totalCount].mapStrInt64 { success := stats[succCount].mapStrInt64[op] diff --git a/log/stat/stat_test.go b/log/stat/stat_test.go new file mode 100644 index 000000000..040ac1493 --- /dev/null +++ b/log/stat/stat_test.go @@ -0,0 +1,104 @@ +package stat + +import ( + "errors" + "sync" + "testing" +) + +func resetStats() { + enabled = false + stats = statistics{} +} + +func TestStatisticsCollect(t *testing.T) { + resetStats() + defer resetStats() + + InitStat() + + var nilErr error + Collect("cp", &nilErr)() + Collect("cp", &nilErr)() + Collect("cp", nil)() // nil *error also counts as success + + anErr := errors.New("some error") + Collect("cp", &anErr)() + + Collect("mv", &nilErr)() + + statMap := map[string]Stat{} + for _, s := range Statistics() { + statMap[s.Operation] = s + } + + cp, ok := statMap["cp"] + if !ok { + t.Fatal("expected 'cp' in statistics") + } + if cp.Success != 3 || cp.Error != 1 { + t.Errorf("cp: expected success=3 error=1, got success=%d error=%d", cp.Success, cp.Error) + } + + mv, ok := statMap["mv"] + if !ok { + t.Fatal("expected 'mv' in statistics") + } + if mv.Success != 1 || mv.Error != 0 { + t.Errorf("mv: expected success=1 error=0, got success=%d error=%d", mv.Success, mv.Error) + } +} + +func TestStatisticsDisabled(t *testing.T) { + resetStats() + + if got := Statistics(); len(got) != 0 { + t.Errorf("expected empty stats when disabled, got %d entries", len(got)) + } +} + +func TestStatisticsConcurrentCollect(t *testing.T) { + resetStats() + defer resetStats() + + InitStat() + + const ( + workers = 8 + iterations = 500 + ) + + var wg sync.WaitGroup + for i := 0; i < workers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + var nilErr error + anErr := errors.New("some error") + for j := 0; j < iterations; j++ { + Collect("cp", &nilErr)() + Collect("cp", &anErr)() + } + }() + } + + // Read statistics while workers are still collecting. + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < iterations; i++ { + _ = Statistics() + } + }() + + wg.Wait() + + result := Statistics() + if len(result) != 1 { + t.Fatalf("expected 1 operation, got %d", len(result)) + } + want := int64(workers * iterations) + if result[0].Success != want || result[0].Error != want { + t.Errorf("expected success=%d error=%d, got success=%d error=%d", want, want, result[0].Success, result[0].Error) + } +} diff --git a/progressbar/progressbar.go b/progressbar/progressbar.go index d8752d47d..cd651cefc 100644 --- a/progressbar/progressbar.go +++ b/progressbar/progressbar.go @@ -60,13 +60,15 @@ func (cp *CommandProgressBar) Finish() { } func (cp *CommandProgressBar) IncrementCompletedObjects() { - atomic.AddInt64(&cp.completedObjects, 1) - cp.progressbar.Set("objects", fmt.Sprintf("(%d/%d)", cp.completedObjects, cp.totalObjects)) + completed := atomic.AddInt64(&cp.completedObjects, 1) + total := atomic.LoadInt64(&cp.totalObjects) + cp.progressbar.Set("objects", fmt.Sprintf("(%d/%d)", completed, total)) } func (cp *CommandProgressBar) IncrementTotalObjects() { - atomic.AddInt64(&cp.totalObjects, 1) - cp.progressbar.Set("objects", fmt.Sprintf("(%d/%d)", cp.completedObjects, cp.totalObjects)) + total := atomic.AddInt64(&cp.totalObjects, 1) + completed := atomic.LoadInt64(&cp.completedObjects) + cp.progressbar.Set("objects", fmt.Sprintf("(%d/%d)", completed, total)) } func (cp *CommandProgressBar) AddCompletedBytes(bytes int64) { diff --git a/progressbar/progressbar_test.go b/progressbar/progressbar_test.go index 8c8a12f9f..57e9bcac1 100644 --- a/progressbar/progressbar_test.go +++ b/progressbar/progressbar_test.go @@ -2,6 +2,7 @@ package progressbar import ( "strings" + "sync" "testing" "gotest.tools/v3/assert" @@ -51,3 +52,35 @@ func TestCommandProgress_AddTotalBytes(t *testing.T) { assert.Equal(t, bytes, cp.progressbar.Total()) assert.Equal(t, true, strings.Contains(cp.progressbar.String(), "102 B")) } + +func TestCommandProgress_ConcurrentIncrement(t *testing.T) { + t.Parallel() + cp := New() + + const ( + workers = 8 + iterations = 500 + ) + + var wg sync.WaitGroup + for i := 0; i < workers; i++ { + wg.Add(2) + go func() { + defer wg.Done() + for j := 0; j < iterations; j++ { + cp.IncrementTotalObjects() + } + }() + go func() { + defer wg.Done() + for j := 0; j < iterations; j++ { + cp.IncrementCompletedObjects() + } + }() + } + wg.Wait() + + assert.Equal(t, int64(workers*iterations), cp.totalObjects) + assert.Equal(t, int64(workers*iterations), cp.completedObjects) + assert.Equal(t, true, strings.Contains(cp.progressbar.String(), "4000/4000")) +}