Class: Taski::Execution::Scheduler

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

Overview

Scheduler manages task dependency state and determines execution order. Both run and clean phases use the same unified state set: :pending, :running, :completed, :failed, :skipped.

State Transitions

Run phase: pending → running → completed | failed pending → skipped (when a dependency fails) Clean phase: pending → running → completed

Responsibilities

  • Load pre-built dependency graph from Executor
  • Track task states: pending, running, completed
  • Determine which tasks are ready to execute (all dependencies completed)
  • Provide next_ready_tasks for the Executor's event loop
  • Build reverse dependency graph for clean operations
  • Track clean states independently from run states
  • Provide next_ready_clean_tasks for reverse dependency order execution

API

Run operations:

Clean operations:

Thread Safety

Scheduler is only accessed from the main thread in Executor, so no synchronization is needed. The Executor serializes all access to the Scheduler through its event loop.

Constant Summary collapse

STATE_PENDING =

Unified task execution states (used by both run and clean phases)

:pending
STATE_RUNNING = 
:running
STATE_COMPLETED = 
:completed
STATE_FAILED = 
:failed
STATE_SKIPPED = 
:skipped

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Scheduler

Initializes internal data structures used to track normal and clean task execution.



58
59
60
61
62
63
64
65
66
67
68
69
 
# File 'lib/taski/execution/scheduler.rb', line 58

def initialize
  # Run execution state
  @dependencies = {}
  @task_states = {}
  @finished_tasks = Set.new
  @run_reverse_deps = {}

  # Clean execution state (independent tracking, same state values)
  @reverse_dependencies = {}
  @clean_task_states = {}
  @clean_finished_tasks = Set.new
end

Instance Method Details

#build_reverse_dependency_graph ⇒ Object

Build reverse dependency graph for clean operations. Clean operations run in reverse order: if A depends on B, then B must be cleaned after A (so A→[B] in reverse graph means B depends on A's clean).

Requires load_graph to have been called first to populate @dependencies. Also initializes clean states for all tasks to STATE_PENDING.



231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
 
# File 'lib/taski/execution/scheduler.rb', line 231

def build_reverse_dependency_graph
  # Clear previous clean state
  @reverse_dependencies.clear
  @clean_task_states.clear
  @clean_finished_tasks.clear

  # Initialize all tasks with empty reverse dependency sets
  @dependencies.each_key do |task_class|
    @reverse_dependencies[task_class] = Set.new
    @clean_task_states[task_class] = STATE_PENDING
  end

  # Build reverse mappings: if A depends on B, then B→[A] in reverse graph
  # This means B's clean depends on A's clean completing first
  @dependencies.each do |task_class, deps|
    deps.each do |dep_class|
      @reverse_dependencies[dep_class].add(task_class)
    end
  end
end

#clean_finished?(task_class) ⇒ Boolean

Check if a task's clean is completed.

Parameters:

  • task_class (Class) —

    The task class to check.

Returns:

  • (Boolean) —

    true if the task's clean is completed, false otherwise.



295
296
297
 
# File 'lib/taski/execution/scheduler.rb', line 295

def clean_finished?(task_class)
  @clean_finished_tasks.include?(task_class)
end

#finished?(task_class) ⇒ Boolean

Check if a task is finished (completed or failed).

Parameters:

  • task_class (Class) —

    The task class to check

Returns:

  • (Boolean) —

    true if the task is finished



149
150
151
 
# File 'lib/taski/execution/scheduler.rb', line 149

def finished?(task_class)
  @finished_tasks.include?(task_class)
end

#load_graph(dependency_graph, root_task_class) ⇒ Object

Load dependency graph from a pre-built DependencyGraph. Populates internal state with all tasks and their dependencies via BFS from root.

Parameters:



76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
 
# File 'lib/taski/execution/scheduler.rb', line 76

def load_graph(dependency_graph, root_task_class)
  # @type var queue: Array[singleton(Taski::Task)]
  queue = [root_task_class]

  while (task_class = queue.shift)
    next if @task_states.key?(task_class)

    deps = dependency_graph.dependencies_for(task_class)
    @dependencies[task_class] = deps.dup
    @task_states[task_class] = STATE_PENDING
    @run_reverse_deps[task_class] ||= Set.new

    deps.each do |dep|
      @run_reverse_deps[dep] ||= Set.new
      @run_reverse_deps[dep].add(task_class)
      log_dependency_resolved(task_class, dep)
      queue << dep
    end
  end
end

#mark_clean_completed(task_class) ⇒ Object

Mark a task as clean completed.

Parameters:

  • task_class (Class) —

    The task class to mark as clean completed.



277
278
279
280
 
# File 'lib/taski/execution/scheduler.rb', line 277

def mark_clean_completed(task_class)
  @clean_task_states[task_class] = STATE_COMPLETED
  @clean_finished_tasks.add(task_class)
end

#mark_clean_failed(task_class) ⇒ Object

Mark a task as clean failed. Adds to @clean_finished_tasks so dependents are not blocked.

Parameters:

  • task_class (Class) —

    The task class to mark as clean failed.



286
287
288
289
 
# File 'lib/taski/execution/scheduler.rb', line 286

def mark_clean_failed(task_class)
  @clean_task_states[task_class] = STATE_FAILED
  @clean_finished_tasks.add(task_class)
end

#mark_clean_running(task_class) ⇒ Object

Mark a task as running for clean execution.

Parameters:

  • task_class (Class) —

    The task class to mark as running for clean.



270
271
272
 
# File 'lib/taski/execution/scheduler.rb', line 270

def mark_clean_running(task_class)
  @clean_task_states[task_class] = STATE_RUNNING
end

#mark_completed(task_class) ⇒ Object

Mark a task as completed.

Parameters:

  • task_class (Class) —

    The task class to mark



122
123
124
125
 
# File 'lib/taski/execution/scheduler.rb', line 122

def mark_completed(task_class)
  @task_states[task_class] = STATE_COMPLETED
  @finished_tasks.add(task_class)
end

#mark_failed(task_class) ⇒ Object

Mark a task as failed. Failed tasks are added to the finished set so dependents can proceed (they will be skipped by the Executor's skip_pending_dependents).

Parameters:

  • task_class (Class) —

    The task class to mark



132
133
134
135
 
# File 'lib/taski/execution/scheduler.rb', line 132

def mark_failed(task_class)
  @task_states[task_class] = STATE_FAILED
  @finished_tasks.add(task_class)
end

#mark_running(task_class) ⇒ Object

Mark a task as running (sent to worker pool). Prevents the task from being selected again by next_ready_tasks.

Parameters:

  • task_class (Class) —

    The task class to mark



115
116
117
 
# File 'lib/taski/execution/scheduler.rb', line 115

def mark_running(task_class)
  @task_states[task_class] = STATE_RUNNING
end

#mark_skipped(task_class) ⇒ Boolean

Mark a task as skipped (never executed). Only transitions from pending.

Parameters:

  • task_class (Class) —

    The task class to mark as skipped

Returns:

  • (Boolean) —

    true if the state was changed



182
183
184
185
186
 
# File 'lib/taski/execution/scheduler.rb', line 182

def mark_skipped(task_class)
  return false unless @task_states[task_class] == STATE_PENDING
  @task_states[task_class] = STATE_SKIPPED
  true
end

#never_started_task_classes ⇒ Array<Class>

Get task classes that were never executed (remained in STATE_PENDING). These are tasks discovered by the static dependency graph (via load_graph) but not reached at runtime — e.g., skipped due to conditional logic inside Task#run or because the root task completed before all statically-discovered tasks were needed.

Returns:

  • (Array<Class>) —

    Array of task classes still pending



174
175
176
 
# File 'lib/taski/execution/scheduler.rb', line 174

def never_started_task_classes
  @task_states.select { |_, state| state == STATE_PENDING }.keys
end

#next_ready_clean_tasks ⇒ Array<Class>

Get all tasks that are ready to clean. A task is ready to clean when it is pending and all its reverse dependencies (dependents) have completed their clean operation.

Returns:

  • (Array<Class>) —

    Array of task classes ready for clean execution.



257
258
259
260
261
262
263
264
265
 
# File 'lib/taski/execution/scheduler.rb', line 257

def next_ready_clean_tasks
  ready = []
  @clean_task_states.each_key do |task_class|
    next unless @clean_task_states[task_class] == STATE_PENDING
    next unless ready_to_clean?(task_class)
    ready << task_class
  end
  ready
end

#next_ready_tasks ⇒ Array<Class>

Get all tasks that are ready to execute. A task is ready when it is pending and all its dependencies are completed.

Returns:

  • (Array<Class>) —

    Array of task classes ready for execution



101
102
103
104
105
106
107
108
109
 
# File 'lib/taski/execution/scheduler.rb', line 101

def next_ready_tasks
  ready = []
  @task_states.each_key do |task_class|
    next unless @task_states[task_class] == STATE_PENDING
    next unless ready_to_execute?(task_class)
    ready << task_class
  end
  ready
end

#pending?(task_class) ⇒ Boolean

Check if a task is in pending state.

Parameters:

  • task_class (Class) —

    The task class to check

Returns:

  • (Boolean) —

    true if the task is pending



141
142
143
 
# File 'lib/taski/execution/scheduler.rb', line 141

def pending?(task_class)
  @task_states[task_class] == STATE_PENDING
end

#pending_dependents_of(task_class) ⇒ Array<Class>

Find all pending tasks that transitively depend on the given task. Traverses the reverse dependency graph (run phase) using BFS. Only returns tasks in STATE_PENDING; running/completed tasks are traversed through but not included in results.

Parameters:

  • task_class (Class) —

    The task to find dependents of

Returns:

  • (Array<Class>) —

    Pending transitive dependents



202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
 
# File 'lib/taski/execution/scheduler.rb', line 202

def pending_dependents_of(task_class)
  result = []
  queue = [task_class]
  visited = Set.new([task_class])

  while (tc = queue.shift)
    dependents = @run_reverse_deps[tc] || Set.new
    dependents.each do |dep|
      next if visited.include?(dep)
      visited.add(dep)

      result << dep if @task_states[dep] == STATE_PENDING
      queue << dep
    end
  end

  result
end

#running_clean_tasks? ⇒ Boolean

Check if there are any running clean tasks.

Returns:

  • (Boolean) —

    true if at least one clean task is running, false otherwise.



302
303
304
 
# File 'lib/taski/execution/scheduler.rb', line 302

def running_clean_tasks?
  @clean_task_states.values.any? { |state| state == STATE_RUNNING }
end

#running_tasks? ⇒ Boolean

Check if there are any running tasks.

Returns:

  • (Boolean) —

    true if any task is running, false otherwise.



156
157
158
 
# File 'lib/taski/execution/scheduler.rb', line 156

def running_tasks?
  @task_states.values.any? { |state| state == STATE_RUNNING }
end

#skipped_count ⇒ Integer

Get the count of tasks in STATE_SKIPPED.

Returns:

  • (Integer) —

    Number of explicitly skipped tasks



191
192
193
 
# File 'lib/taski/execution/scheduler.rb', line 191

def skipped_count
  @task_states.count { |_, state| state == STATE_SKIPPED }
end

#task_count ⇒ Integer

Get the total number of tasks in the dependency graph.

Returns:

  • (Integer) —

    The number of tasks



163
164
165
 
# File 'lib/taski/execution/scheduler.rb', line 163

def task_count
  @task_states.size
end