Class: JobWorkflow::Runner

Inherits:
Object
  • Object
show all
Defined in:
lib/job_workflow/runner.rb,
sig/generated/job_workflow/runner.rbs

Overview

rubocop:disable Metrics/ClassLength

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(context:) ⇒ Runner

: (context: Context) -> void

Parameters:



8
9
10
11
 
# File 'lib/job_workflow/runner.rb', line 8

def initialize(context:)
  @context = context
  @job = context._job || (raise "current job is not set in context")
end

Instance Attribute Details

#context ⇒ Context (readonly)

rubocop:disable Metrics/ClassLength

Returns:



5
6
7
 
# File 'lib/job_workflow/runner.rb', line 5

def context
  @context
end

#job ⇒ _JobInterface (readonly)

Signature:

  • _JobInterface

Returns:

  • (_JobInterface) 


28
29
30
 
# File 'lib/job_workflow/runner.rb', line 28

def job
  @job
end

Instance Method Details

#add_task_output(ctx:, task:, data:, each_index:) ⇒ void

This method returns an undefined value.

: (ctx: Context, task: Task, each_index: Integer, data: untyped) -> void

Parameters:

  • ctx: (Context)
  • task: (Task)
  • each_index: (Integer)
  • data: (Object) 


191
192
193
194
195
 
# File 'lib/job_workflow/runner.rb', line 191

def add_task_output(ctx:, task:, data:, each_index:)
  return if task.output.empty?

  ctx._add_task_output(TaskOutput.from_task(task:, each_index:, data:))
end

#decode_task_cursor(task, task_cursor) ⇒ [ Integer?, untyped ]

: (Task, untyped) -> [Integer?, untyped]

Parameters:

Returns:

  • ([ Integer?, untyped ]) 


84
85
86
87
88
89
90
91
92
93
94
95
 
# File 'lib/job_workflow/runner.rb', line 84

def decode_task_cursor(task, task_cursor)
  return [nil, task_cursor] unless task.each?
  return [task_cursor, nil] if task_cursor.is_a?(Integer)

  if task_cursor.is_a?(Hash) && task_cursor[Context::EACH_TASK_CURSOR_MARKER]
    return [task_cursor.fetch("index"), task_cursor.fetch("cursor")]
  end

  raise "invalid each task cursor: #{task_cursor.inspect}" unless task_cursor.nil?

  [nil, nil]
end

#enqueue_task(task) ⇒ void

This method returns an undefined value.

: (Task) -> void

Parameters:



135
136
137
138
139
140
141
142
143
 
# File 'lib/job_workflow/runner.rb', line 135

def enqueue_task(task)
  sub_jobs = context._with_each_value(task).map do |ctx|
    SubTaskJob.from_parent_context(context: ctx)
  end
  persist_current_job_context
  ActiveJob.perform_all_later(sub_jobs)
  context.job_status.update_task_job_statuses_from_jobs(task_name: task.task_name, jobs: sub_jobs)
  Instrumentation.notify_task_enqueue(job, task, sub_jobs.size)
end

#hooks ⇒ HookRegistry

: () -> HookRegistry

Returns:



41
42
43
 
# File 'lib/job_workflow/runner.rb', line 41

def hooks
  workflow.hooks
end

#persist_current_job_context ⇒ void

This method returns an undefined value.

: () -> void



205
206
207
 
# File 'lib/job_workflow/runner.rb', line 205

def persist_current_job_context
  QueueAdapter.current.persist_job_context(job)
end

#poll_until_complete_or_reschedule(waiting_task, dependent_task, step) ⇒ void

This method returns an undefined value.

: (Task, Task, ActiveJob::Continuation::Step) -> void

Parameters:

  • (Task)
  • (Task)
  • (ActiveJob::Continuation::Step) 


160
161
162
163
164
165
166
167
168
169
170
171
172
173
 
# File 'lib/job_workflow/runner.rb', line 160

