Class: Mutineer::WorkerPool

Inherits:
Object
  • Object
show all
Defined in:
lib/mutineer/worker_pool.rb

Overview

Fixed-size fork pool (KTD1/KTD2). run forks up to size children at once; each child runs the block on one work item, marshals its Result to a private pipe, and exits. The parent reaps any finished child with Process.wait2(-1), opening exactly one slot per reap, then refills. Results are returned in the SAME ORDER as items regardless of finish order, so verdicts are identical to a serial run (R4) and downstream output is stable.

The block is run inside the child via yield(*items[i]); whatever it returns (a Result) is the marshaled payload. Per-mutant timeout is handled one level down by Isolation (KTD2) — the pool adds no separate wall clock.

Instance Method Summary collapse

Constructor Details

#initialize(size) ⇒ WorkerPool

Builds a pool.

Parameters:

  • size (Integer) —

    desired pool size.



21
22
23
 
# File 'lib/mutineer/worker_pool.rb', line 21

def initialize(size)
  @size = [size.to_i, 1].max
end

Instance Method Details

#run(items, stop_when: nil) {|item| ... } ⇒ Array<Mutineer::Result>

Runs the work items through the pool.

Parameters:

  • items (Array<Array>) —

    work items.

  • stop_when (Proc, nil) (defaults to: nil) —

    called with each collected Result; when it returns truthy, no further items are scheduled and the run drains and returns early (#21 --fail-fast). Unscheduled slots stay nil.

Yield Parameters:

  • item (Array) —

    one work item.

Returns:

  • (Array<Mutineer::Result>) —

    results in input order (nil for any item left unscheduled by an early stop).



34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
 
# File 'lib/mutineer/worker_pool.rb', line 34

def run(items, stop_when: nil)
  results  = Array.new(items.size)
  queue    = (0...items.size).to_a
  running  = {} # pid => [index, read_io, buffer]
  stopping = false

  until queue.empty? && running.empty?
    fill(items, queue, running) { |*args| yield(*args) } unless stopping
    result = reap(results, running)
    if !stopping && stop_when && result && stop_when.call(result)
      stopping = true
      queue.clear # schedule no more; let in-flight workers drain
    end
  end

  results
end