openstack/deb-python-taskflow
Code Issues Proposed changes

Do not create TaskAction for each task

Browse Source 
Instead of creating TaskAction for each task we create single TaskAction
in engine, that knows how to run tasks for it. We also got rid of engine
dependency of TaskAction, passing storage and notifier to it instead.

References blueprint task-executor
Co-authored-by: Ivan A. Melnikov <imelnikov@griddynamics.com>
Change-Id: Ie52eba3bba5c730cee091ee24e995e0ba21f9486
This commit is contained in:
Anastasia Karpinska
2013-12-09 19:30:54 +04:00
committed by Ivan A. Melnikov
parent ac8a5882db
commit b56afe66a2
5 changed files with 53 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
taskflow/engines/action_engine/base_action.py
View File

@@ -1,35 +0,0 @@
# -*- coding: utf-8 -*-
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright (C) 2012 Yahoo! Inc. All Rights Reserved.
#
# Licensed 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.
import abc
import six
@six.add_metaclass(abc.ABCMeta)
class Action(object):
"""Base action class"""
@abc.abstractmethod
def execute(self, engine):
"""Run the action"""
@abc.abstractmethod
def revert(self, engine):
"""Undo all side effects of execute method"""

28
taskflow/engines/action_engine/engine.py
View File

@@ -49,7 +49,8 @@ class ActionEngine(base.EngineBase):
reversion to commence. See the valid states in the states module to learn reversion to commence. See the valid states in the states module to learn
more about what other states the tasks & flow being ran can go through. more about what other states the tasks & flow being ran can go through.
""" """
_graph_action = None _graph_action_cls = None
_task_action_cls = task_action.TaskAction
def __init__(self, flow, flow_detail, backend, conf): def __init__(self, flow, flow_detail, backend, conf):
super(ActionEngine, self).__init__(flow, flow_detail, backend, conf) super(ActionEngine, self).__init__(flow, flow_detail, backend, conf)
@@ -58,6 +59,8 @@ class ActionEngine(base.EngineBase):
self._state_lock = threading.RLock() self._state_lock = threading.RLock()
self.notifier = misc.TransitionNotifier() self.notifier = misc.TransitionNotifier()
self.task_notifier = misc.TransitionNotifier() self.task_notifier = misc.TransitionNotifier()
self.task_action = self._task_action_cls(self.storage,
self.task_notifier)
def _revert(self, current_failure=None): def _revert(self, current_failure=None):
self._change_state(states.REVERTING) self._change_state(states.REVERTING)
@@ -139,18 +142,6 @@ class ActionEngine(base.EngineBase):
old_state=old_state) old_state=old_state)
self.notifier.notify(state, details) self.notifier.notify(state, details)
def _on_task_state_change(self, task_name, state, result=None):
"""Notifies the engine that the following task action has completed
a given state with a given result. This is a *internal* to the action
engine and its associated action classes, not for use externally.
"""
task_uuid = self.storage.get_task_uuid(task_name)
details = dict(engine=self,
task_name=task_name,
task_uuid=task_uuid,
result=result)
self.task_notifier.notify(state, details)
def _reset(self): def _reset(self):
for name, uuid in self.storage.reset_tasks(): for name, uuid in self.storage.reset_tasks():
details = dict(engine=self, details = dict(engine=self,
@@ -169,17 +160,16 @@ class ActionEngine(base.EngineBase):
if self._root is not None: if self._root is not None:
return return
assert self._graph_action is not None, ('Graph action class must be' assert self._graph_action_cls is not None, (
' specified') 'Graph action class must be specified')
self._change_state(states.RESUMING) # does nothing in PENDING state self._change_state(states.RESUMING) # does nothing in PENDING state
task_graph = flow_utils.flatten(self._flow) task_graph = flow_utils.flatten(self._flow)
if task_graph.number_of_nodes() == 0: if task_graph.number_of_nodes() == 0:
raise exc.EmptyFlow("Flow %s is empty." % self._flow.name) raise exc.EmptyFlow("Flow %s is empty." % self._flow.name)
self._root = self._graph_action(task_graph) self._root = self._graph_action_cls(task_graph)
for task in task_graph.nodes_iter(): for task in task_graph.nodes_iter():
task_version = misc.get_version_string(task) task_version = misc.get_version_string(task)
self.storage.ensure_task(task.name, task_version, task.save_as) self.storage.ensure_task(task.name, task_version, task.save_as)
self._root.add(task, task_action.TaskAction(task))
self._change_state(states.SUSPENDED) # does nothing in PENDING state self._change_state(states.SUSPENDED) # does nothing in PENDING state
@@ -194,13 +184,13 @@ class ActionEngine(base.EngineBase):
class SingleThreadedActionEngine(ActionEngine): class SingleThreadedActionEngine(ActionEngine):
# NOTE(harlowja): This one attempts to run in a serial manner. # NOTE(harlowja): This one attempts to run in a serial manner.
_graph_action = graph_action.SequentialGraphAction _graph_action_cls = graph_action.SequentialGraphAction
_storage_cls = t_storage.Storage _storage_cls = t_storage.Storage
class MultiThreadedActionEngine(ActionEngine): class MultiThreadedActionEngine(ActionEngine):
# NOTE(harlowja): This one attempts to run in a parallel manner. # NOTE(harlowja): This one attempts to run in a parallel manner.
_graph_action = graph_action.ParallelGraphAction _graph_action_cls = graph_action.ParallelGraphAction
_storage_cls = t_storage.ThreadSafeStorage _storage_cls = t_storage.ThreadSafeStorage
def __init__(self, flow, flow_detail, backend, conf): def __init__(self, flow, flow_detail, backend, conf):

16
taskflow/engines/action_engine/graph_action.py
View File

@@ -22,26 +22,21 @@ import threading
from concurrent import futures from concurrent import futures
from taskflow.engines.action_engine import base_action as base
from taskflow import states as st from taskflow import states as st
from taskflow.utils import misc from taskflow.utils import misc
LOG = logging.getLogger(__name__) LOG = logging.getLogger(__name__)
class GraphAction(base.Action): class GraphAction(object):
def __init__(self, graph): def __init__(self, graph):
self._graph = graph self._graph = graph
self._action_mapping = {}
@property @property
def graph(self): def graph(self):
return self._graph return self._graph
def add(self, node, action):
self._action_mapping[node] = action
def _succ(self, node): def _succ(self, node):
return self._graph.successors(node) return self._graph.successors(node)
@@ -80,8 +75,7 @@ class SequentialGraphAction(GraphAction):
while to_execute and engine.is_running: while to_execute and engine.is_running:
node = to_execute.pop() node = to_execute.pop()
action = self._action_mapping[node] engine.task_action.execute(node)
action.execute(engine) # raises on failure
to_execute += self._resolve_dependencies(node, deps_counter) to_execute += self._resolve_dependencies(node, deps_counter)
if to_execute: if to_execute:
@@ -94,8 +88,7 @@ class SequentialGraphAction(GraphAction):
while to_revert and engine.is_reverting: while to_revert and engine.is_reverting:
node = to_revert.pop() node = to_revert.pop()
action = self._action_mapping[node] engine.task_action.revert(node)
action.revert(engine) # raises on failure
to_revert += self._resolve_dependencies(node, deps_counter, True) to_revert += self._resolve_dependencies(node, deps_counter, True)
if to_revert: if to_revert:
@@ -140,10 +133,9 @@ class ParallelGraphAction(SequentialGraphAction):
if has_failed.is_set(): if has_failed.is_set():
# Someone failed, don't even bother running. # Someone failed, don't even bother running.
return return
action = self._action_mapping[node]
try: try:
if engine.is_running: if engine.is_running:
action.execute(engine) engine.task_action.execute(node)
else: else:
was_suspended.set() was_suspended.set()
return return

89
taskflow/engines/action_engine/task_action.py
View File

@@ -2,7 +2,7 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4 # vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright (C) 2012 Yahoo! Inc. All Rights Reserved. # Copyright (C) 2012-2013 Yahoo! Inc. All Rights Reserved.
# #
# Licensed under the Apache License, Version 2.0 (the "License"); you may # Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain # not use this file except in compliance with the License. You may obtain
@@ -19,7 +19,6 @@
import contextlib import contextlib
import logging import logging
from taskflow.engines.action_engine import base_action as base
from taskflow.openstack.common import excutils from taskflow.openstack.common import excutils
from taskflow import states from taskflow import states
from taskflow.utils import misc from taskflow.utils import misc
@@ -38,79 +37,65 @@ def _autobind(task, bind_name, bind_func, **kwargs):
task.unbind(bind_name, bind_func) task.unbind(bind_name, bind_func)
class TaskAction(base.Action): class TaskAction(object):
def __init__(self, storage, notifier):
self._storage = storage
self._notifier = notifier
def __init__(self, task): def _change_state(self, task, state, result=None, progress=None):
self._task = task old_state = self._storage.get_task_state(task.name)
@property
def name(self):
return self._task.name
def _change_state(self, engine, state, result=None, progress=None):
"""Update result and change state."""
old_state = engine.storage.get_task_state(self.name)
if not states.check_task_transition(old_state, state): if not states.check_task_transition(old_state, state):
return False return False
if state in SAVE_RESULT_STATES: if state in SAVE_RESULT_STATES:
engine.storage.save(self.name, result, state) self._storage.save(task.name, result, state)
else: else:
engine.storage.set_task_state(self.name, state) self._storage.set_task_state(task.name, state)
if progress is not None: if progress is not None:
engine.storage.set_task_progress(self.name, progress) self._storage.set_task_progress(task.name, progress)
engine._on_task_state_change(self.name, state, result=result)
task_uuid = self._storage.get_task_uuid(task.name)
details = dict(task_name=task.name,
task_uuid=task_uuid,
result=result)
self._notifier.notify(state, details)
if progress is not None:
task.update_progress(progress)
return True return True
def _on_update_progress(self, task, event_data, progress, **kwargs): def _on_update_progress(self, task, event_data, progress, **kwargs):
"""Update task progress value that stored in engine.""" """Should be called when task updates its progress"""
try: try:
engine = event_data['engine'] self._storage.set_task_progress(task.name, progress, kwargs)
engine.storage.set_task_progress(self.name, progress, kwargs)
except Exception: except Exception:
# Update progress callbacks should never fail, so capture and log # Update progress callbacks should never fail, so capture and log
# the emitted exception instead of raising it. # the emitted exception instead of raising it.
LOG.exception("Failed setting task progress for %s to %0.3f", LOG.exception("Failed setting task progress for %s to %0.3f",
task, progress) task, progress)
def _change_state_update_task(self, engine, state, progress, result=None): def execute(self, task):
stated_changed = self._change_state(engine, state, if not self._change_state(task, states.RUNNING, progress=0.0):
result=result, progress=progress)
if not stated_changed:
return False
self._task.update_progress(progress)
return True
def execute(self, engine):
if not self._change_state_update_task(engine, states.RUNNING, 0.0):
return return
with _autobind(self._task, with _autobind(task, 'update_progress', self._on_update_progress):
'update_progress', self._on_update_progress,
engine=engine):
try: try:
kwargs = engine.storage.fetch_mapped_args(self._task.rebind) kwargs = self._storage.fetch_mapped_args(task.rebind)
result = self._task.execute(**kwargs) result = task.execute(**kwargs)
except Exception: except Exception:
failure = misc.Failure() failure = misc.Failure()
self._change_state(engine, states.FAILURE, result=failure) self._change_state(task, states.FAILURE, result=failure)
failure.reraise() failure.reraise()
self._change_state_update_task(engine, states.SUCCESS, 1.0,
result=result)
def revert(self, engine): self._change_state(task, states.SUCCESS, result=result, progress=1.0)
if not self._change_state_update_task(engine, states.REVERTING, 0.0):
# NOTE(imelnikov): in all the other states, the task def revert(self, task):
# execution was at least attempted, so we should give if not self._change_state(task, states.REVERTING, progress=0.0):
# task a chance for cleanup
return return
with _autobind(self._task, with _autobind(task, 'update_progress', self._on_update_progress):
'update_progress', self._on_update_progress, kwargs = self._storage.fetch_mapped_args(task.rebind)
engine=engine): kwargs['result'] = self._storage.get(task.name)
kwargs = engine.storage.fetch_mapped_args(self._task.rebind) kwargs['flow_failures'] = self._storage.get_failures()
kwargs['result'] = engine.storage.get(self.name)
kwargs['flow_failures'] = engine.storage.get_failures()
try: try:
self._task.revert(**kwargs) task.revert(**kwargs)
except Exception: except Exception:
with excutils.save_and_reraise_exception(): with excutils.save_and_reraise_exception():
self._change_state(engine, states.FAILURE) self._change_state(task, states.FAILURE)
self._change_state_update_task(engine, states.REVERTED, 1.0) self._change_state(task, states.REVERTED, progress=1.0)

6
taskflow/listeners/base.py
View File

@@ -127,7 +127,7 @@ class LoggingBase(ListenerBase):
def _flow_receiver(self, state, details): def _flow_receiver(self, state, details):
self._log("%s has moved flow '%s' (%s) into state '%s'", self._log("%s has moved flow '%s' (%s) into state '%s'",
details['engine'], details['flow_name'], self._engine, details['flow_name'],
details['flow_uuid'], state) details['flow_uuid'], state)
def _task_receiver(self, state, details): def _task_receiver(self, state, details):
@@ -141,10 +141,10 @@ class LoggingBase(ListenerBase):
was_failure = True was_failure = True
self._log("%s has moved task '%s' (%s) into state '%s'" self._log("%s has moved task '%s' (%s) into state '%s'"
" with result '%s' (failure=%s)", " with result '%s' (failure=%s)",
details['engine'], details['task_name'], self._engine, details['task_name'],
details['task_uuid'], state, result, was_failure, details['task_uuid'], state, result, was_failure,
exc_info=exc_info) exc_info=exc_info)
else: else:
self._log("%s has moved task '%s' (%s) into state '%s'", self._log("%s has moved task '%s' (%s) into state '%s'",
details['engine'], details['task_name'], self._engine, details['task_name'],
details['task_uuid'], state) details['task_uuid'], state)