Class: Fractor::Supervisor

Inherits:
Object
  • Object
show all
Defined in:
lib/fractor/supervisor.rb

Overview

Supervises multiple WrappedRactors, distributes work, and aggregates results.

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(worker_pools: [], continuous_mode: false, debug: false, logger: nil, tracer_enabled: nil, tracer_stream: nil, enable_performance_monitoring: false) ⇒ Supervisor

Initializes the Supervisor.

  • worker_pools: An array of worker pool configurations, each containing:
    • worker_class: The class inheriting from Fractor::Worker (e.g., MyWorker).
    • num_workers: The number of Ractors to spawn for this worker class.
  • continuous_mode: Whether to run in continuous mode without expecting a fixed work count.
  • debug: Enable verbose debugging output for all state changes.
  • logger: Optional logger instance for this Supervisor (defaults to Fractor.logger). Provides isolation when multiple gems use Fractor in the same process.
  • tracer_enabled: Optional override for ExecutionTracer (nil uses global setting).
  • tracer_stream: Optional trace stream for this Supervisor (nil uses global setting).
  • enable_performance_monitoring: Enable performance monitoring (latency, throughput, etc.).


28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
 
# File 'lib/fractor/supervisor.rb', line 28

def initialize(worker_pools: [], continuous_mode: false, debug: false, logger: nil,
               tracer_enabled: nil, tracer_stream: nil, enable_performance_monitoring: false)
  @debug = debug || ENV["FRACTOR_DEBUG"] == "1"
  @logger = logger # Store instance-specific logger for isolation
  @tracer_enabled = tracer_enabled
  @tracer_stream = tracer_stream
  @worker_pools = worker_pools.map.with_index do |pool_config, index|
    worker_class = pool_config[:worker_class]
    num_workers = pool_config[:num_workers] || detect_num_workers

    # Validate worker_class
    unless worker_class.is_a?(Class)
      raise ArgumentError,
            "worker_class must be a Class (got #{worker_class.class}), in worker_pools[#{index}]\n\n" \
            "Expected: { worker_class: MyWorker }\n" \
            "Got:      { worker_class: #{worker_class.inspect} }\n\n" \
            "Fix: Use the class itself, not a symbol or string.\n" \
            "Example: { worker_class: MyWorker }  # Correct\n" \
            "         { worker_class: 'MyWorker' } # Wrong - this is a string"
    end

    unless worker_class < Fractor::Worker
      raise ArgumentError,
            "#{worker_class} must inherit from Fractor::Worker, in worker_pools[#{index}]\n\n" \
            "Your worker class must be defined as:\n" \
            "  class #{worker_class} < Fractor::Worker\n" \
            "    def process(work)\n" \
            "      # ...\n" \
            "    end\n" \
            "  end\n\n" \
            "Did you forget to inherit from Fractor::Worker?"
    end

    # Validate num_workers
    unless num_workers.is_a?(Integer) && num_workers.positive?
      raise ArgumentError,
            "num_workers must be a positive integer (got #{num_workers.inspect}), in worker_pools[#{index}]\n\n" \
            "Valid values: Integer >= 1\n" \
            "Examples: { num_workers: 4 }  # Use 4 workers\n" \
            "          { num_workers: Etc.nprocessors }  # Use available CPUs"
    end

    {
      worker_class: worker_class,
      num_workers: num_workers,
      workers: [], # Will hold the WrappedRactor instances
    }
  end

  @work_queue = Queue.new
  @results = ResultAggregator.new
  @workers = [] # Flattened array of all workers across all pools
  @total_work_count = 0 # Track total items initially added
  @ractors_map = {} # Map Ractor object to WrappedRactor instance
  @continuous_mode = continuous_mode
  @running = false
  @wakeup_ractor = nil # Control ractor for unblocking select
  @timer_thread = nil # Timer thread for periodic wakeup
  @error_reporter = ErrorReporter.new # Track errors and statistics
  @performance_monitor = nil # Performance monitor instance

  # Initialize callback registry for managing work and error callbacks
  @callback_registry = CallbackRegistry.new(debug: @debug)

  # Initialize performance monitor if enabled
  if enable_performance_monitoring
    require_relative "performance_monitor"
    @performance_monitor = PerformanceMonitor.new(self)
    @performance_monitor.start
  end

  # Initialize work distribution manager (handles idle workers and work assignment)
  @work_distribution_manager = WorkDistributionManager.new(
    @work_queue,
    @workers,
    @ractors_map,
    debug: @debug,
    continuous_mode: @continuous_mode,
    performance_monitor: @performance_monitor,
  )

  # Initialize shutdown handler (manages graceful shutdown)
  @shutdown_handler = ShutdownHandler.new(
    @workers,
    @wakeup_ractor,
    @timer_thread,
    @performance_monitor,
    debug: @debug,
    continuous_mode: @continuous_mode,
  )

  # Initialize signal handler for graceful shutdown
  @signal_handler = SignalHandler.new(
    continuous_mode: @continuous_mode,
    debug: @debug,
    status_callback: -> { print_status },
    shutdown_callback: ->(mode) { handle_shutdown_callback(mode) },
  )

  # Initialize error formatter for error messages
  @error_formatter = ErrorFormatter.new
