ariatosca-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mxm...@apache.org
Subject [2/2] incubator-ariatosca git commit: ARIA-278 remove core tasks
Date Tue, 20 Jun 2017 16:14:29 GMT
ARIA-278 remove core tasks


Project: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/commit/3e119df5
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/tree/3e119df5
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/diff/3e119df5

Branch: refs/heads/ARIA-278-Remove-core-tasks
Commit: 3e119df50b87f3f96fc7b17f48756e89fc058364
Parents: 2149a5e
Author: max-orlov <maxim@gigaspaces.com>
Authored: Sun Jun 11 19:05:35 2017 +0300
Committer: max-orlov <maxim@gigaspaces.com>
Committed: Tue Jun 20 19:14:22 2017 +0300

----------------------------------------------------------------------
 aria/modeling/orchestration.py                  | 143 +++++++---
 aria/orchestrator/context/operation.py          |   7 +
 aria/orchestrator/context/workflow.py           |  21 ++
 aria/orchestrator/workflow_runner.py            |  22 +-
 aria/orchestrator/workflows/api/task.py         |   7 +
 aria/orchestrator/workflows/core/__init__.py    |   2 +-
 aria/orchestrator/workflows/core/compile.py     | 116 ++++++++
 aria/orchestrator/workflows/core/engine.py      | 121 +++++----
 .../workflows/core/events_handler.py            | 137 +++++-----
 aria/orchestrator/workflows/core/task.py        | 271 -------------------
 aria/orchestrator/workflows/core/translation.py | 109 --------
 aria/orchestrator/workflows/events_logging.py   |  25 +-
 aria/orchestrator/workflows/executor/base.py    |  31 ++-
 aria/orchestrator/workflows/executor/dry.py     |  57 ++--
 aria/orchestrator/workflows/executor/process.py |  18 +-
 aria/orchestrator/workflows/executor/thread.py  |  18 +-
 tests/__init__.py                               |   2 +
 tests/orchestrator/context/__init__.py          |   9 +-
 tests/orchestrator/context/test_operation.py    |  26 +-
 tests/orchestrator/context/test_serialize.py    |  10 +-
 .../orchestrator/execution_plugin/test_local.py |  10 +-
 tests/orchestrator/execution_plugin/test_ssh.py |  11 +-
 tests/orchestrator/test_workflow_runner.py      |  43 ++-
 .../orchestrator/workflows/core/test_engine.py  |  23 +-
 .../orchestrator/workflows/core/test_events.py  |  20 +-
 tests/orchestrator/workflows/core/test_task.py  |  31 +--
 .../test_task_graph_into_execution_graph.py     |  55 ++--
 .../orchestrator/workflows/executor/__init__.py |  89 +++---
 .../workflows/executor/test_executor.py         |  26 +-
 .../workflows/executor/test_process_executor.py |  26 +-
 .../executor/test_process_executor_extension.py |   7 +-
 .../test_process_executor_tracked_changes.py    |   7 +-
 32 files changed, 677 insertions(+), 823 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e119df5/aria/modeling/orchestration.py
----------------------------------------------------------------------
diff --git a/aria/modeling/orchestration.py b/aria/modeling/orchestration.py
index 995c8c2..17d2476 100644
--- a/aria/modeling/orchestration.py
+++ b/aria/modeling/orchestration.py
@@ -21,7 +21,6 @@ classes:
 """
 
 # pylint: disable=no-self-argument, no-member, abstract-method
-
 from datetime import datetime
 
 from sqlalchemy import (
@@ -34,19 +33,19 @@ from sqlalchemy import (
     String,
     Float,
     orm,
-)
+    PickleType)
 from sqlalchemy.ext.associationproxy import association_proxy
 from sqlalchemy.ext.declarative import declared_attr
 
 from ..orchestrator.exceptions import (TaskAbortException, TaskRetryException)
-from .mixins import ModelMixin, ParameterMixin
+from . import mixins
 from . import (
     relationship,
     types as modeling_types
 )
 
 
-class ExecutionBase(ModelMixin):
+class ExecutionBase(mixins.ModelMixin):
     """
     Execution model representation.
     """
@@ -152,7 +151,7 @@ class ExecutionBase(ModelMixin):
         )
 
 
-class PluginBase(ModelMixin):
+class PluginBase(mixins.ModelMixin):
     """
     An installed plugin.
 
@@ -213,7 +212,7 @@ class PluginBase(ModelMixin):
     uploaded_at = Column(DateTime, nullable=False, index=True)
 
 
-class TaskBase(ModelMixin):
+class TaskBase(mixins.ModelMixin):
     """
     Represents the smallest unit of stateful execution in ARIA. The task state includes inputs,
     outputs, as well as an atomic status, ensuring that the task can only be running once at any
@@ -257,10 +256,24 @@ class TaskBase(ModelMixin):
 
     __tablename__ = 'task'
 
-    __private_fields__ = ['node_fk',
-                          'relationship_fk',
-                          'plugin_fk',
-                          'execution_fk']
+    __private_fields__ = ['dependency_operation_task_fk', 'dependency_stub_task_fk', 'node_fk',
+                          'relationship_fk', 'plugin_fk', 'execution_fk']
+
+    START_WORKFLOW = 'start_workflow'
+    END_WORKFLOW = 'end_workflow'
+    START_SUBWROFKLOW = 'start_subworkflow'
+    END_SUBWORKFLOW = 'end_subworkflow'
+    STUB = 'stub'
+    CONDITIONAL = 'conditional'
+
+    STUB_TYPES = (
+        START_WORKFLOW,
+        START_SUBWROFKLOW,
+        END_WORKFLOW,
+        END_SUBWORKFLOW,
+        STUB,
+        CONDITIONAL,
+    )
 
     PENDING = 'pending'
     RETRYING = 'retrying'
@@ -276,10 +289,28 @@ class TaskBase(ModelMixin):
         SUCCESS,
         FAILED,
     )
-
     INFINITE_RETRIES = -1
 
     @declared_attr
+    def execution(cls):
+        return relationship.many_to_one(cls, 'execution')
+
+    @declared_attr
+    def execution_fk(cls):
+        return relationship.foreign_key('execution', nullable=True)
+
+    status = Column(Enum(*STATES, name='status'), default=PENDING)
+    due_at = Column(DateTime, nullable=False, index=True, default=datetime.utcnow())
+    started_at = Column(DateTime, default=None)
+    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))
+
+    @declared_attr
     def logs(cls):
         return relationship.one_to_many(cls, 'log')
 
@@ -296,10 +327,6 @@ class TaskBase(ModelMixin):
         return relationship.many_to_one(cls, 'plugin')
 
     @declared_attr
-    def execution(cls):
-        return relationship.many_to_one(cls, 'execution')
-
-    @declared_attr
     def arguments(cls):
         return relationship.one_to_many(cls, 'argument', dict_key='name')
 
@@ -307,19 +334,8 @@ class TaskBase(ModelMixin):
     max_attempts = Column(Integer, default=1)
     retry_interval = Column(Float, default=0)
     ignore_failure = Column(Boolean, default=False)
-
-    # State
-    status = Column(Enum(*STATES, name='status'), default=PENDING)
-    due_at = Column(DateTime, nullable=False, index=True, default=datetime.utcnow())
-    started_at = Column(DateTime, default=None)
-    ended_at = Column(DateTime, default=None)
-    attempts_count = Column(Integer, default=1)
-
-    def has_ended(self):
-        return self.status in (self.SUCCESS, self.FAILED)
-
-    def is_waiting(self):
-        return self.status in (self.PENDING, self.RETRYING)
+    interface_name = Column(String)
+    operation_name = Column(String)
 
     @property
     def actor(self):
@@ -351,10 +367,6 @@ class TaskBase(ModelMixin):
     def plugin_fk(cls):
         return relationship.foreign_key('plugin', nullable=True)
 
-    @declared_attr
-    def execution_fk(cls):
-        return relationship.foreign_key('execution', nullable=True)
-
     # endregion
 
     # region association proxies
@@ -376,14 +388,6 @@ class TaskBase(ModelMixin):
 
     # endregion
 
-    @classmethod
-    def for_node(cls, actor, **kwargs):
-        return cls(node=actor, **kwargs)
-
-    @classmethod
-    def for_relationship(cls, actor, **kwargs):
-        return cls(relationship=actor, **kwargs)
-
     @staticmethod
     def abort(message=None):
         raise TaskAbortException(message)
@@ -392,8 +396,63 @@ class TaskBase(ModelMixin):
     def retry(message=None, retry_interval=None):
         raise TaskRetryException(message, retry_interval=retry_interval)
 
+    @declared_attr
+    def dependency_fk(self):
+        return relationship.foreign_key('task', nullable=True)
+
+    @declared_attr
+    def dependencies(cls):
+        # symmetric relationship causes funky graphs
+        return relationship.one_to_many_self(cls, 'dependency_fk')
+
+    def has_ended(self):
+        return self.status in (self.SUCCESS, self.FAILED)
+
+    def is_waiting(self):
+        if self._stub_type:
+            return not self.has_ended()
+        else:
+            return self.status in (self.PENDING, self.RETRYING)
+
+    @classmethod
+    def from_api_task(cls, api_task, executor, **kwargs):
+        instantiation_kwargs = {}
+
+        if hasattr(api_task.actor, 'outbound_relationships'):
+            instantiation_kwargs['node'] = api_task.actor
+        elif hasattr(api_task.actor, 'source_node'):
+            instantiation_kwargs['relationship'] = api_task.actor
+        else:
+            raise RuntimeError('No operation context could be created for {actor.model_cls}'
+                               .format(actor=api_task.actor))
+
+        instantiation_kwargs.update(
+            {
+                'name': api_task.name,
+                'status': cls.PENDING,
+                'max_attempts': api_task.max_attempts,
+                'retry_interval': api_task.retry_interval,
+                'ignore_failure': api_task.ignore_failure,
+                'execution': api_task._workflow_context.execution,
+                'interface_name': api_task.interface_name,
+                'operation_name': api_task.operation_name,
+
+                # Only non-stub tasks have these fields
+                '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,
+            }
+        )
+
+        instantiation_kwargs.update(**kwargs)
+
+        return cls(**instantiation_kwargs)
+
 
-class LogBase(ModelMixin):
+class LogBase(mixins.ModelMixin):
 
     __tablename__ = 'log'
 
@@ -435,7 +494,7 @@ class LogBase(ModelMixin):
         return '{name}: {self.msg}'.format(name=name, self=self)
 
 
-class ArgumentBase(ParameterMixin):
+class ArgumentBase(mixins.ParameterMixin):
 
     __tablename__ = 'argument'
 

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e119df5/aria/orchestrator/context/operation.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/context/operation.py b/aria/orchestrator/context/operation.py
index efdc04d..d43b847 100644
--- a/aria/orchestrator/context/operation.py
+++ b/aria/orchestrator/context/operation.py
@@ -18,6 +18,7 @@ Workflow and operation contexts
 """
 
 import threading
