Class: CDC::Parallel::ProcessorPool::WorkerSlot Private

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

Overview

This class is part of a private API. You should avoid using this class if possible, as it may be removed or be changed in the future.

One supervised worker position in the pool.

A slot keeps its identity while its underlying Ractor can be replaced after unexpected death. The slot also owns in-flight reply ports so a fatal worker exit can be reported back to callers instead of leaving them blocked forever.

rubocop:disable Metrics/ClassLength

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(index:, processor:, supervision:, max_respawns:, respawn_window:, respawn_cooldown:) ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Create a supervised worker slot.

Parameters:

  • index (Integer) —

    Stable slot index inside the owning pool.

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

    Shareable processor instance used by the worker.

  • supervision (Boolean) —

    Whether unexpected worker death should trigger respawn.

  • max_respawns (Integer) —

    Maximum crash count inside the respawn window before cooldown.

  • respawn_window (Numeric) —

    Sliding crash-loop window in seconds.

  • respawn_cooldown (Numeric) —

    Cooldown duration in seconds after repeated crashes.



422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
 
# File 'lib/cdc/parallel/processor_pool.rb', line 422

def initialize(index:, processor:, supervision:, max_respawns:, respawn_window:, respawn_cooldown:)
  @index = index
  @processor = processor
  @supervision = supervision
  @max_respawns = Integer(max_respawns)
  @respawn_window = Float(respawn_window)
  @respawn_cooldown = Float(respawn_cooldown)
  @lock = Mutex.new
  @inflight = {}
  @crashes = []
  @respawns = 0
  @shutdown = false
  @degraded_until = nil
  boot!
  @supervisor_thread = Thread.new { supervise }
end

Instance Attribute Details

#respawns ⇒ Integer (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Number of times this slot has booted a replacement worker.

Returns:

  • (Integer) 


404
405
406
 
# File 'lib/cdc/parallel/processor_pool.rb', line 404

def respawns
  @respawns
end

Instance Method Details

#complete(index) ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Mark an in-flight work item as completed.

Parameters:

  • index (Integer) 


491
492
493
 
# File 'lib/cdc/parallel/processor_pool.rb', line 491

def complete(index)
  @lock.synchronize { @inflight.delete(index) }
end

#degraded? ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Return whether this slot is currently in crash-loop cooldown.

Returns:

  • (Boolean) 


499
500
501
 
# File 'lib/cdc/parallel/processor_pool.rb', line 499

def degraded?
  @lock.synchronize { degraded_locked? }
end

#degraded_locked? ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Return whether this slot is degraded while the caller holds the lock.

Returns:

  • (Boolean) 


507
508
509
510
511
512
513
514
515
516
517
 
# File 'lib/cdc/parallel/processor_pool.rb', line 507

def degraded_locked?
  degraded_until = @degraded_until
  return false unless degraded_until

  if Process.clock_gettime(Process::CLOCK_MONOTONIC) >= degraded_until
    @degraded_until = nil
    false
  else
    true
  end
end

#inbox ⇒ Ractor::Port

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Return the current worker inbox port.

Returns:

  • (Ractor::Port) 


451
452
453
 
# File 'lib/cdc/parallel/processor_pool.rb', line 451

def inbox
  @lock.synchronize { @inbox }
end

#join ⇒ Thread

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Wait for the supervisor thread to finish.

Returns:

  • (Thread) 


571
572
573
 
# File 'lib/cdc/parallel/processor_pool.rb', line 571

def join
  @supervisor_thread.join
end

#require_failure(failure) ⇒ Exception

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Return a non-nil failure object for Steep and runtime dispatch.

Parameters:

  • failure (Exception, nil)

Returns:

  • (Exception)

Raises:

  • (RuntimeError) —

    if no failure was supplied



541
542
543
544
545
 
# File 'lib/cdc/parallel/processor_pool.rb', line 541

def require_failure(failure)
  raise "worker failure unavailable" if failure.nil?

  failure
end

#send_to_inbox(inbox, message) ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Send a message to a worker inbox.

Ractor::Port supports both #send and #<<. Some tests replace the inbox with a small sentinel object that only implements #send so the closed inbox path can be exercised without relying on a live port. Keep this helper deliberately typed as an inbox boundary rather than forcing every caller to narrow the test double shape.

Parameters:

  • inbox (#send, nil)
  • message (Object)

Raises:

  • (Ractor::ClosedError) —

    when the worker inbox has already closed



560
561
562
563
564
565
 
# File 'lib/cdc/parallel/processor_pool.rb', line 560

def send_to_inbox(inbox, message)
  raise Ractor::ClosedError, "worker inbox closed" if inbox.nil?

  inbox.send(message)
  nil
end

#send_work(index, item, reply_port) ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Send one indexed work item to the current worker.

If the slot is cooling down or the worker inbox is already closed, a serialized failure response is sent to the caller reply port.

Parameters:

  • index (Integer)
  • item (Object)
  • reply_port (Ractor::Port) 


465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
 
# File 'lib/cdc/parallel/processor_pool.rb', line 465

def send_work(index, item, reply_port)
  target = nil
  immediate_failure = nil

  @lock.synchronize do
    if degraded_locked?
      immediate_failure = RuntimeError.new("worker slot #{@index} is cooling down after repeated crashes")
    else
      @inflight[index] = reply_port
      target = @inbox
    end
  end

  return send_failure(reply_port, index, require_failure(immediate_failure)) if immediate_failure

  send_to_inbox(target, [index, item, reply_port])
rescue Ractor::ClosedError => e
  complete(index)
  send_failure(reply_port, index, e)
end

#shutdown ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Request worker shutdown and prevent future respawns.



523
524
525
526
527
528
529
530
531
532
533
 
# File 'lib/cdc/parallel/processor_pool.rb', line 523

def shutdown
  target = nil
  @lock.synchronize do
    @shutdown = true
    target = @inbox
  end

  send_to_inbox(target, nil)
rescue Ractor::ClosedError
  nil
end

#worker ⇒ Ractor

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Return the current worker Ractor.

Returns:

  • (Ractor) 


443
444
445
 
# File 'lib/cdc/parallel/processor_pool.rb', line 443

def worker
  @lock.synchronize { @worker }
end