You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@ariatosca.apache.org by mx...@apache.org on 2017/06/15 15:44:56 UTC
incubator-ariatosca git commit: optimization - try1
Repository: incubator-ariatosca
Updated Branches:
refs/heads/ARIA-278-Remove-core-tasks d517b820e -> 15a1f66f9
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/15a1f66f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/tree/15a1f66f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/diff/15a1f66f
Branch: refs/heads/ARIA-278-Remove-core-tasks
Commit: 15a1f66f9ca8c51746a92cea59cf14b3c043b2fd
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 18:44:51 2017 +0300
----------------------------------------------------------------------
aria/orchestrator/workflows/core/engine.py | 9 +-
.../workflows/core/events_handler.py | 129 +++++++++----------
aria/orchestrator/workflows/executor/base.py | 9 +-
3 files changed, 72 insertions(+), 75 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/15a1f66f/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:
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/15a1f66f/aria/orchestrator/workflows/core/events_handler.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/core/events_handler.py b/aria/orchestrator/workflows/core/events_handler.py
index 3a780d5..1f03167 100644
--- a/aria/orchestrator/workflows/core/events_handler.py
+++ b/aria/orchestrator/workflows/core/events_handler.py
@@ -31,108 +31,99 @@ from ... import exceptions
@events.sent_task_signal.connect
def _task_sent(ctx, *args, **kwargs):
- with ctx.track_changes:
- ctx.task.status = ctx.task.SENT
+ ctx.task.status = ctx.task.SENT
@events.start_task_signal.connect
def _task_started(ctx, *args, **kwargs):
- with ctx.track_changes:
- ctx.task.started_at = datetime.utcnow()
- ctx.task.status = ctx.task.STARTED
- _update_node_state_if_necessary(ctx, is_transitional=True)
+ ctx.task.started_at = datetime.utcnow()
+ ctx.task.status = ctx.task.STARTED
+ _update_node_state_if_necessary(ctx, is_transitional=True)
@events.on_failure_task_signal.connect
def _task_failed(ctx, exception, *args, **kwargs):
- with ctx.track_changes:
- should_retry = all([
- not isinstance(exception, exceptions.TaskAbortException),
- ctx.task.attempts_count < ctx.task.max_attempts or
- ctx.task.max_attempts == ctx.task.INFINITE_RETRIES,
- # ignore_failure check here means the task will not be retried and it will be marked
- # as failed. The engine will also look at ignore_failure so it won't fail the
- # workflow.
- not ctx.task.ignore_failure
- ])
- if should_retry:
- retry_interval = None
- if isinstance(exception, exceptions.TaskRetryException):
- retry_interval = exception.retry_interval
- if retry_interval is None:
- retry_interval = ctx.task.retry_interval
- ctx.task.status = ctx.task.RETRYING
- ctx.task.attempts_count += 1
- ctx.task.due_at = datetime.utcnow() + timedelta(seconds=retry_interval)
- else:
- ctx.task.ended_at = datetime.utcnow()
- ctx.task.status = ctx.task.FAILED
+ should_retry = all([
+ not isinstance(exception, exceptions.TaskAbortException),
+ ctx.task.attempts_count < ctx.task.max_attempts or
+ ctx.task.max_attempts == ctx.task.INFINITE_RETRIES,
+ # ignore_failure check here means the task will not be retried and it will be marked
+ # as failed. The engine will also look at ignore_failure so it won't fail the
+ # workflow.
+ not ctx.task.ignore_failure
+ ])
+ if should_retry:
+ retry_interval = None
+ if isinstance(exception, exceptions.TaskRetryException):
+ retry_interval = exception.retry_interval
+ if retry_interval is None:
+ retry_interval = ctx.task.retry_interval
+ ctx.task.status = ctx.task.RETRYING
+ ctx.task.attempts_count += 1
+ ctx.task.due_at = datetime.utcnow() + timedelta(seconds=retry_interval)
+ else:
+ ctx.task.ended_at = datetime.utcnow()
+ ctx.task.status = ctx.task.FAILED
@events.on_success_task_signal.connect
def _task_succeeded(ctx, *args, **kwargs):
- with ctx.track_changes:
- ctx.task.ended_at = datetime.utcnow()
- ctx.task.status = ctx.task.SUCCESS
+ ctx.task.ended_at = datetime.utcnow()
+ ctx.task.status = ctx.task.SUCCESS
- _update_node_state_if_necessary(ctx)
+ _update_node_state_if_necessary(ctx)
@events.start_workflow_signal.connect
def _workflow_started(workflow_context, *args, **kwargs):
- with workflow_context.track_changes:
- execution = workflow_context.execution
- # the execution may already be in the process of cancelling
- if execution.status in (execution.CANCELLING, execution.CANCELLED):
- return
- execution.status = execution.STARTED
- execution.started_at = datetime.utcnow()
+ execution = workflow_context.execution
+ # the execution may already be in the process of cancelling
+ if execution.status in (execution.CANCELLING, execution.CANCELLED):
+ return
+ execution.status = execution.STARTED
+ execution.started_at = datetime.utcnow()
@events.on_failure_workflow_signal.connect
def _workflow_failed(workflow_context, exception, *args, **kwargs):
- with workflow_context.track_changes:
- execution = workflow_context.execution
- execution.error = str(exception)
- execution.status = execution.FAILED
- execution.ended_at = datetime.utcnow()
+ execution = workflow_context.execution
+ execution.error = str(exception)
+ execution.status = execution.FAILED
+ execution.ended_at = datetime.utcnow()
@events.on_success_workflow_signal.connect
def _workflow_succeeded(workflow_context, *args, **kwargs):
- with workflow_context.track_changes:
- execution = workflow_context.execution
- execution.status = execution.SUCCEEDED
- execution.ended_at = datetime.utcnow()
+ execution = workflow_context.execution
+ execution.status = execution.SUCCEEDED
+ execution.ended_at = datetime.utcnow()
@events.on_cancelled_workflow_signal.connect
def _workflow_cancelled(workflow_context, *args, **kwargs):
- with workflow_context.track_changes:
- execution = workflow_context.execution
- # _workflow_cancelling function may have called this function already
- if execution.status == execution.CANCELLED:
- return
- # the execution may have already been finished
- elif execution.status in (execution.SUCCEEDED, execution.FAILED):
- _log_tried_to_cancel_execution_but_it_already_ended(workflow_context, execution.status)
- else:
- execution.status = execution.CANCELLED
- execution.ended_at = datetime.utcnow()
+ execution = workflow_context.execution
+ # _workflow_cancelling function may have called this function already
+ if execution.status == execution.CANCELLED:
+ return
+ # the execution may have already been finished
+ elif execution.status in (execution.SUCCEEDED, execution.FAILED):
+ _log_tried_to_cancel_execution_but_it_already_ended(workflow_context, execution.status)
+ else:
+ execution.status = execution.CANCELLED
+ execution.ended_at = datetime.utcnow()
@events.on_cancelling_workflow_signal.connect
def _workflow_cancelling(workflow_context, *args, **kwargs):
- with workflow_context.track_changes:
- execution = workflow_context.execution
- if execution.status == execution.PENDING:
- return _workflow_cancelled(workflow_context=workflow_context)
- # the execution may have already been finished
- elif execution.status in (execution.SUCCEEDED, execution.FAILED):
- _log_tried_to_cancel_execution_but_it_already_ended(workflow_context, execution.status)
- else:
- execution.status = execution.CANCELLING
- workflow_context.execution = execution
+ execution = workflow_context.execution
+ if execution.status == execution.PENDING:
+ return _workflow_cancelled(workflow_context=workflow_context)
+ # the execution may have already been finished
+ elif execution.status in (execution.SUCCEEDED, execution.FAILED):
+ _log_tried_to_cancel_execution_but_it_already_ended(workflow_context, execution.status)
+ else:
+ execution.status = execution.CANCELLING
+ workflow_context.execution = execution
def _update_node_state_if_necessary(ctx, is_transitional=False):
http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/15a1f66f/aria/orchestrator/workflows/executor/base.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/executor/base.py b/aria/orchestrator/workflows/executor/base.py
index 9e1ce7e..dfb580c 100644
--- a/aria/orchestrator/workflows/executor/base.py
+++ b/aria/orchestrator/workflows/executor/base.py
@@ -51,15 +51,18 @@ class BaseExecutor(logger.LoggerMixin):
@staticmethod
def _task_started(ctx):
- events.start_task_signal.send(ctx)
+ with ctx.track_changes:
+ events.start_task_signal.send(ctx)
@staticmethod
def _task_failed(ctx, exception, traceback=None):
- events.on_failure_task_signal.send(ctx, exception=exception, traceback=traceback)
+ with ctx.track_changes:
+ events.on_failure_task_signal.send(ctx, exception=exception, traceback=traceback)
@staticmethod
def _task_succeeded(ctx):
- events.on_success_task_signal.send(ctx)
+ with ctx.track_changes:
+ events.on_success_task_signal.send(ctx)
class StubTaskExecutor(BaseExecutor): # pylint: disable=abstract-method