Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/tornado/concurrent.py: 47%

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)