end

Instance Attribute Details

#callback_registry ⇒ Object (readonly)

Returns the value of attribute callback_registry.



14
15
16
 
# File 'lib/fractor/supervisor.rb', line 14

def callback_registry
  @callback_registry
end

#debug ⇒ Object (readonly)

Returns the value of attribute debug.



14
15
16
 
# File 'lib/fractor/supervisor.rb', line 14

def debug
  @debug
end

#error_reporter ⇒ Object (readonly)

Returns the value of attribute error_reporter.



14
15
16
 
# File 'lib/fractor/supervisor.rb', line 14

def error_reporter
  @error_reporter
end

#logger ⇒ Object (readonly)

Returns the value of attribute logger.



14
15
16
 
# File 'lib/fractor/supervisor.rb', line 14

def logger
  @logger
end

#performance_monitor ⇒ Object (readonly)

Returns the value of attribute performance_monitor.



14
15
16
 
# File 'lib/fractor/supervisor.rb', line 14

def performance_monitor
  @performance_monitor
end

#results ⇒ Object (readonly)

Returns the value of attribute results.



14
15
16
 
# File 'lib/fractor/supervisor.rb', line 14

def results
  @results
end

#work_queue ⇒ Object (readonly)

Returns the value of attribute work_queue.



14
15
16
 
# File 'lib/fractor/supervisor.rb', line 14

def work_queue
  @work_queue
end

#worker_pools ⇒ Object (readonly)

Returns the value of attribute worker_pools.



14
15
16
 
# File 'lib/fractor/supervisor.rb', line 14

def worker_pools
  @worker_pools
end

#workers ⇒ Object (readonly)

Returns the value of attribute workers.



14
15
16
 
# File 'lib/fractor/supervisor.rb', line 14

def workers
  @workers
end

Class Method Details

.configuration_help ⇒ String

Class-level documentation for Supervisor configuration options. Provides a summary of valid configuration parameters for the initialize method.

Examples:

Print configuration help

puts Fractor::Supervisor.configuration_help

Returns:

  • (String) —

    Configuration documentation



590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
 
# File 'lib/fractor/supervisor.rb', line 590

def self.configuration_help
  <<~HELP
    Fractor::Supervisor Configuration Options
    ==========================================

    The Supervisor accepts the following keyword arguments to initialize():

    worker_pools (Array, required)
      Array of worker pool configuration hashes.
      Each hash must contain:
        - worker_class: Class inheriting from Fractor::Worker (required)
        - num_workers: Positive integer for number of workers (optional, defaults to CPU count)

      Example:
        worker_pools: [
          { worker_class: MyWorker, num_workers: 4 },
          { worker_class: AnotherWorker, num_workers: 2 }
        ]

    continuous_mode (Boolean, optional, default: false)
      Whether to run in continuous mode (long-running) or batch mode.
      - false: Batch mode - processes all work items and exits
      - true: Continuous mode - runs until stopped, accepts work from callbacks

    debug (Boolean, optional, default: false)
      Enable verbose debug output for all state changes.
      Can also be enabled via FRACTOR_DEBUG=1 environment variable.

    logger (Logger, optional, default: Fractor.logger)
      Optional logger instance for this Supervisor.
      Provides isolation when multiple gems use Fractor in the same process.

    tracer_enabled (Boolean, optional)
      Override for ExecutionTracer. nil uses global setting.

    tracer_stream (IO, optional)
      Optional trace stream for this Supervisor. nil uses global setting.

    enable_performance_monitoring (Boolean, optional, default: false)
      Enable performance monitoring (latency, throughput, etc.).
      When enabled, performance_metrics() returns current metrics.

    Validation
    ----------
    All configuration is validated at initialization time with detailed error messages.
    Invalid configurations will raise ArgumentError with helpful fix suggestions.

    For more information, see the Supervisor class documentation.
  HELP
