Coverage for Lib/asyncio/tasks.py: 95%
553 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-25 03:29 +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 elif not isinstance(context, contextvars.Context):
107 # gh-157301: the passed value must be a contextvars.Context
108 self._log_destroy_pending = False
109 raise TypeError('a contextvars.Context was expected, '
110 f'got {type(context).__name__}')
111 else:
112 self._context = context
114 if eager_start and self._loop.is_running():
115 try:
116 self.__eager_start()
117 except:
118 self._log_destroy_pending = False
119 raise
120 else:
121 self._loop.call_soon(self.__step, context=self._context)
122 _py_register_task(self)
124 def __del__(self):
125 if self._state == futures._PENDING and self._log_destroy_pending:
126 context = {
127 'task': self,
128 'message': 'Task was destroyed but it is pending!',
129 }
130 if self._source_traceback:
131 context['source_traceback'] = self._source_traceback
132 self._loop.call_exception_handler(context)
133 super().__del__()
135 __class_getitem__ = classmethod(GenericAlias)
137 def __repr__(self):
138 return base_tasks._task_repr(self)
140 def get_coro(self):
141 return self._coro
143 def get_context(self):
144 return self._context
146 def get_name(self):
147 return self._name
149 def set_name(self, value):
150 self._name = str(value)
152 def set_result(self, result):
153 raise RuntimeError('Task does not support set_result operation')
155 def set_exception(self, exception):
156 raise RuntimeError('Task does not support set_exception operation')
158 def get_stack(self, *, limit=None):
159 """Return the list of stack frames for this task's coroutine.
161 If the coroutine is not done, this returns the stack where it is
162 suspended. If the coroutine has completed successfully or was
163 cancelled, this returns an empty list. If the coroutine was
164 terminated by an exception, this returns the list of traceback
165 frames.
167 The frames are always ordered from oldest to newest.
169 The optional limit gives the maximum number of frames to
170 return; by default all available frames are returned. Its
171 meaning differs depending on whether a stack or a traceback is
172 returned: the newest frames of a stack are returned, but the
173 oldest frames of a traceback are returned. (This matches the
174 behavior of the traceback module.)
176 For reasons beyond our control, only one stack frame is
177 returned for a suspended coroutine.
178 """
179 return base_tasks._task_get_stack(self, limit)
181 def print_stack(self, *, limit=None, file=None):
182 """Print the stack or traceback for this task's coroutine.
184 This produces output similar to that of the traceback module,
185 for the frames retrieved by get_stack(). The limit argument
186 is passed to get_stack(). The file argument is an I/O stream
187 to which the output is written; by default output is written
188 to sys.stderr.
189 """
190 return base_tasks._task_print_stack(self, limit, file)
192 def cancel(self, msg=None):
193 """Request that this task cancel itself.
195 This arranges for a CancelledError to be thrown into the
196 wrapped coroutine on the next cycle through the event loop.
197 The coroutine then has a chance to clean up or even deny
198 the request using try/except/finally.
200 Unlike Future.cancel, this does not guarantee that the
201 task will be cancelled: the exception might be caught and
202 acted upon, delaying cancellation of the task or preventing
203 cancellation completely. The task may also return a value or
204 raise a different exception.
206 Immediately after this method is called, Task.cancelled() will
207 not return True (unless the task was already cancelled). A
208 task will be marked as cancelled when the wrapped coroutine
209 terminates with a CancelledError exception (even if cancel()
210 was not called).
212 This also increases the task's count of cancellation requests.
213 """
214 self._log_traceback = False
215 if self.done():
216 return False
217 self._num_cancels_requested += 1
218 # These two lines are controversial. See discussion starting at
219 # https://github.com/python/cpython/pull/31394#issuecomment-1053545331
220 # Also remember that this is duplicated in _asynciomodule.c.
221 # if self._num_cancels_requested > 1:
222 # return False
223 if self._fut_waiter is not None:
224 if self._fut_waiter.cancel(msg=msg):
225 # Leave self._fut_waiter; it may be a Task that
226 # catches and ignores the cancellation so we may have
227 # to cancel it again later.
228 return True
229 # It must be the case that self.__step is already scheduled.
230 self._must_cancel = True
231 self._cancel_message = msg
232 return True
234 def cancelling(self):
235 """Return the count of the task's cancellation requests.
237 This count is incremented when .cancel() is called
238 and may be decremented using .uncancel().
239 """
240 return self._num_cancels_requested
242 def uncancel(self):
243 """Decrement the task's count of cancellation requests.
245 This should be called by the party that called `cancel()` on the task
246 beforehand.
248 Returns the remaining number of cancellation requests.
249 """
250 if self._num_cancels_requested > 0: 250 ↛ 254line 250 didn't jump to line 254 because the condition on line 250 was always true
251 self._num_cancels_requested -= 1
252 if self._num_cancels_requested == 0: 252 ↛ 254line 252 didn't jump to line 254 because the condition on line 252 was always true
253 self._must_cancel = False
254 return self._num_cancels_requested
256 def __eager_start(self):
257 prev_task = _py_swap_current_task(self._loop, self)
258 try:
259 _py_register_eager_task(self)
260 try:
261 self._context.run(self.__step_run_and_handle_result, None)
262 finally:
263 _py_unregister_eager_task(self)
264 finally:
265 try:
266 curtask = _py_swap_current_task(self._loop, prev_task)
267 assert curtask is self
268 finally:
269 if self.done():
270 self._coro = None
271 self = None # Needed to break cycles when an exception occurs.
272 else:
273 _py_register_task(self)
275 def __step(self, exc=None):
276 if self.done():
277 raise exceptions.InvalidStateError(
278 f'__step(): already done: {self!r}, {exc!r}')
279 if self._must_cancel:
280 # gh-108549: do not swallow SystemExit and KeyboardInterrupt.
281 if not isinstance(exc, (exceptions.CancelledError,
282 SystemExit, KeyboardInterrupt)):
283 exc = self._make_cancelled_error()
284 self._must_cancel = False
285 self._fut_waiter = None
287 _py_enter_task(self._loop, self)
288 try:
289 self.__step_run_and_handle_result(exc)
290 finally:
291 _py_leave_task(self._loop, self)
292 self = None # Needed to break cycles when an exception occurs.
294 def __step_run_and_handle_result(self, exc):
295 coro = self._coro
296 try:
297 if exc is None:
298 # We use the `send` method directly, because coroutines
299 # don't have `__iter__` and `__next__` methods.
300 result = coro.send(None)
301 else:
302 result = coro.throw(exc)
303 except StopIteration as exc:
304 if self._must_cancel:
305 # Task is cancelled right before coro stops.
306 self._must_cancel = False
307 super().cancel(msg=self._cancel_message)
308 else:
309 super().set_result(exc.value)
310 except exceptions.CancelledError as exc:
311 # Save the original exception so we can chain it later.
312 self._cancelled_exc = exc
313 super().cancel() # I.e., Future.cancel(self).
314 except (KeyboardInterrupt, SystemExit) as exc:
315 super().set_exception(exc)
316 raise
317 except BaseException as exc:
318 super().set_exception(exc)
319 else:
320 blocking = getattr(result, '_asyncio_future_blocking', None)
321 if blocking is not None:
322 # Yielded Future must come from Future.__iter__().
323 if futures._get_loop(result) is not self._loop:
324 new_exc = RuntimeError(
325 f'Task {self!r} got Future '
326 f'{result!r} attached to a different loop')
327 self._loop.call_soon(
328 self.__step, new_exc, context=self._context)
329 elif blocking: 329 ↛ 346line 329 didn't jump to line 346 because the condition on line 329 was always true
330 if result is self:
331 new_exc = RuntimeError(
332 f'Task cannot await on itself: {self!r}')
333 self._loop.call_soon(
334 self.__step, new_exc, context=self._context)
335 else:
336 futures.future_add_to_awaited_by(result, self)
337 result._asyncio_future_blocking = False
338 result.add_done_callback(
339 self.__wakeup, context=self._context)
340 self._fut_waiter = result
341 if self._must_cancel:
342 if self._fut_waiter.cancel( 342 ↛ 368line 342 didn't jump to line 368 because the condition on line 342 was always true
343 msg=self._cancel_message):
344 self._must_cancel = False
345 else:
346 new_exc = RuntimeError(
347 f'yield was used instead of yield from '
348 f'in task {self!r} with {result!r}')
349 self._loop.call_soon(
350 self.__step, new_exc, context=self._context)
352 elif result is None: 352 ↛ 355line 352 didn't jump to line 355 because the condition on line 352 was always true
353 # Bare yield relinquishes control for one event loop iteration.
354 self._loop.call_soon(self.__step, context=self._context)
355 elif inspect.isgenerator(result):
356 # Yielding a generator is just wrong.
357 new_exc = RuntimeError(
358 f'yield was used instead of yield from for '
359 f'generator in task {self!r} with {result!r}')
360 self._loop.call_soon(
361 self.__step, new_exc, context=self._context)
362 else:
363 # Yielding something else is an error.
364 new_exc = RuntimeError(f'Task got bad yield: {result!r}')
365 self._loop.call_soon(
366 self.__step, new_exc, context=self._context)
367 finally:
368 self = None # Needed to break cycles when an exception occurs.
370 def __wakeup(self, future):
371 futures.future_discard_from_awaited_by(future, self)
372 try:
373 future.result()
374 except BaseException as exc:
375 # This may also be a cancellation.
376 self.__step(exc)
377 else:
378 # Don't pass the value of `future.result()` explicitly,
379 # as `Future.__iter__` and `Future.__await__` don't need it.
380 # If we call `__step(value, None)` instead of `__step()`,
381 # Python eval loop would use `.send(value)` method call,
382 # instead of `__next__()`, which is slower for futures
383 # that return non-generator iterators from their `__iter__`.
384 self.__step()
385 self = None # Needed to break cycles when an exception occurs.
388_PyTask = Task
391try:
392 import _asyncio
393except ImportError:
394 pass
395else:
396 # _CTask is needed for tests.
397 Task = _CTask = _asyncio.Task
400def create_task(coro, **kwargs):
401 """Schedule the execution of a coroutine object in a spawn task.
403 Return a Task object.
404 """
405 loop = events.get_running_loop()
406 return loop.create_task(coro, **kwargs)
409# wait() and as_completed() similar to those in PEP 3148.
411FIRST_COMPLETED = concurrent.futures.FIRST_COMPLETED
412FIRST_EXCEPTION = concurrent.futures.FIRST_EXCEPTION
413ALL_COMPLETED = concurrent.futures.ALL_COMPLETED
416async def wait(fs, *, timeout=None, return_when=ALL_COMPLETED):
417 """Wait for the Futures or Tasks given by fs to complete.
419 The fs iterable must not be empty.
421 Returns two sets of Future: (done, pending).
423 Usage:
425 done, pending = await asyncio.wait(fs)
427 Note: This does not raise TimeoutError! Futures that aren't done
428 when the timeout occurs are returned in the second set.
429 """
430 if futures.isfuture(fs) or coroutines.iscoroutine(fs):
431 raise TypeError(f"expect a list of futures, not {type(fs).__name__}")
432 if not fs:
433 raise ValueError('Set of Tasks/Futures is empty.')
434 if return_when not in (FIRST_COMPLETED, FIRST_EXCEPTION, ALL_COMPLETED):
435 raise ValueError(f'Invalid return_when value: {return_when}')
437 fs = set(fs)
439 if any(coroutines.iscoroutine(f) for f in fs): 439 ↛ 440line 439 didn't jump to line 440 because the condition on line 439 was never true
440 raise TypeError("Passing coroutines is forbidden, use tasks explicitly.")
442 loop = events.get_running_loop()
443 return await _wait(fs, timeout, return_when, loop)
446def _release_waiter(waiter, *args):
447 if not waiter.done(): 447 ↛ exitline 447 didn't return from function '_release_waiter' because the condition on line 447 was always true
448 waiter.set_result(None)
451async def wait_for(fut, timeout):
452 """Wait for the single Future or coroutine to complete, with timeout.
454 Returns result of the Future or coroutine. When a timeout occurs,
455 it cancels fut and raises TimeoutError. To prevent fut from being
456 cancelled, wrap it in shield().
458 If the wait is cancelled, fut is also cancelled.
460 If fut suppresses the cancellation and returns a value instead,
461 that value is returned.
463 This function is a coroutine.
464 """
465 # The special case for timeout <= 0 is for the following case:
466 #
467 # async def test_waitfor():
468 # func_started = False
469 #
470 # async def func():
471 # nonlocal func_started
472 # func_started = True
473 #
474 # try:
475 # await asyncio.wait_for(func(), 0)
476 # except asyncio.TimeoutError:
477 # assert not func_started
478 # else:
479 # assert False
480 #
481 # asyncio.run(test_waitfor())
484 if timeout is not None and timeout <= 0:
485 fut = ensure_future(fut)
487 if fut.done():
488 return fut.result()
490 await _cancel_and_wait(fut)
491 try:
492 return fut.result()
493 except exceptions.CancelledError as exc:
494 raise TimeoutError from exc
496 async with timeouts.timeout(timeout):
497 return await fut
499async def _wait(fs, timeout, return_when, loop):
500 """Internal helper for wait().
502 The fs argument must be a collection of Futures.
503 """
504 assert fs, 'Set of Futures is empty.'
505 waiter = loop.create_future()
506 timeout_handle = None
507 if timeout is not None:
508 timeout_handle = loop.call_later(timeout, _release_waiter, waiter)
509 counter = len(fs)
510 cur_task = current_task()
512 def _on_completion(f):
513 nonlocal counter
514 counter -= 1
515 if (counter <= 0 or
516 return_when == FIRST_COMPLETED or
517 return_when == FIRST_EXCEPTION and (not f.cancelled() and
518 f.exception() is not None)):
519 if timeout_handle is not None: 519 ↛ 520line 519 didn't jump to line 520 because the condition on line 519 was never true
520 timeout_handle.cancel()
521 if not waiter.done():
522 waiter.set_result(None)
523 futures.future_discard_from_awaited_by(f, cur_task)
525 for f in fs:
526 f.add_done_callback(_on_completion)
527 futures.future_add_to_awaited_by(f, cur_task)
529 try:
530 await waiter
531 finally:
532 if timeout_handle is not None:
533 timeout_handle.cancel()
534 for f in fs:
535 f.remove_done_callback(_on_completion)
536 futures.future_discard_from_awaited_by(f, cur_task)
538 done, pending = set(), set()
539 for f in fs:
540 if f.done():
541 done.add(f)
542 else:
543 pending.add(f)
544 return done, pending
547async def _cancel_and_wait(fut):
548 """Cancel the *fut* future or task and wait until it completes."""
550 loop = events.get_running_loop()
551 waiter = loop.create_future()
552 cb = functools.partial(_release_waiter, waiter)
553 fut.add_done_callback(cb)
555 # gh-157058: awaiting the waiter leaves no edge on fut, add it here
556 cur_task = current_task()
557 futures.future_add_to_awaited_by(fut, cur_task)
559 try:
560 fut.cancel()
561 # We cannot wait on *fut* directly to make
562 # sure _cancel_and_wait itself is reliably cancellable.
563 await waiter
564 finally:
565 fut.remove_done_callback(cb)
566 futures.future_discard_from_awaited_by(fut, cur_task)
569class _AsCompletedIterator:
570 """Iterator of awaitables representing tasks of asyncio.as_completed.
572 As an asynchronous iterator, iteration yields futures as they finish. As a
573 plain iterator, new coroutines are yielded that will return or raise the
574 result of the next underlying future to complete.
575 """
576 def __init__(self, aws, timeout):
577 self._done = queues.Queue()
578 self._timeout_handle = None
580 loop = events.get_event_loop()
581 self._cur_task = current_task()
582 todo = {ensure_future(aw, loop=loop) for aw in set(aws)}
583 for f in todo:
584 f.add_done_callback(self._handle_completion)
585 futures.future_add_to_awaited_by(f, self._cur_task)
586 if todo and timeout is not None:
587 self._timeout_handle = (
588 loop.call_later(timeout, self._handle_timeout)
589 )
590 self._todo = todo
591 self._todo_left = len(todo)
593 def __aiter__(self):
594 return self
596 def __iter__(self):
597 return self
599 async def __anext__(self):
600 if not self._todo_left:
601 raise StopAsyncIteration
602 assert self._todo_left > 0
603 self._todo_left -= 1
604 return await self._wait_for_one()
606 def __next__(self):
607 if not self._todo_left:
608 raise StopIteration
609 assert self._todo_left > 0
610 self._todo_left -= 1
611 return self._wait_for_one(resolve=True)
613 def _handle_timeout(self):
614 for f in self._todo:
615 f.remove_done_callback(self._handle_completion)
616 futures.future_discard_from_awaited_by(f, self._cur_task)
617 self._done.put_nowait(None) # Sentinel for _wait_for_one().
618 self._todo.clear() # Can't do todo.remove(f) in the loop.
620 def _handle_completion(self, f):
621 if not self._todo: 621 ↛ 622line 621 didn't jump to line 622 because the condition on line 621 was never true
622 return # _handle_timeout() was here first.
623 self._todo.remove(f)
624 futures.future_discard_from_awaited_by(f, self._cur_task)
625 self._done.put_nowait(f)
626 if not self._todo and self._timeout_handle is not None:
627 self._timeout_handle.cancel()
629 async def _wait_for_one(self, resolve=False):
630 # Wait for the next future to be done and return it unless resolve is
631 # set, in which case return either the result of the future or raise
632 # an exception.
633 f = await self._done.get()
634 if f is None:
635 # Dummy value from _handle_timeout().
636 raise exceptions.TimeoutError
637 return f.result() if resolve else f
640def as_completed(fs, *, timeout=None):
641 """Create an iterator of awaitables or their results in completion order.
643 Run the supplied awaitables concurrently. The returned object can be
644 iterated to obtain the results of the awaitables as they finish.
646 The object returned can be iterated as an asynchronous iterator or
647 a plain iterator. When asynchronous iteration is used, the
648 originally-supplied awaitables are yielded if they are tasks or
649 futures. This makes it easy to correlate previously-scheduled tasks
650 with their results:
652 ipv4_connect = create_task(open_connection("127.0.0.1", 80))
653 ipv6_connect = create_task(open_connection("::1", 80))
654 tasks = [ipv4_connect, ipv6_connect]
656 async for earliest_connect in as_completed(tasks):
657 # earliest_connect is done. The result can be obtained by
658 # awaiting it or calling earliest_connect.result()
659 reader, writer = await earliest_connect
661 if earliest_connect is ipv6_connect:
662 print("IPv6 connection established.")
663 else:
664 print("IPv4 connection established.")
666 During asynchronous iteration, implicitly-created tasks will be
667 yielded for supplied awaitables that aren't tasks or futures.
669 When used as a plain iterator, each iteration yields a new coroutine
670 that returns the result or raises the exception of the next completed
671 awaitable. This pattern is compatible with Python versions older than
672 3.13:
674 ipv4_connect = create_task(open_connection("127.0.0.1", 80))
675 ipv6_connect = create_task(open_connection("::1", 80))
676 tasks = [ipv4_connect, ipv6_connect]
678 for next_connect in as_completed(tasks):
679 # next_connect is not one of the original task objects. It must
680 # be awaited to obtain the result value or raise the exception
681 # of the awaitable that finishes next.
682 reader, writer = await next_connect
684 A TimeoutError is raised if the timeout occurs before all awaitables
685 are done. This is raised by the async for loop during asynchronous
686 iteration or by the coroutines yielded during plain iteration.
687 """
688 if inspect.isawaitable(fs):
689 raise TypeError(
690 f"expects an iterable of awaitables, not {type(fs).__name__}"
691 )
693 return _AsCompletedIterator(fs, timeout)
696@types.coroutine
697def __sleep0():
698 """Skip one event loop run cycle.
700 This is a private helper for 'asyncio.sleep()', used
701 when the 'delay' is set to 0. It uses a bare 'yield'
702 expression (which Task.__step knows how to handle)
703 instead of creating a Future object.
704 """
705 yield
708async def sleep(delay, result=None):
709 """Coroutine that completes after a given time (in seconds)."""
710 if delay <= 0:
711 await __sleep0()
712 return result
714 if math.isnan(delay):
715 raise ValueError("Invalid delay: NaN (not a number)")
717 loop = events.get_running_loop()
718 future = loop.create_future()
719 h = loop.call_later(delay,
720 futures._set_result_unless_cancelled,
721 future, result)
722 try:
723 return await future
724 finally:
725 h.cancel()
728def ensure_future(coro_or_future, *, loop=None):
729 """Wrap a coroutine or an awaitable in a future.
731 If the argument is a Future, it is returned directly.
732 """
733 if futures.isfuture(coro_or_future):
734 if loop is not None and loop is not futures._get_loop(coro_or_future):
735 raise ValueError('The future belongs to a different loop than '
736 'the one specified as the loop argument')
737 return coro_or_future
738 should_close = True
739 if not coroutines.iscoroutine(coro_or_future):
740 if inspect.isawaitable(coro_or_future):
741 async def _wrap_awaitable(awaitable):
742 return await awaitable
744 coro_or_future = _wrap_awaitable(coro_or_future)
745 should_close = False
746 else:
747 raise TypeError('An asyncio.Future, a coroutine or an awaitable '
748 'is required')
750 if loop is None:
751 loop = events.get_event_loop()
752 try:
753 return loop.create_task(coro_or_future)
754 except RuntimeError:
755 if should_close: 755 ↛ 757line 755 didn't jump to line 757 because the condition on line 755 was always true
756 coro_or_future.close()
757 raise
760class _GatheringFuture(futures.Future):
761 """Helper for gather().
763 This overrides cancel() to cancel all the children and act more
764 like Task.cancel(), which doesn't immediately mark itself as
765 cancelled.
766 """
768 def __init__(self, children, *, loop):
769 assert loop is not None
770 super().__init__(loop=loop)
771 self._children = children
772 self._cancel_requested = False
774 def cancel(self, msg=None):
775 if self.done():
776 return False
777 ret = False
778 for child in self._children:
779 if child.cancel(msg=msg):
780 ret = True
781 if ret:
782 # If any child tasks were actually cancelled, we should
783 # propagate the cancellation request regardless of
784 # *return_exceptions* argument. See issue 32684.
785 self._cancel_requested = True
786 return ret
789def _discard_awaited_by(children, waiter, outer):
790 for fut in children:
791 futures.future_discard_from_awaited_by(fut, waiter)
794def gather(*coros_or_futures, return_exceptions=False):
795 """Return a future aggregating results from the given coroutines/futures.
797 Coroutines will be wrapped in a future and scheduled in the event
798 loop. They will not necessarily be scheduled in the same order as
799 passed in.
801 All futures must share the same event loop. If all the tasks are
802 done successfully, the returned future's result is the list of
803 results (in the order of the original sequence, not necessarily
804 the order of results arrival). If *return_exceptions* is True,
805 exceptions in the tasks are treated the same as successful
806 results, and gathered in the result list; otherwise, the first
807 raised exception will be immediately propagated to the returned
808 future.
810 Cancellation: if the outer Future is cancelled, all children (that
811 have not completed yet) are also cancelled. If any child is
812 cancelled, this is treated as if it raised CancelledError --
813 the outer Future is *not* cancelled in this case. (This is to
814 prevent the cancellation of one child to cause other children to
815 be cancelled.)
817 If *return_exceptions* is False, cancelling gather() after it
818 has been marked done won't cancel any submitted awaitables.
819 For instance, gather can be marked done after propagating an
820 exception to the caller, therefore, calling ``gather.cancel()``
821 after catching an exception (raised by one of the awaitables) from
822 gather won't cancel any other awaitables.
823 """
824 if not coros_or_futures:
825 loop = events.get_event_loop()
826 outer = loop.create_future()
827 outer.set_result([])
828 return outer
830 loop = events._get_running_loop()
831 if loop is not None:
832 cur_task = current_task(loop)
833 else:
834 cur_task = None
836 def _done_callback(fut, cur_task=cur_task):
837 nonlocal nfinished
838 nfinished += 1
840 if cur_task is not None:
841 futures.future_discard_from_awaited_by(fut, cur_task)
843 if outer is None or outer.done():
844 if not fut.cancelled():
845 # Mark exception retrieved.
846 fut.exception()
847 return
849 if not return_exceptions:
850 if fut.cancelled():
851 # Check if 'fut' is cancelled first, as
852 # 'fut.exception()' will *raise* a CancelledError
853 # instead of returning it.
854 exc = fut._make_cancelled_error()
855 outer.set_exception(exc)
856 return
857 else:
858 exc = fut.exception()
859 if exc is not None:
860 outer.set_exception(exc)
861 return
863 if nfinished == nfuts:
864 # All futures are done; create a list of results
865 # and set it to the 'outer' future.
866 results = []
868 for fut in children:
869 if fut.cancelled():
870 # Check if 'fut' is cancelled first, as 'fut.exception()'
871 # will *raise* a CancelledError instead of returning it.
872 # Also, since we're adding the exception return value
873 # to 'results' instead of raising it, don't bother
874 # setting __context__. This also lets us preserve
875 # calling '_make_cancelled_error()' at most once.
876 res = exceptions.CancelledError(
877 '' if fut._cancel_message is None else
878 fut._cancel_message)
879 else:
880 res = fut.exception()
881 if res is None:
882 res = fut.result()
883 results.append(res)
885 if outer._cancel_requested:
886 # If gather is being cancelled we must propagate the
887 # cancellation regardless of *return_exceptions* argument.
888 # See issue 32684.
889 exc = fut._make_cancelled_error()
890 outer.set_exception(exc)
891 else:
892 outer.set_result(results)
894 arg_to_fut = {}
895 children = []
896 nfuts = 0
897 nfinished = 0
898 done_futs = []
899 outer = None # bpo-46672
900 for arg in coros_or_futures:
901 if arg not in arg_to_fut:
902 fut = ensure_future(arg, loop=loop)
903 if loop is None:
904 loop = futures._get_loop(fut)
905 if fut is not arg:
906 # 'arg' was not a Future, therefore, 'fut' is a new
907 # Future created specifically for 'arg'. Since the caller
908 # can't control it, disable the "destroy pending task"
909 # warning.
910 fut._log_destroy_pending = False
911 nfuts += 1
912 arg_to_fut[arg] = fut
913 if fut.done():
914 done_futs.append(fut)
915 else:
916 if cur_task is not None:
917 futures.future_add_to_awaited_by(fut, cur_task)
918 fut.add_done_callback(_done_callback)
920 else:
921 # There's a duplicate Future object in coros_or_futures.
922 fut = arg_to_fut[arg]
924 children.append(fut)
926 outer = _GatheringFuture(children, loop=loop)
927 if cur_task is not None:
928 # gh-157213: a child outliving gather() must lose the awaited-by edge
929 outer.add_done_callback(
930 functools.partial(_discard_awaited_by, children, cur_task))
931 # Run done callbacks after GatheringFuture created so any post-processing
932 # can be performed at this point
933 # optimization: in the special case that *all* futures finished eagerly,
934 # this will effectively complete the gather eagerly, with the last
935 # callback setting the result (or exception) on outer before returning it
936 for fut in done_futs:
937 _done_callback(fut)
938 return outer
941def _log_on_exception(fut):
942 if fut.cancelled(): 942 ↛ 943line 942 didn't jump to line 943 because the condition on line 942 was never true
943 return
945 exc = fut.exception()
946 if exc is None:
947 return
949 context = {
950 'message':
951 f'{exc.__class__.__name__} exception in shielded future',
952 'exception': exc,
953 'future': fut,
954 }
955 if fut._source_traceback: 955 ↛ 956line 955 didn't jump to line 956 because the condition on line 955 was never true
956 context['source_traceback'] = fut._source_traceback
957 fut._loop.call_exception_handler(context)
960def shield(arg):
961 """Wait for a future, shielding it from cancellation.
963 The statement
965 task = asyncio.create_task(something())
966 res = await shield(task)
968 is exactly equivalent to the statement
970 res = await something()
972 *except* that if the coroutine containing it is cancelled, the
973 task running in something() is not cancelled. From the POV of
974 something(), the cancellation did not happen. But its caller is
975 still cancelled, so the yield-from expression still raises
976 CancelledError. Note: If something() is cancelled by other means
977 this will still cancel shield().
979 If you want to completely ignore cancellation (not recommended)
980 you can combine shield() with a try/except clause, as follows:
982 task = asyncio.create_task(something())
983 try:
984 res = await shield(task)
985 except CancelledError:
986 res = None
988 Save a reference to tasks passed to this function, to avoid
989 a task disappearing mid-execution. The event loop only keeps
990 weak references to tasks. A task that isn't referenced elsewhere
991 may get garbage collected at any time, even before it's done.
992 """
993 inner = ensure_future(arg)
994 if inner.done():
995 # Shortcut.
996 return inner
997 loop = futures._get_loop(inner)
998 outer = loop.create_future()
1000 if loop is not None and (cur_task := current_task(loop)) is not None:
1001 futures.future_add_to_awaited_by(inner, cur_task)
1002 else:
1003 cur_task = None
1005 def _clear_awaited_by_callback(inner):
1006 futures.future_discard_from_awaited_by(inner, cur_task)
1008 def _inner_done_callback(inner):
1009 if outer.cancelled(): 1009 ↛ 1010line 1009 didn't jump to line 1010 because the condition on line 1009 was never true
1010 return
1012 if inner.cancelled():
1013 outer.cancel()
1014 else:
1015 exc = inner.exception()
1016 if exc is not None:
1017 outer.set_exception(exc)
1018 else:
1019 outer.set_result(inner.result())
1021 def _outer_done_callback(outer):
1022 if not inner.done():
1023 inner.remove_done_callback(_inner_done_callback)
1024 # Keep only one callback to log on cancel
1025 inner.remove_done_callback(_log_on_exception)
1026 inner.add_done_callback(_log_on_exception)
1027 if cur_task is not None:
1028 inner.remove_done_callback(_clear_awaited_by_callback)
1029 futures.future_discard_from_awaited_by(inner, cur_task)
1031 if cur_task is not None:
1032 inner.add_done_callback(_clear_awaited_by_callback)
1035 inner.add_done_callback(_inner_done_callback)
1036 outer.add_done_callback(_outer_done_callback)
1037 return outer
1040def run_coroutine_threadsafe(coro, loop):
1041 """Submit a coroutine object to a given event loop.
1043 Return a concurrent.futures.Future to access the result.
1044 """
1045 if not coroutines.iscoroutine(coro): 1045 ↛ 1046line 1045 didn't jump to line 1046 because the condition on line 1045 was never true
1046 raise TypeError('A coroutine object is required')
1047 future = concurrent.futures.Future()
1049 def callback():
1050 try:
1051 futures._chain_future(ensure_future(coro, loop=loop), future)
1052 except (SystemExit, KeyboardInterrupt):
1053 raise
1054 except BaseException as exc:
1055 if future.set_running_or_notify_cancel(): 1055 ↛ 1057line 1055 didn't jump to line 1057 because the condition on line 1055 was always true
1056 future.set_exception(exc)
1057 raise
1059 loop.call_soon_threadsafe(callback)
1060 return future
1063def create_eager_task_factory(custom_task_constructor):
1064 """Create a function suitable for use as a task factory on an event-loop.
1066 Example usage:
1068 loop.set_task_factory(
1069 asyncio.create_eager_task_factory(my_task_constructor))
1071 Now, tasks created will be started immediately (rather than being first
1072 scheduled to an event loop). The constructor argument can be any
1073 callable that returns a Task-compatible object and has a signature
1074 compatible with `Task.__init__`; it must have the `eager_start`
1075 keyword argument.
1077 Most applications will use `Task` for `custom_task_constructor` and in
1078 this case there's no need to call `create_eager_task_factory()`
1079 directly. Instead the global `eager_task_factory` instance can be
1080 used. E.g. `loop.set_task_factory(asyncio.eager_task_factory)`.
1081 """
1083 def factory(loop, coro, *, eager_start=True, **kwargs):
1084 return custom_task_constructor(
1085 coro, loop=loop, eager_start=eager_start, **kwargs)
1087 return factory
1090eager_task_factory = create_eager_task_factory(Task)
1093# Collectively these two sets hold references to the complete set of active
1094# tasks. Eagerly executed tasks use a faster regular set as an optimization
1095# but may graduate to a WeakSet if the task blocks on IO.
1096_scheduled_tasks = weakref.WeakSet()
1097_eager_tasks = set()
1099# Dictionary containing tasks that are currently active in
1100# all running event loops. {EventLoop: Task}
1101_current_tasks = {}
1104def _register_task(task):
1105 """Register an asyncio Task scheduled to run on an event loop."""
1106 _scheduled_tasks.add(task)
1109def _register_eager_task(task):
1110 """Register an asyncio Task about to be eagerly executed."""
1111 _eager_tasks.add(task)
1114def _enter_task(loop, task):
1115 current_task = _current_tasks.get(loop)
1116 if current_task is not None:
1117 raise RuntimeError(f"Cannot enter into task {task!r} while another "
1118 f"task {current_task!r} is being executed.")
1119 _current_tasks[loop] = task
1122def _leave_task(loop, task):
1123 current_task = _current_tasks.get(loop)
1124 if current_task is not task:
1125 raise RuntimeError(f"Leaving task {task!r} does not match "
1126 f"the current task {current_task!r}.")
1127 del _current_tasks[loop]
1130def _swap_current_task(loop, task):
1131 prev_task = _current_tasks.get(loop)
1132 if task is None:
1133 del _current_tasks[loop]
1134 else:
1135 _current_tasks[loop] = task
1136 return prev_task
1139def _unregister_task(task):
1140 """Unregister a completed, scheduled Task."""
1141 _scheduled_tasks.discard(task)
1144def _unregister_eager_task(task):
1145 """Unregister a task which finished its first eager step."""
1146 _eager_tasks.discard(task)
1149_py_current_task = current_task
1150_py_register_task = _register_task
1151_py_register_eager_task = _register_eager_task
1152_py_unregister_task = _unregister_task
1153_py_unregister_eager_task = _unregister_eager_task
1154_py_enter_task = _enter_task
1155_py_leave_task = _leave_task
1156_py_swap_current_task = _swap_current_task
1157_py_all_tasks = all_tasks
1159try:
1160 from _asyncio import (_register_task, _register_eager_task,
1161 _unregister_task, _unregister_eager_task,
1162 _enter_task, _leave_task, _swap_current_task,
1163 current_task, all_tasks)
1164except ImportError:
1165 pass
1166else:
1167 _c_current_task = current_task
1168 _c_register_task = _register_task
1169 _c_register_eager_task = _register_eager_task
1170 _c_unregister_task = _unregister_task
1171 _c_unregister_eager_task = _unregister_eager_task
1172 _c_enter_task = _enter_task
1173 _c_leave_task = _leave_task
1174 _c_swap_current_task = _swap_current_task
1175 _c_all_tasks = all_tasks