# File lib/dynflow/executors/parallel.rb, line 8 def initialize(world, pool_size = 10) super(world) @core = Core.spawn name: 'parallel-executor-core', args: [world, pool_size], initialized: @core_initialized = Concurrent.future end
# File lib/dynflow/executors/parallel.rb, line 28 def event(execution_plan_id, step_id, event, future = Concurrent.future) @core.ask([:handle_event, Director::Event[execution_plan_id, step_id, event, future]]) future end
# File lib/dynflow/executors/parallel.rb, line 15 def execute(execution_plan_id, finished = Concurrent.future, wait_for_acceptance = true) accepted = @core.ask([:handle_execution, execution_plan_id, finished]) accepted.value! if wait_for_acceptance finished rescue Concurrent::Actor::ActorTerminated => error dynflow_error = Dynflow::Error.new('executor terminated') finished.fail dynflow_error unless finished.completed? raise dynflow_error rescue => e finished.fail e unless finished.completed? raise e end
# File lib/dynflow/executors/parallel.rb, line 38 def execution_status(execution_plan_id = nil) @core.ask!([:execution_status, execution_plan_id]) end
# File lib/dynflow/executors/parallel.rb, line 42 def initialized @core_initialized end
# File lib/dynflow/executors/parallel.rb, line 33 def terminate(future = Concurrent.future) @core.tell([:start_termination, future]) future end