| """A Future class similar to the one in PEP 3148.""" |
| |
| __all__ = ['CancelledError', 'TimeoutError', 'InvalidStateError', |
| 'Future', 'wrap_future', 'isfuture'] |
| |
| import concurrent.futures |
| import logging |
| import sys |
| import traceback |
| |
| from . import base_futures |
| from . import compat |
| from . import events |
| |
| |
| CancelledError = base_futures.CancelledError |
| InvalidStateError = base_futures.InvalidStateError |
| TimeoutError = base_futures.TimeoutError |
| isfuture = base_futures.isfuture |
| |
| |
| _PENDING = base_futures._PENDING |
| _CANCELLED = base_futures._CANCELLED |
| _FINISHED = base_futures._FINISHED |
| |
| |
| STACK_DEBUG = logging.DEBUG - 1 # heavy-duty debugging |
| |
| |
| class _TracebackLogger: |
| """Helper to log a traceback upon destruction if not cleared. |
| |
| This solves a nasty problem with Futures and Tasks that have an |
| exception set: if nobody asks for the exception, the exception is |
| never logged. This violates the Zen of Python: 'Errors should |
| never pass silently. Unless explicitly silenced.' |
| |
| However, we don't want to log the exception as soon as |
| set_exception() is called: if the calling code is written |
| properly, it will get the exception and handle it properly. But |
| we *do* want to log it if result() or exception() was never called |
| -- otherwise developers waste a lot of time wondering why their |
| buggy code fails silently. |
| |
| An earlier attempt added a __del__() method to the Future class |
| itself, but this backfired because the presence of __del__() |
| prevents garbage collection from breaking cycles. A way out of |
| this catch-22 is to avoid having a __del__() method on the Future |
| class itself, but instead to have a reference to a helper object |
| with a __del__() method that logs the traceback, where we ensure |
| that the helper object doesn't participate in cycles, and only the |
| Future has a reference to it. |
| |
| The helper object is added when set_exception() is called. When |
| the Future is collected, and the helper is present, the helper |
| object is also collected, and its __del__() method will log the |
| traceback. When the Future's result() or exception() method is |
| called (and a helper object is present), it removes the helper |
| object, after calling its clear() method to prevent it from |
| logging. |
| |
| One downside is that we do a fair amount of work to extract the |
| traceback from the exception, even when it is never logged. It |
| would seem cheaper to just store the exception object, but that |
| references the traceback, which references stack frames, which may |
| reference the Future, which references the _TracebackLogger, and |
| then the _TracebackLogger would be included in a cycle, which is |
| what we're trying to avoid! As an optimization, we don't |
| immediately format the exception; we only do the work when |
| activate() is called, which call is delayed until after all the |
| Future's callbacks have run. Since usually a Future has at least |
| one callback (typically set by 'yield from') and usually that |
| callback extracts the callback, thereby removing the need to |
| format the exception. |
| |
| PS. I don't claim credit for this solution. I first heard of it |
| in a discussion about closing files when they are collected. |
| """ |
| |
| __slots__ = ('loop', 'source_traceback', 'exc', 'tb') |
| |
| def __init__(self, future, exc): |
| self.loop = future._loop |
| self.source_traceback = future._source_traceback |
| self.exc = exc |
| self.tb = None |
| |
| def activate(self): |
| exc = self.exc |
| if exc is not None: |
| self.exc = None |
| self.tb = traceback.format_exception(exc.__class__, exc, |
| exc.__traceback__) |
| |
| def clear(self): |
| self.exc = None |
| self.tb = None |
| |
| def __del__(self): |
| if self.tb: |
| msg = 'Future/Task exception was never retrieved\n' |
| if self.source_traceback: |
| src = ''.join(traceback.format_list(self.source_traceback)) |
| msg += 'Future/Task created at (most recent call last):\n' |
| msg += '%s\n' % src.rstrip() |
| msg += ''.join(self.tb).rstrip() |
| self.loop.call_exception_handler({'message': msg}) |
| |
| |
| class Future: |
| """This class is *almost* compatible with concurrent.futures.Future. |
| |
| Differences: |
| |
| - This class is not thread-safe. |
| |
| - result() and exception() do not take a timeout argument and |
| raise an exception when the future isn't done yet. |
| |
| - Callbacks registered with add_done_callback() are always called |
| via the event loop's call_soon(). |
| |
| - This class is not compatible with the wait() and as_completed() |
| methods in the concurrent.futures package. |
| |
| (In Python 3.4 or later we may be able to unify the implementations.) |
| """ |
| |
| # Class variables serving as defaults for instance variables. |
| _state = _PENDING |
| _result = None |
| _exception = None |
| _loop = None |
| _source_traceback = None |
| |
| # This field is used for a dual purpose: |
| # - Its presence is a marker to declare that a class implements |
| # the Future protocol (i.e. is intended to be duck-type compatible). |
| # The value must also be not-None, to enable a subclass to declare |
| # that it is not compatible by setting this to None. |
| # - It is set by __iter__() below so that Task._step() can tell |
| # the difference between `yield from Future()` (correct) vs. |
| # `yield Future()` (incorrect). |
| _asyncio_future_blocking = False |
| |
| _log_traceback = False |
| |
| def __init__(self, *, loop=None): |
| """Initialize the future. |
| |
| The optional event_loop argument allows explicitly setting the event |
| loop object used by the future. If it's not provided, the future uses |
| the default event loop. |
| """ |
| if loop is None: |
| self._loop = events.get_event_loop() |
| else: |
| self._loop = loop |
| self._callbacks = [] |
| if self._loop.get_debug(): |
| self._source_traceback = events.extract_stack(sys._getframe(1)) |
| |
| _repr_info = base_futures._future_repr_info |
| |
| def __repr__(self): |
| return '<%s %s>' % (self.__class__.__name__, ' '.join(self._repr_info())) |
| |
| # On Python 3.3 and older, objects with a destructor part of a reference |
| # cycle are never destroyed. It's not more the case on Python 3.4 thanks |
| # to the PEP 442. |
| if compat.PY34: |
| def __del__(self): |
| if not self._log_traceback: |
| # set_exception() was not called, or result() or exception() |
| # has consumed the exception |
| return |
| exc = self._exception |
| context = { |
| 'message': ('%s exception was never retrieved' |
| % self.__class__.__name__), |
| 'exception': exc, |
| 'future': self, |
| } |
| if self._source_traceback: |
| context['source_traceback'] = self._source_traceback |
| self._loop.call_exception_handler(context) |
| |
| def cancel(self): |
| """Cancel the future and schedule callbacks. |
| |
| If the future is already done or cancelled, return False. Otherwise, |
| change the future's state to cancelled, schedule the callbacks and |
| return True. |
| """ |
| self._log_traceback = False |
| if self._state != _PENDING: |
| return False |
| self._state = _CANCELLED |
| self._schedule_callbacks() |
| return True |
| |
| def _schedule_callbacks(self): |
| """Internal: Ask the event loop to call all callbacks. |
| |
| The callbacks are scheduled to be called as soon as possible. Also |
| clears the callback list. |
| """ |
| callbacks = self._callbacks[:] |
| if not callbacks: |
| return |
| |
| self._callbacks[:] = [] |
| for callback in callbacks: |
| self._loop.call_soon(callback, self) |
| |
| def cancelled(self): |
| """Return True if the future was cancelled.""" |
| return self._state == _CANCELLED |
| |
| # Don't implement running(); see http://bugs.python.org/issue18699 |
| |
| def done(self): |
| """Return True if the future is done. |
| |
| Done means either that a result / exception are available, or that the |
| future was cancelled. |
| """ |
| return self._state != _PENDING |
| |
| def result(self): |
| """Return the result this future represents. |
| |
| If the future has been cancelled, raises CancelledError. If the |
| future's result isn't yet available, raises InvalidStateError. If |
| the future is done and has an exception set, this exception is raised. |
| """ |
| if self._state == _CANCELLED: |
| raise CancelledError |
| if self._state != _FINISHED: |
| raise InvalidStateError('Result is not ready.') |
| self._log_traceback = False |
| if self._exception is not None: |
| raise self._exception |
| return self._result |
| |
| def exception(self): |
| """Return the exception that was set on this future. |
| |
| The exception (or None if no exception was set) is returned only if |
| the future is done. If the future has been cancelled, raises |
| CancelledError. If the future isn't done yet, raises |
| InvalidStateError. |
| """ |
| if self._state == _CANCELLED: |
| raise CancelledError |
| if self._state != _FINISHED: |
| raise InvalidStateError('Exception is not set.') |
| self._log_traceback = False |
| return self._exception |
| |
| def add_done_callback(self, fn): |
| """Add a callback to be run when the future becomes done. |
| |
| The callback is called with a single argument - the future object. If |
| the future is already done when this is called, the callback is |
| scheduled with call_soon. |
| """ |
| if self._state != _PENDING: |
| self._loop.call_soon(fn, self) |
| else: |
| self._callbacks.append(fn) |
| |
| # New method not in PEP 3148. |
| |
| def remove_done_callback(self, fn): |
| """Remove all instances of a callback from the "call when done" list. |
| |
| Returns the number of callbacks removed. |
| """ |
| filtered_callbacks = [f for f in self._callbacks if f != fn] |
| removed_count = len(self._callbacks) - len(filtered_callbacks) |
| if removed_count: |
| self._callbacks[:] = filtered_callbacks |
| return removed_count |
| |
| # So-called internal methods (note: no set_running_or_notify_cancel()). |
| |
| def set_result(self, result): |
| """Mark the future done and set its result. |
| |
| If the future is already done when this method is called, raises |
| InvalidStateError. |
| """ |
| if self._state != _PENDING: |
| raise InvalidStateError('{}: {!r}'.format(self._state, self)) |
| self._result = result |
| self._state = _FINISHED |
| self._schedule_callbacks() |
| |
| def set_exception(self, exception): |
| """Mark the future done and set an exception. |
| |
| If the future is already done when this method is called, raises |
| InvalidStateError. |
| """ |
| if self._state != _PENDING: |
| raise InvalidStateError('{}: {!r}'.format(self._state, self)) |
| if isinstance(exception, type): |
| exception = exception() |
| if type(exception) is StopIteration: |
| raise TypeError("StopIteration interacts badly with generators " |
| "and cannot be raised into a Future") |
| self._exception = exception |
| self._state = _FINISHED |
| self._schedule_callbacks() |
| if compat.PY34: |
| self._log_traceback = True |
| else: |
| self._tb_logger = _TracebackLogger(self, exception) |
| # Arrange for the logger to be activated after all callbacks |
| # have had a chance to call result() or exception(). |
| self._loop.call_soon(self._tb_logger.activate) |
| |
| def __iter__(self): |
| if not self.done(): |
| self._asyncio_future_blocking = True |
| yield self # This tells Task to wait for completion. |
| assert self.done(), "yield from wasn't used with future" |
| return self.result() # May raise too. |
| |
| if compat.PY35: |
| __await__ = __iter__ # make compatible with 'await' expression |
| |
| |
| # Needed for testing purposes. |
| _PyFuture = Future |
| |
| |
| def _set_result_unless_cancelled(fut, result): |
| """Helper setting the result only if the future was not cancelled.""" |
| if fut.cancelled(): |
| return |
| fut.set_result(result) |
| |
| |
| def _set_concurrent_future_state(concurrent, source): |
| """Copy state from a future to a concurrent.futures.Future.""" |
| assert source.done() |
| if source.cancelled(): |
| concurrent.cancel() |
| if not concurrent.set_running_or_notify_cancel(): |
| return |
| exception = source.exception() |
| if exception is not None: |
| concurrent.set_exception(exception) |
| else: |
| result = source.result() |
| concurrent.set_result(result) |
| |
| |
| def _copy_future_state(source, dest): |
| """Internal helper to copy state from another Future. |
| |
| The other Future may be a concurrent.futures.Future. |
| """ |
| assert source.done() |
| if dest.cancelled(): |
| return |
| assert not dest.done() |
| if source.cancelled(): |
| dest.cancel() |
| else: |
| exception = source.exception() |
| if exception is not None: |
| dest.set_exception(exception) |
| else: |
| result = source.result() |
| dest.set_result(result) |
| |
| |
| def _chain_future(source, destination): |
| """Chain two futures so that when one completes, so does the other. |
| |
| The result (or exception) of source will be copied to destination. |
| If destination is cancelled, source gets cancelled too. |
| Compatible with both asyncio.Future and concurrent.futures.Future. |
| """ |
| if not isfuture(source) and not isinstance(source, |
| concurrent.futures.Future): |
| raise TypeError('A future is required for source argument') |
| if not isfuture(destination) and not isinstance(destination, |
| concurrent.futures.Future): |
| raise TypeError('A future is required for destination argument') |
| source_loop = source._loop if isfuture(source) else None |
| dest_loop = destination._loop if isfuture(destination) else None |
| |
| def _set_state(future, other): |
| if isfuture(future): |
| _copy_future_state(other, future) |
| else: |
| _set_concurrent_future_state(future, other) |
| |
| def _call_check_cancel(destination): |
| if destination.cancelled(): |
| if source_loop is None or source_loop is dest_loop: |
| source.cancel() |
| else: |
| source_loop.call_soon_threadsafe(source.cancel) |
| |
| def _call_set_state(source): |
| if (destination.cancelled() and |
| dest_loop is not None and dest_loop.is_closed()): |
| return |
| if dest_loop is None or dest_loop is source_loop: |
| _set_state(destination, source) |
| else: |
| dest_loop.call_soon_threadsafe(_set_state, destination, source) |
| |
| destination.add_done_callback(_call_check_cancel) |
| source.add_done_callback(_call_set_state) |
| |
| |
| def wrap_future(future, *, loop=None): |
| """Wrap concurrent.futures.Future object.""" |
| if isfuture(future): |
| return future |
| assert isinstance(future, concurrent.futures.Future), \ |
| 'concurrent.futures.Future is expected, got {!r}'.format(future) |
| if loop is None: |
| loop = events.get_event_loop() |
| new_future = loop.create_future() |
| _chain_future(future, new_future) |
| return new_future |
| |
| |
| try: |
| import _asyncio |
| except ImportError: |
| pass |
| else: |
| # _CFuture is needed for tests. |
| Future = _CFuture = _asyncio.Future |
