Python code coverage for Lib/asyncio/tasks.py

#	count	content
1	n/a	"""Support for tasks, coroutines and the scheduler."""
2	n/a
3	n/a	__all__ = ['Task',
4	n/a	'FIRST_COMPLETED', 'FIRST_EXCEPTION', 'ALL_COMPLETED',
5	n/a	'wait', 'wait_for', 'as_completed', 'sleep', 'async',
6	n/a	'gather', 'shield', 'ensure_future', 'run_coroutine_threadsafe',
7	n/a	]
8	n/a
9	n/a	import concurrent.futures
10	n/a	import functools
11	n/a	import inspect
12	n/a	import warnings
13	n/a	import weakref
14	n/a
15	n/a	from . import base_tasks
16	n/a	from . import compat
17	n/a	from . import coroutines
18	n/a	from . import events
19	n/a	from . import futures
20	n/a	from .coroutines import coroutine
21	n/a
22	n/a
23	n/a	class Task(futures.Future):
24	n/a	"""A coroutine wrapped in a Future."""
25	n/a
26	n/a	# An important invariant maintained while a Task not done:
27	n/a	#
28	n/a	# - Either _fut_waiter is None, and _step() is scheduled;
29	n/a	# - or _fut_waiter is some Future, and _step() is not scheduled.
30	n/a	#
31	n/a	# The only transition from the latter to the former is through
32	n/a	# _wakeup(). When _fut_waiter is not None, one of its callbacks
33	n/a	# must be _wakeup().
34	n/a
35	n/a	# Weak set containing all tasks alive.
36	n/a	_all_tasks = weakref.WeakSet()
37	n/a
38	n/a	# Dictionary containing tasks that are currently active in
39	n/a	# all running event loops. {EventLoop: Task}
40	n/a	_current_tasks = {}
41	n/a
42	n/a	# If False, don't log a message if the task is destroyed whereas its
43	n/a	# status is still pending
44	n/a	_log_destroy_pending = True
45	n/a
46	n/a	@classmethod
47	n/a	def current_task(cls, loop=None):
48	n/a	"""Return the currently running task in an event loop or None.
49	n/a
50	n/a	By default the current task for the current event loop is returned.
51	n/a
52	n/a	None is returned when called not in the context of a Task.
53	n/a	"""
54	n/a	if loop is None:
55	n/a	loop = events.get_event_loop()
56	n/a	return cls._current_tasks.get(loop)
57	n/a
58	n/a	@classmethod
59	n/a	def all_tasks(cls, loop=None):
60	n/a	"""Return a set of all tasks for an event loop.
61	n/a
62	n/a	By default all tasks for the current event loop are returned.
63	n/a	"""
64	n/a	if loop is None:
65	n/a	loop = events.get_event_loop()
66	n/a	return {t for t in cls._all_tasks if t._loop is loop}
67	n/a
68	n/a	def __init__(self, coro, *, loop=None):
69	n/a	assert coroutines.iscoroutine(coro), repr(coro)
70	n/a	super().__init__(loop=loop)
71	n/a	if self._source_traceback:
72	n/a	del self._source_traceback[-1]
73	n/a	self._coro = coro
74	n/a	self._fut_waiter = None
75	n/a	self._must_cancel = False
76	n/a	self._loop.call_soon(self._step)
77	n/a	self.__class__._all_tasks.add(self)
78	n/a
79	n/a	# On Python 3.3 or older, objects with a destructor that are part of a
80	n/a	# reference cycle are never destroyed. That's not the case any more on
81	n/a	# Python 3.4 thanks to the PEP 442.
82	n/a	if compat.PY34:
83	n/a	def __del__(self):
84	n/a	if self._state == futures._PENDING and self._log_destroy_pending:
85	n/a	context = {
86	n/a	'task': self,
87	n/a	'message': 'Task was destroyed but it is pending!',
88	n/a	}
89	n/a	if self._source_traceback:
90	n/a	context['source_traceback'] = self._source_traceback
91	n/a	self._loop.call_exception_handler(context)
92	n/a	futures.Future.__del__(self)
93	n/a
94	n/a	def _repr_info(self):
95	n/a	return base_tasks._task_repr_info(self)
96	n/a
97	n/a	def get_stack(self, *, limit=None):
98	n/a	"""Return the list of stack frames for this task's coroutine.
99	n/a
100	n/a	If the coroutine is not done, this returns the stack where it is
101	n/a	suspended. If the coroutine has completed successfully or was
102	n/a	cancelled, this returns an empty list. If the coroutine was
103	n/a	terminated by an exception, this returns the list of traceback
104	n/a	frames.
105	n/a
106	n/a	The frames are always ordered from oldest to newest.
107	n/a
108	n/a	The optional limit gives the maximum number of frames to
109	n/a	return; by default all available frames are returned. Its
110	n/a	meaning differs depending on whether a stack or a traceback is
111	n/a	returned: the newest frames of a stack are returned, but the
112	n/a	oldest frames of a traceback are returned. (This matches the
113	n/a	behavior of the traceback module.)
114	n/a
115	n/a	For reasons beyond our control, only one stack frame is
116	n/a	returned for a suspended coroutine.
117	n/a	"""
118	n/a	return base_tasks._task_get_stack(self, limit)
119	n/a
120	n/a	def print_stack(self, *, limit=None, file=None):
121	n/a	"""Print the stack or traceback for this task's coroutine.
122	n/a
123	n/a	This produces output similar to that of the traceback module,
124	n/a	for the frames retrieved by get_stack(). The limit argument
125	n/a	is passed to get_stack(). The file argument is an I/O stream
126	n/a	to which the output is written; by default output is written
127	n/a	to sys.stderr.
128	n/a	"""
129	n/a	return base_tasks._task_print_stack(self, limit, file)
130	n/a
131	n/a	def cancel(self):
132	n/a	"""Request that this task cancel itself.
133	n/a
134	n/a	This arranges for a CancelledError to be thrown into the
135	n/a	wrapped coroutine on the next cycle through the event loop.
136	n/a	The coroutine then has a chance to clean up or even deny
137	n/a	the request using try/except/finally.
138	n/a
139	n/a	Unlike Future.cancel, this does not guarantee that the
140	n/a	task will be cancelled: the exception might be caught and
141	n/a	acted upon, delaying cancellation of the task or preventing
142	n/a	cancellation completely. The task may also return a value or
143	n/a	raise a different exception.
144	n/a
145	n/a	Immediately after this method is called, Task.cancelled() will
146	n/a	not return True (unless the task was already cancelled). A
147	n/a	task will be marked as cancelled when the wrapped coroutine
148	n/a	terminates with a CancelledError exception (even if cancel()
149	n/a	was not called).
150	n/a	"""
151	n/a	if self.done():
152	n/a	return False
153	n/a	if self._fut_waiter is not None:
154	n/a	if self._fut_waiter.cancel():
155	n/a	# Leave self._fut_waiter; it may be a Task that
156	n/a	# catches and ignores the cancellation so we may have
157	n/a	# to cancel it again later.
158	n/a	return True
159	n/a	# It must be the case that self._step is already scheduled.
160	n/a	self._must_cancel = True
161	n/a	return True
162	n/a
163	n/a	def _step(self, exc=None):
164	n/a	assert not self.done(), \
165	n/a	'_step(): already done: {!r}, {!r}'.format(self, exc)
166	n/a	if self._must_cancel:
167	n/a	if not isinstance(exc, futures.CancelledError):
168	n/a	exc = futures.CancelledError()
169	n/a	self._must_cancel = False
170	n/a	coro = self._coro
171	n/a	self._fut_waiter = None
172	n/a
173	n/a	self.__class__._current_tasks[self._loop] = self
174	n/a	# Call either coro.throw(exc) or coro.send(None).
175	n/a	try:
176	n/a	if exc is None:
177	n/a	# We use the `send` method directly, because coroutines
178	n/a	# don't have `__iter__` and `__next__` methods.
179	n/a	result = coro.send(None)
180	n/a	else:
181	n/a	result = coro.throw(exc)
182	n/a	except StopIteration as exc:
183	n/a	self.set_result(exc.value)
184	n/a	except futures.CancelledError:
185	n/a	super().cancel() # I.e., Future.cancel(self).
186	n/a	except Exception as exc:
187	n/a	self.set_exception(exc)
188	n/a	except BaseException as exc:
189	n/a	self.set_exception(exc)
190	n/a	raise
191	n/a	else:
192	n/a	blocking = getattr(result, '_asyncio_future_blocking', None)
193	n/a	if blocking is not None:
194	n/a	# Yielded Future must come from Future.__iter__().
195	n/a	if result._loop is not self._loop:
196	n/a	self._loop.call_soon(
197	n/a	self._step,
198	n/a	RuntimeError(
199	n/a	'Task {!r} got Future {!r} attached to a '
200	n/a	'different loop'.format(self, result)))
201	n/a	elif blocking:
202	n/a	if result is self:
203	n/a	self._loop.call_soon(
204	n/a	self._step,
205	n/a	RuntimeError(
206	n/a	'Task cannot await on itself: {!r}'.format(
207	n/a	self)))
208	n/a	else:
209	n/a	result._asyncio_future_blocking = False
210	n/a	result.add_done_callback(self._wakeup)
211	n/a	self._fut_waiter = result
212	n/a	if self._must_cancel:
213	n/a	if self._fut_waiter.cancel():
214	n/a	self._must_cancel = False
215	n/a	else:
216	n/a	self._loop.call_soon(
217	n/a	self._step,
218	n/a	RuntimeError(
219	n/a	'yield was used instead of yield from '
220	n/a	'in task {!r} with {!r}'.format(self, result)))
221	n/a	elif result is None:
222	n/a	# Bare yield relinquishes control for one event loop iteration.
223	n/a	self._loop.call_soon(self._step)
224	n/a	elif inspect.isgenerator(result):
225	n/a	# Yielding a generator is just wrong.
226	n/a	self._loop.call_soon(
227	n/a	self._step,
228	n/a	RuntimeError(
229	n/a	'yield was used instead of yield from for '
230	n/a	'generator in task {!r} with {}'.format(
231	n/a	self, result)))
232	n/a	else:
233	n/a	# Yielding something else is an error.
234	n/a	self._loop.call_soon(
235	n/a	self._step,
236	n/a	RuntimeError(
237	n/a	'Task got bad yield: {!r}'.format(result)))
238	n/a	finally:
239	n/a	self.__class__._current_tasks.pop(self._loop)
240	n/a	self = None # Needed to break cycles when an exception occurs.
241	n/a
242	n/a	def _wakeup(self, future):
243	n/a	try:
244	n/a	future.result()
245	n/a	except Exception as exc:
246	n/a	# This may also be a cancellation.
247	n/a	self._step(exc)
248	n/a	else:
249	n/a	# Don't pass the value of `future.result()` explicitly,
250	n/a	# as `Future.__iter__` and `Future.__await__` don't need it.
251	n/a	# If we call `_step(value, None)` instead of `_step()`,
252	n/a	# Python eval loop would use `.send(value)` method call,
253	n/a	# instead of `__next__()`, which is slower for futures
254	n/a	# that return non-generator iterators from their `__iter__`.
255	n/a	self._step()
256	n/a	self = None # Needed to break cycles when an exception occurs.
257	n/a
258	n/a
259	n/a	_PyTask = Task
260	n/a
261	n/a
262	n/a	try:
263	n/a	import _asyncio
264	n/a	except ImportError:
265	n/a	pass
266	n/a	else:
267	n/a	# _CTask is needed for tests.
268	n/a	Task = _CTask = _asyncio.Task
269	n/a
270	n/a
271	n/a	# wait() and as_completed() similar to those in PEP 3148.
272	n/a
273	n/a	FIRST_COMPLETED = concurrent.futures.FIRST_COMPLETED
274	n/a	FIRST_EXCEPTION = concurrent.futures.FIRST_EXCEPTION
275	n/a	ALL_COMPLETED = concurrent.futures.ALL_COMPLETED
276	n/a
277	n/a
278	n/a	@coroutine
279	n/a	def wait(fs, *, loop=None, timeout=None, return_when=ALL_COMPLETED):
280	n/a	"""Wait for the Futures and coroutines given by fs to complete.
281	n/a
282	n/a	The sequence futures must not be empty.
283	n/a
284	n/a	Coroutines will be wrapped in Tasks.
285	n/a
286	n/a	Returns two sets of Future: (done, pending).
287	n/a
288	n/a	Usage:
289	n/a
290	n/a	done, pending = yield from asyncio.wait(fs)
291	n/a
292	n/a	Note: This does not raise TimeoutError! Futures that aren't done
293	n/a	when the timeout occurs are returned in the second set.
294	n/a	"""
295	n/a	if futures.isfuture(fs) or coroutines.iscoroutine(fs):
296	n/a	raise TypeError("expect a list of futures, not %s" % type(fs).__name__)
297	n/a	if not fs:
298	n/a	raise ValueError('Set of coroutines/Futures is empty.')
299	n/a	if return_when not in (FIRST_COMPLETED, FIRST_EXCEPTION, ALL_COMPLETED):
300	n/a	raise ValueError('Invalid return_when value: {}'.format(return_when))
301	n/a
302	n/a	if loop is None:
303	n/a	loop = events.get_event_loop()
304	n/a
305	n/a	fs = {ensure_future(f, loop=loop) for f in set(fs)}
306	n/a
307	n/a	return (yield from _wait(fs, timeout, return_when, loop))
308	n/a
309	n/a
310	n/a	def _release_waiter(waiter, *args):
311	n/a	if not waiter.done():
312	n/a	waiter.set_result(None)
313	n/a
314	n/a
315	n/a	@coroutine
316	n/a	def wait_for(fut, timeout, *, loop=None):
317	n/a	"""Wait for the single Future or coroutine to complete, with timeout.
318	n/a
319	n/a	Coroutine will be wrapped in Task.
320	n/a
321	n/a	Returns result of the Future or coroutine. When a timeout occurs,
322	n/a	it cancels the task and raises TimeoutError. To avoid the task
323	n/a	cancellation, wrap it in shield().
324	n/a
325	n/a	If the wait is cancelled, the task is also cancelled.
326	n/a
327	n/a	This function is a coroutine.
328	n/a	"""
329	n/a	if loop is None:
330	n/a	loop = events.get_event_loop()
331	n/a
332	n/a	if timeout is None:
333	n/a	return (yield from fut)
334	n/a
335	n/a	waiter = loop.create_future()
336	n/a	timeout_handle = loop.call_later(timeout, _release_waiter, waiter)
337	n/a	cb = functools.partial(_release_waiter, waiter)
338	n/a
339	n/a	fut = ensure_future(fut, loop=loop)
340	n/a	fut.add_done_callback(cb)
341	n/a
342	n/a	try:
343	n/a	# wait until the future completes or the timeout
344	n/a	try:
345	n/a	yield from waiter
346	n/a	except futures.CancelledError:
347	n/a	fut.remove_done_callback(cb)
348	n/a	fut.cancel()
349	n/a	raise
350	n/a
351	n/a	if fut.done():
352	n/a	return fut.result()
353	n/a	else:
354	n/a	fut.remove_done_callback(cb)
355	n/a	fut.cancel()
356	n/a	raise futures.TimeoutError()
357	n/a	finally:
358	n/a	timeout_handle.cancel()
359	n/a
360	n/a
361	n/a	@coroutine
362	n/a	def _wait(fs, timeout, return_when, loop):
363	n/a	"""Internal helper for wait() and wait_for().
364	n/a
365	n/a	The fs argument must be a collection of Futures.
366	n/a	"""
367	n/a	assert fs, 'Set of Futures is empty.'
368	n/a	waiter = loop.create_future()
369	n/a	timeout_handle = None
370	n/a	if timeout is not None:
371	n/a	timeout_handle = loop.call_later(timeout, _release_waiter, waiter)
372	n/a	counter = len(fs)
373	n/a
374	n/a	def _on_completion(f):
375	n/a	nonlocal counter
376	n/a	counter -= 1
377	n/a	if (counter <= 0 or
378	n/a	return_when == FIRST_COMPLETED or
379	n/a	return_when == FIRST_EXCEPTION and (not f.cancelled() and
380	n/a	f.exception() is not None)):
381	n/a	if timeout_handle is not None:
382	n/a	timeout_handle.cancel()
383	n/a	if not waiter.done():
384	n/a	waiter.set_result(None)
385	n/a
386	n/a	for f in fs:
387	n/a	f.add_done_callback(_on_completion)
388	n/a
389	n/a	try:
390	n/a	yield from waiter
391	n/a	finally:
392	n/a	if timeout_handle is not None:
393	n/a	timeout_handle.cancel()
394	n/a
395	n/a	done, pending = set(), set()
396	n/a	for f in fs:
397	n/a	f.remove_done_callback(_on_completion)
398	n/a	if f.done():
399	n/a	done.add(f)
400	n/a	else:
401	n/a	pending.add(f)
402	n/a	return done, pending
403	n/a
404	n/a
405	n/a	# This is not a @coroutine! It is just an iterator (yielding Futures).
406	n/a	def as_completed(fs, *, loop=None, timeout=None):
407	n/a	"""Return an iterator whose values are coroutines.
408	n/a
409	n/a	When waiting for the yielded coroutines you'll get the results (or
410	n/a	exceptions!) of the original Futures (or coroutines), in the order
411	n/a	in which and as soon as they complete.
412	n/a
413	n/a	This differs from PEP 3148; the proper way to use this is:
414	n/a
415	n/a	for f in as_completed(fs):
416	n/a	result = yield from f # The 'yield from' may raise.
417	n/a	# Use result.
418	n/a
419	n/a	If a timeout is specified, the 'yield from' will raise
420	n/a	TimeoutError when the timeout occurs before all Futures are done.
421	n/a
422	n/a	Note: The futures 'f' are not necessarily members of fs.
423	n/a	"""
424	n/a	if futures.isfuture(fs) or coroutines.iscoroutine(fs):
425	n/a	raise TypeError("expect a list of futures, not %s" % type(fs).__name__)
426	n/a	loop = loop if loop is not None else events.get_event_loop()
427	n/a	todo = {ensure_future(f, loop=loop) for f in set(fs)}
428	n/a	from .queues import Queue # Import here to avoid circular import problem.
429	n/a	done = Queue(loop=loop)
430	n/a	timeout_handle = None
431	n/a
432	n/a	def _on_timeout():
433	n/a	for f in todo:
434	n/a	f.remove_done_callback(_on_completion)
435	n/a	done.put_nowait(None) # Queue a dummy value for _wait_for_one().
436	n/a	todo.clear() # Can't do todo.remove(f) in the loop.
437	n/a
438	n/a	def _on_completion(f):
439	n/a	if not todo:
440	n/a	return # _on_timeout() was here first.
441	n/a	todo.remove(f)
442	n/a	done.put_nowait(f)
443	n/a	if not todo and timeout_handle is not None:
444	n/a	timeout_handle.cancel()
445	n/a
446	n/a	@coroutine
447	n/a	def _wait_for_one():
448	n/a	f = yield from done.get()
449	n/a	if f is None:
450	n/a	# Dummy value from _on_timeout().
451	n/a	raise futures.TimeoutError
452	n/a	return f.result() # May raise f.exception().
453	n/a
454	n/a	for f in todo:
455	n/a	f.add_done_callback(_on_completion)
456	n/a	if todo and timeout is not None:
457	n/a	timeout_handle = loop.call_later(timeout, _on_timeout)
458	n/a	for _ in range(len(todo)):
459	n/a	yield _wait_for_one()
460	n/a
461	n/a
462	n/a	@coroutine
463	n/a	def sleep(delay, result=None, *, loop=None):
464	n/a	"""Coroutine that completes after a given time (in seconds)."""
465	n/a	if delay == 0:
466	n/a	yield
467	n/a	return result
468	n/a
469	n/a	if loop is None:
470	n/a	loop = events.get_event_loop()
471	n/a	future = loop.create_future()
472	n/a	h = future._loop.call_later(delay,
473	n/a	futures._set_result_unless_cancelled,
474	n/a	future, result)
475	n/a	try:
476	n/a	return (yield from future)
477	n/a	finally:
478	n/a	h.cancel()
479	n/a
480	n/a
481	n/a	def async_(coro_or_future, *, loop=None):
482	n/a	"""Wrap a coroutine in a future.
483	n/a
484	n/a	If the argument is a Future, it is returned directly.
485	n/a
486	n/a	This function is deprecated in 3.5. Use asyncio.ensure_future() instead.
487	n/a	"""
488	n/a
489	n/a	warnings.warn("asyncio.async() function is deprecated, use ensure_future()",
490	n/a	DeprecationWarning,
491	n/a	stacklevel=2)
492	n/a
493	n/a	return ensure_future(coro_or_future, loop=loop)
494	n/a
495	n/a	# Silence DeprecationWarning:
496	n/a	globals()['async'] = async_
497	n/a	async_.__name__ = 'async'
498	n/a	del async_
499	n/a
500	n/a
501	n/a	def ensure_future(coro_or_future, *, loop=None):
502	n/a	"""Wrap a coroutine or an awaitable in a future.
503	n/a
504	n/a	If the argument is a Future, it is returned directly.
505	n/a	"""
506	n/a	if futures.isfuture(coro_or_future):
507	n/a	if loop is not None and loop is not coro_or_future._loop:
508	n/a	raise ValueError('loop argument must agree with Future')
509	n/a	return coro_or_future
510	n/a	elif coroutines.iscoroutine(coro_or_future):
511	n/a	if loop is None:
512	n/a	loop = events.get_event_loop()
513	n/a	task = loop.create_task(coro_or_future)
514	n/a	if task._source_traceback:
515	n/a	del task._source_traceback[-1]
516	n/a	return task
517	n/a	elif compat.PY35 and inspect.isawaitable(coro_or_future):
518	n/a	return ensure_future(_wrap_awaitable(coro_or_future), loop=loop)
519	n/a	else:
520	n/a	raise TypeError('A Future, a coroutine or an awaitable is required')
521	n/a
522	n/a
523	n/a	@coroutine
524	n/a	def _wrap_awaitable(awaitable):
525	n/a	"""Helper for asyncio.ensure_future().
526	n/a
527	n/a	Wraps awaitable (an object with __await__) into a coroutine
528	n/a	that will later be wrapped in a Task by ensure_future().
529	n/a	"""
530	n/a	return (yield from awaitable.__await__())
531	n/a
532	n/a
533	n/a	class _GatheringFuture(futures.Future):
534	n/a	"""Helper for gather().
535	n/a
536	n/a	This overrides cancel() to cancel all the children and act more
537	n/a	like Task.cancel(), which doesn't immediately mark itself as
538	n/a	cancelled.
539	n/a	"""
540	n/a
541	n/a	def __init__(self, children, *, loop=None):
542	n/a	super().__init__(loop=loop)
543	n/a	self._children = children
544	n/a
545	n/a	def cancel(self):
546	n/a	if self.done():
547	n/a	return False
548	n/a	ret = False
549	n/a	for child in self._children:
550	n/a	if child.cancel():
551	n/a	ret = True
552	n/a	return ret
553	n/a
554	n/a
555	n/a	def gather(*coros_or_futures, loop=None, return_exceptions=False):
556	n/a	"""Return a future aggregating results from the given coroutines
557	n/a	or futures.
558	n/a
559	n/a	Coroutines will be wrapped in a future and scheduled in the event
560	n/a	loop. They will not necessarily be scheduled in the same order as
561	n/a	passed in.
562	n/a
563	n/a	All futures must share the same event loop. If all the tasks are
564	n/a	done successfully, the returned future's result is the list of
565	n/a	results (in the order of the original sequence, not necessarily
566	n/a	the order of results arrival). If return_exceptions is True,
567	n/a	exceptions in the tasks are treated the same as successful
568	n/a	results, and gathered in the result list; otherwise, the first
569	n/a	raised exception will be immediately propagated to the returned
570	n/a	future.
571	n/a
572	n/a	Cancellation: if the outer Future is cancelled, all children (that
573	n/a	have not completed yet) are also cancelled. If any child is
574	n/a	cancelled, this is treated as if it raised CancelledError --
575	n/a	the outer Future is not cancelled in this case. (This is to
576	n/a	prevent the cancellation of one child to cause other children to
577	n/a	be cancelled.)
578	n/a	"""
579	n/a	if not coros_or_futures:
580	n/a	if loop is None:
581	n/a	loop = events.get_event_loop()
582	n/a	outer = loop.create_future()
583	n/a	outer.set_result([])
584	n/a	return outer
585	n/a
586	n/a	arg_to_fut = {}
587	n/a	for arg in set(coros_or_futures):
588	n/a	if not futures.isfuture(arg):
589	n/a	fut = ensure_future(arg, loop=loop)
590	n/a	if loop is None:
591	n/a	loop = fut._loop
592	n/a	# The caller cannot control this future, the "destroy pending task"
593	n/a	# warning should not be emitted.
594	n/a	fut._log_destroy_pending = False
595	n/a	else:
596	n/a	fut = arg
597	n/a	if loop is None:
598	n/a	loop = fut._loop
599	n/a	elif fut._loop is not loop:
600	n/a	raise ValueError("futures are tied to different event loops")
601	n/a	arg_to_fut[arg] = fut
602	n/a
603	n/a	children = [arg_to_fut[arg] for arg in coros_or_futures]
604	n/a	nchildren = len(children)
605	n/a	outer = _GatheringFuture(children, loop=loop)
606	n/a	nfinished = 0
607	n/a	results = [None] * nchildren
608	n/a
609	n/a	def _done_callback(i, fut):
610	n/a	nonlocal nfinished
611	n/a	if outer.done():
612	n/a	if not fut.cancelled():
613	n/a	# Mark exception retrieved.
614	n/a	fut.exception()
615	n/a	return
616	n/a
617	n/a	if fut.cancelled():
618	n/a	res = futures.CancelledError()
619	n/a	if not return_exceptions:
620	n/a	outer.set_exception(res)
621	n/a	return
622	n/a	elif fut._exception is not None:
623	n/a	res = fut.exception() # Mark exception retrieved.
624	n/a	if not return_exceptions:
625	n/a	outer.set_exception(res)
626	n/a	return
627	n/a	else:
628	n/a	res = fut._result
629	n/a	results[i] = res
630	n/a	nfinished += 1
631	n/a	if nfinished == nchildren:
632	n/a	outer.set_result(results)
633	n/a
634	n/a	for i, fut in enumerate(children):
635	n/a	fut.add_done_callback(functools.partial(_done_callback, i))
636	n/a	return outer
637	n/a
638	n/a
639	n/a	def shield(arg, *, loop=None):
640	n/a	"""Wait for a future, shielding it from cancellation.
641	n/a
642	n/a	The statement
643	n/a
644	n/a	res = yield from shield(something())
645	n/a
646	n/a	is exactly equivalent to the statement
647	n/a
648	n/a	res = yield from something()
649	n/a
650	n/a	except that if the coroutine containing it is cancelled, the
651	n/a	task running in something() is not cancelled. From the POV of
652	n/a	something(), the cancellation did not happen. But its caller is
653	n/a	still cancelled, so the yield-from expression still raises
654	n/a	CancelledError. Note: If something() is cancelled by other means
655	n/a	this will still cancel shield().
656	n/a
657	n/a	If you want to completely ignore cancellation (not recommended)
658	n/a	you can combine shield() with a try/except clause, as follows:
659	n/a
660	n/a	try:
661	n/a	res = yield from shield(something())
662	n/a	except CancelledError:
663	n/a	res = None
664	n/a	"""
665	n/a	inner = ensure_future(arg, loop=loop)
666	n/a	if inner.done():
667	n/a	# Shortcut.
668	n/a	return inner
669	n/a	loop = inner._loop
670	n/a	outer = loop.create_future()
671	n/a
672	n/a	def _done_callback(inner):
673	n/a	if outer.cancelled():
674	n/a	if not inner.cancelled():
675	n/a	# Mark inner's result as retrieved.
676	n/a	inner.exception()
677	n/a	return
678	n/a
679	n/a	if inner.cancelled():
680	n/a	outer.cancel()
681	n/a	else:
682	n/a	exc = inner.exception()
683	n/a	if exc is not None:
684	n/a	outer.set_exception(exc)
685	n/a	else:
686	n/a	outer.set_result(inner.result())
687	n/a
688	n/a	inner.add_done_callback(_done_callback)
689	n/a	return outer
690	n/a
691	n/a
692	n/a	def run_coroutine_threadsafe(coro, loop):
693	n/a	"""Submit a coroutine object to a given event loop.
694	n/a
695	n/a	Return a concurrent.futures.Future to access the result.
696	n/a	"""
697	n/a	if not coroutines.iscoroutine(coro):
698	n/a	raise TypeError('A coroutine object is required')
699	n/a	future = concurrent.futures.Future()
700	n/a
701	n/a	def callback():
702	n/a	try:
703	n/a	futures._chain_future(ensure_future(coro, loop=loop), future)
704	n/a	except Exception as exc:
705	n/a	if future.set_running_or_notify_cancel():
706	n/a	future.set_exception(exc)
707	n/a	raise
708	n/a
709	n/a	loop.call_soon_threadsafe(callback)
710	n/a	return future