Class: Ractor

Inherits:
Object
  • Object
show all
Defined in:
lib/ractor/pipeline.rb,
lib/ractor/pipeline/version.rb,
sig/ractor/pipeline.rbs

Overview

Ractor::Pipeline: a DSL to build stream processing pipelines with Ractors.

include Ractor::Pipeline

stream(File.foreach(name)).
filter_pipe(lanes: 4){ it.include?("foo") }.
reduce([0, 0, 0]) do |(lines, words, bytes), line|
  [lines + 1, words + line.scan(/\S+/).size, bytes + line.bytesize]
end

Execution model:

* stream(source, batch: k) feeds the source into the pipeline from a
Thread in the caller Ractor, k elements per message (default 1).
Batching is transparent: stage blocks always see single elements.
* pipe/filter_pipe/flat_pipe are persistent Ractor stages; each stage
spawns `lanes:` (default 1) worker Ractors at terminal-operation
time. Workers process many elements; no Ractor is created per
element.
* Boundaries into a multi-lane stage (lanes > 1) are demand-driven
(pull): a lane worker sends a ready token upstream when it can take
more work, and producers send a batch only to a worker they hold a
token for (CREDIT tokens per producer/consumer pair). A busy worker
stops sending tokens, so work never piles up behind it. Boundaries
with a single consumer (lanes: 1 stages, and the terminal) are plain
push.
* When the head stage is multi-lane, the feeder reads the source only
while it holds tokens, so a fast or infinite source cannot flood the
pipeline.
* tee broadcasts each element to every branch; branch outputs are
merged (unordered) into one downstream stream.
* reduce/each/to_a/count/first are terminal operations executed in the
caller Ractor.

Semantics notes:

* Ordering: a chain of lanes: 1 stages preserves input order (also
with batch:). Multi-lane stages are unordered.
* End of stream is an in-band EOS message: each worker counts EOS from
its upstream producers, drains its buffered output, and then
broadcasts EOS downstream, so fan-in and fan-out shut down cleanly.
* Cancellation (first, exceptions): the terminal closes its out_port
and broadcasts a cancel message to every worker. Workers exit
immediately, dropping their backlog; sends to closed ports raise
Ractor::ClosedError and cascade the closure upstream (SIGPIPE-style).
* Non-shareable objects are deep-copied at each Ractor boundary
(Ractor::Port#send default). tee copies an element once per branch.
* Stage blocks are isolated with Ractor.shareable_proc at DSL
construction time: they may capture shareable outer values (they are
snapshotted), and capturing a non-shareable value raises
Ractor::IsolationError at pipe/filter_pipe/flat_pipe call time.
* An exception raised in a stage block cancels the pipeline and is
re-raised by the terminal operation in the caller Ractor.
* Push links are unbounded (no backpressure); pull links are bounded
by CREDIT batches per producer/consumer pair.

Message protocol (one input Port per worker; messages are tagged):

[:init, groups, ups, n_producers, batch_size]
       groups: [[ports, pull?], ...] one per fan-out destination
[:data, batch, from]   # from: replenish target on pull links, else nil
[:ready, port]         # demand token (pull links only)
[:eos]                 # counted per upstream producer
[:failure, exception]  # forwarded to the terminal, which raises it
[:cancel]

Defined Under Namespace

Modules: Pipeline