+from contextlib import contextmanager
 
 import aria
 from aria.utils import file
@@ -106,6 +107,12 @@ class BaseOperationContext(common.BaseContext):
             self.model.log._session.remove()
             self.model.log._engine.dispose()
 
+    @property
+    @contextmanager
+    def persist_changes(self):
+        yield
+        self.model.task.update(self.task)
+
 
 class NodeOperationContext(BaseOperationContext):
     """

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e119df5/aria/orchestrator/context/workflow.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/context/workflow.py b/aria/orchestrator/context/workflow.py
index 920b237..aa5a786 100644
--- a/aria/orchestrator/context/workflow.py
+++ b/aria/orchestrator/context/workflow.py
@@ -20,6 +20,8 @@ Workflow and operation contexts
 import threading
 from contextlib import contextmanager
 
+from networkx import DiGraph
+
 from .exceptions import ContextException
 from .common import BaseContext
 
@@ -41,6 +43,7 @@ class WorkflowContext(BaseContext):
         self._task_max_attempts = task_max_attempts
         self._task_retry_interval = task_retry_interval
         self._task_ignore_failure = task_ignore_failure
+        self._execution_graph = None
         self._register_logger()
 
     def __repr__(self):
@@ -92,6 +95,24 @@ class WorkflowContext(BaseContext):
             }
         )
 
+    @property
+    def _graph(self):
+        if self._execution_graph is None:
+            graph = DiGraph()
+            for task in self.execution.tasks:
+                for dependency in task.dependencies:
+                    graph.add_edge(dependency, task)
+
+            self._execution_graph = graph
+
+        return self._execution_graph
+
+    @property
+    @contextmanager
+    def persist_changes(self):
+        yield
+        self._model.execution.update(self.execution)
+
 
 class _CurrentContext(threading.local):
     """

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e119df5/aria/orchestrator/workflow_runner.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflow_runner.py b/aria/orchestrator/workflow_runner.py
index 848c59b..9e6b3ad 100644
--- a/aria/orchestrator/workflow_runner.py
+++ b/aria/orchestrator/workflow_runner.py
@@ -24,7 +24,7 @@ from datetime import datetime
 from . import exceptions
 from .context.workflow import WorkflowContext
 from .workflows import builtin
-from .workflows.core.engine import Engine
+from .workflows.core import engine, compile
 from .workflows.executor.process import ProcessExecutor
 from ..modeling import models
 from ..modeling import utils as modeling_utils
@@ -70,7 +70,7 @@ class WorkflowRunner(object):
         execution = self._create_execution_model(inputs)
         self._execution_id = execution.id
 
-        workflow_context = WorkflowContext(
+        self._workflow_context = WorkflowContext(
             name=self.__class__.__name__,
             model_storage=self._model_storage,
             resource_storage=resource_storage,
@@ -80,15 +80,17 @@ class WorkflowRunner(object):
             task_max_attempts=task_max_attempts,
             task_retry_interval=task_retry_interval)
 
+        # Set default executor and kwargs
+        executor = executor or ProcessExecutor(plugin_manager=plugin_manager)
+
         # transforming the execution inputs to dict, to pass them to the workflow function
         execution_inputs_dict = dict(inp.unwrapped for inp in self.execution.inputs.values())
-        self._tasks_graph = workflow_fn(ctx=workflow_context, **execution_inputs_dict)
 
-        executor = executor or ProcessExecutor(plugin_manager=plugin_manager)
-        self._engine = Engine(
-            executor=executor,
-            workflow_context=workflow_context,
-            tasks_graph=self._tasks_graph)
+        self._tasks_graph = workflow_fn(ctx=self._workflow_context, **execution_inputs_dict)
+        compile.create_execution_tasks(
+            self._workflow_context, self._tasks_graph, executor.__class__)
+
+        self._engine = engine.Engine(executors={executor.__class__: executor})
 
     @property
     def execution_id(self):
@@ -103,10 +105,10 @@ class WorkflowRunner(object):
         return self._model_storage.service.get(self._service_id)
 
     def execute(self):
-        self._engine.execute()
+        self._engine.execute(ctx=self._workflow_context)
 
     def cancel(self):
-        self._engine.cancel_execution()
+        self._engine.cancel_execution(ctx=self._workflow_context)
 
     def _create_execution_model(self, inputs):
         execution = models.Execution(

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e119df5/aria/orchestrator/workflows/api/task.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/api/task.py b/aria/orchestrator/workflows/api/task.py
index ca125a8..f7d2c66 100644
--- a/aria/orchestrator/workflows/api/task.py
+++ b/aria/orchestrator/workflows/api/task.py
@@ -140,6 +140,13 @@ class OperationTask(BaseTask):
         self.arguments = modeling_utils.merge_parameter_values(arguments,
                                                                operation.arguments,
                                                                model_cls=models.Argument)
+        if getattr(self.actor, 'outbound_relationships', None) is not None:
+            self._context_cls = context.operation.NodeOperationContext
+        elif getattr(self.actor, 'source_node', None) is not None:
+            self._context_cls = context.operation.RelationshipOperationContext
+        else:
+            raise exceptions.TaskCreationException('Could not locate valid context for '
+                                                   '{actor.__class__}'.format(actor=self.actor))
 
     def __repr__(self):
         return self.name

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e119df5/aria/orchestrator/workflows/core/__init__.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/core/__init__.py b/aria/orchestrator/workflows/core/__init__.py
index e377153..81db43f 100644
--- a/aria/orchestrator/workflows/core/__init__.py
+++ b/aria/orchestrator/workflows/core/__init__.py
@@ -17,4 +17,4 @@
 Core for the workflow execution mechanism
 """
 
-from . import task, translation, engine
+from . import engine

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e119df5/aria/orchestrator/workflows/core/compile.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/core/compile.py b/aria/orchestrator/workflows/core/compile.py
new file mode 100644
index 0000000..932268a
--- /dev/null
+++ b/aria/orchestrator/workflows/core/compile.py
@@ -0,0 +1,116 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+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

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e119df5/aria/orchestrator/workflows/core/engine.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/core/engine.py b/aria/orchestrator/workflows/core/engine.py
index 3a96804..9f0ddd7 100644
--- a/aria/orchestrator/workflows/core/engine.py
+++ b/aria/orchestrator/workflows/core/engine.py
@@ -20,15 +20,13 @@ The workflow engine. Executes workflows
 import time
 from datetime import datetime
 
-import networkx
-
 from aria import logger
 from aria.modeling import models
 from aria.orchestrator import events
+from aria.orchestrator.context import operation
 
 from .. import exceptions
-from . import task as engine_task
-from . import translation
+from ..executor.base import StubTaskExecutor
 # Import required so all signals are registered
 from . import events_handler  # pylint: disable=unused-import
 
@@ -38,84 +36,105 @@ class Engine(logger.LoggerMixin):
     The workflow engine. Executes workflows
     """
 
-    def __init__(self, executor, workflow_context, tasks_graph, **kwargs):
+    def __init__(self, executors, **kwargs):
         super(Engine, self).__init__(**kwargs)
-        self._workflow_context = workflow_context
-        self._execution_graph = networkx.DiGraph()
-        translation.build_execution_graph(task_graph=tasks_graph,
-                                          execution_graph=self._execution_graph,
-                                          default_executor=executor)
+        self._executors = executors.copy()
+        self._executors.setdefault(StubTaskExecutor, StubTaskExecutor())
 
-    def execute(self):
+    def execute(self, ctx):
         """
         execute the workflow
         """
+        executing_tasks = []
         try:
-            events.start_workflow_signal.send(self._workflow_context)
+            events.start_workflow_signal.send(ctx)
             while True:
-                cancel = self._is_cancel()
+                cancel = self._is_cancel(ctx)
                 if cancel:
                     break
-                for task in self._ended_tasks():
-                    self._handle_ended_tasks(task)
-                for task in self._executable_tasks():
-                    self._handle_executable_task(task)
-                if self._all_tasks_consumed():
+                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):
                     break
                 else:
                     time.sleep(0.1)
             if cancel:
-                events.on_cancelled_workflow_signal.send(self._workflow_context)
+                events.on_cancelled_workflow_signal.send(ctx)
             else:
-                events.on_success_workflow_signal.send(self._workflow_context)
+                events.on_success_workflow_signal.send(ctx)
         except BaseException as e:
-            events.on_failure_workflow_signal.send(self._workflow_context, exception=e)
+            events.on_failure_workflow_signal.send(ctx, exception=e)
             raise
 
-    def cancel_execution(self):
+    @staticmethod
+    def cancel_execution(ctx):
         """
         Send a cancel request to the engine. If execution already started, execution status
         will be modified to 'cancelling' status. If execution is in pending mode, execution status
         will be modified to 'cancelled' directly.
         """
-        events.on_cancelling_workflow_signal.send(self._workflow_context)
+        events.on_cancelling_workflow_signal.send(ctx)
 
-    def _is_cancel(self):
-        return self._workflow_context.execution.status in (models.Execution.CANCELLING,
-                                                           models.Execution.CANCELLED)
+    @staticmethod
+    def _is_cancel(ctx):
+        execution = ctx.model.execution.refresh(ctx.execution)
+        return execution.status in (models.Execution.CANCELLING, models.Execution.CANCELLED)
 
-    def _executable_tasks(self):
+    def _executable_tasks(self, ctx):
         now = datetime.utcnow()
-        return (task for task in self._tasks_iter()
-                if task.is_waiting() and
-                task.due_at <= now and
-                not self._task_has_dependencies(task))
-
-    def _ended_tasks(self):
-        return (task for task in self._tasks_iter() if task.has_ended())
+        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)
+        )
 
