Choosing Concurrency Patterns in Crystal: Fibers, Channels, and When to Escape the Scheduler

The Decision: How to Structure Concurrent Work in Crystal

Crystal's concurrency model centers on lightweight fibers scheduled across a fixed pool of worker threads (one per logical CPU by default). This design delivers high throughput for I/O-bound workloads — 10,000+ concurrent connections on modest hardware — but it imposes hard constraints: a fiber that blocks the worker (via a C call or tight compute loop) stalls every other fiber on that worker. There is no preemption.

The practical decision is not "fibers or threads" but "which pattern keeps work off the scheduler when it must block or compute heavily, while using channels and select for coordination where they fit." This guide compares the supported patterns, their trade-offs, and a validated worker-pool implementation you can adapt.

Supported Concurrency Patterns Compared

Pattern	Use When	Scheduler Impact	Failure Isolation	Complexity
Pure fiber + Channel (CSP)	I/O-bound pipelines, request/response, fan-out/fan-in where every stage yields on I/O	Zero worker blocking if all I/O uses non-blocking APIs	Exception in a fiber stays in that fiber until join; unjoined fibers log at exit	Low — idiomatic, composable
Fiber pool with work channel	Bounded concurrency, backpressure, retry policies, graceful shutdown	Same as pure CSP; pool size caps fibers per worker	Wrapper fiber can supervise pool; ensure blocks for cleanup	Medium — requires supervision logic
Thread.new for blocking calls	Legacy C libraries, DNS via getaddrinfo, sleep, any syscall that blocks	Moves blocking off scheduler entirely; OS thread cost ~1 MiB stack	Thread exceptions don't propagate to fibers; must communicate via channel	Medium — channel bridge required
Process.run for CPU-heavy work	Image processing, crypto, serialization, any compute that exceeds ~1 ms	Full parallelism across cores; no GIL contention	Process isolation — crashes don't affect parent	Higher — serialization overhead, no shared memory
Hybrid: fiber pool + Thread.new bridge	Mixed workloads (e.g., HTTP API calling blocking DB driver)	Workers stay free; threads handle blocking subset	Two supervision domains; bridge fiber monitors thread health	Highest — but only pattern that handles real-world mixed loads safely

Trade-offs at a Glance

• Channels are not thread-safe across workers. A Channel(T) created on worker 0 cannot be safely sent to from a Thread.new block running on another OS thread. Bridge via a dedicated fiber that owns the channel and receives work from threads via a thread-safe queue (e.g., Concurrent::Queue from concurrent-crystal shard) or a second channel owned by the bridge fiber.
• No preemption means cooperative yielding is mandatory. Insert Fiber.yield in compute loops exceeding ~100 µs, or move the work to Process.run. A single tight loop on one worker starves 50k+ other fibers.
• Exception propagation is manual. spawn returns a Fiber handle; you must call fiber.join to re-raise. Unjoined fibers with exceptions only log to stderr at process exit. Build a supervisor fiber that joins all children in an ensure block.
• Channel close semantics are asymmetric. ch.close wakes waiters: receive raises Channel::ClosedError, send returns nil. Handle both explicitly — don't assume send raises.
• No built-in backpressure across channel chains. A fast producer feeding a slow consumer via multiple channels will buffer unboundedly in intermediate channels. Implement credit-based flow control (consumer sends "ready" tokens) or use bounded channels with select else fallback to shed load.

Concrete Implementation: Supervised Worker Pool with Graceful Shutdown

This pattern combines a bounded fiber pool, a work channel, a supervisor fiber, and a signal handler for graceful drain. It runs on Crystal 1.12+ and uses only the standard library.

Worker Pool Code

require "channel"
require "fiber"
require "signal"

class WorkerPool(T)
  # work_ch: receives {job : T, respond_ch : Channel(Result)}
  # worker_count: fibers to spawn (typically CRYSTAL_WORKERS * 2..4)
  # processor: Proc(T, Channel(Result)) -> Nil
  def initialize(@worker_count : Int32, @processor : Proc(T, Channel(Result)) -> Nil)
    @work_ch = Channel({T, Channel(Result)}).new
    @workers = [] of Fiber
    @supervisor = nil
    @shutting_down = false
  end

  def start
    @worker_count.times do |i|
      @workers << spawn(name: "worker-#{i}") do
        loop do
          job, respond_ch = @work_ch.receive
          begin
            @processor.call(job, respond_ch)
          rescue ex
            respond_ch.send(Result.failure(ex))
          end
        end
      end
    end

    @supervisor = spawn(name: "supervisor") do
      @workers.each &.join
    end
  end

  # Submit work; returns a channel that receives Result
  def submit(job : T) : Channel(Result)
    raise "Pool shutting down" if @shutting_down
    respond_ch = Channel(Result).new
    @work_ch.send({job, respond_ch})
    respond_ch
  end

  # Stop accepting new work, wait for in-flight jobs, then close
  def shutdown(timeout : Time::Span = 30.seconds)
    @shutting_down = true
    @work_ch.close  # wakes all workers with ClosedError on receive

    # Wait for supervisor (which joins all workers) with timeout
    deadline = Time.monotonic + timeout
    while Time.monotonic < deadline
      return if @supervisor.try_join
      sleep 10.milliseconds
    end
    # Force-terminate remaining fibers (rare)
    @workers.each &.cancel
    @supervisor.join
  end

  # Result type for typed response channel
  class Result
    private def initialize(@value : T? = nil, @error : Exception? = nil); end
    def self.success(value : T) : Result; new(value); end
    def self.failure(error : Exception) : Result; new(nil, error); end
    def success? : Bool; @error.nil?; end
    def value : T; @value.not_nil!; end
    def error : Exception; @error.not_nil!; end
  end
