Python code coverage for Lib/concurrent/futures/_base.py

#	count	content
1	n/a	# Copyright 2009 Brian Quinlan. All Rights Reserved.
2	n/a	# Licensed to PSF under a Contributor Agreement.
3	n/a
4	n/a	__author__ = 'Brian Quinlan (brian@sweetapp.com)'
5	n/a
6	n/a	import collections
7	n/a	import logging
8	n/a	import threading
9	n/a	import time
10	n/a
11	n/a	FIRST_COMPLETED = 'FIRST_COMPLETED'
12	n/a	FIRST_EXCEPTION = 'FIRST_EXCEPTION'
13	n/a	ALL_COMPLETED = 'ALL_COMPLETED'
14	n/a	_AS_COMPLETED = '_AS_COMPLETED'
15	n/a
16	n/a	# Possible future states (for internal use by the futures package).
17	n/a	PENDING = 'PENDING'
18	n/a	RUNNING = 'RUNNING'
19	n/a	# The future was cancelled by the user...
20	n/a	CANCELLED = 'CANCELLED'
21	n/a	# ...and _Waiter.add_cancelled() was called by a worker.
22	n/a	CANCELLED_AND_NOTIFIED = 'CANCELLED_AND_NOTIFIED'
23	n/a	FINISHED = 'FINISHED'
24	n/a
25	n/a	_FUTURE_STATES = [
26	n/a	PENDING,
27	n/a	RUNNING,
28	n/a	CANCELLED,
29	n/a	CANCELLED_AND_NOTIFIED,
30	n/a	FINISHED
31	n/a	]
32	n/a
33	n/a	_STATE_TO_DESCRIPTION_MAP = {
34	n/a	PENDING: "pending",
35	n/a	RUNNING: "running",
36	n/a	CANCELLED: "cancelled",
37	n/a	CANCELLED_AND_NOTIFIED: "cancelled",
38	n/a	FINISHED: "finished"
39	n/a	}
40	n/a
41	n/a	# Logger for internal use by the futures package.
42	n/a	LOGGER = logging.getLogger("concurrent.futures")
43	n/a
44	n/a	class Error(Exception):
45	n/a	"""Base class for all future-related exceptions."""
46	n/a	pass
47	n/a
48	n/a	class CancelledError(Error):
49	n/a	"""The Future was cancelled."""
50	n/a	pass
51	n/a
52	n/a	class TimeoutError(Error):
53	n/a	"""The operation exceeded the given deadline."""
54	n/a	pass
55	n/a
56	n/a	class _Waiter(object):
57	n/a	"""Provides the event that wait() and as_completed() block on."""
58	n/a	def __init__(self):
59	n/a	self.event = threading.Event()
60	n/a	self.finished_futures = []
61	n/a
62	n/a	def add_result(self, future):
63	n/a	self.finished_futures.append(future)
64	n/a
65	n/a	def add_exception(self, future):
66	n/a	self.finished_futures.append(future)
67	n/a
68	n/a	def add_cancelled(self, future):
69	n/a	self.finished_futures.append(future)
70	n/a
71	n/a	class _AsCompletedWaiter(_Waiter):
72	n/a	"""Used by as_completed()."""
73	n/a
74	n/a	def __init__(self):
75	n/a	super(_AsCompletedWaiter, self).__init__()
76	n/a	self.lock = threading.Lock()
77	n/a
78	n/a	def add_result(self, future):
79	n/a	with self.lock:
80	n/a	super(_AsCompletedWaiter, self).add_result(future)
81	n/a	self.event.set()
82	n/a
83	n/a	def add_exception(self, future):
84	n/a	with self.lock:
85	n/a	super(_AsCompletedWaiter, self).add_exception(future)
86	n/a	self.event.set()
87	n/a
88	n/a	def add_cancelled(self, future):
89	n/a	with self.lock:
90	n/a	super(_AsCompletedWaiter, self).add_cancelled(future)
91	n/a	self.event.set()
92	n/a
93	n/a	class _FirstCompletedWaiter(_Waiter):
94	n/a	"""Used by wait(return_when=FIRST_COMPLETED)."""
95	n/a
96	n/a	def add_result(self, future):
97	n/a	super().add_result(future)
98	n/a	self.event.set()
99	n/a
100	n/a	def add_exception(self, future):
101	n/a	super().add_exception(future)
102	n/a	self.event.set()
103	n/a
104	n/a	def add_cancelled(self, future):
105	n/a	super().add_cancelled(future)
106	n/a	self.event.set()
107	n/a
108	n/a	class _AllCompletedWaiter(_Waiter):
109	n/a	"""Used by wait(return_when=FIRST_EXCEPTION and ALL_COMPLETED)."""
110	n/a
111	n/a	def __init__(self, num_pending_calls, stop_on_exception):
112	n/a	self.num_pending_calls = num_pending_calls
113	n/a	self.stop_on_exception = stop_on_exception
114	n/a	self.lock = threading.Lock()
115	n/a	super().__init__()
116	n/a
117	n/a	def _decrement_pending_calls(self):
118	n/a	with self.lock:
119	n/a	self.num_pending_calls -= 1
120	n/a	if not self.num_pending_calls:
121	n/a	self.event.set()
122	n/a
123	n/a	def add_result(self, future):
124	n/a	super().add_result(future)
125	n/a	self._decrement_pending_calls()
126	n/a
127	n/a	def add_exception(self, future):
128	n/a	super().add_exception(future)
129	n/a	if self.stop_on_exception:
130	n/a	self.event.set()
131	n/a	else:
132	n/a	self._decrement_pending_calls()
133	n/a
134	n/a	def add_cancelled(self, future):
135	n/a	super().add_cancelled(future)
136	n/a	self._decrement_pending_calls()
137	n/a
138	n/a	class _AcquireFutures(object):
139	n/a	"""A context manager that does an ordered acquire of Future conditions."""
140	n/a
141	n/a	def __init__(self, futures):
142	n/a	self.futures = sorted(futures, key=id)
143	n/a
144	n/a	def __enter__(self):
145	n/a	for future in self.futures:
146	n/a	future._condition.acquire()
147	n/a
148	n/a	def __exit__(self, *args):
149	n/a	for future in self.futures:
150	n/a	future._condition.release()
151	n/a
152	n/a	def _create_and_install_waiters(fs, return_when):
153	n/a	if return_when == _AS_COMPLETED:
154	n/a	waiter = _AsCompletedWaiter()
155	n/a	elif return_when == FIRST_COMPLETED:
156	n/a	waiter = _FirstCompletedWaiter()
157	n/a	else:
158	n/a	pending_count = sum(
159	n/a	f._state not in [CANCELLED_AND_NOTIFIED, FINISHED] for f in fs)
160	n/a
161	n/a	if return_when == FIRST_EXCEPTION:
162	n/a	waiter = _AllCompletedWaiter(pending_count, stop_on_exception=True)
163	n/a	elif return_when == ALL_COMPLETED:
164	n/a	waiter = _AllCompletedWaiter(pending_count, stop_on_exception=False)
165	n/a	else:
166	n/a	raise ValueError("Invalid return condition: %r" % return_when)
167	n/a
168	n/a	for f in fs:
169	n/a	f._waiters.append(waiter)
170	n/a
171	n/a	return waiter
172	n/a
173	n/a	def as_completed(fs, timeout=None):
174	n/a	"""An iterator over the given futures that yields each as it completes.
175	n/a
176	n/a	Args:
177	n/a	fs: The sequence of Futures (possibly created by different Executors) to
178	n/a	iterate over.
179	n/a	timeout: The maximum number of seconds to wait. If None, then there
180	n/a	is no limit on the wait time.
181	n/a
182	n/a	Returns:
183	n/a	An iterator that yields the given Futures as they complete (finished or
184	n/a	cancelled). If any given Futures are duplicated, they will be returned
185	n/a	once.
186	n/a
187	n/a	Raises:
188	n/a	TimeoutError: If the entire result iterator could not be generated
189	n/a	before the given timeout.
190	n/a	"""
191	n/a	if timeout is not None:
192	n/a	end_time = timeout + time.time()
193	n/a
194	n/a	fs = set(fs)
195	n/a	with _AcquireFutures(fs):
196	n/a	finished = set(
197	n/a	f for f in fs
198	n/a	if f._state in [CANCELLED_AND_NOTIFIED, FINISHED])
199	n/a	pending = fs - finished
200	n/a	waiter = _create_and_install_waiters(fs, _AS_COMPLETED)
201	n/a
202	n/a	try:
203	n/a	yield from finished
204	n/a
205	n/a	while pending:
206	n/a	if timeout is None:
207	n/a	wait_timeout = None
208	n/a	else:
209	n/a	wait_timeout = end_time - time.time()
210	n/a	if wait_timeout < 0:
211	n/a	raise TimeoutError(
212	n/a	'%d (of %d) futures unfinished' % (
213	n/a	len(pending), len(fs)))
214	n/a
215	n/a	waiter.event.wait(wait_timeout)
216	n/a
217	n/a	with waiter.lock:
218	n/a	finished = waiter.finished_futures
219	n/a	waiter.finished_futures = []
220	n/a	waiter.event.clear()
221	n/a
222	n/a	for future in finished:
223	n/a	yield future
224	n/a	pending.remove(future)
225	n/a
226	n/a	finally:
227	n/a	for f in fs:
228	n/a	with f._condition:
229	n/a	f._waiters.remove(waiter)
230	n/a
231	n/a	DoneAndNotDoneFutures = collections.namedtuple(
232	n/a	'DoneAndNotDoneFutures', 'done not_done')
233	n/a	def wait(fs, timeout=None, return_when=ALL_COMPLETED):
234	n/a	"""Wait for the futures in the given sequence to complete.
235	n/a
236	n/a	Args:
237	n/a	fs: The sequence of Futures (possibly created by different Executors) to
238	n/a	wait upon.
239	n/a	timeout: The maximum number of seconds to wait. If None, then there
240	n/a	is no limit on the wait time.
241	n/a	return_when: Indicates when this function should return. The options
242	n/a	are:
243	n/a
244	n/a	FIRST_COMPLETED - Return when any future finishes or is
245	n/a	cancelled.
246	n/a	FIRST_EXCEPTION - Return when any future finishes by raising an
247	n/a	exception. If no future raises an exception
248	n/a	then it is equivalent to ALL_COMPLETED.
249	n/a	ALL_COMPLETED - Return when all futures finish or are cancelled.
250	n/a
251	n/a	Returns:
252	n/a	A named 2-tuple of sets. The first set, named 'done', contains the
253	n/a	futures that completed (is finished or cancelled) before the wait
254	n/a	completed. The second set, named 'not_done', contains uncompleted
255	n/a	futures.
256	n/a	"""
257	n/a	with _AcquireFutures(fs):
258	n/a	done = set(f for f in fs
259	n/a	if f._state in [CANCELLED_AND_NOTIFIED, FINISHED])
260	n/a	not_done = set(fs) - done
261	n/a
262	n/a	if (return_when == FIRST_COMPLETED) and done:
263	n/a	return DoneAndNotDoneFutures(done, not_done)
264	n/a	elif (return_when == FIRST_EXCEPTION) and done:
265	n/a	if any(f for f in done
266	n/a	if not f.cancelled() and f.exception() is not None):
267	n/a	return DoneAndNotDoneFutures(done, not_done)
268	n/a
269	n/a	if len(done) == len(fs):
270	n/a	return DoneAndNotDoneFutures(done, not_done)
271	n/a
272	n/a	waiter = _create_and_install_waiters(fs, return_when)
273	n/a
274	n/a	waiter.event.wait(timeout)
275	n/a	for f in fs:
276	n/a	with f._condition:
277	n/a	f._waiters.remove(waiter)
278	n/a
279	n/a	done.update(waiter.finished_futures)
280	n/a	return DoneAndNotDoneFutures(done, set(fs) - done)
281	n/a
282	n/a	class Future(object):
283	n/a	"""Represents the result of an asynchronous computation."""
284	n/a
285	n/a	def __init__(self):
286	n/a	"""Initializes the future. Should not be called by clients."""
287	n/a	self._condition = threading.Condition()
288	n/a	self._state = PENDING
289	n/a	self._result = None
290	n/a	self._exception = None
291	n/a	self._waiters = []
292	n/a	self._done_callbacks = []
293	n/a
294	n/a	def _invoke_callbacks(self):
295	n/a	for callback in self._done_callbacks:
296	n/a	try:
297	n/a	callback(self)
298	n/a	except Exception:
299	n/a	LOGGER.exception('exception calling callback for %r', self)
300	n/a
301	n/a	def __repr__(self):
302	n/a	with self._condition:
303	n/a	if self._state == FINISHED:
304	n/a	if self._exception:
305	n/a	return '<%s at %#x state=%s raised %s>' % (
306	n/a	self.__class__.__name__,
307	n/a	id(self),
308	n/a	_STATE_TO_DESCRIPTION_MAP[self._state],
309	n/a	self._exception.__class__.__name__)
310	n/a	else:
311	n/a	return '<%s at %#x state=%s returned %s>' % (
312	n/a	self.__class__.__name__,
313	n/a	id(self),
314	n/a	_STATE_TO_DESCRIPTION_MAP[self._state],
315	n/a	self._result.__class__.__name__)
316	n/a	return '<%s at %#x state=%s>' % (
317	n/a	self.__class__.__name__,
318	n/a	id(self),
319	n/a	_STATE_TO_DESCRIPTION_MAP[self._state])
320	n/a
321	n/a	def cancel(self):
322	n/a	"""Cancel the future if possible.
323	n/a
324	n/a	Returns True if the future was cancelled, False otherwise. A future
325	n/a	cannot be cancelled if it is running or has already completed.
326	n/a	"""
327	n/a	with self._condition:
328	n/a	if self._state in [RUNNING, FINISHED]:
329	n/a	return False
330	n/a
331	n/a	if self._state in [CANCELLED, CANCELLED_AND_NOTIFIED]:
332	n/a	return True
333	n/a
334	n/a	self._state = CANCELLED
335	n/a	self._condition.notify_all()
336	n/a
337	n/a	self._invoke_callbacks()
338	n/a	return True
339	n/a
340	n/a	def cancelled(self):
341	n/a	"""Return True if the future was cancelled."""
342	n/a	with self._condition:
343	n/a	return self._state in [CANCELLED, CANCELLED_AND_NOTIFIED]
344	n/a
345	n/a	def running(self):
346	n/a	"""Return True if the future is currently executing."""
347	n/a	with self._condition:
348	n/a	return self._state == RUNNING
349	n/a
350	n/a	def done(self):
351	n/a	"""Return True of the future was cancelled or finished executing."""
352	n/a	with self._condition:
353	n/a	return self._state in [CANCELLED, CANCELLED_AND_NOTIFIED, FINISHED]
354	n/a
355	n/a	def __get_result(self):
356	n/a	if self._exception:
357	n/a	raise self._exception
358	n/a	else:
359	n/a	return self._result
360	n/a
361	n/a	def add_done_callback(self, fn):
362	n/a	"""Attaches a callable that will be called when the future finishes.
363	n/a
364	n/a	Args:
365	n/a	fn: A callable that will be called with this future as its only
366	n/a	argument when the future completes or is cancelled. The callable
367	n/a	will always be called by a thread in the same process in which
368	n/a	it was added. If the future has already completed or been
369	n/a	cancelled then the callable will be called immediately. These
370	n/a	callables are called in the order that they were added.
371	n/a	"""
372	n/a	with self._condition:
373	n/a	if self._state not in [CANCELLED, CANCELLED_AND_NOTIFIED, FINISHED]:
374	n/a	self._done_callbacks.append(fn)
375	n/a	return
376	n/a	fn(self)
377	n/a
378	n/a	def result(self, timeout=None):
379	n/a	"""Return the result of the call that the future represents.
380	n/a
381	n/a	Args:
382	n/a	timeout: The number of seconds to wait for the result if the future
383	n/a	isn't done. If None, then there is no limit on the wait time.
384	n/a
385	n/a	Returns:
386	n/a	The result of the call that the future represents.
387	n/a
388	n/a	Raises:
389	n/a	CancelledError: If the future was cancelled.
390	n/a	TimeoutError: If the future didn't finish executing before the given
391	n/a	timeout.
392	n/a	Exception: If the call raised then that exception will be raised.
393	n/a	"""
394	n/a	with self._condition:
395	n/a	if self._state in [CANCELLED, CANCELLED_AND_NOTIFIED]:
396	n/a	raise CancelledError()
397	n/a	elif self._state == FINISHED:
398	n/a	return self.__get_result()
399	n/a
400	n/a	self._condition.wait(timeout)
401	n/a
402	n/a	if self._state in [CANCELLED, CANCELLED_AND_NOTIFIED]:
403	n/a	raise CancelledError()
404	n/a	elif self._state == FINISHED:
405	n/a	return self.__get_result()
406	n/a	else:
407	n/a	raise TimeoutError()
408	n/a
409	n/a	def exception(self, timeout=None):
410	n/a	"""Return the exception raised by the call that the future represents.
411	n/a
412	n/a	Args:
413	n/a	timeout: The number of seconds to wait for the exception if the
414	n/a	future isn't done. If None, then there is no limit on the wait
415	n/a	time.
416	n/a
417	n/a	Returns:
418	n/a	The exception raised by the call that the future represents or None
419	n/a	if the call completed without raising.
420	n/a
421	n/a	Raises:
422	n/a	CancelledError: If the future was cancelled.
423	n/a	TimeoutError: If the future didn't finish executing before the given
424	n/a	timeout.
425	n/a	"""
426	n/a
427	n/a	with self._condition:
428	n/a	if self._state in [CANCELLED, CANCELLED_AND_NOTIFIED]:
429	n/a	raise CancelledError()
430	n/a	elif self._state == FINISHED:
431	n/a	return self._exception
432	n/a
433	n/a	self._condition.wait(timeout)
434	n/a
435	n/a	if self._state in [CANCELLED, CANCELLED_AND_NOTIFIED]:
436	n/a	raise CancelledError()
437	n/a	elif self._state == FINISHED:
438	n/a	return self._exception
439	n/a	else:
440	n/a	raise TimeoutError()
441	n/a
442	n/a	# The following methods should only be used by Executors and in tests.
443	n/a	def set_running_or_notify_cancel(self):
444	n/a	"""Mark the future as running or process any cancel notifications.
445	n/a
446	n/a	Should only be used by Executor implementations and unit tests.
447	n/a
448	n/a	If the future has been cancelled (cancel() was called and returned
449	n/a	True) then any threads waiting on the future completing (though calls
450	n/a	to as_completed() or wait()) are notified and False is returned.
451	n/a
452	n/a	If the future was not cancelled then it is put in the running state
453	n/a	(future calls to running() will return True) and True is returned.
454	n/a
455	n/a	This method should be called by Executor implementations before
456	n/a	executing the work associated with this future. If this method returns
457	n/a	False then the work should not be executed.
458	n/a
459	n/a	Returns:
460	n/a	False if the Future was cancelled, True otherwise.
461	n/a
462	n/a	Raises:
463	n/a	RuntimeError: if this method was already called or if set_result()
464	n/a	or set_exception() was called.
465	n/a	"""
466	n/a	with self._condition:
467	n/a	if self._state == CANCELLED:
468	n/a	self._state = CANCELLED_AND_NOTIFIED
469	n/a	for waiter in self._waiters:
470	n/a	waiter.add_cancelled(self)
471	n/a	# self._condition.notify_all() is not necessary because
472	n/a	# self.cancel() triggers a notification.
473	n/a	return False
474	n/a	elif self._state == PENDING:
475	n/a	self._state = RUNNING
476	n/a	return True
477	n/a	else:
478	n/a	LOGGER.critical('Future %s in unexpected state: %s',
479	n/a	id(self),
480	n/a	self._state)
481	n/a	raise RuntimeError('Future in unexpected state')
482	n/a
483	n/a	def set_result(self, result):
484	n/a	"""Sets the return value of work associated with the future.
485	n/a
486	n/a	Should only be used by Executor implementations and unit tests.
487	n/a	"""
488	n/a	with self._condition:
489	n/a	self._result = result
490	n/a	self._state = FINISHED
491	n/a	for waiter in self._waiters:
492	n/a	waiter.add_result(self)
493	n/a	self._condition.notify_all()
494	n/a	self._invoke_callbacks()
495	n/a
496	n/a	def set_exception(self, exception):
497	n/a	"""Sets the result of the future as being the given exception.
498	n/a
499	n/a	Should only be used by Executor implementations and unit tests.
500	n/a	"""
501	n/a	with self._condition:
502	n/a	self._exception = exception
503	n/a	self._state = FINISHED
504	n/a	for waiter in self._waiters:
505	n/a	waiter.add_exception(self)
506	n/a	self._condition.notify_all()
507	n/a	self._invoke_callbacks()
508	n/a
509	n/a	class Executor(object):
510	n/a	"""This is an abstract base class for concrete asynchronous executors."""
511	n/a
512	n/a	def submit(self, fn, args, *kwargs):
513	n/a	"""Submits a callable to be executed with the given arguments.
514	n/a
515	n/a	Schedules the callable to be executed as fn(args, *kwargs) and returns
516	n/a	a Future instance representing the execution of the callable.
517	n/a
518	n/a	Returns:
519	n/a	A Future representing the given call.
520	n/a	"""
521	n/a	raise NotImplementedError()
522	n/a
523	n/a	def map(self, fn, *iterables, timeout=None, chunksize=1):
524	n/a	"""Returns an iterator equivalent to map(fn, iter).
525	n/a
526	n/a	Args:
527	n/a	fn: A callable that will take as many arguments as there are
528	n/a	passed iterables.
529	n/a	timeout: The maximum number of seconds to wait. If None, then there
530	n/a	is no limit on the wait time.
531	n/a	chunksize: The size of the chunks the iterable will be broken into
532	n/a	before being passed to a child process. This argument is only
533	n/a	used by ProcessPoolExecutor; it is ignored by
534	n/a	ThreadPoolExecutor.
535	n/a
536	n/a	Returns:
537	n/a	An iterator equivalent to: map(func, *iterables) but the calls may
538	n/a	be evaluated out-of-order.
539	n/a
540	n/a	Raises:
541	n/a	TimeoutError: If the entire result iterator could not be generated
542	n/a	before the given timeout.
543	n/a	Exception: If fn(*args) raises for any values.
544	n/a	"""
545	n/a	if timeout is not None:
546	n/a	end_time = timeout + time.time()
547	n/a
548	n/a	fs = [self.submit(fn, args) for args in zip(iterables)]
549	n/a
550	n/a	# Yield must be hidden in closure so that the futures are submitted
551	n/a	# before the first iterator value is required.
552	n/a	def result_iterator():
553	n/a	try:
554	n/a	for future in fs:
555	n/a	if timeout is None:
556	n/a	yield future.result()
557	n/a	else:
558	n/a	yield future.result(end_time - time.time())
559	n/a	finally:
560	n/a	for future in fs:
561	n/a	future.cancel()
562	n/a	return result_iterator()
563	n/a
564	n/a	def shutdown(self, wait=True):
565	n/a	"""Clean-up the resources associated with the Executor.
566	n/a
567	n/a	It is safe to call this method several times. Otherwise, no other
568	n/a	methods can be called after this one.
569	n/a
570	n/a	Args:
571	n/a	wait: If True then shutdown will not return until all running
572	n/a	futures have finished executing and the resources used by the
573	n/a	executor have been reclaimed.
574	n/a	"""
575	n/a	pass
576	n/a
577	n/a	def __enter__(self):
578	n/a	return self
579	n/a
580	n/a	def __exit__(self, exc_type, exc_val, exc_tb):
581	n/a	self.shutdown(wait=True)
582	n/a	return False