Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 1 | """Support for tasks, coroutines and the scheduler.""" |
| 2 | |
Victor Stinner | f951d28 | 2014-06-29 00:46:45 +0200 | [diff] [blame] | 3 | __all__ = ['Task', |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 4 | 'FIRST_COMPLETED', 'FIRST_EXCEPTION', 'ALL_COMPLETED', |
| 5 | 'wait', 'wait_for', 'as_completed', 'sleep', 'async', |
Guido van Rossum | 841d9ee | 2015-10-03 08:31:42 -0700 | [diff] [blame] | 6 | 'gather', 'shield', 'ensure_future', 'run_coroutine_threadsafe', |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 7 | ] |
| 8 | |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 9 | import concurrent.futures |
| 10 | import functools |
| 11 | import inspect |
| 12 | import linecache |
| 13 | import traceback |
Yury Selivanov | 59eb9a4 | 2015-05-11 14:48:38 -0400 | [diff] [blame] | 14 | import warnings |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 15 | import weakref |
| 16 | |
Victor Stinner | 71080fc | 2015-07-25 02:23:21 +0200 | [diff] [blame] | 17 | from . import compat |
Victor Stinner | f951d28 | 2014-06-29 00:46:45 +0200 | [diff] [blame] | 18 | from . import coroutines |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 19 | from . import events |
| 20 | from . import futures |
Victor Stinner | f951d28 | 2014-06-29 00:46:45 +0200 | [diff] [blame] | 21 | from .coroutines import coroutine |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 22 | |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 23 | |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 24 | class Task(futures.Future): |
| 25 | """A coroutine wrapped in a Future.""" |
| 26 | |
| 27 | # An important invariant maintained while a Task not done: |
| 28 | # |
| 29 | # - Either _fut_waiter is None, and _step() is scheduled; |
| 30 | # - or _fut_waiter is some Future, and _step() is *not* scheduled. |
| 31 | # |
| 32 | # The only transition from the latter to the former is through |
| 33 | # _wakeup(). When _fut_waiter is not None, one of its callbacks |
| 34 | # must be _wakeup(). |
| 35 | |
| 36 | # Weak set containing all tasks alive. |
| 37 | _all_tasks = weakref.WeakSet() |
| 38 | |
Guido van Rossum | 1a605ed | 2013-12-06 12:57:40 -0800 | [diff] [blame] | 39 | # Dictionary containing tasks that are currently active in |
| 40 | # all running event loops. {EventLoop: Task} |
| 41 | _current_tasks = {} |
| 42 | |
Victor Stinner | fe22e09 | 2014-12-04 23:00:13 +0100 | [diff] [blame] | 43 | # If False, don't log a message if the task is destroyed whereas its |
| 44 | # status is still pending |
| 45 | _log_destroy_pending = True |
| 46 | |
Guido van Rossum | 1a605ed | 2013-12-06 12:57:40 -0800 | [diff] [blame] | 47 | @classmethod |
| 48 | def current_task(cls, loop=None): |
| 49 | """Return the currently running task in an event loop or None. |
| 50 | |
| 51 | By default the current task for the current event loop is returned. |
| 52 | |
| 53 | None is returned when called not in the context of a Task. |
| 54 | """ |
| 55 | if loop is None: |
| 56 | loop = events.get_event_loop() |
| 57 | return cls._current_tasks.get(loop) |
| 58 | |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 59 | @classmethod |
| 60 | def all_tasks(cls, loop=None): |
| 61 | """Return a set of all tasks for an event loop. |
| 62 | |
| 63 | By default all tasks for the current event loop are returned. |
| 64 | """ |
| 65 | if loop is None: |
| 66 | loop = events.get_event_loop() |
| 67 | return {t for t in cls._all_tasks if t._loop is loop} |
| 68 | |
| 69 | def __init__(self, coro, *, loop=None): |
Victor Stinner | 15cc678 | 2015-01-09 00:09:10 +0100 | [diff] [blame] | 70 | assert coroutines.iscoroutine(coro), repr(coro) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 71 | super().__init__(loop=loop) |
Victor Stinner | 80f53aa | 2014-06-27 13:52:20 +0200 | [diff] [blame] | 72 | if self._source_traceback: |
| 73 | del self._source_traceback[-1] |
Yury Selivanov | 1ad08a5 | 2015-05-28 10:52:19 -0400 | [diff] [blame] | 74 | self._coro = coro |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 75 | self._fut_waiter = None |
| 76 | self._must_cancel = False |
| 77 | self._loop.call_soon(self._step) |
| 78 | self.__class__._all_tasks.add(self) |
| 79 | |
R David Murray | 8e069d5 | 2014-09-24 13:13:45 -0400 | [diff] [blame] | 80 | # On Python 3.3 or older, objects with a destructor that are part of a |
| 81 | # reference cycle are never destroyed. That's not the case any more on |
| 82 | # Python 3.4 thanks to the PEP 442. |
Victor Stinner | 71080fc | 2015-07-25 02:23:21 +0200 | [diff] [blame] | 83 | if compat.PY34: |
Victor Stinner | a02f81f | 2014-06-24 22:37:53 +0200 | [diff] [blame] | 84 | def __del__(self): |
Victor Stinner | 98b6391 | 2014-06-30 14:51:04 +0200 | [diff] [blame] | 85 | if self._state == futures._PENDING and self._log_destroy_pending: |
Victor Stinner | 80f53aa | 2014-06-27 13:52:20 +0200 | [diff] [blame] | 86 | context = { |
Victor Stinner | a02f81f | 2014-06-24 22:37:53 +0200 | [diff] [blame] | 87 | 'task': self, |
| 88 | 'message': 'Task was destroyed but it is pending!', |
Victor Stinner | 80f53aa | 2014-06-27 13:52:20 +0200 | [diff] [blame] | 89 | } |
| 90 | if self._source_traceback: |
| 91 | context['source_traceback'] = self._source_traceback |
| 92 | self._loop.call_exception_handler(context) |
Victor Stinner | a02f81f | 2014-06-24 22:37:53 +0200 | [diff] [blame] | 93 | futures.Future.__del__(self) |
| 94 | |
Victor Stinner | 313a980 | 2014-07-29 12:58:23 +0200 | [diff] [blame] | 95 | def _repr_info(self): |
| 96 | info = super()._repr_info() |
| 97 | |
Victor Stinner | 975735f | 2014-06-25 21:41:58 +0200 | [diff] [blame] | 98 | if self._must_cancel: |
Victor Stinner | 313a980 | 2014-07-29 12:58:23 +0200 | [diff] [blame] | 99 | # replace status |
| 100 | info[0] = 'cancelling' |
Victor Stinner | 975735f | 2014-06-25 21:41:58 +0200 | [diff] [blame] | 101 | |
Victor Stinner | c39ba7d | 2014-07-11 00:21:27 +0200 | [diff] [blame] | 102 | coro = coroutines._format_coroutine(self._coro) |
Victor Stinner | 313a980 | 2014-07-29 12:58:23 +0200 | [diff] [blame] | 103 | info.insert(1, 'coro=<%s>' % coro) |
Victor Stinner | 975735f | 2014-06-25 21:41:58 +0200 | [diff] [blame] | 104 | |
Victor Stinner | 2dba23a | 2014-07-03 00:59:00 +0200 | [diff] [blame] | 105 | if self._fut_waiter is not None: |
Victor Stinner | 313a980 | 2014-07-29 12:58:23 +0200 | [diff] [blame] | 106 | info.insert(2, 'wait_for=%r' % self._fut_waiter) |
| 107 | return info |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 108 | |
| 109 | def get_stack(self, *, limit=None): |
| 110 | """Return the list of stack frames for this task's coroutine. |
| 111 | |
Victor Stinner | d87de83 | 2014-12-02 17:57:04 +0100 | [diff] [blame] | 112 | If the coroutine is not done, this returns the stack where it is |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 113 | suspended. If the coroutine has completed successfully or was |
| 114 | cancelled, this returns an empty list. If the coroutine was |
| 115 | terminated by an exception, this returns the list of traceback |
| 116 | frames. |
| 117 | |
| 118 | The frames are always ordered from oldest to newest. |
| 119 | |
Yury Selivanov | b0b0e62 | 2014-02-18 22:27:48 -0500 | [diff] [blame] | 120 | The optional limit gives the maximum number of frames to |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 121 | return; by default all available frames are returned. Its |
| 122 | meaning differs depending on whether a stack or a traceback is |
| 123 | returned: the newest frames of a stack are returned, but the |
| 124 | oldest frames of a traceback are returned. (This matches the |
| 125 | behavior of the traceback module.) |
| 126 | |
| 127 | For reasons beyond our control, only one stack frame is |
| 128 | returned for a suspended coroutine. |
| 129 | """ |
| 130 | frames = [] |
Yury Selivanov | 2339833 | 2015-08-14 15:30:59 -0400 | [diff] [blame] | 131 | try: |
| 132 | # 'async def' coroutines |
| 133 | f = self._coro.cr_frame |
| 134 | except AttributeError: |
| 135 | f = self._coro.gi_frame |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 136 | if f is not None: |
| 137 | while f is not None: |
| 138 | if limit is not None: |
| 139 | if limit <= 0: |
| 140 | break |
| 141 | limit -= 1 |
| 142 | frames.append(f) |
| 143 | f = f.f_back |
| 144 | frames.reverse() |
| 145 | elif self._exception is not None: |
| 146 | tb = self._exception.__traceback__ |
| 147 | while tb is not None: |
| 148 | if limit is not None: |
| 149 | if limit <= 0: |
| 150 | break |
| 151 | limit -= 1 |
| 152 | frames.append(tb.tb_frame) |
| 153 | tb = tb.tb_next |
| 154 | return frames |
| 155 | |
| 156 | def print_stack(self, *, limit=None, file=None): |
| 157 | """Print the stack or traceback for this task's coroutine. |
| 158 | |
| 159 | This produces output similar to that of the traceback module, |
| 160 | for the frames retrieved by get_stack(). The limit argument |
| 161 | is passed to get_stack(). The file argument is an I/O stream |
R David Murray | 8e069d5 | 2014-09-24 13:13:45 -0400 | [diff] [blame] | 162 | to which the output is written; by default output is written |
| 163 | to sys.stderr. |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 164 | """ |
| 165 | extracted_list = [] |
| 166 | checked = set() |
| 167 | for f in self.get_stack(limit=limit): |
| 168 | lineno = f.f_lineno |
| 169 | co = f.f_code |
| 170 | filename = co.co_filename |
| 171 | name = co.co_name |
| 172 | if filename not in checked: |
| 173 | checked.add(filename) |
| 174 | linecache.checkcache(filename) |
| 175 | line = linecache.getline(filename, lineno, f.f_globals) |
| 176 | extracted_list.append((filename, lineno, name, line)) |
| 177 | exc = self._exception |
| 178 | if not extracted_list: |
| 179 | print('No stack for %r' % self, file=file) |
| 180 | elif exc is not None: |
| 181 | print('Traceback for %r (most recent call last):' % self, |
| 182 | file=file) |
| 183 | else: |
| 184 | print('Stack for %r (most recent call last):' % self, |
| 185 | file=file) |
| 186 | traceback.print_list(extracted_list, file=file) |
| 187 | if exc is not None: |
| 188 | for line in traceback.format_exception_only(exc.__class__, exc): |
| 189 | print(line, file=file, end='') |
| 190 | |
| 191 | def cancel(self): |
R David Murray | 8e069d5 | 2014-09-24 13:13:45 -0400 | [diff] [blame] | 192 | """Request that this task cancel itself. |
Victor Stinner | 4bd652a | 2014-04-07 11:18:06 +0200 | [diff] [blame] | 193 | |
Victor Stinner | 8d21357 | 2014-06-02 23:06:46 +0200 | [diff] [blame] | 194 | This arranges for a CancelledError to be thrown into the |
Victor Stinner | 4bd652a | 2014-04-07 11:18:06 +0200 | [diff] [blame] | 195 | wrapped coroutine on the next cycle through the event loop. |
| 196 | The coroutine then has a chance to clean up or even deny |
| 197 | the request using try/except/finally. |
| 198 | |
R David Murray | 8e069d5 | 2014-09-24 13:13:45 -0400 | [diff] [blame] | 199 | Unlike Future.cancel, this does not guarantee that the |
Victor Stinner | 4bd652a | 2014-04-07 11:18:06 +0200 | [diff] [blame] | 200 | task will be cancelled: the exception might be caught and |
R David Murray | 8e069d5 | 2014-09-24 13:13:45 -0400 | [diff] [blame] | 201 | acted upon, delaying cancellation of the task or preventing |
| 202 | cancellation completely. The task may also return a value or |
| 203 | raise a different exception. |
Victor Stinner | 4bd652a | 2014-04-07 11:18:06 +0200 | [diff] [blame] | 204 | |
| 205 | Immediately after this method is called, Task.cancelled() will |
| 206 | not return True (unless the task was already cancelled). A |
| 207 | task will be marked as cancelled when the wrapped coroutine |
| 208 | terminates with a CancelledError exception (even if cancel() |
| 209 | was not called). |
| 210 | """ |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 211 | if self.done(): |
| 212 | return False |
| 213 | if self._fut_waiter is not None: |
| 214 | if self._fut_waiter.cancel(): |
| 215 | # Leave self._fut_waiter; it may be a Task that |
| 216 | # catches and ignores the cancellation so we may have |
| 217 | # to cancel it again later. |
| 218 | return True |
| 219 | # It must be the case that self._step is already scheduled. |
| 220 | self._must_cancel = True |
| 221 | return True |
| 222 | |
Yury Selivanov | d59bba8 | 2015-11-20 12:41:03 -0500 | [diff] [blame] | 223 | def _step(self, exc=None): |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 224 | assert not self.done(), \ |
Yury Selivanov | d59bba8 | 2015-11-20 12:41:03 -0500 | [diff] [blame] | 225 | '_step(): already done: {!r}, {!r}'.format(self, exc) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 226 | if self._must_cancel: |
| 227 | if not isinstance(exc, futures.CancelledError): |
| 228 | exc = futures.CancelledError() |
| 229 | self._must_cancel = False |
| 230 | coro = self._coro |
| 231 | self._fut_waiter = None |
Guido van Rossum | 1a605ed | 2013-12-06 12:57:40 -0800 | [diff] [blame] | 232 | |
| 233 | self.__class__._current_tasks[self._loop] = self |
Yury Selivanov | d59bba8 | 2015-11-20 12:41:03 -0500 | [diff] [blame] | 234 | # Call either coro.throw(exc) or coro.send(None). |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 235 | try: |
Yury Selivanov | d59bba8 | 2015-11-20 12:41:03 -0500 | [diff] [blame] | 236 | if exc is None: |
| 237 | # We use the `send` method directly, because coroutines |
| 238 | # don't have `__iter__` and `__next__` methods. |
| 239 | result = coro.send(None) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 240 | else: |
Yury Selivanov | d59bba8 | 2015-11-20 12:41:03 -0500 | [diff] [blame] | 241 | result = coro.throw(exc) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 242 | except StopIteration as exc: |
| 243 | self.set_result(exc.value) |
| 244 | except futures.CancelledError as exc: |
| 245 | super().cancel() # I.e., Future.cancel(self). |
| 246 | except Exception as exc: |
| 247 | self.set_exception(exc) |
| 248 | except BaseException as exc: |
| 249 | self.set_exception(exc) |
| 250 | raise |
| 251 | else: |
Guido van Rossum | 1140a03 | 2016-09-09 12:54:54 -0700 | [diff] [blame] | 252 | blocking = getattr(result, '_asyncio_future_blocking', None) |
| 253 | if blocking is not None: |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 254 | # Yielded Future must come from Future.__iter__(). |
Yury Selivanov | 0ac3a0c | 2015-12-11 11:33:59 -0500 | [diff] [blame] | 255 | if result._loop is not self._loop: |
| 256 | self._loop.call_soon( |
| 257 | self._step, |
| 258 | RuntimeError( |
| 259 | 'Task {!r} got Future {!r} attached to a ' |
| 260 | 'different loop'.format(self, result))) |
Guido van Rossum | 1140a03 | 2016-09-09 12:54:54 -0700 | [diff] [blame] | 261 | elif blocking: |
| 262 | result._asyncio_future_blocking = False |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 263 | result.add_done_callback(self._wakeup) |
| 264 | self._fut_waiter = result |
Yury Selivanov | 4c0a09a | 2015-08-02 16:49:31 -0400 | [diff] [blame] | 265 | if self._must_cancel: |
| 266 | if self._fut_waiter.cancel(): |
| 267 | self._must_cancel = False |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 268 | else: |
| 269 | self._loop.call_soon( |
Yury Selivanov | d59bba8 | 2015-11-20 12:41:03 -0500 | [diff] [blame] | 270 | self._step, |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 271 | RuntimeError( |
| 272 | 'yield was used instead of yield from ' |
| 273 | 'in task {!r} with {!r}'.format(self, result))) |
| 274 | elif result is None: |
| 275 | # Bare yield relinquishes control for one event loop iteration. |
| 276 | self._loop.call_soon(self._step) |
| 277 | elif inspect.isgenerator(result): |
| 278 | # Yielding a generator is just wrong. |
| 279 | self._loop.call_soon( |
Yury Selivanov | d59bba8 | 2015-11-20 12:41:03 -0500 | [diff] [blame] | 280 | self._step, |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 281 | RuntimeError( |
| 282 | 'yield was used instead of yield from for ' |
| 283 | 'generator in task {!r} with {}'.format( |
| 284 | self, result))) |
| 285 | else: |
| 286 | # Yielding something else is an error. |
| 287 | self._loop.call_soon( |
Yury Selivanov | d59bba8 | 2015-11-20 12:41:03 -0500 | [diff] [blame] | 288 | self._step, |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 289 | RuntimeError( |
| 290 | 'Task got bad yield: {!r}'.format(result))) |
Guido van Rossum | 1a605ed | 2013-12-06 12:57:40 -0800 | [diff] [blame] | 291 | finally: |
| 292 | self.__class__._current_tasks.pop(self._loop) |
Victor Stinner | d74ac82 | 2014-03-04 23:07:08 +0100 | [diff] [blame] | 293 | self = None # Needed to break cycles when an exception occurs. |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 294 | |
| 295 | def _wakeup(self, future): |
| 296 | try: |
Yury Selivanov | a4afc48 | 2015-11-16 15:12:10 -0500 | [diff] [blame] | 297 | future.result() |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 298 | except Exception as exc: |
| 299 | # This may also be a cancellation. |
Yury Selivanov | d59bba8 | 2015-11-20 12:41:03 -0500 | [diff] [blame] | 300 | self._step(exc) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 301 | else: |
Yury Selivanov | a4afc48 | 2015-11-16 15:12:10 -0500 | [diff] [blame] | 302 | # Don't pass the value of `future.result()` explicitly, |
| 303 | # as `Future.__iter__` and `Future.__await__` don't need it. |
| 304 | # If we call `_step(value, None)` instead of `_step()`, |
| 305 | # Python eval loop would use `.send(value)` method call, |
| 306 | # instead of `__next__()`, which is slower for futures |
| 307 | # that return non-generator iterators from their `__iter__`. |
| 308 | self._step() |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 309 | self = None # Needed to break cycles when an exception occurs. |
| 310 | |
| 311 | |
| 312 | # wait() and as_completed() similar to those in PEP 3148. |
| 313 | |
| 314 | FIRST_COMPLETED = concurrent.futures.FIRST_COMPLETED |
| 315 | FIRST_EXCEPTION = concurrent.futures.FIRST_EXCEPTION |
| 316 | ALL_COMPLETED = concurrent.futures.ALL_COMPLETED |
| 317 | |
| 318 | |
| 319 | @coroutine |
| 320 | def wait(fs, *, loop=None, timeout=None, return_when=ALL_COMPLETED): |
| 321 | """Wait for the Futures and coroutines given by fs to complete. |
| 322 | |
Victor Stinner | db74d98 | 2014-06-10 11:16:05 +0200 | [diff] [blame] | 323 | The sequence futures must not be empty. |
| 324 | |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 325 | Coroutines will be wrapped in Tasks. |
| 326 | |
| 327 | Returns two sets of Future: (done, pending). |
| 328 | |
| 329 | Usage: |
| 330 | |
| 331 | done, pending = yield from asyncio.wait(fs) |
| 332 | |
| 333 | Note: This does not raise TimeoutError! Futures that aren't done |
| 334 | when the timeout occurs are returned in the second set. |
| 335 | """ |
Guido van Rossum | 7b3b3dc | 2016-09-09 14:26:31 -0700 | [diff] [blame] | 336 | if futures.isfuture(fs) or coroutines.iscoroutine(fs): |
Victor Stinner | eb74876 | 2014-02-11 11:54:08 +0100 | [diff] [blame] | 337 | raise TypeError("expect a list of futures, not %s" % type(fs).__name__) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 338 | if not fs: |
| 339 | raise ValueError('Set of coroutines/Futures is empty.') |
Victor Stinner | e931f7b | 2014-07-16 18:50:39 +0200 | [diff] [blame] | 340 | if return_when not in (FIRST_COMPLETED, FIRST_EXCEPTION, ALL_COMPLETED): |
| 341 | raise ValueError('Invalid return_when value: {}'.format(return_when)) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 342 | |
| 343 | if loop is None: |
| 344 | loop = events.get_event_loop() |
| 345 | |
Yury Selivanov | 59eb9a4 | 2015-05-11 14:48:38 -0400 | [diff] [blame] | 346 | fs = {ensure_future(f, loop=loop) for f in set(fs)} |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 347 | |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 348 | return (yield from _wait(fs, timeout, return_when, loop)) |
| 349 | |
| 350 | |
Victor Stinner | 59e0802 | 2014-08-28 11:19:25 +0200 | [diff] [blame] | 351 | def _release_waiter(waiter, *args): |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 352 | if not waiter.done(): |
Victor Stinner | 59e0802 | 2014-08-28 11:19:25 +0200 | [diff] [blame] | 353 | waiter.set_result(None) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 354 | |
| 355 | |
| 356 | @coroutine |
| 357 | def wait_for(fut, timeout, *, loop=None): |
| 358 | """Wait for the single Future or coroutine to complete, with timeout. |
| 359 | |
| 360 | Coroutine will be wrapped in Task. |
| 361 | |
Victor Stinner | 421e49b | 2014-01-23 17:40:59 +0100 | [diff] [blame] | 362 | Returns result of the Future or coroutine. When a timeout occurs, |
| 363 | it cancels the task and raises TimeoutError. To avoid the task |
| 364 | cancellation, wrap it in shield(). |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 365 | |
Victor Stinner | 922bc2c | 2015-01-15 16:29:10 +0100 | [diff] [blame] | 366 | If the wait is cancelled, the task is also cancelled. |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 367 | |
Victor Stinner | 922bc2c | 2015-01-15 16:29:10 +0100 | [diff] [blame] | 368 | This function is a coroutine. |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 369 | """ |
| 370 | if loop is None: |
| 371 | loop = events.get_event_loop() |
| 372 | |
Guido van Rossum | 48c66c3 | 2014-01-29 14:30:38 -0800 | [diff] [blame] | 373 | if timeout is None: |
| 374 | return (yield from fut) |
| 375 | |
Yury Selivanov | 7661db6 | 2016-05-16 15:38:39 -0400 | [diff] [blame] | 376 | waiter = loop.create_future() |
Victor Stinner | 59e0802 | 2014-08-28 11:19:25 +0200 | [diff] [blame] | 377 | timeout_handle = loop.call_later(timeout, _release_waiter, waiter) |
| 378 | cb = functools.partial(_release_waiter, waiter) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 379 | |
Yury Selivanov | 59eb9a4 | 2015-05-11 14:48:38 -0400 | [diff] [blame] | 380 | fut = ensure_future(fut, loop=loop) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 381 | fut.add_done_callback(cb) |
| 382 | |
| 383 | try: |
Victor Stinner | 59e0802 | 2014-08-28 11:19:25 +0200 | [diff] [blame] | 384 | # wait until the future completes or the timeout |
Victor Stinner | 922bc2c | 2015-01-15 16:29:10 +0100 | [diff] [blame] | 385 | try: |
| 386 | yield from waiter |
| 387 | except futures.CancelledError: |
| 388 | fut.remove_done_callback(cb) |
| 389 | fut.cancel() |
| 390 | raise |
Victor Stinner | 59e0802 | 2014-08-28 11:19:25 +0200 | [diff] [blame] | 391 | |
| 392 | if fut.done(): |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 393 | return fut.result() |
| 394 | else: |
| 395 | fut.remove_done_callback(cb) |
Victor Stinner | 421e49b | 2014-01-23 17:40:59 +0100 | [diff] [blame] | 396 | fut.cancel() |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 397 | raise futures.TimeoutError() |
| 398 | finally: |
| 399 | timeout_handle.cancel() |
| 400 | |
| 401 | |
| 402 | @coroutine |
| 403 | def _wait(fs, timeout, return_when, loop): |
Victor Stinner | 2ba8ece | 2016-04-01 21:39:09 +0200 | [diff] [blame] | 404 | """Internal helper for wait() and wait_for(). |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 405 | |
| 406 | The fs argument must be a collection of Futures. |
| 407 | """ |
| 408 | assert fs, 'Set of Futures is empty.' |
Yury Selivanov | 7661db6 | 2016-05-16 15:38:39 -0400 | [diff] [blame] | 409 | waiter = loop.create_future() |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 410 | timeout_handle = None |
| 411 | if timeout is not None: |
| 412 | timeout_handle = loop.call_later(timeout, _release_waiter, waiter) |
| 413 | counter = len(fs) |
| 414 | |
| 415 | def _on_completion(f): |
| 416 | nonlocal counter |
| 417 | counter -= 1 |
| 418 | if (counter <= 0 or |
| 419 | return_when == FIRST_COMPLETED or |
| 420 | return_when == FIRST_EXCEPTION and (not f.cancelled() and |
| 421 | f.exception() is not None)): |
| 422 | if timeout_handle is not None: |
| 423 | timeout_handle.cancel() |
| 424 | if not waiter.done(): |
Victor Stinner | 59e0802 | 2014-08-28 11:19:25 +0200 | [diff] [blame] | 425 | waiter.set_result(None) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 426 | |
| 427 | for f in fs: |
| 428 | f.add_done_callback(_on_completion) |
| 429 | |
| 430 | try: |
| 431 | yield from waiter |
| 432 | finally: |
| 433 | if timeout_handle is not None: |
| 434 | timeout_handle.cancel() |
| 435 | |
| 436 | done, pending = set(), set() |
| 437 | for f in fs: |
| 438 | f.remove_done_callback(_on_completion) |
| 439 | if f.done(): |
| 440 | done.add(f) |
| 441 | else: |
| 442 | pending.add(f) |
| 443 | return done, pending |
| 444 | |
| 445 | |
| 446 | # This is *not* a @coroutine! It is just an iterator (yielding Futures). |
| 447 | def as_completed(fs, *, loop=None, timeout=None): |
Guido van Rossum | b58f053 | 2014-02-12 17:58:19 -0800 | [diff] [blame] | 448 | """Return an iterator whose values are coroutines. |
| 449 | |
| 450 | When waiting for the yielded coroutines you'll get the results (or |
| 451 | exceptions!) of the original Futures (or coroutines), in the order |
| 452 | in which and as soon as they complete. |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 453 | |
| 454 | This differs from PEP 3148; the proper way to use this is: |
| 455 | |
| 456 | for f in as_completed(fs): |
| 457 | result = yield from f # The 'yield from' may raise. |
| 458 | # Use result. |
| 459 | |
Guido van Rossum | b58f053 | 2014-02-12 17:58:19 -0800 | [diff] [blame] | 460 | If a timeout is specified, the 'yield from' will raise |
| 461 | TimeoutError when the timeout occurs before all Futures are done. |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 462 | |
| 463 | Note: The futures 'f' are not necessarily members of fs. |
| 464 | """ |
Guido van Rossum | 7b3b3dc | 2016-09-09 14:26:31 -0700 | [diff] [blame] | 465 | if futures.isfuture(fs) or coroutines.iscoroutine(fs): |
Victor Stinner | eb74876 | 2014-02-11 11:54:08 +0100 | [diff] [blame] | 466 | raise TypeError("expect a list of futures, not %s" % type(fs).__name__) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 467 | loop = loop if loop is not None else events.get_event_loop() |
Yury Selivanov | 59eb9a4 | 2015-05-11 14:48:38 -0400 | [diff] [blame] | 468 | todo = {ensure_future(f, loop=loop) for f in set(fs)} |
Guido van Rossum | b58f053 | 2014-02-12 17:58:19 -0800 | [diff] [blame] | 469 | from .queues import Queue # Import here to avoid circular import problem. |
| 470 | done = Queue(loop=loop) |
| 471 | timeout_handle = None |
| 472 | |
| 473 | def _on_timeout(): |
| 474 | for f in todo: |
| 475 | f.remove_done_callback(_on_completion) |
| 476 | done.put_nowait(None) # Queue a dummy value for _wait_for_one(). |
| 477 | todo.clear() # Can't do todo.remove(f) in the loop. |
| 478 | |
| 479 | def _on_completion(f): |
| 480 | if not todo: |
| 481 | return # _on_timeout() was here first. |
| 482 | todo.remove(f) |
| 483 | done.put_nowait(f) |
| 484 | if not todo and timeout_handle is not None: |
| 485 | timeout_handle.cancel() |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 486 | |
| 487 | @coroutine |
| 488 | def _wait_for_one(): |
Guido van Rossum | b58f053 | 2014-02-12 17:58:19 -0800 | [diff] [blame] | 489 | f = yield from done.get() |
| 490 | if f is None: |
| 491 | # Dummy value from _on_timeout(). |
| 492 | raise futures.TimeoutError |
| 493 | return f.result() # May raise f.exception(). |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 494 | |
Guido van Rossum | b58f053 | 2014-02-12 17:58:19 -0800 | [diff] [blame] | 495 | for f in todo: |
| 496 | f.add_done_callback(_on_completion) |
| 497 | if todo and timeout is not None: |
| 498 | timeout_handle = loop.call_later(timeout, _on_timeout) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 499 | for _ in range(len(todo)): |
| 500 | yield _wait_for_one() |
| 501 | |
| 502 | |
| 503 | @coroutine |
| 504 | def sleep(delay, result=None, *, loop=None): |
| 505 | """Coroutine that completes after a given time (in seconds).""" |
Yury Selivanov | ade0412 | 2015-11-05 14:29:04 -0500 | [diff] [blame] | 506 | if delay == 0: |
| 507 | yield |
| 508 | return result |
| 509 | |
Yury Selivanov | 7661db6 | 2016-05-16 15:38:39 -0400 | [diff] [blame] | 510 | if loop is None: |
| 511 | loop = events.get_event_loop() |
| 512 | future = loop.create_future() |
Victor Stinner | a9acbe8 | 2014-07-05 15:29:41 +0200 | [diff] [blame] | 513 | h = future._loop.call_later(delay, |
Yury Selivanov | 5d7e3b6 | 2015-11-17 12:19:41 -0500 | [diff] [blame] | 514 | futures._set_result_unless_cancelled, |
| 515 | future, result) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 516 | try: |
| 517 | return (yield from future) |
| 518 | finally: |
| 519 | h.cancel() |
| 520 | |
| 521 | |
Yury Selivanov | 4357cf6 | 2016-09-15 13:49:08 -0400 | [diff] [blame] | 522 | def async_(coro_or_future, *, loop=None): |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 523 | """Wrap a coroutine in a future. |
| 524 | |
| 525 | If the argument is a Future, it is returned directly. |
Yury Selivanov | 59eb9a4 | 2015-05-11 14:48:38 -0400 | [diff] [blame] | 526 | |
| 527 | This function is deprecated in 3.5. Use asyncio.ensure_future() instead. |
| 528 | """ |
| 529 | |
| 530 | warnings.warn("asyncio.async() function is deprecated, use ensure_future()", |
| 531 | DeprecationWarning) |
| 532 | |
| 533 | return ensure_future(coro_or_future, loop=loop) |
| 534 | |
Yury Selivanov | 4357cf6 | 2016-09-15 13:49:08 -0400 | [diff] [blame] | 535 | # Silence DeprecationWarning: |
| 536 | globals()['async'] = async_ |
| 537 | async_.__name__ = 'async' |
| 538 | del async_ |
| 539 | |
Yury Selivanov | 59eb9a4 | 2015-05-11 14:48:38 -0400 | [diff] [blame] | 540 | |
| 541 | def ensure_future(coro_or_future, *, loop=None): |
Yury Selivanov | 620279b | 2015-10-02 15:00:19 -0400 | [diff] [blame] | 542 | """Wrap a coroutine or an awaitable in a future. |
Yury Selivanov | 59eb9a4 | 2015-05-11 14:48:38 -0400 | [diff] [blame] | 543 | |
| 544 | If the argument is a Future, it is returned directly. |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 545 | """ |
Guido van Rossum | 7b3b3dc | 2016-09-09 14:26:31 -0700 | [diff] [blame] | 546 | if futures.isfuture(coro_or_future): |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 547 | if loop is not None and loop is not coro_or_future._loop: |
| 548 | raise ValueError('loop argument must agree with Future') |
| 549 | return coro_or_future |
Victor Stinner | f951d28 | 2014-06-29 00:46:45 +0200 | [diff] [blame] | 550 | elif coroutines.iscoroutine(coro_or_future): |
Victor Stinner | 896a25a | 2014-07-08 11:29:25 +0200 | [diff] [blame] | 551 | if loop is None: |
| 552 | loop = events.get_event_loop() |
| 553 | task = loop.create_task(coro_or_future) |
Victor Stinner | 80f53aa | 2014-06-27 13:52:20 +0200 | [diff] [blame] | 554 | if task._source_traceback: |
| 555 | del task._source_traceback[-1] |
| 556 | return task |
Yury Selivanov | 620279b | 2015-10-02 15:00:19 -0400 | [diff] [blame] | 557 | elif compat.PY35 and inspect.isawaitable(coro_or_future): |
| 558 | return ensure_future(_wrap_awaitable(coro_or_future), loop=loop) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 559 | else: |
Yury Selivanov | 620279b | 2015-10-02 15:00:19 -0400 | [diff] [blame] | 560 | raise TypeError('A Future, a coroutine or an awaitable is required') |
| 561 | |
| 562 | |
| 563 | @coroutine |
| 564 | def _wrap_awaitable(awaitable): |
| 565 | """Helper for asyncio.ensure_future(). |
| 566 | |
| 567 | Wraps awaitable (an object with __await__) into a coroutine |
| 568 | that will later be wrapped in a Task by ensure_future(). |
| 569 | """ |
| 570 | return (yield from awaitable.__await__()) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 571 | |
| 572 | |
| 573 | class _GatheringFuture(futures.Future): |
| 574 | """Helper for gather(). |
| 575 | |
| 576 | This overrides cancel() to cancel all the children and act more |
| 577 | like Task.cancel(), which doesn't immediately mark itself as |
| 578 | cancelled. |
| 579 | """ |
| 580 | |
| 581 | def __init__(self, children, *, loop=None): |
| 582 | super().__init__(loop=loop) |
| 583 | self._children = children |
| 584 | |
| 585 | def cancel(self): |
| 586 | if self.done(): |
| 587 | return False |
| 588 | for child in self._children: |
| 589 | child.cancel() |
| 590 | return True |
| 591 | |
| 592 | |
| 593 | def gather(*coros_or_futures, loop=None, return_exceptions=False): |
| 594 | """Return a future aggregating results from the given coroutines |
| 595 | or futures. |
| 596 | |
| 597 | All futures must share the same event loop. If all the tasks are |
| 598 | done successfully, the returned future's result is the list of |
| 599 | results (in the order of the original sequence, not necessarily |
Yury Selivanov | f317cb7 | 2014-02-06 12:03:53 -0500 | [diff] [blame] | 600 | the order of results arrival). If *return_exceptions* is True, |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 601 | exceptions in the tasks are treated the same as successful |
| 602 | results, and gathered in the result list; otherwise, the first |
| 603 | raised exception will be immediately propagated to the returned |
| 604 | future. |
| 605 | |
| 606 | Cancellation: if the outer Future is cancelled, all children (that |
| 607 | have not completed yet) are also cancelled. If any child is |
| 608 | cancelled, this is treated as if it raised CancelledError -- |
| 609 | the outer Future is *not* cancelled in this case. (This is to |
| 610 | prevent the cancellation of one child to cause other children to |
| 611 | be cancelled.) |
| 612 | """ |
Victor Stinner | f03b3c7 | 2014-07-16 18:36:24 +0200 | [diff] [blame] | 613 | if not coros_or_futures: |
Yury Selivanov | 7661db6 | 2016-05-16 15:38:39 -0400 | [diff] [blame] | 614 | if loop is None: |
| 615 | loop = events.get_event_loop() |
| 616 | outer = loop.create_future() |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 617 | outer.set_result([]) |
| 618 | return outer |
Victor Stinner | f03b3c7 | 2014-07-16 18:36:24 +0200 | [diff] [blame] | 619 | |
| 620 | arg_to_fut = {} |
| 621 | for arg in set(coros_or_futures): |
Guido van Rossum | 7b3b3dc | 2016-09-09 14:26:31 -0700 | [diff] [blame] | 622 | if not futures.isfuture(arg): |
Yury Selivanov | 59eb9a4 | 2015-05-11 14:48:38 -0400 | [diff] [blame] | 623 | fut = ensure_future(arg, loop=loop) |
Victor Stinner | f03b3c7 | 2014-07-16 18:36:24 +0200 | [diff] [blame] | 624 | if loop is None: |
| 625 | loop = fut._loop |
| 626 | # The caller cannot control this future, the "destroy pending task" |
| 627 | # warning should not be emitted. |
| 628 | fut._log_destroy_pending = False |
| 629 | else: |
| 630 | fut = arg |
| 631 | if loop is None: |
| 632 | loop = fut._loop |
| 633 | elif fut._loop is not loop: |
| 634 | raise ValueError("futures are tied to different event loops") |
| 635 | arg_to_fut[arg] = fut |
| 636 | |
| 637 | children = [arg_to_fut[arg] for arg in coros_or_futures] |
| 638 | nchildren = len(children) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 639 | outer = _GatheringFuture(children, loop=loop) |
| 640 | nfinished = 0 |
Victor Stinner | f03b3c7 | 2014-07-16 18:36:24 +0200 | [diff] [blame] | 641 | results = [None] * nchildren |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 642 | |
| 643 | def _done_callback(i, fut): |
| 644 | nonlocal nfinished |
Victor Stinner | 3531d90 | 2015-01-09 01:42:52 +0100 | [diff] [blame] | 645 | if outer.done(): |
| 646 | if not fut.cancelled(): |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 647 | # Mark exception retrieved. |
| 648 | fut.exception() |
| 649 | return |
Victor Stinner | 3531d90 | 2015-01-09 01:42:52 +0100 | [diff] [blame] | 650 | |
Victor Stinner | 2934262 | 2015-01-29 14:15:19 +0100 | [diff] [blame] | 651 | if fut.cancelled(): |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 652 | res = futures.CancelledError() |
| 653 | if not return_exceptions: |
| 654 | outer.set_exception(res) |
| 655 | return |
| 656 | elif fut._exception is not None: |
| 657 | res = fut.exception() # Mark exception retrieved. |
| 658 | if not return_exceptions: |
| 659 | outer.set_exception(res) |
| 660 | return |
| 661 | else: |
| 662 | res = fut._result |
| 663 | results[i] = res |
| 664 | nfinished += 1 |
Victor Stinner | f03b3c7 | 2014-07-16 18:36:24 +0200 | [diff] [blame] | 665 | if nfinished == nchildren: |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 666 | outer.set_result(results) |
| 667 | |
| 668 | for i, fut in enumerate(children): |
| 669 | fut.add_done_callback(functools.partial(_done_callback, i)) |
| 670 | return outer |
| 671 | |
| 672 | |
| 673 | def shield(arg, *, loop=None): |
| 674 | """Wait for a future, shielding it from cancellation. |
| 675 | |
| 676 | The statement |
| 677 | |
| 678 | res = yield from shield(something()) |
| 679 | |
| 680 | is exactly equivalent to the statement |
| 681 | |
| 682 | res = yield from something() |
| 683 | |
| 684 | *except* that if the coroutine containing it is cancelled, the |
| 685 | task running in something() is not cancelled. From the POV of |
| 686 | something(), the cancellation did not happen. But its caller is |
| 687 | still cancelled, so the yield-from expression still raises |
| 688 | CancelledError. Note: If something() is cancelled by other means |
| 689 | this will still cancel shield(). |
| 690 | |
| 691 | If you want to completely ignore cancellation (not recommended) |
| 692 | you can combine shield() with a try/except clause, as follows: |
| 693 | |
| 694 | try: |
| 695 | res = yield from shield(something()) |
| 696 | except CancelledError: |
| 697 | res = None |
| 698 | """ |
Yury Selivanov | 59eb9a4 | 2015-05-11 14:48:38 -0400 | [diff] [blame] | 699 | inner = ensure_future(arg, loop=loop) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 700 | if inner.done(): |
| 701 | # Shortcut. |
| 702 | return inner |
| 703 | loop = inner._loop |
Yury Selivanov | 7661db6 | 2016-05-16 15:38:39 -0400 | [diff] [blame] | 704 | outer = loop.create_future() |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 705 | |
| 706 | def _done_callback(inner): |
| 707 | if outer.cancelled(): |
Victor Stinner | 3531d90 | 2015-01-09 01:42:52 +0100 | [diff] [blame] | 708 | if not inner.cancelled(): |
| 709 | # Mark inner's result as retrieved. |
| 710 | inner.exception() |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 711 | return |
Victor Stinner | 3531d90 | 2015-01-09 01:42:52 +0100 | [diff] [blame] | 712 | |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 713 | if inner.cancelled(): |
| 714 | outer.cancel() |
| 715 | else: |
| 716 | exc = inner.exception() |
| 717 | if exc is not None: |
| 718 | outer.set_exception(exc) |
| 719 | else: |
| 720 | outer.set_result(inner.result()) |
| 721 | |
| 722 | inner.add_done_callback(_done_callback) |
| 723 | return outer |
Guido van Rossum | 841d9ee | 2015-10-03 08:31:42 -0700 | [diff] [blame] | 724 | |
| 725 | |
| 726 | def run_coroutine_threadsafe(coro, loop): |
| 727 | """Submit a coroutine object to a given event loop. |
| 728 | |
| 729 | Return a concurrent.futures.Future to access the result. |
| 730 | """ |
| 731 | if not coroutines.iscoroutine(coro): |
| 732 | raise TypeError('A coroutine object is required') |
| 733 | future = concurrent.futures.Future() |
| 734 | |
| 735 | def callback(): |
Guido van Rossum | 601953b | 2015-10-05 16:20:00 -0700 | [diff] [blame] | 736 | try: |
| 737 | futures._chain_future(ensure_future(coro, loop=loop), future) |
| 738 | except Exception as exc: |
| 739 | if future.set_running_or_notify_cancel(): |
| 740 | future.set_exception(exc) |
| 741 | raise |
Guido van Rossum | 841d9ee | 2015-10-03 08:31:42 -0700 | [diff] [blame] | 742 | |
| 743 | loop.call_soon_threadsafe(callback) |
| 744 | return future |