text: # Copyright 2015 OpenStack Foundation
  # 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 contextlib
  
  from oslo_config import cfg
  from oslo_log import log as logging
  from oslo_utils import excutils
  from stevedore import driver
  from taskflow import engines
  from taskflow.listeners import logging as llistener
  from taskflow.types import futures
  from taskflow.utils import eventlet_utils
  
  import daisy.async
  from daisy.common.scripts import utils as script_utils
  from daisy import i18n
  
  _ = i18n._
  _LE = i18n._LE
  LOG = logging.getLogger(__name__)
  
  _deprecated_opt = cfg.DeprecatedOpt('eventlet_executor_pool_size',
                                      group='task')
  
  taskflow_executor_opts = [
      cfg.StrOpt('engine_mode',
                 default='parallel',
                 choices=('serial', 'parallel'),
                 help=_("The mode in which the engine will run. "
                        "Can be 'serial' or 'parallel'.")),
      cfg.IntOpt('max_workers',
                 default=10,
                 help=_("The number of parallel activities executed at the "
                        "same time by the engine. The value can be greater "
                        "than one when the engine mode is 'parallel'."),
                 deprecated_opts=[_deprecated_opt])
  ]
  
  
  CONF = cfg.CONF
  CONF.register_opts(taskflow_executor_opts, group='taskflow_executor')
  
  
  class TaskExecutor(daisy.async.TaskExecutor):
  
      def __init__(self, context, task_repo, image_repo, image_factory):
          self.context = context
          self.task_repo = task_repo
          self.image_repo = image_repo
          self.image_factory = image_factory
          self.engine_conf = {
              'engine': CONF.taskflow_executor.engine_mode,
          }
          self.engine_kwargs = {}
          if CONF.taskflow_executor.engine_mode == 'parallel':
              self.engine_kwargs['max_workers'] = (
                  CONF.taskflow_executor.max_workers)
          super(TaskExecutor, self).__init__(context, task_repo, image_repo,
                                             image_factory)
  
      @contextlib.contextmanager
      def _executor(self):
          if CONF.taskflow_executor.engine_mode != 'parallel':
              yield None
          else:
              max_workers = CONF.taskflow_executor.max_workers
              if eventlet_utils.EVENTLET_AVAILABLE:
                  yield futures.GreenThreadPoolExecutor(max_workers=max_workers)
              else:
                  yield futures.ThreadPoolExecutor(max_workers=max_workers)
  
      def _get_flow(self, task):
          try:
              task_input = script_utils.unpack_task_input(task)
              uri = script_utils.validate_location_uri(
                  task_input.get('import_from'))
  
              kwds = {
                  'uri': uri,
                  'task_id': task.task_id,
                  'task_type': task.type,
                  'context': self.context,
                  'task_repo': self.task_repo,
                  'image_repo': self.image_repo,
                  'image_factory': self.image_factory
              }
  
              return driver.DriverManager('daisy.flows', task.type,
                                          invoke_on_load=True,
                                          invoke_kwds=kwds).driver
          except RuntimeError:
              raise NotImplementedError()
  
      def _run(self, task_id, task_type):
          LOG.debug('Taskflow executor picked up the execution of task ID '
                    '%(task_id)s of task type '
                    '%(task_type)s' % {'task_id': task_id,
                                       'task_type': task_type})
  
          task = script_utils.get_task(self.task_repo, task_id)
          if task is None:
              # NOTE: This happens if task is not found in the database. In
              # such cases, there is no way to update the task status so,
              # it's ignored here.
              return
  
          flow = self._get_flow(task)
  
          try:
              with self._executor() as executor:
                  engine = engines.load(flow, self.engine_conf,
                                        executor=executor, **self.engine_kwargs)
                  with llistener.DynamicLoggingListener(engine, log=LOG):
                      engine.run()
          except Exception as exc:
              with excutils.save_and_reraise_exception():
                  LOG.error(_LE('Failed to execute task %(task_id)s: %(exc)s') %
                            {'task_id': task_id, 'exc': exc.message})
  