1#
2# Copyright 2012 Facebook
3#
4# Licensed under the Apache License, Version 2.0 (the "License"); you may
5# not use this file except in compliance with the License. You may obtain
6# a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13# License for the specific language governing permissions and limitations
14# under the License.
15"""Utilities for working with ``Future`` objects.
16
17Tornado previously provided its own ``Future`` class, but now uses
18`asyncio.Future`. This module contains utility functions for working
19with `asyncio.Future` in a way that is backwards-compatible with
20Tornado's old ``Future`` implementation.
21
22While this module is an important part of Tornado's internal
23implementation, applications rarely need to interact with it
24directly.
25
26"""
27
28import asyncio
29from concurrent import futures
30import functools
31import sys
32import types
33
34from tornado.log import app_log
35
36import typing
37from typing import Any, Callable, Optional, Tuple, Union
38
39_T = typing.TypeVar("_T")
40
41
42class ReturnValueIgnoredError(Exception):
43 # No longer used; was previously used by @return_future
44 pass
45
46
47Future = asyncio.Future
48
49FUTURES = (futures.Future, Future)
50
51
52def is_future(x: Any) -> bool:
53 return isinstance(x, FUTURES)
54
55
56class DummyExecutor(futures.Executor):
57 def submit( # type: ignore[override]
58 self, fn: Callable[..., _T], *args: Any, **kwargs: Any
59 ) -> "futures.Future[_T]":
60 future = futures.Future() # type: futures.Future[_T]
61 try:
62 future_set_result_unless_cancelled(future, fn(*args, **kwargs))
63 except Exception:
64 future_set_exc_info(future, sys.exc_info())
65 return future
66
67 if sys.version_info >= (3, 9):
68
69 def shutdown(self, wait: bool = True, cancel_futures: bool = False) -> None:
70 pass
71
72 else:
73
74 def shutdown(self, wait: bool = True) -> None:
75 pass
76
77
78dummy_executor = DummyExecutor()
79
80
81def run_on_executor(*args: Any, **kwargs: Any) -> Callable:
82 """Decorator to run a synchronous method asynchronously on an executor.
83
84 Returns a future.
85
86 The executor to be used is determined by the ``executor``
87 attributes of ``self``. To use a different attribute name, pass a
88 keyword argument to the decorator::
89
90 @run_on_executor(executor='_thread_pool')
91 def foo(self):
92 pass
93
94 This decorator should not be confused with the similarly-named
95 `.IOLoop.run_in_executor`. In general, using ``run_in_executor``
96 when *calling* a blocking method is recommended instead of using
97 this decorator when *defining* a method. If compatibility with older
98 versions of Tornado is required, consider defining an executor
99 and using ``executor.submit()`` at the call site.
100
101 .. versionchanged:: 4.2
102 Added keyword arguments to use alternative attributes.
103
104 .. versionchanged:: 5.0
105 Always uses the current IOLoop instead of ``self.io_loop``.
106
107 .. versionchanged:: 5.1
108 Returns a `.Future` compatible with ``await`` instead of a
109 `concurrent.futures.Future`.
110
111 .. deprecated:: 5.1
112
113 The ``callback`` argument is deprecated and will be removed in
114 6.0. The decorator itself is discouraged in new code but will
115 not be removed in 6.0.
116
117 .. versionchanged:: 6.0
118
119 The ``callback`` argument was removed.
120 """
121
122 # Fully type-checking decorators is tricky, and this one is
123 # discouraged anyway so it doesn't have all the generic magic.
124 def run_on_executor_decorator(fn: Callable) -> Callable[..., Future]:
125 executor = kwargs.get("executor", "executor")
126
127 @functools.wraps(fn)
128 def wrapper(self: Any, *args: Any, **kwargs: Any) -> Future:
129 async_future = Future() # type: Future
130 conc_future = getattr(self, executor).submit(fn, self, *args, **kwargs)
131 chain_future(conc_future, async_future)
132 return async_future
133
134 return wrapper
135
136 if args and kwargs:
137 raise ValueError("cannot combine positional and keyword args")
138 if len(args) == 1:
139 return run_on_executor_decorator(args[0])
140 elif len(args) != 0:
141 raise ValueError("expected 1 argument, got %d", len(args))
142 return run_on_executor_decorator
143
144
145_NO_RESULT = object()
146
147
148def chain_future(a: "Future[_T]", b: "Future[_T]") -> None:
149 """Chain two futures together so that when one completes, so does the other.
150
151 The result (success or failure) of ``a`` will be copied to ``b``, unless
152 ``b`` has already been completed or cancelled by the time ``a`` finishes.
153
154 .. versionchanged:: 5.0
155
156 Now accepts both Tornado/asyncio `Future` objects and
157 `concurrent.futures.Future`.
158
159 """
160
161 def copy(a: "Future[_T]") -> None:
162 if b.done():
163 return
164 if hasattr(a, "exc_info") and a.exc_info() is not None: # type: ignore
165 future_set_exc_info(b, a.exc_info()) # type: ignore
166 else:
167 a_exc = a.exception()
168 if a_exc is not None:
169 b.set_exception(a_exc)
170 else:
171 b.set_result(a.result())
172
173 if isinstance(a, Future):
174 future_add_done_callback(a, copy)
175 else:
176 # concurrent.futures.Future
177 from tornado.ioloop import IOLoop
178
179 IOLoop.current().add_future(a, copy)
180
181
182def future_set_result_unless_cancelled(
183 future: "Union[futures.Future[_T], Future[_T]]", value: _T
184) -> None:
185 """Set the given ``value`` as the `Future`'s result, if not cancelled.
186
187 Avoids ``asyncio.InvalidStateError`` when calling ``set_result()`` on
188 a cancelled `asyncio.Future`.
189
190 .. versionadded:: 5.0
191 """
192 if not future.cancelled():
193 future.set_result(value)
194
195
196def future_set_exception_unless_cancelled(
197 future: "Union[futures.Future[_T], Future[_T]]", exc: BaseException
198) -> None:
199 """Set the given ``exc`` as the `Future`'s exception.
200
201 If the Future is already canceled, logs the exception instead. If
202 this logging is not desired, the caller should explicitly check
203 the state of the Future and call ``Future.set_exception`` instead of
204 this wrapper.
205
206 Avoids ``asyncio.InvalidStateError`` when calling ``set_exception()`` on
207 a cancelled `asyncio.Future`.
208
209 .. versionadded:: 6.0
210
211 """
212 if not future.cancelled():
213 future.set_exception(exc)
214 else:
215 app_log.error("Exception after Future was cancelled", exc_info=exc)
216
217
218def future_set_exc_info(
219 future: "Union[futures.Future[_T], Future[_T]]",
220 exc_info: Tuple[
221 Optional[type], Optional[BaseException], Optional[types.TracebackType]
222 ],
223) -> None:
224 """Set the given ``exc_info`` as the `Future`'s exception.
225
226 Understands both `asyncio.Future` and the extensions in older
227 versions of Tornado to enable better tracebacks on Python 2.
228
229 .. versionadded:: 5.0
230
231 .. versionchanged:: 6.0
232
233 If the future is already cancelled, this function is a no-op.
234 (previously ``asyncio.InvalidStateError`` would be raised)
235
236 """
237 if exc_info[1] is None:
238 raise Exception("future_set_exc_info called with no exception")
239 future_set_exception_unless_cancelled(future, exc_info[1])
240
241
242@typing.overload
243def future_add_done_callback(
244 future: "futures.Future[_T]", callback: Callable[["futures.Future[_T]"], None]
245) -> None:
246 pass
247
248
249@typing.overload # noqa: F811
250def future_add_done_callback(
251 future: "Future[_T]", callback: Callable[["Future[_T]"], None]
252) -> None:
253 pass
254
255
256def future_add_done_callback( # noqa: F811
257 future: "Union[futures.Future[_T], Future[_T]]", callback: Callable[..., None]
258) -> None:
259 """Arrange to call ``callback`` when ``future`` is complete.
260
261 ``callback`` is invoked with one argument, the ``future``.
262
263 If ``future`` is already done, ``callback`` is invoked immediately.
264 This may differ from the behavior of ``Future.add_done_callback``,
265 which makes no such guarantee.
266
267 .. versionadded:: 5.0
268 """
269 if future.done():
270 callback(future)
271 else:
272 future.add_done_callback(callback)