def poll_until_complete_or_reschedule(waiting_task, dependent_task, step)
  poll_state = { count: 0, started_at: Time.current }
  dependency_wait = waiting_task.dependency_wait

  loop do
    step.checkpoint!
    context.job_status.update_task_job_statuses_from_db(dependent_task.task_name)
    break if context.job_status.needs_waiting?(dependent_task.task_name)

    poll_state[:count] += 1
    reschedule_if_needed(dependent_task, dependency_wait, poll_state)
    sleep dependency_wait.poll_interval
  end
end

#reschedule_if_needed(dependent_task, dependency_wait, poll_state) ⇒ void

This method returns an undefined value.

: (Task, TaskDependencyWait, Hash[Symbol, untyped]) -> void

Parameters:



176
177
178
179
180
181
182
183
184
185
186
187
188
 
# File 'lib/job_workflow/runner.rb', line 176

def reschedule_if_needed(dependent_task, dependency_wait, poll_state)
  return if dependency_wait.polling_only?
  return if dependency_wait.polling_keep?(poll_state[:started_at])
  return unless QueueAdapter.current.reschedule_job(job, dependency_wait.reschedule_delay)

  Instrumentation.notify_dependent_reschedule(
    job,
    dependent_task,
    dependency_wait.reschedule_delay,
    poll_state[:count]
  )
  throw :rescheduled
end

#run ⇒ void

This method returns an undefined value.

: () -> void



14
15
16
17
18
19
20
21
22
23
24
 
# File 'lib/job_workflow/runner.rb', line 14

def run
  task = context._task_context.task
  if !task.nil? && context.sub_job?
    job.step(task.task_name) { |step| run_task(task, step:) }
    persist_current_job_context
    return
  end

  catch(:rescheduled) { run_workflow }
  persist_current_job_context
end

#run_around_hooks(task, ctx, around_hooks) { ... } ⇒ void

This method returns an undefined value.

: (Task, Context, Array) { () -> void } -> void

Parameters:

Yields:

Yield Returns:

  • (void) 


119
120
121
122
123
124
125
126
127
 
# File 'lib/job_workflow/runner.rb', line 119

def run_around_hooks(task, ctx, around_hooks, &)
  return yield if around_hooks.empty?

  hook = around_hooks.first
  remaining = around_hooks[1..] || []
  callable = TaskCallable.new { run_around_hooks(task, ctx, remaining, &) }
  hook.block.call(ctx, callable)
  raise TaskCallable::NotCalledError, task.task_name unless callable.called?
end

#run_each_task(task, ctx, step:, cursor: nil) ⇒ void

This method returns an undefined value.

: (Task, Context, step: ActiveJob::Continuation::Step, ?cursor: untyped) -> void

Parameters:

  • (Task)
  • (Context)
  • step: (ActiveJob::Continuation::Step)
  • cursor: (Object) (defaults to: nil) 


98
99
100
101
102
103
104
105
106
107
108
109
 
# File 'lib/job_workflow/runner.rb', line 98

def run_each_task(task, ctx, step:, cursor: nil)
  Instrumentation.instrument_task(job, task, ctx) do
    ctx._with_current_step(step, cursor:) do
      ctx._with_task_throttle do
        run_hooks(task, ctx) do
          data = task.block.call(ctx)
          add_task_output(ctx:, task:, each_index: ctx._task_context.index, data:)
        end
      end
    end
  end
end

#run_error_hooks(task, ctx, error) ⇒ void

This method returns an undefined value.

: (Task, Context, StandardError) -> void

Parameters:



130
131
132
 
# File 'lib/job_workflow/runner.rb', line 130

def run_error_hooks(task, ctx, error)
  hooks.error_hooks_for(task.task_name).each { |hook| hook.block.call(ctx, error, task) }
end

#run_hooks(task, ctx) { ... } ⇒ void

This method returns an undefined value.

: (Task, Context) { () -> void } -> void

Parameters:

Yields:

Yield Returns:

  • (void) 


112
113
114
115
116
 
# File 'lib/job_workflow/runner.rb', line 112

def run_hooks(task, ctx, &)
  hooks.before_hooks_for(task.task_name).each { |hook| hook.block.call(ctx) }
  run_around_hooks(task, ctx, hooks.around_hooks_for(task.task_name), &)
  hooks.after_hooks_for(task.task_name).each { |hook| hook.block.call(ctx) }
end

#run_task(task, step:) ⇒ void

This method returns an undefined value.

: (Task, step: ActiveJob::Continuation::Step) -> void

Parameters:

  • (Task)
  • step: (ActiveJob::Continuation::Step) 


67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
 
# File 'lib/job_workflow/runner.rb', line 67

def run_task(task, step:) # rubocop:disable Metrics/AbcSize, Metrics/MethodLength
  context._load_parent_task_output
  start_index, task_cursor = decode_task_cursor(task, step.cursor)

  context._with_each_value(task, start_index:).each do |ctx|
    iteration_cursor = task_cursor
    iteration_cursor = nil if task.each? && (task_cursor.nil? || start_index != ctx._task_context.index)

    run_each_task(task, ctx, step:, cursor: iteration_cursor)
    step.set!(ctx._task_context.index + 1) if task.each?
  rescue StandardError => e
    run_error_hooks(task, ctx, e)
    raise
  end
end

#run_workflow ⇒ void

This method returns an undefined value.

: () -> void



46
47
48
49
50
51
52
53
54
55
56
57
 
# File 'lib/job_workflow/runner.rb', line 46

def run_workflow
  Instrumentation.instrument_workflow(job) do
    tasks.each do |task|
      next if skip_task?(task)

      job.step(task.task_name) do |step|
        wait_for_dependent_tasks(task, step)
        task.enqueue.should_enqueue?(context) ? enqueue_task(task) : run_task(task, step:)
      end
    end
  end
end

#skip_task?(task) ⇒ Boolean

: (Task) -> bool

Parameters:

Returns:

  • (Boolean) 


60
61
62
63
64
 
# File 'lib/job_workflow/runner.rb', line 60

def skip_task?(task)
  result = !task.condition.call(context)
  Instrumentation.notify_task_skip(job, task, "condition_not_met") if result
  result
end

#tasks ⇒ Array[Task]

: () -> Array

Returns:



36
37
38
 
# File 'lib/job_workflow/runner.rb', line 36

def tasks
  workflow.tasks
end

#update_task_outputs(task) ⇒ void

This method returns an undefined value.

: (Task) -> void

Parameters:



198
199
200
201
202
 
# File 'lib/job_workflow/runner.rb', line 198

def update_task_outputs(task)
  finished_job_ids = context.job_status.finished_job_ids(task_name: task.task_name)
  context_data_list = QueueAdapter.current.fetch_job_contexts(finished_job_ids)
  context.output.update_task_outputs_from_contexts(context_data_list, context.workflow)
end

#wait_for_dependent_tasks(waiting_task, step) ⇒ void

This method returns an undefined value.

: (Task, ActiveJob::Continuation::Step) -> void

Parameters:

  • (Task)
  • (ActiveJob::Continuation::Step) 


146
147
148
149
150
151
152
153
154
155
156
157
 
# File 'lib/job_workflow/runner.rb', line 146

def wait_for_dependent_tasks(waiting_task, step)
  waiting_task.depends_on.each do |dependent_task_name|
    dependent_task = workflow.fetch_task(dependent_task_name)
    next if dependent_task.nil? || context.job_status.needs_waiting?(dependent_task.task_name)

    Instrumentation.instrument_dependent_wait(job, dependent_task) do
      poll_until_complete_or_reschedule(waiting_task, dependent_task, step)
    end

    update_task_outputs(dependent_task)
  end
end

#workflow ⇒ Workflow

: () -> Workflow

Returns:



31
32
33
 
# File 'lib/job_workflow/runner.rb', line 31

def workflow
  context.workflow
end