Coverage for Lib/asyncio/tasks.py: 95%
531 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-09 01:31 +0000
1"""Support for tasks, coroutines and the scheduler."""
3__all__ = (
4 'Task', 'create_task',
5 'FIRST_COMPLETED', 'FIRST_EXCEPTION', 'ALL_COMPLETED',
6 'wait', 'wait_for', 'as_completed', 'sleep',
7 'gather', 'shield', 'ensure_future', 'run_coroutine_threadsafe',
8 'current_task', 'all_tasks',
9 'create_eager_task_factory', 'eager_task_factory',
10 '_register_task', '_unregister_task', '_enter_task', '_leave_task',
11)
13import concurrent.futures
14import contextvars
15import functools
16import inspect
17import itertools
18import math
19import types
20import weakref
21from types import GenericAlias
23from . import base_tasks
24from . import coroutines
25from . import events
26from . import exceptions
27from . import futures
28from . import queues
29from . import timeouts
31# Helper to generate new task names
32# This uses itertools.count() instead of a "+= 1" operation because the latter
33# is not thread safe. See bpo-11866 for a longer explanation.
34_task_name_counter = itertools.count(1).__next__
37def current_task(loop=None):
38 """Return a currently executed task."""
39 if loop is None:
40 loop = events.get_running_loop()
41 return _current_tasks.get(loop)
44def all_tasks(loop=None):
45 """Return a set of all tasks for the loop."""
46 if loop is None:
47 loop = events.get_running_loop()
48 # capturing the set of eager tasks first, so if an eager task "graduates"
49 # to a regular task in another thread, we don't risk missing it.
50 eager_tasks = list(_eager_tasks)
52 return {t for t in itertools.chain(_scheduled_tasks, eager_tasks)
53 if futures._get_loop(t) is loop and not t.done()}
56class Task(futures._PyFuture): # Inherit Python Task implementation
57 # from a Python Future implementation.
59 """A coroutine wrapped in a Future."""
61 # An important invariant maintained while a Task not done:
62 # _fut_waiter is either None or a Future. The Future
63 # can be either done() or not done().
64 # The task can be in any of 3 states:
65 #
66 # - 1: _fut_waiter is not None and not _fut_waiter.done():
67 # __step() is *not* scheduled and the Task is waiting for _fut_waiter.
68 # - 2: (_fut_waiter is None or _fut_waiter.done()) and __step() is scheduled:
69 # the Task is waiting for __step() to be executed.
70 # - 3: _fut_waiter is None and __step() is *not* scheduled:
71 # the Task is currently executing (in __step()).
72 #
73 # * In state 1, one of the callbacks of __fut_waiter must be __wakeup().
74 # * The transition from 1 to 2 happens when _fut_waiter becomes done(),
75 # as it schedules __wakeup() to be called (which calls __step() so
76 # we way that __step() is scheduled).
77 # * It transitions from 2 to 3 when __step() is executed, and it clears
78 # _fut_waiter to None.
80 # If False, don't log a message if the task is destroyed while its
81 # status is still pending
82 _log_destroy_pending = True
84 def __init__(self, coro, *, loop=None, name=None, context=None,
85 eager_start=False):
86 super().__init__(loop=loop)
87 if self._source_traceback:
88 del self._source_traceback[-1]
89 if not coroutines.iscoroutine(coro):
90 # raise after Future.__init__(), attrs are required for __del__
91 # prevent logging for pending task in __del__
92 self._log_destroy_pending = False
93 raise TypeError(f"a coroutine was expected, got {coro!r}")
95 if name is None:
96 self._name = f'Task-{_task_name_counter()}'
97 else:
98 self._name = str(name)
100 self._num_cancels_requested = 0
101 self._must_cancel = False
102 self._fut_waiter = None
103 self._coro = coro
104 if context is None:
105 self._context = contextvars.copy_context()
106 else:
107 self._context = context
109 if eager_start and self._loop.is_running():
110 self.__eager_start()
111 else:
112 self._loop.call_soon(self.__step, context=self._context)
113 _py_register_task(self)
115 def __del__(self):
116 if self._state == futures._PENDING and self._log_destroy_pending:
117 context = {
118 'task': self,
119 'message': 'Task was destroyed but it is pending!',
120 }
121 if self._source_traceback:
122 context['source_traceback'] = self._source_traceback
123 self._loop.call_exception_handler(context)
124 super().__del__()
126 __class_getitem__ = classmethod(GenericAlias)
128 def __repr__(self):
129 return base_tasks._task_repr(self)
131 def get_coro(self):
132 return self._coro
134 def get_context(self):
135 return self._context
137 def get_name(self):
138 return self._name
140 def set_name(self, value):
141 self._name = str(value)
143 def set_result(self, result):
144 raise RuntimeError('Task does not support set_result operation')
146 def set_exception(self, exception):
147 raise RuntimeError('Task does not support set_exception operation')
149 def get_stack(self, *, limit=None):
150 """Return the list of stack frames for this task's coroutine.
152 If the coroutine is not done, this returns the stack where it is
153 suspended. If the coroutine has completed successfully or was
154 cancelled, this returns an empty list. If the coroutine was
155 terminated by an exception, this returns the list of traceback
156 frames.
158 The frames are always ordered from oldest to newest.
160 The optional limit gives the maximum number of frames to
161 return; by default all available frames are returned. Its
162 meaning differs depending on whether a stack or a traceback is
163 returned: the newest frames of a stack are returned, but the
164 oldest frames of a traceback are returned. (This matches the
165 behavior of the traceback module.)
167 For reasons beyond our control, only one stack frame is
168 returned for a suspended coroutine.
169 """
170 return base_tasks._task_get_stack(self, limit)
172 def print_stack(self, *, limit=None, file=None):
173 """Print the stack or traceback for this task's coroutine.
175 This produces output similar to that of the traceback module,
176 for the frames retrieved by get_stack(). The limit argument
177 is passed to get_stack(). The file argument is an I/O stream
178 to which the output is written; by default output is written
179 to sys.stderr.
180 """
181 return base_tasks._task_print_stack(self, limit, file)
183 def cancel(self, msg=None):
184 """Request that this task cancel itself.
186 This arranges for a CancelledError to be thrown into the
187 wrapped coroutine on the next cycle through the event loop.
188 The coroutine then has a chance to clean up or even deny
189 the request using try/except/finally.
191 Unlike Future.cancel, this does not guarantee that the
192 task will be cancelled: the exception might be caught and
193 acted upon, delaying cancellation of the task or preventing
194 cancellation completely. The task may also return a value or
195 raise a different exception.
197 Immediately after this method is called, Task.cancelled() will
198 not return True (unless the task was already cancelled). A
199 task will be marked as cancelled when the wrapped coroutine
200 terminates with a CancelledError exception (even if cancel()
201 was not called).
203 This also increases the task's count of cancellation requests.
204 """
205 self._log_traceback = False
206 if self.done():
207 return False
208 self._num_cancels_requested += 1
209 # These two lines are controversial. See discussion starting at
210 # https://github.com/python/cpython/pull/31394#issuecomment-1053545331
211 # Also remember that this is duplicated in _asynciomodule.c.
212 # if self._num_cancels_requested > 1:
213 # return False
214 if self._fut_waiter is not None:
215 if self._fut_waiter.cancel(msg=msg):
216 # Leave self._fut_waiter; it may be a Task that
217 # catches and ignores the cancellation so we may have
218 # to cancel it again later.
219 return True
220 # It must be the case that self.__step is already scheduled.
221 self._must_cancel = True
222 self._cancel_message = msg
223 return True
225 def cancelling(self):
226 """Return the count of the task's cancellation requests.
228 This count is incremented when .cancel() is called
229 and may be decremented using .uncancel().
230 """
231 return self._num_cancels_requested
233 def uncancel(self):
234 """Decrement the task's count of cancellation requests.
236 This should be called by the party that called `cancel()` on the task
237 beforehand.
239 Returns the remaining number of cancellation requests.
240 """
241 if self._num_cancels_requested > 0: 241 ↛ 245line 241 didn't jump to line 245 because the condition on line 241 was always true
242 self._num_cancels_requested -= 1
243 if self._num_cancels_requested == 0: 243 ↛ 245line 243 didn't jump to line 245 because the condition on line 243 was always true
244 self._must_cancel = False
245 return self._num_cancels_requested
247 def __eager_start(self):
248 prev_task = _py_swap_current_task(self._loop, self)
249 try:
250 _py_register_eager_task(self)
251 try:
252 self._context.run(self.__step_run_and_handle_result, None)
253 finally:
254 _py_unregister_eager_task(self)
255 finally:
256 try:
257 curtask = _py_swap_current_task(self._loop, prev_task)
258 assert curtask is self
259 finally:
260 if self.done():
261 self._coro = None
262 self = None # Needed to break cycles when an exception occurs.
263 else:
264 _py_register_task(self)
266 def __step(self, exc=None):
267 if self.done():
268 raise exceptions.InvalidStateError(
269 f'__step(): already done: {self!r}, {exc!r}')
270 if self._must_cancel:
271 if not isinstance(exc, exceptions.CancelledError):
272 exc = self._make_cancelled_error()
273 self._must_cancel = False
274 self._fut_waiter = None
276 _py_enter_task(self._loop, self)
277 try:
278 self.__step_run_and_handle_result(exc)
279 finally:
280 _py_leave_task(self._loop, self)
281 self = None # Needed to break cycles when an exception occurs.
283 def __step_run_and_handle_result(self, exc):
284 coro = self._coro
285 try:
286 if exc is None:
287 # We use the `send` method directly, because coroutines
288 # don't have `__iter__` and `__next__` methods.
289 result = coro.send(None)
290 else:
291 result = coro.throw(exc)
292 except StopIteration as exc:
293 if self._must_cancel:
294 # Task is cancelled right before coro stops.
295 self._must_cancel = False
296 super().cancel(msg=self._cancel_message)
297 else:
298 super().set_result(exc.value)
299 except exceptions.CancelledError as exc:
300 # Save the original exception so we can chain it later.
301 self._cancelled_exc = exc
302 super().cancel() # I.e., Future.cancel(self).
303 except (KeyboardInterrupt, SystemExit) as exc:
304 super().set_exception(exc)
305 raise
306 except BaseException as exc:
307 super().set_exception(exc)
308 else:
309 blocking = getattr(result, '_asyncio_future_blocking', None)
310 if blocking is not None:
311 # Yielded Future must come from Future.__iter__().
312 if futures._get_loop(result) is not self._loop:
313 new_exc = RuntimeError(
314 f'Task {self!r} got Future '
315 f'{result!r} attached to a different loop')
316 self._loop.call_soon(
317 self.__step, new_exc, context=self._context)
318 elif blocking: 318 ↛ 335line 318 didn't jump to line 335 because the condition on line 318 was always true
319 if result is self:
320 new_exc = RuntimeError(
321 f'Task cannot await on itself: {self!r}')
322 self._loop.call_soon(
323 self.__step, new_exc, context=self._context)
324 else:
325 futures.future_add_to_awaited_by(result, self)
326 result._asyncio_future_blocking = False
327 result.add_done_callback(
328 self.__wakeup, context=self._context)
329 self._fut_waiter = result
330 if self._must_cancel:
331 if self._fut_waiter.cancel( 331 ↛ 357line 331 didn't jump to line 357 because the condition on line 331 was always true
332 msg=self._cancel_message):
333 self._must_cancel = False
334 else:
335 new_exc = RuntimeError(
336 f'yield was used instead of yield from '
337 f'in task {self!r} with {result!r}')
338 self._loop.call_soon(
339 self.__step, new_exc, context=self._context)
341 elif result is None: 341 ↛ 344line 341 didn't jump to line 344 because the condition on line 341 was always true
342 # Bare yield relinquishes control for one event loop iteration.
343 self._loop.call_soon(self.__step, context=self._context)
344 elif inspect.isgenerator(result):
345 # Yielding a generator is just wrong.
346 new_exc = RuntimeError(
347 f'yield was used instead of yield from for '
348 f'generator in task {self!r} with {result!r}')
349 self._loop.call_soon(
350 self.__step, new_exc, context=self._context)
351 else:
352 # Yielding something else is an error.
353 new_exc = RuntimeError(f'Task got bad yield: {result!r}')
354 self._loop.call_soon(
355 self.__step, new_exc, context=self._context)
356 finally:
357 self = None # Needed to break cycles when an exception occurs.
359 def __wakeup(self, future):
360 futures.future_discard_from_awaited_by(future, self)
361 try:
362 future.result()
363 except BaseException as exc:
364 # This may also be a cancellation.
365 self.__step(exc)
366 else:
367 # Don't pass the value of `future.result()` explicitly,
368 # as `Future.__iter__` and `Future.__await__` don't need it.
369 # If we call `__step(value, None)` instead of `__step()`,
370 # Python eval loop would use `.send(value)` method call,
371 # instead of `__next__()`, which is slower for futures
372 # that return non-generator iterators from their `__iter__`.
373 self.__step()
374 self = None # Needed to break cycles when an exception occurs.
377_PyTask = Task
380try:
381 import _asyncio
382except ImportError:
383 pass
384else:
385 # _CTask is needed for tests.
386 Task = _CTask = _asyncio.Task
389def create_task(coro, **kwargs):
390 """Schedule the execution of a coroutine object in a spawn task.
392 Return a Task object.
393 """
394 loop = events.get_running_loop()
395 return loop.create_task(coro, **kwargs)
398# wait() and as_completed() similar to those in PEP 3148.
400FIRST_COMPLETED = concurrent.futures.FIRST_COMPLETED
401FIRST_EXCEPTION = concurrent.futures.FIRST_EXCEPTION
402ALL_COMPLETED = concurrent.futures.ALL_COMPLETED
405async def wait(fs, *, timeout=None, return_when=ALL_COMPLETED):
406 """Wait for the Futures or Tasks given by fs to complete.
408 The fs iterable must not be empty.
410 Returns two sets of Future: (done, pending).
412 Usage:
414 done, pending = await asyncio.wait(fs)
416 Note: This does not raise TimeoutError! Futures that aren't done
417 when the timeout occurs are returned in the second set.
418 """
419 if futures.isfuture(fs) or coroutines.iscoroutine(fs):
420 raise TypeError(f"expect a list of futures, not {type(fs).__name__}")
421 if not fs:
422 raise ValueError('Set of Tasks/Futures is empty.')
423 if return_when not in (FIRST_COMPLETED, FIRST_EXCEPTION, ALL_COMPLETED):
424 raise ValueError(f'Invalid return_when value: {return_when}')
426 fs = set(fs)
428 if any(coroutines.iscoroutine(f) for f in fs): 428 ↛ 429line 428 didn't jump to line 429 because the condition on line 428 was never true
429 raise TypeError("Passing coroutines is forbidden, use tasks explicitly.")
431 loop = events.get_running_loop()
432 return await _wait(fs, timeout, return_when, loop)
435def _release_waiter(waiter, *args):
436 if not waiter.done(): 436 ↛ exitline 436 didn't return from function '_release_waiter' because the condition on line 436 was always true
437 waiter.set_result(None)
440async def wait_for(fut, timeout):
441 """Wait for the single Future or coroutine to complete, with timeout.
443 Coroutine will be wrapped in Task.
445 Returns result of the Future or coroutine. When a timeout occurs,
446 it cancels the task and raises TimeoutError. To avoid the task
447 cancellation, wrap it in shield().
449 If the wait is cancelled, the task is also cancelled.
451 If the task suppresses the cancellation and returns a value instead,
452 that value is returned.
454 This function is a coroutine.
455 """
456 # The special case for timeout <= 0 is for the following case:
457 #
458 # async def test_waitfor():
459 # func_started = False
460 #
461 # async def func():
462 # nonlocal func_started
463 # func_started = True
464 #
465 # try:
466 # await asyncio.wait_for(func(), 0)
467 # except asyncio.TimeoutError:
468 # assert not func_started
469 # else:
470 # assert False
471 #
472 # asyncio.run(test_waitfor())
475 if timeout is not None and timeout <= 0:
476 fut = ensure_future(fut)
478 if fut.done():
479 return fut.result()
481 await _cancel_and_wait(fut)
482 try:
483 return fut.result()
484 except exceptions.CancelledError as exc:
485 raise TimeoutError from exc
487 async with timeouts.timeout(timeout):
488 return await fut
490async def _wait(fs, timeout, return_when, loop):
491 """Internal helper for wait().
493 The fs argument must be a collection of Futures.
494 """
495 assert fs, 'Set of Futures is empty.'
496 waiter = loop.create_future()
497 timeout_handle = None
498 if timeout is not None:
499 timeout_handle = loop.call_later(timeout, _release_waiter, waiter)
500 counter = len(fs)
501 cur_task = current_task()
503 def _on_completion(f):
504 nonlocal counter
505 counter -= 1
506 if (counter <= 0 or
507 return_when == FIRST_COMPLETED or
508 return_when == FIRST_EXCEPTION and (not f.cancelled() and
509 f.exception() is not None)):
510 if timeout_handle is not None: 510 ↛ 511line 510 didn't jump to line 511 because the condition on line 510 was never true
511 timeout_handle.cancel()
512 if not waiter.done():
513 waiter.set_result(None)
514 futures.future_discard_from_awaited_by(f, cur_task)
516 for f in fs:
517 f.add_done_callback(_on_completion)
518 futures.future_add_to_awaited_by(f, cur_task)
520 try:
521 await waiter
522 finally:
523 if timeout_handle is not None:
524 timeout_handle.cancel()
525 for f in fs:
526 f.remove_done_callback(_on_completion)
527 futures.future_discard_from_awaited_by(f, cur_task)
529 done, pending = set(), set()
530 for f in fs:
531 if f.done():
532 done.add(f)
533 else:
534 pending.add(f)
535 return done, pending
538async def _cancel_and_wait(fut):
539 """Cancel the *fut* future or task and wait until it completes."""
541 loop = events.get_running_loop()
542 waiter = loop.create_future()
543 cb = functools.partial(_release_waiter, waiter)
544 fut.add_done_callback(cb)
546 try:
547 fut.cancel()
548 # We cannot wait on *fut* directly to make
549 # sure _cancel_and_wait itself is reliably cancellable.
550 await waiter
551 finally:
552 fut.remove_done_callback(cb)
555class _AsCompletedIterator:
556 """Iterator of awaitables representing tasks of asyncio.as_completed.
558 As an asynchronous iterator, iteration yields futures as they finish. As a
559 plain iterator, new coroutines are yielded that will return or raise the
560 result of the next underlying future to complete.
561 """
562 def __init__(self, aws, timeout):
563 self._done = queues.Queue()
564 self._timeout_handle = None
566 loop = events.get_event_loop()
567 todo = {ensure_future(aw, loop=loop) for aw in set(aws)}
568 for f in todo:
569 f.add_done_callback(self._handle_completion)
570 if todo and timeout is not None:
571 self._timeout_handle = (
572 loop.call_later(timeout, self._handle_timeout)
573 )
574 self._todo = todo
575 self._todo_left = len(todo)
577 def __aiter__(self):
578 return self
580 def __iter__(self):
581 return self
583 async def __anext__(self):
584 if not self._todo_left:
585 raise StopAsyncIteration
586 assert self._todo_left > 0
587 self._todo_left -= 1
588 return await self._wait_for_one()
590 def __next__(self):
591 if not self._todo_left:
592 raise StopIteration
593 assert self._todo_left > 0
594 self._todo_left -= 1
595 return self._wait_for_one(resolve=True)
597 def _handle_timeout(self):
598 for f in self._todo:
599 f.remove_done_callback(self._handle_completion)
600 self._done.put_nowait(None) # Sentinel for _wait_for_one().
601 self._todo.clear() # Can't do todo.remove(f) in the loop.
603 def _handle_completion(self, f):
604 if not self._todo: 604 ↛ 605line 604 didn't jump to line 605 because the condition on line 604 was never true
605 return # _handle_timeout() was here first.
606 self._todo.remove(f)
607 self._done.put_nowait(f)
608 if not self._todo and self._timeout_handle is not None:
609 self._timeout_handle.cancel()
611 async def _wait_for_one(self, resolve=False):
612 # Wait for the next future to be done and return it unless resolve is
613 # set, in which case return either the result of the future or raise
614 # an exception.
615 f = await self._done.get()
616 if f is None:
617 # Dummy value from _handle_timeout().
618 raise exceptions.TimeoutError
619 return f.result() if resolve else f
622def as_completed(fs, *, timeout=None):
623 """Create an iterator of awaitables or their results in completion order.
625 Run the supplied awaitables concurrently. The returned object can be
626 iterated to obtain the results of the awaitables as they finish.
628 The object returned can be iterated as an asynchronous iterator or
629 a plain iterator. When asynchronous iteration is used, the
630 originally-supplied awaitables are yielded if they are tasks or
631 futures. This makes it easy to correlate previously-scheduled tasks
632 with their results:
634 ipv4_connect = create_task(open_connection("127.0.0.1", 80))
635 ipv6_connect = create_task(open_connection("::1", 80))
636 tasks = [ipv4_connect, ipv6_connect]
638 async for earliest_connect in as_completed(tasks):
639 # earliest_connect is done. The result can be obtained by
640 # awaiting it or calling earliest_connect.result()
641 reader, writer = await earliest_connect
643 if earliest_connect is ipv6_connect:
644 print("IPv6 connection established.")
645 else:
646 print("IPv4 connection established.")
648 During asynchronous iteration, implicitly-created tasks will be
649 yielded for supplied awaitables that aren't tasks or futures.
651 When used as a plain iterator, each iteration yields a new coroutine
652 that returns the result or raises the exception of the next completed
653 awaitable. This pattern is compatible with Python versions older than
654 3.13:
656 ipv4_connect = create_task(open_connection("127.0.0.1", 80))
657 ipv6_connect = create_task(open_connection("::1", 80))
658 tasks = [ipv4_connect, ipv6_connect]
660 for next_connect in as_completed(tasks):
661 # next_connect is not one of the original task objects. It must
662 # be awaited to obtain the result value or raise the exception
663 # of the awaitable that finishes next.
664 reader, writer = await next_connect
666 A TimeoutError is raised if the timeout occurs before all awaitables
667 are done. This is raised by the async for loop during asynchronous
668 iteration or by the coroutines yielded during plain iteration.
669 """
670 if inspect.isawaitable(fs):
671 raise TypeError(
672 f"expects an iterable of awaitables, not {type(fs).__name__}"
673 )
675 return _AsCompletedIterator(fs, timeout)
678@types.coroutine
679def __sleep0():
680 """Skip one event loop run cycle.
682 This is a private helper for 'asyncio.sleep()', used
683 when the 'delay' is set to 0. It uses a bare 'yield'
684 expression (which Task.__step knows how to handle)
685 instead of creating a Future object.
686 """
687 yield
690async def sleep(delay, result=None):
691 """Coroutine that completes after a given time (in seconds)."""
692 if delay <= 0:
693 await __sleep0()
694 return result
696 if math.isnan(delay):
697 raise ValueError("Invalid delay: NaN (not a number)")
699 loop = events.get_running_loop()
700 future = loop.create_future()
701 h = loop.call_later(delay,
702 futures._set_result_unless_cancelled,
703 future, result)
704 try:
705 return await future
706 finally:
707 h.cancel()
710def ensure_future(coro_or_future, *, loop=None):
711 """Wrap a coroutine or an awaitable in a future.
713 If the argument is a Future, it is returned directly.
714 """
715 if futures.isfuture(coro_or_future):
716 if loop is not None and loop is not futures._get_loop(coro_or_future):
717 raise ValueError('The future belongs to a different loop than '
718 'the one specified as the loop argument')
719 return coro_or_future
720 should_close = True
721 if not coroutines.iscoroutine(coro_or_future):
722 if inspect.isawaitable(coro_or_future):
723 async def _wrap_awaitable(awaitable):
724 return await awaitable
726 coro_or_future = _wrap_awaitable(coro_or_future)
727 should_close = False
728 else:
729 raise TypeError('An asyncio.Future, a coroutine or an awaitable '
730 'is required')
732 if loop is None:
733 loop = events.get_event_loop()
734 try:
735 return loop.create_task(coro_or_future)
736 except RuntimeError:
737 if should_close: 737 ↛ 739line 737 didn't jump to line 739 because the condition on line 737 was always true
738 coro_or_future.close()
739 raise
742class _GatheringFuture(futures.Future):
743 """Helper for gather().
745 This overrides cancel() to cancel all the children and act more
746 like Task.cancel(), which doesn't immediately mark itself as
747 cancelled.
748 """
750 def __init__(self, children, *, loop):
751 assert loop is not None
752 super().__init__(loop=loop)
753 self._children = children
754 self._cancel_requested = False
756 def cancel(self, msg=None):
757 if self.done():
758 return False
759 ret = False
760 for child in self._children:
761 if child.cancel(msg=msg):
762 ret = True
763 if ret:
764 # If any child tasks were actually cancelled, we should
765 # propagate the cancellation request regardless of
766 # *return_exceptions* argument. See issue 32684.
767 self._cancel_requested = True
768 return ret
771def gather(*coros_or_futures, return_exceptions=False):
772 """Return a future aggregating results from the given coroutines/futures.
774 Coroutines will be wrapped in a future and scheduled in the event
775 loop. They will not necessarily be scheduled in the same order as
776 passed in.
778 All futures must share the same event loop. If all the tasks are
779 done successfully, the returned future's result is the list of
780 results (in the order of the original sequence, not necessarily
781 the order of results arrival). If *return_exceptions* is True,
782 exceptions in the tasks are treated the same as successful
783 results, and gathered in the result list; otherwise, the first
784 raised exception will be immediately propagated to the returned
785 future.
787 Cancellation: if the outer Future is cancelled, all children (that
788 have not completed yet) are also cancelled. If any child is
789 cancelled, this is treated as if it raised CancelledError --
790 the outer Future is *not* cancelled in this case. (This is to
791 prevent the cancellation of one child to cause other children to
792 be cancelled.)
794 If *return_exceptions* is False, cancelling gather() after it
795 has been marked done won't cancel any submitted awaitables.
796 For instance, gather can be marked done after propagating an
797 exception to the caller, therefore, calling ``gather.cancel()``
798 after catching an exception (raised by one of the awaitables) from
799 gather won't cancel any other awaitables.
800 """
801 if not coros_or_futures:
802 loop = events.get_event_loop()
803 outer = loop.create_future()
804 outer.set_result([])
805 return outer
807 loop = events._get_running_loop()
808 if loop is not None:
809 cur_task = current_task(loop)
810 else:
811 cur_task = None
813 def _done_callback(fut, cur_task=cur_task):
814 nonlocal nfinished
815 nfinished += 1
817 if cur_task is not None:
818 futures.future_discard_from_awaited_by(fut, cur_task)
820 if outer is None or outer.done():
821 if not fut.cancelled():
822 # Mark exception retrieved.
823 fut.exception()
824 return
826 if not return_exceptions:
827 if fut.cancelled():
828 # Check if 'fut' is cancelled first, as
829 # 'fut.exception()' will *raise* a CancelledError
830 # instead of returning it.
831 exc = fut._make_cancelled_error()
832 outer.set_exception(exc)
833 return
834 else:
835 exc = fut.exception()
836 if exc is not None:
837 outer.set_exception(exc)
838 return
840 if nfinished == nfuts:
841 # All futures are done; create a list of results
842 # and set it to the 'outer' future.
843 results = []
845 for fut in children:
846 if fut.cancelled():
847 # Check if 'fut' is cancelled first, as 'fut.exception()'
848 # will *raise* a CancelledError instead of returning it.
849 # Also, since we're adding the exception return value
850 # to 'results' instead of raising it, don't bother
851 # setting __context__. This also lets us preserve
852 # calling '_make_cancelled_error()' at most once.
853 res = exceptions.CancelledError(
854 '' if fut._cancel_message is None else
855 fut._cancel_message)
856 else:
857 res = fut.exception()
858 if res is None:
859 res = fut.result()
860 results.append(res)
862 if outer._cancel_requested:
863 # If gather is being cancelled we must propagate the
864 # cancellation regardless of *return_exceptions* argument.
865 # See issue 32684.
866 exc = fut._make_cancelled_error()
867 outer.set_exception(exc)
868 else:
869 outer.set_result(results)
871 arg_to_fut = {}
872 children = []
873 nfuts = 0
874 nfinished = 0
875 done_futs = []
876 outer = None # bpo-46672
877 for arg in coros_or_futures:
878 if arg not in arg_to_fut:
879 fut = ensure_future(arg, loop=loop)
880 if loop is None:
881 loop = futures._get_loop(fut)
882 if fut is not arg:
883 # 'arg' was not a Future, therefore, 'fut' is a new
884 # Future created specifically for 'arg'. Since the caller
885 # can't control it, disable the "destroy pending task"
886 # warning.
887 fut._log_destroy_pending = False
888 nfuts += 1
889 arg_to_fut[arg] = fut
890 if fut.done():
891 done_futs.append(fut)
892 else:
893 if cur_task is not None:
894 futures.future_add_to_awaited_by(fut, cur_task)
895 fut.add_done_callback(_done_callback)
897 else:
898 # There's a duplicate Future object in coros_or_futures.
899 fut = arg_to_fut[arg]
901 children.append(fut)
903 outer = _GatheringFuture(children, loop=loop)
904 # Run done callbacks after GatheringFuture created so any post-processing
905 # can be performed at this point
906 # optimization: in the special case that *all* futures finished eagerly,
907 # this will effectively complete the gather eagerly, with the last
908 # callback setting the result (or exception) on outer before returning it
909 for fut in done_futs:
910 _done_callback(fut)
911 return outer
914def _log_on_exception(fut):
915 if fut.cancelled(): 915 ↛ 916line 915 didn't jump to line 916 because the condition on line 915 was never true
916 return
918 exc = fut.exception()
919 if exc is None:
920 return
922 context = {
923 'message':
924 f'{exc.__class__.__name__} exception in shielded future',
925 'exception': exc,
926 'future': fut,
927 }
928 if fut._source_traceback: 928 ↛ 929line 928 didn't jump to line 929 because the condition on line 928 was never true
929 context['source_traceback'] = fut._source_traceback
930 fut._loop.call_exception_handler(context)
933def shield(arg):
934 """Wait for a future, shielding it from cancellation.
936 The statement
938 task = asyncio.create_task(something())
939 res = await shield(task)
941 is exactly equivalent to the statement
943 res = await something()
945 *except* that if the coroutine containing it is cancelled, the
946 task running in something() is not cancelled. From the POV of
947 something(), the cancellation did not happen. But its caller is
948 still cancelled, so the yield-from expression still raises
949 CancelledError. Note: If something() is cancelled by other means
950 this will still cancel shield().
952 If you want to completely ignore cancellation (not recommended)
953 you can combine shield() with a try/except clause, as follows:
955 task = asyncio.create_task(something())
956 try:
957 res = await shield(task)
958 except CancelledError:
959 res = None
961 Save a reference to tasks passed to this function, to avoid
962 a task disappearing mid-execution. The event loop only keeps
963 weak references to tasks. A task that isn't referenced elsewhere
964 may get garbage collected at any time, even before it's done.
965 """
966 inner = ensure_future(arg)
967 if inner.done():
968 # Shortcut.
969 return inner
970 loop = futures._get_loop(inner)
971 outer = loop.create_future()
973 if loop is not None and (cur_task := current_task(loop)) is not None:
974 futures.future_add_to_awaited_by(inner, cur_task)
975 else:
976 cur_task = None
978 def _clear_awaited_by_callback(inner):
979 futures.future_discard_from_awaited_by(inner, cur_task)
981 def _inner_done_callback(inner):
982 if outer.cancelled(): 982 ↛ 983line 982 didn't jump to line 983 because the condition on line 982 was never true
983 return
985 if inner.cancelled():
986 outer.cancel()
987 else:
988 exc = inner.exception()
989 if exc is not None:
990 outer.set_exception(exc)
991 else:
992 outer.set_result(inner.result())
994 def _outer_done_callback(outer):
995 if not inner.done():
996 inner.remove_done_callback(_inner_done_callback)
997 # Keep only one callback to log on cancel
998 inner.remove_done_callback(_log_on_exception)
999 inner.add_done_callback(_log_on_exception)
1001 if cur_task is not None:
1002 inner.add_done_callback(_clear_awaited_by_callback)
1005 inner.add_done_callback(_inner_done_callback)
1006 outer.add_done_callback(_outer_done_callback)
1007 return outer
1010def run_coroutine_threadsafe(coro, loop):
1011 """Submit a coroutine object to a given event loop.
1013 Return a concurrent.futures.Future to access the result.
1014 """
1015 if not coroutines.iscoroutine(coro): 1015 ↛ 1016line 1015 didn't jump to line 1016 because the condition on line 1015 was never true
1016 raise TypeError('A coroutine object is required')
1017 future = concurrent.futures.Future()
1019 def callback():
1020 try:
1021 futures._chain_future(ensure_future(coro, loop=loop), future)
1022 except (SystemExit, KeyboardInterrupt):
1023 raise
1024 except BaseException as exc:
1025 if future.set_running_or_notify_cancel(): 1025 ↛ 1027line 1025 didn't jump to line 1027 because the condition on line 1025 was always true
1026 future.set_exception(exc)
1027 raise
1029 loop.call_soon_threadsafe(callback)
1030 return future
1033def create_eager_task_factory(custom_task_constructor):
1034 """Create a function suitable for use as a task factory on an event-loop.
1036 Example usage:
1038 loop.set_task_factory(
1039 asyncio.create_eager_task_factory(my_task_constructor))
1041 Now, tasks created will be started immediately (rather than being first
1042 scheduled to an event loop). The constructor argument can be any
1043 callable that returns a Task-compatible object and has a signature
1044 compatible with `Task.__init__`; it must have the `eager_start`
1045 keyword argument.
1047 Most applications will use `Task` for `custom_task_constructor` and in
1048 this case there's no need to call `create_eager_task_factory()`
1049 directly. Instead the global `eager_task_factory` instance can be
1050 used. E.g. `loop.set_task_factory(asyncio.eager_task_factory)`.
1051 """
1053 def factory(loop, coro, *, eager_start=True, **kwargs):
1054 return custom_task_constructor(
1055 coro, loop=loop, eager_start=eager_start, **kwargs)
1057 return factory
1060eager_task_factory = create_eager_task_factory(Task)
1063# Collectively these two sets hold references to the complete set of active
1064# tasks. Eagerly executed tasks use a faster regular set as an optimization
1065# but may graduate to a WeakSet if the task blocks on IO.
1066_scheduled_tasks = weakref.WeakSet()
1067_eager_tasks = set()
1069# Dictionary containing tasks that are currently active in
1070# all running event loops. {EventLoop: Task}
1071_current_tasks = {}
1074def _register_task(task):
1075 """Register an asyncio Task scheduled to run on an event loop."""
1076 _scheduled_tasks.add(task)
1079def _register_eager_task(task):
1080 """Register an asyncio Task about to be eagerly executed."""
1081 _eager_tasks.add(task)
1084def _enter_task(loop, task):
1085 current_task = _current_tasks.get(loop)
1086 if current_task is not None:
1087 raise RuntimeError(f"Cannot enter into task {task!r} while another "
1088 f"task {current_task!r} is being executed.")
1089 _current_tasks[loop] = task
1092def _leave_task(loop, task):
1093 current_task = _current_tasks.get(loop)
1094 if current_task is not task:
1095 raise RuntimeError(f"Leaving task {task!r} does not match "
1096 f"the current task {current_task!r}.")
1097 del _current_tasks[loop]
1100def _swap_current_task(loop, task):
1101 prev_task = _current_tasks.get(loop)
1102 if task is None:
1103 del _current_tasks[loop]
1104 else:
1105 _current_tasks[loop] = task
1106 return prev_task
1109def _unregister_task(task):
1110 """Unregister a completed, scheduled Task."""
1111 _scheduled_tasks.discard(task)
1114def _unregister_eager_task(task):
1115 """Unregister a task which finished its first eager step."""
1116 _eager_tasks.discard(task)
1119_py_current_task = current_task
1120_py_register_task = _register_task
1121_py_register_eager_task = _register_eager_task
1122_py_unregister_task = _unregister_task
1123_py_unregister_eager_task = _unregister_eager_task
1124_py_enter_task = _enter_task
1125_py_leave_task = _leave_task
1126_py_swap_current_task = _swap_current_task
1127_py_all_tasks = all_tasks
1129try:
1130 from _asyncio import (_register_task, _register_eager_task,
1131 _unregister_task, _unregister_eager_task,
1132 _enter_task, _leave_task, _swap_current_task,
1133 current_task, all_tasks)
1134except ImportError:
1135 pass
1136else:
1137 _c_current_task = current_task
1138 _c_register_task = _register_task
1139 _c_register_eager_task = _register_eager_task
1140 _c_unregister_task = _unregister_task
1141 _c_unregister_eager_task = _unregister_eager_task
1142 _c_enter_task = _enter_task
1143 _c_leave_task = _leave_task
1144 _c_swap_current_task = _swap_current_task
1145 _c_all_tasks = all_tasks