-    def _task_has_dependencies(self, task):
-        return len(self._execution_graph.pred.get(task.id, {})) > 0
+    @staticmethod
+    def _ended_tasks(ctx, executing_tasks):
+        for task in executing_tasks:
+            if task.has_ended() and task in ctx._graph:
+                yield task
 
-    def _all_tasks_consumed(self):
-        return len(self._execution_graph.node) == 0
+    @staticmethod
+    def _task_has_dependencies(ctx, task):
+        return len(ctx._graph.pred.get(task, [])) > 0
 
-    def _tasks_iter(self):
-        for _, data in self._execution_graph.nodes_iter(data=True):
-            task = data['task']
-            if isinstance(task, engine_task.OperationTask):
-                if not task.model_task.has_ended():
-                    self._workflow_context.model.task.refresh(task.model_task)
-            yield task
+    @staticmethod
+    def _all_tasks_consumed(ctx):
+        return len(ctx._graph.node) == 0
 
     @staticmethod
-    def _handle_executable_task(task):
-        if isinstance(task, engine_task.OperationTask):
-            events.sent_task_signal.send(task)
-        task.execute()
+    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):
+        task_executor = self._executors[task._executor]
+
+        # If the task is a stub, a default context is provided, else it should hold the context cls
+        context_cls = operation.BaseOperationContext if task._stub_type else task._context_cls
+        op_ctx = context_cls(
+            model_storage=ctx.model,
+            resource_storage=ctx.resource,
+            workdir=ctx._workdir,
+            task_id=task.id,
+            actor_id=task.actor.id if task.actor else None,
+            service_id=task.execution.service.id,
+            execution_id=task.execution.id,
+            name=task.name
+        )
+
+        executing_tasks.append(task)
+
+        if not task._stub_type:
+            events.sent_task_signal.send(op_ctx)
+        task_executor.execute(op_ctx)
 
-    def _handle_ended_tasks(self, task):
+    @staticmethod
+    def _handle_ended_tasks(ctx, task, executing_tasks):
+        executing_tasks.remove(task)
         if task.status == models.Task.FAILED and not task.ignore_failure:
             raise exceptions.ExecutorException('Workflow failed')
         else:
-            self._execution_graph.remove_node(task.id)
+            ctx._graph.remove_node(task)

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e119df5/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 669fb43..2d71d2a 100644
--- a/aria/orchestrator/workflows/core/events_handler.py
+++ b/aria/orchestrator/workflows/core/events_handler.py
@@ -30,120 +30,121 @@ from ... import exceptions
 
 
 @events.sent_task_signal.connect
-def _task_sent(task, *args, **kwargs):
-    with task._update():
-        task.status = task.SENT
+def _task_sent(ctx, *args, **kwargs):
+    with ctx.persist_changes:
+        ctx.task.status = ctx.task.SENT
 
 
 @events.start_task_signal.connect
-def _task_started(task, *args, **kwargs):
-    with task._update():
-        task.started_at = datetime.utcnow()
-        task.status = task.STARTED
-    _update_node_state_if_necessary(task, is_transitional=True)
+def _task_started(ctx, *args, **kwargs):
+    with ctx.persist_changes:
+        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(task, exception, *args, **kwargs):
-    with task._update():
+def _task_failed(ctx, exception, *args, **kwargs):
+    with ctx.persist_changes:
         should_retry = all([
             not isinstance(exception, exceptions.TaskAbortException),
-            task.attempts_count < task.max_attempts or task.max_attempts == task.INFINITE_RETRIES,
-            # ignore_failure check here means the task will not be retries and it will be marked
+            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 task.ignore_failure
+            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 = task.retry_interval
-            task.status = task.RETRYING
-            task.attempts_count += 1
-            task.due_at = datetime.utcnow() + timedelta(seconds=retry_interval)
+                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:
-            task.ended_at = datetime.utcnow()
-            task.status = task.FAILED
+            ctx.task.ended_at = datetime.utcnow()
+            ctx.task.status = ctx.task.FAILED
 
 
 @events.on_success_task_signal.connect
-def _task_succeeded(task, *args, **kwargs):
-    with task._update():
-        task.ended_at = datetime.utcnow()
-        task.status = task.SUCCESS
+def _task_succeeded(ctx, *args, **kwargs):
+    with ctx.persist_changes:
+        ctx.task.ended_at = datetime.utcnow()
+        ctx.task.status = ctx.task.SUCCESS
 
-    _update_node_state_if_necessary(task)
+        _update_node_state_if_necessary(ctx)
 
 
 @events.start_workflow_signal.connect
 def _workflow_started(workflow_context, *args, **kwargs):
-    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()
-    workflow_context.execution = execution
+    with workflow_context.persist_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()
 
 
 @events.on_failure_workflow_signal.connect
 def _workflow_failed(workflow_context, exception, *args, **kwargs):
-    execution = workflow_context.execution
-    execution.error = str(exception)
-    execution.status = execution.FAILED
-    execution.ended_at = datetime.utcnow()
-    workflow_context.execution = execution
+    with workflow_context.persist_changes:
+        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):
-    execution = workflow_context.execution
-    execution.status = execution.SUCCEEDED
-    execution.ended_at = datetime.utcnow()
-    workflow_context.execution = execution
+    with workflow_context.persist_changes:
+        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):
-    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()
-        workflow_context.execution = execution
+    with workflow_context.persist_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()
 
 
 @events.on_cancelling_workflow_signal.connect
 def _workflow_cancelling(workflow_context, *args, **kwargs):
-    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(task, is_transitional=False):
+    with workflow_context.persist_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
+
+
+def _update_node_state_if_necessary(ctx, is_transitional=False):
     # TODO: this is not the right way to check! the interface name is arbitrary
     # and also will *never* be the type name
