improvement: Adding workerpool package and implementation #14

2025-12-17T20:54:05Z

k-simons commented

2025-12-17 20:54:05 +00:00

(Migrated from github.com)

Adding workerpool package and implementation
- Adds a new workerpool package providing generic worker pool primitives for concurrent task execution
- Implements four worker pool types: SupplierWorkerPool[R], FunctionWorkerPool[T,R], RunnableWorkerPool, and ConsumerWorkerPool[T]
- Supports configurable concurrency limits via WithMaxNumberOfThreads and optional metrics via WithMetricTags

- Adding workerpool package and implementation - Adds a new workerpool package providing generic worker pool primitives for concurrent task execution - Implements four worker pool types: SupplierWorkerPool[R], FunctionWorkerPool[T,R], RunnableWorkerPool, and ConsumerWorkerPool[T] - Supports configurable concurrency limits via WithMaxNumberOfThreads and optional metrics via WithMetricTags

changelog-app[bot] commented

2025-12-17 20:54:08 +00:00

(Migrated from github.com)

Generate changelog in `changelog/@unreleased`

Type (Select exactly one)

Feature (Adding new functionality)
Improvement (Improving existing functionality)
Fix (Fixing an issue with existing functionality)
Break (Creating a new major version by breaking public APIs)
Deprecation (Removing functionality in a non-breaking way)
Migration (Automatically moving data/functionality to a new system)

Description Ksimons/worker pool final

Check the box to generate changelog(s)

Generate changelog entry

### Generate changelog in <changelog-dir>`changelog/@unreleased`</changelog-dir> **Type** _(Select exactly one)_ <changelog-type> - [ ] **Feature** (Adding new functionality) - [ ] **Improvement** (Improving existing functionality) - [ ] **Fix** (Fixing an issue with existing functionality) - [ ] **Break** (Creating a _new major version_ by breaking public APIs) - [ ] **Deprecation** (Removing functionality in a non-breaking way) - [ ] **Migration** (Automatically moving data/functionality to a new system) </changelog-type> **Description** <changelog-description> Ksimons/worker pool final </changelog-description><changelog-links></changelog-links> **Check the box to generate changelog(s)** - [ ] Generate changelog entry

changelog-app[bot] commented

2025-12-17 21:57:15 +00:00

(Migrated from github.com)

✅ Successfully generated changelog entry!

Entry generated via PR title

To modify this entry, edit PR title using proper format.

📋Changelog Preview

💡 Improvements

