Class: Taski::Execution::WorkerPool

Inherits:
Object
  • Object
show all
Includes:
TestHelper::WorkerPoolExtension
Defined in:
lib/taski/execution/worker_pool.rb

Overview

WorkerPool manages N threads, each with its own command Queue. Tasks are executed within Fibers on worker threads.

Fiber protocol supports two yield types (FiberProtocol Data classes):

  • StartDep(task_class) → non-blocking. Starts dep on another thread and resumes the Fiber immediately. Used for speculative prestart.
  • NeedDep(task_class, method) → blocking. Resolves dependency via TaskWrapper#request_value:
    • :completed → resume Fiber immediately with the value
    • :wait → park the Fiber (it will be resumed later via the thread's queue)
    • :start → start the dependency as a nested Fiber on the same thread

Worker threads process these commands (FiberProtocol Data classes):

  • Execute(task_class, wrapper) → create and drive a new Fiber
  • ExecuteClean(task_class, wrapper) → run clean directly (no Fiber)
  • Resume(fiber, value) → resume a parked Fiber with a value
  • ResumeError(fiber, error) → resume a parked Fiber with an error
  • :shutdown → exit the worker loop

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(registry:, execution_facade:, completion_queue:, worker_count: nil) ⇒ WorkerPool

Returns a new instance of WorkerPool.



32
33
34
35
36
37
38
39
40
41
42
43
44
45
 
# File 'lib/taski/execution/worker_pool.rb', line 32

def initialize(registry:, execution_facade:, completion_queue:, worker_count: nil)
  @registry = registry
  @execution_facade = execution_facade
  @worker_count = worker_count || Execution.default_worker_count
  @completion_queue = completion_queue
  @threads = []
  @thread_queues = []
  @next_thread_index = 0
  @fiber_contexts_mutex = Mutex.new
  @fiber_contexts = {}
  @task_start_times_mutex = Mutex.new
  @task_start_times = {}
  @enqueue_mutex = Mutex.new
end

Instance Attribute Details

#worker_count ⇒ Object (readonly)

Returns the value of attribute worker_count.



30
31
32
 
# File 'lib/taski/execution/worker_pool.rb', line 30

def worker_count
  @worker_count
end

Instance Method Details

#enqueue(task_class, wrapper) ⇒ Object

Round-robins across worker threads.



58
59
60
61
62
63
64
65
 
# File 'lib/taski/execution/worker_pool.rb', line 58

def enqueue(task_class, wrapper)
  @enqueue_mutex.synchronize do
    queue = @thread_queues[@next_thread_index % @worker_count]
    @next_thread_index += 1
    queue.push(FiberProtocol::Execute.new(task_class, wrapper))
    Taski::Logging.debug(Taski::Logging::Events::WORKER_POOL_ENQUEUED, task: task_class.name, thread_index: (@next_thread_index - 1) % @worker_count)
  end
end

#enqueue_clean(task_class, wrapper) ⇒ Object

Clean tasks run directly without Fiber wrapping.



68
69
70
71
72
73
74
 
# File 'lib/taski/execution/worker_pool.rb', line 68

def enqueue_clean(task_class, wrapper)
  @enqueue_mutex.synchronize do
    queue = @thread_queues[@next_thread_index % @worker_count]
    @next_thread_index += 1
    queue.push(FiberProtocol::ExecuteClean.new(task_class, wrapper))
  end
end

#shutdown ⇒ Object 



76
77
78
79
 
# File 'lib/taski/execution/worker_pool.rb', line 76

def shutdown
  @thread_queues.each { |q| q.push(:shutdown) }
  @threads.each(&:join)
end

#start ⇒ Object 



47
48
49
50
51
52
53
54
55
 
# File 'lib/taski/execution/worker_pool.rb', line 47

def start
  @worker_count.times do
    queue = Queue.new
    @thread_queues << queue
    thread = Thread.new(queue) { |q| worker_loop(q) }
    @threads << thread
    @registry.register_thread(thread)
  end
end