Per-process MLX stream management for concurrent inference.

MLX dispatches GPU work through Metal command queues. By default every op shares a single command queue (the default worker thread). Emily.Stream lets each BEAM process use its own worker thread — its own Metal command queue — so multiple processes can run inference concurrently on a shared model.

Public API

• new/1 — create a stream on :gpu or :cpu. Each stream allocates a dedicated OS thread that owns the MLX stream object; the thread is joined when the stream reference is garbage collected.
• with_stream/2 — install a stream for the current process for the duration of a function call, then restore the previous stream (or the default) on exit. Nesting is safe.
• close/1 — stop a stream's worker deterministically instead of waiting for garbage collection: queued ops are cancelled (their callers raise) and the OS thread is joined off the BEAM schedulers.

How it works

with_stream/2 stores the worker reference in the process dictionary under :emily_worker. Emily.Backend reads it and passes it to every NIF call. Each NIF dispatches its work to the worker's dedicated OS thread where the MLX stream lives. Tensors allocated by one stream can be read by another (MLX arrays are refcounted and thread-safe for reads), but lazy tensors must be evaluated on the stream that created them.

Configuration

Two application-env keys tune worker behaviour (set them in your config/config.exs):

• :worker_queue_limit (default 8192) — the maximum number of operations that may be queued on a single worker before further dispatch is rejected with a RuntimeError. Each op is awaited synchronously, so a process holds at most one queued item; this cap is reached only by many processes dispatching to one worker concurrently, and provides back-pressure against a runaway producer.
• :await_timeout (default :infinity) — milliseconds to wait for a native result before raising. :infinity never times out; set a finite value to bound how long a caller can block on one operation.

config :emily, worker_queue_limit: 8192, await_timeout: :infinity

Concurrent serving patterns

Stream-per-process (shared model, per-process queues):

stream = Emily.Stream.new(:gpu)
Emily.Stream.with_stream(stream, fn ->
  Nx.Serving.batched_run(my_serving, input)
end)

Each serving worker allocates its own stream once at init. Weights are shared — no duplication.

Pooled servings (K instances behind a pool):

Start K Nx.Serving instances behind poolboy / Registry / etc. Each loads its own weights and runs on the default stream. No Emily.Stream needed. Trade-off: each pool member holds its own weight copy, so memory scales with K.

For small models the pool approach is simpler. For large models (Qwen3-7B+) where duplicating weights is impractical, use stream-per-process.

Examples

iex> stream = Emily.Stream.new(:gpu)
iex> Emily.Stream.with_stream(stream, fn -> 42 end)
42