feat(worker): bounded queue, drop-with-error stop, off-scheduler join - #132
Merged
Conversation
Reworks the per-stream worker so teardown never blocks a BEAM scheduler
and a flood of operations can't grow the queue without bound.
- Move worker state into a shared_ptr<State> so the thread outlives the
BEAM resource. On stop the worker cancels its queued tasks (each posts
{:error, :stopped}) instead of draining them, and a dedicated Reaper
joins finished worker threads off-scheduler — it tracks live workers
and is itself joined in the NIF unload callback. The resource
destructor just hands the thread to the reaper (non-blocking).
- Bound the queue (config :emily, worker_queue_limit:, default 8192);
run_async rejects when full so memory can't grow without limit.
- Add Emily.Stream.close/1 for deterministic teardown, an opt-in await
timeout (config :emily, await_timeout:, default :infinity), and
Emily.Native.worker_queue_depth/1. Remove the dead run_sync.
Closes #121
Closes #122
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Closes #121 and #122 — reworks the per-stream worker so teardown never blocks a
BEAM scheduler and the async queue can't grow without bound. One PR, since both
rework
WorkerThread's internals.#121 — non-blocking teardown
The resource destructor used to drain the entire queue and
join()the OSthread inline, so collecting a busy stream during GC stalled a scheduler.
shared_ptr<State>, so the thread safely outlivesthe BEAM resource.
{:error, :stopped}) instead of running them, so a join waits for at most theone in-flight kernel — never the backlog.
live workers so the NIF unload callback can stop+join stragglers (no
thread is ever abandoned — detaching isn't allowed under the BEAM). The
destructor just hands the thread to the reaper, non-blocking.
Emily.Stream.close/1releases a stream's worker deterministically off the GCpath (a plain, non-blocking NIF — the reaper does the join).
#122 — back-pressure, cancellation, timeout
config :emily, worker_queue_limit:, default 8192);run_asyncrejects when full, so a flood can't pin host/GPU memory. Every opis awaited, so a process holds ≤1 queued item — the cap only trips under a
many-process flood.
{:error, :stopped}to every queuedcaller instead of leaving it blocked forever.
config :emily, await_timeout:, default:infinity—no behaviour change) and
Emily.Native.worker_queue_depth/1for observability.run_sync.De-scoped (as discussed): true in-flight cancellation (can't interrupt a Metal
kernel) and liveness-check-before-reply (
enif_sendalready no-ops on dead pids).Verification
mix precommitgreen, stable across multiple seeds (543 tests + 39 doctests +79 properties, 0 failures, credo
--strictclean).test/emily/worker_lifecycle_test.exs: non-positive limit rejected,idempotent
stop_worker, drop-cancels-with-:stopped-and-never-hangs,back-pressure rejection,
Stream.close/1, and a churn test(100 create/drop busy streams) asserting the reaper reaps without crashing or
leaking.
GC'd / half closed): no crash, allocator memory returned to baseline (delta 0).