Repository: incubator-ariatosca Updated Branches: refs/heads/ARIA-278-Remove-core-tasks 15a1f66f9 -> f198fc05e (forced update)
optimization - try1 Project: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/commit/f198fc05 Tree: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/tree/f198fc05 Diff: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/diff/f198fc05 Branch: refs/heads/ARIA-278-Remove-core-tasks Commit: f198fc05e29aacc4c7d822be4a63e68233375df7 Parents: d517b82 Author: max-orlov <ma...@gigaspaces.com> Authored: Thu Jun 15 18:44:51 2017 +0300 Committer: max-orlov <ma...@gigaspaces.com> Committed: Thu Jun 15 19:08:08 2017 +0300 ---------------------------------------------------------------------- aria/orchestrator/workflows/core/engine.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/f198fc05/aria/orchestrator/workflows/core/engine.py ---------------------------------------------------------------------- diff --git a/aria/orchestrator/workflows/core/engine.py b/aria/orchestrator/workflows/core/engine.py index 48fb60a..e547aa1 100644 --- a/aria/orchestrator/workflows/core/engine.py +++ b/aria/orchestrator/workflows/core/engine.py @@ -38,6 +38,7 @@ class Engine(logger.LoggerMixin): def __init__(self, executor, **kwargs): super(Engine, self).__init__(**kwargs) self._executors = {executor.__class__: executor} + self._executing_tasks = [] def execute(self, ctx): """ @@ -88,7 +89,7 @@ class Engine(logger.LoggerMixin): ) def _ended_tasks(self, ctx): - for task in self._tasks_iter(ctx): + for task in self._executing_tasks: if task.has_ended() and task in ctx._graph: yield task @@ -122,12 +123,14 @@ class Engine(logger.LoggerMixin): name=task.name ) + self._executing_tasks.append(task) + if not task.stub_type: events.sent_task_signal.send(op_ctx) executor.execute(op_ctx) - @staticmethod - def _handle_ended_tasks(ctx, task): + def _handle_ended_tasks(self, ctx, task): + self._executing_tasks.remove(task) if task.status == models.Task.FAILED and not task.ignore_failure: raise exceptions.ExecutorException('Workflow failed') else: