Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 1 | """Support for tasks, coroutines and the scheduler.""" |
| 2 | |
Yury Selivanov | 6370f34 | 2017-12-10 18:36:12 -0500 | [diff] [blame] | 3 | __all__ = ( |
Andrew Svetlov | f74ef45 | 2017-12-15 07:04:38 +0200 | [diff] [blame] | 4 | 'Task', 'create_task', |
Yury Selivanov | 6370f34 | 2017-12-10 18:36:12 -0500 | [diff] [blame] | 5 | 'FIRST_COMPLETED', 'FIRST_EXCEPTION', 'ALL_COMPLETED', |
Yury Selivanov | 9edad3c | 2017-12-11 10:03:48 -0500 | [diff] [blame] | 6 | 'wait', 'wait_for', 'as_completed', 'sleep', |
Yury Selivanov | 6370f34 | 2017-12-10 18:36:12 -0500 | [diff] [blame] | 7 | 'gather', 'shield', 'ensure_future', 'run_coroutine_threadsafe', |
Andrew Svetlov | 44d1a59 | 2017-12-16 21:58:38 +0200 | [diff] [blame] | 8 | 'current_task', 'all_tasks', |
| 9 | '_register_task', '_unregister_task', '_enter_task', '_leave_task', |
Yury Selivanov | 6370f34 | 2017-12-10 18:36:12 -0500 | [diff] [blame] | 10 | ) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 11 | |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 12 | import concurrent.futures |
Yury Selivanov | f23746a | 2018-01-22 19:11:18 -0500 | [diff] [blame] | 13 | import contextvars |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 14 | import functools |
| 15 | import inspect |
Alex Grönholm | cca4eec | 2018-08-09 00:06:47 +0300 | [diff] [blame] | 16 | import itertools |
Andrew Svetlov | 5f841b5 | 2017-12-09 00:23:48 +0200 | [diff] [blame] | 17 | import types |
Yury Selivanov | 59eb9a4 | 2015-05-11 14:48:38 -0400 | [diff] [blame] | 18 | import warnings |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 19 | import weakref |
| 20 | |
Yury Selivanov | a0c1ba6 | 2016-10-28 12:52:37 -0400 | [diff] [blame] | 21 | from . import base_tasks |
Victor Stinner | f951d28 | 2014-06-29 00:46:45 +0200 | [diff] [blame] | 22 | from . import coroutines |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 23 | from . import events |
Andrew Svetlov | 0baa72f | 2018-09-11 10:13:04 -0700 | [diff] [blame] | 24 | from . import exceptions |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 25 | from . import futures |
Victor Stinner | f951d28 | 2014-06-29 00:46:45 +0200 | [diff] [blame] | 26 | from .coroutines import coroutine |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 27 | |
Alex Grönholm | cca4eec | 2018-08-09 00:06:47 +0300 | [diff] [blame] | 28 | # Helper to generate new task names |
| 29 | # This uses itertools.count() instead of a "+= 1" operation because the latter |
| 30 | # is not thread safe. See bpo-11866 for a longer explanation. |
| 31 | _task_name_counter = itertools.count(1).__next__ |
| 32 | |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 33 | |
Andrew Svetlov | 44d1a59 | 2017-12-16 21:58:38 +0200 | [diff] [blame] | 34 | def current_task(loop=None): |
| 35 | """Return a currently executed task.""" |
| 36 | if loop is None: |
| 37 | loop = events.get_running_loop() |
| 38 | return _current_tasks.get(loop) |
| 39 | |
| 40 | |
| 41 | def all_tasks(loop=None): |
| 42 | """Return a set of all tasks for the loop.""" |
| 43 | if loop is None: |
Yury Selivanov | 416c1eb | 2018-05-28 17:54:02 -0400 | [diff] [blame] | 44 | loop = events.get_running_loop() |
Andrew Svetlov | 97cf082 | 2018-10-13 21:12:40 +0300 | [diff] [blame] | 45 | # NB: set(_all_tasks) is required to protect |
| 46 | # from https://bugs.python.org/issue34970 bug |
| 47 | return {t for t in list(_all_tasks) |
Yury Selivanov | 416c1eb | 2018-05-28 17:54:02 -0400 | [diff] [blame] | 48 | if futures._get_loop(t) is loop and not t.done()} |
| 49 | |
| 50 | |
| 51 | def _all_tasks_compat(loop=None): |
| 52 | # Different from "all_task()" by returning *all* Tasks, including |
| 53 | # the completed ones. Used to implement deprecated "Tasks.all_task()" |
| 54 | # method. |
| 55 | if loop is None: |
Andrew Svetlov | 44d1a59 | 2017-12-16 21:58:38 +0200 | [diff] [blame] | 56 | loop = events.get_event_loop() |
Andrew Svetlov | 97cf082 | 2018-10-13 21:12:40 +0300 | [diff] [blame] | 57 | # NB: set(_all_tasks) is required to protect |
| 58 | # from https://bugs.python.org/issue34970 bug |
| 59 | return {t for t in list(_all_tasks) if futures._get_loop(t) is loop} |
Andrew Svetlov | 44d1a59 | 2017-12-16 21:58:38 +0200 | [diff] [blame] | 60 | |
| 61 | |
Alex Grönholm | cca4eec | 2018-08-09 00:06:47 +0300 | [diff] [blame] | 62 | def _set_task_name(task, name): |
| 63 | if name is not None: |
| 64 | try: |
| 65 | set_name = task.set_name |
| 66 | except AttributeError: |
| 67 | pass |
| 68 | else: |
| 69 | set_name(name) |
| 70 | |
| 71 | |
Yury Selivanov | 0cf16f9 | 2017-12-25 10:48:15 -0500 | [diff] [blame] | 72 | class Task(futures._PyFuture): # Inherit Python Task implementation |
| 73 | # from a Python Future implementation. |
| 74 | |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 75 | """A coroutine wrapped in a Future.""" |
| 76 | |
| 77 | # An important invariant maintained while a Task not done: |
| 78 | # |
| 79 | # - Either _fut_waiter is None, and _step() is scheduled; |
| 80 | # - or _fut_waiter is some Future, and _step() is *not* scheduled. |
| 81 | # |
| 82 | # The only transition from the latter to the former is through |
| 83 | # _wakeup(). When _fut_waiter is not None, one of its callbacks |
| 84 | # must be _wakeup(). |
| 85 | |
Victor Stinner | fe22e09 | 2014-12-04 23:00:13 +0100 | [diff] [blame] | 86 | # If False, don't log a message if the task is destroyed whereas its |
| 87 | # status is still pending |
| 88 | _log_destroy_pending = True |
| 89 | |
Guido van Rossum | 1a605ed | 2013-12-06 12:57:40 -0800 | [diff] [blame] | 90 | @classmethod |
| 91 | def current_task(cls, loop=None): |
| 92 | """Return the currently running task in an event loop or None. |
| 93 | |
| 94 | By default the current task for the current event loop is returned. |
| 95 | |
| 96 | None is returned when called not in the context of a Task. |
| 97 | """ |
Andrew Svetlov | 44d1a59 | 2017-12-16 21:58:38 +0200 | [diff] [blame] | 98 | warnings.warn("Task.current_task() is deprecated, " |
| 99 | "use asyncio.current_task() instead", |
| 100 | PendingDeprecationWarning, |
| 101 | stacklevel=2) |
Guido van Rossum | 1a605ed | 2013-12-06 12:57:40 -0800 | [diff] [blame] | 102 | if loop is None: |
| 103 | loop = events.get_event_loop() |
Andrew Svetlov | 44d1a59 | 2017-12-16 21:58:38 +0200 | [diff] [blame] | 104 | return current_task(loop) |
Guido van Rossum | 1a605ed | 2013-12-06 12:57:40 -0800 | [diff] [blame] | 105 | |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 106 | @classmethod |
| 107 | def all_tasks(cls, loop=None): |
| 108 | """Return a set of all tasks for an event loop. |
| 109 | |
| 110 | By default all tasks for the current event loop are returned. |
| 111 | """ |
Andrew Svetlov | 44d1a59 | 2017-12-16 21:58:38 +0200 | [diff] [blame] | 112 | warnings.warn("Task.all_tasks() is deprecated, " |
| 113 | "use asyncio.all_tasks() instead", |
| 114 | PendingDeprecationWarning, |
| 115 | stacklevel=2) |
Yury Selivanov | 416c1eb | 2018-05-28 17:54:02 -0400 | [diff] [blame] | 116 | return _all_tasks_compat(loop) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 117 | |
Alex Grönholm | cca4eec | 2018-08-09 00:06:47 +0300 | [diff] [blame] | 118 | def __init__(self, coro, *, loop=None, name=None): |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 119 | super().__init__(loop=loop) |
Victor Stinner | 80f53aa | 2014-06-27 13:52:20 +0200 | [diff] [blame] | 120 | if self._source_traceback: |
| 121 | del self._source_traceback[-1] |
Andrew Svetlov | f74ef45 | 2017-12-15 07:04:38 +0200 | [diff] [blame] | 122 | if not coroutines.iscoroutine(coro): |
| 123 | # raise after Future.__init__(), attrs are required for __del__ |
| 124 | # prevent logging for pending task in __del__ |
| 125 | self._log_destroy_pending = False |
| 126 | raise TypeError(f"a coroutine was expected, got {coro!r}") |
| 127 | |
Alex Grönholm | cca4eec | 2018-08-09 00:06:47 +0300 | [diff] [blame] | 128 | if name is None: |
| 129 | self._name = f'Task-{_task_name_counter()}' |
| 130 | else: |
| 131 | self._name = str(name) |
| 132 | |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 133 | self._must_cancel = False |
Andrew Svetlov | f74ef45 | 2017-12-15 07:04:38 +0200 | [diff] [blame] | 134 | self._fut_waiter = None |
| 135 | self._coro = coro |
Yury Selivanov | f23746a | 2018-01-22 19:11:18 -0500 | [diff] [blame] | 136 | self._context = contextvars.copy_context() |
Andrew Svetlov | f74ef45 | 2017-12-15 07:04:38 +0200 | [diff] [blame] | 137 | |
Yury Selivanov | 22feeb8 | 2018-01-24 11:31:01 -0500 | [diff] [blame] | 138 | self._loop.call_soon(self.__step, context=self._context) |
Yury Selivanov | ca9b36c | 2017-12-23 15:04:15 -0500 | [diff] [blame] | 139 | _register_task(self) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 140 | |
INADA Naoki | 3e2ad8e | 2017-04-25 10:57:18 +0900 | [diff] [blame] | 141 | def __del__(self): |
| 142 | if self._state == futures._PENDING and self._log_destroy_pending: |
| 143 | context = { |
| 144 | 'task': self, |
| 145 | 'message': 'Task was destroyed but it is pending!', |
| 146 | } |
| 147 | if self._source_traceback: |
| 148 | context['source_traceback'] = self._source_traceback |
| 149 | self._loop.call_exception_handler(context) |
Yury Selivanov | 0cf16f9 | 2017-12-25 10:48:15 -0500 | [diff] [blame] | 150 | super().__del__() |
Victor Stinner | a02f81f | 2014-06-24 22:37:53 +0200 | [diff] [blame] | 151 | |
Victor Stinner | 313a980 | 2014-07-29 12:58:23 +0200 | [diff] [blame] | 152 | def _repr_info(self): |
Yury Selivanov | a0c1ba6 | 2016-10-28 12:52:37 -0400 | [diff] [blame] | 153 | return base_tasks._task_repr_info(self) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 154 | |
Alex Grönholm | cca4eec | 2018-08-09 00:06:47 +0300 | [diff] [blame] | 155 | def get_name(self): |
| 156 | return self._name |
| 157 | |
| 158 | def set_name(self, value): |
| 159 | self._name = str(value) |
| 160 | |
Yury Selivanov | 0cf16f9 | 2017-12-25 10:48:15 -0500 | [diff] [blame] | 161 | def set_result(self, result): |
| 162 | raise RuntimeError('Task does not support set_result operation') |
| 163 | |
| 164 | def set_exception(self, exception): |
| 165 | raise RuntimeError('Task does not support set_exception operation') |
| 166 | |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 167 | def get_stack(self, *, limit=None): |
| 168 | """Return the list of stack frames for this task's coroutine. |
| 169 | |
Victor Stinner | d87de83 | 2014-12-02 17:57:04 +0100 | [diff] [blame] | 170 | 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] | 171 | suspended. If the coroutine has completed successfully or was |
| 172 | cancelled, this returns an empty list. If the coroutine was |
| 173 | terminated by an exception, this returns the list of traceback |
| 174 | frames. |
| 175 | |
| 176 | The frames are always ordered from oldest to newest. |
| 177 | |
Yury Selivanov | b0b0e62 | 2014-02-18 22:27:48 -0500 | [diff] [blame] | 178 | The optional limit gives the maximum number of frames to |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 179 | return; by default all available frames are returned. Its |
| 180 | meaning differs depending on whether a stack or a traceback is |
| 181 | returned: the newest frames of a stack are returned, but the |
| 182 | oldest frames of a traceback are returned. (This matches the |
| 183 | behavior of the traceback module.) |
| 184 | |
| 185 | For reasons beyond our control, only one stack frame is |
| 186 | returned for a suspended coroutine. |
| 187 | """ |
Yury Selivanov | a0c1ba6 | 2016-10-28 12:52:37 -0400 | [diff] [blame] | 188 | return base_tasks._task_get_stack(self, limit) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 189 | |
| 190 | def print_stack(self, *, limit=None, file=None): |
| 191 | """Print the stack or traceback for this task's coroutine. |
| 192 | |
| 193 | This produces output similar to that of the traceback module, |
| 194 | for the frames retrieved by get_stack(). The limit argument |
| 195 | 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] | 196 | to which the output is written; by default output is written |
| 197 | to sys.stderr. |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 198 | """ |
Yury Selivanov | a0c1ba6 | 2016-10-28 12:52:37 -0400 | [diff] [blame] | 199 | return base_tasks._task_print_stack(self, limit, file) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 200 | |
| 201 | def cancel(self): |
R David Murray | 8e069d5 | 2014-09-24 13:13:45 -0400 | [diff] [blame] | 202 | """Request that this task cancel itself. |
Victor Stinner | 4bd652a | 2014-04-07 11:18:06 +0200 | [diff] [blame] | 203 | |
Victor Stinner | 8d21357 | 2014-06-02 23:06:46 +0200 | [diff] [blame] | 204 | This arranges for a CancelledError to be thrown into the |
Victor Stinner | 4bd652a | 2014-04-07 11:18:06 +0200 | [diff] [blame] | 205 | wrapped coroutine on the next cycle through the event loop. |
| 206 | The coroutine then has a chance to clean up or even deny |
| 207 | the request using try/except/finally. |
| 208 | |
R David Murray | 8e069d5 | 2014-09-24 13:13:45 -0400 | [diff] [blame] | 209 | Unlike Future.cancel, this does not guarantee that the |
Victor Stinner | 4bd652a | 2014-04-07 11:18:06 +0200 | [diff] [blame] | 210 | task will be cancelled: the exception might be caught and |
R David Murray | 8e069d5 | 2014-09-24 13:13:45 -0400 | [diff] [blame] | 211 | acted upon, delaying cancellation of the task or preventing |
| 212 | cancellation completely. The task may also return a value or |
| 213 | raise a different exception. |
Victor Stinner | 4bd652a | 2014-04-07 11:18:06 +0200 | [diff] [blame] | 214 | |
| 215 | Immediately after this method is called, Task.cancelled() will |
| 216 | not return True (unless the task was already cancelled). A |
| 217 | task will be marked as cancelled when the wrapped coroutine |
| 218 | terminates with a CancelledError exception (even if cancel() |
| 219 | was not called). |
| 220 | """ |
Yury Selivanov | 7ce1c6f | 2017-06-11 13:49:18 +0000 | [diff] [blame] | 221 | self._log_traceback = False |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 222 | if self.done(): |
| 223 | return False |
| 224 | if self._fut_waiter is not None: |
| 225 | if self._fut_waiter.cancel(): |
| 226 | # Leave self._fut_waiter; it may be a Task that |
| 227 | # catches and ignores the cancellation so we may have |
| 228 | # to cancel it again later. |
| 229 | return True |
Yury Selivanov | 22feeb8 | 2018-01-24 11:31:01 -0500 | [diff] [blame] | 230 | # It must be the case that self.__step is already scheduled. |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 231 | self._must_cancel = True |
| 232 | return True |
| 233 | |
Yury Selivanov | 22feeb8 | 2018-01-24 11:31:01 -0500 | [diff] [blame] | 234 | def __step(self, exc=None): |
Yury Selivanov | 0cf16f9 | 2017-12-25 10:48:15 -0500 | [diff] [blame] | 235 | if self.done(): |
Andrew Svetlov | 0baa72f | 2018-09-11 10:13:04 -0700 | [diff] [blame] | 236 | raise exceptions.InvalidStateError( |
Yury Selivanov | 0cf16f9 | 2017-12-25 10:48:15 -0500 | [diff] [blame] | 237 | f'_step(): already done: {self!r}, {exc!r}') |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 238 | if self._must_cancel: |
Andrew Svetlov | 0baa72f | 2018-09-11 10:13:04 -0700 | [diff] [blame] | 239 | if not isinstance(exc, exceptions.CancelledError): |
| 240 | exc = exceptions.CancelledError() |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 241 | self._must_cancel = False |
| 242 | coro = self._coro |
| 243 | self._fut_waiter = None |
Guido van Rossum | 1a605ed | 2013-12-06 12:57:40 -0800 | [diff] [blame] | 244 | |
Andrew Svetlov | 44d1a59 | 2017-12-16 21:58:38 +0200 | [diff] [blame] | 245 | _enter_task(self._loop, self) |
Yury Selivanov | d59bba8 | 2015-11-20 12:41:03 -0500 | [diff] [blame] | 246 | # Call either coro.throw(exc) or coro.send(None). |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 247 | try: |
Yury Selivanov | d59bba8 | 2015-11-20 12:41:03 -0500 | [diff] [blame] | 248 | if exc is None: |
| 249 | # We use the `send` method directly, because coroutines |
| 250 | # don't have `__iter__` and `__next__` methods. |
| 251 | result = coro.send(None) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 252 | else: |
Yury Selivanov | d59bba8 | 2015-11-20 12:41:03 -0500 | [diff] [blame] | 253 | result = coro.throw(exc) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 254 | except StopIteration as exc: |
INADA Naoki | 991adca | 2017-05-11 21:18:38 +0900 | [diff] [blame] | 255 | if self._must_cancel: |
| 256 | # Task is cancelled right before coro stops. |
| 257 | self._must_cancel = False |
Andrew Svetlov | 0baa72f | 2018-09-11 10:13:04 -0700 | [diff] [blame] | 258 | super().set_exception(exceptions.CancelledError()) |
INADA Naoki | 991adca | 2017-05-11 21:18:38 +0900 | [diff] [blame] | 259 | else: |
Yury Selivanov | 0cf16f9 | 2017-12-25 10:48:15 -0500 | [diff] [blame] | 260 | super().set_result(exc.value) |
Andrew Svetlov | 0baa72f | 2018-09-11 10:13:04 -0700 | [diff] [blame] | 261 | except exceptions.CancelledError: |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 262 | super().cancel() # I.e., Future.cancel(self). |
| 263 | except Exception as exc: |
Yury Selivanov | 0cf16f9 | 2017-12-25 10:48:15 -0500 | [diff] [blame] | 264 | super().set_exception(exc) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 265 | except BaseException as exc: |
Yury Selivanov | 0cf16f9 | 2017-12-25 10:48:15 -0500 | [diff] [blame] | 266 | super().set_exception(exc) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 267 | raise |
| 268 | else: |
Guido van Rossum | 1140a03 | 2016-09-09 12:54:54 -0700 | [diff] [blame] | 269 | blocking = getattr(result, '_asyncio_future_blocking', None) |
| 270 | if blocking is not None: |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 271 | # Yielded Future must come from Future.__iter__(). |
Yury Selivanov | ca9b36c | 2017-12-23 15:04:15 -0500 | [diff] [blame] | 272 | if futures._get_loop(result) is not self._loop: |
Yury Selivanov | 6370f34 | 2017-12-10 18:36:12 -0500 | [diff] [blame] | 273 | new_exc = RuntimeError( |
| 274 | f'Task {self!r} got Future ' |
| 275 | f'{result!r} attached to a different loop') |
Yury Selivanov | f23746a | 2018-01-22 19:11:18 -0500 | [diff] [blame] | 276 | self._loop.call_soon( |
Yury Selivanov | 22feeb8 | 2018-01-24 11:31:01 -0500 | [diff] [blame] | 277 | self.__step, new_exc, context=self._context) |
Guido van Rossum | 1140a03 | 2016-09-09 12:54:54 -0700 | [diff] [blame] | 278 | elif blocking: |
Yury Selivanov | 4145c83 | 2016-10-09 12:19:12 -0400 | [diff] [blame] | 279 | if result is self: |
Yury Selivanov | 6370f34 | 2017-12-10 18:36:12 -0500 | [diff] [blame] | 280 | new_exc = RuntimeError( |
| 281 | f'Task cannot await on itself: {self!r}') |
Yury Selivanov | f23746a | 2018-01-22 19:11:18 -0500 | [diff] [blame] | 282 | self._loop.call_soon( |
Yury Selivanov | 22feeb8 | 2018-01-24 11:31:01 -0500 | [diff] [blame] | 283 | self.__step, new_exc, context=self._context) |
Yury Selivanov | 4145c83 | 2016-10-09 12:19:12 -0400 | [diff] [blame] | 284 | else: |
| 285 | result._asyncio_future_blocking = False |
Yury Selivanov | f23746a | 2018-01-22 19:11:18 -0500 | [diff] [blame] | 286 | result.add_done_callback( |
Yury Selivanov | 22feeb8 | 2018-01-24 11:31:01 -0500 | [diff] [blame] | 287 | self.__wakeup, context=self._context) |
Yury Selivanov | 4145c83 | 2016-10-09 12:19:12 -0400 | [diff] [blame] | 288 | self._fut_waiter = result |
| 289 | if self._must_cancel: |
| 290 | if self._fut_waiter.cancel(): |
| 291 | self._must_cancel = False |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 292 | else: |
Yury Selivanov | 6370f34 | 2017-12-10 18:36:12 -0500 | [diff] [blame] | 293 | new_exc = RuntimeError( |
| 294 | f'yield was used instead of yield from ' |
| 295 | f'in task {self!r} with {result!r}') |
Yury Selivanov | f23746a | 2018-01-22 19:11:18 -0500 | [diff] [blame] | 296 | self._loop.call_soon( |
Yury Selivanov | 22feeb8 | 2018-01-24 11:31:01 -0500 | [diff] [blame] | 297 | self.__step, new_exc, context=self._context) |
Yury Selivanov | 6370f34 | 2017-12-10 18:36:12 -0500 | [diff] [blame] | 298 | |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 299 | elif result is None: |
| 300 | # Bare yield relinquishes control for one event loop iteration. |
Yury Selivanov | 22feeb8 | 2018-01-24 11:31:01 -0500 | [diff] [blame] | 301 | self._loop.call_soon(self.__step, context=self._context) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 302 | elif inspect.isgenerator(result): |
| 303 | # Yielding a generator is just wrong. |
Yury Selivanov | 6370f34 | 2017-12-10 18:36:12 -0500 | [diff] [blame] | 304 | new_exc = RuntimeError( |
| 305 | f'yield was used instead of yield from for ' |
Serhiy Storchaka | 6655354 | 2018-05-20 16:30:31 +0300 | [diff] [blame] | 306 | f'generator in task {self!r} with {result!r}') |
Yury Selivanov | f23746a | 2018-01-22 19:11:18 -0500 | [diff] [blame] | 307 | self._loop.call_soon( |
Yury Selivanov | 22feeb8 | 2018-01-24 11:31:01 -0500 | [diff] [blame] | 308 | self.__step, new_exc, context=self._context) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 309 | else: |
| 310 | # Yielding something else is an error. |
Yury Selivanov | 6370f34 | 2017-12-10 18:36:12 -0500 | [diff] [blame] | 311 | new_exc = RuntimeError(f'Task got bad yield: {result!r}') |
Yury Selivanov | f23746a | 2018-01-22 19:11:18 -0500 | [diff] [blame] | 312 | self._loop.call_soon( |
Yury Selivanov | 22feeb8 | 2018-01-24 11:31:01 -0500 | [diff] [blame] | 313 | self.__step, new_exc, context=self._context) |
Guido van Rossum | 1a605ed | 2013-12-06 12:57:40 -0800 | [diff] [blame] | 314 | finally: |
Andrew Svetlov | 44d1a59 | 2017-12-16 21:58:38 +0200 | [diff] [blame] | 315 | _leave_task(self._loop, self) |
Victor Stinner | d74ac82 | 2014-03-04 23:07:08 +0100 | [diff] [blame] | 316 | self = None # Needed to break cycles when an exception occurs. |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 317 | |
Yury Selivanov | 22feeb8 | 2018-01-24 11:31:01 -0500 | [diff] [blame] | 318 | def __wakeup(self, future): |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 319 | try: |
Yury Selivanov | a4afc48 | 2015-11-16 15:12:10 -0500 | [diff] [blame] | 320 | future.result() |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 321 | except Exception as exc: |
| 322 | # This may also be a cancellation. |
Yury Selivanov | 22feeb8 | 2018-01-24 11:31:01 -0500 | [diff] [blame] | 323 | self.__step(exc) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 324 | else: |
Yury Selivanov | a4afc48 | 2015-11-16 15:12:10 -0500 | [diff] [blame] | 325 | # Don't pass the value of `future.result()` explicitly, |
| 326 | # as `Future.__iter__` and `Future.__await__` don't need it. |
| 327 | # If we call `_step(value, None)` instead of `_step()`, |
| 328 | # Python eval loop would use `.send(value)` method call, |
| 329 | # instead of `__next__()`, which is slower for futures |
| 330 | # that return non-generator iterators from their `__iter__`. |
Yury Selivanov | 22feeb8 | 2018-01-24 11:31:01 -0500 | [diff] [blame] | 331 | self.__step() |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 332 | self = None # Needed to break cycles when an exception occurs. |
| 333 | |
| 334 | |
Yury Selivanov | a0c1ba6 | 2016-10-28 12:52:37 -0400 | [diff] [blame] | 335 | _PyTask = Task |
| 336 | |
| 337 | |
| 338 | try: |
| 339 | import _asyncio |
| 340 | except ImportError: |
| 341 | pass |
| 342 | else: |
| 343 | # _CTask is needed for tests. |
| 344 | Task = _CTask = _asyncio.Task |
| 345 | |
| 346 | |
Alex Grönholm | cca4eec | 2018-08-09 00:06:47 +0300 | [diff] [blame] | 347 | def create_task(coro, *, name=None): |
Andrew Svetlov | f74ef45 | 2017-12-15 07:04:38 +0200 | [diff] [blame] | 348 | """Schedule the execution of a coroutine object in a spawn task. |
| 349 | |
| 350 | Return a Task object. |
| 351 | """ |
| 352 | loop = events.get_running_loop() |
Alex Grönholm | cca4eec | 2018-08-09 00:06:47 +0300 | [diff] [blame] | 353 | task = loop.create_task(coro) |
| 354 | _set_task_name(task, name) |
| 355 | return task |
Andrew Svetlov | f74ef45 | 2017-12-15 07:04:38 +0200 | [diff] [blame] | 356 | |
| 357 | |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 358 | # wait() and as_completed() similar to those in PEP 3148. |
| 359 | |
| 360 | FIRST_COMPLETED = concurrent.futures.FIRST_COMPLETED |
| 361 | FIRST_EXCEPTION = concurrent.futures.FIRST_EXCEPTION |
| 362 | ALL_COMPLETED = concurrent.futures.ALL_COMPLETED |
| 363 | |
| 364 | |
Andrew Svetlov | 5f841b5 | 2017-12-09 00:23:48 +0200 | [diff] [blame] | 365 | async def wait(fs, *, loop=None, timeout=None, return_when=ALL_COMPLETED): |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 366 | """Wait for the Futures and coroutines given by fs to complete. |
| 367 | |
Victor Stinner | db74d98 | 2014-06-10 11:16:05 +0200 | [diff] [blame] | 368 | The sequence futures must not be empty. |
| 369 | |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 370 | Coroutines will be wrapped in Tasks. |
| 371 | |
| 372 | Returns two sets of Future: (done, pending). |
| 373 | |
| 374 | Usage: |
| 375 | |
Andrew Svetlov | 5f841b5 | 2017-12-09 00:23:48 +0200 | [diff] [blame] | 376 | done, pending = await asyncio.wait(fs) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 377 | |
| 378 | Note: This does not raise TimeoutError! Futures that aren't done |
| 379 | when the timeout occurs are returned in the second set. |
| 380 | """ |
Guido van Rossum | 7b3b3dc | 2016-09-09 14:26:31 -0700 | [diff] [blame] | 381 | if futures.isfuture(fs) or coroutines.iscoroutine(fs): |
Yury Selivanov | 6370f34 | 2017-12-10 18:36:12 -0500 | [diff] [blame] | 382 | raise TypeError(f"expect a list of futures, not {type(fs).__name__}") |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 383 | if not fs: |
| 384 | raise ValueError('Set of coroutines/Futures is empty.') |
Victor Stinner | e931f7b | 2014-07-16 18:50:39 +0200 | [diff] [blame] | 385 | if return_when not in (FIRST_COMPLETED, FIRST_EXCEPTION, ALL_COMPLETED): |
Yury Selivanov | 6370f34 | 2017-12-10 18:36:12 -0500 | [diff] [blame] | 386 | raise ValueError(f'Invalid return_when value: {return_when}') |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 387 | |
| 388 | if loop is None: |
João Júnior | 558c49b | 2018-09-24 06:51:22 -0300 | [diff] [blame] | 389 | loop = events.get_running_loop() |
| 390 | else: |
Yury Selivanov | fad6af2 | 2018-09-25 17:44:52 -0400 | [diff] [blame] | 391 | warnings.warn("The loop argument is deprecated and scheduled for " |
| 392 | "removal in Python 3.10.", |
João Júnior | 558c49b | 2018-09-24 06:51:22 -0300 | [diff] [blame] | 393 | DeprecationWarning, stacklevel=2) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 394 | |
Yury Selivanov | 59eb9a4 | 2015-05-11 14:48:38 -0400 | [diff] [blame] | 395 | fs = {ensure_future(f, loop=loop) for f in set(fs)} |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 396 | |
Andrew Svetlov | 5f841b5 | 2017-12-09 00:23:48 +0200 | [diff] [blame] | 397 | return await _wait(fs, timeout, return_when, loop) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 398 | |
| 399 | |
Victor Stinner | 59e0802 | 2014-08-28 11:19:25 +0200 | [diff] [blame] | 400 | def _release_waiter(waiter, *args): |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 401 | if not waiter.done(): |
Victor Stinner | 59e0802 | 2014-08-28 11:19:25 +0200 | [diff] [blame] | 402 | waiter.set_result(None) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 403 | |
| 404 | |
Andrew Svetlov | 5f841b5 | 2017-12-09 00:23:48 +0200 | [diff] [blame] | 405 | async def wait_for(fut, timeout, *, loop=None): |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 406 | """Wait for the single Future or coroutine to complete, with timeout. |
| 407 | |
| 408 | Coroutine will be wrapped in Task. |
| 409 | |
Victor Stinner | 421e49b | 2014-01-23 17:40:59 +0100 | [diff] [blame] | 410 | Returns result of the Future or coroutine. When a timeout occurs, |
| 411 | it cancels the task and raises TimeoutError. To avoid the task |
| 412 | cancellation, wrap it in shield(). |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 413 | |
Victor Stinner | 922bc2c | 2015-01-15 16:29:10 +0100 | [diff] [blame] | 414 | If the wait is cancelled, the task is also cancelled. |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 415 | |
Victor Stinner | 922bc2c | 2015-01-15 16:29:10 +0100 | [diff] [blame] | 416 | This function is a coroutine. |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 417 | """ |
| 418 | if loop is None: |
João Júnior | 558c49b | 2018-09-24 06:51:22 -0300 | [diff] [blame] | 419 | loop = events.get_running_loop() |
| 420 | else: |
Yury Selivanov | fad6af2 | 2018-09-25 17:44:52 -0400 | [diff] [blame] | 421 | warnings.warn("The loop argument is deprecated and scheduled for " |
| 422 | "removal in Python 3.10.", |
João Júnior | 558c49b | 2018-09-24 06:51:22 -0300 | [diff] [blame] | 423 | DeprecationWarning, stacklevel=2) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 424 | |
Guido van Rossum | 48c66c3 | 2014-01-29 14:30:38 -0800 | [diff] [blame] | 425 | if timeout is None: |
Andrew Svetlov | 5f841b5 | 2017-12-09 00:23:48 +0200 | [diff] [blame] | 426 | return await fut |
Guido van Rossum | 48c66c3 | 2014-01-29 14:30:38 -0800 | [diff] [blame] | 427 | |
Victor K | 4d07189 | 2017-10-05 19:04:39 +0300 | [diff] [blame] | 428 | if timeout <= 0: |
| 429 | fut = ensure_future(fut, loop=loop) |
| 430 | |
| 431 | if fut.done(): |
| 432 | return fut.result() |
| 433 | |
| 434 | fut.cancel() |
Andrew Svetlov | 0baa72f | 2018-09-11 10:13:04 -0700 | [diff] [blame] | 435 | raise exceptions.TimeoutError() |
Victor K | 4d07189 | 2017-10-05 19:04:39 +0300 | [diff] [blame] | 436 | |
Yury Selivanov | 7661db6 | 2016-05-16 15:38:39 -0400 | [diff] [blame] | 437 | waiter = loop.create_future() |
Victor Stinner | 59e0802 | 2014-08-28 11:19:25 +0200 | [diff] [blame] | 438 | timeout_handle = loop.call_later(timeout, _release_waiter, waiter) |
| 439 | cb = functools.partial(_release_waiter, waiter) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 440 | |
Yury Selivanov | 59eb9a4 | 2015-05-11 14:48:38 -0400 | [diff] [blame] | 441 | fut = ensure_future(fut, loop=loop) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 442 | fut.add_done_callback(cb) |
| 443 | |
| 444 | try: |
Victor Stinner | 59e0802 | 2014-08-28 11:19:25 +0200 | [diff] [blame] | 445 | # wait until the future completes or the timeout |
Victor Stinner | 922bc2c | 2015-01-15 16:29:10 +0100 | [diff] [blame] | 446 | try: |
Andrew Svetlov | 5f841b5 | 2017-12-09 00:23:48 +0200 | [diff] [blame] | 447 | await waiter |
Andrew Svetlov | 0baa72f | 2018-09-11 10:13:04 -0700 | [diff] [blame] | 448 | except exceptions.CancelledError: |
Victor Stinner | 922bc2c | 2015-01-15 16:29:10 +0100 | [diff] [blame] | 449 | fut.remove_done_callback(cb) |
| 450 | fut.cancel() |
| 451 | raise |
Victor Stinner | 59e0802 | 2014-08-28 11:19:25 +0200 | [diff] [blame] | 452 | |
| 453 | if fut.done(): |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 454 | return fut.result() |
| 455 | else: |
| 456 | fut.remove_done_callback(cb) |
Elvis Pranskevichus | e2b340a | 2018-05-29 17:31:01 -0400 | [diff] [blame] | 457 | # We must ensure that the task is not running |
| 458 | # after wait_for() returns. |
| 459 | # See https://bugs.python.org/issue32751 |
| 460 | await _cancel_and_wait(fut, loop=loop) |
Andrew Svetlov | 0baa72f | 2018-09-11 10:13:04 -0700 | [diff] [blame] | 461 | raise exceptions.TimeoutError() |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 462 | finally: |
| 463 | timeout_handle.cancel() |
| 464 | |
| 465 | |
Andrew Svetlov | 5f841b5 | 2017-12-09 00:23:48 +0200 | [diff] [blame] | 466 | async def _wait(fs, timeout, return_when, loop): |
Elvis Pranskevichus | e2b340a | 2018-05-29 17:31:01 -0400 | [diff] [blame] | 467 | """Internal helper for wait(). |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 468 | |
| 469 | The fs argument must be a collection of Futures. |
| 470 | """ |
| 471 | assert fs, 'Set of Futures is empty.' |
Yury Selivanov | 7661db6 | 2016-05-16 15:38:39 -0400 | [diff] [blame] | 472 | waiter = loop.create_future() |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 473 | timeout_handle = None |
| 474 | if timeout is not None: |
| 475 | timeout_handle = loop.call_later(timeout, _release_waiter, waiter) |
| 476 | counter = len(fs) |
| 477 | |
| 478 | def _on_completion(f): |
| 479 | nonlocal counter |
| 480 | counter -= 1 |
| 481 | if (counter <= 0 or |
| 482 | return_when == FIRST_COMPLETED or |
| 483 | return_when == FIRST_EXCEPTION and (not f.cancelled() and |
| 484 | f.exception() is not None)): |
| 485 | if timeout_handle is not None: |
| 486 | timeout_handle.cancel() |
| 487 | if not waiter.done(): |
Victor Stinner | 59e0802 | 2014-08-28 11:19:25 +0200 | [diff] [blame] | 488 | waiter.set_result(None) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 489 | |
| 490 | for f in fs: |
| 491 | f.add_done_callback(_on_completion) |
| 492 | |
| 493 | try: |
Andrew Svetlov | 5f841b5 | 2017-12-09 00:23:48 +0200 | [diff] [blame] | 494 | await waiter |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 495 | finally: |
| 496 | if timeout_handle is not None: |
| 497 | timeout_handle.cancel() |
| 498 | |
| 499 | done, pending = set(), set() |
| 500 | for f in fs: |
| 501 | f.remove_done_callback(_on_completion) |
| 502 | if f.done(): |
| 503 | done.add(f) |
| 504 | else: |
| 505 | pending.add(f) |
| 506 | return done, pending |
| 507 | |
| 508 | |
Elvis Pranskevichus | e2b340a | 2018-05-29 17:31:01 -0400 | [diff] [blame] | 509 | async def _cancel_and_wait(fut, loop): |
| 510 | """Cancel the *fut* future or task and wait until it completes.""" |
| 511 | |
| 512 | waiter = loop.create_future() |
| 513 | cb = functools.partial(_release_waiter, waiter) |
| 514 | fut.add_done_callback(cb) |
| 515 | |
| 516 | try: |
| 517 | fut.cancel() |
| 518 | # We cannot wait on *fut* directly to make |
| 519 | # sure _cancel_and_wait itself is reliably cancellable. |
| 520 | await waiter |
| 521 | finally: |
| 522 | fut.remove_done_callback(cb) |
| 523 | |
| 524 | |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 525 | # This is *not* a @coroutine! It is just an iterator (yielding Futures). |
| 526 | def as_completed(fs, *, loop=None, timeout=None): |
Guido van Rossum | b58f053 | 2014-02-12 17:58:19 -0800 | [diff] [blame] | 527 | """Return an iterator whose values are coroutines. |
| 528 | |
| 529 | When waiting for the yielded coroutines you'll get the results (or |
| 530 | exceptions!) of the original Futures (or coroutines), in the order |
| 531 | in which and as soon as they complete. |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 532 | |
| 533 | This differs from PEP 3148; the proper way to use this is: |
| 534 | |
| 535 | for f in as_completed(fs): |
Andrew Svetlov | 5f841b5 | 2017-12-09 00:23:48 +0200 | [diff] [blame] | 536 | result = await f # The 'await' may raise. |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 537 | # Use result. |
| 538 | |
Andrew Svetlov | 5f841b5 | 2017-12-09 00:23:48 +0200 | [diff] [blame] | 539 | If a timeout is specified, the 'await' will raise |
Guido van Rossum | b58f053 | 2014-02-12 17:58:19 -0800 | [diff] [blame] | 540 | TimeoutError when the timeout occurs before all Futures are done. |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 541 | |
| 542 | Note: The futures 'f' are not necessarily members of fs. |
| 543 | """ |
Guido van Rossum | 7b3b3dc | 2016-09-09 14:26:31 -0700 | [diff] [blame] | 544 | if futures.isfuture(fs) or coroutines.iscoroutine(fs): |
Yury Selivanov | 6370f34 | 2017-12-10 18:36:12 -0500 | [diff] [blame] | 545 | raise TypeError(f"expect a list of futures, not {type(fs).__name__}") |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 546 | loop = loop if loop is not None else events.get_event_loop() |
Yury Selivanov | 59eb9a4 | 2015-05-11 14:48:38 -0400 | [diff] [blame] | 547 | todo = {ensure_future(f, loop=loop) for f in set(fs)} |
Guido van Rossum | b58f053 | 2014-02-12 17:58:19 -0800 | [diff] [blame] | 548 | from .queues import Queue # Import here to avoid circular import problem. |
| 549 | done = Queue(loop=loop) |
| 550 | timeout_handle = None |
| 551 | |
| 552 | def _on_timeout(): |
| 553 | for f in todo: |
| 554 | f.remove_done_callback(_on_completion) |
| 555 | done.put_nowait(None) # Queue a dummy value for _wait_for_one(). |
| 556 | todo.clear() # Can't do todo.remove(f) in the loop. |
| 557 | |
| 558 | def _on_completion(f): |
| 559 | if not todo: |
| 560 | return # _on_timeout() was here first. |
| 561 | todo.remove(f) |
| 562 | done.put_nowait(f) |
| 563 | if not todo and timeout_handle is not None: |
| 564 | timeout_handle.cancel() |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 565 | |
Andrew Svetlov | 5f841b5 | 2017-12-09 00:23:48 +0200 | [diff] [blame] | 566 | async def _wait_for_one(): |
| 567 | f = await done.get() |
Guido van Rossum | b58f053 | 2014-02-12 17:58:19 -0800 | [diff] [blame] | 568 | if f is None: |
| 569 | # Dummy value from _on_timeout(). |
Andrew Svetlov | 0baa72f | 2018-09-11 10:13:04 -0700 | [diff] [blame] | 570 | raise exceptions.TimeoutError |
Guido van Rossum | b58f053 | 2014-02-12 17:58:19 -0800 | [diff] [blame] | 571 | return f.result() # May raise f.exception(). |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 572 | |
Guido van Rossum | b58f053 | 2014-02-12 17:58:19 -0800 | [diff] [blame] | 573 | for f in todo: |
| 574 | f.add_done_callback(_on_completion) |
| 575 | if todo and timeout is not None: |
| 576 | timeout_handle = loop.call_later(timeout, _on_timeout) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 577 | for _ in range(len(todo)): |
| 578 | yield _wait_for_one() |
| 579 | |
| 580 | |
Andrew Svetlov | 5f841b5 | 2017-12-09 00:23:48 +0200 | [diff] [blame] | 581 | @types.coroutine |
| 582 | def __sleep0(): |
| 583 | """Skip one event loop run cycle. |
| 584 | |
| 585 | This is a private helper for 'asyncio.sleep()', used |
| 586 | when the 'delay' is set to 0. It uses a bare 'yield' |
Yury Selivanov | 22feeb8 | 2018-01-24 11:31:01 -0500 | [diff] [blame] | 587 | expression (which Task.__step knows how to handle) |
Andrew Svetlov | 5f841b5 | 2017-12-09 00:23:48 +0200 | [diff] [blame] | 588 | instead of creating a Future object. |
| 589 | """ |
| 590 | yield |
| 591 | |
| 592 | |
| 593 | async def sleep(delay, result=None, *, loop=None): |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 594 | """Coroutine that completes after a given time (in seconds).""" |
Andrew Svetlov | 5382c05 | 2017-12-17 16:41:30 +0200 | [diff] [blame] | 595 | if delay <= 0: |
Andrew Svetlov | 5f841b5 | 2017-12-09 00:23:48 +0200 | [diff] [blame] | 596 | await __sleep0() |
Yury Selivanov | ade0412 | 2015-11-05 14:29:04 -0500 | [diff] [blame] | 597 | return result |
| 598 | |
Yury Selivanov | 7661db6 | 2016-05-16 15:38:39 -0400 | [diff] [blame] | 599 | if loop is None: |
João Júnior | 558c49b | 2018-09-24 06:51:22 -0300 | [diff] [blame] | 600 | loop = events.get_running_loop() |
| 601 | else: |
Yury Selivanov | fad6af2 | 2018-09-25 17:44:52 -0400 | [diff] [blame] | 602 | warnings.warn("The loop argument is deprecated and scheduled for " |
| 603 | "removal in Python 3.10.", |
João Júnior | 558c49b | 2018-09-24 06:51:22 -0300 | [diff] [blame] | 604 | DeprecationWarning, stacklevel=2) |
| 605 | |
Yury Selivanov | 7661db6 | 2016-05-16 15:38:39 -0400 | [diff] [blame] | 606 | future = loop.create_future() |
Yury Selivanov | ca9b36c | 2017-12-23 15:04:15 -0500 | [diff] [blame] | 607 | h = loop.call_later(delay, |
| 608 | futures._set_result_unless_cancelled, |
| 609 | future, result) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 610 | try: |
Andrew Svetlov | 5f841b5 | 2017-12-09 00:23:48 +0200 | [diff] [blame] | 611 | return await future |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 612 | finally: |
| 613 | h.cancel() |
| 614 | |
| 615 | |
Yury Selivanov | 59eb9a4 | 2015-05-11 14:48:38 -0400 | [diff] [blame] | 616 | def ensure_future(coro_or_future, *, loop=None): |
Yury Selivanov | 620279b | 2015-10-02 15:00:19 -0400 | [diff] [blame] | 617 | """Wrap a coroutine or an awaitable in a future. |
Yury Selivanov | 59eb9a4 | 2015-05-11 14:48:38 -0400 | [diff] [blame] | 618 | |
| 619 | If the argument is a Future, it is returned directly. |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 620 | """ |
jimmylai | e549c4b | 2018-05-28 06:42:05 -1000 | [diff] [blame] | 621 | if coroutines.iscoroutine(coro_or_future): |
Victor Stinner | 896a25a | 2014-07-08 11:29:25 +0200 | [diff] [blame] | 622 | if loop is None: |
| 623 | loop = events.get_event_loop() |
| 624 | task = loop.create_task(coro_or_future) |
Victor Stinner | 80f53aa | 2014-06-27 13:52:20 +0200 | [diff] [blame] | 625 | if task._source_traceback: |
| 626 | del task._source_traceback[-1] |
| 627 | return task |
jimmylai | e549c4b | 2018-05-28 06:42:05 -1000 | [diff] [blame] | 628 | elif futures.isfuture(coro_or_future): |
| 629 | if loop is not None and loop is not futures._get_loop(coro_or_future): |
| 630 | raise ValueError('loop argument must agree with Future') |
| 631 | return coro_or_future |
Victor Stinner | 3f438a9 | 2017-11-28 14:43:52 +0100 | [diff] [blame] | 632 | elif inspect.isawaitable(coro_or_future): |
Yury Selivanov | 620279b | 2015-10-02 15:00:19 -0400 | [diff] [blame] | 633 | return ensure_future(_wrap_awaitable(coro_or_future), loop=loop) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 634 | else: |
Charles Renwick | ae5b326 | 2017-04-21 16:49:48 -0400 | [diff] [blame] | 635 | raise TypeError('An asyncio.Future, a coroutine or an awaitable is ' |
| 636 | 'required') |
Yury Selivanov | 620279b | 2015-10-02 15:00:19 -0400 | [diff] [blame] | 637 | |
| 638 | |
| 639 | @coroutine |
| 640 | def _wrap_awaitable(awaitable): |
| 641 | """Helper for asyncio.ensure_future(). |
| 642 | |
| 643 | Wraps awaitable (an object with __await__) into a coroutine |
| 644 | that will later be wrapped in a Task by ensure_future(). |
| 645 | """ |
| 646 | return (yield from awaitable.__await__()) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 647 | |
| 648 | |
| 649 | class _GatheringFuture(futures.Future): |
| 650 | """Helper for gather(). |
| 651 | |
| 652 | This overrides cancel() to cancel all the children and act more |
| 653 | like Task.cancel(), which doesn't immediately mark itself as |
| 654 | cancelled. |
| 655 | """ |
| 656 | |
| 657 | def __init__(self, children, *, loop=None): |
| 658 | super().__init__(loop=loop) |
| 659 | self._children = children |
Yury Selivanov | 863b674 | 2018-05-29 17:20:02 -0400 | [diff] [blame] | 660 | self._cancel_requested = False |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 661 | |
| 662 | def cancel(self): |
| 663 | if self.done(): |
| 664 | return False |
Yury Selivanov | 3d67615 | 2016-10-21 17:22:17 -0400 | [diff] [blame] | 665 | ret = False |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 666 | for child in self._children: |
Yury Selivanov | 3d67615 | 2016-10-21 17:22:17 -0400 | [diff] [blame] | 667 | if child.cancel(): |
| 668 | ret = True |
Yury Selivanov | 863b674 | 2018-05-29 17:20:02 -0400 | [diff] [blame] | 669 | if ret: |
| 670 | # If any child tasks were actually cancelled, we should |
| 671 | # propagate the cancellation request regardless of |
| 672 | # *return_exceptions* argument. See issue 32684. |
| 673 | self._cancel_requested = True |
Yury Selivanov | 3d67615 | 2016-10-21 17:22:17 -0400 | [diff] [blame] | 674 | return ret |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 675 | |
| 676 | |
| 677 | def gather(*coros_or_futures, loop=None, return_exceptions=False): |
Yury Selivanov | 36c2c04 | 2017-12-19 07:19:53 -0500 | [diff] [blame] | 678 | """Return a future aggregating results from the given coroutines/futures. |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 679 | |
Guido van Rossum | e3c65a7 | 2016-09-30 08:17:15 -0700 | [diff] [blame] | 680 | Coroutines will be wrapped in a future and scheduled in the event |
| 681 | loop. They will not necessarily be scheduled in the same order as |
| 682 | passed in. |
| 683 | |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 684 | All futures must share the same event loop. If all the tasks are |
| 685 | done successfully, the returned future's result is the list of |
| 686 | results (in the order of the original sequence, not necessarily |
Yury Selivanov | f317cb7 | 2014-02-06 12:03:53 -0500 | [diff] [blame] | 687 | the order of results arrival). If *return_exceptions* is True, |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 688 | exceptions in the tasks are treated the same as successful |
| 689 | results, and gathered in the result list; otherwise, the first |
| 690 | raised exception will be immediately propagated to the returned |
| 691 | future. |
| 692 | |
| 693 | Cancellation: if the outer Future is cancelled, all children (that |
| 694 | have not completed yet) are also cancelled. If any child is |
| 695 | cancelled, this is treated as if it raised CancelledError -- |
| 696 | the outer Future is *not* cancelled in this case. (This is to |
| 697 | prevent the cancellation of one child to cause other children to |
| 698 | be cancelled.) |
| 699 | """ |
Victor Stinner | f03b3c7 | 2014-07-16 18:36:24 +0200 | [diff] [blame] | 700 | if not coros_or_futures: |
Yury Selivanov | 7661db6 | 2016-05-16 15:38:39 -0400 | [diff] [blame] | 701 | if loop is None: |
| 702 | loop = events.get_event_loop() |
| 703 | outer = loop.create_future() |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 704 | outer.set_result([]) |
| 705 | return outer |
Victor Stinner | f03b3c7 | 2014-07-16 18:36:24 +0200 | [diff] [blame] | 706 | |
Yury Selivanov | 36c2c04 | 2017-12-19 07:19:53 -0500 | [diff] [blame] | 707 | def _done_callback(fut): |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 708 | nonlocal nfinished |
Yury Selivanov | 36c2c04 | 2017-12-19 07:19:53 -0500 | [diff] [blame] | 709 | nfinished += 1 |
| 710 | |
Victor Stinner | 3531d90 | 2015-01-09 01:42:52 +0100 | [diff] [blame] | 711 | if outer.done(): |
| 712 | if not fut.cancelled(): |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 713 | # Mark exception retrieved. |
| 714 | fut.exception() |
| 715 | return |
Victor Stinner | 3531d90 | 2015-01-09 01:42:52 +0100 | [diff] [blame] | 716 | |
Yury Selivanov | 36c2c04 | 2017-12-19 07:19:53 -0500 | [diff] [blame] | 717 | if not return_exceptions: |
| 718 | if fut.cancelled(): |
| 719 | # Check if 'fut' is cancelled first, as |
| 720 | # 'fut.exception()' will *raise* a CancelledError |
| 721 | # instead of returning it. |
Andrew Svetlov | 0baa72f | 2018-09-11 10:13:04 -0700 | [diff] [blame] | 722 | exc = exceptions.CancelledError() |
Yury Selivanov | 36c2c04 | 2017-12-19 07:19:53 -0500 | [diff] [blame] | 723 | outer.set_exception(exc) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 724 | return |
Yury Selivanov | 36c2c04 | 2017-12-19 07:19:53 -0500 | [diff] [blame] | 725 | else: |
| 726 | exc = fut.exception() |
| 727 | if exc is not None: |
| 728 | outer.set_exception(exc) |
| 729 | return |
| 730 | |
| 731 | if nfinished == nfuts: |
| 732 | # All futures are done; create a list of results |
| 733 | # and set it to the 'outer' future. |
| 734 | results = [] |
| 735 | |
| 736 | for fut in children: |
| 737 | if fut.cancelled(): |
| 738 | # Check if 'fut' is cancelled first, as |
| 739 | # 'fut.exception()' will *raise* a CancelledError |
| 740 | # instead of returning it. |
Andrew Svetlov | 0baa72f | 2018-09-11 10:13:04 -0700 | [diff] [blame] | 741 | res = exceptions.CancelledError() |
Yury Selivanov | 36c2c04 | 2017-12-19 07:19:53 -0500 | [diff] [blame] | 742 | else: |
| 743 | res = fut.exception() |
| 744 | if res is None: |
| 745 | res = fut.result() |
| 746 | results.append(res) |
| 747 | |
Yury Selivanov | 863b674 | 2018-05-29 17:20:02 -0400 | [diff] [blame] | 748 | if outer._cancel_requested: |
| 749 | # If gather is being cancelled we must propagate the |
| 750 | # cancellation regardless of *return_exceptions* argument. |
| 751 | # See issue 32684. |
Andrew Svetlov | 0baa72f | 2018-09-11 10:13:04 -0700 | [diff] [blame] | 752 | outer.set_exception(exceptions.CancelledError()) |
Yury Selivanov | 863b674 | 2018-05-29 17:20:02 -0400 | [diff] [blame] | 753 | else: |
| 754 | outer.set_result(results) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 755 | |
Yury Selivanov | 36c2c04 | 2017-12-19 07:19:53 -0500 | [diff] [blame] | 756 | arg_to_fut = {} |
| 757 | children = [] |
| 758 | nfuts = 0 |
| 759 | nfinished = 0 |
| 760 | for arg in coros_or_futures: |
| 761 | if arg not in arg_to_fut: |
| 762 | fut = ensure_future(arg, loop=loop) |
| 763 | if loop is None: |
Yury Selivanov | ca9b36c | 2017-12-23 15:04:15 -0500 | [diff] [blame] | 764 | loop = futures._get_loop(fut) |
Yury Selivanov | 36c2c04 | 2017-12-19 07:19:53 -0500 | [diff] [blame] | 765 | if fut is not arg: |
| 766 | # 'arg' was not a Future, therefore, 'fut' is a new |
| 767 | # Future created specifically for 'arg'. Since the caller |
| 768 | # can't control it, disable the "destroy pending task" |
| 769 | # warning. |
| 770 | fut._log_destroy_pending = False |
| 771 | |
| 772 | nfuts += 1 |
| 773 | arg_to_fut[arg] = fut |
| 774 | fut.add_done_callback(_done_callback) |
| 775 | |
| 776 | else: |
| 777 | # There's a duplicate Future object in coros_or_futures. |
| 778 | fut = arg_to_fut[arg] |
| 779 | |
| 780 | children.append(fut) |
| 781 | |
| 782 | outer = _GatheringFuture(children, loop=loop) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 783 | return outer |
| 784 | |
| 785 | |
| 786 | def shield(arg, *, loop=None): |
| 787 | """Wait for a future, shielding it from cancellation. |
| 788 | |
| 789 | The statement |
| 790 | |
Andrew Svetlov | 5f841b5 | 2017-12-09 00:23:48 +0200 | [diff] [blame] | 791 | res = await shield(something()) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 792 | |
| 793 | is exactly equivalent to the statement |
| 794 | |
Andrew Svetlov | 5f841b5 | 2017-12-09 00:23:48 +0200 | [diff] [blame] | 795 | res = await something() |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 796 | |
| 797 | *except* that if the coroutine containing it is cancelled, the |
| 798 | task running in something() is not cancelled. From the POV of |
| 799 | something(), the cancellation did not happen. But its caller is |
| 800 | still cancelled, so the yield-from expression still raises |
| 801 | CancelledError. Note: If something() is cancelled by other means |
| 802 | this will still cancel shield(). |
| 803 | |
| 804 | If you want to completely ignore cancellation (not recommended) |
| 805 | you can combine shield() with a try/except clause, as follows: |
| 806 | |
| 807 | try: |
Andrew Svetlov | 5f841b5 | 2017-12-09 00:23:48 +0200 | [diff] [blame] | 808 | res = await shield(something()) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 809 | except CancelledError: |
| 810 | res = None |
| 811 | """ |
Yury Selivanov | 59eb9a4 | 2015-05-11 14:48:38 -0400 | [diff] [blame] | 812 | inner = ensure_future(arg, loop=loop) |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 813 | if inner.done(): |
| 814 | # Shortcut. |
| 815 | return inner |
Yury Selivanov | ca9b36c | 2017-12-23 15:04:15 -0500 | [diff] [blame] | 816 | loop = futures._get_loop(inner) |
Yury Selivanov | 7661db6 | 2016-05-16 15:38:39 -0400 | [diff] [blame] | 817 | outer = loop.create_future() |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 818 | |
| 819 | def _done_callback(inner): |
| 820 | if outer.cancelled(): |
Victor Stinner | 3531d90 | 2015-01-09 01:42:52 +0100 | [diff] [blame] | 821 | if not inner.cancelled(): |
| 822 | # Mark inner's result as retrieved. |
| 823 | inner.exception() |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 824 | return |
Victor Stinner | 3531d90 | 2015-01-09 01:42:52 +0100 | [diff] [blame] | 825 | |
Guido van Rossum | 27b7c7e | 2013-10-17 13:40:50 -0700 | [diff] [blame] | 826 | if inner.cancelled(): |
| 827 | outer.cancel() |
| 828 | else: |
| 829 | exc = inner.exception() |
| 830 | if exc is not None: |
| 831 | outer.set_exception(exc) |
| 832 | else: |
| 833 | outer.set_result(inner.result()) |
| 834 | |
| 835 | inner.add_done_callback(_done_callback) |
| 836 | return outer |
Guido van Rossum | 841d9ee | 2015-10-03 08:31:42 -0700 | [diff] [blame] | 837 | |
| 838 | |
| 839 | def run_coroutine_threadsafe(coro, loop): |
| 840 | """Submit a coroutine object to a given event loop. |
| 841 | |
| 842 | Return a concurrent.futures.Future to access the result. |
| 843 | """ |
| 844 | if not coroutines.iscoroutine(coro): |
| 845 | raise TypeError('A coroutine object is required') |
| 846 | future = concurrent.futures.Future() |
| 847 | |
| 848 | def callback(): |
Guido van Rossum | 601953b | 2015-10-05 16:20:00 -0700 | [diff] [blame] | 849 | try: |
| 850 | futures._chain_future(ensure_future(coro, loop=loop), future) |
| 851 | except Exception as exc: |
| 852 | if future.set_running_or_notify_cancel(): |
| 853 | future.set_exception(exc) |
| 854 | raise |
Guido van Rossum | 841d9ee | 2015-10-03 08:31:42 -0700 | [diff] [blame] | 855 | |
| 856 | loop.call_soon_threadsafe(callback) |
| 857 | return future |
Andrew Svetlov | 44d1a59 | 2017-12-16 21:58:38 +0200 | [diff] [blame] | 858 | |
| 859 | |
Yury Selivanov | ca9b36c | 2017-12-23 15:04:15 -0500 | [diff] [blame] | 860 | # WeakSet containing all alive tasks. |
| 861 | _all_tasks = weakref.WeakSet() |
Andrew Svetlov | 44d1a59 | 2017-12-16 21:58:38 +0200 | [diff] [blame] | 862 | |
| 863 | # Dictionary containing tasks that are currently active in |
| 864 | # all running event loops. {EventLoop: Task} |
| 865 | _current_tasks = {} |
| 866 | |
| 867 | |
Yury Selivanov | ca9b36c | 2017-12-23 15:04:15 -0500 | [diff] [blame] | 868 | def _register_task(task): |
| 869 | """Register a new task in asyncio as executed by loop.""" |
| 870 | _all_tasks.add(task) |
Andrew Svetlov | 44d1a59 | 2017-12-16 21:58:38 +0200 | [diff] [blame] | 871 | |
| 872 | |
| 873 | def _enter_task(loop, task): |
| 874 | current_task = _current_tasks.get(loop) |
| 875 | if current_task is not None: |
| 876 | raise RuntimeError(f"Cannot enter into task {task!r} while another " |
| 877 | f"task {current_task!r} is being executed.") |
| 878 | _current_tasks[loop] = task |
| 879 | |
| 880 | |
| 881 | def _leave_task(loop, task): |
| 882 | current_task = _current_tasks.get(loop) |
| 883 | if current_task is not task: |
| 884 | raise RuntimeError(f"Leaving task {task!r} does not match " |
| 885 | f"the current task {current_task!r}.") |
| 886 | del _current_tasks[loop] |
| 887 | |
| 888 | |
Yury Selivanov | ca9b36c | 2017-12-23 15:04:15 -0500 | [diff] [blame] | 889 | def _unregister_task(task): |
| 890 | """Unregister a task.""" |
| 891 | _all_tasks.discard(task) |
Andrew Svetlov | 44d1a59 | 2017-12-16 21:58:38 +0200 | [diff] [blame] | 892 | |
| 893 | |
| 894 | _py_register_task = _register_task |
| 895 | _py_unregister_task = _unregister_task |
| 896 | _py_enter_task = _enter_task |
| 897 | _py_leave_task = _leave_task |
| 898 | |
| 899 | |
| 900 | try: |
| 901 | from _asyncio import (_register_task, _unregister_task, |
| 902 | _enter_task, _leave_task, |
| 903 | _all_tasks, _current_tasks) |
| 904 | except ImportError: |
| 905 | pass |
| 906 | else: |
| 907 | _c_register_task = _register_task |
| 908 | _c_unregister_task = _unregister_task |
| 909 | _c_enter_task = _enter_task |
| 910 | _c_leave_task = _leave_task |