Opensource internals

Two Sets and a FIFO: The Whole Concurrency Model of the Kubernetes Workqueue

13 min read Jay Hubbert

client-go's workqueue is three fields - a FIFO queue, a dirty set and a processing set - and that is enough to give coalescing, mutual exclusion and fair draining.

TL;DR: The entire concurrency model of client-go’s workqueue is three fields: a FIFO queue, a dirty set, and a processing set. Add marks an item dirty and pushes it only if it is not currently being processed; Get moves it from dirty to processing; Done removes it from processing and, if it was re-added while in flight, pushes it back. The result: one item is never processed by two workers at once, and a thousand rapid Adds of the same key collapse into one or two processing runs.

Every controller in Kubernetes ends up with the same problem. Something happens in the cluster — a Pod is created, a Node goes away — and the controller has to react. But “react” means “call the API server”, and the API server has a limited capacity to be called. Worse, the same event tends to arrive in bursts: a Node’s status update touches fifty Pods at once, and a reconcile loop that re-adds the same key on every error will re-add it forever if nobody stops it.

The workqueue is the small piece of machinery that sits between “an event happened” and “the controller does something about it”. It has three properties, and the package doc states them without elaboration:

go 
// staging/src/k8s.io/client-go/util/workqueue/doc.go:19
//   - Fair: items processed in the order in which they are added.
//   - Stingy: a single item will not be processed multiple times concurrently,
//     and if an item is added multiple times before it can be processed, it
//     will only be processed once.
//   - Multiple consumers and producers. In particular, it is allowed for an
//     item to be reenqueued while it is being processed.
//     In this case it will be processed again.
//   - Shutdown notifications.

“Fair”, “stingy”, and “reenqueueable”. Everything else in the package is bookkeeping for those three.

The three fields

go 
// staging/src/k8s.io/client-go/util/workqueue/queue.go:194
	// queue defines the order in which we will work on items. Every
	// element of queue should be in the dirty set and not in the
	// processing set.
	queue Queue[t]

	// dirty defines all of the items that need to be processed.
	dirty sets.Set[t]

	// Things that are currently being processed are in the processing set.
	// These things may be simultaneously in the dirty set. When we finish
	// processing something and remove it from this set, we'll check if
	// it's in the dirty set, and if so, add it to the queue.
	processing sets.Set[t]

The invariant, stated in the comment on queue: every element of the FIFO is in the dirty set and not in the processing set. A dirty item means “needs to be processed”. A processing item means “a worker has claimed it”. An item can be in both, which is precisely the state that says “it was re-added while a worker was busy with it”.

dirty and processing are sets.Set[T] from k8s.io/apimachinery/pkg/util/sets — plain Go maps with a Set interface. The FIFO is pluggable via the Queue interface, which is how you get a priority queue or a rate-limited variant.

Add: mark dirty, push once

go 
// staging/src/k8s.io/client-go/util/workqueue/queue.go:230
func (q *Typed[T]) Add(item T) {
	q.cond.L.Lock()
	defer q.cond.L.Unlock()
	if q.shuttingDown {
		return
	}
	if q.dirty.Has(item) {
		// the same item is added again before it is processed, call the Touch
		// function if the queue cares about it (for e.g, reset its priority)
		if !q.processing.Has(item) {
			q.queue.Touch(item)
		}
		return
	}

	q.metrics.add(item)

	q.dirty.Insert(item)
	if q.processing.Has(item) {
		return
	}

	q.queue.Push(item)
	q.cond.Signal()
}

Three branches, and each one is a distinct case:

  1. Already dirty and not in flight. Return without pushing. This is the coalescing: the item is already sitting in the FIFO exactly once, and adding it again would be a duplicate. The Touch call gives priority-queue implementations a chance to boost the item, but only when it is not currently being processed — you cannot reprioritise something a worker is already holding.
  2. Already dirty and in flight. Also return without pushing — but the item stays dirty, and it was already removed from the FIFO when the worker claimed it. This is the “re-enqueued while being processed” case.
  3. Not dirty. Insert into dirty, and push to the FIFO — unless the item is currently being processed, in which case do not push.

