Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/tornado/gen.py: 25%
317 statements
« prev ^ index » next coverage.py v7.2.7, created at 2023-07-01 06:54 +0000
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")
32.. testoutput::
33 :hide:
35Asynchronous functions in Tornado return an ``Awaitable`` or `.Future`;
36yielding this object returns its result.
38You can also yield a list or dict of other yieldable objects, which
39will be started at the same time and run in parallel; a list or dict
40of results will be returned when they are all finished:
42.. testcode::
44 @gen.coroutine
45 def get(self):
46 http_client = AsyncHTTPClient()
47 response1, response2 = yield [http_client.fetch(url1),
48 http_client.fetch(url2)]
49 response_dict = yield dict(response3=http_client.fetch(url3),
50 response4=http_client.fetch(url4))
51 response3 = response_dict['response3']
52 response4 = response_dict['response4']
54.. testoutput::
55 :hide:
57If ``tornado.platform.twisted`` is imported, it is also possible to
58yield Twisted's ``Deferred`` objects. See the `convert_yielded`
59function to extend this mechanism.
61.. versionchanged:: 3.2
62 Dict support added.
64.. versionchanged:: 4.1
65 Support added for yielding ``asyncio`` Futures and Twisted Deferreds
66 via ``singledispatch``.
68"""
69import asyncio
70import builtins
71import collections
72from collections.abc import Generator
73import concurrent.futures
74import datetime
75import functools
76from functools import singledispatch
77from inspect import isawaitable
78import sys
79import types
81from tornado.concurrent import (
82 Future,
83 is_future,
84 chain_future,
85 future_set_exc_info,
86 future_add_done_callback,
87 future_set_result_unless_cancelled,
88)
89from tornado.ioloop import IOLoop
90from tornado.log import app_log
91from tornado.util import TimeoutError
93try:
94 import contextvars
95except ImportError:
96 contextvars = None # type: ignore
98import typing
99from typing import Union, Any, Callable, List, Type, Tuple, Awaitable, Dict, overload
101if typing.TYPE_CHECKING:
102 from typing import Sequence, Deque, Optional, Set, Iterable # noqa: F401
104_T = typing.TypeVar("_T")
106_Yieldable = Union[
107 None, Awaitable, List[Awaitable], Dict[Any, Awaitable], concurrent.futures.Future
108]
111class KeyReuseError(Exception):
112 pass
115class UnknownKeyError(Exception):
116 pass
119class LeakedCallbackError(Exception):
120 pass
123class BadYieldError(Exception):
124 pass
127class ReturnValueIgnoredError(Exception):
128 pass
131def _value_from_stopiteration(e: Union[StopIteration, "Return"]) -> Any:
132 try:
133 # StopIteration has a value attribute beginning in py33.
134 # So does our Return class.
135 return e.value
136 except AttributeError:
137 pass
138 try:
139 # Cython backports coroutine functionality by putting the value in
140 # e.args[0].
141 return e.args[0]
142 except (AttributeError, IndexError):
143 return None
146def _create_future() -> Future:
147 future = Future() # type: Future
148 # Fixup asyncio debug info by removing extraneous stack entries
149 source_traceback = getattr(future, "_source_traceback", ())
150 while source_traceback:
151 # Each traceback entry is equivalent to a
152 # (filename, self.lineno, self.name, self.line) tuple
153 filename = source_traceback[-1][0]
154 if filename == __file__:
155 del source_traceback[-1]
156 else:
157 break
158 return future
161def _fake_ctx_run(f: Callable[..., _T], *args: Any, **kw: Any) -> _T:
162 return f(*args, **kw)
165@overload
166def coroutine(
167 func: Callable[..., "Generator[Any, Any, _T]"]
168) -> Callable[..., "Future[_T]"]:
169 ...
172@overload
173def coroutine(func: Callable[..., _T]) -> Callable[..., "Future[_T]"]:
174 ...
177def coroutine(
178 func: Union[Callable[..., "Generator[Any, Any, _T]"], Callable[..., _T]]
179) -> Callable[..., "Future[_T]"]:
180 """Decorator for asynchronous generators.
182 For compatibility with older versions of Python, coroutines may
183 also "return" by raising the special exception `Return(value)
184 <Return>`.
186 Functions with this decorator return a `.Future`.
188 .. warning::
190 When exceptions occur inside a coroutine, the exception
191 information will be stored in the `.Future` object. You must
192 examine the result of the `.Future` object, or the exception
193 may go unnoticed by your code. This means yielding the function
194 if called from another coroutine, using something like
195 `.IOLoop.run_sync` for top-level calls, or passing the `.Future`
196 to `.IOLoop.add_future`.
198 .. versionchanged:: 6.0
200 The ``callback`` argument was removed. Use the returned
201 awaitable object instead.
203 """
205 @functools.wraps(func)
206 def wrapper(*args, **kwargs):
207 # type: (*Any, **Any) -> Future[_T]
208 # This function is type-annotated with a comment to work around
209 # https://bitbucket.org/pypy/pypy/issues/2868/segfault-with-args-type-annotation-in
210 future = _create_future()
211 if contextvars is not None:
212 ctx_run = contextvars.copy_context().run # type: Callable
213 else:
214 ctx_run = _fake_ctx_run
215 try:
216 result = ctx_run(func, *args, **kwargs)
217 except (Return, StopIteration) as e:
218 result = _value_from_stopiteration(e)
219 except Exception:
220 future_set_exc_info(future, sys.exc_info())
221 try:
222 return future
223 finally:
224 # Avoid circular references
225 future = None # type: ignore
226 else:
227 if isinstance(result, Generator):
228 # Inline the first iteration of Runner.run. This lets us
229 # avoid the cost of creating a Runner when the coroutine
230 # never actually yields, which in turn allows us to
231 # use "optional" coroutines in critical path code without
232 # performance penalty for the synchronous case.
233 try:
234 yielded = ctx_run(next, result)
235 except (StopIteration, Return) as e:
236 future_set_result_unless_cancelled(
237 future, _value_from_stopiteration(e)
238 )
239 except Exception:
240 future_set_exc_info(future, sys.exc_info())
241 else:
242 # Provide strong references to Runner objects as long
243 # as their result future objects also have strong
244 # references (typically from the parent coroutine's
245 # Runner). This keeps the coroutine's Runner alive.
246 # We do this by exploiting the public API
247 # add_done_callback() instead of putting a private
248 # attribute on the Future.
249 # (GitHub issues #1769, #2229).
250 runner = Runner(ctx_run, result, future, yielded)
251 future.add_done_callback(lambda _: runner)
252 yielded = None
253 try:
254 return future
255 finally:
256 # Subtle memory optimization: if next() raised an exception,
257 # the future's exc_info contains a traceback which
258 # includes this stack frame. This creates a cycle,
259 # which will be collected at the next full GC but has
260 # been shown to greatly increase memory usage of
261 # benchmarks (relative to the refcount-based scheme
262 # used in the absence of cycles). We can avoid the
263 # cycle by clearing the local variable after we return it.
264 future = None # type: ignore
265 future_set_result_unless_cancelled(future, result)
266 return future
268 wrapper.__wrapped__ = func # type: ignore
269 wrapper.__tornado_coroutine__ = True # type: ignore
270 return wrapper
273def is_coroutine_function(func: Any) -> bool:
274 """Return whether *func* is a coroutine function, i.e. a function
275 wrapped with `~.gen.coroutine`.
277 .. versionadded:: 4.5
278 """
279 return getattr(func, "__tornado_coroutine__", False)
282class Return(Exception):
283 """Special exception to return a value from a `coroutine`.
285 If this exception is raised, its value argument is used as the
286 result of the coroutine::
288 @gen.coroutine
289 def fetch_json(url):
290 response = yield AsyncHTTPClient().fetch(url)
291 raise gen.Return(json_decode(response.body))
293 In Python 3.3, this exception is no longer necessary: the ``return``
294 statement can be used directly to return a value (previously
295 ``yield`` and ``return`` with a value could not be combined in the
296 same function).
298 By analogy with the return statement, the value argument is optional,
299 but it is never necessary to ``raise gen.Return()``. The ``return``
300 statement can be used with no arguments instead.
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(object):
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 On Python 3.5, `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 = {} # type: Dict[Future, Union[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 if kwargs:
372 self._unfinished = dict((f, k) for (k, f) in kwargs.items())
373 futures = list(kwargs.values()) # type: Sequence[Future]
374 else:
375 self._unfinished = dict((f, i) for (i, f) in enumerate(args))
376 futures = args
378 self._finished = collections.deque() # type: Deque[Future]
379 self.current_index = None # type: Optional[Union[str, int]]
380 self.current_future = None # type: Optional[Future]
381 self._running_future = None # type: Optional[Future]
383 for future in futures:
384 future_add_done_callback(future, self._done_callback)
386 def done(self) -> bool:
387 """Returns True if this iterator has no more results."""
388 if self._finished or self._unfinished:
389 return False
390 # Clear the 'current' values when iteration is done.
391 self.current_index = self.current_future = None
392 return True
394 def next(self) -> Future:
395 """Returns a `.Future` that will yield the next available result.
397 Note that this `.Future` will not be the same object as any of
398 the inputs.
399 """
400 self._running_future = Future()
402 if self._finished:
403 return self._return_result(self._finished.popleft())
405 return self._running_future
407 def _done_callback(self, done: Future) -> None:
408 if self._running_future and not self._running_future.done():
409 self._return_result(done)
410 else:
411 self._finished.append(done)
413 def _return_result(self, done: Future) -> Future:
414 """Called set the returned future's state that of the future
415 we yielded, and set the current future for the iterator.
416 """
417 if self._running_future is None:
418 raise Exception("no future is running")
419 chain_future(done, self._running_future)
421 res = self._running_future
422 self._running_future = None
423 self.current_future = done
424 self.current_index = self._unfinished.pop(done)
426 return res
428 def __aiter__(self) -> typing.AsyncIterator:
429 return self
431 def __anext__(self) -> Future:
432 if self.done():
433 # Lookup by name to silence pyflakes on older versions.
434 raise getattr(builtins, "StopAsyncIteration")()
435 return self.next()
438def multi(
439 children: Union[List[_Yieldable], Dict[Any, _Yieldable]],
440 quiet_exceptions: "Union[Type[Exception], Tuple[Type[Exception], ...]]" = (),
441) -> "Union[Future[List], Future[Dict]]":
442 """Runs multiple asynchronous operations in parallel.
444 ``children`` may either be a list or a dict whose values are
445 yieldable objects. ``multi()`` returns a new yieldable
446 object that resolves to a parallel structure containing their
447 results. If ``children`` is a list, the result is a list of
448 results in the same order; if it is a dict, the result is a dict
449 with the same keys.
451 That is, ``results = yield multi(list_of_futures)`` is equivalent
452 to::
454 results = []
455 for future in list_of_futures:
456 results.append(yield future)
458 If any children raise exceptions, ``multi()`` will raise the first
459 one. All others will be logged, unless they are of types
460 contained in the ``quiet_exceptions`` argument.
462 In a ``yield``-based coroutine, it is not normally necessary to
463 call this function directly, since the coroutine runner will
464 do it automatically when a list or dict is yielded. However,
465 it is necessary in ``await``-based coroutines, or to pass
466 the ``quiet_exceptions`` argument.
468 This function is available under the names ``multi()`` and ``Multi()``
469 for historical reasons.
471 Cancelling a `.Future` returned by ``multi()`` does not cancel its
472 children. `asyncio.gather` is similar to ``multi()``, but it does
473 cancel its children.
475 .. versionchanged:: 4.2
476 If multiple yieldables fail, any exceptions after the first
477 (which is raised) will be logged. Added the ``quiet_exceptions``
478 argument to suppress this logging for selected exception types.
480 .. versionchanged:: 4.3
481 Replaced the class ``Multi`` and the function ``multi_future``
482 with a unified function ``multi``. Added support for yieldables
483 other than ``YieldPoint`` and `.Future`.
485 """
486 return multi_future(children, quiet_exceptions=quiet_exceptions)
489Multi = multi
492def multi_future(
493 children: Union[List[_Yieldable], Dict[Any, _Yieldable]],
494 quiet_exceptions: "Union[Type[Exception], Tuple[Type[Exception], ...]]" = (),
495) -> "Union[Future[List], Future[Dict]]":
496 """Wait for multiple asynchronous futures in parallel.
498 Since Tornado 6.0, this function is exactly the same as `multi`.
500 .. versionadded:: 4.0
502 .. versionchanged:: 4.2
503 If multiple ``Futures`` fail, any exceptions after the first (which is
504 raised) will be logged. Added the ``quiet_exceptions``
505 argument to suppress this logging for selected exception types.
507 .. deprecated:: 4.3
508 Use `multi` instead.
509 """
510 if isinstance(children, dict):
511 keys = list(children.keys()) # type: Optional[List]
512 children_seq = children.values() # type: Iterable
513 else:
514 keys = None
515 children_seq = children
516 children_futs = list(map(convert_yielded, children_seq))
517 assert all(is_future(i) or isinstance(i, _NullFuture) for i in children_futs)
518 unfinished_children = set(children_futs)
520 future = _create_future()
521 if not children_futs:
522 future_set_result_unless_cancelled(future, {} if keys is not None else [])
524 def callback(fut: Future) -> None:
525 unfinished_children.remove(fut)
526 if not unfinished_children:
527 result_list = []
528 for f in children_futs:
529 try:
530 result_list.append(f.result())
531 except Exception as e:
532 if future.done():
533 if not isinstance(e, quiet_exceptions):
534 app_log.error(
535 "Multiple exceptions in yield list", exc_info=True
536 )
537 else:
538 future_set_exc_info(future, sys.exc_info())
539 if not future.done():
540 if keys is not None:
541 future_set_result_unless_cancelled(
542 future, dict(zip(keys, result_list))
543 )
544 else:
545 future_set_result_unless_cancelled(future, result_list)
547 listening = set() # type: Set[Future]
548 for f in children_futs:
549 if f not in listening:
550 listening.add(f)
551 future_add_done_callback(f, callback)
552 return future
555def maybe_future(x: Any) -> Future:
556 """Converts ``x`` into a `.Future`.
558 If ``x`` is already a `.Future`, it is simply returned; otherwise
559 it is wrapped in a new `.Future`. This is suitable for use as
560 ``result = yield gen.maybe_future(f())`` when you don't know whether
561 ``f()`` returns a `.Future` or not.
563 .. deprecated:: 4.3
564 This function only handles ``Futures``, not other yieldable objects.
565 Instead of `maybe_future`, check for the non-future result types
566 you expect (often just ``None``), and ``yield`` anything unknown.
567 """
568 if is_future(x):
569 return x
570 else:
571 fut = _create_future()
572 fut.set_result(x)
573 return fut
576def with_timeout(
577 timeout: Union[float, datetime.timedelta],
578 future: _Yieldable,
579 quiet_exceptions: "Union[Type[Exception], Tuple[Type[Exception], ...]]" = (),
580) -> Future:
581 """Wraps a `.Future` (or other yieldable object) in a timeout.
583 Raises `tornado.util.TimeoutError` if the input future does not
584 complete before ``timeout``, which may be specified in any form
585 allowed by `.IOLoop.add_timeout` (i.e. a `datetime.timedelta` or
586 an absolute time relative to `.IOLoop.time`)
588 If the wrapped `.Future` fails after it has timed out, the exception
589 will be logged unless it is either of a type contained in
590 ``quiet_exceptions`` (which may be an exception type or a sequence of
591 types), or an ``asyncio.CancelledError``.
593 The wrapped `.Future` is not canceled when the timeout expires,
594 permitting it to be reused. `asyncio.wait_for` is similar to this
595 function but it does cancel the wrapped `.Future` on timeout.
597 .. versionadded:: 4.0
599 .. versionchanged:: 4.1
600 Added the ``quiet_exceptions`` argument and the logging of unhandled
601 exceptions.
603 .. versionchanged:: 4.4
604 Added support for yieldable objects other than `.Future`.
606 .. versionchanged:: 6.0.3
607 ``asyncio.CancelledError`` is now always considered "quiet".
609 .. versionchanged:: 6.2
610 ``tornado.util.TimeoutError`` is now an alias to ``asyncio.TimeoutError``.
612 """
613 # It's tempting to optimize this by cancelling the input future on timeout
614 # instead of creating a new one, but A) we can't know if we are the only
615 # one waiting on the input future, so cancelling it might disrupt other
616 # callers and B) concurrent futures can only be cancelled while they are
617 # in the queue, so cancellation cannot reliably bound our waiting time.
618 future_converted = convert_yielded(future)
619 result = _create_future()
620 chain_future(future_converted, result)
621 io_loop = IOLoop.current()
623 def error_callback(future: Future) -> None:
624 try:
625 future.result()
626 except asyncio.CancelledError:
627 pass
628 except Exception as e:
629 if not isinstance(e, quiet_exceptions):
630 app_log.error(
631 "Exception in Future %r after timeout", future, exc_info=True
632 )
634 def timeout_callback() -> None:
635 if not result.done():
636 result.set_exception(TimeoutError("Timeout"))
637 # In case the wrapped future goes on to fail, log it.
638 future_add_done_callback(future_converted, error_callback)
640 timeout_handle = io_loop.add_timeout(timeout, timeout_callback)
641 if isinstance(future_converted, Future):
642 # We know this future will resolve on the IOLoop, so we don't
643 # need the extra thread-safety of IOLoop.add_future (and we also
644 # don't care about StackContext here.
645 future_add_done_callback(
646 future_converted, lambda future: io_loop.remove_timeout(timeout_handle)
647 )
648 else:
649 # concurrent.futures.Futures may resolve on any thread, so we
650 # need to route them back to the IOLoop.
651 io_loop.add_future(
652 future_converted, lambda future: io_loop.remove_timeout(timeout_handle)
653 )
654 return result
657def sleep(duration: float) -> "Future[None]":
658 """Return a `.Future` that resolves after the given number of seconds.
660 When used with ``yield`` in a coroutine, this is a non-blocking
661 analogue to `time.sleep` (which should not be used in coroutines
662 because it is blocking)::
664 yield gen.sleep(0.5)
666 Note that calling this function on its own does nothing; you must
667 wait on the `.Future` it returns (usually by yielding it).
669 .. versionadded:: 4.1
670 """
671 f = _create_future()
672 IOLoop.current().call_later(
673 duration, lambda: future_set_result_unless_cancelled(f, None)
674 )
675 return f
678class _NullFuture(object):
679 """_NullFuture resembles a Future that finished with a result of None.
681 It's not actually a `Future` to avoid depending on a particular event loop.
682 Handled as a special case in the coroutine runner.
684 We lie and tell the type checker that a _NullFuture is a Future so
685 we don't have to leak _NullFuture into lots of public APIs. But
686 this means that the type checker can't warn us when we're passing
687 a _NullFuture into a code path that doesn't understand what to do
688 with it.
689 """
691 def result(self) -> None:
692 return None
694 def done(self) -> bool:
695 return True
698# _null_future is used as a dummy value in the coroutine runner. It differs
699# from moment in that moment always adds a delay of one IOLoop iteration
700# while _null_future is processed as soon as possible.
701_null_future = typing.cast(Future, _NullFuture())
703moment = typing.cast(Future, _NullFuture())
704moment.__doc__ = """A special object which may be yielded to allow the IOLoop to run for
705one iteration.
707This is not needed in normal use but it can be helpful in long-running
708coroutines that are likely to yield Futures that are ready instantly.
710Usage: ``yield gen.moment``
712In native coroutines, the equivalent of ``yield gen.moment`` is
713``await asyncio.sleep(0)``.
715.. versionadded:: 4.0
717.. deprecated:: 4.5
718 ``yield None`` (or ``yield`` with no argument) is now equivalent to
719 ``yield gen.moment``.
720"""
723class Runner(object):
724 """Internal implementation of `tornado.gen.coroutine`.
726 Maintains information about pending callbacks and their results.
728 The results of the generator are stored in ``result_future`` (a
729 `.Future`)
730 """
732 def __init__(
733 self,
734 ctx_run: Callable,
735 gen: "Generator[_Yieldable, Any, _T]",
736 result_future: "Future[_T]",
737 first_yielded: _Yieldable,
738 ) -> None:
739 self.ctx_run = ctx_run
740 self.gen = gen
741 self.result_future = result_future
742 self.future = _null_future # type: Union[None, Future]
743 self.running = False
744 self.finished = False
745 self.io_loop = IOLoop.current()
746 if self.ctx_run(self.handle_yield, first_yielded):
747 gen = result_future = first_yielded = None # type: ignore
748 self.ctx_run(self.run)
750 def run(self) -> None:
751 """Starts or resumes the generator, running until it reaches a
752 yield point that is not ready.
753 """
754 if self.running or self.finished:
755 return
756 try:
757 self.running = True
758 while True:
759 future = self.future
760 if future is None:
761 raise Exception("No pending future")
762 if not future.done():
763 return
764 self.future = None
765 try:
766 try:
767 value = future.result()
768 except Exception as e:
769 # Save the exception for later. It's important that
770 # gen.throw() not be called inside this try/except block
771 # because that makes sys.exc_info behave unexpectedly.
772 exc: Optional[Exception] = e
773 else:
774 exc = None
775 finally:
776 future = None
778 if exc is not None:
779 try:
780 yielded = self.gen.throw(exc)
781 finally:
782 # Break up a circular reference for faster GC on
783 # CPython.
784 del exc
785 else:
786 yielded = self.gen.send(value)
788 except (StopIteration, Return) as e:
789 self.finished = True
790 self.future = _null_future
791 future_set_result_unless_cancelled(
792 self.result_future, _value_from_stopiteration(e)
793 )
794 self.result_future = None # type: ignore
795 return
796 except Exception:
797 self.finished = True
798 self.future = _null_future
799 future_set_exc_info(self.result_future, sys.exc_info())
800 self.result_future = None # type: ignore
801 return
802 if not self.handle_yield(yielded):
803 return
804 yielded = None
805 finally:
806 self.running = False
808 def handle_yield(self, yielded: _Yieldable) -> bool:
809 try:
810 self.future = convert_yielded(yielded)
811 except BadYieldError:
812 self.future = Future()
813 future_set_exc_info(self.future, sys.exc_info())
815 if self.future is moment:
816 self.io_loop.add_callback(self.ctx_run, self.run)
817 return False
818 elif self.future is None:
819 raise Exception("no pending future")
820 elif not self.future.done():
822 def inner(f: Any) -> None:
823 # Break a reference cycle to speed GC.
824 f = None # noqa: F841
825 self.ctx_run(self.run)
827 self.io_loop.add_future(self.future, inner)
828 return False
829 return True
831 def handle_exception(
832 self, typ: Type[Exception], value: Exception, tb: types.TracebackType
833 ) -> bool:
834 if not self.running and not self.finished:
835 self.future = Future()
836 future_set_exc_info(self.future, (typ, value, tb))
837 self.ctx_run(self.run)
838 return True
839 else:
840 return False
843# Convert Awaitables into Futures.
844try:
845 _wrap_awaitable = asyncio.ensure_future
846except AttributeError:
847 # asyncio.ensure_future was introduced in Python 3.4.4, but
848 # Debian jessie still ships with 3.4.2 so try the old name.
849 _wrap_awaitable = getattr(asyncio, "async")
852def convert_yielded(yielded: _Yieldable) -> Future:
853 """Convert a yielded object into a `.Future`.
855 The default implementation accepts lists, dictionaries, and
856 Futures. This has the side effect of starting any coroutines that
857 did not start themselves, similar to `asyncio.ensure_future`.
859 If the `~functools.singledispatch` library is available, this function
860 may be extended to support additional types. For example::
862 @convert_yielded.register(asyncio.Future)
863 def _(asyncio_future):
864 return tornado.platform.asyncio.to_tornado_future(asyncio_future)
866 .. versionadded:: 4.1
868 """
869 if yielded is None or yielded is moment:
870 return moment
871 elif yielded is _null_future:
872 return _null_future
873 elif isinstance(yielded, (list, dict)):
874 return multi(yielded) # type: ignore
875 elif is_future(yielded):
876 return typing.cast(Future, yielded)
877 elif isawaitable(yielded):
878 return _wrap_awaitable(yielded) # type: ignore
879 else:
880 raise BadYieldError("yielded unknown object %r" % (yielded,))
883convert_yielded = singledispatch(convert_yielded)