Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/tornado/gen.py: 24%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""``tornado.gen`` implements generator-based coroutines.
3.. note::
5 The "decorator and generator" approach in this module is a
6 precursor to native coroutines (using ``async def`` and ``await``)
7 which were introduced in Python 3.5. Applications that do not
8 require compatibility with older versions of Python should use
9 native coroutines instead. Some parts of this module are still
10 useful with native coroutines, notably `multi`, `sleep`,
11 `WaitIterator`, and `with_timeout`. Some of these functions have
12 counterparts in the `asyncio` module which may be used as well,
13 although the two may not necessarily be 100% compatible.
15Coroutines provide an easier way to work in an asynchronous
16environment than chaining callbacks. Code using coroutines is
17technically asynchronous, but it is written as a single generator
18instead of a collection of separate functions.
20For example, here's a coroutine-based handler:
22.. testcode::
24 class GenAsyncHandler(RequestHandler):
25 @gen.coroutine
26 def get(self):
27 http_client = AsyncHTTPClient()
28 response = yield http_client.fetch("http://example.com")
29 do_something_with_response(response)
30 self.render("template.html")
32Asynchronous functions in Tornado return an ``Awaitable`` or `.Future`;
33yielding this object returns its result.
35You can also yield a list or dict of other yieldable objects, which
36will be started at the same time and run in parallel; a list or dict
37of results will be returned when they are all finished:
39.. testcode::
41 @gen.coroutine
42 def get(self):
43 http_client = AsyncHTTPClient()
44 response1, response2 = yield [http_client.fetch(url1),
45 http_client.fetch(url2)]
46 response_dict = yield dict(response3=http_client.fetch(url3),
47 response4=http_client.fetch(url4))
48 response3 = response_dict['response3']
49 response4 = response_dict['response4']
51If ``tornado.platform.twisted`` is imported, it is also possible to
52yield Twisted's ``Deferred`` objects. See the `convert_yielded`
53function to extend this mechanism.
55.. versionchanged:: 3.2
56 Dict support added.
58.. versionchanged:: 4.1
59 Support added for yielding ``asyncio`` Futures and Twisted Deferreds
60 via ``singledispatch``.
62"""
64import asyncio
65import builtins
66import collections
67import concurrent.futures
68import contextvars
69import datetime
70import functools
71import sys
72import types
73import typing
74from collections.abc import Awaitable, Callable, Generator, Iterable, Mapping, Sequence
75from functools import singledispatch
76from inspect import isawaitable
77from typing import (
78 Any,
79 Dict,
80 List,
81 Tuple,
82 Type,
83 Union,
84 overload,
85)
87from tornado.concurrent import (
88 Future,
89 chain_future,
90 future_add_done_callback,
91 future_set_exc_info,
92 future_set_result_unless_cancelled,
93 is_future,
94)
95from tornado.ioloop import IOLoop
96from tornado.log import app_log
97from tornado.util import TimeoutError # noqa: F401 (re-exported for compatibility)
99_T = typing.TypeVar("_T")
101_Yieldable = Union[
102 None, Awaitable, list[Awaitable], dict[Any, Awaitable], concurrent.futures.Future
103]
106class KeyReuseError(Exception):
107 pass
110class UnknownKeyError(Exception):
111 pass
114class LeakedCallbackError(Exception):
115 pass
118class BadYieldError(Exception):
119 pass
122class ReturnValueIgnoredError(Exception):
123 pass
126def _value_from_stopiteration(e: Union[StopIteration, "Return"]) -> Any:
127 try:
128 # StopIteration has a value attribute beginning in py33.
129 # So does our Return class.
130 return e.value
131 except AttributeError:
132 pass
133 try:
134 # Cython backports coroutine functionality by putting the value in
135 # e.args[0].
136 return e.args[0]
137 except (AttributeError, IndexError):
138 return None
141def _create_future() -> Future:
142 future: Future = Future()
143 # Fixup asyncio debug info by removing extraneous stack entries
144 source_traceback = getattr(future, "_source_traceback", ())
145 while source_traceback:
146 # Each traceback entry is equivalent to a
147 # (filename, self.lineno, self.name, self.line) tuple
148 filename = source_traceback[-1][0]
149 if filename == __file__:
150 del source_traceback[-1]
151 else:
152 break
153 return future
156def _fake_ctx_run(f: Callable[..., _T], *args: Any, **kw: Any) -> _T:
157 return f(*args, **kw)
160@overload
161def coroutine(
162 func: Callable[..., "Generator[Any, Any, _T]"],
163) -> Callable[..., "Future[_T]"]: ...
166@overload
167def coroutine(func: Callable[..., _T]) -> Callable[..., "Future[_T]"]: ...
170def coroutine(
171 func: Callable[..., "Generator[Any, Any, _T]"] | Callable[..., _T],
172) -> Callable[..., "Future[_T]"]:
173 """Decorator for asynchronous generators.
175 For compatibility with older versions of Python, coroutines may
176 also "return" by raising the special exception `Return(value)
177 <Return>`.
179 Functions with this decorator return a `.Future`.
181 .. warning::
183 When exceptions occur inside a coroutine, the exception
184 information will be stored in the `.Future` object. You must
185 examine the result of the `.Future` object, or the exception
186 may go unnoticed by your code. This means yielding the function
187 if called from another coroutine, using something like
188 `.IOLoop.run_sync` for top-level calls, or passing the `.Future`
189 to `.IOLoop.add_future`.
191 .. versionchanged:: 6.0
193 The ``callback`` argument was removed. Use the returned
194 awaitable object instead.
196 .. versionchanged:: 6.6
198 If the coroutine raises `asyncio.CancelledError` (including by
199 yielding a cancelled `.Future` without catching the exception), the
200 returned `.Future` is now cancelled. Previously the coroutine would
201 never complete.
203 """
205 @functools.wraps(func)
206 def wrapper(*args: Any, **kwargs: Any) -> Future[_T]:
207 # This function is type-annotated with a comment to work around
208 # https://bitbucket.org/pypy/pypy/issues/2868/segfault-with-args-type-annotation-in
209 future = _create_future()
210 ctx_run: Callable = contextvars.copy_context().run
211 try:
212 result = ctx_run(func, *args, **kwargs)
213 except (Return, StopIteration) as e:
214 result = _value_from_stopiteration(e)
215 except asyncio.CancelledError:
216 # Like asyncio.Task, a coroutine that raises CancelledError
217 # is treated as cancelled.
218 future.cancel()
219 return future
220 except Exception:
221 future_set_exc_info(future, sys.exc_info())
222 try:
223 return future
224 finally:
225 # Avoid circular references
226 future = None # type: ignore
227 else:
228 if isinstance(result, Generator):
229 # Inline the first iteration of Runner.run. This lets us
230 # avoid the cost of creating a Runner when the coroutine
231 # never actually yields, which in turn allows us to
232 # use "optional" coroutines in critical path code without
233 # performance penalty for the synchronous case.
234 try:
235 yielded = ctx_run(next, result)
236 except (StopIteration, Return) as e:
237 future_set_result_unless_cancelled(
238 future, _value_from_stopiteration(e)
239 )
240 except asyncio.CancelledError:
241 future.cancel()
242 except Exception:
243 future_set_exc_info(future, sys.exc_info())
244 else:
245 # Provide strong references to Runner objects as long
246 # as their result future objects also have strong
247 # references (typically from the parent coroutine's
248 # Runner). This keeps the coroutine's Runner alive.
249 # We do this by exploiting the public API
250 # add_done_callback() instead of putting a private
251 # attribute on the Future.
252 # (GitHub issues #1769, #2229).
253 runner = Runner(ctx_run, result, future, yielded)
254 future.add_done_callback(lambda _: runner)
255 yielded = None
256 try:
257 return future
258 finally:
259 # Subtle memory optimization: if next() raised an exception,
260 # the future's exc_info contains a traceback which
261 # includes this stack frame. This creates a cycle,
262 # which will be collected at the next full GC but has
263 # been shown to greatly increase memory usage of
264 # benchmarks (relative to the refcount-based scheme
265 # used in the absence of cycles). We can avoid the
266 # cycle by clearing the local variable after we return it.
267 future = None # type: ignore
268 future_set_result_unless_cancelled(future, result)
269 return future
271 wrapper.__wrapped__ = func # type: ignore
272 wrapper.__tornado_coroutine__ = True # type: ignore
273 return wrapper
276def is_coroutine_function(func: Any) -> bool:
277 """Return whether *func* is a coroutine function, i.e. a function
278 wrapped with `~.gen.coroutine`.
280 .. versionadded:: 4.5
281 """
282 return getattr(func, "__tornado_coroutine__", False)
285class Return(Exception):
286 """Special exception to return a value from a `coroutine`.
288 This exception exists for compatibility with older versions of
289 Python (before 3.3). In newer code use the ``return`` statement
290 instead.
292 If this exception is raised, its value argument is used as the
293 result of the coroutine::
295 @gen.coroutine
296 def fetch_json(url):
297 response = yield AsyncHTTPClient().fetch(url)
298 raise gen.Return(json_decode(response.body))
300 By analogy with the return statement, the value argument is optional.
301 """
303 def __init__(self, value: Any = None) -> None:
304 super().__init__()
305 self.value = value
306 # Cython recognizes subclasses of StopIteration with a .args tuple.
307 self.args = (value,)
310class WaitIterator:
311 """Provides an iterator to yield the results of awaitables as they finish.
313 Yielding a set of awaitables like this:
315 ``results = yield [awaitable1, awaitable2]``
317 pauses the coroutine until both ``awaitable1`` and ``awaitable2``
318 return, and then restarts the coroutine with the results of both
319 awaitables. If either awaitable raises an exception, the
320 expression will raise that exception and all the results will be
321 lost.
323 If you need to get the result of each awaitable as soon as possible,
324 or if you need the result of some awaitables even if others produce
325 errors, you can use ``WaitIterator``::
327 wait_iterator = gen.WaitIterator(awaitable1, awaitable2)
328 while not wait_iterator.done():
329 try:
330 result = yield wait_iterator.next()
331 except Exception as e:
332 print("Error {} from {}".format(e, wait_iterator.current_future))
333 else:
334 print("Result {} received from {} at {}".format(
335 result, wait_iterator.current_future,
336 wait_iterator.current_index))
338 Because results are returned as soon as they are available the
339 output from the iterator *will not be in the same order as the
340 input arguments*. If you need to know which future produced the
341 current result, you can use the attributes
342 ``WaitIterator.current_future``, or ``WaitIterator.current_index``
343 to get the index of the awaitable from the input list. (if keyword
344 arguments were used in the construction of the `WaitIterator`,
345 ``current_index`` will use the corresponding keyword).
347 `WaitIterator` implements the async iterator
348 protocol, so it can be used with the ``async for`` statement (note
349 that in this version the entire iteration is aborted if any value
350 raises an exception, while the previous example can continue past
351 individual errors)::
353 async for result in gen.WaitIterator(future1, future2):
354 print("Result {} received from {} at {}".format(
355 result, wait_iterator.current_future,
356 wait_iterator.current_index))
358 .. versionadded:: 4.1
360 .. versionchanged:: 4.3
361 Added ``async for`` support in Python 3.5.
363 """
365 _unfinished: dict[Future, collections.deque[int | str]] = {}
367 def __init__(self, *args: Future, **kwargs: Future) -> None:
368 if args and kwargs:
369 raise ValueError("You must provide args or kwargs, not both")
371 self._unfinished = {}
372 if kwargs:
373 futures: Sequence[Future] = list(kwargs.values())
374 for key, future in kwargs.items():
375 self._unfinished.setdefault(future, collections.deque()).append(key)
376 else:
377 futures = args
378 for index, future in enumerate(args):
379 self._unfinished.setdefault(future, collections.deque()).append(index)
381 self._finished: collections.deque[Future] = collections.deque()
382 self.current_index: str | int | None = None
383 self.current_future: Future | None = None
384 self._running_future: Future | None = None
386 for future in futures:
387 future_add_done_callback(future, self._done_callback)
389 def done(self) -> bool:
390 """Returns True if this iterator has no more results."""
391 if self._finished or self._unfinished:
392 return False
393 # Clear the 'current' values when iteration is done.
394 self.current_index = self.current_future = None
395 return True
397 def next(self) -> Future:
398 """Returns a `.Future` that will yield the next available result.
400 Note that this `.Future` will not be the same object as any of
401 the inputs.
402 """
403 self._running_future = Future()
405 if self._finished:
406 return self._return_result(self._finished.popleft())
408 return self._running_future
410 def _done_callback(self, done: Future) -> None:
411 if self._running_future and not self._running_future.done():
412 self._return_result(done)
413 else:
414 self._finished.append(done)
416 def _return_result(self, done: Future) -> Future:
417 """Called set the returned future's state that of the future
418 we yielded, and set the current future for the iterator.
419 """
420 if self._running_future is None:
421 raise Exception("no future is running")
422 chain_future(done, self._running_future)
424 res = self._running_future
425 self._running_future = None
426 self.current_future = done
427 indices = self._unfinished[done]
428 self.current_index = indices.popleft()
429 if not indices:
430 del self._unfinished[done]
432 return res
434 def __aiter__(self) -> typing.AsyncIterator:
435 return self
437 def __anext__(self) -> Future:
438 if self.done():
439 # Lookup by name to silence pyflakes on older versions.
440 raise getattr(builtins, "StopAsyncIteration")()
441 return self.next()
444@overload
445def multi(
446 children: Sequence[_Yieldable],
447 quiet_exceptions: type[Exception] | tuple[type[Exception], ...] = (),
448) -> Future[list]: ...
451@overload
452def multi(
453 children: Mapping[Any, _Yieldable],
454 quiet_exceptions: type[Exception] | tuple[type[Exception], ...] = (),
455) -> Future[dict]: ...
458def multi(
459 children: Sequence[_Yieldable] | Mapping[Any, _Yieldable],
460 quiet_exceptions: "Union[Type[Exception], Tuple[Type[Exception], ...]]" = (),
461) -> "Union[Future[List], Future[Dict]]":
462 """Runs multiple asynchronous operations in parallel.
464 ``children`` may either be a list or a dict whose values are
465 yieldable objects. ``multi()`` returns a new yieldable
466 object that resolves to a parallel structure containing their
467 results. If ``children`` is a list, the result is a list of
468 results in the same order; if it is a dict, the result is a dict
469 with the same keys.
471 That is, ``results = yield multi(list_of_futures)`` is equivalent
472 to::
474 results = []
475 for future in list_of_futures:
476 results.append(yield future)
478 If any children raise exceptions, ``multi()`` will raise the first
479 one. All others will be logged, unless they are of types
480 contained in the ``quiet_exceptions`` argument.
482 In a ``yield``-based coroutine, it is not normally necessary to
483 call this function directly, since the coroutine runner will
484 do it automatically when a list or dict is yielded. However,
485 it is necessary in ``await``-based coroutines, or to pass
486 the ``quiet_exceptions`` argument.
488 This function is available under the names ``multi()`` and ``Multi()``
489 for historical reasons.
491 Cancelling a `.Future` returned by ``multi()`` does not cancel its
492 children. `asyncio.gather` is similar to ``multi()``, but it does
493 cancel its children.
495 .. versionchanged:: 4.2
496 If multiple yieldables fail, any exceptions after the first
497 (which is raised) will be logged. Added the ``quiet_exceptions``
498 argument to suppress this logging for selected exception types.
500 .. versionchanged:: 4.3
501 Replaced the class ``Multi`` and the function ``multi_future``
502 with a unified function ``multi``. Added support for yieldables
503 other than ``YieldPoint`` and `.Future`.
505 .. versionchanged:: 6.6
506 If a child is cancelled, ``multi()`` now raises `asyncio.CancelledError`
507 (without logging it if it is not the first failure), matching
508 `asyncio.gather`. Previously ``multi()`` would never complete.
510 """
511 return multi_future(children, quiet_exceptions=quiet_exceptions)
514Multi = multi
517def multi_future(
518 children: Sequence[_Yieldable] | Mapping[Any, _Yieldable],
519 quiet_exceptions: "Union[Type[Exception], Tuple[Type[Exception], ...]]" = (),
520) -> "Union[Future[List], Future[Dict]]":
521 """Wait for multiple asynchronous futures in parallel.
523 Since Tornado 6.0, this function is exactly the same as `multi`.
525 .. versionadded:: 4.0
527 .. versionchanged:: 4.2
528 If multiple ``Futures`` fail, any exceptions after the first (which is
529 raised) will be logged. Added the ``quiet_exceptions``
530 argument to suppress this logging for selected exception types.
532 .. deprecated:: 4.3
533 Use `multi` instead.
534 """
535 if isinstance(children, dict):
536 keys: list | None = list(children.keys())
537 children_seq: Iterable = children.values()
538 else:
539 keys = None
540 children_seq = children
541 children_futs = list(map(convert_yielded, children_seq))
542 assert all(is_future(i) or isinstance(i, _NullFuture) for i in children_futs)
543 unfinished_children = set(children_futs)
545 future = _create_future()
546 if not children_futs:
547 future_set_result_unless_cancelled(future, {} if keys is not None else [])
549 def callback(fut: Future) -> None:
550 unfinished_children.remove(fut)
551 if not unfinished_children:
552 result_list = []
553 for f in children_futs:
554 try:
555 result_list.append(f.result())
556 except (Exception, asyncio.CancelledError) as e:
557 # As in asyncio.gather, a cancelled child is treated as
558 # an error, and does not cancel the combined future.
559 if future.done():
560 if not isinstance(
561 e, (asyncio.CancelledError, quiet_exceptions)
562 ):
563 app_log.error(
564 "Multiple exceptions in yield list", exc_info=True
565 )
566 else:
567 future_set_exc_info(future, sys.exc_info())
568 if not future.done():
569 if keys is not None:
570 future_set_result_unless_cancelled(
571 future, dict(zip(keys, result_list))
572 )
573 else:
574 future_set_result_unless_cancelled(future, result_list)
576 listening: set[Future] = set()
577 for f in children_futs:
578 if f not in listening:
579 listening.add(f)
580 future_add_done_callback(f, callback)
581 return future
584def maybe_future(x: Any) -> Future:
585 """Converts ``x`` into a `.Future`.
587 If ``x`` is already a `.Future`, it is simply returned; otherwise
588 it is wrapped in a new `.Future`. This is suitable for use as
589 ``result = yield gen.maybe_future(f())`` when you don't know whether
590 ``f()`` returns a `.Future` or not.
592 .. deprecated:: 4.3
593 This function only handles ``Futures``, not other yieldable objects.
594 Instead of `maybe_future`, check for the non-future result types
595 you expect (often just ``None``), and ``yield`` anything unknown.
596 """
597 if is_future(x):
598 return x
599 else:
600 fut = _create_future()
601 fut.set_result(x)
602 return fut
605def with_timeout(
606 timeout: float | datetime.timedelta,
607 future: _Yieldable,
608 quiet_exceptions: "Union[Type[Exception], Tuple[Type[Exception], ...]]" = (),
609 *,
610 message: str | Callable[[], str] | None = None,
611) -> Future:
612 """Wraps a `.Future` (or other yieldable object) in a timeout.
614 Raises `TimeoutError` if the input future does not
615 complete before ``timeout``, which may be specified in any form
616 allowed by `.IOLoop.add_timeout` (i.e. a `datetime.timedelta` or
617 an absolute time relative to `.IOLoop.time`)
619 ``message`` is used as the message of the `TimeoutError`. It may be
620 a string or a callable returning a string; a callable is only called
621 if the timeout expires, so it may be used to defer the cost of
622 formatting the message.
624 If the wrapped `.Future` fails after it has timed out, the exception
625 will be logged unless it is either of a type contained in
626 ``quiet_exceptions`` (which may be an exception type or a sequence of
627 types), or an ``asyncio.CancelledError``.
629 The wrapped `.Future` is not canceled when the timeout expires,
630 permitting it to be reused. `asyncio.wait_for` is similar to this
631 function but it does cancel the wrapped `.Future` on timeout.
633 .. versionadded:: 4.0
635 .. versionchanged:: 4.1
636 Added the ``quiet_exceptions`` argument and the logging of unhandled
637 exceptions.
639 .. versionchanged:: 4.4
640 Added support for yieldable objects other than `.Future`.
642 .. versionchanged:: 6.0.3
643 ``asyncio.CancelledError`` is now always considered "quiet".
645 .. versionchanged:: 6.2
646 ``tornado.util.TimeoutError`` is now an alias to ``asyncio.TimeoutError``.
648 .. versionchanged:: 6.6
649 Added the ``message`` argument.
651 """
652 # It's tempting to optimize this by cancelling the input future on timeout
653 # instead of creating a new one, but A) we can't know if we are the only
654 # one waiting on the input future, so cancelling it might disrupt other
655 # callers and B) concurrent futures can only be cancelled while they are
656 # in the queue, so cancellation cannot reliably bound our waiting time.
657 future_converted = convert_yielded(future)
658 result = _create_future()
659 chain_future(future_converted, result)
660 io_loop = IOLoop.current()
662 def error_callback(future: Future) -> None:
663 try:
664 future.result()
665 except asyncio.CancelledError:
666 pass
667 except Exception as e:
668 if not isinstance(e, quiet_exceptions):
669 app_log.error(
670 "Exception in Future %r after timeout", future, exc_info=True
671 )
673 def timeout_callback() -> None:
674 if not result.done():
675 if message is None:
676 msg = "Timeout"
677 elif callable(message):
678 msg = message()
679 else:
680 msg = message
681 result.set_exception(TimeoutError(msg))
682 # In case the wrapped future goes on to fail, log it.
683 future_add_done_callback(future_converted, error_callback)
685 timeout_handle = io_loop.add_timeout(timeout, timeout_callback)
686 if isinstance(future_converted, Future):
687 # We know this future will resolve on the IOLoop, so we don't
688 # need the extra thread-safety of IOLoop.add_future (and we also
689 # don't care about StackContext here.
690 future_add_done_callback(
691 future_converted, lambda future: io_loop.remove_timeout(timeout_handle)
692 )
693 else:
694 # concurrent.futures.Futures may resolve on any thread, so we
695 # need to route them back to the IOLoop.
696 io_loop.add_future(
697 future_converted, lambda future: io_loop.remove_timeout(timeout_handle)
698 )
699 return result
702def sleep(duration: float) -> "Future[None]":
703 """Return a `.Future` that resolves after the given number of seconds.
705 When used with ``yield`` in a coroutine, this is a non-blocking
706 analogue to `time.sleep` (which should not be used in coroutines
707 because it is blocking)::
709 yield gen.sleep(0.5)
711 Note that calling this function on its own does nothing; you must
712 wait on the `.Future` it returns (usually by yielding it).
714 .. versionadded:: 4.1
715 """
716 f = _create_future()
717 IOLoop.current().call_later(
718 duration, lambda: future_set_result_unless_cancelled(f, None)
719 )
720 return f
723class _NullFuture:
724 """_NullFuture resembles a Future that finished with a result of None.
726 It's not actually a `Future` to avoid depending on a particular event loop.
727 Handled as a special case in the coroutine runner.
729 We lie and tell the type checker that a _NullFuture is a Future so
730 we don't have to leak _NullFuture into lots of public APIs. But
731 this means that the type checker can't warn us when we're passing
732 a _NullFuture into a code path that doesn't understand what to do
733 with it.
734 """
736 def result(self) -> None:
737 return None
739 def done(self) -> bool:
740 return True
743# _null_future is used as a dummy value in the coroutine runner. It differs
744# from moment in that moment always adds a delay of one IOLoop iteration
745# while _null_future is processed as soon as possible.
746_null_future = typing.cast(Future, _NullFuture())
748moment = typing.cast(Future, _NullFuture())
749moment.__doc__ = """A special object which may be yielded to allow the IOLoop to run for
750one iteration.
752This is not needed in normal use but it can be helpful in long-running
753coroutines that are likely to yield Futures that are ready instantly.
755Usage: ``yield gen.moment``
757In native coroutines, the equivalent of ``yield gen.moment`` is
758``await asyncio.sleep(0)``.
760.. versionadded:: 4.0
762.. deprecated:: 4.5
763 ``yield None`` (or ``yield`` with no argument) is now equivalent to
764 ``yield gen.moment``.
765"""
768class Runner:
769 """Internal implementation of `tornado.gen.coroutine`.
771 Maintains information about pending callbacks and their results.
773 The results of the generator are stored in ``result_future`` (a
774 `.Future`)
775 """
777 def __init__(
778 self,
779 ctx_run: Callable,
780 gen: "Generator[_Yieldable, Any, _T]",
781 result_future: "Future[_T]",
782 first_yielded: _Yieldable,
783 ) -> None:
784 self.ctx_run = ctx_run
785 self.gen = gen
786 self.result_future = result_future
787 self.future: None | Future = _null_future
788 self.running = False
789 self.finished = False
790 self.io_loop = IOLoop.current()
791 if self.ctx_run(self.handle_yield, first_yielded):
792 gen = result_future = first_yielded = None # type: ignore
793 self.ctx_run(self.run)
795 def run(self) -> None:
796 """Starts or resumes the generator, running until it reaches a
797 yield point that is not ready.
798 """
799 if self.running or self.finished:
800 return
801 try:
802 self.running = True
803 while True:
804 future = self.future
805 if future is None:
806 raise Exception("No pending future")
807 if not future.done():
808 return
809 self.future = None
810 try:
811 try:
812 value = future.result()
813 except (Exception, asyncio.CancelledError) as e:
814 # Save the exception for later. It's important that
815 # gen.throw() not be called inside this try/except block
816 # because that makes sys.exc_info behave unexpectedly.
817 exc: BaseException | None = e
818 else:
819 exc = None
820 finally:
821 future = None
823 if exc is not None:
824 try:
825 yielded = self.gen.throw(exc)
826 finally:
827 # Break up a circular reference for faster GC on
828 # CPython.
829 del exc
830 else:
831 yielded = self.gen.send(value)
833 except (StopIteration, Return) as e:
834 self.finished = True
835 self.future = _null_future
836 future_set_result_unless_cancelled(
837 self.result_future, _value_from_stopiteration(e)
838 )
839 self.result_future = None # type: ignore
840 return
841 except asyncio.CancelledError:
842 self.finished = True
843 self.future = _null_future
844 self.result_future.cancel()
845 self.result_future = None # type: ignore
846 return
847 except Exception:
848 self.finished = True
849 self.future = _null_future
850 future_set_exc_info(self.result_future, sys.exc_info())
851 self.result_future = None # type: ignore
852 return
853 if not self.handle_yield(yielded):
854 return
855 yielded = None
856 finally:
857 self.running = False
859 def handle_yield(self, yielded: _Yieldable) -> bool:
860 try:
861 self.future = convert_yielded(yielded)
862 except BadYieldError:
863 self.future = Future()
864 future_set_exc_info(self.future, sys.exc_info())
866 if self.future is moment:
867 self.io_loop.add_callback(self.ctx_run, self.run)
868 return False
869 elif self.future is None:
870 raise Exception("no pending future")
871 elif not self.future.done():
873 def inner(f: Any) -> None:
874 # Break a reference cycle to speed GC.
875 f = None # noqa: F841
876 self.ctx_run(self.run)
878 self.io_loop.add_future(self.future, inner)
879 return False
880 return True
882 def handle_exception(
883 self, typ: type[Exception], value: Exception, tb: types.TracebackType
884 ) -> bool:
885 if not self.running and not self.finished:
886 self.future = Future()
887 future_set_exc_info(self.future, (typ, value, tb))
888 self.ctx_run(self.run)
889 return True
890 else:
891 return False
894def _wrap_awaitable(awaitable: Awaitable) -> Future:
895 # Convert Awaitables into Futures.
896 # Note that we use ensure_future, which handles both awaitables
897 # and coroutines, rather than create_task, which only accepts
898 # coroutines. (ensure_future calls create_task if given a coroutine)
899 fut = asyncio.ensure_future(awaitable)
900 # See comments on IOLoop._pending_tasks.
901 loop = IOLoop.current()
902 loop._register_task(fut)
903 fut.add_done_callback(lambda f: loop._unregister_task(f))
904 return fut
907def convert_yielded(yielded: _Yieldable) -> Future:
908 """Convert a yielded object into a `.Future`.
910 The default implementation accepts lists, dictionaries, and
911 Futures. This has the side effect of starting any coroutines that
912 did not start themselves, similar to `asyncio.ensure_future`.
914 If the `~functools.singledispatch` library is available, this function
915 may be extended to support additional types. For example::
917 @convert_yielded.register(asyncio.Future)
918 def _(asyncio_future):
919 return tornado.platform.asyncio.to_tornado_future(asyncio_future)
921 .. versionadded:: 4.1
923 """
924 if yielded is None or yielded is moment:
925 return moment
926 elif yielded is _null_future:
927 return _null_future
928 elif isinstance(yielded, (list, dict)):
929 return multi(yielded) # type: ignore
930 elif is_future(yielded):
931 return typing.cast(Future, yielded)
932 elif isawaitable(yielded):
933 return _wrap_awaitable(yielded) # type: ignore
934 else:
935 raise BadYieldError(f"yielded unknown object {yielded!r}")
938convert_yielded = singledispatch(convert_yielded)