Case 3’s exception is the important one. If a worker is holding item X and someone calls Add(X), the item must not go into the FIFO, because then two workers could pick up X — violating the “stingy” guarantee. Instead, X is left marked dirty, and the re-add is honoured later, by Done.

Get: dirty to processing

go 
// staging/src/k8s.io/client-go/util/workqueue/queue.go:268
func (q *Typed[T]) Get() (item T, shutdown bool) {
	q.cond.L.Lock()
	defer q.cond.L.Unlock()
	for q.queue.Len() == 0 && !q.shuttingDown {
		q.cond.Wait()
	}
	if q.queue.Len() == 0 {
		// We must be shutting down.
		return *new(T), true
	}

	item = q.queue.Pop()

	q.metrics.get(item)

	q.processing.Insert(item)
	q.dirty.Delete(item)

	return item, false
}

Get pops from the FIFO and immediately flips the item’s membership: out of dirty, into processing. That single pair of set operations is the mutual exclusion. Because the item is no longer dirty, a subsequent Add will take the “not dirty” branch, reach the q.processing.Has(item) check, and decline to push. So the item cannot be handed to a second worker. The sync.Cond handles the blocking: Get waits while the queue is empty and not shutting down, Add signals, and ShutDown broadcasts to wake every waiter.

Done: the deferred re-add.

go 
// staging/src/k8s.io/client-go/util/workqueue/queue.go:292
func (q *Typed[T]) Done(item T) {
	q.cond.L.Lock()
	defer q.cond.L.Unlock()

	q.metrics.done(item)

	q.processing.Delete(item)
	if q.dirty.Has(item) {
		q.queue.Push(item)
		q.cond.Signal()
	} else if q.processing.Len() == 0 {
		q.cond.Signal()
	}
}

Done is where the promise is kept. If the item was re-added while the worker was busy, it is still in dirty, so it goes back into the FIFO and another worker picks it up. If it was not re-added, it leaves processing and the queue is now idle — the else if branch signals the condition so that a Get blocked on an empty queue wakes up.

The end-to-end behaviour: N rapid Adds of the same key collapse into one processing run if they all land before the worker calls Get, and into two if one of them lands while the worker is holding the item. Not N. That is the coalescing property, and it is why controllers can safely Add the same key from a dozen different code paths.

The slice queue and its GC trick

The default FIFO is a plain Go slice:

go 
// staging/src/k8s.io/client-go/util/workqueue/queue.go:62
func (q *queue[T]) Touch(item T) {}

func (q *queue[T]) Push(item T) {
	*q = append(*q, item)
}

func (q *queue[T]) Len() int {
	return len(*q)
}

func (q *queue[T]) Pop() (item T) {
	item = (*q)[0]

	// The underlying array still exists and reference this object, so the object will not be garbage collected.
	(*q)[0] = *new(T)
	*q = (*q)[1:]

	return item
}

The Pop implementation is the interesting bit. *q = (*q)[1:] slides the slice header forward, which means the popped slot is still inside the backing array — and the backing array still holds a reference to the object. The object cannot be collected, even though it has been popped.

The fix is one line: (*q)[0] = *new(T) writes the zero value of T over the slot, dropping the reference so the GC can reclaim the object. It costs one store per pop and saves an entire object graph from being pinned for the lifetime of the queue’s backing array.

The same trick appears in the ring buffer behind every informer handler (vendor/k8s.io/utils/buffer/ring_growing.go:69).

AddAfter: one heap, one ticker, zero timers

The delaying queue is what you use when a retry must not be immediate — AddAfter(item, 30*time.Second) after a failed reconcile. The obvious implementation is one time.Timer per pending item. The actual implementation uses one heap and one heartbeat ticker:

go 
// staging/src/k8s.io/client-go/util/workqueue/delaying_queue.go:151
	ret := &delayingType[T]{
		TypedInterface:  q,
		clock:           clock,
		heartbeat:       clock.NewTicker(maxWait),
		stopCh:          make(chan struct{}),
		waitingForAddCh: make(chan *waitFor[T], 1000),
		metrics:         newRetryMetrics(name, provider),
	}

	go ret.waitingLoop(logger)
	return ret
}

