Upgrade to go 1.25

Author: bogcon

.github/workflows/build.yml

@@ -13,7 +13,7 @@ jobs:
     timeout-minutes: 10
     strategy:
       matrix:
-        go-version: [1.23.x, 1.24.x]
+        go-version: [1.25.x, 1.26.x]
         platform: [ubuntu-latest, macos-latest, windows-latest]
     runs-on: ${{ matrix.platform }}
 
@@ -24,10 +24,10 @@ jobs:
         go-version: ${{ matrix.go-version }}
 
     - name: Checkout code
-      uses: actions/checkout@v5
+      uses: actions/checkout@v6
 
     - name: Load cached dependencies
-      uses: actions/cache@v4
+      uses: actions/cache@v5
       with:
         path: ~/go/pkg/mod
         key: ${{ runner.os }}-go-${{ matrix.go-version }}-${{ hashFiles('**/go.sum') }}
@@ -44,9 +44,9 @@ jobs:
       run: make clean cover
 
     - name: Upload coverage to coveralls.io
-      if: matrix.platform == 'ubuntu-latest' && matrix.go-version == '1.24.x'
+      if: matrix.platform == 'ubuntu-latest' && matrix.go-version == '1.26.x'
       uses: coverallsapp/github-action@v2
       with:
-        file: cover.out
+        file: bin/cover.out
         flag-name: ${{ runner.os }}-go-${{ matrix.go-version }}
         fail-on-error: false

.gitignore

