Class: Fractor::WorkDistributionManager

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

Overview

Manages work distribution to workers in a Supervisor. Responsible for tracking idle workers and assigning work from the queue.

This class extracts work distribution logic from Supervisor to follow the Single Responsibility Principle.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(work_queue, workers, ractors_map, debug: false, continuous_mode: false, performance_monitor: nil) ⇒ WorkDistributionManager

Returns a new instance of WorkDistributionManager.



12
13
14
15
16
17
18
19
20
21
22
 
# File 'lib/fractor/work_distribution_manager.rb', line 12

def initialize(work_queue, workers, ractors_map, debug: false,
               continuous_mode: false, performance_monitor: nil)
  @work_queue = work_queue
  @workers = workers
  @ractors_map = ractors_map
  @debug = debug
  @continuous_mode = continuous_mode
  @performance_monitor = performance_monitor
  @idle_workers = []
  @work_start_times = {}
end

Instance Attribute Details

#idle_workers ⇒ Object (readonly)

Returns the value of attribute idle_workers.



10
11
12
 
# File 'lib/fractor/work_distribution_manager.rb', line 10

def idle_workers
  @idle_workers
end

Instance Method Details

#assign_work_to_worker(wrapped_ractor) ⇒ Boolean

Assign work to a specific worker if work is available.

Parameters:

  • wrapped_ractor (WrappedRactor) —

    The worker to assign work to

Returns:

  • (Boolean) —

    true if work was sent, false otherwise



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
 
# File 'lib/fractor/work_distribution_manager.rb', line 36

def assign_work_to_worker(wrapped_ractor)
  # Ensure the wrapped_ractor instance is valid and its underlying ractor is not closed
  if wrapped_ractor && !wrapped_ractor.closed?
    if @work_queue.empty?
      puts "Work queue empty. Not sending new work to Ractor #{wrapped_ractor.name}." if @debug
      false
    else
      work_item = @work_queue.pop # Now directly a Work object

      # Track start time for performance monitoring
      if @performance_monitor
        @work_start_times[work_item.object_id] = Time.now
      end

      puts "Sending next work #{work_item.inspect} to Ractor: #{wrapped_ractor.name}." if @debug
      wrapped_ractor.send(work_item) # Send the Work object
      puts "Work sent to #{wrapped_ractor.name}." if @debug

      # Remove from idle workers list since it's now busy
      @idle_workers.delete(wrapped_ractor)
      true
    end
  else
    puts "Attempted to send work to an invalid or closed Ractor: #{wrapped_ractor&.name || 'unknown'}." if @debug
    # Remove from map if found but closed
    if wrapped_ractor && @ractors_map.key?(wrapped_ractor.ractor)
      @ractors_map.delete(wrapped_ractor.ractor)
    end
    false
  end
end

#busy_count ⇒ Integer

Get the count of busy workers.

Returns:

  • (Integer) —

    Number of busy workers



123
124
125
 
# File 'lib/fractor/work_distribution_manager.rb', line 123

def busy_count
  @workers.size - @idle_workers.size
end

#busy_workers_list ⇒ Array<WrappedRactor>

Get the list of busy (processing) workers.

Returns:



109
110
111
 
# File 'lib/fractor/work_distribution_manager.rb', line 109

def busy_workers_list
  @workers.reject { |w| @idle_workers.include?(w) }
end

#clear_work_start_times ⇒ Object

Clear all tracked work start times. Useful for cleanup or testing.



137
138
139
 
# File 'lib/fractor/work_distribution_manager.rb', line 137

def clear_work_start_times
  @work_start_times.clear
end

#distribute_to_idle_workers ⇒ Integer

Distribute available work to idle workers. Useful when new work is added to the queue.

Returns:

  • (Integer) —

    Number of workers that received work



87
88
89
90
91
92
93
94
95
96
97
 
# File 'lib/fractor/work_distribution_manager.rb', line 87

def distribute_to_idle_workers
  distributed = 0
  while !@work_queue.empty? && !@idle_workers.empty?
    worker = @idle_workers.shift
    if assign_work_to_worker(worker)
      puts "Sent work to idle worker #{worker.name}" if @debug
      distributed += 1
    end
  end
  distributed
end

#get_work_start_time(work_object_id) ⇒ Time?

Get the work start time for a specific work item.

Parameters:

  • work_object_id (Integer) —

    The object_id of the work item

Returns:

  • (Time, nil) —

    The start time, or nil if not found



131
132
133
 
# File 'lib/fractor/work_distribution_manager.rb', line 131

def get_work_start_time(work_object_id)
  @work_start_times.delete(work_object_id)
end

#idle_count ⇒ Integer

Get the count of idle workers.

Returns:

  • (Integer) —

    Number of idle workers



116
117
118
 
# File 'lib/fractor/work_distribution_manager.rb', line 116

def idle_count
  @idle_workers.size
end

#idle_workers_list ⇒ Array<WrappedRactor>

Get the list of idle (available) workers.

Returns:



102
103
104
 
# File 'lib/fractor/work_distribution_manager.rb', line 102

def idle_workers_list
  @idle_workers.dup
end

#mark_worker_busy(wrapped_ractor) ⇒ Object

Mark a worker as busy (not available for work).

Parameters:



79
80
81
 
# File 'lib/fractor/work_distribution_manager.rb', line 79

def mark_worker_busy(wrapped_ractor)
  @idle_workers.delete(wrapped_ractor)
end

#mark_worker_idle(wrapped_ractor) ⇒ Object

Mark a worker as idle (available for work).

Parameters:



71
72
73
74
 
# File 'lib/fractor/work_distribution_manager.rb', line 71

def mark_worker_idle(wrapped_ractor)
  @idle_workers << wrapped_ractor unless @idle_workers.include?(wrapped_ractor)
  puts "Worker #{wrapped_ractor.name} marked as idle." if @debug
end

#status_summary ⇒ Hash

Get current worker status summary.

Returns:

  • (Hash) —

    Worker status summary with :idle and :busy counts



144
145
146
147
148
149
 
# File 'lib/fractor/work_distribution_manager.rb', line 144

def status_summary
  {
    idle: @idle_workers.size,
    busy: @workers.size - @idle_workers.size,
  }
end

#update_workers(workers) ⇒ Object

Update the workers reference after workers are created This is needed when @workers is reassigned in Supervisor.start_workers

Parameters:



28
29
30
 
# File 'lib/fractor/work_distribution_manager.rb', line 28

def update_workers(workers)
  @workers = workers
end