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

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 

29import functools 

30import sys 

31import types 

32import typing 

33from collections.abc import Callable 

34from concurrent import futures 

35from typing import Any, Union 

36 

37from tornado.log import app_log 

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[_T] = futures.Future() 

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 def shutdown(self, wait: bool = True, cancel_futures: bool = False) -> None: 

68 pass 

69 

70 

71dummy_executor = DummyExecutor() 

72 

73 

74def run_on_executor(*args: Any, **kwargs: Any) -> Callable: 

75 """Decorator to run a synchronous method asynchronously on an executor. 

76 

77 Returns a future. 

78 

79 The executor to be used is determined by the ``executor`` 

80 attributes of ``self``. To use a different attribute name, pass a 

81 keyword argument to the decorator:: 

82 

83 @run_on_executor(executor='_thread_pool') 

84 def foo(self): 

85 pass 

86 

87 This decorator should not be confused with the similarly-named 

88 `.IOLoop.run_in_executor`. In general, using ``run_in_executor`` 

89 when *calling* a blocking method is recommended instead of using 

90 this decorator when *defining* a method. If compatibility with older 

91 versions of Tornado is required, consider defining an executor 

92 and using ``executor.submit()`` at the call site. 

93 

94 .. versionchanged:: 4.2 

95 Added keyword arguments to use alternative attributes. 

96 

97 .. versionchanged:: 5.0 

98 Always uses the current IOLoop instead of ``self.io_loop``. 

99 

100 .. versionchanged:: 5.1 

101 Returns a `.Future` compatible with ``await`` instead of a 

102 `concurrent.futures.Future`. 

103 

104 .. deprecated:: 5.1 

105 

106 The ``callback`` argument is deprecated and will be removed in 

107 6.0. The decorator itself is discouraged in new code but will 

108 not be removed in 6.0. 

109 

110 .. versionchanged:: 6.0 

111 

112 The ``callback`` argument was removed. 

113 """ 

114 

115 # Fully type-checking decorators is tricky, and this one is 

116 # discouraged anyway so it doesn't have all the generic magic. 

117 def run_on_executor_decorator(fn: Callable) -> Callable[..., Future]: 

118 executor = kwargs.get("executor", "executor") 

119 

120 @functools.wraps(fn) 

121 def wrapper(self: Any, *args: Any, **kwargs: Any) -> Future: 

122 async_future: Future = Future() 

123 conc_future = getattr(self, executor).submit(fn, self, *args, **kwargs) 

124 chain_future(conc_future, async_future) 

125 return async_future 

126 

127 return wrapper 

128 

129 if args and kwargs: 

130 raise ValueError("cannot combine positional and keyword args") 

131 if len(args) == 1: 

132 return run_on_executor_decorator(args[0]) 

133 elif len(args) != 0: 

134 raise ValueError("expected 1 argument, got %d" % len(args)) 

135 return run_on_executor_decorator 

136 

137 

138_NO_RESULT = object() 

139 

140 

141def chain_future( 

142 a: Union["Future[_T]", "futures.Future[_T]"], 

143 b: Union["Future[_T]", "futures.Future[_T]"], 

144) -> None: 

145 """Chain two futures together so that when one completes, so does the other. 

146 

147 The result (success or failure) of ``a`` will be copied to ``b``, unless 

148 ``b`` has already been completed or cancelled by the time ``a`` finishes. 

149 

150 .. versionchanged:: 5.0 

151 

152 Now accepts both Tornado/asyncio `Future` objects and 

153 `concurrent.futures.Future`. 

154 

155 .. versionchanged:: 6.6 

156 

157 If ``a`` is cancelled, ``b`` is now cancelled too. Previously ``b`` 

158 would never complete. 

159 

160 """ 

161 

162 def copy(a: "Future[_T]") -> None: 

163 if b.done(): 

164 return 

165 if a.cancelled(): 

166 b.cancel() 

167 return 

168 if hasattr(a, "exc_info") and a.exc_info() is not None: # type: ignore 

169 future_set_exc_info(b, a.exc_info()) # type: ignore 

170 else: 

171 a_exc = a.exception() 

172 if a_exc is not None: 

173 b.set_exception(a_exc) 

174 else: 

175 b.set_result(a.result()) 

176 

177 if isinstance(a, Future): 

178 future_add_done_callback(a, copy) 

179 else: 

180 # concurrent.futures.Future 

181 from tornado.ioloop import IOLoop 

182 

183 IOLoop.current().add_future(a, copy) 

184 

185 

186def future_set_result_unless_cancelled( 

187 future: "Union[futures.Future[_T], Future[_T]]", value: _T 

188) -> None: 

189 """Set the given ``value`` as the `Future`'s result, if not cancelled. 

190 

191 Avoids ``asyncio.InvalidStateError`` when calling ``set_result()`` on 

192 a cancelled `asyncio.Future`. 

193 

194 .. versionadded:: 5.0 

195 """ 

196 if not future.cancelled(): 

197 future.set_result(value) 

198 

199 

200def future_set_exception_unless_cancelled( 

201 future: "Union[futures.Future[_T], Future[_T]]", exc: BaseException 

202) -> None: 

203 """Set the given ``exc`` as the `Future`'s exception. 

204 

205 If the Future is already canceled, logs the exception instead. If 

206 this logging is not desired, the caller should explicitly check 

207 the state of the Future and call ``Future.set_exception`` instead of 

208 this wrapper. 

209 

210 Avoids ``asyncio.InvalidStateError`` when calling ``set_exception()`` on 

211 a cancelled `asyncio.Future`. 

212 

213 .. versionadded:: 6.0 

214 

215 """ 

216 if not future.cancelled(): 

217 future.set_exception(exc) 

218 else: 

219 app_log.error("Exception after Future was cancelled", exc_info=exc) 

220 

221 

222def future_set_exc_info( 

223 future: "Union[futures.Future[_T], Future[_T]]", 

224 exc_info: tuple[type | None, BaseException | None, types.TracebackType | None], 

225) -> None: 

226 """Set the given ``exc_info`` as the `Future`'s exception. 

227 

228 Understands both `asyncio.Future` and the extensions in older 

229 versions of Tornado to enable better tracebacks on Python 2. 

230 

231 .. versionadded:: 5.0 

232 

233 .. versionchanged:: 6.0 

234 

235 If the future is already cancelled, this function is a no-op. 

236 (previously ``asyncio.InvalidStateError`` would be raised) 

237 

238 """ 

239 if exc_info[1] is None: 

240 raise Exception("future_set_exc_info called with no exception") 

241 future_set_exception_unless_cancelled(future, exc_info[1]) 

242 

243 

244@typing.overload 

245def future_add_done_callback( 

246 future: "futures.Future[_T]", callback: Callable[["futures.Future[_T]"], None] 

247) -> None: 

248 pass 

249 

250 

251@typing.overload 

252def future_add_done_callback( 

253 future: "Future[_T]", callback: Callable[["Future[_T]"], None] 

254) -> None: 

255 pass 

256 

257 

258def future_add_done_callback( 

259 future: "Union[futures.Future[_T], Future[_T]]", callback: Callable[..., None] 

260) -> None: 

261 """Arrange to call ``callback`` when ``future`` is complete. 

262 

263 ``callback`` is invoked with one argument, the ``future``. 

264 

265 If ``future`` is already done, ``callback`` is invoked immediately. 

266 This may differ from the behavior of ``Future.add_done_callback``, 

267 which makes no such guarantee. 

268 

269 .. versionadded:: 5.0 

270 """ 

271 if future.done(): 

272 callback(future) 

273 else: 

274 future.add_done_callback(callback)