Class: Dynflow::Director::ExecutionPlanManager

Inherits:
Object
  • Object
show all
Includes:
Algebrick::Matching, Algebrick::TypeCheck
Defined in:
lib/dynflow/director/execution_plan_manager.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(world, execution_plan, future) ⇒ ExecutionPlanManager

Returns a new instance of ExecutionPlanManager.



11
12
13
14
15
16
17
18
19
20
21
22
 
# File 'lib/dynflow/director/execution_plan_manager.rb', line 11

def initialize(world, execution_plan, future)
  @world                 = Type! world, World
  @execution_plan        = Type! execution_plan, ExecutionPlan
  @future                = Type! future, Concurrent::Promises::ResolvableFuture
  @running_steps_manager = RunningStepsManager.new(world)
  @halted                = false

  unless [:planned, :paused].include? execution_plan.state
    raise "execution_plan is not in pending or paused state, it's #{execution_plan.state}"
  end
  execution_plan.update_state(:running)
end

Instance Attribute Details

#execution_plan ⇒ Object (readonly)

Returns the value of attribute execution_plan.



9
10
11
 
# File 'lib/dynflow/director/execution_plan_manager.rb', line 9

def execution_plan
  @execution_plan
end

#future ⇒ Object (readonly)

Returns the value of attribute future.



9
10
11
 
# File 'lib/dynflow/director/execution_plan_manager.rb', line 9

def future
  @future
end

Instance Method Details

#done? ⇒ Boolean

Returns:

  • (Boolean) 


80
81
82
 
# File 'lib/dynflow/director/execution_plan_manager.rb', line 80

def done?
  @halted || (!@run_manager || @run_manager.done?) && (!@finalize_manager || @finalize_manager.done?)
end

#event(event) ⇒ Object 



72
73
74
75
76
77
78
 
# File 'lib/dynflow/director/execution_plan_manager.rb', line 72

def event(event)
  Type! event, Event
  unless event.execution_plan_id == @execution_plan.id
    raise "event #{event.inspect} doesn't belong to plan #{@execution_plan.id}"
  end
  @running_steps_manager.event(event)
end

#halt ⇒ Object 



29
30
31
32
 
# File 'lib/dynflow/director/execution_plan_manager.rb', line 29

def halt
  @halted = true
  @running_steps_manager.terminate
end

#prepare_next_step(step) ⇒ Object 



40
41
42
43
44
 
# File 'lib/dynflow/director/execution_plan_manager.rb', line 40

def prepare_next_step(step)
  StepWorkItem.new(execution_plan.id, step, step.queue, @world.id).tap do |work|
    @running_steps_manager.add(step, work)
  end
end

#restart ⇒ Object 



34
35
36
37
38
 
# File 'lib/dynflow/director/execution_plan_manager.rb', line 34

def restart
  @run_manager = nil
  @finalize_manager = nil
  start
end

#start ⇒ Object 



24
25
26
27
 
# File 'lib/dynflow/director/execution_plan_manager.rb', line 24

def start
  raise "The future was already set" if @future.resolved?
  start_run or start_finalize or finish
end

#terminate ⇒ Object 



84
85
86
 
# File 'lib/dynflow/director/execution_plan_manager.rb', line 84

def terminate
  @running_steps_manager.terminate
end

#what_is_next(work) ⇒ Array<WorkItem>

Returns of Work items to continue with.

Returns:

  • (Array<WorkItem>) —

    of Work items to continue with



47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/dynflow/director/execution_plan_manager.rb', line 47

def what_is_next(work)
  Type! work, WorkItem

  case work
  when StepWorkItem
    step = work.step
    update_steps([step])
    suspended, work = @running_steps_manager.done(step)
    work = compute_next_from_step(step) unless suspended
    work
  when FinalizeWorkItem
    if work.finalize_steps_data
      steps = work.finalize_steps_data.map do |step_data|
        Serializable.from_hash(step_data, execution_plan.id, @world)
      end
      update_steps(steps)
    end
    raise "Finalize work item without @finalize_manager ready" unless @finalize_manager
    @finalize_manager.done!
    finish
  else
    raise "Unexpected work #{work}"
  end
end