Enhanced README and added error handling and result retrieval

- Updated README to include sections on task timeout, error handling, and result retrieval.
- Modified task function signature to return a result and an error.
- Added resultCallback and errorCallback to the goPool struct.
- Updated NewGoPool function to accept options for result and error callbacks.
- Modified worker start function to handle task results and errors.
- Updated tests to reflect changes in task function signature and to test error and result handling.

Signed-off-by: Daniel Hu <[email protected]>

README.md

@@ -12,7 +12,7 @@ GoPool is a high-performance, feature-rich, and easy-to-use worker pool library
 
 - **Graceful Shutdown**: GoPool can shut down gracefully. It stops accepting new tasks and waits for all ongoing tasks to complete before shutting down when there are no more tasks or a shutdown signal is received.
 
-- **Error Handling**: GoPool can handle errors that occur during task execution.
+- **Task Error Handling**: GoPool can handle errors that occur during task execution.
 
 - **Task Timeout Handling**: GoPool can handle task execution timeouts. If a task is not completed within the specified timeout period, the task is considered failed and a timeout error is returned.
 
@@ -112,6 +112,96 @@ func main() {
 
 In this example, the pool starts with 50 workers. If the number of tasks in the queue exceeds (MaxWorkers - MinWorkers) / 2 + MinWorkers, the pool will add more workers. If the number of tasks in the queue is less than MinWorkers, the pool will remove some workers.
 
+## Task Timeout Handling
+
+GoPool supports task timeout. If a task takes longer than the specified timeout, it will be cancelled. This feature can be enabled by setting the `WithTimeout` option when creating the pool.
+
+Here is an example of how to use GoPool with task timeout:
+
+```go
+package main
+
+import (
+ "time"
+
+ "github.com/devchat-ai/gopool"
+)
+
+func main() {
+ pool := gopool.NewGoPool(100, gopool.WithTimeout(1*time.Second))
+ for i := 0; i < 1000; i++ {
+ pool.AddTask(func() (interface{}, error) {
+ time.Sleep(2 * time.Second)
+ return nil, nil
+ })
+ }
+ pool.Release()
+}
+```
+
+In this example, the task will be cancelled if it takes longer than 1 second.
+
+## Task Error Handling
+
+GoPool supports task error handling. If a task returns an error, the error callback function will be called. This feature can be enabled by setting the `WithErrorCallback` option when creating the pool.
+
+Here is an example of how to use GoPool with error handling:
+
+```go
+package main
+
+import (
+ "errors"
+ "fmt"
+
+ "github.com/devchat-ai/gopool"
+)
+
+func main() {
+ pool := gopool.NewGoPool(100, gopool.WithErrorCallback(func(err error) {
+ fmt.Println("Task error:", err)
+ }))
+ for i := 0; i < 1000; i++ {
+ pool.AddTask(func() (interface{}, error) {
+ return nil, errors.New("task error")
+ })
+ }
+ pool.Release()
+}
+```
+
+In this example, if a task returns an error, the error will be printed to the console.
+
+## Task Result Retrieval
+
+GoPool supports task result retrieval. If a task returns a result, the result callback function will be called. This feature can be enabled by setting the `WithResultCallback` option when creating the pool.
+
+Here is an example of how to use GoPool with task result retrieval:
+
+```go
+package main
+
+import (
+ "fmt"
+
+ "github.com/devchat-ai/gopool"
+)
+
+func main() {
+ pool := gopool.NewGoPool(100, gopool.WithResultCallback(func(result interface{}) {
+ fmt.Println("Task result:", result)
+ }))
+ for i := 0; i < 1000; i++ {
+ pool.AddTask(func() (interface{}, error) {
+ return "task result", nil
+ })
+ }
+ pool.Release()
+}
+```
+
+In this example, if a task returns a result, the result will be printed to the console.
+
 ## Performance Testing
 
 We have conducted several performance tests to evaluate the efficiency and performance of GoPool. Here are the results:

gopool.go

@@ -5,8 +5,9 @@ import (
  "time"
 )
 
-// Task represents a function that will be executed by a worker.
-type task func()
+// task represents a function that will be executed by a worker.
+// It returns a result and an error.
+type task func() (interface{}, error)
 
 // goPool represents a pool of workers.
 type goPool struct {
@@ -18,6 +19,8 @@ type goPool struct {
  lock sync.Locker
  cond *sync.Cond
  timeout time.Duration
+ resultCallback func(interface{})
+ errorCallback func(error)
 }
 
 // NewGoPool creates a new pool of workers.

gopool_test.go

@@ -1,84 +1,120 @@
 package gopool
 
 import (
- "sync"
- "testing"
- "time"
+ "sync"
+ "testing"
+ "time"
+ "errors"
 
- "github.com/daniel-hutao/spinlock"
+ "github.com/daniel-hutao/spinlock"
 )
 
 func TestGoPoolWithMutex(t *testing.T) {
- pool := NewGoPool(100, WithLock(new(sync.Mutex)))
- for i := 0; i < 1000; i++ {
- pool.AddTask(func() {
- time.Sleep(10 * time.Millisecond)
- })
- }
- pool.Release()
+ pool := NewGoPool(100, WithLock(new(sync.Mutex)))
+ for i := 0; i < 1000; i++ {
+ pool.AddTask(func() (interface{}, error) {
+ time.Sleep(10 * time.Millisecond)
+ return nil, nil
+ })
+ }
+ pool.Release()
 }
 
 func TestGoPoolWithSpinLock(t *testing.T) {
- pool := NewGoPool(100, WithLock(new(spinlock.SpinLock)))
- for i := 0; i < 1000; i++ {
- pool.AddTask(func() {
- time.Sleep(10 * time.Millisecond)
- })
- }
- pool.Release()
+ pool := NewGoPool(100, WithLock(new(spinlock.SpinLock)))
+ for i := 0; i < 1000; i++ {
+ pool.AddTask(func() (interface{}, error) {
+ time.Sleep(10 * time.Millisecond)
+ return nil, nil
+ })
+ }
+ pool.Release()
 }
 
 func BenchmarkGoPoolWithMutex(b *testing.B) {
- var wg sync.WaitGroup
- var taskNum = int(1e6)
- pool := NewGoPool(5e4, WithLock(new(sync.Mutex)))
+ var wg sync.WaitGroup
+ var taskNum = int(1e6)
+ pool := NewGoPool(5e4, WithLock(new(sync.Mutex)))
 
- b.ResetTimer()
- for i := 0; i < b.N; i++ {
- wg.Add(taskNum)
- for num := 0; num < taskNum; num++ {
- pool.AddTask(func() {
- time.Sleep(10 * time.Millisecond)
- wg.Done()
- })
- }
- }
- wg.Wait()
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ wg.Add(taskNum)
+ for num := 0; num < taskNum; num++ {
+ pool.AddTask(func() (interface{}, error) {
+ time.Sleep(10 * time.Millisecond)
+ wg.Done()
+ return nil, nil
+ })
+ }
+ }
+ wg.Wait()
  b.StopTimer()
- pool.Release()
+ pool.Release()
 }
 
 func BenchmarkGoPoolWithSpinLock(b *testing.B) {
- var wg sync.WaitGroup
- var taskNum = int(1e6)
- pool := NewGoPool(5e4, WithLock(new(spinlock.SpinLock)))
+ var wg sync.WaitGroup
+ var taskNum = int(1e6)
+ pool := NewGoPool(5e4, WithLock(new(spinlock.SpinLock)))
 
- b.ResetTimer()
- for i := 0; i < b.N; i++ {
- wg.Add(taskNum)
- for num := 0; num < taskNum; num++ {
- pool.AddTask(func() {
- time.Sleep(10 * time.Millisecond)
- wg.Done()
- })
- }
- }
- wg.Wait()
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ wg.Add(taskNum)
+ for num := 0; num < taskNum; num++ {
+ pool.AddTask(func() (interface{}, error) {
+ time.Sleep(10 * time.Millisecond)
+ wg.Done()
+ return nil, nil
+ })
+ }
+ }
+ wg.Wait()
  b.StopTimer()
- pool.Release()
+ pool.Release()
 }
 
 func BenchmarkGoroutines(b *testing.B) {
- var wg sync.WaitGroup
- var taskNum = int(1e6)
+ var wg sync.WaitGroup
+ var taskNum = int(1e6)
+
+ for i := 0; i < b.N; i++ {
+ wg.Add(taskNum)
+ for num := 0; num < taskNum; num++ {
+ go func() (interface{}, error) {
+ time.Sleep(10 * time.Millisecond)
+ wg.Done()
+ return nil, nil
+ }()
+ }
+ }
+}
 