One goroutine (waitingLoop) owns one heap. AddAfter does not touch the heap directly; it pushes a waitFor struct onto a buffered channel of capacity 1000 and lets the waiting loop own all the ordering. Two details matter: duration <= 0 short-circuits to a plain Add — no channel, no heap, no waiting loop involvement — and the send is guarded by select on q.stopCh, so a ShutDown racing with an AddAfter cannot wedge the caller:

go 
// staging/src/k8s.io/client-go/util/workqueue/delaying_queue.go:258
	q.metrics.retry()

	// immediately add things with no delay
	if duration <= 0 {
		q.Add(item)
		return
	}

	select {
	case <-q.stopCh:
		// unblock if ShutDown() is called
	case q.waitingForAddCh <- &waitFor[T]{data: item, readyAt: q.clock.Now().Add(duration)}:
	}
}

The heartbeat exists because the waiting loop is driven by events, and the heap head’s timer is the only thing that can wake it. The bound on that wait is ten seconds:

go 
// staging/src/k8s.io/client-go/util/workqueue/delaying_queue.go:276
const maxWait = 10 * time.Second

The waiting loop itself pops everything that is ready, then arms a single timer for the head of the heap:

go 
// staging/src/k8s.io/client-go/util/workqueue/delaying_queue.go:300
		// Add ready entries
		for waitingForQueue.Len() > 0 {
			entry := waitingForQueue.Peek().(*waitFor[T])
			if entry.readyAt.After(now) {
				break
			}

			entry = heap.Pop(waitingForQueue).(*waitFor[T])
			q.Add(entry.data)
			delete(waitingEntryByData, entry.data)
		}

		// Set up a wait for the first item's readyAt (if one exists)
		nextReadyAt := never
		if waitingForQueue.Len() > 0 {
			if nextReadyAtTimer != nil {
				nextReadyAtTimer.Stop()
			}
			entry := waitingForQueue.Peek().(*waitFor[T])
			nextReadyAtTimer = q.clock.NewTimer(entry.readyAt.Sub(now))
			nextReadyAt = nextReadyAtTimer.C()
		}

The select that follows has four arms: stopCh, the 10-second heartbeat.C(), the nextReadyAt timer for the heap head, and waitingForAddCh for new entries — with a nested drain loop that empties the channel of anything that arrived while it was being read.

The heartbeat.C() case is the insurance: every 10 seconds the loop wakes up and re-checks the heap head, even if the timer somehow did not fire. The comment says exactly why: “we know that we’ll never end up with an expired item sitting for more than 10 seconds”. The nextReadyAt timer is stopped and recreated each time the head changes, so at any moment there is exactly one timer, not one per item.

The heap is deduplicated by item:

go 
// staging/src/k8s.io/client-go/util/workqueue/delaying_queue.go:360
	existing, exists := knownEntries[entry.data]
	if exists {
		if existing.readyAt.After(entry.readyAt) {
			existing.readyAt = entry.readyAt
			heap.Fix(q, existing.index)
		}

		return
	}

	heap.Push(q, entry)
	knownEntries[entry.data] = entry
}

knownEntries maps item -> *waitFor, and the waitFor struct caches its own heap index, which is what makes heap.Fix an O(log n) reposition rather than a remove-and-reinsert. And the update rule is earlier wins: if the item is already scheduled for a later time, the new, earlier time replaces it, but if it was scheduled for an earlier time, the later request is ignored.

The rate limiter: max of two

The default controller rate limiter composes two independent limiters and takes the worst answer:

go 
// staging/src/k8s.io/client-go/util/workqueue/default_rate_limiters.go:48
// DefaultTypedControllerRateLimiter is a no-arg constructor for a default rate limiter for a workqueue.  It has
// both overall and per-item rate limiting.  The overall is a token bucket and the per-item is exponential
func DefaultTypedControllerRateLimiter[T comparable]() TypedRateLimiter[T] {
	return NewTypedMaxOfRateLimiter(
		NewTypedItemExponentialFailureRateLimiter[T](5*time.Millisecond, 1000*time.Second),
		// 10 qps, 100 bucket size.  This is only for retry speed and its only the overall factor (not per item)
		&TypedBucketRateLimiter[T]{Limiter: rate.NewLimiter(rate.Limit(10), 100)},
	)
}
  • Per-item exponential: 5ms base, doubling per failure of that item, capped at 1000 seconds.
  • Overall token bucket: 10 QPS sustained, burst of 100, shared by all items.

