logo

oasis-root

Compiled tree of Oasis Linux based on own branch at <https://hacktivis.me/git/oasis/> git clone https://anongit.hacktivis.me/git/oasis-root.git

tasks.py (31950B)


  1. """Support for tasks, coroutines and the scheduler."""
  2. __all__ = (
  3. 'Task', 'create_task',
  4. 'FIRST_COMPLETED', 'FIRST_EXCEPTION', 'ALL_COMPLETED',
  5. 'wait', 'wait_for', 'as_completed', 'sleep',
  6. 'gather', 'shield', 'ensure_future', 'run_coroutine_threadsafe',
  7. 'current_task', 'all_tasks',
  8. '_register_task', '_unregister_task', '_enter_task', '_leave_task',
  9. )
  10. import concurrent.futures
  11. import contextvars
  12. import functools
  13. import inspect
  14. import itertools
  15. import types
  16. import warnings
  17. import weakref
  18. from . import base_tasks
  19. from . import coroutines
  20. from . import events
  21. from . import exceptions
  22. from . import futures
  23. from .coroutines import _is_coroutine
  24. # Helper to generate new task names
  25. # This uses itertools.count() instead of a "+= 1" operation because the latter
  26. # is not thread safe. See bpo-11866 for a longer explanation.
  27. _task_name_counter = itertools.count(1).__next__
  28. def current_task(loop=None):
  29. """Return a currently executed task."""
  30. if loop is None:
  31. loop = events.get_running_loop()
  32. return _current_tasks.get(loop)
  33. def all_tasks(loop=None):
  34. """Return a set of all tasks for the loop."""
  35. if loop is None:
  36. loop = events.get_running_loop()
  37. # Looping over a WeakSet (_all_tasks) isn't safe as it can be updated from another
  38. # thread while we do so. Therefore we cast it to list prior to filtering. The list
  39. # cast itself requires iteration, so we repeat it several times ignoring
  40. # RuntimeErrors (which are not very likely to occur). See issues 34970 and 36607 for
  41. # details.
  42. i = 0
  43. while True:
  44. try:
  45. tasks = list(_all_tasks)
  46. except RuntimeError:
  47. i += 1
  48. if i >= 1000:
  49. raise
  50. else:
  51. break
  52. return {t for t in tasks
  53. if futures._get_loop(t) is loop and not t.done()}
  54. def _set_task_name(task, name):
  55. if name is not None:
  56. try:
  57. set_name = task.set_name
  58. except AttributeError:
  59. pass
  60. else:
  61. set_name(name)
  62. class Task(futures._PyFuture): # Inherit Python Task implementation
  63. # from a Python Future implementation.
  64. """A coroutine wrapped in a Future."""
  65. # An important invariant maintained while a Task not done:
  66. #
  67. # - Either _fut_waiter is None, and _step() is scheduled;
  68. # - or _fut_waiter is some Future, and _step() is *not* scheduled.
  69. #
  70. # The only transition from the latter to the former is through
  71. # _wakeup(). When _fut_waiter is not None, one of its callbacks
  72. # must be _wakeup().
  73. # If False, don't log a message if the task is destroyed whereas its
  74. # status is still pending
  75. _log_destroy_pending = True
  76. def __init__(self, coro, *, loop=None, name=None):
  77. super().__init__(loop=loop)
  78. if self._source_traceback:
  79. del self._source_traceback[-1]
  80. if not coroutines.iscoroutine(coro):
  81. # raise after Future.__init__(), attrs are required for __del__
  82. # prevent logging for pending task in __del__
  83. self._log_destroy_pending = False
  84. raise TypeError(f"a coroutine was expected, got {coro!r}")
  85. if name is None:
  86. self._name = f'Task-{_task_name_counter()}'
  87. else:
  88. self._name = str(name)
  89. self._must_cancel = False
  90. self._fut_waiter = None
  91. self._coro = coro
  92. self._context = contextvars.copy_context()
  93. self._loop.call_soon(self.__step, context=self._context)
  94. _register_task(self)
  95. def __del__(self):
  96. if self._state == futures._PENDING and self._log_destroy_pending:
  97. context = {
  98. 'task': self,
  99. 'message': 'Task was destroyed but it is pending!',
  100. }
  101. if self._source_traceback:
  102. context['source_traceback'] = self._source_traceback
  103. self._loop.call_exception_handler(context)
  104. super().__del__()
  105. def __class_getitem__(cls, type):
  106. return cls
  107. def _repr_info(self):
  108. return base_tasks._task_repr_info(self)
  109. def get_coro(self):
  110. return self._coro
  111. def get_name(self):
  112. return self._name
  113. def set_name(self, value):
  114. self._name = str(value)
  115. def set_result(self, result):
  116. raise RuntimeError('Task does not support set_result operation')
  117. def set_exception(self, exception):
  118. raise RuntimeError('Task does not support set_exception operation')
  119. def get_stack(self, *, limit=None):
  120. """Return the list of stack frames for this task's coroutine.
  121. If the coroutine is not done, this returns the stack where it is
  122. suspended. If the coroutine has completed successfully or was
  123. cancelled, this returns an empty list. If the coroutine was
  124. terminated by an exception, this returns the list of traceback
  125. frames.
  126. The frames are always ordered from oldest to newest.
  127. The optional limit gives the maximum number of frames to
  128. return; by default all available frames are returned. Its
  129. meaning differs depending on whether a stack or a traceback is
  130. returned: the newest frames of a stack are returned, but the
  131. oldest frames of a traceback are returned. (This matches the
  132. behavior of the traceback module.)
  133. For reasons beyond our control, only one stack frame is
  134. returned for a suspended coroutine.
  135. """
  136. return base_tasks._task_get_stack(self, limit)
  137. def print_stack(self, *, limit=None, file=None):
  138. """Print the stack or traceback for this task's coroutine.
  139. This produces output similar to that of the traceback module,
  140. for the frames retrieved by get_stack(). The limit argument
  141. is passed to get_stack(). The file argument is an I/O stream
  142. to which the output is written; by default output is written
  143. to sys.stderr.
  144. """
  145. return base_tasks._task_print_stack(self, limit, file)
  146. def cancel(self, msg=None):
  147. """Request that this task cancel itself.
  148. This arranges for a CancelledError to be thrown into the
  149. wrapped coroutine on the next cycle through the event loop.
  150. The coroutine then has a chance to clean up or even deny
  151. the request using try/except/finally.
  152. Unlike Future.cancel, this does not guarantee that the
  153. task will be cancelled: the exception might be caught and
  154. acted upon, delaying cancellation of the task or preventing
  155. cancellation completely. The task may also return a value or
  156. raise a different exception.
  157. Immediately after this method is called, Task.cancelled() will
  158. not return True (unless the task was already cancelled). A
  159. task will be marked as cancelled when the wrapped coroutine
  160. terminates with a CancelledError exception (even if cancel()
  161. was not called).
  162. """
  163. self._log_traceback = False
  164. if self.done():
  165. return False
  166. if self._fut_waiter is not None:
  167. if self._fut_waiter.cancel(msg=msg):
  168. # Leave self._fut_waiter; it may be a Task that
  169. # catches and ignores the cancellation so we may have
  170. # to cancel it again later.
  171. return True
  172. # It must be the case that self.__step is already scheduled.
  173. self._must_cancel = True
  174. self._cancel_message = msg
  175. return True
  176. def __step(self, exc=None):
  177. if self.done():
  178. raise exceptions.InvalidStateError(
  179. f'_step(): already done: {self!r}, {exc!r}')
  180. if self._must_cancel:
  181. if not isinstance(exc, exceptions.CancelledError):
  182. exc = self._make_cancelled_error()
  183. self._must_cancel = False
  184. coro = self._coro
  185. self._fut_waiter = None
  186. _enter_task(self._loop, self)
  187. # Call either coro.throw(exc) or coro.send(None).
  188. try:
  189. if exc is None:
  190. # We use the `send` method directly, because coroutines
  191. # don't have `__iter__` and `__next__` methods.
  192. result = coro.send(None)
  193. else:
  194. result = coro.throw(exc)
  195. except StopIteration as exc:
  196. if self._must_cancel:
  197. # Task is cancelled right before coro stops.
  198. self._must_cancel = False
  199. super().cancel(msg=self._cancel_message)
  200. else:
  201. super().set_result(exc.value)
  202. except exceptions.CancelledError as exc:
  203. # Save the original exception so we can chain it later.
  204. self._cancelled_exc = exc
  205. super().cancel() # I.e., Future.cancel(self).
  206. except (KeyboardInterrupt, SystemExit) as exc:
  207. super().set_exception(exc)
  208. raise
  209. except BaseException as exc:
  210. super().set_exception(exc)
  211. else:
  212. blocking = getattr(result, '_asyncio_future_blocking', None)
  213. if blocking is not None:
  214. # Yielded Future must come from Future.__iter__().
  215. if futures._get_loop(result) is not self._loop:
  216. new_exc = RuntimeError(
  217. f'Task {self!r} got Future '
  218. f'{result!r} attached to a different loop')
  219. self._loop.call_soon(
  220. self.__step, new_exc, context=self._context)
  221. elif blocking:
  222. if result is self:
  223. new_exc = RuntimeError(
  224. f'Task cannot await on itself: {self!r}')
  225. self._loop.call_soon(
  226. self.__step, new_exc, context=self._context)
  227. else:
  228. result._asyncio_future_blocking = False
  229. result.add_done_callback(
  230. self.__wakeup, context=self._context)
  231. self._fut_waiter = result
  232. if self._must_cancel:
  233. if self._fut_waiter.cancel(
  234. msg=self._cancel_message):
  235. self._must_cancel = False
  236. else:
  237. new_exc = RuntimeError(
  238. f'yield was used instead of yield from '
  239. f'in task {self!r} with {result!r}')
  240. self._loop.call_soon(
  241. self.__step, new_exc, context=self._context)
  242. elif result is None:
  243. # Bare yield relinquishes control for one event loop iteration.
  244. self._loop.call_soon(self.__step, context=self._context)
  245. elif inspect.isgenerator(result):
  246. # Yielding a generator is just wrong.
  247. new_exc = RuntimeError(
  248. f'yield was used instead of yield from for '
  249. f'generator in task {self!r} with {result!r}')
  250. self._loop.call_soon(
  251. self.__step, new_exc, context=self._context)
  252. else:
  253. # Yielding something else is an error.
  254. new_exc = RuntimeError(f'Task got bad yield: {result!r}')
  255. self._loop.call_soon(
  256. self.__step, new_exc, context=self._context)
  257. finally:
  258. _leave_task(self._loop, self)
  259. self = None # Needed to break cycles when an exception occurs.
  260. def __wakeup(self, future):
  261. try:
  262. future.result()
  263. except BaseException as exc:
  264. # This may also be a cancellation.
  265. self.__step(exc)
  266. else:
  267. # Don't pass the value of `future.result()` explicitly,
  268. # as `Future.__iter__` and `Future.__await__` don't need it.
  269. # If we call `_step(value, None)` instead of `_step()`,
  270. # Python eval loop would use `.send(value)` method call,
  271. # instead of `__next__()`, which is slower for futures
  272. # that return non-generator iterators from their `__iter__`.
  273. self.__step()
  274. self = None # Needed to break cycles when an exception occurs.
  275. _PyTask = Task
  276. try:
  277. import _asyncio
  278. except ImportError:
  279. pass
  280. else:
  281. # _CTask is needed for tests.
  282. Task = _CTask = _asyncio.Task
  283. def create_task(coro, *, name=None):
  284. """Schedule the execution of a coroutine object in a spawn task.
  285. Return a Task object.
  286. """
  287. loop = events.get_running_loop()
  288. task = loop.create_task(coro)
  289. _set_task_name(task, name)
  290. return task
  291. # wait() and as_completed() similar to those in PEP 3148.
  292. FIRST_COMPLETED = concurrent.futures.FIRST_COMPLETED
  293. FIRST_EXCEPTION = concurrent.futures.FIRST_EXCEPTION
  294. ALL_COMPLETED = concurrent.futures.ALL_COMPLETED
  295. async def wait(fs, *, timeout=None, return_when=ALL_COMPLETED):
  296. """Wait for the Futures and coroutines given by fs to complete.
  297. The fs iterable must not be empty.
  298. Coroutines will be wrapped in Tasks.
  299. Returns two sets of Future: (done, pending).
  300. Usage:
  301. done, pending = await asyncio.wait(fs)
  302. Note: This does not raise TimeoutError! Futures that aren't done
  303. when the timeout occurs are returned in the second set.
  304. """
  305. if futures.isfuture(fs) or coroutines.iscoroutine(fs):
  306. raise TypeError(f"expect a list of futures, not {type(fs).__name__}")
  307. if not fs:
  308. raise ValueError('Set of coroutines/Futures is empty.')
  309. if return_when not in (FIRST_COMPLETED, FIRST_EXCEPTION, ALL_COMPLETED):
  310. raise ValueError(f'Invalid return_when value: {return_when}')
  311. loop = events.get_running_loop()
  312. fs = set(fs)
  313. if any(coroutines.iscoroutine(f) for f in fs):
  314. warnings.warn("The explicit passing of coroutine objects to "
  315. "asyncio.wait() is deprecated since Python 3.8, and "
  316. "scheduled for removal in Python 3.11.",
  317. DeprecationWarning, stacklevel=2)
  318. fs = {ensure_future(f, loop=loop) for f in fs}
  319. return await _wait(fs, timeout, return_when, loop)
  320. def _release_waiter(waiter, *args):
  321. if not waiter.done():
  322. waiter.set_result(None)
  323. async def wait_for(fut, timeout):
  324. """Wait for the single Future or coroutine to complete, with timeout.
  325. Coroutine will be wrapped in Task.
  326. Returns result of the Future or coroutine. When a timeout occurs,
  327. it cancels the task and raises TimeoutError. To avoid the task
  328. cancellation, wrap it in shield().
  329. If the wait is cancelled, the task is also cancelled.
  330. This function is a coroutine.
  331. """
  332. loop = events.get_running_loop()
  333. if timeout is None:
  334. return await fut
  335. if timeout <= 0:
  336. fut = ensure_future(fut, loop=loop)
  337. if fut.done():
  338. return fut.result()
  339. await _cancel_and_wait(fut, loop=loop)
  340. try:
  341. fut.result()
  342. except exceptions.CancelledError as exc:
  343. raise exceptions.TimeoutError() from exc
  344. else:
  345. raise exceptions.TimeoutError()
  346. waiter = loop.create_future()
  347. timeout_handle = loop.call_later(timeout, _release_waiter, waiter)
  348. cb = functools.partial(_release_waiter, waiter)
  349. fut = ensure_future(fut, loop=loop)
  350. fut.add_done_callback(cb)
  351. try:
  352. # wait until the future completes or the timeout
  353. try:
  354. await waiter
  355. except exceptions.CancelledError:
  356. if fut.done():
  357. return fut.result()
  358. else:
  359. fut.remove_done_callback(cb)
  360. # We must ensure that the task is not running
  361. # after wait_for() returns.
  362. # See https://bugs.python.org/issue32751
  363. await _cancel_and_wait(fut, loop=loop)
  364. raise
  365. if fut.done():
  366. return fut.result()
  367. else:
  368. fut.remove_done_callback(cb)
  369. # We must ensure that the task is not running
  370. # after wait_for() returns.
  371. # See https://bugs.python.org/issue32751
  372. await _cancel_and_wait(fut, loop=loop)
  373. # In case task cancellation failed with some
  374. # exception, we should re-raise it
  375. # See https://bugs.python.org/issue40607
  376. try:
  377. fut.result()
  378. except exceptions.CancelledError as exc:
  379. raise exceptions.TimeoutError() from exc
  380. else:
  381. raise exceptions.TimeoutError()
  382. finally:
  383. timeout_handle.cancel()
  384. async def _wait(fs, timeout, return_when, loop):
  385. """Internal helper for wait().
  386. The fs argument must be a collection of Futures.
  387. """
  388. assert fs, 'Set of Futures is empty.'
  389. waiter = loop.create_future()
  390. timeout_handle = None
  391. if timeout is not None:
  392. timeout_handle = loop.call_later(timeout, _release_waiter, waiter)
  393. counter = len(fs)
  394. def _on_completion(f):
  395. nonlocal counter
  396. counter -= 1
  397. if (counter <= 0 or
  398. return_when == FIRST_COMPLETED or
  399. return_when == FIRST_EXCEPTION and (not f.cancelled() and
  400. f.exception() is not None)):
  401. if timeout_handle is not None:
  402. timeout_handle.cancel()
  403. if not waiter.done():
  404. waiter.set_result(None)
  405. for f in fs:
  406. f.add_done_callback(_on_completion)
  407. try:
  408. await waiter
  409. finally:
  410. if timeout_handle is not None:
  411. timeout_handle.cancel()
  412. for f in fs:
  413. f.remove_done_callback(_on_completion)
  414. done, pending = set(), set()
  415. for f in fs:
  416. if f.done():
  417. done.add(f)
  418. else:
  419. pending.add(f)
  420. return done, pending
  421. async def _cancel_and_wait(fut, loop):
  422. """Cancel the *fut* future or task and wait until it completes."""
  423. waiter = loop.create_future()
  424. cb = functools.partial(_release_waiter, waiter)
  425. fut.add_done_callback(cb)
  426. try:
  427. fut.cancel()
  428. # We cannot wait on *fut* directly to make
  429. # sure _cancel_and_wait itself is reliably cancellable.
  430. await waiter
  431. finally:
  432. fut.remove_done_callback(cb)
  433. # This is *not* a @coroutine! It is just an iterator (yielding Futures).
  434. def as_completed(fs, *, timeout=None):
  435. """Return an iterator whose values are coroutines.
  436. When waiting for the yielded coroutines you'll get the results (or
  437. exceptions!) of the original Futures (or coroutines), in the order
  438. in which and as soon as they complete.
  439. This differs from PEP 3148; the proper way to use this is:
  440. for f in as_completed(fs):
  441. result = await f # The 'await' may raise.
  442. # Use result.
  443. If a timeout is specified, the 'await' will raise
  444. TimeoutError when the timeout occurs before all Futures are done.
  445. Note: The futures 'f' are not necessarily members of fs.
  446. """
  447. if futures.isfuture(fs) or coroutines.iscoroutine(fs):
  448. raise TypeError(f"expect an iterable of futures, not {type(fs).__name__}")
  449. from .queues import Queue # Import here to avoid circular import problem.
  450. done = Queue()
  451. loop = events._get_event_loop()
  452. todo = {ensure_future(f, loop=loop) for f in set(fs)}
  453. timeout_handle = None
  454. def _on_timeout():
  455. for f in todo:
  456. f.remove_done_callback(_on_completion)
  457. done.put_nowait(None) # Queue a dummy value for _wait_for_one().
  458. todo.clear() # Can't do todo.remove(f) in the loop.
  459. def _on_completion(f):
  460. if not todo:
  461. return # _on_timeout() was here first.
  462. todo.remove(f)
  463. done.put_nowait(f)
  464. if not todo and timeout_handle is not None:
  465. timeout_handle.cancel()
  466. async def _wait_for_one():
  467. f = await done.get()
  468. if f is None:
  469. # Dummy value from _on_timeout().
  470. raise exceptions.TimeoutError
  471. return f.result() # May raise f.exception().
  472. for f in todo:
  473. f.add_done_callback(_on_completion)
  474. if todo and timeout is not None:
  475. timeout_handle = loop.call_later(timeout, _on_timeout)
  476. for _ in range(len(todo)):
  477. yield _wait_for_one()
  478. @types.coroutine
  479. def __sleep0():
  480. """Skip one event loop run cycle.
  481. This is a private helper for 'asyncio.sleep()', used
  482. when the 'delay' is set to 0. It uses a bare 'yield'
  483. expression (which Task.__step knows how to handle)
  484. instead of creating a Future object.
  485. """
  486. yield
  487. async def sleep(delay, result=None):
  488. """Coroutine that completes after a given time (in seconds)."""
  489. if delay <= 0:
  490. await __sleep0()
  491. return result
  492. loop = events.get_running_loop()
  493. future = loop.create_future()
  494. h = loop.call_later(delay,
  495. futures._set_result_unless_cancelled,
  496. future, result)
  497. try:
  498. return await future
  499. finally:
  500. h.cancel()
  501. def ensure_future(coro_or_future, *, loop=None):
  502. """Wrap a coroutine or an awaitable in a future.
  503. If the argument is a Future, it is returned directly.
  504. """
  505. return _ensure_future(coro_or_future, loop=loop)
  506. def _ensure_future(coro_or_future, *, loop=None):
  507. if futures.isfuture(coro_or_future):
  508. if loop is not None and loop is not futures._get_loop(coro_or_future):
  509. raise ValueError('The future belongs to a different loop than '
  510. 'the one specified as the loop argument')
  511. return coro_or_future
  512. if not coroutines.iscoroutine(coro_or_future):
  513. if inspect.isawaitable(coro_or_future):
  514. coro_or_future = _wrap_awaitable(coro_or_future)
  515. else:
  516. raise TypeError('An asyncio.Future, a coroutine or an awaitable '
  517. 'is required')
  518. if loop is None:
  519. loop = events._get_event_loop(stacklevel=4)
  520. return loop.create_task(coro_or_future)
  521. @types.coroutine
  522. def _wrap_awaitable(awaitable):
  523. """Helper for asyncio.ensure_future().
  524. Wraps awaitable (an object with __await__) into a coroutine
  525. that will later be wrapped in a Task by ensure_future().
  526. """
  527. return (yield from awaitable.__await__())
  528. _wrap_awaitable._is_coroutine = _is_coroutine
  529. class _GatheringFuture(futures.Future):
  530. """Helper for gather().
  531. This overrides cancel() to cancel all the children and act more
  532. like Task.cancel(), which doesn't immediately mark itself as
  533. cancelled.
  534. """
  535. def __init__(self, children, *, loop):
  536. assert loop is not None
  537. super().__init__(loop=loop)
  538. self._children = children
  539. self._cancel_requested = False
  540. def cancel(self, msg=None):
  541. if self.done():
  542. return False
  543. ret = False
  544. for child in self._children:
  545. if child.cancel(msg=msg):
  546. ret = True
  547. if ret:
  548. # If any child tasks were actually cancelled, we should
  549. # propagate the cancellation request regardless of
  550. # *return_exceptions* argument. See issue 32684.
  551. self._cancel_requested = True
  552. return ret
  553. def gather(*coros_or_futures, return_exceptions=False):
  554. """Return a future aggregating results from the given coroutines/futures.
  555. Coroutines will be wrapped in a future and scheduled in the event
  556. loop. They will not necessarily be scheduled in the same order as
  557. passed in.
  558. All futures must share the same event loop. If all the tasks are
  559. done successfully, the returned future's result is the list of
  560. results (in the order of the original sequence, not necessarily
  561. the order of results arrival). If *return_exceptions* is True,
  562. exceptions in the tasks are treated the same as successful
  563. results, and gathered in the result list; otherwise, the first
  564. raised exception will be immediately propagated to the returned
  565. future.
  566. Cancellation: if the outer Future is cancelled, all children (that
  567. have not completed yet) are also cancelled. If any child is
  568. cancelled, this is treated as if it raised CancelledError --
  569. the outer Future is *not* cancelled in this case. (This is to
  570. prevent the cancellation of one child to cause other children to
  571. be cancelled.)
  572. If *return_exceptions* is False, cancelling gather() after it
  573. has been marked done won't cancel any submitted awaitables.
  574. For instance, gather can be marked done after propagating an
  575. exception to the caller, therefore, calling ``gather.cancel()``
  576. after catching an exception (raised by one of the awaitables) from
  577. gather won't cancel any other awaitables.
  578. """
  579. if not coros_or_futures:
  580. loop = events._get_event_loop()
  581. outer = loop.create_future()
  582. outer.set_result([])
  583. return outer
  584. def _done_callback(fut):
  585. nonlocal nfinished
  586. nfinished += 1
  587. if outer.done():
  588. if not fut.cancelled():
  589. # Mark exception retrieved.
  590. fut.exception()
  591. return
  592. if not return_exceptions:
  593. if fut.cancelled():
  594. # Check if 'fut' is cancelled first, as
  595. # 'fut.exception()' will *raise* a CancelledError
  596. # instead of returning it.
  597. exc = fut._make_cancelled_error()
  598. outer.set_exception(exc)
  599. return
  600. else:
  601. exc = fut.exception()
  602. if exc is not None:
  603. outer.set_exception(exc)
  604. return
  605. if nfinished == nfuts:
  606. # All futures are done; create a list of results
  607. # and set it to the 'outer' future.
  608. results = []
  609. for fut in children:
  610. if fut.cancelled():
  611. # Check if 'fut' is cancelled first, as 'fut.exception()'
  612. # will *raise* a CancelledError instead of returning it.
  613. # Also, since we're adding the exception return value
  614. # to 'results' instead of raising it, don't bother
  615. # setting __context__. This also lets us preserve
  616. # calling '_make_cancelled_error()' at most once.
  617. res = exceptions.CancelledError(
  618. '' if fut._cancel_message is None else
  619. fut._cancel_message)
  620. else:
  621. res = fut.exception()
  622. if res is None:
  623. res = fut.result()
  624. results.append(res)
  625. if outer._cancel_requested:
  626. # If gather is being cancelled we must propagate the
  627. # cancellation regardless of *return_exceptions* argument.
  628. # See issue 32684.
  629. exc = fut._make_cancelled_error()
  630. outer.set_exception(exc)
  631. else:
  632. outer.set_result(results)
  633. arg_to_fut = {}
  634. children = []
  635. nfuts = 0
  636. nfinished = 0
  637. loop = None
  638. for arg in coros_or_futures:
  639. if arg not in arg_to_fut:
  640. fut = _ensure_future(arg, loop=loop)
  641. if loop is None:
  642. loop = futures._get_loop(fut)
  643. if fut is not arg:
  644. # 'arg' was not a Future, therefore, 'fut' is a new
  645. # Future created specifically for 'arg'. Since the caller
  646. # can't control it, disable the "destroy pending task"
  647. # warning.
  648. fut._log_destroy_pending = False
  649. nfuts += 1
  650. arg_to_fut[arg] = fut
  651. fut.add_done_callback(_done_callback)
  652. else:
  653. # There's a duplicate Future object in coros_or_futures.
  654. fut = arg_to_fut[arg]
  655. children.append(fut)
  656. outer = _GatheringFuture(children, loop=loop)
  657. return outer
  658. def shield(arg):
  659. """Wait for a future, shielding it from cancellation.
  660. The statement
  661. res = await shield(something())
  662. is exactly equivalent to the statement
  663. res = await something()
  664. *except* that if the coroutine containing it is cancelled, the
  665. task running in something() is not cancelled. From the POV of
  666. something(), the cancellation did not happen. But its caller is
  667. still cancelled, so the yield-from expression still raises
  668. CancelledError. Note: If something() is cancelled by other means
  669. this will still cancel shield().
  670. If you want to completely ignore cancellation (not recommended)
  671. you can combine shield() with a try/except clause, as follows:
  672. try:
  673. res = await shield(something())
  674. except CancelledError:
  675. res = None
  676. """
  677. inner = _ensure_future(arg)
  678. if inner.done():
  679. # Shortcut.
  680. return inner
  681. loop = futures._get_loop(inner)
  682. outer = loop.create_future()
  683. def _inner_done_callback(inner):
  684. if outer.cancelled():
  685. if not inner.cancelled():
  686. # Mark inner's result as retrieved.
  687. inner.exception()
  688. return
  689. if inner.cancelled():
  690. outer.cancel()
  691. else:
  692. exc = inner.exception()
  693. if exc is not None:
  694. outer.set_exception(exc)
  695. else:
  696. outer.set_result(inner.result())
  697. def _outer_done_callback(outer):
  698. if not inner.done():
  699. inner.remove_done_callback(_inner_done_callback)
  700. inner.add_done_callback(_inner_done_callback)
  701. outer.add_done_callback(_outer_done_callback)
  702. return outer
  703. def run_coroutine_threadsafe(coro, loop):
  704. """Submit a coroutine object to a given event loop.
  705. Return a concurrent.futures.Future to access the result.
  706. """
  707. if not coroutines.iscoroutine(coro):
  708. raise TypeError('A coroutine object is required')
  709. future = concurrent.futures.Future()
  710. def callback():
  711. try:
  712. futures._chain_future(ensure_future(coro, loop=loop), future)
  713. except (SystemExit, KeyboardInterrupt):
  714. raise
  715. except BaseException as exc:
  716. if future.set_running_or_notify_cancel():
  717. future.set_exception(exc)
  718. raise
  719. loop.call_soon_threadsafe(callback)
  720. return future
  721. # WeakSet containing all alive tasks.
  722. _all_tasks = weakref.WeakSet()
  723. # Dictionary containing tasks that are currently active in
  724. # all running event loops. {EventLoop: Task}
  725. _current_tasks = {}
  726. def _register_task(task):
  727. """Register a new task in asyncio as executed by loop."""
  728. _all_tasks.add(task)
  729. def _enter_task(loop, task):
  730. current_task = _current_tasks.get(loop)
  731. if current_task is not None:
  732. raise RuntimeError(f"Cannot enter into task {task!r} while another "
  733. f"task {current_task!r} is being executed.")
  734. _current_tasks[loop] = task
  735. def _leave_task(loop, task):
  736. current_task = _current_tasks.get(loop)
  737. if current_task is not task:
  738. raise RuntimeError(f"Leaving task {task!r} does not match "
  739. f"the current task {current_task!r}.")
  740. del _current_tasks[loop]
  741. def _unregister_task(task):
  742. """Unregister a task."""
  743. _all_tasks.discard(task)
  744. _py_register_task = _register_task
  745. _py_unregister_task = _unregister_task
  746. _py_enter_task = _enter_task
  747. _py_leave_task = _leave_task
  748. try:
  749. from _asyncio import (_register_task, _unregister_task,
  750. _enter_task, _leave_task,
  751. _all_tasks, _current_tasks)
  752. except ImportError:
  753. pass
  754. else:
  755. _c_register_task = _register_task
  756. _c_unregister_task = _unregister_task
  757. _c_enter_task = _enter_task
  758. _c_leave_task = _leave_task