- for i := 0; i < b.N; i++ {
- wg.Add(taskNum)
- for num := 0; num < taskNum; num++ {
- go func() {
- time.Sleep(10 * time.Millisecond)
- wg.Done()
- }()
+func TestGoPoolWithError(t *testing.T) {
+ var errTaskError = errors.New("task error")
+ pool := NewGoPool(100, WithErrorCallback(func(err error) {
+ if err != errTaskError {
+ t.Errorf("Expected error %v, but got %v", errTaskError, err)
  }
+ }))
+ for i := 0; i< 1000; i++ {
+ pool.AddTask(func() (interface{}, error) {
+ return nil, errTaskError
+ })
  }
+ pool.Release()
+}
+
+func TestGoPoolWithResult(t *testing.T) {
+ var expectedResult = "task result"
+ pool := NewGoPool(100, WithResultCallback(func(result interface{}) {
+ if result != expectedResult {
+ t.Errorf("Expected result %v, but got %v", expectedResult, result)
+ }
+ }))
+ for i := 0; i< 1000; i++ {
+ pool.AddTask(func() (interface{}, error) {
+ return expectedResult, nil
+ })
+ }
+ pool.Release()
 }

option.go

@@ -29,3 +29,17 @@ func WithTimeout(timeout time.Duration) Option {
  p.timeout = timeout
  }
 }
+
+// WithResultCallback sets the result callback for the pool.
+func WithResultCallback(callback func(interface{})) Option {
+ return func(p *goPool) {
+ p.resultCallback = callback
+ }
+}
+
+// WithErrorCallback sets the error callback for the pool.
+func WithErrorCallback(callback func(error)) Option {
+ return func(p *goPool) {
+ p.errorCallback = callback
+ }
+}

worker.go

@@ -20,6 +20,9 @@ func (w *worker) start(pool *goPool, workerIndex int) {
  go func() {
  for t := range w.taskQueue {
  if t != nil {
+ var result interface{}
+ var err error
+
  if pool.timeout > 0 {
  // Create a context with timeout
  ctx, cancel := context.WithTimeout(context.Background(), pool.timeout)
@@ -30,21 +33,33 @@ func (w *worker) start(pool *goPool, workerIndex int) {
 
  // Run the task in a separate goroutine
  go func() {
- t()
+ result, err = t()
  close(done)
  }()
 
  // Wait for the task to finish or for the context to timeout
  select {
  case <-done:
  // The task finished successfully
+ if err != nil && pool.errorCallback != nil {
+ pool.errorCallback(err)
+ } else if pool.resultCallback != nil {
+ pool.resultCallback(result)
+ }
  case <-ctx.Done():
  // The context timed out, the task took too long
- fmt.Println("Task timed out")
+ if pool.errorCallback != nil {
+ pool.errorCallback(fmt.Errorf("Task timed out"))
+ }
  }
  } else {
  // If timeout is not set or is zero, just run the task
- t()
+ result, err = t()
+ if err != nil && pool.errorCallback != nil {
+ pool.errorCallback(err)
+ } else if pool.resultCallback != nil {
+ pool.resultCallback(result)
+ }
  }
  }
  pool.pushWorker(workerIndex)