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")) +}