Class: Taski::Execution::Executor

Inherits:
Object
  • Object
show all
Defined in:
lib/taski/execution/executor.rb

Overview

Orchestrates run (Fiber-based) and clean (direct) phases of task execution. Delegates to Scheduler (state tracking / advisory proposals), WorkerPool (worker threads), and ExecutionFacade (observer notifications).

Task execution is driven by the Fiber pull model — tasks start only when requested via Fiber.yield FiberProtocol::NeedDep. Scheduler may propose tasks, but Executor/Wrapper can reject proposals not backed by actual Fiber requests.

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(registry:, execution_facade:, worker_count: nil) ⇒ Executor

Returns a new instance of Executor.



25
26
27
28
29
30
31
 
# File 'lib/taski/execution/executor.rb', line 25

def initialize(registry:, execution_facade:, worker_count: nil)
  @registry = registry
  @completion_queue = Queue.new
  @execution_facade = execution_facade
  @scheduler = Scheduler.new
  @effective_worker_count = worker_count || Taski.args_worker_count
end

Class Method Details

.execute(root_task_class, registry:, execution_facade:) ⇒ Object 



16
17
18
 
# File 'lib/taski/execution/executor.rb', line 16

def execute(root_task_class, registry:, execution_facade:)
  new(registry: registry, execution_facade: execution_facade).execute(root_task_class)
end

.execute_clean(root_task_class, registry:, execution_facade:) ⇒ Object 



20
21
22
 
# File 'lib/taski/execution/executor.rb', line 20

def execute_clean(root_task_class, registry:, execution_facade:)
  new(registry: registry, execution_facade: execution_facade).execute_clean(root_task_class)
end

Instance Method Details

#execute(root_task_class) ⇒ Object 



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
 
# File 'lib/taski/execution/executor.rb', line 33

def execute(root_task_class)
  start_time = Time.now

  log_execution_started(root_task_class)

  @scheduler.load_graph(@execution_facade.dependency_graph, root_task_class)

  with_display_lifecycle(root_task_class) do
    @worker_pool = WorkerPool.new(
      registry: @registry,
      execution_facade: @execution_facade,
      worker_count: @effective_worker_count,
      completion_queue: @completion_queue
    )

    @worker_pool.start

    enqueue_root_if_needed(root_task_class)

    run_main_loop(root_task_class)

    @worker_pool.shutdown

    notify_skipped_tasks
  end

  log_execution_completed(root_task_class, start_time)

  raise_if_any_failures
end

#execute_clean(root_task_class) ⇒ Object 



64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
 
# File 'lib/taski/execution/executor.rb', line 64

def execute_clean(root_task_class)
  @scheduler.load_graph(@execution_facade.dependency_graph, root_task_class)
  @scheduler.build_reverse_dependency_graph

  with_display_lifecycle(root_task_class) do
    @worker_pool = WorkerPool.new(
      registry: @registry,
      execution_facade: @execution_facade,
      worker_count: @effective_worker_count,
      completion_queue: @completion_queue
    )
    @worker_pool.start
    enqueue_ready_clean_tasks
    run_clean_main_loop
    @worker_pool.shutdown
  end

  raise_if_any_clean_failures
end