@@ -7,7 +7,3 @@ bin/*
 *.exe
 *.exe~
 *.so
-
-cover.out
-cover.html
-*.prof

Makefile

@@ -1,4 +1,4 @@
-LINTER_VERSION=v2.1.6
+LINTER_VERSION=v2.9.0
 LINTER=./bin/golangci-lint
 ifeq ($(OS),Windows_NT)
 	LINTER=./bin/golangci-lint.exe
@@ -19,6 +19,7 @@ lint: ## Run linter and detect go mod tidy changes.
 
 .PHONY: setup
 setup: ## Download dependencies.
+	@mkdir -p bin
 	go mod download
 	@if [ ! -f "$(LINTER)" ]; then \
 		curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s $(LINTER_VERSION); \
@@ -34,8 +35,8 @@ bench: ## Run benchmarks.
 
 .PHONY: cover
 cover: ## Run tests with coverage. Generates "cover.out" profile and its html representation.
-	go test -race -timeout=10m -coverprofile=cover.out -coverpkg=./... $(pkgs)
-	go tool cover -html=cover.out -o cover.html
+	go test -race -timeout=10m -coverprofile=./bin/cover.out -coverpkg=./... $(pkgs)
+	go tool cover -html=./bin/cover.out -o ./bin/cover.html
 
 .PHONY: tidy
 tidy: ## Simply runs 'go mod tidy'.
@@ -44,12 +45,8 @@ tidy: ## Simply runs 'go mod tidy'.
 .PHONY: clean
 clean: ## Clean up go tests cache and coverage generated files.
 	go clean -testcache
-	@for file in cover.html cover.out; do \
-		if [ -f $$file ]; then \
-			echo "rm -f $$file"; \
-			rm -f $$file; \
-		fi \
-	done
+	@rm -f ./bin/cover*
+	@rm -f ./bin/*.prof
 
 .PHONY: help
 # Absolutely awesome: https://marmelab.com/blog/2016/02/29/auto-documented-makefile.html

cmd/pprof/main.go

@@ -38,7 +38,7 @@ func main() {
 	defer tearDownTmpCsvFile(fName)
 
 	// enable cpu profiling
-	fCPU, err := os.Create("./cpu_" + *generateProfileFor + ".prof")
+	fCPU, err := os.Create("./bin/cpu_" + *generateProfileFor + ".prof")
 	if err != nil {
 		log.Println("could not create CPU profile: ", err)
 
@@ -62,7 +62,7 @@ func main() {
 	}
 
 	// enable memory profiling
-	fMem, err := os.Create("./mem_" + *generateProfileFor + ".prof")
+	fMem, err := os.Create("./bin/mem_" + *generateProfileFor + ".prof")
 	if err != nil {
 		log.Println("could not create memory profile: ", err)
 
@@ -206,25 +206,22 @@ func consumeBigCsvReaderResults(rowsChans []bigcsvreader.RowsChan, errsChan bigc
 	)
 
 	for i := range rowsChans {
-		wg.Add(1)
-		go func(rowsChan bigcsvreader.RowsChan, waitGr *sync.WaitGroup) {
+		wg.Go(func() {
+			rowsChan := rowsChans[i]
 			var localCount int64
 			for record := range rowsChan {
 				localCount++
 				_ = record
 			}
 			atomic.AddInt64(&count, localCount)
-			waitGr.Done()
-		}(rowsChans[i], &wg)
+		})
 	}
 
-	wg.Add(1)
-	go func(errsCh bigcsvreader.ErrsChan, waitGr *sync.WaitGroup) {
-		for err := range errsCh {
+	wg.Go(func() {
+		for err := range errsChan {
 			log.Println("Read error: ", err)
 		}
-		waitGr.Done()
-	}(errsChan, &wg)
+	})
 
 	wg.Wait()
 

example_test.go

@@ -47,15 +47,14 @@ func ExampleCsvReader() {
 	rowsChans, errsChan := bigCSV.Read(ctx)
 
 	// process rows and errors:
-
 	for i := range rowsChans {
-		wg.Add(1)
-		go rowWorker(rowsChans[i], &wg)
+		wg.Go(func() {
+			rowWorker(rowsChans[i])
+		})
 	}
-
-	wg.Add(1)
-	go errWorker(errsChan, &wg)
-
+	wg.Go(func() {
+		errWorker(errsChan)
+	})
 	wg.Wait()
 
 	// Unordered output:
@@ -66,18 +65,16 @@ func ExampleCsvReader() {
 	// {ID:5 Name:Logitech Mouse G203 Desc:Lorem ipsum dolor sit amet, consectetur adipiscing elit. Nunc eleifend felis quis magna auctor, ut lacinia eros efficitur. Maecenas mattis dolor a pharetra gravida. Aenean at eros sed metus posuere feugiat in vitae libero. Morbi a diam volutpat, tempor lacus sed, sagittis velit. Donec eget dignissim mauris, sed aliquam ex. Duis eros dolor, vestibulum ac aliquam eget, viverra in enim. Aenean ut turpis quis purus porta lobortis. Etiam sollicitudin lectus vitae velit tincidunt, ut volutpat justo aliquam. Aenean vitae vehicula arcu. Interdum et malesuada fames ac ante ipsum primis in faucibus. Nunc viverra enim nec risus mollis elementum nec dictum ex. Nunc lorem eros, vulputate a rutrum nec, scelerisque non augue. Sed in egestas eros. Quisque felis lorem, vehicula ac venenatis vel, tristique id sapien. Morbi vitae odio eget orci facilisis suscipit. Cras sodales, augue vitae tincidunt tempus, diam turpis volutpat est, vitae fringilla augue leo semper augue. Integer scelerisque tempor mauris, ac posuere sem aenean Price:30.5 Qty:35}
 }
 
-func rowWorker(rowsChan bigcsvreader.RowsChan, waitGr *sync.WaitGroup) {
+func rowWorker(rowsChan bigcsvreader.RowsChan) {
 	for row := range rowsChan {
 		processRow(row)
 	}
-	waitGr.Done()
 }
 
-func errWorker(errsChan bigcsvreader.ErrsChan, waitGr *sync.WaitGroup) {
+func errWorker(errsChan bigcsvreader.ErrsChan) {
 	for err := range errsChan {
 		handleError(err)
 	}
-	waitGr.Done()
 }
 
 // processRow can be used to implement business logic

go.mod

@@ -1,3 +1,3 @@
 module github.com/actforgood/bigcsvreader
 
-go 1.23
+go 1.25

reader.go

@@ -153,18 +153,18 @@ func (cr *CsvReader) readAsync(
 
 	// create a wait group pool as we need to wait for all goroutines to terminate.
 	var wg sync.WaitGroup
-	wg.Add(totalThreads)
 	worker := cr.readBetweenOffsetsAsync
 	for thread := range totalThreads {
-		go worker(
-			ctx,
-			thread+1,
-			threadsInfo[thread][0], // start offset
-			threadsInfo[thread][1], // end offset
-			&wg,
-			rowsChans[thread],
-			errsChan,
-		)
+		wg.Go(func() {
+			worker(
+				ctx,
+				thread+1,
+				threadsInfo[thread][0], // start offset
+				threadsInfo[thread][1], // end offset
+				rowsChans[thread],
+				errsChan,
+			)
+		})
 	}
 	wg.Wait()
 
@@ -175,12 +175,9 @@ func (cr *CsvReader) readAsync(
 func (cr *CsvReader) readBetweenOffsetsAsync(
 	ctx context.Context,
 	currentThreadNo, offsetStart, offsetEnd int,
-	wg *sync.WaitGroup,
 	rowsChan chan<- []string,
 	errsChan chan<- error,
 ) {
-	defer wg.Done()
-
 	f := cr.openFile(currentThreadNo, errsChan)
 	if f == nil {
 		return

reader_test.go

@@ -18,6 +18,7 @@ import (
 	"sync"
 	"sync/atomic"
 	"testing"
+	"testing/synctest"
 	"time"
 
 	"github.com/actforgood/bigcsvreader"
@@ -129,54 +130,53 @@ func testCsvReaderWithDifferentFileSizesAndMaxGoroutines(rowsCount int64) func(t
 	return func(t *testing.T) {
 		t.Parallel()
 
-		// arrange
-		fName, err := setUpTmpCsvFile(t.TempDir(), rowsCount)
-		if err != nil {
-			t.Fatalf("prerequisite failed: could not generate CSV file: %v", err)
-		}
-		subject := bigcsvreader.New()
-		subject.SetFilePath(fName)
-		subject.ColumnsCount = 5
-		ctx, cancelCtx := context.WithCancel(context.Background())
-		defer cancelCtx()
-		var sumIDs int64
-		var wg sync.WaitGroup
-
-		for maxGoroutines := 1; maxGoroutines <= 16; maxGoroutines++ {
-			subject.MaxGoroutinesNo = maxGoroutines
-			sumIDs = 0
-			expectedSumIDs := rowsCount * (rowsCount + 1) / 2
-
-			// act
-			rowsChans, errsChan := subject.Read(ctx)
-
-			// assert
-			for i := range rowsChans {
-				wg.Add(1)
-				go func(rowsChan bigcsvreader.RowsChan, waitGr *sync.WaitGroup) {
-					var localSumIDs int64
-					for record := range rowsChan {
-						if !assertEqual(t, 5, len(record)) {
-							continue
-						}
-						id, _ := strconv.ParseInt(record[colID], 10, 64)
-						localSumIDs += id
-						expectedColName := colValueNamePrefix + record[colID]
-						assertEqual(t, expectedColName, record[colName])
-						assertEqual(t, colValueDescription, record[colDescription])
-						assertEqual(t, colValuePrice, record[colPrice])
-						assertEqual(t, colValueStock, record[colStock])
-					}
-					atomic.AddInt64(&sumIDs, localSumIDs)
-					waitGr.Done()
-				}(rowsChans[i], &wg)
+		synctest.Test(t, func(*testing.T) {
+			// arrange
+			fName, err := setUpTmpCsvFile(t.TempDir(), rowsCount)
+			if err != nil {
+				t.Fatalf("prerequisite failed: could not generate CSV file: %v", err)
 			}
-			for err := range errsChan {
-				assertNil(t, err)
+			subject := bigcsvreader.New()
+			subject.SetFilePath(fName)
+			subject.ColumnsCount = 5
+			ctx, cancelCtx := context.WithCancel(context.Background())
+			defer cancelCtx()
+			var sumIDs int64
+
+			for maxGoroutines := range 16 {
+				subject.MaxGoroutinesNo = maxGoroutines + 1
+				sumIDs = 0
+				expectedSumIDs := rowsCount * (rowsCount + 1) / 2
+
+				// act
+				rowsChans, errsChan := subject.Read(ctx)
+
+				// assert
+				for i := range rowsChans {
+					go func(rowsChan bigcsvreader.RowsChan) {
+						var localSumIDs int64
+						for record := range rowsChan {
+							if !assertEqual(t, 5, len(record)) {
+								continue
+							}
+							id, _ := strconv.ParseInt(record[colID], 10, 64)
+							localSumIDs += id
+							expectedColName := colValueNamePrefix + record[colID]
+							assertEqual(t, expectedColName, record[colName])
+							assertEqual(t, colValueDescription, record[colDescription])
+							assertEqual(t, colValuePrice, record[colPrice])
+							assertEqual(t, colValueStock, record[colStock])
+						}
+						atomic.AddInt64(&sumIDs, localSumIDs)
+					}(rowsChans[i])
+				}
+				for err := range errsChan {
+					assertNil(t, err)
+				}
+				synctest.Wait()
+				assertEqual(t, expectedSumIDs, sumIDs)
 			}
-			wg.Wait()
-			assertEqual(t, expectedSumIDs, sumIDs)
-		}
+		})
 	}
 }
 
@@ -297,15 +297,14 @@ func gatherRecords(rowsChans []bigcsvreader.RowsChan, errsChan bigcsvreader.Errs
 		records = make([][]string, 0)
 	)
 	for i := range rowsChans {
-		wg.Add(1)
-		go func(rowsChan bigcsvreader.RowsChan, mutex *sync.Mutex, waitGr *sync.WaitGroup) {
+		wg.Go(func() {
+			rowsChan := rowsChans[i]
 			for record := range rowsChan {
-				mutex.Lock()
+				mu.Lock()
 				records = append(records, record)
 				mu.Unlock()
 			}
-			waitGr.Done()
-		}(rowsChans[i], &mu, &wg)
+		})
 	}
 
 	for err := range errsChan {
@@ -416,16 +415,15 @@ func consumeBenchResults(rowsChans []bigcsvreader.RowsChan, _ bigcsvreader.ErrsC
 	)
 
 	for i := range rowsChans {
-		wg.Add(1)
-		go func(rowsChan bigcsvreader.RowsChan, waitGr *sync.WaitGroup) {
+		wg.Go(func() {
+			rowsChan := rowsChans[i]
 			var localCount int64
 			for record := range rowsChan {
 				localCount++
 				fakeProcessRow(record)
 			}
 			atomic.AddInt64(&count, localCount)
-			waitGr.Done()
-		}(rowsChans[i], &wg)
+		})
 	}
 	wg.Wait()
 
@@ -531,16 +529,14 @@ func benchmarkStdGoCsvReaderReadOneByOneProcessParalell(rowsCount int64) func(b
 				wg    sync.WaitGroup
 			)
 			for range numWorkers {
-				wg.Add(1)
-				go func() {
+				wg.Go(func() {
 					var localCount int64
 					for record := range rowsChan {
 						localCount++
 						fakeProcessRow(record)
 					}
 					atomic.AddInt64(&count, localCount)
-					wg.Done()
-				}()
+				})
 			}
 
 			// sequential reading