Adding workerpool package and implementation (#14)

### ✅ **Successfully generated changelog entry!** #### Entry generated via PR title To modify this entry, edit PR title using proper format. --- ### 📋Changelog Preview ### 💡 Improvements * Adding workerpool package and implementation ([#14](https://github.com/palantir/witchcraft-go-tasks/pull/14))

nmiyake (Migrated from github.com) reviewed

2025-12-17 22:08:17 +00:00

nmiyake (Migrated from github.com) left a comment

Submitted some feedback for computing_future.go -- will look at worker pool files later.

Submitted some feedback for `computing_future.go` -- will look at worker pool files later.

workerpool/internal/computing_future.go

Should this be in another directory/package under internal so that package name is not just internal?

Should this be in another directory/package under `internal` so that package name is not just `internal`?

// ComputingFuture implements async.Future[T] and defines the Compute(ctx context.Context) function.

```suggestion // ComputingFuture implements async.Future[T] and defines the Compute(ctx context.Context) function. ```

// NewDefaultComputingFuture returns the default implementation of ComputingFuture[T].
// The Compute function of this implementation runs the provided supplier exactly once and stores the
// result. Any panics that occur in the supplier are caught and treated as an error of the supplier.
// The Get function of this implementation returns the result computed by Compute (and blocks until Compute has been called and completes if it has not yet been called).
// Neither the Compute nor the Get implementations check for context cancellation: if cancellation support is needed, it should be handled by the provided supplier.

```suggestion // NewDefaultComputingFuture returns the default implementation of ComputingFuture[T]. // The Compute function of this implementation runs the provided supplier exactly once and stores the // result. Any panics that occur in the supplier are caught and treated as an error of the supplier. // The Get function of this implementation returns the result computed by Compute (and blocks until Compute has been called and completes if it has not yet been called). // Neither the Compute nor the Get implementations check for context cancellation: if cancellation support is needed, it should be handled by the provided supplier. ```

k-simons (Migrated from github.com) reviewed

2025-12-17 22:30:24 +00:00

workerpool/internal/computing_future.go

yeah that makes more sense; done

nmiyake (Migrated from github.com) requested changes

2025-12-18 20:05:55 +00:00

nmiyake (Migrated from github.com) left a comment

Thanks -- mostly looks good. Minor feedback and questions.

workerpool/options.go

					
				@ -0,0 +1,45 @@

				// Copyright (c) 2025 Palantir Technologies. All rights reserved.

// Each worker is a single goroutine

```suggestion // Each worker is a single goroutine ```

workerpool/options.go

					
				@ -0,0 +30,4 @@

						if maxNumberOfThreads < 1 {

							return c

						}

						c.maxNumberOfWorkers = &maxNumberOfThreads

What is the behavior if the provided value is negative? Should probably either:

Interpret anything 0 or less as "unlimited/no limit"
If a value 0 or less is provided, set to 1

Should also document the behavior as well.

What is the behavior if the provided value is negative? Should probably either: * Interpret anything 0 or less as "unlimited/no limit" * If a value 0 or less is provided, set to 1 Should also document the behavior as well.

workerpool/options.go

					
				@ -0,0 +39,4 @@

				// Metrics will only be emitted for the pool if at least 1 tag is supplied

				func WithMetricTags(tags []metrics.Tag) Option {

					return func(c *Config) *Config {

						c.tags = tags

Is it fine that we're using the provided slice directly rather than cloning? Current approach is not necessarily incorrect, but does mean that callers should generally not mutate the slice they provide here unless they want to modify behavior of the worker pool

workerpool/supplier_workerpool.go

					
				@ -0,0 +1,171 @@

				// Copyright (c) 2025 Palantir Technologies. All rights reserved.

	if d.config.maxNumberOfWorkers != nil && *d.config.maxNumberOfWorkers <= mapSize {

Wouldn't <= be safer in case mapSize somehow grows beyond maxNumberOfWorkers? Don't see locking around this logic so seems like it might be possible

```suggestion if d.config.maxNumberOfWorkers != nil && *d.config.maxNumberOfWorkers <= mapSize { ``` Wouldn't `<=` be safer in case `mapSize` somehow grows beyond `maxNumberOfWorkers`? Don't see locking around this logic so seems like it might be possible

workerpool/supplier_workerpool.go

					
				@ -0,0 +106,4 @@

					go d.startWorker(ctx)

				}

				func (d *defaultSupplierWorkerPool[R]) runWorkerLoop(workerContext context.Context) {

}

func (d *defaultSupplierWorkerPool[R]) runWorkerLoop(workerContext context.Context) {

```suggestion } func (d *defaultSupplierWorkerPool[R]) runWorkerLoop(workerContext context.Context) { ```

k-simons (Migrated from github.com) reviewed

2025-12-18 20:26:26 +00:00

workerpool/options.go

					
				@ -0,0 +39,4 @@

				// Metrics will only be emitted for the pool if at least 1 tag is supplied

				func WithMetricTags(tags []metrics.Tag) Option {

					return func(c *Config) *Config {

						c.tags = tags

I think this is fine tbh

nmiyake (Migrated from github.com) requested changes

2025-12-18 20:43:26 +00:00

workerpool/options.go

					
				@ -0,0 +1,45 @@

				// Copyright (c) 2025 Palantir Technologies. All rights reserved.

func WithMaxNumberOfWorkers(maxNumberOfThreads int) Option {

Sorry I missed this the first time around -- can we use Workers since that's more accurate

```suggestion func WithMaxNumberOfWorkers(maxNumberOfThreads int) Option { ``` Sorry I missed this the first time around -- can we use `Workers` since that's more accurate

workerpool/options.go

					
				@ -0,0 +27,4 @@

				// If a value of less than 1 is provided, it is ignored

				func WithMaxNumberOfWorkers(maxNumberOfThreads int) Option {

					return func(c *Config) *Config {

						if maxNumberOfThreads < 1 {

This should be fine, but just noting that this means that once a max has been configured, it's not possible to use this option function to configure back to "unlimited"