end

Instance Method Details

#add_work_item(work) ⇒ Object

Adds a single work item to the queue. The item must be an instance of Fractor::Work or a subclass.



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
161
162
163
164
 
# File 'lib/fractor/supervisor.rb', line 133

def add_work_item(work)
  unless work.is_a?(Fractor::Work)
    raise ArgumentError,
          "#{work.class} must be an instance of Fractor::Work.\n\n" \
          "Received: #{work.inspect}\n\n" \
          "To create a valid work item:\n" \
          "  class MyWork < Fractor::Work\n" \
          "    def initialize(data)\n" \
          "      super({ value: data })\n" \
          "    end\n" \
          "  end\n\n" \
          "  work = MyWork.new(42)\n" \
          "  supervisor.add_work_item(work)"
  end

  @work_queue << work
  @total_work_count += 1

  # Trace work item queued
  trace_work(:queued, work, queue_size: @work_queue.size)

  # Distribute to idle workers if work_distribution_manager is available
  # This ensures work added from callbacks gets picked up immediately
  if @work_distribution_manager && @running
    distributed = @work_distribution_manager.distribute_to_idle_workers
    puts "Distributed work to #{distributed} idle workers after add_work_item" if @debug && distributed.positive?
  end

  return unless @debug

  puts "Work item added. Initial work count: #{@total_work_count}, Queue size: #{@work_queue.size}"
end

#add_work_items(works) ⇒ Object

Adds multiple work items to the queue. Each item must be an instance of Fractor::Work or a subclass.



168
169
170
171
172
 
# File 'lib/fractor/supervisor.rb', line 168

def add_work_items(works)
  works.each do |work|
    add_work_item(work)
  end
end

#debug! ⇒ Object

Enable debug mode for verbose output



561
562
563
 
# File 'lib/fractor/supervisor.rb', line 561

def debug!
  @debug = true
end

#debug? ⇒ Boolean

Check if debug mode is enabled

Returns:

  • (Boolean) 


571
572
573
 
# File 'lib/fractor/supervisor.rb', line 571

def debug?
  @debug
end

#debug_off! ⇒ Object

Disable debug mode



566
567
568
 
# File 'lib/fractor/supervisor.rb', line 566

def debug_off!
  @debug = false
end

#handle_shutdown_callback(mode) ⇒ Object

Callback for signal handler shutdown requests.



278
279
280
281
282
283
284
285
 
# File 'lib/fractor/supervisor.rb', line 278

def handle_shutdown_callback(mode)
  if mode == :graceful
    stop
  else
    # Immediate shutdown - raise signal in current thread
    Thread.current.raise(ShutdownSignal, "Interrupted by signal")
  end
end

#inspect_queue ⇒ Object

Inspect the current state of the work queue Returns a hash with queue information and items



507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
 
# File 'lib/fractor/supervisor.rb', line 507

def inspect_queue
  items = []
  # Queue doesn't have to_a, need to iterate
  temp_queue = Queue.new
  until @work_queue.empty?
    item = @work_queue.pop
    items << item
    temp_queue.push(item)
  end
  # Restore the queue
  until temp_queue.empty?
    @work_queue.push(temp_queue.pop)
  end

  {
    size: @work_queue.size,
    total_added: @total_work_count,
    items: items.map do |work|
      {
        class: work.class.name,
        input: work.input,
        inspect: work.inspect,
      }
    end,
  }
end

#on_error(&callback) ⇒ Object

Register a callback to handle errors The callback receives (error_result, worker_name, worker_class) Example: supervisor.on_error { |err, worker, klass| puts "Error in #klass: #Fractor::Supervisor.errerr.error" }



183
184
185
 
# File 'lib/fractor/supervisor.rb', line 183

def on_error(&callback)
  @callback_registry.register_error_callback(&callback)
end

#performance_metrics ⇒ Object