end

Usage Example: HTTP Endpoint with Bounded Concurrency

require "http/server"
require "json"

pool = WorkerPool(String).new(worker_count: 8) do |payload, respond_ch|
  # Simulate I/O-bound work (non-blocking HTTP client, DB, etc.)
  response = HTTP::Client.get("https://api.example.com/transform", form: "data=" + payload)
  respond_ch.send(WorkerPool(String)::Result.success(response.body))
rescue ex
  respond_ch.send(WorkerPool(String)::Result.failure(ex))
end
pool.start

# Graceful shutdown on SIGINT/SIGTERM
Signal::INT.trap { pool.shutdown; exit 0 }
Signal::TERM.trap { pool.shutdown; exit 0 }

server = HTTP::Server.new([
  HTTP::ErrorHandler.new,
  HTTP::LogHandler.new,
  HTTP::CompressHandler.new,
]) do |context|
  payload = context.request.body.try_gets_to_end || ""
  result_ch = pool.submit(payload)
  result = result_ch.receive
  if result.success?
    context.response.content_type = "application/json"
    context.response.print(result.value)
  else
    context.response.status_code = 500
    context.response.print({error: result.error.message}.to_json)
  end
end

server.bind_tcp(8080)
puts "Listening on http://0.0.0.0:8080"
server.listen

Why This Works

• Bounded fibers: worker_count caps concurrent in-flight jobs. Excess requests queue in @work_ch (unbounded here; add capacity for backpressure).
• Structured error handling: Each worker wraps @processor.call in begin/rescue and sends a typed Result — no exception leaks.
• Graceful drain: shutdown closes the work channel, which delivers Channel::ClosedError to every blocked receive. Workers exit their loop cleanly; supervisor joins them.
• Signal safety: Trap handlers call pool.shutdown (non-blocking) then exit. The main fiber continues serving until shutdown returns.

Validation Checklist

Run these checks to confirm the pattern behaves as expected on your target Crystal version (1.12+).

1. Version: crystal --version → expect 1.12 or later.
2. Worker count: puts ENV["CRYSTAL_WORKERS"] || "default (#{Process.cpu_count})" — matches your container/VM CPU allocation.
3. Fiber spawn/channel throughput: Build a microbenchmark with Benchmark.ips spawning 10k fibers that send/receive on a channel; expect >100k ops/sec on modern hardware.
4. Channel close behavior: Run crystal eval 'ch = Channel(Int32).new; ch.close; begin ch.receive rescue puts "ClosedError: #{$!}" end' — should print ClosedError.
5. Non-blocking I/O verification: Start the HTTP server above, hit it with wrk -t4 -c100 -d10s http://localhost:8080 — latency should stay flat as concurrency rises (no worker starvation).
6. Shutdown drain: Send SIGINT during load; verify in-flight requests complete (check logs) and process exits within timeout.

When to Escalate Beyond This Pattern

• CPU-bound stages: If @processor does image resizing, JSON serialization of large objects, or crypto, move that step to Process.run and have the worker fiber wait on the subprocess via a channel.
• Blocking C libraries: If you must call a blocking C function (e.g., legacy DB driver), wrap it in Thread.new { ... } and bridge the result back to a fiber-owned channel via a dedicated bridge fiber.
• Multi-socket NUMA: Pin workers with taskset -c 0-3 crystal app or container CPU affinity; Crystal's scheduler is not NUMA-aware.
• Observability: Add Fiber.current.name logging, channel depth metrics, and worker busy/idle counters for production visibility.

Limitations

• The pool uses an unbounded work channel; under sustained overload memory grows. Add Channel({T, Channel(Result)}).new(capacity: 1000) and handle send returning nil (channel full) with select else fallback.
• No built-in retry/backoff — implement in @processor or wrap submit.
• Crystal 1.12's Channel#close is used; older versions require a separate "poison pill" message.