fix: data races in progress bar counters and stat maps

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.

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

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]

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

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) {

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