Get performance metrics snapshot if performance monitoring is enabled Returns nil if performance monitoring is not enabled



577
578
579
580
581
 
# File 'lib/fractor/supervisor.rb', line 577

def performance_metrics
  return nil unless @performance_monitor

  @performance_monitor.snapshot
end

#print_status ⇒ Object

Prints current supervisor status



288
289
290
291
292
293
294
295
296
297
298
299
 
# File 'lib/fractor/supervisor.rb', line 288

def print_status
  status = @work_distribution_manager.status_summary
  puts "\n=== Fractor Supervisor Status ==="
  puts "Mode: #{@continuous_mode ? 'Continuous' : 'Batch'}"
  puts "Running: #{@running}"
  puts "Workers: #{@workers.size}"
  puts "Idle workers: #{status[:idle]}"
  puts "Queue size: #{@work_queue.size}"
  puts "Results: #{@results.results.size}"
  puts "Errors: #{@results.errors.size}"
  puts "================================\n"
end

#register_work_source(&callback) ⇒ Object

Register a callback to provide new work items The callback should return nil or empty array when no new work is available



176
177
178
 
# File 'lib/fractor/supervisor.rb', line 176

def register_work_source(&callback)
  @callback_registry.register_work_source(&callback)
end

#run ⇒ Object

Runs the main processing loop.



310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
 
# File 'lib/fractor/supervisor.rb', line 310

def run
  setup_signal_handler
  start_workers

  @running = true

  # Distribute any work that was added before run() was called
  # This is critical for Ruby 4.0 where workers need explicit work distribution
  if @work_distribution_manager
    distributed = @work_distribution_manager.distribute_to_idle_workers
    puts "Distributed initial work to #{distributed} idle workers (work_queue.size: #{@work_queue.size})" if @debug
  end

  # Start timer thread for continuous mode to periodically check work sources
  # CRITICAL: Always start timer thread in continuous mode to ensure main loop
  # can periodically check for worker termination during shutdown
  start_timer_thread if @continuous_mode

  begin
    # Run the main event loop through MainLoopHandler
    @main_loop_handler = MainLoopHandler.create(self, debug: @debug)
    @main_loop_handler.run_loop
  rescue ShutdownSignal => e
    puts "Shutdown signal caught: #{e.message}" if @debug
    puts "Sending shutdown message to all Ractors..." if @debug

    # Send shutdown message to each worker Ractor
    @workers.each do |w|
      w.send(:shutdown)
      puts "Sent shutdown to Ractor: #{w.name}" if @debug
    rescue StandardError => send_error
      puts "Error sending shutdown to Ractor #{w.name}: #{send_error.message}" if @debug
    end

    puts "Exiting due to shutdown signal." if @debug
    exit!(1) # Force exit immediately
  end

  return if @continuous_mode

  return unless @debug

  puts "Final Aggregated Results: #{@results.inspect}"
end

#setup_signal_handler ⇒ Object

Sets up signal handlers for graceful shutdown. Uses SignalHandler to manage signal handling logic.



273
274
275
 
# File 'lib/fractor/supervisor.rb', line 273

def setup_signal_handler
  @signal_handler.setup
end

#start ⇒ Object

Starts the supervisor (alias for run). Provides a consistent API with stop method.

See Also:



305
306
307
 
# File 'lib/fractor/supervisor.rb', line 305

def start
  run
end

#start_workers ⇒ Object

Starts the worker Ractors for all worker pools.



188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
 
# File 'lib/fractor/supervisor.rb', line 188