-    model_task = task.model_task
-    node = model_task.node if model_task is not None else None
+    node = ctx.task.node if ctx.task is not None else None
     if (node is not None) and \
-        (task.interface_name in ('Standard', 'tosca.interfaces.node.lifecycle.Standard')):
-        state = node.determine_state(op_name=task.operation_name, is_transitional=is_transitional)
+        (ctx.task.interface_name in ('Standard', 'tosca.interfaces.node.lifecycle.Standard')):
+        state = node.determine_state(op_name=ctx.task.operation_name,
+                                     is_transitional=is_transitional)
         if state:
             node.state = state
-            task.context.model.node.update(node)
+            ctx.model.node.update(node)
 
 
 def _log_tried_to_cancel_execution_but_it_already_ended(workflow_context, status):

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e119df5/aria/orchestrator/workflows/core/task.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/core/task.py b/aria/orchestrator/workflows/core/task.py
deleted file mode 100644
index d732f09..0000000
--- a/aria/orchestrator/workflows/core/task.py
+++ /dev/null
@@ -1,271 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License.  You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-"""
-Workflow tasks
-"""
-
-from contextlib import contextmanager
-from datetime import datetime
-from functools import (
-    partial,
-    wraps,
-)
-
-
-from ....modeling import models
-from ...context import operation as operation_context
-from .. import exceptions
-
-
-def _locked(func=None):
-    if func is None:
-        return partial(_locked, func=_locked)
-
-    @wraps(func)
-    def _wrapper(self, value, **kwargs):
-        if self._update_fields is None:
-            raise exceptions.TaskException('Task is not in update mode')
-        return func(self, value, **kwargs)
-    return _wrapper
-
-
-class BaseTask(object):
-    """
-    Base class for Task objects
-    """
-
-    def __init__(self, id, executor, *args, **kwargs):
-        super(BaseTask, self).__init__(*args, **kwargs)
-        self._id = id
-        self._executor = executor
-
-    def execute(self):
-        return self._executor.execute(self)
-
-    @property
-    def id(self):
-        """
-        :return: the task's id
-        """
-        return self._id
-
-
-class StubTask(BaseTask):
-    """
-    Base stub task for marker user tasks that only mark the start/end of a workflow
-    or sub-workflow
-    """
-    STARTED = models.Task.STARTED
-    SUCCESS = models.Task.SUCCESS
-
-    def __init__(self, *args, **kwargs):
-        super(StubTask, self).__init__(*args, **kwargs)
-        self.status = models.Task.PENDING
-        self.due_at = datetime.utcnow()
-
-    def has_ended(self):
-        return self.status == self.SUCCESS
-
-    def is_waiting(self):
-        return not self.has_ended()
-
-
-class StartWorkflowTask(StubTask):
-    """
-    Task marking a workflow start
-    """
-    pass
-
-
-class EndWorkflowTask(StubTask):
-    """
-    Task marking a workflow end
-    """
-    pass
-
-
-class StartSubWorkflowTask(StubTask):
-    """
-    Task marking a subworkflow start
-    """
-    pass
-
-
-class EndSubWorkflowTask(StubTask):
-    """
-    Task marking a subworkflow end
-    """
-    pass
-
-
-class OperationTask(BaseTask):
-    """
-    Operation task
-    """
-    def __init__(self, api_task, *args, **kwargs):
-        # If no executor is provided, we infer that this is an empty task which does not need to be
-        # executed.
-        super(OperationTask, self).__init__(id=api_task.id, *args, **kwargs)
-        self._workflow_context = api_task._workflow_context
-        self.interface_name = api_task.interface_name
-        self.operation_name = api_task.operation_name
-        model_storage = api_task._workflow_context.model
-
-        actor = getattr(api_task.actor, '_wrapped', api_task.actor)
-
-        base_task_model = model_storage.task.model_cls
-        if isinstance(actor, models.Node):
-            context_cls = operation_context.NodeOperationContext
-            create_task_model = base_task_model.for_node
-        elif isinstance(actor, models.Relationship):
-            context_cls = operation_context.RelationshipOperationContext
-            create_task_model = base_task_model.for_relationship
-        else:
-            raise RuntimeError('No operation context could be created for {actor.model_cls}'
-                               .format(actor=actor))
-
-        task_model = create_task_model(
-            name=api_task.name,
-            actor=actor,
-            status=base_task_model.PENDING,
-            max_attempts=api_task.max_attempts,
-            retry_interval=api_task.retry_interval,
-            ignore_failure=api_task.ignore_failure,
-            execution=self._workflow_context.execution,
-
-            # Only non-stub tasks have these fields
-            plugin=api_task.plugin,
-            function=api_task.function,
-            arguments=api_task.arguments
-        )
-        self._workflow_context.model.task.put(task_model)
-
-        self._ctx = context_cls(name=api_task.name,
-                                model_storage=self._workflow_context.model,
-                                resource_storage=self._workflow_context.resource,
-                                service_id=self._workflow_context._service_id,
-                                task_id=task_model.id,
-                                actor_id=actor.id,
-                                execution_id=self._workflow_context._execution_id,
-                                workdir=self._workflow_context._workdir)
-        self._task_id = task_model.id
-        self._update_fields = None
-
-    @contextmanager
-    def _update(self):
-        """
-        A context manager which puts the task into update mode, enabling fields update.
-        :yields: None
-        """
-        self._update_fields = {}
-        try:
-            yield
-            for key, value in self._update_fields.items():
-                setattr(self.model_task, key, value)
-            self.model_task = self.model_task
-        finally:
-            self._update_fields = None
-
-    @property
-    def model_task(self):
-        """
-        Returns the task model in storage
-        :return: task in storage
-        """
-        return self._workflow_context.model.task.get(self._task_id)
-
-    @model_task.setter
-    def model_task(self, value):
-        self._workflow_context.model.task.put(value)
-
-    @property
-    def context(self):
-        """
-        Contexts for the operation
-        :return:
-        """
-        return self._ctx
-
-    @property
-    def status(self):
-        """
-        Returns the task status
-        :return: task status
-        """
-        return self.model_task.status
-
-    @status.setter
-    @_locked
-    def status(self, value):
-        self._update_fields['status'] = value
-
-    @property
-    def started_at(self):
-        """
-        Returns when the task started
-        :return: when task started
-        """
-        return self.model_task.started_at
-
-    @started_at.setter
-    @_locked
-    def started_at(self, value):
-        self._update_fields['started_at'] = value
-
-    @property
-    def ended_at(self):
-        """
-        Returns when the task ended
-        :return: when task ended
-        """
-        return self.model_task.ended_at
-
-    @ended_at.setter
-    @_locked
-    def ended_at(self, value):
-        self._update_fields['ended_at'] = value
-
-    @property
-    def attempts_count(self):
-        """
-        Returns the attempts count for the task
-        :return: attempts count
-        """
-        return self.model_task.attempts_count
-
-    @attempts_count.setter
-    @_locked
-    def attempts_count(self, value):
-        self._update_fields['attempts_count'] = value
-
-    @property
-    def due_at(self):
-        """
-        Returns the minimum datetime in which the task can be executed
-        :return: eta
-        """
-        return self.model_task.due_at
-
-    @due_at.setter
-    @_locked
-    def due_at(self, value):
-        self._update_fields['due_at'] = value
-
-    def __getattr__(self, attr):
-        try:
-            return getattr(self.model_task, attr)
-        except AttributeError:
-            return super(OperationTask, self).__getattribute__(attr)

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e119df5/aria/orchestrator/workflows/core/translation.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/core/translation.py b/aria/orchestrator/workflows/core/translation.py
deleted file mode 100644
index fec108b..0000000
--- a/aria/orchestrator/workflows/core/translation.py
+++ /dev/null
@@ -1,109 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements.  See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License.  You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-"""
-Translation of user graph's API to the execution graph
-"""
-
-from .. import api
-from ..executor import base
-from . import task as core_task
-
-
-def build_execution_graph(
-        task_graph,
-        execution_graph,
-        default_executor,
-        start_cls=core_task.StartWorkflowTask,
-        end_cls=core_task.EndWorkflowTask,
-        depends_on=()):
-    """
-    Translates the user graph to the execution graph
-    :param task_graph: The user's graph
-    :param workflow_context: The workflow
-    :param execution_graph: The execution graph that is being built
-    :param start_cls: internal use
-    :param end_cls: internal use
-    :param depends_on: internal use
-    """
-    # Insert start marker
-    start_task = start_cls(id=_start_graph_suffix(task_graph.id), executor=base.StubTaskExecutor())
-    _add_task_and_dependencies(execution_graph, start_task, depends_on)
-
-    for api_task in task_graph.topological_order(reverse=True):
-        dependencies = task_graph.get_dependencies(api_task)
-        operation_dependencies = _get_tasks_from_dependencies(
-            execution_graph, dependencies, default=[start_task])
-
-        if isinstance(api_task, api.task.OperationTask):
-            operation_task = core_task.OperationTask(api_task, executor=default_executor)
-            _add_task_and_dependencies(execution_graph, operation_task, operation_dependencies)
-        elif isinstance(api_task, api.task.WorkflowTask):
-            # Build the graph recursively while adding start and end markers
-            build_execution_graph(
-                task_graph=api_task,
-                execution_graph=execution_graph,
-                default_executor=default_executor,
-                start_cls=core_task.StartSubWorkflowTask,
-                end_cls=core_task.EndSubWorkflowTask,
-                depends_on=operation_dependencies
-            )
-        elif isinstance(api_task, api.task.StubTask):
-            stub_task = core_task.StubTask(id=api_task.id, executor=base.StubTaskExecutor())
-            _add_task_and_dependencies(execution_graph, stub_task, operation_dependencies)
-        else:
-            raise RuntimeError('Undefined state')
-
-    # Insert end marker
-    workflow_dependencies = _get_tasks_from_dependencies(
-        execution_graph,
-        _get_non_dependency_tasks(task_graph),
-        default=[start_task])
-    end_task = end_cls(id=_end_graph_suffix(task_graph.id), executor=base.StubTaskExecutor())
-    _add_task_and_dependencies(execution_graph, end_task, workflow_dependencies)
-
-
-def _add_task_and_dependencies(execution_graph, operation_task, operation_dependencies=()):
-    execution_graph.add_node(operation_task.id, task=operation_task)
-    for dependency in operation_dependencies:
-        execution_graph.add_edge(dependency.id, operation_task.id)
-
-
-def _get_tasks_from_dependencies(execution_graph, dependencies, default=()):
-    """
-    Returns task list from dependencies.
-    """
-    tasks = []
-    for dependency in dependencies:
-        if isinstance(dependency, (api.task.OperationTask, api.task.StubTask)):
-            dependency_id = dependency.id
-        else:
-            dependency_id = _end_graph_suffix(dependency.id)
-        tasks.append(execution_graph.node[dependency_id]['task'])
-    return tasks or default
-
-
-def _start_graph_suffix(id):
-    return '{0}-Start'.format(id)
-
-
-def _end_graph_suffix(id):
-    return '{0}-End'.format(id)
-
-
-def _get_non_dependency_tasks(graph):
-    for task in graph.tasks:
-        if len(list(graph.get_dependents(task))) == 0:
-            yield task

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e119df5/aria/orchestrator/workflows/events_logging.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/events_logging.py b/aria/orchestrator/workflows/events_logging.py
index 036c1f7..4cee867 100644
--- a/aria/orchestrator/workflows/events_logging.py
+++ b/aria/orchestrator/workflows/events_logging.py
@@ -34,31 +34,32 @@ def _get_task_name(task):
 
 
 @events.start_task_signal.connect
