Class: CDC::Parallel::ProcessorPool

Inherits:
Object
  • Object
show all
Defined in:
lib/cdc/parallel/processor_pool.rb

Overview

Note:

ProcessorPool preserves the order of returned results, not the order in which independent items execute. If a sink needs strict ordering by transaction, relation, or primary key, use the ecosystem ordering contract and an ordered dispatcher/runtime above this primitive.

Executes one Ractor-safe cdc-core processor across a fixed set of pre-warmed Ractor workers.

ProcessorPool is the low-level execution primitive used by Runtime. It accepts normalized cdc-core work items, sends them across Ractor boundaries, invokes the configured processor, and returns CDC::Core::ProcessorResult objects in input order.

This class is intentionally focused on CPU-bound parallel execution. Use it when the processor spends most of its time doing Ruby work such as transformation, enrichment, serialization, compression, scoring, or other in-memory computation. For I/O-heavy work, the CDC Ecosystem boundary is a future fiber-friendly runtime such as cdc-concurrent.

Processor safety contract

The supplied processor must declare ractor_safe! on its class. That declaration is treated as the processor author's explicit promise that the processor object and its dependencies can safely cross a Ractor boundary.

ProcessorPool validates this declaration before booting workers:

Declaring ractor_safe! does not make unsafe code safe. It only allows the processor to be passed into worker Ractors. Mutable global state, database connections, sockets, caches, file handles, and non-shareable objects still need to be designed carefully by the processor implementor.

Execution model

Workers are created during initialization and reused for all dispatches. This pays Ractor startup cost once and keeps the pool stable even when individual processor calls fail.

The pool uses a fan-out / fan-in pattern:

work items
     |
     v
ProcessorPool
     |
     +----> Worker Ractor 1
     +----> Worker Ractor 2
     +----> Worker Ractor N
                   |
                   v
            ProcessorResult
                   |
                   v
             ordered results

Fan-out uses round-robin worker selection. Fan-in collects responses from a reply port and reorders them by submission index, so process_many always returns results in the same order as the input array even when work completes out of order.

Examples:

Declaring a processor as Ractor-safe

class AnalyticsProcessor < CDC::Core::Processor
  ractor_safe!

  def process(event)
    CDC::Core::ProcessorResult.success(event)
  end
end

pool = CDC::Parallel::ProcessorPool.new(
  processor: AnalyticsProcessor.new,
  size: 4
)

Processing one item

result = pool.process(event)
result.success? #=> true

Processing a batch while preserving result order

results = pool.process_many([event_a, event_b, event_c])
results.map(&:success?)

Shutting down explicitly

pool.shutdown

See Also:

Defined Under Namespace

Classes: WorkerSlot

Instance Method Summary collapse

Constructor Details

#initialize(processor:, size: Etc.nprocessors, timeout: nil, supervision: true, max_respawns: 3, respawn_window: 60, respawn_cooldown: 5, manage_lifecycle: true) ⇒ void

Create a new pool and boot its worker Ractors.

rubocop:disable Metrics/MethodLength

Parameters:

  • processor (CDC::Core::Processor) —

    Processor instance used by every worker. Its class must respond to ractor_safe? and return true.

  • size (Integer) (defaults to: Etc.nprocessors) —

    Number of worker Ractors to boot. Defaults to Etc.nprocessors.

  • timeout (Numeric, nil) (defaults to: nil) —

    Optional timeout, in seconds, used when waiting for worker results and during shutdown. nil means wait indefinitely.

  • supervision (Boolean) (defaults to: true) —

    Whether worker Ractors should be respawned after unexpected death.

  • max_respawns (Integer) (defaults to: 3) —

    Maximum crash count inside respawn_window before a worker slot enters cooldown.

  • respawn_window (Numeric) (defaults to: 60) —

    Time window, in seconds, used by the crash-loop circuit breaker.

  • respawn_cooldown (Numeric) (defaults to: 5) —

    Cooldown, in seconds, before a crash-looping slot is revived again.

  • manage_lifecycle (Boolean) (defaults to: true) —

    When true (default), the pool calls processor.start during initialization and processor.flush + processor.stop during shutdown. Set to false when a higher-level runtime (e.g. Runtime) owns the processor lifecycle so that start/stop/flush are not called multiple times when the same processor is shared across pools.

Raises:

  • (UnsafeProcessorError) —

    Raised when the processor class has not declared ractor_safe!.

  • (ArgumentError) —

    Raised by Configuration when size or timeout is invalid.



128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
 
# File 'lib/cdc/parallel/processor_pool.rb', line 128

def initialize(
  processor:,
  size: Etc.nprocessors,
  timeout: nil,
  supervision: true,
  max_respawns: 3,
  respawn_window: 60,
  respawn_cooldown: 5,
  manage_lifecycle: true
)
  validate_processor!(processor)

  processor.start if manage_lifecycle
  @processor = ::Ractor.make_shareable(processor)
  @manage_lifecycle = manage_lifecycle
  @configuration = Configuration.new(size:, timeout:)
  @slots = Array.new(@configuration.size) do |index|
    WorkerSlot.new(
      index:,
      processor: @processor,
      supervision:,
      max_respawns:,
      respawn_window:,
      respawn_cooldown:
    )
  end.freeze
  @workers = @slots.map(&:worker).freeze
  @worker_inboxes = @slots.map(&:inbox).freeze

  @next_worker = 0
  @dispatch_mutex = Mutex.new
  @shutdown = false
end

Instance Method Details

#degraded? ⇒ Boolean

Return whether any worker slot is currently in crash-loop cooldown.

Returns:

  • (Boolean) 


173
174
175
 
# File 'lib/cdc/parallel/processor_pool.rb', line 173

def degraded?
  @slots.any?(&:degraded?)
end

#process(item) ⇒ CDC::Core::ProcessorResult

Process one work item synchronously.

This is a convenience wrapper around #process_many. The work still executes inside a worker Ractor; the call blocks until the corresponding CDC::Core::ProcessorResult is available or until the optional timeout is reached.

Parameters:

  • item (Object) —

    Shareable work item, usually a CDC::Core::ChangeEvent.

Returns:

  • (CDC::Core::ProcessorResult) —

    Normalized processor result. Processor exceptions are captured as failure results rather than escaping directly from the worker Ractor.

Raises:



191
192
193
 
# File 'lib/cdc/parallel/processor_pool.rb', line 191

def process(item)
  process_many([item]).fetch(0)
end

#process_many(items) ⇒ Array<CDC::Core::ProcessorResult>

Process many work items using the pre-warmed worker pool.

Each item is made shareable before dispatch. Items are assigned to worker inboxes using round-robin selection. Responses are collected through a per-call reply port and returned in the same order as the input array.

Parameters:

  • items (Array<Object>) —

    Work items to process. Empty arrays are valid and return an empty frozen array.

Returns:

  • (Array<CDC::Core::ProcessorResult>) —

    Frozen array of normalized results, ordered to match items.

Raises:



208
209
210
211
212
213
214
215
216
217
 
# File 'lib/cdc/parallel/processor_pool.rb', line 208

def process_many(items)
  work_items = items.map { |item| ::Ractor.make_shareable(item) }
  reply_port = ::Ractor::Port.new

  assignments = dispatch(work_items, reply_port)

  collect_results(reply_port, work_items.length, assignments).compact.freeze
ensure
  reply_port&.close
end

#respawns ⇒ Integer

Return total worker-slot respawns since this pool booted.

Returns:

  • (Integer) 


166
167
168
 
# File 'lib/cdc/parallel/processor_pool.rb', line 166

def respawns
  @slots.sum(&:respawns)
end

#shutdown ⇒ void

This method returns an undefined value.

Shut down the pool and wait for worker Ractors to exit.

Shutdown is idempotent. The first caller signals all worker inboxes with a stop message and waits for workers to join. Later calls return without doing anything.



226
227
228
229
230
231
232
233
234
235
236
237
238
239
 
# File 'lib/cdc/parallel/processor_pool.rb', line 226

def shutdown
  @dispatch_mutex.synchronize do
    return if @shutdown

    @shutdown = true
    signal_workers
  end

  wait_for_workers
  return unless @manage_lifecycle

  @processor.flush
  @processor.stop
end