Bounded In-Process Pipelines with Ruby's SizedQueue and Worker Threads

The Memory Problem with Unbounded Queues

When a Ruby producer outruns its consumer, an unbounded Queue turns into a memory leak. The useful takeaway is that SizedQueue caps the number of pending items and blocks the producer when that limit is reached, giving you backpressure without adding an external broker. Pair it with a fixed pool of worker threads and explicit error handling, and you get a bounded, in-process pipeline suitable for I/O-bound work.

Requirements: Bounded Memory and Visible Backpressure

A minimal pipeline must satisfy four constraints:

• Bound memory so a slow consumer cannot exhaust the heap.
• Decouple I/O-bound producers and consumers so neither blocks the other unnecessarily.
• Surface worker failures instead of letting threads die silently.
• Avoid external dependencies for simple, single-process workloads.

The Smallest Suitable Design: SizedQueue with a Fixed Worker Pool

Use SizedQueue.new(capacity) as the only coordination point. Spawn a fixed number of worker threads that loop on queue.pop. Producers call queue.push(item), which blocks when the queue is full. This is the smallest design that satisfies the requirements: one queue, a fixed thread count, and no shared locks beyond the queue itself.

require 'thread'

queue = SizedQueue.new(50)
workers = 4.times.map do |worker_id|
  Thread.new do
    loop do
      payload = queue.pop
      begin
        result = transform(payload)
        store(result)
      rescue StandardError => e
        warn "[worker-#{worker_id}] #{e.class}: #{e.message}"
      end
    end
  end
end

# Producer: blocks here if 50 items are already queued
items.each { |item| queue.push(item) }

The rescue sits inside the loop so a single bad payload does not kill the thread. Thread.abort_on_exception is process-global; if one worker hits an unexpected error, the entire application crashes. Explicit per-worker rescue blocks are safer for a supervised pool.

Trust and Data Boundaries: Shared Heap, Shared Mutable State

Ruby threads share a single heap. There is no memory isolation between workers. Any object passed through the queue is shared mutable state unless you make it otherwise. If a producer pushes a Hash and then mutates it, a worker may observe the change mid-processing. When ownership is unclear, freeze the payload before pushing, or deep-copy it with Marshal.load(Marshal.dump(payload)). Treat the queue as a message boundary: once an item is pushed, the producer should not touch it again.

Operational Checks: Queue Depth, Thread Liveness, and Worker Errors

Under production-like traffic, watch three signals:

• queue.size — the number of items waiting. If it sits at capacity, consumers are the bottleneck.
• Thread.list.size — total live threads. A slow upward trend means threads are leaking.
• Worker liveness — workers.all?(&:alive?) or equivalent health checks.

Run these from a Rails console, IRB session attached to the process, or a metrics endpoint. queue.size is a snapshot; reading it while workers mutate shared state is safe for the integer itself, but do not assume it reflects a consistent point-in-time view of the whole pipeline. If queue.size is pinned at maximum and RSS is climbing, the consumer is too slow or blocked.

Failure Modes: Memory, Silent Death, Deadlock, and the GVL

Unbounded memory growth

Queue.new has no limit. A burst of fast producers will allocate until the process is killed. SizedQueue is the direct fix.

Silent worker death

An exception raised outside a rescue inside the worker loop terminates that thread. The queue still accepts items, but throughput drops to zero without any obvious error. Always wrap the processing logic, not just the pop.

Deadlock from nested locks

If a producer holds a mutex and then blocks on queue.push because the queue is full, and a consumer needs that same mutex to process an item, the pipeline freezes. Never hold a lock across a blocking queue operation.

GVL ceiling

Ruby's Global VM Lock (GVL) allows only one thread to execute Ruby bytecode at a time. Threads improve concurrency for I/O-bound work because I/O releases the GVL, but they do not speed up pure CPU computation. A pool of eight workers processing CPU-bound data will not finish faster than one worker.

Conditions That Change the Design

Switch away from threads when:

• Tasks are CPU-bound. Use process-based parallelism (forked workers, the parallel gem) or move work to an external job system.
• Durability or cross-machine distribution is required. An in-memory queue loses everything on restart. A Redis-backed system such as Sidekiq provides persistence, retries, and multi-process consumers.
• You need memory isolation between tasks. Threads share a heap; a memory leak or segfault in one worker affects the whole process.

Verifying the Design

To confirm the pattern works in your application:

1. Load test with a fast producer and a deliberately slow consumer (for example, a sleep(0.1) inside the worker). Verify that queue.size caps at the configured capacity and that resident memory stays flat rather than growing with the producer rate.
2. Inject an exception inside a worker and confirm the rescue block logs the error and the thread remains alive.
3. Profile Thread.list.size and queue.size under sustained traffic to validate pool sizing and detect leaks.
4. For CPU-bound work, compare wall-clock time across different thread counts. If adding threads does not reduce runtime, the GVL is the limiting factor and the design needs processes or external workers.

These checks are conservative: they validate backpressure, error handling, and the GVL boundary without requiring a specific framework or deployment target.