README
¶
AtomicGo | pool
Documentation | Contributing | Code of Conduct
go get atomicgo.dev/pool
pool
import "atomicgo.dev/pool"
Package pool provides a generic and concurrent worker pool implementation. It allows you to enqueue tasks and process them using a fixed number of workers.
Example (Demo)
package main
import (
"atomicgo.dev/pool"
"context"
"log"
"time"
)
func main() {
// Create a new pool with 3 workers
p := pool.New[int](pool.Config{
MaxWorkers: 3,
})
// Set the task handler to process integers, simulating a 1-second task
p.SetHandler(func(ctx context.Context, i int) error {
log.Printf("Processing %d", i)
time.Sleep(time.Second)
return nil
})
// Start the pool
p.Start()
// Add 10 tasks to the pool
for i := 0; i < 10; i++ {
p.Add(i)
}
// Close the pool and wait for all tasks to complete
p.Close()
}
Example (Error Handling)
package main
import (
"atomicgo.dev/pool"
"context"
"fmt"
"log"
)
func main() {
// Create a new pool
p := pool.New[int](pool.Config{
MaxWorkers: 2,
})
// Set the task handler, which returns an error for even numbers
p.SetHandler(func(ctx context.Context, i int) error {
if i%2 == 0 {
return fmt.Errorf("error processing %d", i)
}
log.Printf("Successfully processed %d", i)
return nil
})
// Set a custom error handler that logs errors
p.SetErrorHandler(func(err error, pool *pool.Pool[int]) {
log.Printf("Encountered an error: %v", err)
// Optional: you can call pool.Kill() here if you want to stop the whole pool on the first error
})
// Start the pool
p.Start()
// Add 5 tasks to the pool
for i := 0; i < 5; i++ {
p.Add(i)
}
// Close the pool and wait for all tasks to complete
p.Close()
}
Example (Timeout)
package main
import (
"atomicgo.dev/pool"
"context"
"log"
"time"
)
func main() {
// Create a new pool with a short timeout of 2 seconds
p := pool.New[int](pool.Config{
MaxWorkers: 3,
Timeout: time.Second * 2,
})
// Set the task handler with a task that may exceed the timeout
p.SetHandler(func(ctx context.Context, i int) error {
log.Printf("Processing %d", i)
time.Sleep(time.Second * 3) // This sleep is longer than the timeout
return nil
})
// Start the pool
p.Start()
// Add tasks to the pool
for i := 0; i < 5; i++ {
p.Add(i)
}
// Close the pool and wait for all tasks to complete
p.Close()
}
Index
type Config
Config struct defines the configuration parameters for the Pool. - MaxWorkers: The maximum number of concurrent workers in the pool. - Timeout: The maximum duration for processing a single task. If a task takes longer, it will be terminated.
If the Timeout is set to 0, tasks will not be terminated. This is the default behavior. If MaxWorkers is set to 0, it will be set to the number of logical CPUs on the machine.
type Config struct {
MaxWorkers int
Timeout time.Duration
}
type ErrorHandler
ErrorHandler is a function type for handling errors. It provides a way for users to define custom error handling logic. The function receives an error and a pointer to the pool, allowing users to log errors, modify the pool, or perform other actions.
type ErrorHandler[T any] func(error, *Pool[T])
type Pool
Pool struct represents a pool of workers processing tasks of type T. It encapsulates the task handling logic, error handling, and synchronization mechanisms.
type Pool[T any] struct {
// contains filtered or unexported fields
}
func New
func New[T any](config Config) *Pool[T]
New creates and returns a new pool with the specified configuration. It initializes the internal structures but does not start the worker goroutines. - config: Configuration settings for the pool, including max workers and task timeout.
func (*Pool[T]) Add
func (p *Pool[T]) Add(item T)
Add enqueues a task into the pool. If the pool's worker goroutines are running, the task will be picked up for processing. If the pool is not running or has been closed, the behavior of Add is undefined and may result in a deadlock or panic. - item: The task to be added to the pool for processing.
func (*Pool[T]) Close
func (p *Pool[T]) Close()
Close gracefully shuts down the pool. It stops accepting new tasks and waits for all ongoing tasks to complete. This method should be called to ensure a clean shutdown of the pool.
func (*Pool[T]) Kill
func (p *Pool[T]) Kill()
Kill immediately stops all workers in the pool. It cancels the context, causing all worker goroutines to exit. Any ongoing tasks may be left unfinished. This method is useful for emergency shutdown scenarios.
func (*Pool[T]) SetErrorHandler
func (p *Pool[T]) SetErrorHandler(handler ErrorHandler[T]) *Pool[T]
SetErrorHandler sets a custom error handling function for the pool. It is optional. The error handler allows custom logic to be executed when a task processing results in an error. - handler: The function to be called when a task encounters an error.
func (*Pool[T]) SetHandler
func (p *Pool[T]) SetHandler(handler func(context.Context, T) error) *Pool[T]
SetHandler sets the task handling function for the pool. It must be called before starting the pool. The handler function takes a context (for timeout control) and a task of type T, and returns an error. - handler: The function that will be called to process each task.
func (*Pool[T]) Start
func (p *Pool[T]) Start()
Start initiates the worker goroutines. It should be called after setting up the task handler and optionally the error handler. This method spins up MaxWorkers number of goroutines, each listening for tasks to process.
Generated by gomarkdoc
AtomicGo.dev · with ❤️ by @MarvinJWendt | MarvinJWendt.com
Documentation
¶
Overview ¶
Package pool provides a generic and concurrent worker pool implementation. It allows you to enqueue tasks and process them using a fixed number of workers.
Example (Demo) ¶
package main
import (
"atomicgo.dev/pool"
"context"
"log"
"time"
)
func main() {
// Create a new pool with 3 workers
p := pool.New[int](pool.Config{
MaxWorkers: 3,
})
// Set the task handler to process integers, simulating a 1-second task
p.SetHandler(func(ctx context.Context, i int) error {
log.Printf("Processing %d", i)
time.Sleep(time.Second)
return nil
})
// Start the pool
p.Start()
// Add 10 tasks to the pool
for i := 0; i < 10; i++ {
p.Add(i)
}
// Close the pool and wait for all tasks to complete
p.Close()
}
Output:
Example (ErrorHandling) ¶
package main
import (
"atomicgo.dev/pool"
"context"
"fmt"
"log"
)
func main() {
// Create a new pool
p := pool.New[int](pool.Config{
MaxWorkers: 2,
})
// Set the task handler, which returns an error for even numbers
p.SetHandler(func(ctx context.Context, i int) error {
if i%2 == 0 {
return fmt.Errorf("error processing %d", i)
}
log.Printf("Successfully processed %d", i)
return nil
})
// Set a custom error handler that logs errors
p.SetErrorHandler(func(err error, pool *pool.Pool[int]) {
log.Printf("Encountered an error: %v", err)
// Optional: you can call pool.Kill() here if you want to stop the whole pool on the first error
})
// Start the pool
p.Start()
// Add 5 tasks to the pool
for i := 0; i < 5; i++ {
p.Add(i)
}
// Close the pool and wait for all tasks to complete
p.Close()
}
Output:
Example (Timeout) ¶
package main
import (
"atomicgo.dev/pool"
"context"
"log"
"time"
)
func main() {
// Create a new pool with a short timeout of 2 seconds
p := pool.New[int](pool.Config{
MaxWorkers: 3,
Timeout: time.Second * 2,
})
// Set the task handler with a task that may exceed the timeout
p.SetHandler(func(ctx context.Context, i int) error {
log.Printf("Processing %d", i)
time.Sleep(time.Second * 3) // This sleep is longer than the timeout
return nil
})
// Start the pool
p.Start()
// Add tasks to the pool
for i := 0; i < 5; i++ {
p.Add(i)
}
// Close the pool and wait for all tasks to complete
p.Close()
}
Output:
Index ¶
Examples ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Config ¶
Config struct defines the configuration parameters for the Pool. - MaxWorkers: The maximum number of concurrent workers in the pool. - Timeout: The maximum duration for processing a single task. If a task takes longer, it will be terminated.
If the Timeout is set to 0, tasks will not be terminated. This is the default behavior. If MaxWorkers is set to 0, it will be set to the number of logical CPUs on the machine.
type ErrorHandler ¶
ErrorHandler is a function type for handling errors. It provides a way for users to define custom error handling logic. The function receives an error and a pointer to the pool, allowing users to log errors, modify the pool, or perform other actions.
type Pool ¶
type Pool[T any] struct { // contains filtered or unexported fields }
Pool struct represents a pool of workers processing tasks of type T. It encapsulates the task handling logic, error handling, and synchronization mechanisms.
func New ¶
New creates and returns a new pool with the specified configuration. It initializes the internal structures but does not start the worker goroutines. - config: Configuration settings for the pool, including max workers and task timeout.
func (*Pool[T]) Add ¶
func (p *Pool[T]) Add(item T)
Add enqueues a task into the pool. If the pool's worker goroutines are running, the task will be picked up for processing. If the pool is not running or has been closed, the behavior of Add is undefined and may result in a deadlock or panic. - item: The task to be added to the pool for processing.
func (*Pool[T]) Close ¶
func (p *Pool[T]) Close()
Close gracefully shuts down the pool. It stops accepting new tasks and waits for all ongoing tasks to complete. This method should be called to ensure a clean shutdown of the pool.
func (*Pool[T]) Kill ¶
func (p *Pool[T]) Kill()
Kill immediately stops all workers in the pool. It cancels the context, causing all worker goroutines to exit. Any ongoing tasks may be left unfinished. This method is useful for emergency shutdown scenarios.
func (*Pool[T]) SetErrorHandler ¶
func (p *Pool[T]) SetErrorHandler(handler ErrorHandler[T]) *Pool[T]
SetErrorHandler sets a custom error handling function for the pool. It is optional. The error handler allows custom logic to be executed when a task processing results in an error. - handler: The function to be called when a task encounters an error.
func (*Pool[T]) SetHandler ¶
SetHandler sets the task handling function for the pool. It must be called before starting the pool. The handler function takes a context (for timeout control) and a task of type T, and returns an error. - handler: The function that will be called to process each task.
Source Files