Class: Roast::ExecutionManager

Inherits:
Object
  • Object
show all
Includes:
SystemCogs::Call::Manager, SystemCogs::Map::Manager, SystemCogs::Repeat::Manager
Defined in:
lib/roast/execution_manager.rb

Overview

Context in which the ‘execute` block of a workflow is evaluated

Defined Under Namespace

Classes: ExecutionManagerAlreadyPreparedError, ExecutionManagerCurrentlyRunningError, ExecutionManagerError, ExecutionManagerNotPreparedError, ExecutionScopeDoesNotExistError, ExecutionScopeNotSpecifiedError, IllegalCogNameError, OutputsAlreadyDefinedError

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(cog_registry, config_manager, all_execution_procs, workflow_context, scope: nil, scope_value: nil, scope_index: 0) ⇒ ExecutionManager

: ( | Cog::Registry, | ConfigManager, | Hash[Symbol?, Array[^() -> void]], | WorkflowContext, | ?scope: Symbol?, | ?scope_value: untyped?, | ?scope_index: Integer | ) -> void



51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/roast/execution_manager.rb', line 51

def initialize(
  cog_registry,
  config_manager,
  all_execution_procs,
  workflow_context,
  scope: nil,
  scope_value: nil,
  scope_index: 0
)
  @cog_registry = cog_registry
  @config_manager = config_manager
  @all_execution_procs = all_execution_procs
  @workflow_context = workflow_context
  @scope = scope
  @scope_value = scope_value
  @scope_index = scope_index
  @cogs = Cog::Store.new #: Cog::Store
  @cog_stack = Cog::Stack.new #: Cog::Stack
  @execution_context = ExecutionContext.new #: ExecutionContext
  @cog_input_manager = CogInputManager.new(@cog_registry, @cogs, @workflow_context) #: CogInputManager
  @barrier = Async::Barrier.new #: Async::Barrier
  @final_output = nil #: untyped
  @final_output_computed = false #: bool
end

Instance Attribute Details

#final_output ⇒ Object (readonly)

: untyped



40
41
42
 
# File 'lib/roast/execution_manager.rb', line 40

def final_output
  @final_output
end

#scope ⇒ Object (readonly)

: Symbol?



31
32
33
 
# File 'lib/roast/execution_manager.rb', line 31

def scope
  @scope
end

#scope_index ⇒ Object (readonly)

: Integer



37
38
39
 
# File 'lib/roast/execution_manager.rb', line 37

def scope_index
  @scope_index
end

#scope_value ⇒ Object (readonly)

: untyped



34
35
36
 
# File 'lib/roast/execution_manager.rb', line 34

def scope_value
  @scope_value
end

#workflow_context ⇒ Object (readonly)

: WorkflowContext



28
29
30
 
# File 'lib/roast/execution_manager.rb', line 28

def workflow_context
  @workflow_context
end

Instance Method Details

#cog_input_context ⇒ Object

: () -> CogInputContext

Raises:



139
140
141
142
143
 
# File 'lib/roast/execution_manager.rb', line 139

def cog_input_context
  raise ExecutionManagerNotPreparedError unless prepared?

  @cog_input_manager.context
end

#prepare! ⇒ Object

: () -> void

Raises:



77
78
79
80
81
82
83
84
85
 
# File 'lib/roast/execution_manager.rb', line 77

def prepare!
  raise ExecutionManagerAlreadyPreparedError if preparing? || prepared?

  @preparing = true
  bind_outputs
  bind_registered_cogs
  my_execution_procs.each { |ep| @execution_context.instance_eval(&ep) }
  @prepared = true
end

#prepared? ⇒ Boolean

: () -> bool

Returns:

  • (Boolean) 


129
130
131
 
# File 'lib/roast/execution_manager.rb', line 129

def prepared?
  @prepared ||= false
end

#preparing? ⇒ Boolean

: () -> bool

Returns:

  • (Boolean) 


124
125
126
 
# File 'lib/roast/execution_manager.rb', line 124

def preparing?
  @preparing ||= false
end

#run! ⇒ Object

Raises:



87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
 
# File 'lib/roast/execution_manager.rb', line 87

def run!
  raise ExecutionManagerNotPreparedError unless prepared?
  raise ExecutionManagerCurrentlyRunningError if running?

  @running = true
  Sync do |sync_task|
    sync_task.annotate("ExecutionManager #{@scope}")
    TaskContext.begin_execution_manager(self)
    @cog_stack.each do |cog|
      cog_config = @config_manager.config_for(cog.class, cog.name)
      cog_task = cog.run!(
        @barrier,
        cog_config.deep_dup,
        cog_input_context,
        @scope_value.deep_dup,
        @scope_index,
      )
      cog_task.wait unless cog_config.async?
    end
    # Wait on the tasks in their completion order, so that an exception in a task will be raised as soon as it occurs
    # noinspection RubyArgCount
    @barrier.wait { |task| wait_for_task_with_exception_handling(task) }
    compute_final_output # eagerly compute the final output (so it, too, can 'break!' subsequent executions in a loop)
  ensure
    @barrier.stop
    compute_final_output
    TaskContext.end
    @running = false
  end
end

#running? ⇒ Boolean

: () -> bool

Returns:

  • (Boolean) 


134
135
136
 
# File 'lib/roast/execution_manager.rb', line 134

def running?
  @running ||= false
end

#stop! ⇒ Object

: () -> void



119
120
121
 
# File 'lib/roast/execution_manager.rb', line 119

def stop!
  @barrier.stop
end