1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
|
# Copyright 2016 Rackspace Inc.
#
# Author: Eric Larson <eric.larson@rackspace.com>
#
# 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 time
import futurist
from oslo_config import cfg
from oslo_log import log as logging
from designate import exceptions
LOG = logging.getLogger(__name__)
CONF = cfg.CONF
def default_executor():
# TODO(mugsie): if (when) we move away from eventlet this may have to
# revert back to ThreadPoolExecutor - this is changing due to
# https://bugs.launchpad.net/bugs/1782647 (eventlet + py37 issues)
return futurist.GreenThreadPoolExecutor(CONF['service:worker'].threads)
class Executor(object):
"""
Object to facilitate the running of a task, or a set of tasks on an
executor that can map multiple tasks across a configurable number of
threads
"""
def __init__(self, executor=None):
self._executor = executor or default_executor()
@staticmethod
def do(task):
try:
return task()
except exceptions.BadAction as e:
LOG.warning(e)
@staticmethod
def task_name(task):
if hasattr(task, 'task_name'):
return str(task.task_name)
if hasattr(task, 'func_name'):
return str(task.func_name)
return 'UnnamedTask'
def run(self, tasks):
"""
Run task or set of tasks
:param tasks: the task or tasks you want to execute in the
executor's pool
:return: The results of the tasks (list)
If a single task is pass
"""
if callable(tasks):
tasks = [tasks]
start_time = time.monotonic()
results = [r for r in self._executor.map(self.do, tasks)]
elapsed_time = time.monotonic() - start_time
LOG.debug(
'Finished Task(s): %(tasks)s in %(time)fs',
{
'tasks': ', '.join([self.task_name(t) for t in tasks]),
'time': elapsed_time
}
)
return results
|
