diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2a94727..8718f8e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -2,8 +2,8 @@ name: build on: push: - branches: - tags: + branches: ["**"] + tags: ["**"] pull_request: jobs: @@ -12,10 +12,10 @@ jobs: steps: - name: checkout - uses: actions/checkout@v4 + uses: actions/checkout@v7 - name: set up go - uses: actions/setup-go@v5 + uses: actions/setup-go@v7 with: go-version: "1.24" id: go @@ -29,9 +29,9 @@ jobs: go build -race - name: golangci-lint - uses: golangci/golangci-lint-action@v7 + uses: golangci/golangci-lint-action@v9 with: - version: v2.7 + version: v2.12.2 - name: install goveralls run: | diff --git a/bench_test.go b/bench_test.go index 3100678..ee66f90 100644 --- a/bench_test.go +++ b/bench_test.go @@ -36,7 +36,7 @@ func TestPoolPerf(t *testing.T) { var egDuration time.Duration t.Run("errgroup", func(t *testing.T) { - var count2 int32 + var count2 atomic.Int32 st := time.Now() defer func() { egDuration = time.Since(st) @@ -47,20 +47,20 @@ func TestPoolPerf(t *testing.T) { for range 1000000 { g.Go(func() error { benchTask(n) - atomic.AddInt32(&count2, 1) + count2.Add(1) return nil }) } require.NoError(t, g.Wait()) - assert.Equal(t, int32(1000000), atomic.LoadInt32(&count2)) + assert.Equal(t, int32(1000000), count2.Load()) }) t.Run("pool default", func(t *testing.T) { // pool with 8 workers - var count1 int32 + var count1 atomic.Int32 worker := WorkerFunc[int](func(context.Context, int) error { benchTask(n) - atomic.AddInt32(&count1, 1) + count1.Add(1) return nil }) @@ -74,17 +74,17 @@ func TestPoolPerf(t *testing.T) { assert.NoError(t, p.Close(ctx)) }() require.NoError(t, p.Wait(ctx)) - assert.Equal(t, int32(1000000), atomic.LoadInt32(&count1)) + assert.Equal(t, int32(1000000), count1.Load()) t.Logf("elapsed pool: %v", time.Since(st)) assert.Less(t, time.Since(st), egDuration) }) t.Run("pool with 100 chan size", func(t *testing.T) { // pool with 8 workers - var count1 int32 + var count1 atomic.Int32 worker := WorkerFunc[int](func(context.Context, int) error { benchTask(n) - atomic.AddInt32(&count1, 1) + count1.Add(1) return nil }) @@ -98,17 +98,17 @@ func TestPoolPerf(t *testing.T) { assert.NoError(t, p.Close(ctx)) }() require.NoError(t, p.Wait(ctx)) - assert.Equal(t, int32(1000000), atomic.LoadInt32(&count1)) + assert.Equal(t, int32(1000000), count1.Load()) t.Logf("elapsed pool: %v", time.Since(st)) assert.Less(t, time.Since(st), egDuration) }) t.Run("pool with 100 chan size and 100 batch size", func(t *testing.T) { // pool with 8 workers - var count1 int32 + var count1 atomic.Int32 worker := WorkerFunc[int](func(context.Context, int) error { benchTask(n) - atomic.AddInt32(&count1, 1) + count1.Add(1) return nil }) @@ -122,17 +122,17 @@ func TestPoolPerf(t *testing.T) { assert.NoError(t, p.Close(ctx)) }() require.NoError(t, p.Wait(ctx)) - assert.Equal(t, int32(1000000), atomic.LoadInt32(&count1)) + assert.Equal(t, int32(1000000), count1.Load()) t.Logf("elapsed pool: %v", time.Since(st)) assert.Less(t, time.Since(st), egDuration) }) t.Run("pool with 100 chan size and 100 batch size and chunking", func(t *testing.T) { // pool with 8 workers - var count1 int32 + var count1 atomic.Int32 worker := WorkerFunc[int](func(context.Context, int) error { benchTask(n) - atomic.AddInt32(&count1, 1) + count1.Add(1) return nil }) @@ -148,7 +148,7 @@ func TestPoolPerf(t *testing.T) { assert.NoError(t, p.Close(ctx)) }() require.NoError(t, p.Wait(ctx)) - assert.Equal(t, int32(1000000), atomic.LoadInt32(&count1)) + assert.Equal(t, int32(1000000), count1.Load()) t.Logf("elapsed pool: %v", time.Since(st)) assert.Less(t, time.Since(st), egDuration) }) @@ -164,29 +164,29 @@ func BenchmarkPoolCompare(b *testing.B) { b.Run("errgroup", func(b *testing.B) { b.ResetTimer() for range b.N { - var count int32 + var count atomic.Int32 g, _ := errgroup.WithContext(ctx) g.SetLimit(workers) for range iterations { g.Go(func() error { benchTask(n) - atomic.AddInt32(&count, 1) + count.Add(1) return nil }) } require.NoError(b, g.Wait()) - require.Equal(b, int32(iterations), atomic.LoadInt32(&count)) + require.Equal(b, int32(iterations), count.Load()) } }) b.Run("pool default", func(b *testing.B) { b.ResetTimer() for range b.N { - var count int32 + var count atomic.Int32 p := New[int](workers, WorkerFunc[int](func(context.Context, int) error { benchTask(n) - atomic.AddInt32(&count, 1) + count.Add(1) return nil })) @@ -198,17 +198,17 @@ func BenchmarkPoolCompare(b *testing.B) { p.Close(ctx) }() require.NoError(b, p.Wait(ctx)) - require.Equal(b, int32(iterations), atomic.LoadInt32(&count)) + require.Equal(b, int32(iterations), count.Load()) } }) b.Run("pool with chan=100", func(b *testing.B) { b.ResetTimer() for range b.N { - var count int32 + var count atomic.Int32 p := New[int](workers, WorkerFunc[int](func(context.Context, int) error { benchTask(n) - atomic.AddInt32(&count, 1) + count.Add(1) return nil })).WithWorkerChanSize(100) @@ -220,17 +220,17 @@ func BenchmarkPoolCompare(b *testing.B) { p.Close(ctx) }() require.NoError(b, p.Wait(ctx)) - require.Equal(b, int32(iterations), atomic.LoadInt32(&count)) + require.Equal(b, int32(iterations), count.Load()) } }) b.Run("pool with batching", func(b *testing.B) { b.ResetTimer() for range b.N { - var count int32 + var count atomic.Int32 p := New[int](workers, WorkerFunc[int](func(context.Context, int) error { benchTask(n) - atomic.AddInt32(&count, 1) + count.Add(1) return nil })).WithWorkerChanSize(100).WithBatchSize(100) @@ -242,17 +242,17 @@ func BenchmarkPoolCompare(b *testing.B) { p.Close(ctx) }() require.NoError(b, p.Wait(ctx)) - require.Equal(b, int32(iterations), atomic.LoadInt32(&count)) + require.Equal(b, int32(iterations), count.Load()) } }) b.Run("pool with batching and chunking", func(b *testing.B) { b.ResetTimer() for range b.N { - var count int32 + var count atomic.Int32 p := New[int](workers, WorkerFunc[int](func(context.Context, int) error { benchTask(n) - atomic.AddInt32(&count, 1) + count.Add(1) return nil })).WithWorkerChanSize(100).WithBatchSize(100).WithChunkFn(func(v int) string { return strconv.Itoa(v % workers) @@ -266,7 +266,7 @@ func BenchmarkPoolCompare(b *testing.B) { p.Close(ctx) }() require.NoError(b, p.Wait(ctx)) - require.Equal(b, int32(iterations), atomic.LoadInt32(&count)) + require.Equal(b, int32(iterations), count.Load()) } }) } diff --git a/pool.go b/pool.go index 0e9ded7..8aea9c1 100644 --- a/pool.go +++ b/pool.go @@ -6,6 +6,7 @@ import ( "fmt" "hash/fnv" "math/rand" + "slices" "sync" "sync/atomic" "time" @@ -566,9 +567,9 @@ func (p *WorkerGroup[T]) Use(middlewares ...Middleware[T]) *WorkerGroup[T] { // apply middlewares in order from last to first // this makes first middleware outermost wrapped := worker - for i := len(middlewares) - 1; i >= 0; i-- { + for _, v := range slices.Backward(middlewares) { prev := wrapped - wrapped = middlewares[i](prev) + wrapped = v(prev) } return wrapped } @@ -577,9 +578,9 @@ func (p *WorkerGroup[T]) Use(middlewares ...Middleware[T]) *WorkerGroup[T] { // for stateless worker, just wrap it directly wrapped := p.worker - for i := len(middlewares) - 1; i >= 0; i-- { + for _, v := range slices.Backward(middlewares) { prev := wrapped - wrapped = middlewares[i](prev) + wrapped = v(prev) } p.worker = wrapped return p diff --git a/pool_test.go b/pool_test.go index 1d8ca62..994b152 100644 --- a/pool_test.go +++ b/pool_test.go @@ -333,10 +333,10 @@ func TestPool_Distribution(t *testing.T) { func TestPool_Metrics(t *testing.T) { t.Run("basic metrics", func(t *testing.T) { - var processed int32 + var processed atomic.Int32 worker := WorkerFunc[int](func(ctx context.Context, _ int) error { time.Sleep(time.Millisecond) // simulate work - atomic.AddInt32(&processed, 1) + processed.Add(1) return nil }) @@ -349,7 +349,7 @@ func TestPool_Metrics(t *testing.T) { require.NoError(t, p.Close(context.Background())) stats := p.Metrics().GetStats() - assert.Equal(t, int(atomic.LoadInt32(&processed)), stats.Processed) + assert.Equal(t, int(processed.Load()), stats.Processed) assert.Equal(t, 0, stats.Errors) assert.Equal(t, 0, stats.Dropped) assert.Greater(t, stats.ProcessingTime, time.Duration(0)) @@ -948,12 +948,12 @@ func TestMiddleware_Practical(t *testing.T) { return nil }) - var totalTime int64 + var totalTime atomic.Int64 timingMiddleware := func(next Worker[string]) Worker[string] { return WorkerFunc[string](func(ctx context.Context, v string) error { start := time.Now() err := next.Do(ctx, v) - atomic.AddInt64(&totalTime, time.Since(start).Microseconds()) + totalTime.Add(time.Since(start).Microseconds()) return err }) } @@ -964,7 +964,7 @@ func TestMiddleware_Practical(t *testing.T) { p.Submit("test") require.NoError(t, p.Close(context.Background())) - assert.Greater(t, atomic.LoadInt64(&totalTime), int64(1000), + assert.Greater(t, totalTime.Load(), int64(1000), "should measure time greater than 1ms") }) }