The MaxOf composition is what makes them complementary rather than redundant. The exponential limiter stops one pathological key from hot-looping; the token bucket stops a thousand different keys from collectively overwhelming the apiserver. MaxOf is not a sum and not a min — it takes the larger delay, so whichever constraint is tighter wins:

go 
// staging/src/k8s.io/client-go/util/workqueue/default_rate_limiters.go:224
	ret := time.Duration(0)
	for _, limiter := range r.limiters {
		curr := limiter.When(item)
		if curr > ret {
			ret = curr
		}
	}

	return ret
}

The doc comment is honest about the consequence: “When used with a token bucket limiter, the burst could be apparently exceeded in cases where particular items were separately delayed a longer time.”

The exponential limiter itself:

go 
// staging/src/k8s.io/client-go/util/workqueue/default_rate_limiters.go:120
	exp := r.failures[item]
	r.failures[item] = r.failures[item] + 1

	// The backoff is capped such that 'calculated' value never overflows.
	backoff := float64(r.baseDelay.Nanoseconds()) * math.Pow(2, float64(exp))
	if backoff > math.MaxInt64 {
		return r.maxDelay
	}

	calculated := time.Duration(backoff)
	if calculated > r.maxDelay {
		return r.maxDelay
	}

	return calculated
}

Note the overflow guard: math.Pow(2, exp) grows without bound, and casting a huge float to time.Duration is undefined behaviour in Go, so the math.MaxInt64 check comes first.

Forget is the other half, and it is the caller’s job: it deletes the failure count for that item out of the limiter’s map. If a controller never calls Forget on success, a key that failed once and then succeeded for a week still carries its first failure count, and its next failure jumps straight to a long delay. The standard controller pattern is defer queue.Forget(key) on the success path and queue.AddRateLimited(key) on the failure path.

The trade-off

What this design buys: per-item mutual exclusion for free, automatic coalescing of duplicate work, fairness (FIFO), and a retry mechanism that needs one goroutine and one heap rather than one timer per item.

What it costs:

  • The FIFO is not a priority queue. Items are processed in the order they were added. If a critical object and a batch of unimportant ones are queued together, the critical one waits.
  • Add is O(1) but Done is not free. Done does a set lookup and possibly a Push, all under the same mutex as Add and Get. The queue is a contention point when many workers hammer it.
  • The rate limiter’s map grows. failures is keyed by item and only shrinks on Forget. A controller that requeues a new key on every failure without ever calling Forget leaks map entries — the delay is capped, but the memory is not.
  • The delaying queue’s channel is a bottleneck. waitingForAddCh has capacity 1000; beyond that, AddAfter blocks until the waiting loop drains it, and the waiting loop is a single goroutine.
  • Coalescing loses information. If you need to know that an event happened five times, the workqueue cannot tell you.

Wrapping up

The one thing to remember: the queue, the dirty set and the processing set are the whole model. Add marks dirty, Get moves dirty to processing, Done moves processing back to the queue if the item was re-added. Everything else — the heap, the token bucket, the heartbeat ticker — is an implementation detail layered on top of those three fields.

The part people get wrong: assuming that calling Add twice means the handler runs twice. It usually does not. If the key is already dirty and not in flight, the second Add is a no-op; if it is in flight, the re-add is honoured once, after Done. Controllers that need “one run per change” get that; controllers that need “one run per event” do not.

What to go do about it: call queue.Forget(key) on the success path of every processNextWorkItem, or the exponential limiter’s failure count never resets and your next transient error costs you a 1000-second delay. If AddAfter is your retry path, remember the steady-state delay is baseDelay * 2^failures capped at maxDelay — with the default DefaultTypedControllerRateLimiter that is 5ms doubling, and the overall 10 QPS / burst 100 token bucket is the constraint that actually protects the apiserver when many keys fail at once. Watch workqueue_retries_total and workqueue_queue_duration_seconds to see which side of max(exponential, token bucket) you are living on.

KubernetesInternalsClient go