Module: Ractor::Pipeline

Defined in:
lib/ractor/pipeline.rb,
lib/ractor/pipeline/version.rb,
sig/ractor/pipeline.rbs

Defined Under Namespace

Modules: Stages Classes: Branch, Error, Source, Stage, TokenPool

Constant Summary collapse

CREDIT =

ready tokens per (producer, consumer) pair on pull links

2
VERSION =

Returns:

  • (String) 
"0.2.0"

Class Method Summary collapse

Class Method Details

.filter_pipe(lanes: 1, &block) ⇒ Object 



457
 
# File 'lib/ractor/pipeline.rb', line 457

def filter_pipe(lanes: 1, &block) = Branch.new.filter_pipe(lanes:, &block)

.flat_pipe(lanes: 1, &block) ⇒ Object 



458
 
# File 'lib/ractor/pipeline.rb', line 458

def flat_pipe(lanes: 1, &block) = Branch.new.flat_pipe(lanes:, &block)

.pipe(lanes: 1, &block) ⇒ Object

Source-less fragments for tee branches.



456
 
# File 'lib/ractor/pipeline.rb', line 456

def pipe(lanes: 1, &block) = Branch.new.pipe(lanes:, &block)

.stream(source, batch: 1) ⇒ Object

-- DSL entry points ------------------------------------------------



449
 
# File 'lib/ractor/pipeline.rb', line 449

def stream(source, batch: 1) = Source.new(source, batch:)

.stream1(obj) ⇒ Object

A stream of exactly one element: stream1(obj) == stream([obj]). Use it to flow an object (even an each-able one) as a single element.



453
 
# File 'lib/ractor/pipeline.rb', line 453

def stream1(obj) = Source.new([obj])

.token_pools(groups) ⇒ Object 



358
359
360
 
# File 'lib/ractor/pipeline.rb', line 358

def token_pools(groups)
  groups.map { TokenPool.new }
end

.worker_loop(ctrl, job, kind) ⇒ Object

The main loop of a stage worker. Stateless between streams except for the demand bookkeeping of its (per-run) wiring.



364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
 
# File 'lib/ractor/pipeline.rb', line 364

def worker_loop(ctrl, job, kind)
  in_port = Ractor::Port.new
  ctrl << in_port

  # wait for [:init]; queue anything that arrives earlier
  early = []
  msg = in_port.receive
  until msg[0] == :init
    early << msg
    msg = in_port.receive
  end
  _, groups, ups, n_producers, batch_size = msg

  tokens = Pipeline.token_pools(groups) # demand tokens, per pull group
  pending = groups.map { [] }           # batches waiting for a token
  port_group = {}
  groups.each_with_index do |(ports, pull), gi|
    ports.each { |port| port_group[port] = gi } if pull
  end
  eos = 0

  dispatch = lambda do |batch|
    groups.each_with_index do |(ports, pull), gi|
      if !pull
        ports.first << [:data, batch, nil]
      elsif (port = tokens[gi].take)
        port << [:data, batch, in_port]
      else
        pending[gi] << batch
      end
    end
  end

  handle = lambda do |m|
    case m[0]
    when :data
      begin
        case kind
        when :pipe
          dispatch.call(m[1].map { |obj| job.call(obj) })
        when :filter_pipe
          out = m[1].select { |obj| job.call(obj) }
          dispatch.call(out) unless out.empty?
        when :flat_pipe
          out = []
          m[1].each { |obj| job.call(obj).each { |o| out << o } }
          out.each_slice(batch_size) { |slice| dispatch.call(slice) }
        end
      rescue Ractor::ClosedError
        raise
      rescue Exception => e
        groups.first.first.first << [:failure, e] rescue nil
      end
      (m[2] << [:ready, in_port] rescue nil) if m[2] # replenish credit
    when :ready
      gi = port_group[m[1]]
      if (batch = pending[gi].shift)
        m[1] << [:data, batch, in_port]
      else
        tokens[gi].add(m[1])
      end
    when :eos
      eos += 1
    when :failure
      groups.first.first.first << m rescue nil # pass through
    when :cancel
      throw :cancelled
    end
  end

  catch(:cancelled) do
    ups.each { |up| CREDIT.times { up << [:ready, in_port] rescue nil } }
    early.each { |m| handle.call(m) }
    handle.call(in_port.receive) until eos == n_producers
    # all producers finished: drain buffered batches, then EOS
    handle.call(in_port.receive) until pending.all?(&:empty?)
    groups.each { |ports, _| ports.each { |port| port << [:eos] rescue nil } }
  end
rescue Ractor::ClosedError
  # a consumer is gone: the stream was cancelled (SIGPIPE-style)
end