improvement: Add internal queue package and external set package #11

2025-12-16T19:57:57Z

k-simons commented

2025-12-16 19:57:57 +00:00

(Migrated from github.com)

Adds a new internal queue package with thread-safe generic work queues for producer-consumer patterns
Adds a util/set package providing a generic set implementation
Queue[T any] - simple FIFO work queue that accepts any type
CollapsingQueue[T comparable] - deduplicating work queue that collapses duplicate items added while processing

- Adds a new internal queue package with thread-safe generic work queues for producer-consumer patterns - Adds a util/set package providing a generic set implementation - Queue[T any] - simple FIFO work queue that accepts any type - CollapsingQueue[T comparable] - deduplicating work queue that collapses duplicate items added while processing

changelog-app[bot] commented

2025-12-16 19:58:01 +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/add queue

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/add queue </changelog-description><changelog-links></changelog-links> **Check the box to generate changelog(s)** - [ ] Generate changelog entry

changelog-app[bot] commented

2025-12-16 19:58:34 +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

Add internal queue package and external set package (#11)

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

nmiyake (Migrated from github.com) requested changes

2025-12-17 00:14:50 +00:00

nmiyake (Migrated from github.com) left a comment

Thanks -- mostly done with review, but ran out of time. However, wanted to submit what I have so far, especially since the queue comparable type issue is a major one.

internal/queue/collapsing_queue.go

					
				@ -0,0 +1,182 @@

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

Comment is on wrong function (on GetWithCallback instead of Get).

Not sure if you want to duplicate the documentation from interface functions or not.

Comment is on wrong function (on `GetWithCallback` instead of `Get`). Not sure if you want to duplicate the documentation from interface functions or not.

type collapsingQueue[T comparable] struct {

```suggestion type collapsingQueue[T comparable] struct { ```

internal/queue/queue.go

					
				@ -0,0 +1,114 @@

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

	// Get returns the next item in the queue. If the queue is empty, Get blocks until an item is
	// available and then returns it. If the shutdown return value is true, it indicates that the queue
	// has been shut down -- in this case, the returned item will be the zero value of the type T.
	// Once a queue has been shut down, it will never return any more elements, so the caller should
	// exit their processing loop.
	Get() (item T, shutdown bool)

```suggestion // Get returns the next item in the queue. If the queue is empty, Get blocks until an item is // available and then returns it. If the shutdown return value is true, it indicates that the queue // has been shut down -- in this case, the returned item will be the zero value of the type T. // Once a queue has been shut down, it will never return any more elements, so the caller should // exit their processing loop. Get() (item T, shutdown bool) ```

This doesn't work -- you're basically tricking the compiler, but ultimately queueItem is the type used for CollapsingQueue, and if the type stored in item is not Comparable, then this will panic at runtime.

You can observe this by adding the following code to the TestQueue_AddAndGet test:

	type MyFunc func() string
	q2 := NewQueue[MyFunc]()
	q2.Add(func() string { return "hello, world" })
	q2.Add(func() string { return "goodbye, world" })
	got, shutdown := q2.Get()
	fmt.Println(got(), shutdown)
	got, shutdown = q2.Get()
	fmt.Println(got(), shutdown)

Results in:

=== RUN   TestQueue_AddAndGet
--- FAIL: TestQueue_AddAndGet (0.00s)
panic: hash of unhashable type: queue.MyFunc [recovered, repanicked]

goroutine 8 [running]:
testing.tRunner.func1.2({0x102da3de0, 0x102d82540})
	/usr/local/go-1.25.5/src/testing/testing.go:1872 +0x3a4
testing.tRunner.func1()
	/usr/local/go-1.25.5/src/testing/testing.go:1875 +0x4a8
panic({0x102da3de0?, 0x102d82540?})
	/usr/local/go-1.25.5/src/runtime/panic.go:783 +0xec
github.com/palantir/witchcraft-go-tasks/util/set.Set[...].Has(0x102dd5280, {0x1, {0x102d82540, 0x102dcef88}})
	/Volumes/git/go/src/github.com/palantir/witchcraft-go-tasks/util/set/set.go:28 +0x48
github.com/palantir/witchcraft-go-tasks/internal/queue.(*collapsingQueue[...]).Add(0x102dd4f80, {0x1, {0x102d82540, 0x102dcef88}})
	/Volumes/git/go/src/github.com/palantir/witchcraft-go-tasks/internal/queue/collapsing_queue.go:90 +0x10c
github.com/palantir/witchcraft-go-tasks/internal/queue.(*queue[...]).Add(0x102dd3fc0, 0x102dcef88)
	/Volumes/git/go/src/github.com/palantir/witchcraft-go-tasks/internal/queue/queue.go:81 +0x88
github.com/palantir/witchcraft-go-tasks/internal/queue.TestQueue_AddAndGet(0x14000003dc0)
	/Volumes/git/go/src/github.com/palantir/witchcraft-go-tasks/internal/queue/queue_test.go:42 +0x2b0
testing.tRunner(0x14000003dc0, 0x102dcee90)
	/usr/local/go-1.25.5/src/testing/testing.go:1934 +0x140
created by testing.(*T).Run in goroutine 1
	/usr/local/go-1.25.5/src/testing/testing.go:1997 +0x74c

This doesn't work -- you're basically tricking the compiler, but ultimately `queueItem` is the type used for `CollapsingQueue`, and if the type stored in `item` is not `Comparable`, then this will panic at runtime. You can observe this by adding the following code to the `TestQueue_AddAndGet` test: ``` type MyFunc func() string q2 := NewQueue[MyFunc]() q2.Add(func() string { return "hello, world" }) q2.Add(func() string { return "goodbye, world" }) got, shutdown := q2.Get() fmt.Println(got(), shutdown) got, shutdown = q2.Get() fmt.Println(got(), shutdown) ``` Results in: ``` === RUN TestQueue_AddAndGet --- FAIL: TestQueue_AddAndGet (0.00s) panic: hash of unhashable type: queue.MyFunc [recovered, repanicked] goroutine 8 [running]: testing.tRunner.func1.2({0x102da3de0, 0x102d82540}) /usr/local/go-1.25.5/src/testing/testing.go:1872 +0x3a4 testing.tRunner.func1() /usr/local/go-1.25.5/src/testing/testing.go:1875 +0x4a8 panic({0x102da3de0?, 0x102d82540?}) /usr/local/go-1.25.5/src/runtime/panic.go:783 +0xec github.com/palantir/witchcraft-go-tasks/util/set.Set[...].Has(0x102dd5280, {0x1, {0x102d82540, 0x102dcef88}}) /Volumes/git/go/src/github.com/palantir/witchcraft-go-tasks/util/set/set.go:28 +0x48 github.com/palantir/witchcraft-go-tasks/internal/queue.(*collapsingQueue[...]).Add(0x102dd4f80, {0x1, {0x102d82540, 0x102dcef88}}) /Volumes/git/go/src/github.com/palantir/witchcraft-go-tasks/internal/queue/collapsing_queue.go:90 +0x10c github.com/palantir/witchcraft-go-tasks/internal/queue.(*queue[...]).Add(0x102dd3fc0, 0x102dcef88) /Volumes/git/go/src/github.com/palantir/witchcraft-go-tasks/internal/queue/queue.go:81 +0x88 github.com/palantir/witchcraft-go-tasks/internal/queue.TestQueue_AddAndGet(0x14000003dc0) /Volumes/git/go/src/github.com/palantir/witchcraft-go-tasks/internal/queue/queue_test.go:42 +0x2b0 testing.tRunner(0x14000003dc0, 0x102dcee90) /usr/local/go-1.25.5/src/testing/testing.go:1934 +0x140 created by testing.(*T).Run in goroutine 1 /usr/local/go-1.25.5/src/testing/testing.go:1997 +0x74c ```

Can reply in more detail later, but here are a few ways I can think of to approach this:

Restrict Queue type to comparable (not great because it's more restrictive)
Update CollapsingQueue so that it basically has a Key type and Value type -- if you use a map instead of Set for internal storage, then without too much overhead/change you should be able to do all the enqueue/dequeue logic on keys, but associate with value. However, will make API a bit heavier.
I think there's a way we could use type constraints to effectively say CollapsingQueue takes either a comparable OR an interface that implements Key() -- if this is done effectively and decomped right, I think it gives the best of both worlds, where you can create queues with comparable types directly, or allow objects to be queued as long as they have a Key() function that returns a comparable to use as the queue key
Similar to above, but just require all items that are enqueued to be of type/interface QueueItem[T comparable] with function Key() T and provide a convenience function/constructor that accepts any comparable type directly without having to have clients define their own types

I think 4 might be the most straightforward.

For 3, started playing around with the following, but didn't have time to finish locally:

type keyType[T comparable] interface {
	comparable
	Key() T
}

// NewCollapsingQueue constructs a new deduplicating work queue.
func newCollapsingQueue[K comparable, T keyType[K]]() CollapsingQueue[T] {

Can reply in more detail later, but here are a few ways I can think of to approach this: 1. Restrict Queue type to `comparable` (not great because it's more restrictive) 2. Update `CollapsingQueue` so that it basically has a `Key` type and `Value` type -- if you use a `map` instead of `Set` for internal storage, then without too much overhead/change you should be able to do all the enqueue/dequeue logic on keys, but associate with value. However, will make API a bit heavier. 3. I think there's a way we could use type constraints to effectively say `CollapsingQueue` takes either a `comparable` OR an interface that implements `Key()` -- if this is done effectively and decomped right, I think it gives the best of both worlds, where you can create queues with comparable types directly, or allow objects to be queued as long as they have a `Key()` function that returns a comparable to use as the queue key 4. Similar to above, but just require all items that are enqueued to be of type/interface `QueueItem[T comparable]` with function `Key() T` and provide a convenience function/constructor that accepts any `comparable` type directly without having to have clients define their own types I think 4 might be the most straightforward. For 3, started playing around with the following, but didn't have time to finish locally: ``` type keyType[T comparable] interface { comparable Key() T } // NewCollapsingQueue constructs a new deduplicating work queue. func newCollapsingQueue[K comparable, T keyType[K]]() CollapsingQueue[T] { ```

util/set/set.go

Would recommend adding an Iterator function (although not firm on naming -- can be All() or Elements() or something like that instead), since that's standard in Go now and will allow for easier usage with packages like slices.

// Iterator returns an iterator that returns the elements in the set.
// The order of the returned elements is random.
func (s Set[T]) Iterator() iter.Seq[T] {
	return func(yield func(T) bool) {
		for k := range s {
			if !yield(k) {
				return
			}
		}
	}
}

Would recommend adding an `Iterator` function (although not firm on naming -- can be `All()` or `Elements()` or something like that instead), since that's standard in Go now and will allow for easier usage with packages like `slices`. ```suggestion // Iterator returns an iterator that returns the elements in the set. // The order of the returned elements is random. func (s Set[T]) Iterator() iter.Seq[T] { return func(yield func(T) bool) { for k := range s { if !yield(k) { return } } } } ```

Fine to keep this, but if we add the Iterator function above, not sure if it's needed -- we can range over elements using for elem := range s.Iterator(), and if we actually do want a slice, we can still do it as a one-liner using slices.Collect(s.Iterator()).

Fine to keep this, but if we add the `Iterator` function above, not sure if it's needed -- we can range over elements using `for elem := range s.Iterator()`, and if we actually do want a slice, we can still do it as a one-liner using `slices.Collect(s.Iterator())`.

nmiyake (Migrated from github.com) requested changes

2025-12-17 19:10:23 +00:00

nmiyake (Migrated from github.com) left a comment

Thanks! Mostly looks good -- just a functional question about whether it is intended that callback function is called even when shutdown=true and no item is dequeued, and a few suggestions for documentation.

Thanks! Mostly looks good -- just a functional question about whether it is intended that callback function is called even when `shutdown=true` and no item is dequeued, and a few suggestions for documentation.

internal/queue/collapsing_queue.go

					
				@ -0,0 +1,182 @@

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

	// Done marks the item as finished processing, and must be called once for each
	// item returned by Get or GetWithCallback. If the item was re-added to the
	// queue while it was being processed, it will be re-queued for another processing
	// cycle.

```suggestion // Done marks the item as finished processing, and must be called once for each // item returned by Get or GetWithCallback. If the item was re-added to the // queue while it was being processed, it will be re-queued for another processing // cycle. ```

As structured, callback() will be called even if the queue is shutting down and this function returns *new(T), true -- is that intentional/desired?

As structured, `callback()` will be called even if the queue is shutting down and this function returns `*new(T), true` -- is that intentional/desired?

internal/queue/queue.go

					
				@ -0,0 +1,114 @@

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

	// Len returns the current number of items waiting to be processed in the queue.
	// This is informational only and should not be used for synchronization decisions.

```suggestion // Len returns the current number of items waiting to be processed in the queue. // This is informational only and should not be used for synchronization decisions. ```

// item wraps a value. Under the hood, the queue stores *item[T] elements.
// Because pointers are always comparable (by address), *item[T] satisfies comparable
// even when T doesn't. Each Add creates a new pointer, ensuring uniqueness
// (which also prevents collapsing elements).

```suggestion // item wraps a value. Under the hood, the queue stores *item[T] elements. // Because pointers are always comparable (by address), *item[T] satisfies comparable // even when T doesn't. Each Add creates a new pointer, ensuring uniqueness // (which also prevents collapsing elements). ```

	// Get blocks until an item is available and returns it. If shutdown is true,
	// the queue has been shut down: in this case, the returned item is the zero
	// value of T and should be ignored, and the caller should exit their processing loop.

```suggestion // Get blocks until an item is available and returns it. If shutdown is true, // the queue has been shut down: in this case, the returned item is the zero // value of T and should be ignored, and the caller should exit their processing loop. ```

	// ShutDownWithDrain initiates shutdown and blocks until all queued items have
	// been retrieved via Get(). Once this function is called, new items cannot be added
	// (Add() becomes a noop). Any elements in the queue that
	// were present before it started shutting down will still be returned
	// by Get() calls with shutdown=false. Once the queue is empty,
	// Get() calls will return with shutdown=true. If Shutdown() is called, it will interrupt
	// any ShutDownWithDrain calls and cause them to return immediately.

It took me a while to understand the differences between the functions even when looking at implementation, so tried to update documentation to be a bit clearer. However, feel free to modify as needed (or if anything here is incorrect).

```suggestion // ShutDownWithDrain initiates shutdown and blocks until all queued items have // been retrieved via Get(). Once this function is called, new items cannot be added // (Add() becomes a noop). Any elements in the queue that // were present before it started shutting down will still be returned // by Get() calls with shutdown=false. Once the queue is empty, // Get() calls will return with shutdown=true. If Shutdown() is called, it will interrupt // any ShutDownWithDrain calls and cause them to return immediately. ``` It took me a while to understand the differences between the functions even when looking at implementation, so tried to update documentation to be a bit clearer. However, feel free to modify as needed (or if anything here is incorrect).

	// ShutDown initiates shutdown and returns (does not block).
	// Once this function is called, new items cannot be added
	// (Add() becomes a noop). Any elements in the queue that
	// were present before it started shutting down will still be returned
	// by Get() calls with shutdown=false. Once the queue is empty,
	// Get() calls will return with shutdown=true.

```suggestion // ShutDown initiates shutdown and returns (does not block). // Once this function is called, new items cannot be added // (Add() becomes a noop). Any elements in the queue that // were present before it started shutting down will still be returned // by Get() calls with shutdown=false. Once the queue is empty, // Get() calls will return with shutdown=true. ```

internal/queue/queue.go

					
				@ -0,0 +21,4 @@

				// by a corresponding Get(). This is suitable for task queues where every submitted

				// item must be processed exactly once.

				//

				// The type parameter T can be any type since no comparison is needed.

// Queue is a thread-safe generic work queue that provides blocking Get operations
// and graceful shutdown capabilities. Items are processed in FIFO order.
//
// Unlike CollapsingQueue, this queue does not deduplicate items - every call to Add() adds an element that can be retrieved
// by a corresponding Get(). This is suitable for task queues where every submitted
// item must be processed exactly once.
//
// The type parameter T can be any type since no comparison is needed.

```suggestion // Queue is a thread-safe generic work queue that provides blocking Get operations // and graceful shutdown capabilities. Items are processed in FIFO order. // // Unlike CollapsingQueue, this queue does not deduplicate items - every call to Add() adds an element that can be retrieved // by a corresponding Get(). This is suitable for task queues where every submitted // item must be processed exactly once. // // The type parameter T can be any type since no comparison is needed. ```

internal/queue/queue.go

					
				@ -0,0 +35,4 @@

					Get() (item T, shutdown bool)

					// GetWithCallback is like Get but invokes the callback while holding the queue

					// lock, before returning. This can be used to perform atomic operations when

					// an item is dequeued.

	// GetWithCallback is like Get but invokes the callback while holding the queue
	// lock, before returning. This can be used to perform atomic operations when
	// an item is dequeued. The callback is always invoked, including when
	// GetWithCallback returns shutdown=true.

Per other comment, not sure if this behavior is intended, but if it is then I think it should be documented since it's not obvious.

If this is not intended, we can probably update documentation to state that callback will only be invoked if function actually dequeues an element and returns with shutdown=false.

```suggestion // GetWithCallback is like Get but invokes the callback while holding the queue // lock, before returning. This can be used to perform atomic operations when // an item is dequeued. The callback is always invoked, including when // GetWithCallback returns shutdown=true. ``` Per other comment, not sure if this behavior is intended, but if it is then I think it should be documented since it's not obvious. If this is not intended, we can probably update documentation to state that callback will only be invoked if function actually dequeues an element and returns with shutdown=false.

internal/queue/queue_type_wrapper.go

					
				@ -0,0 +1,34 @@

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

// queueWrapper is a named type for a slice that implements queue functions.

Fixed up comment to reflect type name and clarify (in the context of this package, Queue is a job queue interface, and this type doesn't implement that)

```suggestion // queueWrapper is a named type for a slice that implements queue functions. ``` Fixed up comment to reflect type name and clarify (in the context of this package, `Queue` is a job queue interface, and this type doesn't implement that)

	// manually zero out the old element to allow it to be garbage collected

```suggestion // manually zero out the old element to allow it to be garbage collected ```

nmiyake (Migrated from github.com) reviewed

2025-12-17 19:12:32 +00:00

internal/queue/queue_test.go

					
				@ -0,0 +47,4 @@

					got, shutdown = q2.Get()

					assert.Equal(t, got(), "goodbye, world")

					assert.False(t, shutdown)

				}

Can we add logic to test the "Get" portion as well per name of test?

	got, shutdown := q2.Get()
	assert.Equal(t, got(), "hello, world")
	assert.False(t, shutdown)
	got, shutdown = q2.Get()
	assert.Equal(t, got(), "goodbye, world")
	assert.False(t, shutdown)
}

Can we add logic to test the "Get" portion as well per name of test? ```suggestion got, shutdown := q2.Get() assert.Equal(t, got(), "hello, world") assert.False(t, shutdown) got, shutdown = q2.Get() assert.Equal(t, got(), "goodbye, world") assert.False(t, shutdown) } ```

k-simons (Migrated from github.com) reviewed

2025-12-17 19:51:32 +00:00

internal/queue/queue.go

					
				@ -0,0 +35,4 @@

					Get() (item T, shutdown bool)

					// GetWithCallback is like Get but invokes the callback while holding the queue

					// lock, before returning. This can be used to perform atomic operations when

					// an item is dequeued.

Changed the comment to reflect that we no-long invoke the callback

k-simons (Migrated from github.com) reviewed

2025-12-17 19:52:15 +00:00

internal/queue/collapsing_queue.go

					
				@ -0,0 +1,182 @@

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

Nope, modified to not invoke if its shutting down

nmiyake (Migrated from github.com) approved these changes

2025-12-17 19:56:24 +00:00

autorelease3[bot] commented

2025-12-17 19:56:38 +00:00

(Migrated from github.com)

Released v0.2.0

Released [v0.2.0](https://github.com/palantir/witchcraft-go-tasks/releases/tag/v0.2.0)

Sign in to join this conversation.

No reviewers

nmiyake

No labels

No milestone

No project

No assignees

1 participant

Notifications

Due date

The due date is invalid or out of range. Please use the format "yyyy-mm-dd".

No due date set.

Dependencies

No dependencies set

Reference

golibs/go-tasks!11

				`@ -0,0 +1,182 @@`
				`// Copyright (c) 2025 Palantir Technologies. All rights reserved.`

				`@ -0,0 +1,114 @@`
				`// Copyright (c) 2025 Palantir Technologies. All rights reserved.`

				`@ -0,0 +1,34 @@`
				`// Copyright (c) 2025 Palantir Technologies. All rights reserved.`

Rows
Columns

improvement: Add internal queue package and external set package #11

Generate changelog in changelog/@unreleased

✅ Successfully generated changelog entry!

Entry generated via PR title

📋Changelog Preview

💡 Improvements

Generate changelog in `changelog/@unreleased`