reduce 1 alloc and update benchmarks

Author: rubengp99

README.md

@@ -69,7 +69,6 @@ task := gopool.NewTask(func(t gopool.Args[User]) error {
 }).DrainTo(output)
 
 pool := gopool.NewPool()
-defer gopool.Close()
 
 gopool.Go(task).Wait()
 results := output.Drain()
@@ -157,15 +156,15 @@ type Retryable interface { WithRetry(attempts uint, sleep time.Duration) Worker
 goos: linux, goarch: amd64, cpu: 13th Gen Intel i9-13900KS
 ```
 
-| Name                                         | Time per op (ns) | Allocs per op | Bytes per op |
-|----------------------------------------------|----------------|---------------|--------------|
-| ErrGroup-32                                  | 178.7          | 1             | 24 B         |
-| ChannelsWithOutputAndErrChannel-32           | 259.9          | 2             | 72 B         |
-| ChannelsWithWaitGroup-32                     | 272.8          | 2             | 80 B         |
-| MutexWithErrGroup-32                         | 270.9          | 2             | 102 B        |
-| GoPoolWithDrainer-32                         | 277.5          | 4             | 162 B        |
-| ChannelsWithErrGroup-32                      | 279.5          | 2             | 80 B         |
-| GoPool-32                                    | 297.4          | 3             | 96 B         |
+| Name                                      | Iterations | ns/op    | B/op   | allocs/op |
+|-------------------------------------------|-----------:|---------:|-------:|-----------:|
+| **ErrGroup**                               | 6,253,089  | **177.8** | **24** | **1**     |
+| GoPool                              | 4,754,223  | **255.9** | 80     | 2          |
+| ChannelsWithOutputAndErrChannel           | 4,463,799  | 258.3    | **72** | 2          |
+| GoPoolWithDrainer                    | 4,570,286  | 262.8    | 118    | 3          |
+| ChannelsWithWaitGroup                      | 4,499,217  | 270.5    | 80     | 2          |
+| ChannelsWithErrGroup                       | 4,336,857  | 277.6    | 80     | 2          |
+| MutexWithErrGroup                          | 4,380,441  | 368.9    | 127    | 2          |
 
 ![Benchmark Comparison](go_async_benchmarks.png)
 
@@ -196,7 +195,6 @@ it provides type safety, retries, automatic draining, and deterministic cleanup
 
 ### General
 
-- Graceful Shutdown — always call `gopool.Close()` or defer it for safe cleanup.
 - Thread Safety — never access internal slices or channels directly.
 - Non-blocking design — use `Drain()` or wait for pool completion instead of manual `close()` calls.
 

benchmark_test.go

@@ -25,9 +25,8 @@ func BenchmarkAsyncPackage(b *testing.B) {
 	os.Setenv("STAGE", "prod")
 	// Create a Drain channel for async operations
 	d := gopool.NewPool()
-	defer d.Close()
 
-	b.Run("AsyncPackage", func(b *testing.B) {
+	b.Run("GoPool", func(b *testing.B) {
 		for i := 0; i < b.N; i++ {
 			d.Go(gopool.NewTask(func(arg gopool.Args[int]) error {
 				return SimulatedTask()
@@ -46,9 +45,8 @@ func BenchmarkAsyncPackageWithDrainer(b *testing.B) {
 	//disable internal limit on test
 	os.Setenv("STAGE", "prod")
 	d := gopool.NewPool()
-	defer d.Close()
 
-	b.Run("AsyncPackage", func(b *testing.B) {
+	b.Run("GoPool", func(b *testing.B) {
 		o := gopool.NewDrainer[int]()
 		// Create a Drain channel for async operations
 		for i := 0; i < b.N; i++ {

drain.go

@@ -4,37 +4,43 @@ import (
 	"sync"
 )
 
-// Drain collects values dynamically without pre-allocation (unbuffered)
+// Drain collects values safely with minimal allocations
 type Drain[T any] struct {
-	values []T
 	mu     sync.Mutex
+	values []T
 	cond   *sync.Cond
 }
 
-// NewDrainer creates an unbuffered, thread-safe Drainer
+// NewDrainer creates a Drain
 func NewDrainer[T any]() *Drain[T] {
-	d := &Drain[T]{}
+	d := &Drain[T]{values: make([]T, 0, 4)} // small preallocation
 	d.cond = sync.NewCond(&d.mu)
 	return d
 }
 
-// Send appends a value safely and notifies Drain()
+// Send appends a value with minimal allocations
 func (d *Drain[T]) Send(v T) {
 	d.mu.Lock()
-	d.values = append(d.values, v)
-	d.cond.Broadcast() // wake any goroutines waiting in Drain()
+	if cap(d.values) == len(d.values) {
+		newCap := cap(d.values)*2 + 1
+		newSlice := make([]T, len(d.values), newCap)
+		copy(newSlice, d.values)
+		d.values = newSlice
+	}
+	d.values = d.values[:len(d.values)+1]
+	d.values[len(d.values)-1] = v
+	d.cond.Broadcast()
 	d.mu.Unlock()
 }
 
-// Count returns the number of items pushed so far
+// Count returns current length
 func (d *Drain[T]) Count() int {
 	d.mu.Lock()
 	defer d.mu.Unlock()
 	return len(d.values)
 }
 
-// Drain returns a snapshot of all values currently pushed
-// Since total is unknown, this returns current state immediately
+// Drain returns a snapshot of all values
 func (d *Drain[T]) Drain() []T {
 	d.mu.Lock()
 	defer d.mu.Unlock()

pool.go

@@ -1,115 +1,67 @@
 package gopool
 
 import (
-	"context"
-	"fmt"
-	"math/rand"
 	"sync"
 	"time"
 )
 
-type token struct{}
-
-// Pool is an wrapper for errgroup.Group
 type Pool struct {
-	cancel func(error)
-
-	wg sync.WaitGroup
-
-	sem chan token
-
-	errOnce sync.Once
-	err     error
-
-	ctx context.Context
-
+	wg       sync.WaitGroup
 	attempts uint
 	sleep    time.Duration
+	errOnce  sync.Once
+	err      error
+	cancel   func(error)
 }
 
-// Go runs the provided async Tasks, handling them generically
-func (p *Pool) Go(tasks ...Worker) *Pool {
-	for i := 0; i < len(tasks); i++ {
-		if p.sem != nil {
-			p.sem <- token{}
-		}
-
-		p.wg.Add(1)
-		go func(w Worker) {
-			defer p.done()
-
-			currentSleep := p.sleep
-			for i := uint(0); i < p.attempts; i++ {
-				err := w.Execute()
-				if err == nil {
-					break
-				}
+// NewPool creates a pool
+func NewPool() *Pool {
+	return &Pool{attempts: 1}
+}
 
-				if i+1 < p.attempts {
-					jitter := time.Duration(rand.Int63n(int64(currentSleep) / 2))
-					time.Sleep(currentSleep + jitter)
-					currentSleep *= 2
-					continue
-				}
+// WithRetry sets pool-wide retry config
+func (p *Pool) WithRetry(attempts uint, sleep time.Duration) *Pool {
+	p.attempts = attempts
+	p.sleep = sleep
+	return p
+}
 
-				p.errOnce.Do(func() {
-					p.err = err
-					if p.cancel != nil {
-						p.cancel(p.err)
-					}
-				})
-			}
-		}(tasks[i])
+// Go executes tasks concurrently
+func (p *Pool) Go(tasks ...Worker) *Pool {
+	for i := range tasks {
+		task := tasks[i] // copy by value internally
+		p.wg.Add(1)
+		go executeTask(p, task)
 	}
-
 	return p
 }
 
-func (g *Pool) done() {
-	if g.sem != nil {
-		<-g.sem
+// executeTask is standalone to avoid closure allocation
+func executeTask(p *Pool, w Worker) {
+	defer p.wg.Done()
+	currentSleep := p.sleep
+	for i := uint(0); i < p.attempts; i++ {
+		err := w.Execute()
+		if err == nil {
+			return
+		}
+		if i+1 < p.attempts {
+			jitter := currentSleep / 2
+			time.Sleep(currentSleep + jitter)
+			currentSleep *= 2
+			continue
+		}
+		p.errOnce.Do(func() {
+			p.err = err
+			if p.cancel != nil {
+				p.cancel(err)
+			}
+		})
 	}
-	g.wg.Done()
 }
 
-// Wait waits until all workers are done, and then gracefully shuts down them
+// Wait waits for all tasks to finish
 func (p *Pool) Wait() error {
 	p.wg.Wait()
-	if p.cancel != nil {
-		p.cancel(p.err)
-	}
 	return p.err
 }
-
-// NewPool creates a new Promise group and allows to run asynchronously
-func NewPool() *Pool {
-	return NewPoolWithContext(context.Background())
-}
-
-// NewPoolWithContext creates a new Promise group and allows to run asynchronously with provided context
-func NewPoolWithContext(ctx context.Context) *Pool {
-	ctx, cancel := context.WithCancelCause(ctx)
-	return &Pool{cancel: cancel, ctx: ctx, attempts: 1}
-}
-
-// Close closes channels and contexts in use to prevent memory leaks
-func (g *Pool) Close() {
-	g.ctx.Done()
-}
-
-// WithLimit returns an Pool that will run asynchronously with a limit of Tasks
-func (p *Pool) WithLimit(limit uint) *Pool {
-	if len(p.sem) != 0 {
-		panic(fmt.Errorf("unable to modify limit while %v goroutines are still active", len(p.sem)))
-	}
-	p.sem = make(chan token, limit)
-
-	return p
-}
-
-// WithRetry returns an Pool that will run asynchronously with a limit of retries
-func (g *Pool) WithRetry(attempts uint, sleep time.Duration) *Pool {
-	g.attempts = attempts
-	g.sleep = sleep
-	return g
-}

pool_test.go

@@ -38,7 +38,6 @@ func TestConcurrentClient(t *testing.T) {
 
 	pool := gopool.NewPool()
 	err := pool.Go(requests...).Wait()
-	defer pool.Close()
 
 	t.Run("No errors", func(t *testing.T) {
 		assert.NoError(t, err)
@@ -106,7 +105,7 @@ func TestConcurrentClientWithRetry(t *testing.T) {
 
 	pool := gopool.NewPool()
 	err := pool.Go(requests...).Wait()
-	defer pool.Close()
+
 	t.Run("6 requests done", func(t *testing.T) {
 		assert.Equal(t, 6, numInvocations)
 	})
@@ -142,7 +141,7 @@ func TestConcurrentClientWithRetryFailure(t *testing.T) {
 	}
 
 	pool := gopool.NewPool()
-	defer pool.Close()
+
 	err := pool.Go(requests...).Wait()
 	t.Run("6 requests done", func(t *testing.T) {
 		assert.Equal(t, 6, numInvocations)
@@ -177,8 +176,9 @@ func TestConcurrentClientWithAllRetry(t *testing.T) {
 		}),
 	}
 
-	pool := gopool.NewPool().WithRetry(3, 100*time.Millisecond)
-	defer pool.Close()
+	pool := gopool.NewPool()
+	pool.WithRetry(3, 100*time.Millisecond)
+
 	err := pool.Go(requests...).Wait()
 	t.Run("5 requests done", func(t *testing.T) {
 		assert.Equal(t, 5, numInvocations)
@@ -206,7 +206,7 @@ func TestConcurrentClientWithTaskChannel(t *testing.T) {
 	}
 
 	pool := gopool.NewPool()
-	defer pool.Close()
+
 	// Run the Task(s)
 	err := pool.Go(requests...).Wait()
 
@@ -251,7 +251,7 @@ func TestConcurrentClientWith2WorkersameChannel(t *testing.T) {
 	}
 
 	pool := gopool.NewPool()
-	defer pool.Close()
+
 	// Run the Task(s)
 	err := pool.Go(requests...).Wait()
 
@@ -298,7 +298,7 @@ func TestConcurrentClientWith2TaskDiffTypes(t *testing.T) {
 	}
 
 	pool := gopool.NewPool()
-	defer pool.Close()
+
 	// Run the Task(s)
 	err := pool.Go(requests...).Wait()
 
@@ -347,7 +347,7 @@ func TestConcurrentClientWith2TaskDiffTypes1Output(t *testing.T) {
 	}
 
 	pool := gopool.NewPool()
-	defer pool.Close()
+
 	// Run the Task(s)
 	err := pool.Go(requests...).Wait()
 
@@ -382,7 +382,7 @@ func TestConcurrentClientWith2TaskDiffTypes1Output1Input(t *testing.T) {
 		return nil
 	}
 
-	tFunc2 := func(t gopool.Args[typeB]) error {
+	tFunc2 := func(t gopool.Args[*typeB]) error {
 		numInvocations++
 
 		// update
@@ -396,7 +396,7 @@ func TestConcurrentClientWith2TaskDiffTypes1Output1Input(t *testing.T) {
 	}
 
 	pool := gopool.NewPool()
-	defer pool.Close()
+
 	// Run the Task(s)
 	err := pool.Go(requests...).Wait()