Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 22 additions & 0 deletions lib/emily/stream.ex
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,9 @@ defmodule Emily.Stream do
* `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

Expand All @@ -28,6 +31,25 @@ defmodule Emily.Stream do
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.

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

## Concurrent serving patterns

**Stream-per-process** (shared model, per-process queues):
Expand Down