forked from jamesgao/ipython
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtaskclient.py
More file actions
198 lines (158 loc) · 6.35 KB
/
taskclient.py
File metadata and controls
198 lines (158 loc) · 6.35 KB
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
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
# encoding: utf-8
# -*- test-case-name: IPython.kernel.tests.test_taskcontrollerxmlrpc -*-
"""
A blocking version of the task client.
"""
__docformat__ = "restructuredtext en"
#-------------------------------------------------------------------------------
# Copyright (C) 2008 The IPython Development Team
#
# Distributed under the terms of the BSD License. The full license is in
# the file COPYING, distributed as part of this software.
#-------------------------------------------------------------------------------
#-------------------------------------------------------------------------------
# Imports
#-------------------------------------------------------------------------------
from zope.interface import Interface, implements
from twisted.python import components
try:
from foolscap.api import DeadReferenceError
except ImportError:
from foolscap import DeadReferenceError
from IPython.kernel.twistedutil import blockingCallFromThread
from IPython.kernel import task, error
from IPython.kernel.mapper import (
SynchronousTaskMapper,
ITaskMapperFactory,
IMapper
)
from IPython.kernel.parallelfunction import (
ParallelFunction,
ITaskParallelDecorator
)
#-------------------------------------------------------------------------------
# The task client
#-------------------------------------------------------------------------------
class IBlockingTaskClient(Interface):
"""
A vague interface of the blocking task client
"""
pass
class BlockingTaskClient(object):
"""
A blocking task client that adapts a non-blocking one.
"""
implements(
IBlockingTaskClient,
ITaskMapperFactory,
IMapper,
ITaskParallelDecorator
)
def __init__(self, task_controller):
self.task_controller = task_controller
self.block = True
def _bcft(self, *args, **kwargs):
try:
result = blockingCallFromThread(*args, **kwargs)
except DeadReferenceError:
raise error.ConnectionError(
"""A connection error has occurred in trying to connect to the
controller. This is usually caused by the controller dying or
being restarted. To resolve this issue try recreating the
task client."""
)
else:
return result
def run(self, task, block=False):
"""Run a task on the `TaskController`.
See the documentation of the `MapTask` and `StringTask` classes for
details on how to build a task of different types.
:Parameters:
task : an `ITask` implementer
:Returns: The int taskid of the submitted task. Pass this to
`get_task_result` to get the `TaskResult` object.
"""
tid = self._bcft(self.task_controller.run, task)
if block:
return self.get_task_result(tid, block=True)
else:
return tid
def get_task_result(self, taskid, block=False):
"""
Get a task result by taskid.
:Parameters:
taskid : int
The taskid of the task to be retrieved.
block : boolean
Should I block until the task is done?
:Returns: A `TaskResult` object that encapsulates the task result.
"""
return self._bcft(self.task_controller.get_task_result,
taskid, block)
def abort(self, taskid):
"""
Abort a task by taskid.
:Parameters:
taskid : int
The taskid of the task to be aborted.
"""
return self._bcft(self.task_controller.abort, taskid)
def barrier(self, taskids):
"""Block until a set of tasks are completed.
:Parameters:
taskids : list, tuple
A sequence of taskids to block on.
"""
return self._bcft(self.task_controller.barrier, taskids)
def spin(self):
"""
Touch the scheduler, to resume scheduling without submitting a task.
This method only needs to be called in unusual situations where the
scheduler is idle for some reason.
"""
return self._bcft(self.task_controller.spin)
def queue_status(self, verbose=False):
"""
Get a dictionary with the current state of the task queue.
:Parameters:
verbose : boolean
If True, return a list of taskids. If False, simply give
the number of tasks with each status.
:Returns:
A dict with the queue status.
"""
return self._bcft(self.task_controller.queue_status, verbose)
def clear(self):
"""
Clear all previously run tasks from the task controller.
This is needed because the task controller keep all task results
in memory. This can be a problem is there are many completed
tasks. Users should call this periodically to clean out these
cached task results.
"""
return self._bcft(self.task_controller.clear)
def map(self, func, *sequences):
"""
Apply func to *sequences elementwise. Like Python's builtin map.
This version is load balanced.
"""
return self.mapper().map(func, *sequences)
def mapper(self, clear_before=False, clear_after=False, retries=0,
recovery_task=None, depend=None, block=True):
"""
Create an `IMapper` implementer with a given set of arguments.
The `IMapper` created using a task controller is load balanced.
See the documentation for `IPython.kernel.task.BaseTask` for
documentation on the arguments to this method.
"""
return SynchronousTaskMapper(self, clear_before=clear_before,
clear_after=clear_after, retries=retries,
recovery_task=recovery_task, depend=depend, block=block)
def parallel(self, clear_before=False, clear_after=False, retries=0,
recovery_task=None, depend=None, block=True):
mapper = self.mapper(clear_before, clear_after, retries,
recovery_task, depend, block)
pf = ParallelFunction(mapper)
return pf
components.registerAdapter(BlockingTaskClient,
task.ITaskController, IBlockingTaskClient)