-def _start_task_handler(task, **kwargs):
+def _start_task_handler(ctx, **kwargs):
     # If the task has no function this is an empty task.
-    if task.function:
+    if ctx.task.function:
         suffix = 'started...'
-        logger = task.context.logger.info
+        logger = ctx.logger.info
     else:
         suffix = 'has no implementation'
-        logger = task.context.logger.debug
+        logger = ctx.logger.debug
 
     logger('{name} {task.interface_name}.{task.operation_name} {suffix}'.format(
-        name=_get_task_name(task), task=task, suffix=suffix))
+        name=_get_task_name(ctx.task), task=ctx.task, suffix=suffix))
+
 
 @events.on_success_task_signal.connect
-def _success_task_handler(task, **kwargs):
-    if not task.function:
+def _success_task_handler(ctx, **kwargs):
+    if not ctx.task.function:
         return
-    task.context.logger.info('{name} {task.interface_name}.{task.operation_name} successful'
-                             .format(name=_get_task_name(task), task=task))
+    ctx.logger.info('{name} {task.interface_name}.{task.operation_name} successful'
+                    .format(name=_get_task_name(ctx.task), task=ctx.task))
 
 
 @events.on_failure_task_signal.connect
-def _failure_operation_handler(task, traceback, **kwargs):
-    task.context.logger.error(
+def _failure_operation_handler(ctx, traceback, **kwargs):
+    ctx.logger.error(
         '{name} {task.interface_name}.{task.operation_name} failed'
-        .format(name=_get_task_name(task), task=task), extra=dict(traceback=traceback)
+        .format(name=_get_task_name(ctx.task), task=ctx.task), extra=dict(traceback=traceback)
     )
 
 

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e119df5/aria/orchestrator/workflows/executor/base.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/executor/base.py b/aria/orchestrator/workflows/executor/base.py
index 7fece6f..6a3c9d2 100644
--- a/aria/orchestrator/workflows/executor/base.py
+++ b/aria/orchestrator/workflows/executor/base.py
@@ -28,19 +28,19 @@ class BaseExecutor(logger.LoggerMixin):
     def _execute(self, task):
         raise NotImplementedError
 
-    def execute(self, task):
+    def execute(self, ctx):
         """
         Execute a task
         :param task: task to execute
         """
-        if task.function:
-            self._execute(task)
+        if ctx.task.function:
+            self._execute(ctx)
         else:
             # In this case the task is missing a function. This task still gets to an
-            # executor, but since there is nothing to run, we by default simply skip the execution
-            # itself.
-            self._task_started(task)
-            self._task_succeeded(task)
+            # executor, but since there is nothing to run, we by default simply skip the
+            # execution itself.
+            self._task_started(ctx)
+            self._task_succeeded(ctx)
 
     def close(self):
         """
@@ -49,18 +49,19 @@ class BaseExecutor(logger.LoggerMixin):
         pass
 
     @staticmethod
-    def _task_started(task):
-        events.start_task_signal.send(task)
+    def _task_started(ctx):
+        events.start_task_signal.send(ctx)
 
     @staticmethod
-    def _task_failed(task, exception, traceback=None):
-        events.on_failure_task_signal.send(task, exception=exception, traceback=traceback)
+    def _task_failed(ctx, exception, traceback=None):
+        events.on_failure_task_signal.send(ctx, exception=exception, traceback=traceback)
 
     @staticmethod
-    def _task_succeeded(task):
-        events.on_success_task_signal.send(task)
+    def _task_succeeded(ctx):
+        events.on_success_task_signal.send(ctx)
 
 
 class StubTaskExecutor(BaseExecutor):                                                               # pylint: disable=abstract-method
-    def execute(self, task):
-        task.status = task.SUCCESS
+    def execute(self, ctx, *args, **kwargs):
+        with ctx.persist_changes:
+            ctx.task.status = ctx.task.SUCCESS

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e119df5/aria/orchestrator/workflows/executor/dry.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/executor/dry.py b/aria/orchestrator/workflows/executor/dry.py
index 72080b4..9d86125 100644
--- a/aria/orchestrator/workflows/executor/dry.py
+++ b/aria/orchestrator/workflows/executor/dry.py
@@ -18,37 +18,36 @@ Dry executor
 """
 from datetime import datetime
 
-from .base import BaseExecutor
+from . import base
 
 
-class DryExecutor(BaseExecutor):                                                                    # pylint: disable=abstract-method
+class DryExecutor(base.BaseExecutor):                                                                    # pylint: disable=abstract-method
     """
     Executor which dry runs tasks - prints task information without causing any side effects
     """
-    def execute(self, task):
-        # updating the task manually instead of calling self._task_started(task),
-        # to avoid any side effects raising that event might cause
-        with task._update():
-            task.started_at = datetime.utcnow()
-            task.status = task.STARTED
-
-        dry_msg = '<dry> {name} {task.interface_name}.{task.operation_name} {suffix}'
-        logger = task.context.logger.info if task.function else task.context.logger.debug
-
-        if hasattr(task.actor, 'source_node'):
-            name = '{source_node.name}->{target_node.name}'.format(
-                source_node=task.actor.source_node, target_node=task.actor.target_node)
-        else:
-            name = task.actor.name
-
-        if task.function:
-            logger(dry_msg.format(name=name, task=task, suffix='started...'))
-            logger(dry_msg.format(name=name, task=task, suffix='successful'))
-        else:
-            logger(dry_msg.format(name=name, task=task, suffix='has no implementation'))
-
-        # updating the task manually instead of calling self._task_succeeded(task),
-        # to avoid any side effects raising that event might cause
-        with task._update():
-            task.ended_at = datetime.utcnow()
-            task.status = task.SUCCESS
+    def execute(self, ctx):
+        with ctx.persist_changes:
+            # updating the task manually instead of calling self._task_started(task),
+            # to avoid any side effects raising that event might cause
+            ctx.task.started_at = datetime.utcnow()
+            ctx.task.status = ctx.task.STARTED
+
+            dry_msg = '<dry> {name} {task.interface_name}.{task.operation_name} {suffix}'
+            logger = ctx.logger.info if ctx.task.function else ctx.logger.debug
+
+            if hasattr(ctx.task.actor, 'source_node'):
+                name = '{source_node.name}->{target_node.name}'.format(
+                    source_node=ctx.task.actor.source_node, target_node=ctx.task.actor.target_node)
+            else:
+                name = ctx.task.actor.name
+
+            if ctx.task.function:
+                logger(dry_msg.format(name=name, task=ctx.task, suffix='started...'))
+                logger(dry_msg.format(name=name, task=ctx.task, suffix='successful'))
+            else:
+                logger(dry_msg.format(name=name, task=ctx.task, suffix='has no implementation'))
+
+            # updating the task manually instead of calling self._task_succeeded(task),
+            # to avoid any side effects raising that event might cause
+            ctx.task.ended_at = datetime.utcnow()
+            ctx.task.status = ctx.task.SUCCESS

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e119df5/aria/orchestrator/workflows/executor/process.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/executor/process.py b/aria/orchestrator/workflows/executor/process.py
index 634f1f2..8518b33 100644
--- a/aria/orchestrator/workflows/executor/process.py
+++ b/aria/orchestrator/workflows/executor/process.py
@@ -113,17 +113,17 @@ class ProcessExecutor(base.BaseExecutor):
         self._server_socket.close()
         self._listener_thread.join(timeout=60)
 
-    def _execute(self, task):
+    def _execute(self, ctx):
         self._check_closed()
-        self._tasks[task.id] = task
+        self._tasks[ctx.task.id] = ctx
 
         # Temporary file used to pass arguments to the started subprocess
         file_descriptor, arguments_json_path = tempfile.mkstemp(prefix='executor-', suffix='.json')
         os.close(file_descriptor)
         with open(arguments_json_path, 'wb') as f:
-            f.write(pickle.dumps(self._create_arguments_dict(task)))
+            f.write(pickle.dumps(self._create_arguments_dict(ctx)))
 
-        env = self._construct_subprocess_env(task=task)
+        env = self._construct_subprocess_env(task=ctx.task)
         # Asynchronously start the operation in a subprocess
         subprocess.Popen(
             '{0} {1} {2}'.format(sys.executable, __file__, arguments_json_path),
@@ -137,13 +137,13 @@ class ProcessExecutor(base.BaseExecutor):
         if self._stopped:
             raise RuntimeError('Executor closed')
 
-    def _create_arguments_dict(self, task):
+    def _create_arguments_dict(self, ctx):
         return {
-            'task_id': task.id,
-            'function': task.function,
-            'operation_arguments': dict(arg.unwrapped for arg in task.arguments.values()),
+            'task_id': ctx.task.id,
+            'function': ctx.task.function,
+            'operation_arguments': dict(arg.unwrapped for arg in ctx.task.arguments.values()),
             'port': self._server_port,
-            'context': task.context.serialization_dict,
+            'context': ctx.serialization_dict,
         }
 
     def _construct_subprocess_env(self, task):

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e119df5/aria/orchestrator/workflows/executor/thread.py
----------------------------------------------------------------------
diff --git a/aria/orchestrator/workflows/executor/thread.py b/aria/orchestrator/workflows/executor/thread.py
index 56a56a5..8c447b6 100644
--- a/aria/orchestrator/workflows/executor/thread.py
+++ b/aria/orchestrator/workflows/executor/thread.py
@@ -46,8 +46,8 @@ class ThreadExecutor(BaseExecutor):
             thread.start()
             self._pool.append(thread)
 
-    def _execute(self, task):
-        self._queue.put(task)
+    def _execute(self, ctx):
+        self._queue.put(ctx)
 
     def close(self):
         self._stopped = True
@@ -57,15 +57,15 @@ class ThreadExecutor(BaseExecutor):
     def _processor(self):
         while not self._stopped:
             try:
-                task = self._queue.get(timeout=1)
-                self._task_started(task)
+                ctx = self._queue.get(timeout=1)
+                self._task_started(ctx)
                 try:
-                    task_func = imports.load_attribute(task.function)
-                    arguments = dict(arg.unwrapped for arg in task.arguments.values())
-                    task_func(ctx=task.context, **arguments)
-                    self._task_succeeded(task)
+                    task_func = imports.load_attribute(ctx.task.function)
+                    arguments = dict(arg.unwrapped for arg in ctx.task.arguments.values())
+                    task_func(ctx=ctx, **arguments)
+                    self._task_succeeded(ctx)
                 except BaseException as e:
-                    self._task_failed(task,
+                    self._task_failed(ctx,
                                       exception=e,
                                       traceback=exceptions.get_exception_as_string(*sys.exc_info()))
             # Daemon threads

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e119df5/tests/__init__.py
----------------------------------------------------------------------
diff --git a/tests/__init__.py b/tests/__init__.py
index d2858d2..ace30c8 100644
--- a/tests/__init__.py
+++ b/tests/__init__.py
@@ -15,4 +15,6 @@
 
 import os
 
+from . import storage, mock
+
 ROOT_DIR = os.path.dirname(os.path.dirname(__file__))

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e119df5/tests/orchestrator/context/__init__.py
----------------------------------------------------------------------
diff --git a/tests/orchestrator/context/__init__.py b/tests/orchestrator/context/__init__.py
index 4fde0a7..086a066 100644
--- a/tests/orchestrator/context/__init__.py
+++ b/tests/orchestrator/context/__init__.py
@@ -15,7 +15,7 @@
 
 import sys
 
-from aria.orchestrator.workflows.core import engine
+from aria.orchestrator.workflows.core import engine, compile
 
 
 def op_path(func, module_path=None):
@@ -25,5 +25,8 @@ def op_path(func, module_path=None):
 
 def execute(workflow_func, workflow_context, executor):
     graph = workflow_func(ctx=workflow_context)
-    eng = engine.Engine(executor=executor, workflow_context=workflow_context, tasks_graph=graph)
-    eng.execute()
+
+    compile.create_execution_tasks(workflow_context, graph, executor.__class__)
+    eng = engine.Engine(executors={executor.__class__: executor})
+
+    eng.execute(workflow_context)

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e119df5/tests/orchestrator/context/test_operation.py
----------------------------------------------------------------------
diff --git a/tests/orchestrator/context/test_operation.py b/tests/orchestrator/context/test_operation.py
index 3dcfaa2..9550d12 100644
--- a/tests/orchestrator/context/test_operation.py
+++ b/tests/orchestrator/context/test_operation.py
@@ -50,21 +50,12 @@ def ctx(tmpdir):
 
 
 @pytest.fixture
-def process_executor():
-    ex = process.ProcessExecutor(**dict(python_path=tests.ROOT_DIR))
-    try:
-        yield ex
-    finally:
-        ex.close()
-
-
-@pytest.fixture
 def thread_executor():
-    ex = thread.ThreadExecutor()
+    result = thread.ThreadExecutor()
     try:
-        yield ex
+        yield result
     finally:
-        ex.close()
+        result.close()
 
 
 @pytest.fixture
@@ -266,12 +257,12 @@ def test_plugin_workdir(ctx, thread_executor, tmpdir):
     (process.ProcessExecutor, {'python_path': [tests.ROOT_DIR]}),
 ])
 def executor(request):
-    executor_cls, executor_kwargs = request.param
-    result = executor_cls(**executor_kwargs)
+    ex_cls, kwargs = request.param
+    ex = ex_cls(**kwargs)
     try:
-        yield result
+        yield ex
     finally:
-        result.close()
+        ex.close()
 
 
 def test_node_operation_logging(ctx, executor):
@@ -304,7 +295,6 @@ def test_node_operation_logging(ctx, executor):
                 arguments=arguments
             )
         )
-
     execute(workflow_func=basic_workflow, workflow_context=ctx, executor=executor)
     _assert_loggins(ctx, arguments)
 
@@ -418,7 +408,7 @@ def _assert_loggins(ctx, arguments):
     assert len(executions) == 1
     execution = executions[0]
 
-    tasks = ctx.model.task.list()
+    tasks = ctx.model.task.list(filters={'_stub_type': None})
     assert len(tasks) == 1
     task = tasks[0]
     assert len(task.logs) == 4

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e119df5/tests/orchestrator/context/test_serialize.py
----------------------------------------------------------------------
diff --git a/tests/orchestrator/context/test_serialize.py b/tests/orchestrator/context/test_serialize.py
index 0919e81..5db5b63 100644
--- a/tests/orchestrator/context/test_serialize.py
+++ b/tests/orchestrator/context/test_serialize.py
@@ -16,7 +16,7 @@
 import pytest
 
 from aria.orchestrator.workflows import api
-from aria.orchestrator.workflows.core import engine
+from aria.orchestrator.workflows.core import engine, compile
 from aria.orchestrator.workflows.executor import process
 from aria.orchestrator import workflow, operation
 import tests
@@ -48,15 +48,15 @@ def test_serialize_operation_context(context, executor, tmpdir):
     context.model.node.update(node)
 
     graph = _mock_workflow(ctx=context)  # pylint: disable=no-value-for-parameter
-    eng = engine.Engine(executor=executor, workflow_context=context, tasks_graph=graph)
-    eng.execute()
+    compile.create_execution_tasks(context, graph, executor.__class__)
+    eng = engine.Engine({executor.__class__: executor})
+    eng.execute(context)
 
 
 @workflow
 def _mock_workflow(ctx, graph):
     node = ctx.model.node.get_by_name(mock.models.DEPENDENCY_NODE_NAME)
-    task = api.task.OperationTask(node, interface_name='test', operation_name='op')
-    graph.add_tasks(task)
+    graph.add_tasks(api.task.OperationTask(node, interface_name='test', operation_name='op'))
     return graph
 
 

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e119df5/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 f667460..1695320 100644
--- a/tests/orchestrator/execution_plugin/test_local.py
+++ b/tests/orchestrator/execution_plugin/test_local.py
@@ -28,7 +28,7 @@ from aria.orchestrator.execution_plugin.exceptions import ProcessException
 from aria.orchestrator.execution_plugin import local
 from aria.orchestrator.execution_plugin import constants
 from aria.orchestrator.workflows.executor import process
-from aria.orchestrator.workflows.core import engine
+from aria.orchestrator.workflows.core import engine, compile
 
 from tests import mock
 from tests import storage
@@ -500,11 +500,9 @@ if __name__ == '__main__':
                 arguments=arguments))
             return graph
         tasks_graph = mock_workflow(ctx=workflow_context)  # pylint: disable=no-value-for-parameter
-        eng = engine.Engine(
-            executor=executor,
-            workflow_context=workflow_context,
-            tasks_graph=tasks_graph)
-        eng.execute()
+        compile.create_execution_tasks(workflow_context, tasks_graph, executor.__class__)
+        eng = engine.Engine({executor.__class__: executor})
+        eng.execute(workflow_context)
         return workflow_context.model.node.get_by_name(
             mock.models.DEPENDENCY_NODE_NAME).attributes
 

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e119df5/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 8c4dd2d..fb1dc09 100644
--- a/tests/orchestrator/execution_plugin/test_ssh.py
+++ b/tests/orchestrator/execution_plugin/test_ssh.py
@@ -29,7 +29,7 @@ from aria.orchestrator import events
 from aria.orchestrator import workflow
 from aria.orchestrator.workflows import api
 from aria.orchestrator.workflows.executor import process
-from aria.orchestrator.workflows.core import engine
+from aria.orchestrator.workflows.core import engine, compile
 from aria.orchestrator.workflows.exceptions import ExecutorException
 from aria.orchestrator.exceptions import TaskAbortException, TaskRetryException
 from aria.orchestrator.execution_plugin import operations
@@ -254,11 +254,10 @@ class TestWithActualSSHServer(object):
             graph.sequence(*ops)
             return graph
         tasks_graph = mock_workflow(ctx=self._workflow_context)  # pylint: disable=no-value-for-parameter
-        eng = engine.Engine(
-            executor=self._executor,
-            workflow_context=self._workflow_context,
-            tasks_graph=tasks_graph)
-        eng.execute()
+        compile.create_execution_tasks(
+            self._workflow_context, tasks_graph, self._executor.__class__)
+        eng = engine.Engine({self._executor.__class__: self._executor})
+        eng.execute(self._workflow_context)
         return self._workflow_context.model.node.get_by_name(
             mock.models.DEPENDENCY_NODE_NAME).attributes
 

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e119df5/tests/orchestrator/test_workflow_runner.py
----------------------------------------------------------------------
diff --git a/tests/orchestrator/test_workflow_runner.py b/tests/orchestrator/test_workflow_runner.py
index c5a62ae..40f9035 100644
--- a/tests/orchestrator/test_workflow_runner.py
+++ b/tests/orchestrator/test_workflow_runner.py
@@ -96,20 +96,20 @@ def test_default_executor(request):
     # validates the ProcessExecutor is used by the workflow runner by default
     mock_workflow = _setup_mock_workflow_in_service(request)
 
-    with mock.patch('aria.orchestrator.workflow_runner.Engine') as mock_engine_cls:
+    with mock.patch('aria.orchestrator.workflow_runner.engine.Engine') as mock_engine_cls:
         _create_workflow_runner(request, mock_workflow)
         _, engine_kwargs = mock_engine_cls.call_args
-        assert isinstance(engine_kwargs.get('executor'), ProcessExecutor)
+        assert isinstance(engine_kwargs.get('executors').values()[0], ProcessExecutor)
 
 
 def test_custom_executor(request):
     mock_workflow = _setup_mock_workflow_in_service(request)
 
     custom_executor = mock.MagicMock()
-    with mock.patch('aria.orchestrator.workflow_runner.Engine') as mock_engine_cls:
+    with mock.patch('aria.orchestrator.workflow_runner.engine.Engine') as mock_engine_cls:
         _create_workflow_runner(request, mock_workflow, executor=custom_executor)
         _, engine_kwargs = mock_engine_cls.call_args
-        assert engine_kwargs.get('executor') == custom_executor
+        assert engine_kwargs.get('executors').values()[0] == custom_executor
 
 
 def test_task_configuration_parameters(request):
@@ -117,48 +117,47 @@ def test_task_configuration_parameters(request):
 
     task_max_attempts = 5
     task_retry_interval = 7
-    with mock.patch('aria.orchestrator.workflow_runner.Engine') as mock_engine_cls:
+    with mock.patch('aria.orchestrator.workflow_runner.engine.Engine.execute') as \
+            mock_engine_execute:
         _create_workflow_runner(request, mock_workflow, task_max_attempts=task_max_attempts,
-                                task_retry_interval=task_retry_interval)
-        _, engine_kwargs = mock_engine_cls.call_args
-        assert engine_kwargs['workflow_context']._task_max_attempts == task_max_attempts
-        assert engine_kwargs['workflow_context']._task_retry_interval == task_retry_interval
+                                task_retry_interval=task_retry_interval).execute()
+        _, engine_kwargs = mock_engine_execute.call_args
+        assert engine_kwargs['ctx']._task_max_attempts == task_max_attempts
+        assert engine_kwargs['ctx']._task_retry_interval == task_retry_interval
 
 
 def test_execute(request, service):
     mock_workflow = _setup_mock_workflow_in_service(request)
 
     mock_engine = mock.MagicMock()
-    with mock.patch('aria.orchestrator.workflow_runner.Engine', return_value=mock_engine) \
-            as mock_engine_cls:
+    with mock.patch('aria.orchestrator.workflow_runner.engine.Engine.execute',
+                    return_value=mock_engine) as mock_engine_execute:
         workflow_runner = _create_workflow_runner(request, mock_workflow)
+        workflow_runner.execute()
 
-        _, engine_kwargs = mock_engine_cls.call_args
-        assert engine_kwargs['workflow_context'].service.id == service.id
-        assert engine_kwargs['workflow_context'].execution.workflow_name == 'test_workflow'
+        _, engine_kwargs = mock_engine_execute.call_args
+        assert engine_kwargs['ctx'].service.id == service.id
+        assert engine_kwargs['ctx'].execution.workflow_name == 'test_workflow'
 
-        workflow_runner.execute()
-        mock_engine.execute.assert_called_once_with()
+        mock_engine_execute.assert_called_once_with(ctx=workflow_runner._workflow_context)
 
 
 def test_cancel_execution(request):
     mock_workflow = _setup_mock_workflow_in_service(request)
 
     mock_engine = mock.MagicMock()
-    with mock.patch('aria.orchestrator.workflow_runner.Engine', return_value=mock_engine):
+    with mock.patch('aria.orchestrator.workflow_runner.engine.Engine', return_value=mock_engine):
         workflow_runner = _create_workflow_runner(request, mock_workflow)
         workflow_runner.cancel()
-        mock_engine.cancel_execution.assert_called_once_with()
+        mock_engine.cancel_execution.assert_called_once_with(ctx=workflow_runner._workflow_context)
 
 
 def test_execution_model_creation(request, service, model):
     mock_workflow = _setup_mock_workflow_in_service(request)
 
-    with mock.patch('aria.orchestrator.workflow_runner.Engine') as mock_engine_cls:
+    with mock.patch('aria.orchestrator.workflow_runner.engine.Engine'):
         workflow_runner = _create_workflow_runner(request, mock_workflow)
 
-        _, engine_kwargs = mock_engine_cls.call_args
-        assert engine_kwargs['workflow_context'].execution == workflow_runner.execution
         assert model.execution.get(workflow_runner.execution.id) == workflow_runner.execution
         assert workflow_runner.execution.service.id == service.id
         assert workflow_runner.execution.workflow_name == mock_workflow
@@ -173,7 +172,7 @@ def test_execution_inputs_override_workflow_inputs(request):
         inputs=dict((name, models.Input.wrap(name, val)) for name, val
                     in wf_inputs.iteritems()))
 
-    with mock.patch('aria.orchestrator.workflow_runner.Engine'):
+    with mock.patch('aria.orchestrator.workflow_runner.engine.Engine'):
         workflow_runner = _create_workflow_runner(
             request, mock_workflow, inputs={'input2': 'overriding-value2', 'input3': 7})
 

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e119df5/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 0438544..b77d284 100644
--- a/tests/orchestrator/workflows/core/test_engine.py
+++ b/tests/orchestrator/workflows/core/test_engine.py
@@ -28,7 +28,7 @@ from aria.orchestrator.workflows import (
     api,
     exceptions,
 )
-from aria.orchestrator.workflows.core import engine
+from aria.orchestrator.workflows.core import engine, compile
 from aria.orchestrator.workflows.executor import thread
 
 from tests import mock, storage
@@ -44,15 +44,15 @@ class BaseTest(object):
         eng = cls._engine(workflow_func=workflow_func,
                           workflow_context=workflow_context,
                           executor=executor)
-        eng.execute()
+        eng.execute(ctx=workflow_context)
         return eng
 
     @staticmethod
     def _engine(workflow_func, workflow_context, executor):
         graph = workflow_func(ctx=workflow_context)
-        return engine.Engine(executor=executor,
-                             workflow_context=workflow_context,
-                             tasks_graph=graph)
+        compile.create_execution_tasks(workflow_context, graph, executor.__class__)
+
+        return engine.Engine(executors={executor.__class__: executor})
 
     @staticmethod
     def _create_interface(ctx, func, arguments=None):
@@ -98,9 +98,10 @@ class BaseTest(object):
 
     @pytest.fixture(autouse=True)
     def signals_registration(self, ):
-        def sent_task_handler(*args, **kwargs):
-            calls = global_test_holder.setdefault('sent_task_signal_calls', 0)
-            global_test_holder['sent_task_signal_calls'] = calls + 1
+        def sent_task_handler(ctx, *args, **kwargs):
+            if ctx.task._stub_type is None:
+                calls = global_test_holder.setdefault('sent_task_signal_calls', 0)
+                global_test_holder['sent_task_signal_calls'] = calls + 1
 
         def start_workflow_handler(workflow_context, *args, **kwargs):
             workflow_context.states.append('start')
@@ -255,10 +256,10 @@ class TestCancel(BaseTest):
         eng = self._engine(workflow_func=mock_workflow,
                            workflow_context=workflow_context,
                            executor=executor)
-        t = threading.Thread(target=eng.execute)
+        t = threading.Thread(target=eng.execute, kwargs=dict(ctx=workflow_context))
         t.start()
         time.sleep(10)
-        eng.cancel_execution()
+        eng.cancel_execution(workflow_context)
         t.join(timeout=60) # we need to give this a *lot* of time because Travis can be *very* slow
         assert not t.is_alive() # if join is timed out it will not raise an exception
         assert workflow_context.states == ['start', 'cancel']
@@ -277,7 +278,7 @@ class TestCancel(BaseTest):
         eng = self._engine(workflow_func=mock_workflow,
                            workflow_context=workflow_context,
                            executor=executor)
-        eng.cancel_execution()
+        eng.cancel_execution(workflow_context)
         execution = workflow_context.execution
         assert execution.status == models.Execution.CANCELLED
 

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e119df5/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 6d542e9..2b82443 100644
--- a/tests/orchestrator/workflows/core/test_events.py
+++ b/tests/orchestrator/workflows/core/test_events.py
@@ -15,12 +15,13 @@
 
 import pytest
 
-from tests import mock, storage
-from aria.modeling.service_instance import NodeBase
 from aria.orchestrator.decorators import operation, workflow
-from aria.orchestrator.workflows.core import engine
+from aria.orchestrator.workflows.core import engine, compile
 from aria.orchestrator.workflows.executor.thread import ThreadExecutor
 from aria.orchestrator.workflows import api
+from aria.modeling.service_instance import NodeBase
+
+from tests import mock, storage
 
 global_test_dict = {}  # used to capture transitional node state changes
 
@@ -112,14 +113,13 @@ 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)
 
-    eng = engine.Engine(executor=ThreadExecutor(),
-                        workflow_context=ctx,
-                        tasks_graph=single_operation_workflow(ctx=ctx,  # pylint: disable=no-value-for-parameter
-                                                              node=node,
-                                                              interface_name=interface_name,
-                                                              op_name=op_name))
-    eng.execute()
+    eng = engine.Engine(executors={ThreadExecutor: ThreadExecutor()})
+    eng.execute(ctx)
     return node
 
 

http://git-wip-us.apache.org/repos/asf/incubator-ariatosca/blob/3e119df5/tests/orchestrator/workflows/core/test_task.py
----------------------------------------------------------------------
diff --git a/tests/orchestrator/workflows/core/test_task.py b/tests/orchestrator/workflows/core/test_task.py
index c0d3616..2b3f7d7 100644
--- a/tests/orchestrator/workflows/core/test_task.py
+++ b/tests/orchestrator/workflows/core/test_task.py
@@ -22,9 +22,9 @@ import pytest
 from aria.orchestrator.context import workflow as workflow_context
 from aria.orchestrator.workflows import (
     api,
-    core,
     exceptions,
 )
+from aria.modeling import models
 
 from tests import mock, storage
 
@@ -70,8 +70,8 @@ class TestOperationTask(object):
                 node,
                 interface_name=NODE_INTERFACE_NAME,
                 operation_name=NODE_OPERATION_NAME)
-            core_task = core.task.OperationTask(api_task=api_task, executor=None)
-        return api_task, core_task
+            model_task = models.Task.from_api_task(api_task, None)
+        return api_task, model_task
 
     def _create_relationship_operation_task(self, ctx, relationship):
         with workflow_context.current.push(ctx):
@@ -79,7 +79,7 @@ class TestOperationTask(object):
                 relationship,
                 interface_name=RELATIONSHIP_INTERFACE_NAME,
                 operation_name=RELATIONSHIP_OPERATION_NAME)
-            core_task = core.task.OperationTask(api_task=api_task, executor=None)
+            core_task = models.Task.from_api_task(api_task, None)
         return api_task, core_task
 
     def test_node_operation_task_creation(self, ctx):
@@ -96,25 +96,21 @@ class TestOperationTask(object):
         )
         node.interfaces[interface.name] = interface
         ctx.model.node.update(node)
-        api_task, core_task = self._create_node_operation_task(ctx, node)
-        storage_task = ctx.model.task.get_by_name(core_task.name)
-        assert storage_task.plugin is storage_plugin
-        assert storage_task.execution_name == ctx.execution.name
-        assert storage_task.actor == core_task.context.node
-        assert core_task.model_task == storage_task
-        assert core_task.name == api_task.name
-        assert core_task.function == api_task.function
-        assert core_task.actor == api_task.actor == node
-        assert core_task.arguments == api_task.arguments == storage_task.arguments
-        assert core_task.plugin == storage_plugin
+        api_task, model_task = self._create_node_operation_task(ctx, node)
+        assert model_task.name == api_task.name
+        assert model_task.function == api_task.function
+        assert model_task.actor == api_task.actor == node
+        assert model_task.arguments == api_task.arguments
+        assert model_task.plugin == storage_plugin
 
     def test_relationship_operation_task_creation(self, ctx):
         relationship = ctx.model.relationship.list()[0]
         ctx.model.relationship.update(relationship)
-        _, core_task = self._create_relationship_operation_task(
+        _, model_task = self._create_relationship_operation_task(
             ctx, relationship)
-        assert core_task.model_task.actor == relationship
+        assert model_task.actor == relationship
 
+    @pytest.mark.skip("Currently not supported for model tasks")
     def test_operation_task_edit_locked_attribute(self, ctx):
         node = ctx.model.node.get_by_name(mock.models.DEPENDENCY_NODE_NAME)
 
@@ -131,6 +127,7 @@ class TestOperationTask(object):
         with pytest.raises(exceptions.TaskException):
             core_task.due_at = now
 
+    @pytest.mark.skip("Currently not supported for model tasks")
     def test_operation_task_edit_attributes(self, ctx):
         node = ctx.model.node.get_by_name(mock.models.DEPENDENCY_NODE_NAME)
 


Mime
View raw message