module BoundedFanOut
Result = Struct.new(:value, :error) do
def ok?
error.nil?
end
end
def self.map(items, size: 4)
queue = Queue.new
items.each_with_index { |item, i| queue << [i, item] }
results = Array.new(items.size)
mutex = Mutex.new
workers = Array.new([size, items.size].min) do
Thread.new do
loop do
index, item = queue.pop(true)
value = yield(item)
mutex.synchronize { results[index] = Result.new(value, nil) }
rescue ThreadError
break # queue empty
rescue => e
mutex.synchronize { results[index] = Result.new(nil, e) }
end
end
end
workers.each(&:join)
results
end
end
require "timeout"
class EnrichmentService
CALL_TIMEOUT = 3 # seconds
def initialize(user_id, client: ApiClient.new)
@user_id = user_id
@client = client
end
def call
tasks = {
profile: -> { @client.get("/users/#{@user_id}") },
credit: -> { @client.get("/scores/#{@user_id}") },
orders: -> { @client.get("/users/#{@user_id}/orders?limit=5") }
}
keys = tasks.keys
results = BoundedFanOut.map(tasks.values, size: 3) do |task|
Timeout.timeout(CALL_TIMEOUT) { task.call }
end
keys.zip(results).each_with_object({}) do |(key, result), acc|
if result.ok?
acc[key] = result.value
else
Rails.logger.warn("enrichment #{key} failed: #{result.error.class}")
acc[key] = nil
end
end
end
end
class EnrichmentController < ApplicationController
def show
data = EnrichmentService.new(params[:id]).call
if data.values.all?(&:nil?)
render json: { error: "upstream unavailable" }, status: :bad_gateway
else
render json: {
profile: data[:profile],
credit_score: data[:credit],
recent_orders: data[:orders],
partial: data.values.any?(&:nil?)
}
end
end
end
When a request needs data from several independent upstream services, calling them one after another wastes time: total latency becomes the sum of every call. Because these calls do not depend on one another, they can run concurrently, turning the total latency into roughly the slowest single call. The catch is that unbounded concurrency is dangerous — spinning up a thread per item can exhaust file descriptors, overwhelm a downstream service, or trip its rate limiter. The pattern shown here fans out work across a fixed-size pool so parallelism is capped at a safe level.
The BoundedFanOut tab implements the pool using Queue, which is thread-safe, as the work channel. It enqueues each item alongside its original index, then starts exactly size worker threads. Each worker loops, popping items with a non-blocking pop(true) until the queue raises ThreadError (empty), so no sentinel values are needed. Results are written into a pre-sized results array at the item's index, preserving input order regardless of completion order. A Mutex guards the shared array. Exceptions are captured rather than raised, so one failing call cannot silently kill the whole batch; the caller decides what to do.
The EnrichmentService tab is the real-world consumer. It gathers a profile, credit score, and recent orders for a user — three unrelated HTTP calls — and passes them as lambdas to BoundedFanOut.map. Each lambda wraps its call in Timeout.timeout so a hung upstream cannot block a worker forever, and the results are matched back to their keys by position. Note that IO-bound work like HTTP benefits from threads even on MRI, because the GIL is released during blocking IO.
The trade-offs matter. A fixed pool bounds resource usage and gives a predictable ceiling on load sent to any dependency, but it is not a substitute for per-host rate limiting or circuit breaking. Capturing exceptions means partial success is possible, so callers must inspect results. Timeout.timeout is blunt and can interrupt at awkward points, so it is best reserved for genuinely external, idempotent reads. This approach fits read-heavy aggregation endpoints where several slow, independent lookups would otherwise serialize.
Related snips
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
Share this code
Here's the card — post it anywhere.
