ยปCore Development>Code coverage>Lib/asyncio/futures.py

Python code coverage for Lib/asyncio/futures.py

#countcontent
1n/a"""A Future class similar to the one in PEP 3148."""
2n/a
3n/a__all__ = ['CancelledError', 'TimeoutError', 'InvalidStateError',
4n/a 'Future', 'wrap_future', 'isfuture']
5n/a
6n/aimport concurrent.futures
7n/aimport logging
8n/aimport sys
9n/aimport traceback
10n/a
11n/afrom . import base_futures
12n/afrom . import compat
13n/afrom . import events
14n/a
15n/a
16n/aCancelledError = base_futures.CancelledError
17n/aInvalidStateError = base_futures.InvalidStateError
18n/aTimeoutError = base_futures.TimeoutError
19n/aisfuture = base_futures.isfuture
20n/a
21n/a
22n/a_PENDING = base_futures._PENDING
23n/a_CANCELLED = base_futures._CANCELLED
24n/a_FINISHED = base_futures._FINISHED
25n/a
26n/a
27n/aSTACK_DEBUG = logging.DEBUG - 1 # heavy-duty debugging
28n/a
29n/a
30n/aclass _TracebackLogger:
31n/a """Helper to log a traceback upon destruction if not cleared.
32n/a
33n/a This solves a nasty problem with Futures and Tasks that have an
34n/a exception set: if nobody asks for the exception, the exception is
35n/a never logged. This violates the Zen of Python: 'Errors should
36n/a never pass silently. Unless explicitly silenced.'
37n/a
38n/a However, we don't want to log the exception as soon as
39n/a set_exception() is called: if the calling code is written
40n/a properly, it will get the exception and handle it properly. But
41n/a we *do* want to log it if result() or exception() was never called
42n/a -- otherwise developers waste a lot of time wondering why their
43n/a buggy code fails silently.
44n/a
45n/a An earlier attempt added a __del__() method to the Future class
46n/a itself, but this backfired because the presence of __del__()
47n/a prevents garbage collection from breaking cycles. A way out of
48n/a this catch-22 is to avoid having a __del__() method on the Future
49n/a class itself, but instead to have a reference to a helper object
50n/a with a __del__() method that logs the traceback, where we ensure
51n/a that the helper object doesn't participate in cycles, and only the
52n/a Future has a reference to it.
53n/a
54n/a The helper object is added when set_exception() is called. When
55n/a the Future is collected, and the helper is present, the helper
56n/a object is also collected, and its __del__() method will log the
57n/a traceback. When the Future's result() or exception() method is
58n/a called (and a helper object is present), it removes the helper
59n/a object, after calling its clear() method to prevent it from
60n/a logging.
61n/a
62n/a One downside is that we do a fair amount of work to extract the
63n/a traceback from the exception, even when it is never logged. It
64n/a would seem cheaper to just store the exception object, but that
65n/a references the traceback, which references stack frames, which may
66n/a reference the Future, which references the _TracebackLogger, and
67n/a then the _TracebackLogger would be included in a cycle, which is
68n/a what we're trying to avoid! As an optimization, we don't
69n/a immediately format the exception; we only do the work when
70n/a activate() is called, which call is delayed until after all the
71n/a Future's callbacks have run. Since usually a Future has at least
72n/a one callback (typically set by 'yield from') and usually that
73n/a callback extracts the callback, thereby removing the need to
74n/a format the exception.
75n/a
76n/a PS. I don't claim credit for this solution. I first heard of it
77n/a in a discussion about closing files when they are collected.
78n/a """
79n/a
80n/a __slots__ = ('loop', 'source_traceback', 'exc', 'tb')
81n/a
82n/a def __init__(self, future, exc):
83n/a self.loop = future._loop
84n/a self.source_traceback = future._source_traceback
85n/a self.exc = exc
86n/a self.tb = None
87n/a
88n/a def activate(self):
89n/a exc = self.exc
90n/a if exc is not None:
91n/a self.exc = None
92n/a self.tb = traceback.format_exception(exc.__class__, exc,
93n/a exc.__traceback__)
94n/a
95n/a def clear(self):
96n/a self.exc = None
97n/a self.tb = None
98n/a
99n/a def __del__(self):
100n/a if self.tb:
101n/a msg = 'Future/Task exception was never retrieved\n'
102n/a if self.source_traceback:
103n/a src = ''.join(traceback.format_list(self.source_traceback))
104n/a msg += 'Future/Task created at (most recent call last):\n'
105n/a msg += '%s\n' % src.rstrip()
106n/a msg += ''.join(self.tb).rstrip()
107n/a self.loop.call_exception_handler({'message': msg})
108n/a
109n/a
110n/aclass Future:
111n/a """This class is *almost* compatible with concurrent.futures.Future.
112n/a
113n/a Differences:
114n/a
115n/a - result() and exception() do not take a timeout argument and
116n/a raise an exception when the future isn't done yet.
117n/a
118n/a - Callbacks registered with add_done_callback() are always called
119n/a via the event loop's call_soon_threadsafe().
120n/a
121n/a - This class is not compatible with the wait() and as_completed()
122n/a methods in the concurrent.futures package.
123n/a
124n/a (In Python 3.4 or later we may be able to unify the implementations.)
125n/a """
126n/a
127n/a # Class variables serving as defaults for instance variables.
128n/a _state = _PENDING
129n/a _result = None
130n/a _exception = None
131n/a _loop = None
132n/a _source_traceback = None
133n/a
134n/a # This field is used for a dual purpose:
135n/a # - Its presence is a marker to declare that a class implements
136n/a # the Future protocol (i.e. is intended to be duck-type compatible).
137n/a # The value must also be not-None, to enable a subclass to declare
138n/a # that it is not compatible by setting this to None.
139n/a # - It is set by __iter__() below so that Task._step() can tell
140n/a # the difference between `yield from Future()` (correct) vs.
141n/a # `yield Future()` (incorrect).
142n/a _asyncio_future_blocking = False
143n/a
144n/a _log_traceback = False # Used for Python 3.4 and later
145n/a _tb_logger = None # Used for Python 3.3 only
146n/a
147n/a def __init__(self, *, loop=None):
148n/a """Initialize the future.
149n/a
150n/a The optional event_loop argument allows explicitly setting the event
151n/a loop object used by the future. If it's not provided, the future uses
152n/a the default event loop.
153n/a """
154n/a if loop is None:
155n/a self._loop = events.get_event_loop()
156n/a else:
157n/a self._loop = loop
158n/a self._callbacks = []
159n/a if self._loop.get_debug():
160n/a self._source_traceback = traceback.extract_stack(sys._getframe(1))
161n/a
162n/a _repr_info = base_futures._future_repr_info
163n/a
164n/a def __repr__(self):
165n/a return '<%s %s>' % (self.__class__.__name__, ' '.join(self._repr_info()))
166n/a
167n/a # On Python 3.3 and older, objects with a destructor part of a reference
168n/a # cycle are never destroyed. It's not more the case on Python 3.4 thanks
169n/a # to the PEP 442.
170n/a if compat.PY34:
171n/a def __del__(self):
172n/a if not self._log_traceback:
173n/a # set_exception() was not called, or result() or exception()
174n/a # has consumed the exception
175n/a return
176n/a exc = self._exception
177n/a context = {
178n/a 'message': ('%s exception was never retrieved'
179n/a % self.__class__.__name__),
180n/a 'exception': exc,
181n/a 'future': self,
182n/a }
183n/a if self._source_traceback:
184n/a context['source_traceback'] = self._source_traceback
185n/a self._loop.call_exception_handler(context)
186n/a
187n/a def cancel(self):
188n/a """Cancel the future and schedule callbacks.
189n/a
190n/a If the future is already done or cancelled, return False. Otherwise,
191n/a change the future's state to cancelled, schedule the callbacks and
192n/a return True.
193n/a """
194n/a if self._state != _PENDING:
195n/a return False
196n/a self._state = _CANCELLED
197n/a self._schedule_callbacks()
198n/a return True
199n/a
200n/a def _schedule_callbacks(self):
201n/a """Internal: Ask the event loop to call all callbacks.
202n/a
203n/a The callbacks are scheduled to be called as soon as possible. Also
204n/a clears the callback list.
205n/a """
206n/a callbacks = self._callbacks[:]
207n/a if not callbacks:
208n/a return
209n/a
210n/a self._callbacks[:] = []
211n/a for callback in callbacks:
212n/a self._loop.call_soon(callback, self)
213n/a
214n/a def cancelled(self):
215n/a """Return True if the future was cancelled."""
216n/a return self._state == _CANCELLED
217n/a
218n/a # Don't implement running(); see http://bugs.python.org/issue18699
219n/a
220n/a def done(self):
221n/a """Return True if the future is done.
222n/a
223n/a Done means either that a result / exception are available, or that the
224n/a future was cancelled.
225n/a """
226n/a return self._state != _PENDING
227n/a
228n/a def result(self):
229n/a """Return the result this future represents.
230n/a
231n/a If the future has been cancelled, raises CancelledError. If the
232n/a future's result isn't yet available, raises InvalidStateError. If
233n/a the future is done and has an exception set, this exception is raised.
234n/a """
235n/a if self._state == _CANCELLED:
236n/a raise CancelledError
237n/a if self._state != _FINISHED:
238n/a raise InvalidStateError('Result is not ready.')
239n/a self._log_traceback = False
240n/a if self._tb_logger is not None:
241n/a self._tb_logger.clear()
242n/a self._tb_logger = None
243n/a if self._exception is not None:
244n/a raise self._exception
245n/a return self._result
246n/a
247n/a def exception(self):
248n/a """Return the exception that was set on this future.
249n/a
250n/a The exception (or None if no exception was set) is returned only if
251n/a the future is done. If the future has been cancelled, raises
252n/a CancelledError. If the future isn't done yet, raises
253n/a InvalidStateError.
254n/a """
255n/a if self._state == _CANCELLED:
256n/a raise CancelledError
257n/a if self._state != _FINISHED:
258n/a raise InvalidStateError('Exception is not set.')
259n/a self._log_traceback = False
260n/a if self._tb_logger is not None:
261n/a self._tb_logger.clear()
262n/a self._tb_logger = None
263n/a return self._exception
264n/a
265n/a def add_done_callback(self, fn):
266n/a """Add a callback to be run when the future becomes done.
267n/a
268n/a The callback is called with a single argument - the future object. If
269n/a the future is already done when this is called, the callback is
270n/a scheduled with call_soon.
271n/a """
272n/a if self._state != _PENDING:
273n/a self._loop.call_soon(fn, self)
274n/a else:
275n/a self._callbacks.append(fn)
276n/a
277n/a # New method not in PEP 3148.
278n/a
279n/a def remove_done_callback(self, fn):
280n/a """Remove all instances of a callback from the "call when done" list.
281n/a
282n/a Returns the number of callbacks removed.
283n/a """
284n/a filtered_callbacks = [f for f in self._callbacks if f != fn]
285n/a removed_count = len(self._callbacks) - len(filtered_callbacks)
286n/a if removed_count:
287n/a self._callbacks[:] = filtered_callbacks
288n/a return removed_count
289n/a
290n/a # So-called internal methods (note: no set_running_or_notify_cancel()).
291n/a
292n/a def set_result(self, result):
293n/a """Mark the future done and set its result.
294n/a
295n/a If the future is already done when this method is called, raises
296n/a InvalidStateError.
297n/a """
298n/a if self._state != _PENDING:
299n/a raise InvalidStateError('{}: {!r}'.format(self._state, self))
300n/a self._result = result
301n/a self._state = _FINISHED
302n/a self._schedule_callbacks()
303n/a
304n/a def set_exception(self, exception):
305n/a """Mark the future done and set an exception.
306n/a
307n/a If the future is already done when this method is called, raises
308n/a InvalidStateError.
309n/a """
310n/a if self._state != _PENDING:
311n/a raise InvalidStateError('{}: {!r}'.format(self._state, self))
312n/a if isinstance(exception, type):
313n/a exception = exception()
314n/a if type(exception) is StopIteration:
315n/a raise TypeError("StopIteration interacts badly with generators "
316n/a "and cannot be raised into a Future")
317n/a self._exception = exception
318n/a self._state = _FINISHED
319n/a self._schedule_callbacks()
320n/a if compat.PY34:
321n/a self._log_traceback = True
322n/a else:
323n/a self._tb_logger = _TracebackLogger(self, exception)
324n/a # Arrange for the logger to be activated after all callbacks
325n/a # have had a chance to call result() or exception().
326n/a self._loop.call_soon(self._tb_logger.activate)
327n/a
328n/a def __iter__(self):
329n/a if not self.done():
330n/a self._asyncio_future_blocking = True
331n/a yield self # This tells Task to wait for completion.
332n/a assert self.done(), "yield from wasn't used with future"
333n/a return self.result() # May raise too.
334n/a
335n/a if compat.PY35:
336n/a __await__ = __iter__ # make compatible with 'await' expression
337n/a
338n/a
339n/a# Needed for testing purposes.
340n/a_PyFuture = Future
341n/a
342n/a
343n/adef _set_result_unless_cancelled(fut, result):
344n/a """Helper setting the result only if the future was not cancelled."""
345n/a if fut.cancelled():
346n/a return
347n/a fut.set_result(result)
348n/a
349n/a
350n/adef _set_concurrent_future_state(concurrent, source):
351n/a """Copy state from a future to a concurrent.futures.Future."""
352n/a assert source.done()
353n/a if source.cancelled():
354n/a concurrent.cancel()
355n/a if not concurrent.set_running_or_notify_cancel():
356n/a return
357n/a exception = source.exception()
358n/a if exception is not None:
359n/a concurrent.set_exception(exception)
360n/a else:
361n/a result = source.result()
362n/a concurrent.set_result(result)
363n/a
364n/a
365n/adef _copy_future_state(source, dest):
366n/a """Internal helper to copy state from another Future.
367n/a
368n/a The other Future may be a concurrent.futures.Future.
369n/a """
370n/a assert source.done()
371n/a if dest.cancelled():
372n/a return
373n/a assert not dest.done()
374n/a if source.cancelled():
375n/a dest.cancel()
376n/a else:
377n/a exception = source.exception()
378n/a if exception is not None:
379n/a dest.set_exception(exception)
380n/a else:
381n/a result = source.result()
382n/a dest.set_result(result)
383n/a
384n/a
385n/adef _chain_future(source, destination):
386n/a """Chain two futures so that when one completes, so does the other.
387n/a
388n/a The result (or exception) of source will be copied to destination.
389n/a If destination is cancelled, source gets cancelled too.
390n/a Compatible with both asyncio.Future and concurrent.futures.Future.
391n/a """
392n/a if not isfuture(source) and not isinstance(source,
393n/a concurrent.futures.Future):
394n/a raise TypeError('A future is required for source argument')
395n/a if not isfuture(destination) and not isinstance(destination,
396n/a concurrent.futures.Future):
397n/a raise TypeError('A future is required for destination argument')
398n/a source_loop = source._loop if isfuture(source) else None
399n/a dest_loop = destination._loop if isfuture(destination) else None
400n/a
401n/a def _set_state(future, other):
402n/a if isfuture(future):
403n/a _copy_future_state(other, future)
404n/a else:
405n/a _set_concurrent_future_state(future, other)
406n/a
407n/a def _call_check_cancel(destination):
408n/a if destination.cancelled():
409n/a if source_loop is None or source_loop is dest_loop:
410n/a source.cancel()
411n/a else:
412n/a source_loop.call_soon_threadsafe(source.cancel)
413n/a
414n/a def _call_set_state(source):
415n/a if dest_loop is None or dest_loop is source_loop:
416n/a _set_state(destination, source)
417n/a else:
418n/a dest_loop.call_soon_threadsafe(_set_state, destination, source)
419n/a
420n/a destination.add_done_callback(_call_check_cancel)
421n/a source.add_done_callback(_call_set_state)
422n/a
423n/a
424n/adef wrap_future(future, *, loop=None):
425n/a """Wrap concurrent.futures.Future object."""
426n/a if isfuture(future):
427n/a return future
428n/a assert isinstance(future, concurrent.futures.Future), \
429n/a 'concurrent.futures.Future is expected, got {!r}'.format(future)
430n/a if loop is None:
431n/a loop = events.get_event_loop()
432n/a new_future = loop.create_future()
433n/a _chain_future(future, new_future)
434n/a return new_future
435n/a
436n/a
437n/atry:
438n/a import _asyncio
439n/aexcept ImportError:
440n/a pass
441n/aelse:
442n/a # _CFuture is needed for tests.
443n/a Future = _CFuture = _asyncio.Future