Class: Omnizip::Parallel::Engine

Inherits:
Object
  • Object
show all
Defined in:
lib/omnizip/parallel/engine.rb

Overview

Engine owns the Fractor worker-pool dance: build the pool, submit a batch of work, run it to completion, return successful and failed results. Callers (ParallelCompressor, ParallelExtractor, future operations) supply the worker class and the work items; Engine handles the threading.

This is the deep module behind the parallel API: the threading shape is identical across every consumer, so it lives here once. Domain-specific concerns (which file to read, which archive to open, what stats to track) stay with the caller.

Instance Method Summary collapse

Constructor Details

#initialize(worker_class:, threads:) ⇒ Engine

Returns a new instance of Engine.

Parameters:

  • worker_class (Class) —

    a Fractor::Worker subclass whose #process(work) turns one work item into a result.

  • threads (Integer) —

    worker count.



21
22
23
24
 
# File 'lib/omnizip/parallel/engine.rb', line 21

def initialize(worker_class:, threads:)
  @worker_class = worker_class
  @threads = threads
end

Instance Method Details

#run(work_items) {|result| ... } ⇒ Array<Fractor::WorkResult>

Run work_items through the pool and yield each successful Fractor::WorkResult to the caller-supplied block.

Parameters:

  • work_items (Array<Fractor::Work>) —

    batch to submit.

Yield Parameters:

  • result (Fractor::WorkResult) —

    a successful result.

Returns:

  • (Array<Fractor::WorkResult>) —

    the failed results, for the caller to decide how to surface.



33
34
35
36
37
38
39
40
41
42
43
44
45
 
# File 'lib/omnizip/parallel/engine.rb', line 33

def run(work_items)
  pool = Fractor::WorkerPool.new(
    worker_class: @worker_class,
    num_workers: @threads,
    continuous: false,
  )
  pool.start
  pool.submit_batch(work_items)
  pool.run

  pool.successful_results.each { |r| yield r if block_given? }
  pool.failed_results
end