Rafael Romão's Coding Labs

Lab L01 · Java + Go

Implementing a bounded blocking queue

Build the hand-off buffer that sits between threads producing work and threads consuming it. The queue holds at most capacity items: put blocks while it is full, take blocks while it is empty, and close() stops new puts while letting consumers finish the items already accepted before they see “end of stream”.

The task

An ingestion service receives GPS pings on network threads and writes them to a database on writer threads. If the writers slow down, an unbounded buffer grows until the process runs out of memory. A bounded queue applies backpressure instead: when it is full, the network threads block and intake slows to the speed of the writers. At shutdown you want to stop accepting new pings but still write every ping already buffered. That is the whole component: a fixed-size FIFO with two blocking operations and a clean shutdown.

producer 1producer 2put()capacity = 3banewest → oldesttake()consumer 1consumer 2wait() while fullwait() while emptyOPENput ✓ take ✓CLOSED, items leftput ✗ take → next itemCLOSED, drainedtake → Optional.empty()close()last take
Producers append on one side and consumers remove from the other; both wait on the same monitor. The lower strip is the lifecycle: after close() takes still succeed until the queue is empty.

What you implement

public final class BlockingQueue<T> {
    public BlockingQueue(int capacity);                  // capacity <= 0 -> IllegalArgumentException
    public void put(T value) throws InterruptedException;      // waits while full
    public Optional<T> take() throws InterruptedException;     // waits while empty
    public void close();                                       // stop accepting puts; wake all waiters
}

Rules

  • Items come out in the order they went in (FIFO). The queue never holds more than capacity items.
  • put(null) throws NullPointerException.
  • put on a full queue waits until a take frees a slot. If the queue is closed while it waits (or was already closed), it throws IllegalStateException and adds nothing.
  • take on an empty queue waits until a put or a close. It returns Optional.empty() only when the queue is closed and drained; until then a closed queue still hands out its remaining items.
  • close() is idempotent and must wake every blocked producer and consumer.
  • Interrupting a thread blocked in put or take makes that call throw InterruptedException.
  • Thread-safe: check the condition and change the contents under the same lock, and re-check the condition after every wake-up (another thread may have taken the slot first).

Worked example

  1. new BlockingQueue<>(2); put(a), put(b) → the queue is full.
  2. A third thread calls put(c) → it blocks.
  3. take() → a; the blocked put(c) wakes up and appends c.
  4. close(). Now take() → b, take() → c, take() → Optional.empty() (and keeps returning empty).
  5. put(d) after close → IllegalStateException.

Files and commands

Read the tests first, then replace each TODO in the starter until they pass. Java starter: BlockingQueue.java · tests: BlockingQueueTest.java. Go starter: queue.go · tests: queue_test.go.

cd java
./gradlew test --tests 'labs.blockingqueue.*'

cd go
go test -race -count=1 ./blockingqueue
Reveal the reference solutions

Java: BlockingQueue.java
Go: queue.go

When to use

Use this lab when producers and consumers need a bounded first-in, first-out buffer. The queue holds at most its positive capacity in entries. Producers wait while full; consumers wait while empty. Closing rejects future puts, wakes waiters, and lets consumers drain accepted entries before reporting closure. Close is idempotent. This bounds queued entries, not the number of waiting callers or the size of stored values.

The examples live under solutions/. Java rejects null values and uses interruption to cancel a wait. Go permits zero and nil values where the type allows them, and uses a context for cancellation. In Go, cancellation racing with an operation becoming possible may result in either outcome; callers must handle both. Neither implementation guarantees fairness among waiters.

Steps

  1. Write the state machine. The queue is open or closed; its FIFO contents remain available after closure. A put linearizes when its value is appended; a take linearizes when its value is removed. Closure linearizes when the closed flag changes under the lock.
  2. Preserve the capacity invariant. Check the predicate and mutate contents under the same lock. Recheck after waking because another caller may have taken the available slot first.
  3. Wake the right population. Java uses a monitor and notifyAll. Go closes a generation-specific notification channel under the mutex and creates the next one; each waiter captures the current channel before unlocking. The shared lock prevents missing a transition between checking and subscribing.
  4. Define termination. A blocked producer must return after close. A consumer drains accepted items and then observes closure. Java signals closure with an empty result; Go returns a distinct closed error. Interrupted Java waits throw; Go returns the context error.
  5. Exercise boundaries. Use capacity one to expose the full and empty cases, test FIFO order and close-and-drain, and test invalid capacity. Coordinate actual blocked waiters when testing wakeups; a task started concurrently with close may instead observe the already-closed state and does not prove it was woken.
  6. Examine resource costs. The Go example uses a slice and may retain backing-array capacity temporarily; removed element slots are cleared. The Java example uses an array deque. Waking every waiter keeps the implementation understandable but can cause contention under high fan-in. Neither queue is a persistent message broker.

Checklist

Recall and variation

Rebuild close-and-drain from memory using Go and Java only after your attempt. Variation: cancel a producer waiting on a full queue. Acceptance: the accepted item remains available and the cancelled put adds nothing.

  • Capacity is positive and the entry count never exceeds it
  • Waiters check predicates again after waking
  • Closure wakes producers and consumers and preserves accepted FIFO items
  • Cancellation and closure races are described, not hidden
  • Waiting callers and value sizes have separate resource policies

Signals

Working: accepted items are removed in FIFO order; close prevents further puts while allowing drain; callers do not depend on fairness.

Going wrong: a waiter remains blocked after close, a notification is missed between checking and waiting, or a cancellation result is mistaken for proof that no concurrent operation succeeded.

Related