Python code coverage for Lib/asyncio/futures.py

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