Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/asyncio/futures.py: 26%
206 statements
« prev ^ index » next coverage.py v7.0.1, created at 2022-12-25 06:11 +0000
1"""A Future class similar to the one in PEP 3148."""
3__all__ = (
4 'Future', 'wrap_future', 'isfuture',
5)
7import concurrent.futures
8import contextvars
9import logging
10import sys
12from . import base_futures
13from . import events
14from . import exceptions
15from . import format_helpers
18isfuture = base_futures.isfuture
21_PENDING = base_futures._PENDING
22_CANCELLED = base_futures._CANCELLED
23_FINISHED = base_futures._FINISHED
26STACK_DEBUG = logging.DEBUG - 1 # heavy-duty debugging
29class Future:
30 """This class is *almost* compatible with concurrent.futures.Future.
32 Differences:
34 - This class is not thread-safe.
36 - result() and exception() do not take a timeout argument and
37 raise an exception when the future isn't done yet.
39 - Callbacks registered with add_done_callback() are always called
40 via the event loop's call_soon().
42 - This class is not compatible with the wait() and as_completed()
43 methods in the concurrent.futures package.
45 (In Python 3.4 or later we may be able to unify the implementations.)
46 """
48 # Class variables serving as defaults for instance variables.
49 _state = _PENDING
50 _result = None
51 _exception = None
52 _loop = None
53 _source_traceback = None
55 # This field is used for a dual purpose:
56 # - Its presence is a marker to declare that a class implements
57 # the Future protocol (i.e. is intended to be duck-type compatible).
58 # The value must also be not-None, to enable a subclass to declare
59 # that it is not compatible by setting this to None.
60 # - It is set by __iter__() below so that Task._step() can tell
61 # the difference between
62 # `await Future()` or`yield from Future()` (correct) vs.
63 # `yield Future()` (incorrect).
64 _asyncio_future_blocking = False
66 __log_traceback = False
68 def __init__(self, *, loop=None):
69 """Initialize the future.
71 The optional event_loop argument allows explicitly setting the event
72 loop object used by the future. If it's not provided, the future uses
73 the default event loop.
74 """
75 if loop is None:
76 self._loop = events.get_event_loop()
77 else:
78 self._loop = loop
79 self._callbacks = []
80 if self._loop.get_debug():
81 self._source_traceback = format_helpers.extract_stack(
82 sys._getframe(1))
84 _repr_info = base_futures._future_repr_info
86 def __repr__(self):
87 return '<{} {}>'.format(self.__class__.__name__,
88 ' '.join(self._repr_info()))
90 def __del__(self):
91 if not self.__log_traceback:
92 # set_exception() was not called, or result() or exception()
93 # has consumed the exception
94 return
95 exc = self._exception
96 context = {
97 'message':
98 f'{self.__class__.__name__} exception was never retrieved',
99 'exception': exc,
100 'future': self,
101 }
102 if self._source_traceback:
103 context['source_traceback'] = self._source_traceback
104 self._loop.call_exception_handler(context)
106 @property
107 def _log_traceback(self):
108 return self.__log_traceback
110 @_log_traceback.setter
111 def _log_traceback(self, val):
112 if bool(val):
113 raise ValueError('_log_traceback can only be set to False')
114 self.__log_traceback = False
116 def get_loop(self):
117 """Return the event loop the Future is bound to."""
118 loop = self._loop
119 if loop is None:
120 raise RuntimeError("Future object is not initialized.")
121 return loop
123 def cancel(self):
124 """Cancel the future and schedule callbacks.
126 If the future is already done or cancelled, return False. Otherwise,
127 change the future's state to cancelled, schedule the callbacks and
128 return True.
129 """
130 self.__log_traceback = False
131 if self._state != _PENDING:
132 return False
133 self._state = _CANCELLED
134 self.__schedule_callbacks()
135 return True
137 def __schedule_callbacks(self):
138 """Internal: Ask the event loop to call all callbacks.
140 The callbacks are scheduled to be called as soon as possible. Also
141 clears the callback list.
142 """
143 callbacks = self._callbacks[:]
144 if not callbacks:
145 return
147 self._callbacks[:] = []
148 for callback, ctx in callbacks:
149 self._loop.call_soon(callback, self, context=ctx)
151 def cancelled(self):
152 """Return True if the future was cancelled."""
153 return self._state == _CANCELLED
155 # Don't implement running(); see http://bugs.python.org/issue18699
157 def done(self):
158 """Return True if the future is done.
160 Done means either that a result / exception are available, or that the
161 future was cancelled.
162 """
163 return self._state != _PENDING
165 def result(self):
166 """Return the result this future represents.
168 If the future has been cancelled, raises CancelledError. If the
169 future's result isn't yet available, raises InvalidStateError. If
170 the future is done and has an exception set, this exception is raised.
171 """
172 if self._state == _CANCELLED:
173 raise exceptions.CancelledError
174 if self._state != _FINISHED:
175 raise exceptions.InvalidStateError('Result is not ready.')
176 self.__log_traceback = False
177 if self._exception is not None:
178 raise self._exception
179 return self._result
181 def exception(self):
182 """Return the exception that was set on this future.
184 The exception (or None if no exception was set) is returned only if
185 the future is done. If the future has been cancelled, raises
186 CancelledError. If the future isn't done yet, raises
187 InvalidStateError.
188 """
189 if self._state == _CANCELLED:
190 raise exceptions.CancelledError
191 if self._state != _FINISHED:
192 raise exceptions.InvalidStateError('Exception is not set.')
193 self.__log_traceback = False
194 return self._exception
196 def add_done_callback(self, fn, *, context=None):
197 """Add a callback to be run when the future becomes done.
199 The callback is called with a single argument - the future object. If
200 the future is already done when this is called, the callback is
201 scheduled with call_soon.
202 """
203 if self._state != _PENDING:
204 self._loop.call_soon(fn, self, context=context)
205 else:
206 if context is None:
207 context = contextvars.copy_context()
208 self._callbacks.append((fn, context))
210 # New method not in PEP 3148.
212 def remove_done_callback(self, fn):
213 """Remove all instances of a callback from the "call when done" list.
215 Returns the number of callbacks removed.
216 """
217 filtered_callbacks = [(f, ctx)
218 for (f, ctx) in self._callbacks
219 if f != fn]
220 removed_count = len(self._callbacks) - len(filtered_callbacks)
221 if removed_count:
222 self._callbacks[:] = filtered_callbacks
223 return removed_count
225 # So-called internal methods (note: no set_running_or_notify_cancel()).
227 def set_result(self, result):
228 """Mark the future done and set its result.
230 If the future is already done when this method is called, raises
231 InvalidStateError.
232 """
233 if self._state != _PENDING:
234 raise exceptions.InvalidStateError(f'{self._state}: {self!r}')
235 self._result = result
236 self._state = _FINISHED
237 self.__schedule_callbacks()
239 def set_exception(self, exception):
240 """Mark the future done and set an exception.
242 If the future is already done when this method is called, raises
243 InvalidStateError.
244 """
245 if self._state != _PENDING:
246 raise exceptions.InvalidStateError(f'{self._state}: {self!r}')
247 if isinstance(exception, type):
248 exception = exception()
249 if type(exception) is StopIteration:
250 raise TypeError("StopIteration interacts badly with generators "
251 "and cannot be raised into a Future")
252 self._exception = exception
253 self._state = _FINISHED
254 self.__schedule_callbacks()
255 self.__log_traceback = True
257 def __await__(self):
258 if not self.done():
259 self._asyncio_future_blocking = True
260 yield self # This tells Task to wait for completion.
261 if not self.done():
262 raise RuntimeError("await wasn't used with future")
263 return self.result() # May raise too.
265 __iter__ = __await__ # make compatible with 'yield from'.
268# Needed for testing purposes.
269_PyFuture = Future
272def _get_loop(fut):
273 # Tries to call Future.get_loop() if it's available.
274 # Otherwise fallbacks to using the old '_loop' property.
275 try:
276 get_loop = fut.get_loop
277 except AttributeError:
278 pass
279 else:
280 return get_loop()
281 return fut._loop
284def _set_result_unless_cancelled(fut, result):
285 """Helper setting the result only if the future was not cancelled."""
286 if fut.cancelled():
287 return
288 fut.set_result(result)
291def _convert_future_exc(exc):
292 exc_class = type(exc)
293 if exc_class is concurrent.futures.CancelledError:
294 return exceptions.CancelledError(*exc.args)
295 elif exc_class is concurrent.futures.TimeoutError:
296 return exceptions.TimeoutError(*exc.args)
297 elif exc_class is concurrent.futures.InvalidStateError:
298 return exceptions.InvalidStateError(*exc.args)
299 else:
300 return exc
303def _set_concurrent_future_state(concurrent, source):
304 """Copy state from a future to a concurrent.futures.Future."""
305 assert source.done()
306 if source.cancelled():
307 concurrent.cancel()
308 if not concurrent.set_running_or_notify_cancel():
309 return
310 exception = source.exception()
311 if exception is not None:
312 concurrent.set_exception(_convert_future_exc(exception))
313 else:
314 result = source.result()
315 concurrent.set_result(result)
318def _copy_future_state(source, dest):
319 """Internal helper to copy state from another Future.
321 The other Future may be a concurrent.futures.Future.
322 """
323 assert source.done()
324 if dest.cancelled():
325 return
326 assert not dest.done()
327 if source.cancelled():
328 dest.cancel()
329 else:
330 exception = source.exception()
331 if exception is not None:
332 dest.set_exception(_convert_future_exc(exception))
333 else:
334 result = source.result()
335 dest.set_result(result)
338def _chain_future(source, destination):
339 """Chain two futures so that when one completes, so does the other.
341 The result (or exception) of source will be copied to destination.
342 If destination is cancelled, source gets cancelled too.
343 Compatible with both asyncio.Future and concurrent.futures.Future.
344 """
345 if not isfuture(source) and not isinstance(source,
346 concurrent.futures.Future):
347 raise TypeError('A future is required for source argument')
348 if not isfuture(destination) and not isinstance(destination,
349 concurrent.futures.Future):
350 raise TypeError('A future is required for destination argument')
351 source_loop = _get_loop(source) if isfuture(source) else None
352 dest_loop = _get_loop(destination) if isfuture(destination) else None
354 def _set_state(future, other):
355 if isfuture(future):
356 _copy_future_state(other, future)
357 else:
358 _set_concurrent_future_state(future, other)
360 def _call_check_cancel(destination):
361 if destination.cancelled():
362 if source_loop is None or source_loop is dest_loop:
363 source.cancel()
364 else:
365 source_loop.call_soon_threadsafe(source.cancel)
367 def _call_set_state(source):
368 if (destination.cancelled() and
369 dest_loop is not None and dest_loop.is_closed()):
370 return
371 if dest_loop is None or dest_loop is source_loop:
372 _set_state(destination, source)
373 else:
374 dest_loop.call_soon_threadsafe(_set_state, destination, source)
376 destination.add_done_callback(_call_check_cancel)
377 source.add_done_callback(_call_set_state)
380def wrap_future(future, *, loop=None):
381 """Wrap concurrent.futures.Future object."""
382 if isfuture(future):
383 return future
384 assert isinstance(future, concurrent.futures.Future), \
385 f'concurrent.futures.Future is expected, got {future!r}'
386 if loop is None:
387 loop = events.get_event_loop()
388 new_future = loop.create_future()
389 _chain_future(future, new_future)
390 return new_future
393try:
394 import _asyncio
395except ImportError:
396 pass
397else:
398 # _CFuture is needed for tests.
399 Future = _CFuture = _asyncio.Future