wip
Project: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/commit/827230da Tree: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/tree/827230da Diff: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/diff/827230da Branch: refs/heads/ARIA-284-Cleanup-and-optimize-the-task-execution Commit: 827230da2f515c2c087d847a68137f27241de161 Parents: 75112ab Author: max-orlov <ma...@gigaspaces.com> Authored: Wed Jun 21 12:41:33 2017 +0300 Committer: max-orlov <ma...@gigaspaces.com> Committed: Thu Jun 22 14:28:55 2017 +0300 ---------------------------------------------------------------------- aria/modeling/orchestration.py | 2 - aria/orchestrator/context/workflow.py | 19 -- aria/orchestrator/workflow_runner.py | 5 +- aria/orchestrator/workflows/core/compile.py | 198 ++++++++++--------- aria/orchestrator/workflows/core/engine.py | 110 +++++++---- tests/orchestrator/context/__init__.py | 2 +- tests/orchestrator/context/test_serialize.py | 2 +- .../orchestrator/execution_plugin/test_local.py | 2 +- tests/orchestrator/execution_plugin/test_ssh.py | 3 +- .../orchestrator/workflows/core/test_engine.py | 2 +- .../orchestrator/workflows/core/test_events.py | 7 +- .../test_task_graph_into_execution_graph.py | 19 +- .../executor/test_process_executor_extension.py | 2 +- .../test_process_executor_tracked_changes.py | 2 +- 14 files changed, 198 insertions(+), 177 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/827230da/aria/modeling/orchestration.py ---------------------------------------------------------------------- diff --git a/aria/modeling/orchestration.py b/aria/modeling/orchestration.py index 276b68e..5b02d1b 100644 --- a/aria/modeling/orchestration.py +++ b/aria/modeling/orchestration.py @@ -306,7 +306,6 @@ class TaskBase(mixins.ModelMixin): ended_at = Column(DateTime, default=None) attempts_count = Column(Integer, default=1) - _api_id = Column(String) _executor = Column(PickleType) _context_cls = Column(PickleType) _stub_type = Column(Enum(*STUB_TYPES)) @@ -442,7 +441,6 @@ class TaskBase(mixins.ModelMixin): 'plugin': api_task.plugin, 'function': api_task.function, 'arguments': api_task.arguments, - '_api_id': api_task.id, '_context_cls': api_task._context_cls, '_executor': executor, } http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/827230da/aria/orchestrator/context/workflow.py ---------------------------------------------------------------------- diff --git a/aria/orchestrator/context/workflow.py b/aria/orchestrator/context/workflow.py index adcd635..18334f3 100644 --- a/aria/orchestrator/context/workflow.py +++ b/aria/orchestrator/context/workflow.py @@ -20,8 +20,6 @@ Workflow and operation contexts import threading from contextlib import contextmanager -from networkx import DiGraph - from .exceptions import ContextException from .common import BaseContext @@ -96,23 +94,6 @@ class WorkflowContext(BaseContext): ) @property - def _graph(self): - # Constructing a graph with only not ended nodes - if self._execution_graph is None: - graph = DiGraph() - for task in self.execution.tasks: - if task.has_ended(): - continue - for dependency in task.dependencies: - if dependency.has_ended(): - continue - graph.add_edge(dependency, task) - - self._execution_graph = graph - - return self._execution_graph - - @property @contextmanager def persist_changes(self): yield http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/827230da/aria/orchestrator/workflow_runner.py ---------------------------------------------------------------------- diff --git a/aria/orchestrator/workflow_runner.py b/aria/orchestrator/workflow_runner.py index 3ccb1ee..b3f100d 100644 --- a/aria/orchestrator/workflow_runner.py +++ b/aria/orchestrator/workflow_runner.py @@ -96,8 +96,9 @@ class WorkflowRunner(object): if not self._is_resume: workflow_fn = self._get_workflow_fn() - tasks_graph = workflow_fn(ctx=self._workflow_context, **execution_inputs_dict) - compile.create_execution_tasks(self._workflow_context, tasks_graph, executor.__class__) + self._tasks_graph = workflow_fn(ctx=self._workflow_context, **execution_inputs_dict) + compile.GraphCompiler(self._workflow_context, executor.__class__).compile( + self._tasks_graph) self._engine = engine.Engine(executors={executor.__class__: executor}) http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/827230da/aria/orchestrator/workflows/core/compile.py ---------------------------------------------------------------------- diff --git a/aria/orchestrator/workflows/core/compile.py b/aria/orchestrator/workflows/core/compile.py index 932268a..83de22c 100644 --- a/aria/orchestrator/workflows/core/compile.py +++ b/aria/orchestrator/workflows/core/compile.py @@ -18,99 +18,105 @@ from ....modeling import models from .. import executor, api -def create_execution_tasks(ctx, task_graph, default_executor): - execution = ctx.execution - _construct_execution_tasks(execution, task_graph, default_executor) - ctx.model.execution.update(execution) - return execution.tasks - - -def _construct_execution_tasks(execution, - task_graph, - default_executor, - stub_executor=executor.base.StubTaskExecutor, - start_stub_type=models.Task.START_WORKFLOW, - end_stub_type=models.Task.END_WORKFLOW, - depends_on=()): - """ - Translates the user graph to the execution graph - :param task_graph: The user's graph - :param start_stub_type: internal use - :param end_stub_type: internal use - :param depends_on: internal use - """ - depends_on = list(depends_on) - - # Insert start marker - start_task = models.Task(execution=execution, - dependencies=depends_on, - _api_id=_start_graph_suffix(task_graph.id), - _stub_type=start_stub_type, - _executor=stub_executor) - - for task in task_graph.topological_order(reverse=True): - operation_dependencies = _get_tasks_from_dependencies( - execution, task_graph.get_dependencies(task), [start_task]) - - if isinstance(task, api.task.OperationTask): - models.Task.from_api_task(api_task=task, - executor=default_executor, - dependencies=operation_dependencies) - - elif isinstance(task, api.task.WorkflowTask): - # Build the graph recursively while adding start and end markers - _construct_execution_tasks( - execution=execution, - task_graph=task, - default_executor=default_executor, - stub_executor=stub_executor, - start_stub_type=models.Task.START_SUBWROFKLOW, - end_stub_type=models.Task.END_SUBWORKFLOW, - depends_on=operation_dependencies - ) - elif isinstance(task, api.task.StubTask): - models.Task(execution=execution, - dependencies=operation_dependencies, - _api_id=task.id, - _executor=stub_executor, - _stub_type=models.Task.STUB, - ) - else: - raise RuntimeError('Undefined state') - - # Insert end marker - models.Task(dependencies=_get_non_dependent_tasks(execution) or [start_task], - execution=execution, - _api_id=_end_graph_suffix(task_graph.id), - _executor=stub_executor, - _stub_type=end_stub_type) - - -def _start_graph_suffix(api_id): - return '{0}-Start'.format(api_id) - - -def _end_graph_suffix(api_id): - return '{0}-End'.format(api_id) - - -def _get_non_dependent_tasks(execution): - tasks_with_dependencies = set() - for task in execution.tasks: - tasks_with_dependencies.update(task.dependencies) - return list(set(execution.tasks) - set(tasks_with_dependencies)) - - -def _get_tasks_from_dependencies(execution, dependencies, default=()): - """ - Returns task list from dependencies. - """ - tasks = [] - for dependency in dependencies: - if getattr(dependency, 'actor', False): - # This is - dependency_name = dependency.id - else: - dependency_name = _end_graph_suffix(dependency.id) - tasks.extend(task for task in execution.tasks if task._api_id == dependency_name) - return tasks or default +# TODO: is class really needed? + +class GraphCompiler(object): + def __init__(self, ctx, default_executor): + self._ctx = ctx + self._default_executor = default_executor + self._stub_executor = executor.base.StubTaskExecutor + self._model_to_api_id = {} + + def compile(self, + task_graph, + start_stub_type=models.Task.START_WORKFLOW, + end_stub_type=models.Task.END_WORKFLOW, + depends_on=()): + """ + Translates the user graph to the execution graph + :param task_graph: The user's graph + :param start_stub_type: internal use + :param end_stub_type: internal use + :param depends_on: internal use + """ + task_graph = task_graph or self._task_graph + depends_on = list(depends_on) + + # Insert start marker + start_task = self._create_stub_task( + start_stub_type, depends_on, self._start_graph_suffix(task_graph.id), task_graph.name, + ) + + for task in task_graph.topological_order(reverse=True): + dependencies = \ + (self._get_tasks_from_dependencies(task_graph.get_dependencies(task)) + or [start_task]) + + if isinstance(task, api.task.OperationTask): + self._create_operation_task(task, dependencies) + + elif isinstance(task, api.task.WorkflowTask): + # Build the graph recursively while adding start and end markers + self.compile( + task, models.Task.START_SUBWROFKLOW, models.Task.END_SUBWORKFLOW, dependencies + ) + elif isinstance(task, api.task.StubTask): + self._create_stub_task(models.Task.STUB, dependencies, task.id) + else: + raise RuntimeError('Undefined state') + + # Insert end marker + self._create_stub_task( + end_stub_type, + self._get_non_dependent_tasks(self._ctx.execution) or [start_task], + self._end_graph_suffix(task_graph.id), + task_graph.name + ) + + def _create_stub_task(self, stub_type, dependencies, api_id, name=None): + model_task = models.Task( + name=name, + dependencies=dependencies, + execution=self._ctx.execution, + _executor=self._stub_executor, + _stub_type=stub_type) + self._ctx.model.task.put(model_task) + self._model_to_api_id[model_task.id] = api_id + return model_task + + def _create_operation_task(self, api_task, dependencies): + model_task = models.Task.from_api_task( + api_task, self._default_executor, dependencies=dependencies) + self._ctx.model.task.put(model_task) + self._model_to_api_id[model_task.id] = api_task.id + return model_task + + @staticmethod + def _start_graph_suffix(api_id): + return '{0}-Start'.format(api_id) + + @staticmethod + def _end_graph_suffix(api_id): + return '{0}-End'.format(api_id) + + @staticmethod + def _get_non_dependent_tasks(execution): + tasks_with_dependencies = set() + for task in execution.tasks: + tasks_with_dependencies.update(task.dependencies) + return list(set(execution.tasks) - set(tasks_with_dependencies)) + + def _get_tasks_from_dependencies(self, dependencies): + """ + Returns task list from dependencies. + """ + tasks = [] + for dependency in dependencies: + if getattr(dependency, 'actor', False): + # This is + dependency_name = dependency.id + else: + dependency_name = self._end_graph_suffix(dependency.id) + tasks.extend(task for task in self._ctx.execution.tasks + if self._model_to_api_id.get(task.id, None) == dependency_name) + return tasks http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/827230da/aria/orchestrator/workflows/core/engine.py ---------------------------------------------------------------------- diff --git a/aria/orchestrator/workflows/core/engine.py b/aria/orchestrator/workflows/core/engine.py index d5a6e70..f594e36 100644 --- a/aria/orchestrator/workflows/core/engine.py +++ b/aria/orchestrator/workflows/core/engine.py @@ -45,22 +45,23 @@ class Engine(logger.LoggerMixin): """ execute the workflow """ - executing_tasks = [] - if resuming: events.on_resume_workflow_signal.send(ctx) + task_tracker = _TasksTracker(ctx) try: events.start_workflow_signal.send(ctx) while True: cancel = self._is_cancel(ctx) if cancel: break - for task in self._ended_tasks(ctx, executing_tasks): - self._handle_ended_tasks(ctx, task, executing_tasks) - for task in self._executable_tasks(ctx): - self._handle_executable_task(ctx, task, executing_tasks) - if self._all_tasks_consumed(ctx): + for task in task_tracker.ended_tasks: + self._handle_ended_tasks(task) + task_tracker.finished_(task) + for task in task_tracker.executable_tasks: + task_tracker.executing_(task) + self._handle_executable_task(ctx, task) + if task_tracker.all_tasks_consumed: break else: time.sleep(0.1) @@ -86,34 +87,7 @@ class Engine(logger.LoggerMixin): execution = ctx.model.execution.refresh(ctx.execution) return execution.status in (models.Execution.CANCELLING, models.Execution.CANCELLED) - def _executable_tasks(self, ctx): - now = datetime.utcnow() - return ( - task for task in self._tasks_iter(ctx) - if task.is_waiting() and task.due_at <= now and \ - not self._task_has_dependencies(ctx, task) - ) - - @staticmethod - def _ended_tasks(ctx, executing_tasks): - for task in executing_tasks: - if task.has_ended() and task in ctx._graph: - yield task - - @staticmethod - def _task_has_dependencies(ctx, task): - return len(ctx._graph.pred.get(task, [])) > 0 - - @staticmethod - def _all_tasks_consumed(ctx): - return len(ctx._graph.node) == 0 - - @staticmethod - def _tasks_iter(ctx): - for task in ctx.execution.tasks: - yield ctx.model.task.refresh(task) - - def _handle_executable_task(self, ctx, task, executing_tasks): + def _handle_executable_task(self, ctx, task): task_executor = self._executors[task._executor] # If the task is a stub, a default context is provided, else it should hold the context cls @@ -129,16 +103,70 @@ class Engine(logger.LoggerMixin): name=task.name ) - executing_tasks.append(task) - if not task._stub_type: events.sent_task_signal.send(op_ctx) task_executor.execute(op_ctx) @staticmethod - def _handle_ended_tasks(ctx, task, executing_tasks): - executing_tasks.remove(task) + def _handle_ended_tasks(task): if task.status == models.Task.FAILED and not task.ignore_failure: raise exceptions.ExecutorException('Workflow failed') - else: - ctx._graph.remove_node(task) + + +class _TasksTracker(object): + def __init__(self, ctx): + self._ctx = ctx + self._tasks = ctx.execution.tasks + self._executed_tasks = [task for task in self._tasks if task.has_ended()] + self._executable_tasks = list(set(self._tasks) - set(self._executed_tasks)) + self._executing_tasks = [] + + @property + def all_tasks_consumed(self): + return len(self._executed_tasks) == len(self._tasks) and len(self._executing_tasks) == 0 + + def executing_(self, task): + # Task executing could be retrying (thus removed and added earlier) + if task not in self._executing_tasks: + self._executable_tasks.remove(task) + self._executing_tasks.append(task) + + def finished_(self, task): + self._executing_tasks.remove(task) + self._executed_tasks.append(task) + + @property + def ended_tasks(self): + for task in self.executing_tasks: + if task.has_ended(): + yield task + + @property + def executable_tasks(self): + now = datetime.utcnow() + # we need both list since retrying task are in the executing task list. + for task in self._update_tasks(self._executing_tasks + self._executable_tasks): + if all([task.is_waiting(), + task.due_at <= now, + all(dependency in self._executed_tasks for dependency in task.dependencies) + ]): + yield task + + @property + def executing_tasks(self): + for task in self._update_tasks(self._executing_tasks): + yield task + + @property + def executed_tasks(self): + for task in self._update_tasks(self._executed_tasks): + yield task + + @property + def tasks(self): + for task in self._update_tasks(self._tasks): + yield task + + def _update_tasks(self, tasks): + for task in tasks: + yield self._ctx.model.task.refresh(task) http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/827230da/tests/orchestrator/context/__init__.py ---------------------------------------------------------------------- diff --git a/tests/orchestrator/context/__init__.py b/tests/orchestrator/context/__init__.py index 086a066..752706e 100644 --- a/tests/orchestrator/context/__init__.py +++ b/tests/orchestrator/context/__init__.py @@ -26,7 +26,7 @@ def op_path(func, module_path=None): def execute(workflow_func, workflow_context, executor): graph = workflow_func(ctx=workflow_context) - compile.create_execution_tasks(workflow_context, graph, executor.__class__) + compile.GraphCompiler(workflow_context, executor.__class__).compile(graph) eng = engine.Engine(executors={executor.__class__: executor}) eng.execute(workflow_context) http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/827230da/tests/orchestrator/context/test_serialize.py ---------------------------------------------------------------------- diff --git a/tests/orchestrator/context/test_serialize.py b/tests/orchestrator/context/test_serialize.py index 5db5b63..b7335a0 100644 --- a/tests/orchestrator/context/test_serialize.py +++ b/tests/orchestrator/context/test_serialize.py @@ -48,7 +48,7 @@ def test_serialize_operation_context(context, executor, tmpdir): context.model.node.update(node) graph = _mock_workflow(ctx=context) # pylint: disable=no-value-for-parameter - compile.create_execution_tasks(context, graph, executor.__class__) + compile.GraphCompiler(context, executor.__class__).compile(graph) eng = engine.Engine({executor.__class__: executor}) eng.execute(context) http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/827230da/tests/orchestrator/execution_plugin/test_local.py ---------------------------------------------------------------------- diff --git a/tests/orchestrator/execution_plugin/test_local.py b/tests/orchestrator/execution_plugin/test_local.py index 1695320..ab6310c 100644 --- a/tests/orchestrator/execution_plugin/test_local.py +++ b/tests/orchestrator/execution_plugin/test_local.py @@ -500,7 +500,7 @@ if __name__ == '__main__': arguments=arguments)) return graph tasks_graph = mock_workflow(ctx=workflow_context) # pylint: disable=no-value-for-parameter - compile.create_execution_tasks(workflow_context, tasks_graph, executor.__class__) + compile.GraphCompiler(workflow_context, executor.__class__).compile(tasks_graph) eng = engine.Engine({executor.__class__: executor}) eng.execute(workflow_context) return workflow_context.model.node.get_by_name( http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/827230da/tests/orchestrator/execution_plugin/test_ssh.py ---------------------------------------------------------------------- diff --git a/tests/orchestrator/execution_plugin/test_ssh.py b/tests/orchestrator/execution_plugin/test_ssh.py index fb1dc09..13ad1a3 100644 --- a/tests/orchestrator/execution_plugin/test_ssh.py +++ b/tests/orchestrator/execution_plugin/test_ssh.py @@ -254,8 +254,7 @@ class TestWithActualSSHServer(object): graph.sequence(*ops) return graph tasks_graph = mock_workflow(ctx=self._workflow_context) # pylint: disable=no-value-for-parameter - compile.create_execution_tasks( - self._workflow_context, tasks_graph, self._executor.__class__) + compile.GraphCompiler(self._workflow_context, self._executor.__class__).compile(tasks_graph) eng = engine.Engine({self._executor.__class__: self._executor}) eng.execute(self._workflow_context) return self._workflow_context.model.node.get_by_name( http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/827230da/tests/orchestrator/workflows/core/test_engine.py ---------------------------------------------------------------------- diff --git a/tests/orchestrator/workflows/core/test_engine.py b/tests/orchestrator/workflows/core/test_engine.py index b77d284..7275723 100644 --- a/tests/orchestrator/workflows/core/test_engine.py +++ b/tests/orchestrator/workflows/core/test_engine.py @@ -50,7 +50,7 @@ class BaseTest(object): @staticmethod def _engine(workflow_func, workflow_context, executor): graph = workflow_func(ctx=workflow_context) - compile.create_execution_tasks(workflow_context, graph, executor.__class__) + compile.GraphCompiler(workflow_context, executor.__class__).compile(graph) return engine.Engine(executors={executor.__class__: executor}) http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/827230da/tests/orchestrator/workflows/core/test_events.py ---------------------------------------------------------------------- diff --git a/tests/orchestrator/workflows/core/test_events.py b/tests/orchestrator/workflows/core/test_events.py index 2b82443..32a6b7b 100644 --- a/tests/orchestrator/workflows/core/test_events.py +++ b/tests/orchestrator/workflows/core/test_events.py @@ -113,10 +113,9 @@ def run_operation_on_node(ctx, op_name, interface_name): operation_name=op_name, operation_kwargs=dict(function='{name}.{func.__name__}'.format(name=__name__, func=func))) node.interfaces[interface.name] = interface - compile.create_execution_tasks( - ctx, - single_operation_workflow(ctx, node=node, interface_name=interface_name, op_name=op_name), - ThreadExecutor) + compile.GraphCompiler(ctx, ThreadExecutor).compile( + single_operation_workflow(ctx, node=node, interface_name=interface_name, op_name=op_name) + ) eng = engine.Engine(executors={ThreadExecutor: ThreadExecutor()}) eng.execute(ctx) http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/827230da/tests/orchestrator/workflows/core/test_task_graph_into_execution_graph.py ---------------------------------------------------------------------- diff --git a/tests/orchestrator/workflows/core/test_task_graph_into_execution_graph.py b/tests/orchestrator/workflows/core/test_task_graph_into_execution_graph.py index f5fb17a..3d47d54 100644 --- a/tests/orchestrator/workflows/core/test_task_graph_into_execution_graph.py +++ b/tests/orchestrator/workflows/core/test_task_graph_into_execution_graph.py @@ -13,7 +13,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -from networkx import topological_sort +from networkx import topological_sort, DiGraph from aria.modeling import models from aria.orchestrator import context @@ -65,9 +65,10 @@ def test_task_graph_into_execution_graph(tmpdir): test_task_graph.add_dependency(inner_task_graph, simple_before_task) test_task_graph.add_dependency(simple_after_task, inner_task_graph) - compile.create_execution_tasks(workflow_context, test_task_graph, base.StubTaskExecutor) + graph_compiler = compile.GraphCompiler(workflow_context, base.StubTaskExecutor) + graph_compiler.compile(test_task_graph) - execution_tasks = topological_sort(workflow_context._graph) + execution_tasks = topological_sort(_graph(workflow_context.execution.tasks)) assert len(execution_tasks) == 7 @@ -81,7 +82,7 @@ def test_task_graph_into_execution_graph(tmpdir): '{0}-End'.format(test_task_graph.id) ] - assert expected_tasks_names == [t._api_id for t in execution_tasks] + assert expected_tasks_names == [graph_compiler._model_to_api_id[t.id] for t in execution_tasks] assert all(isinstance(task, models.Task) for task in execution_tasks) execution_tasks = iter(execution_tasks) @@ -97,7 +98,6 @@ def test_task_graph_into_execution_graph(tmpdir): def _assert_execution_is_api_task(execution_task, api_task): - assert execution_task._api_id == api_task.id assert execution_task.name == api_task.name assert execution_task.function == api_task.function assert execution_task.actor == api_task.actor @@ -106,3 +106,12 @@ def _assert_execution_is_api_task(execution_task, api_task): def _get_task_by_name(task_name, graph): return graph.node[task_name]['task'] + + +def _graph(tasks): + graph = DiGraph() + for task in tasks: + for dependency in task.dependencies: + graph.add_edge(dependency, task) + + return graph http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/827230da/tests/orchestrator/workflows/executor/test_process_executor_extension.py ---------------------------------------------------------------------- diff --git a/tests/orchestrator/workflows/executor/test_process_executor_extension.py b/tests/orchestrator/workflows/executor/test_process_executor_extension.py index ba98c4f..aa08685 100644 --- a/tests/orchestrator/workflows/executor/test_process_executor_extension.py +++ b/tests/orchestrator/workflows/executor/test_process_executor_extension.py @@ -57,7 +57,7 @@ def test_decorate_extension(context, executor): graph.add_tasks(task) return graph graph = mock_workflow(ctx=context) # pylint: disable=no-value-for-parameter - compile.create_execution_tasks(context, graph, executor.__class__) + compile.GraphCompiler(context, executor.__class__).compile(graph) eng = engine.Engine({executor.__class__: executor}) eng.execute(context) out = get_node(context).attributes.get('out').value http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/827230da/tests/orchestrator/workflows/executor/test_process_executor_tracked_changes.py ---------------------------------------------------------------------- diff --git a/tests/orchestrator/workflows/executor/test_process_executor_tracked_changes.py b/tests/orchestrator/workflows/executor/test_process_executor_tracked_changes.py index 2f1c325..7102b13 100644 --- a/tests/orchestrator/workflows/executor/test_process_executor_tracked_changes.py +++ b/tests/orchestrator/workflows/executor/test_process_executor_tracked_changes.py @@ -107,7 +107,7 @@ def _run_workflow(context, executor, op_func, arguments=None): graph.add_tasks(task) return graph graph = mock_workflow(ctx=context) # pylint: disable=no-value-for-parameter - compile.create_execution_tasks(context, graph, executor.__class__) + compile.GraphCompiler(context, executor.__class__).compile(graph) eng = engine.Engine({executor.__class__: executor}) eng.execute(context) out = context.model.node.get_by_name(mock.models.DEPENDENCY_NODE_NAME).attributes.get('out')