def start_workers
  # Capture debug flag for Ractor isolation (Ruby 4.0)
  # Pass as parameter to avoid isolation error
  debug_mode = @debug

  # Create a wakeup Ractor for unblocking Ractor.select
  if Fractor::RUBY_4_0_OR_HIGHER
    # In Ruby 4.0, wakeup uses ports too
    @wakeup_port = Ractor::Port.new
    @wakeup_ractor = Ractor.new(@wakeup_port, debug_mode) do |port, debug|
      puts "Wakeup Ractor started" if debug
      loop do
        msg = Ractor.receive
        puts "Wakeup Ractor received: #{msg.inspect}" if debug
        if %i[wakeup shutdown].include?(msg)
          port << { type: :wakeup, message: msg }
          break if msg == :shutdown
        end
      end
      puts "Wakeup Ractor shutting down" if debug
    end
  else
    @wakeup_ractor = Ractor.new(debug_mode) do |debug|
      puts "Wakeup Ractor started" if debug
      loop do
        msg = Ractor.receive
        puts "Wakeup Ractor received: #{msg.inspect}" if debug
        if %i[wakeup shutdown].include?(msg)
          Ractor.yield({ type: :wakeup, message: msg })
          break if msg == :shutdown
        end
      end
      puts "Wakeup Ractor shutting down" if debug
    end
  end

  # Add wakeup ractor to the map with a special marker
  @ractors_map[@wakeup_ractor] = :wakeup

  @worker_pools.each do |pool|
    worker_class = pool[:worker_class]
    num_workers = pool[:num_workers]

    pool[:workers] = (1..num_workers).map do |i|
      # In Ruby 4.0, create a response port for each worker
      response_port = if Fractor::RUBY_4_0_OR_HIGHER
                        Ractor::Port.new
                      end

      # Use the factory method to create the appropriate implementation
      wrapped_ractor = WrappedRactor.create(
        "worker #{worker_class}:#{i}",
        worker_class,
        response_port: response_port,
      )
      wrapped_ractor.start # Start the underlying Ractor
      # Map the actual Ractor object to the WrappedRactor instance
      @ractors_map[wrapped_ractor.ractor] = wrapped_ractor if wrapped_ractor.ractor
      wrapped_ractor
    end.compact
  end

  # Flatten all workers for easier access
  @workers = @worker_pools.flat_map { |pool| pool[:workers] }
  @ractors_map.compact! # Ensure map doesn't contain nil keys/values

  # Update work distribution manager's workers reference
  # This is critical because @workers was reassigned, and WorkDistributionManager
  # needs the updated reference to properly track idle workers
  @work_distribution_manager.update_workers(@workers)

  # Mark all workers as idle initially so they can receive work
  # This is critical for Ruby 4.0 where workers don't send :initialize messages
  @workers.each do |worker|
    @work_distribution_manager.mark_worker_idle(worker)
  end

  return unless @debug

  puts "Workers started: #{@workers.size} active across #{@worker_pools.size} pools."
  puts "All workers marked as idle and ready for work."
end

#stop ⇒ Object

Stop the supervisor (for continuous mode)



356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
 
# File 'lib/fractor/supervisor.rb', line 356

def stop
  puts "Stopping supervisor..." if @debug

  # Initiate shutdown in main loop handler first, so it continues
  # processing shutdown acknowledgments even after @running = false
  @main_loop_handler&.initiate_shutdown

  @running = false

  # CRITICAL: Send immediate wakeup signal to unblock main loop from Ractor.select
  # This is especially important for Ruby 3.4+ where Ractor.select may block indefinitely
  # without periodic checks of @running. The timer thread might take time to exit,
  # so we send the signal here immediately.
  if @wakeup_ractor
    begin
      @wakeup_ractor.send(:shutdown)
    rescue StandardError => e
      puts "Error sending shutdown to wakeup ractor: #{e.message}" if @debug
    end
  end

  # Update shutdown handler with current references before shutdown
  @shutdown_handler.instance_variable_set(:@workers, @workers)
  @shutdown_handler.instance_variable_set(:@wakeup_ractor, @wakeup_ractor)
  @shutdown_handler.instance_variable_set(:@timer_thread, @timer_thread)
  @shutdown_handler.instance_variable_set(:@performance_monitor,
                                          @performance_monitor)

  # Send shutdown signals but don't wait for workers to close
  # The caller (e.g., ContinuousServer) should wait for the main loop thread
  @shutdown_handler.shutdown
end

#workers_status ⇒ Object

Get current worker status Returns a hash with worker statistics



536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
 
# File 'lib/fractor/supervisor.rb', line 536

def workers_status
  status = @work_distribution_manager.status_summary
  idle_count = status[:idle]
  busy_count = status[:busy]

  {
    total: @workers.size,
    idle: idle_count,
    busy: busy_count,
    pools: @worker_pools.map do |pool|
      {
        worker_class: pool[:worker_class].name,
        num_workers: pool[:num_workers],
        workers: pool[:workers].map do |w|
          {
            name: w.name,
            idle: @work_distribution_manager.idle_workers_list.include?(w),
          }
        end,
      }
    end,
  }
end