Python code coverage for Lib/asyncio/unix_events.py

#	count	content
1	n/a	"""Selector event loop for Unix with signal handling."""
2	n/a
3	n/a	import errno
4	n/a	import os
5	n/a	import signal
6	n/a	import socket
7	n/a	import stat
8	n/a	import subprocess
9	n/a	import sys
10	n/a	import threading
11	n/a	import warnings
12	n/a
13	n/a
14	n/a	from . import base_events
15	n/a	from . import base_subprocess
16	n/a	from . import compat
17	n/a	from . import constants
18	n/a	from . import coroutines
19	n/a	from . import events
20	n/a	from . import futures
21	n/a	from . import selector_events
22	n/a	from . import selectors
23	n/a	from . import transports
24	n/a	from .coroutines import coroutine
25	n/a	from .log import logger
26	n/a
27	n/a
28	n/a	__all__ = ['SelectorEventLoop',
29	n/a	'AbstractChildWatcher', 'SafeChildWatcher',
30	n/a	'FastChildWatcher', 'DefaultEventLoopPolicy',
31	n/a	]
32	n/a
33	n/a	if sys.platform == 'win32': # pragma: no cover
34	n/a	raise ImportError('Signals are not really supported on Windows')
35	n/a
36	n/a
37	n/a	def _sighandler_noop(signum, frame):
38	n/a	"""Dummy signal handler."""
39	n/a	pass
40	n/a
41	n/a
42	n/a	try:
43	n/a	_fspath = os.fspath
44	n/a	except AttributeError:
45	n/a	# Python 3.5 or earlier
46	n/a	_fspath = lambda path: path
47	n/a
48	n/a
49	n/a	class _UnixSelectorEventLoop(selector_events.BaseSelectorEventLoop):
50	n/a	"""Unix event loop.
51	n/a
52	n/a	Adds signal handling and UNIX Domain Socket support to SelectorEventLoop.
53	n/a	"""
54	n/a
55	n/a	def __init__(self, selector=None):
56	n/a	super().__init__(selector)
57	n/a	self._signal_handlers = {}
58	n/a
59	n/a	def _socketpair(self):
60	n/a	return socket.socketpair()
61	n/a
62	n/a	def close(self):
63	n/a	super().close()
64	n/a	for sig in list(self._signal_handlers):
65	n/a	self.remove_signal_handler(sig)
66	n/a
67	n/a	def _process_self_data(self, data):
68	n/a	for signum in data:
69	n/a	if not signum:
70	n/a	# ignore null bytes written by _write_to_self()
71	n/a	continue
72	n/a	self._handle_signal(signum)
73	n/a
74	n/a	def add_signal_handler(self, sig, callback, *args):
75	n/a	"""Add a handler for a signal. UNIX only.
76	n/a
77	n/a	Raise ValueError if the signal number is invalid or uncatchable.
78	n/a	Raise RuntimeError if there is a problem setting up the handler.
79	n/a	"""
80	n/a	if (coroutines.iscoroutine(callback)
81	n/a	or coroutines.iscoroutinefunction(callback)):
82	n/a	raise TypeError("coroutines cannot be used "
83	n/a	"with add_signal_handler()")
84	n/a	self._check_signal(sig)
85	n/a	self._check_closed()
86	n/a	try:
87	n/a	# set_wakeup_fd() raises ValueError if this is not the
88	n/a	# main thread. By calling it early we ensure that an
89	n/a	# event loop running in another thread cannot add a signal
90	n/a	# handler.
91	n/a	signal.set_wakeup_fd(self._csock.fileno())
92	n/a	except (ValueError, OSError) as exc:
93	n/a	raise RuntimeError(str(exc))
94	n/a
95	n/a	handle = events.Handle(callback, args, self)
96	n/a	self._signal_handlers[sig] = handle
97	n/a
98	n/a	try:
99	n/a	# Register a dummy signal handler to ask Python to write the signal
100	n/a	# number in the wakup file descriptor. _process_self_data() will
101	n/a	# read signal numbers from this file descriptor to handle signals.
102	n/a	signal.signal(sig, _sighandler_noop)
103	n/a
104	n/a	# Set SA_RESTART to limit EINTR occurrences.
105	n/a	signal.siginterrupt(sig, False)
106	n/a	except OSError as exc:
107	n/a	del self._signal_handlers[sig]
108	n/a	if not self._signal_handlers:
109	n/a	try:
110	n/a	signal.set_wakeup_fd(-1)
111	n/a	except (ValueError, OSError) as nexc:
112	n/a	logger.info('set_wakeup_fd(-1) failed: %s', nexc)
113	n/a
114	n/a	if exc.errno == errno.EINVAL:
115	n/a	raise RuntimeError('sig {} cannot be caught'.format(sig))
116	n/a	else:
117	n/a	raise
118	n/a
119	n/a	def _handle_signal(self, sig):
120	n/a	"""Internal helper that is the actual signal handler."""
121	n/a	handle = self._signal_handlers.get(sig)
122	n/a	if handle is None:
123	n/a	return # Assume it's some race condition.
124	n/a	if handle._cancelled:
125	n/a	self.remove_signal_handler(sig) # Remove it properly.
126	n/a	else:
127	n/a	self._add_callback_signalsafe(handle)
128	n/a
129	n/a	def remove_signal_handler(self, sig):
130	n/a	"""Remove a handler for a signal. UNIX only.
131	n/a
132	n/a	Return True if a signal handler was removed, False if not.
133	n/a	"""
134	n/a	self._check_signal(sig)
135	n/a	try:
136	n/a	del self._signal_handlers[sig]
137	n/a	except KeyError:
138	n/a	return False
139	n/a
140	n/a	if sig == signal.SIGINT:
141	n/a	handler = signal.default_int_handler
142	n/a	else:
143	n/a	handler = signal.SIG_DFL
144	n/a
145	n/a	try:
146	n/a	signal.signal(sig, handler)
147	n/a	except OSError as exc:
148	n/a	if exc.errno == errno.EINVAL:
149	n/a	raise RuntimeError('sig {} cannot be caught'.format(sig))
150	n/a	else:
151	n/a	raise
152	n/a
153	n/a	if not self._signal_handlers:
154	n/a	try:
155	n/a	signal.set_wakeup_fd(-1)
156	n/a	except (ValueError, OSError) as exc:
157	n/a	logger.info('set_wakeup_fd(-1) failed: %s', exc)
158	n/a
159	n/a	return True
160	n/a
161	n/a	def _check_signal(self, sig):
162	n/a	"""Internal helper to validate a signal.
163	n/a
164	n/a	Raise ValueError if the signal number is invalid or uncatchable.
165	n/a	Raise RuntimeError if there is a problem setting up the handler.
166	n/a	"""
167	n/a	if not isinstance(sig, int):
168	n/a	raise TypeError('sig must be an int, not {!r}'.format(sig))
169	n/a
170	n/a	if not (1 <= sig < signal.NSIG):
171	n/a	raise ValueError(
172	n/a	'sig {} out of range(1, {})'.format(sig, signal.NSIG))
173	n/a
174	n/a	def _make_read_pipe_transport(self, pipe, protocol, waiter=None,
175	n/a	extra=None):
176	n/a	return _UnixReadPipeTransport(self, pipe, protocol, waiter, extra)
177	n/a
178	n/a	def _make_write_pipe_transport(self, pipe, protocol, waiter=None,
179	n/a	extra=None):
180	n/a	return _UnixWritePipeTransport(self, pipe, protocol, waiter, extra)
181	n/a
182	n/a	@coroutine
183	n/a	def _make_subprocess_transport(self, protocol, args, shell,
184	n/a	stdin, stdout, stderr, bufsize,
185	n/a	extra=None, **kwargs):
186	n/a	with events.get_child_watcher() as watcher:
187	n/a	waiter = self.create_future()
188	n/a	transp = _UnixSubprocessTransport(self, protocol, args, shell,
189	n/a	stdin, stdout, stderr, bufsize,
190	n/a	waiter=waiter, extra=extra,
191	n/a	**kwargs)
192	n/a
193	n/a	watcher.add_child_handler(transp.get_pid(),
194	n/a	self._child_watcher_callback, transp)
195	n/a	try:
196	n/a	yield from waiter
197	n/a	except Exception as exc:
198	n/a	# Workaround CPython bug #23353: using yield/yield-from in an
199	n/a	# except block of a generator doesn't clear properly
200	n/a	# sys.exc_info()
201	n/a	err = exc
202	n/a	else:
203	n/a	err = None
204	n/a
205	n/a	if err is not None:
206	n/a	transp.close()
207	n/a	yield from transp._wait()
208	n/a	raise err
209	n/a
210	n/a	return transp
211	n/a
212	n/a	def _child_watcher_callback(self, pid, returncode, transp):
213	n/a	self.call_soon_threadsafe(transp._process_exited, returncode)
214	n/a
215	n/a	@coroutine
216	n/a	def create_unix_connection(self, protocol_factory, path, *,
217	n/a	ssl=None, sock=None,
218	n/a	server_hostname=None):
219	n/a	assert server_hostname is None or isinstance(server_hostname, str)
220	n/a	if ssl:
221	n/a	if server_hostname is None:
222	n/a	raise ValueError(
223	n/a	'you have to pass server_hostname when using ssl')
224	n/a	else:
225	n/a	if server_hostname is not None:
226	n/a	raise ValueError('server_hostname is only meaningful with ssl')
227	n/a
228	n/a	if path is not None:
229	n/a	if sock is not None:
230	n/a	raise ValueError(
231	n/a	'path and sock can not be specified at the same time')
232	n/a
233	n/a	sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM, 0)
234	n/a	try:
235	n/a	sock.setblocking(False)
236	n/a	yield from self.sock_connect(sock, path)
237	n/a	except:
238	n/a	sock.close()
239	n/a	raise
240	n/a
241	n/a	else:
242	n/a	if sock is None:
243	n/a	raise ValueError('no path and sock were specified')
244	n/a	if (sock.family != socket.AF_UNIX or
245	n/a	not base_events._is_stream_socket(sock)):
246	n/a	raise ValueError(
247	n/a	'A UNIX Domain Stream Socket was expected, got {!r}'
248	n/a	.format(sock))
249	n/a	sock.setblocking(False)
250	n/a
251	n/a	transport, protocol = yield from self._create_connection_transport(
252	n/a	sock, protocol_factory, ssl, server_hostname)
253	n/a	return transport, protocol
254	n/a
255	n/a	@coroutine
256	n/a	def create_unix_server(self, protocol_factory, path=None, *,
257	n/a	sock=None, backlog=100, ssl=None):
258	n/a	if isinstance(ssl, bool):
259	n/a	raise TypeError('ssl argument must be an SSLContext or None')
260	n/a
261	n/a	if path is not None:
262	n/a	if sock is not None:
263	n/a	raise ValueError(
264	n/a	'path and sock can not be specified at the same time')
265	n/a
266	n/a	path = _fspath(path)
267	n/a	sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
268	n/a
269	n/a	# Check for abstract socket. `str` and `bytes` paths are supported.
270	n/a	if path[0] not in (0, '\x00'):
271	n/a	try:
272	n/a	if stat.S_ISSOCK(os.stat(path).st_mode):
273	n/a	os.remove(path)
274	n/a	except FileNotFoundError:
275	n/a	pass
276	n/a	except OSError as err:
277	n/a	# Directory may have permissions only to create socket.
278	n/a	logger.error('Unable to check or remove stale UNIX socket %r: %r', path, err)
279	n/a
280	n/a	try:
281	n/a	sock.bind(path)
282	n/a	except OSError as exc:
283	n/a	sock.close()
284	n/a	if exc.errno == errno.EADDRINUSE:
285	n/a	# Let's improve the error message by adding
286	n/a	# with what exact address it occurs.
287	n/a	msg = 'Address {!r} is already in use'.format(path)
288	n/a	raise OSError(errno.EADDRINUSE, msg) from None
289	n/a	else:
290	n/a	raise
291	n/a	except:
292	n/a	sock.close()
293	n/a	raise
294	n/a	else:
295	n/a	if sock is None:
296	n/a	raise ValueError(
297	n/a	'path was not specified, and no sock specified')
298	n/a
299	n/a	if (sock.family != socket.AF_UNIX or
300	n/a	not base_events._is_stream_socket(sock)):
301	n/a	raise ValueError(
302	n/a	'A UNIX Domain Stream Socket was expected, got {!r}'
303	n/a	.format(sock))
304	n/a
305	n/a	server = base_events.Server(self, [sock])
306	n/a	sock.listen(backlog)
307	n/a	sock.setblocking(False)
308	n/a	self._start_serving(protocol_factory, sock, ssl, server)
309	n/a	return server
310	n/a
311	n/a
312	n/a	if hasattr(os, 'set_blocking'):
313	n/a	def _set_nonblocking(fd):
314	n/a	os.set_blocking(fd, False)
315	n/a	else:
316	n/a	import fcntl
317	n/a
318	n/a	def _set_nonblocking(fd):
319	n/a	flags = fcntl.fcntl(fd, fcntl.F_GETFL)
320	n/a	flags = flags \| os.O_NONBLOCK
321	n/a	fcntl.fcntl(fd, fcntl.F_SETFL, flags)
322	n/a
323	n/a
324	n/a	class _UnixReadPipeTransport(transports.ReadTransport):
325	n/a
326	n/a	max_size = 256 * 1024 # max bytes we read in one event loop iteration
327	n/a
328	n/a	def __init__(self, loop, pipe, protocol, waiter=None, extra=None):
329	n/a	super().__init__(extra)
330	n/a	self._extra['pipe'] = pipe
331	n/a	self._loop = loop
332	n/a	self._pipe = pipe
333	n/a	self._fileno = pipe.fileno()
334	n/a	self._protocol = protocol
335	n/a	self._closing = False
336	n/a
337	n/a	mode = os.fstat(self._fileno).st_mode
338	n/a	if not (stat.S_ISFIFO(mode) or
339	n/a	stat.S_ISSOCK(mode) or
340	n/a	stat.S_ISCHR(mode)):
341	n/a	self._pipe = None
342	n/a	self._fileno = None
343	n/a	self._protocol = None
344	n/a	raise ValueError("Pipe transport is for pipes/sockets only.")
345	n/a
346	n/a	_set_nonblocking(self._fileno)
347	n/a
348	n/a	self._loop.call_soon(self._protocol.connection_made, self)
349	n/a	# only start reading when connection_made() has been called
350	n/a	self._loop.call_soon(self._loop._add_reader,
351	n/a	self._fileno, self._read_ready)
352	n/a	if waiter is not None:
353	n/a	# only wake up the waiter when connection_made() has been called
354	n/a	self._loop.call_soon(futures._set_result_unless_cancelled,
355	n/a	waiter, None)
356	n/a
357	n/a	def __repr__(self):
358	n/a	info = [self.__class__.__name__]
359	n/a	if self._pipe is None:
360	n/a	info.append('closed')
361	n/a	elif self._closing:
362	n/a	info.append('closing')
363	n/a	info.append('fd=%s' % self._fileno)
364	n/a	selector = getattr(self._loop, '_selector', None)
365	n/a	if self._pipe is not None and selector is not None:
366	n/a	polling = selector_events._test_selector_event(
367	n/a	selector,
368	n/a	self._fileno, selectors.EVENT_READ)
369	n/a	if polling:
370	n/a	info.append('polling')
371	n/a	else:
372	n/a	info.append('idle')
373	n/a	elif self._pipe is not None:
374	n/a	info.append('open')
375	n/a	else:
376	n/a	info.append('closed')
377	n/a	return '<%s>' % ' '.join(info)
378	n/a
379	n/a	def _read_ready(self):
380	n/a	try:
381	n/a	data = os.read(self._fileno, self.max_size)
382	n/a	except (BlockingIOError, InterruptedError):
383	n/a	pass
384	n/a	except OSError as exc:
385	n/a	self._fatal_error(exc, 'Fatal read error on pipe transport')
386	n/a	else:
387	n/a	if data:
388	n/a	self._protocol.data_received(data)
389	n/a	else:
390	n/a	if self._loop.get_debug():
391	n/a	logger.info("%r was closed by peer", self)
392	n/a	self._closing = True
393	n/a	self._loop._remove_reader(self._fileno)
394	n/a	self._loop.call_soon(self._protocol.eof_received)
395	n/a	self._loop.call_soon(self._call_connection_lost, None)
396	n/a
397	n/a	def pause_reading(self):
398	n/a	self._loop._remove_reader(self._fileno)
399	n/a
400	n/a	def resume_reading(self):
401	n/a	self._loop._add_reader(self._fileno, self._read_ready)
402	n/a
403	n/a	def set_protocol(self, protocol):
404	n/a	self._protocol = protocol
405	n/a
406	n/a	def get_protocol(self):
407	n/a	return self._protocol
408	n/a
409	n/a	def is_closing(self):
410	n/a	return self._closing
411	n/a
412	n/a	def close(self):
413	n/a	if not self._closing:
414	n/a	self._close(None)
415	n/a
416	n/a	# On Python 3.3 and older, objects with a destructor part of a reference
417	n/a	# cycle are never destroyed. It's not more the case on Python 3.4 thanks
418	n/a	# to the PEP 442.
419	n/a	if compat.PY34:
420	n/a	def __del__(self):
421	n/a	if self._pipe is not None:
422	n/a	warnings.warn("unclosed transport %r" % self, ResourceWarning,
423	n/a	source=self)
424	n/a	self._pipe.close()
425	n/a
426	n/a	def _fatal_error(self, exc, message='Fatal error on pipe transport'):
427	n/a	# should be called by exception handler only
428	n/a	if (isinstance(exc, OSError) and exc.errno == errno.EIO):
429	n/a	if self._loop.get_debug():
430	n/a	logger.debug("%r: %s", self, message, exc_info=True)
431	n/a	else:
432	n/a	self._loop.call_exception_handler({
433	n/a	'message': message,
434	n/a	'exception': exc,
435	n/a	'transport': self,
436	n/a	'protocol': self._protocol,
437	n/a	})
438	n/a	self._close(exc)
439	n/a
440	n/a	def _close(self, exc):
441	n/a	self._closing = True
442	n/a	self._loop._remove_reader(self._fileno)
443	n/a	self._loop.call_soon(self._call_connection_lost, exc)
444	n/a
445	n/a	def _call_connection_lost(self, exc):
446	n/a	try:
447	n/a	self._protocol.connection_lost(exc)
448	n/a	finally:
449	n/a	self._pipe.close()
450	n/a	self._pipe = None
451	n/a	self._protocol = None
452	n/a	self._loop = None
453	n/a
454	n/a
455	n/a	class _UnixWritePipeTransport(transports._FlowControlMixin,
456	n/a	transports.WriteTransport):
457	n/a
458	n/a	def __init__(self, loop, pipe, protocol, waiter=None, extra=None):
459	n/a	super().__init__(extra, loop)
460	n/a	self._extra['pipe'] = pipe
461	n/a	self._pipe = pipe
462	n/a	self._fileno = pipe.fileno()
463	n/a	self._protocol = protocol
464	n/a	self._buffer = bytearray()
465	n/a	self._conn_lost = 0
466	n/a	self._closing = False # Set when close() or write_eof() called.
467	n/a
468	n/a	mode = os.fstat(self._fileno).st_mode
469	n/a	is_char = stat.S_ISCHR(mode)
470	n/a	is_fifo = stat.S_ISFIFO(mode)
471	n/a	is_socket = stat.S_ISSOCK(mode)
472	n/a	if not (is_char or is_fifo or is_socket):
473	n/a	self._pipe = None
474	n/a	self._fileno = None
475	n/a	self._protocol = None
476	n/a	raise ValueError("Pipe transport is only for "
477	n/a	"pipes, sockets and character devices")
478	n/a
479	n/a	_set_nonblocking(self._fileno)
480	n/a	self._loop.call_soon(self._protocol.connection_made, self)
481	n/a
482	n/a	# On AIX, the reader trick (to be notified when the read end of the
483	n/a	# socket is closed) only works for sockets. On other platforms it
484	n/a	# works for pipes and sockets. (Exception: OS X 10.4? Issue #19294.)
485	n/a	if is_socket or (is_fifo and not sys.platform.startswith("aix")):
486	n/a	# only start reading when connection_made() has been called
487	n/a	self._loop.call_soon(self._loop._add_reader,
488	n/a	self._fileno, self._read_ready)
489	n/a
490	n/a	if waiter is not None:
491	n/a	# only wake up the waiter when connection_made() has been called
492	n/a	self._loop.call_soon(futures._set_result_unless_cancelled,
493	n/a	waiter, None)
494	n/a
495	n/a	def __repr__(self):
496	n/a	info = [self.__class__.__name__]
497	n/a	if self._pipe is None:
498	n/a	info.append('closed')
499	n/a	elif self._closing:
500	n/a	info.append('closing')
501	n/a	info.append('fd=%s' % self._fileno)
502	n/a	selector = getattr(self._loop, '_selector', None)
503	n/a	if self._pipe is not None and selector is not None:
504	n/a	polling = selector_events._test_selector_event(
505	n/a	selector,
506	n/a	self._fileno, selectors.EVENT_WRITE)
507	n/a	if polling:
508	n/a	info.append('polling')
509	n/a	else:
510	n/a	info.append('idle')
511	n/a
512	n/a	bufsize = self.get_write_buffer_size()
513	n/a	info.append('bufsize=%s' % bufsize)
514	n/a	elif self._pipe is not None:
515	n/a	info.append('open')
516	n/a	else:
517	n/a	info.append('closed')
518	n/a	return '<%s>' % ' '.join(info)
519	n/a
520	n/a	def get_write_buffer_size(self):
521	n/a	return len(self._buffer)
522	n/a
523	n/a	def _read_ready(self):
524	n/a	# Pipe was closed by peer.
525	n/a	if self._loop.get_debug():
526	n/a	logger.info("%r was closed by peer", self)
527	n/a	if self._buffer:
528	n/a	self._close(BrokenPipeError())
529	n/a	else:
530	n/a	self._close()
531	n/a
532	n/a	def write(self, data):
533	n/a	assert isinstance(data, (bytes, bytearray, memoryview)), repr(data)
534	n/a	if isinstance(data, bytearray):
535	n/a	data = memoryview(data)
536	n/a	if not data:
537	n/a	return
538	n/a
539	n/a	if self._conn_lost or self._closing:
540	n/a	if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES:
541	n/a	logger.warning('pipe closed by peer or '
542	n/a	'os.write(pipe, data) raised exception.')
543	n/a	self._conn_lost += 1
544	n/a	return
545	n/a
546	n/a	if not self._buffer:
547	n/a	# Attempt to send it right away first.
548	n/a	try:
549	n/a	n = os.write(self._fileno, data)
550	n/a	except (BlockingIOError, InterruptedError):
551	n/a	n = 0
552	n/a	except Exception as exc:
553	n/a	self._conn_lost += 1
554	n/a	self._fatal_error(exc, 'Fatal write error on pipe transport')
555	n/a	return
556	n/a	if n == len(data):
557	n/a	return
558	n/a	elif n > 0:
559	n/a	data = memoryview(data)[n:]
560	n/a	self._loop._add_writer(self._fileno, self._write_ready)
561	n/a
562	n/a	self._buffer += data
563	n/a	self._maybe_pause_protocol()
564	n/a
565	n/a	def _write_ready(self):
566	n/a	assert self._buffer, 'Data should not be empty'
567	n/a
568	n/a	try:
569	n/a	n = os.write(self._fileno, self._buffer)
570	n/a	except (BlockingIOError, InterruptedError):
571	n/a	pass
572	n/a	except Exception as exc:
573	n/a	self._buffer.clear()
574	n/a	self._conn_lost += 1
575	n/a	# Remove writer here, _fatal_error() doesn't it
576	n/a	# because _buffer is empty.
577	n/a	self._loop._remove_writer(self._fileno)
578	n/a	self._fatal_error(exc, 'Fatal write error on pipe transport')
579	n/a	else:
580	n/a	if n == len(self._buffer):
581	n/a	self._buffer.clear()
582	n/a	self._loop._remove_writer(self._fileno)
583	n/a	self._maybe_resume_protocol() # May append to buffer.
584	n/a	if self._closing:
585	n/a	self._loop._remove_reader(self._fileno)
586	n/a	self._call_connection_lost(None)
587	n/a	return
588	n/a	elif n > 0:
589	n/a	del self._buffer[:n]
590	n/a
591	n/a	def can_write_eof(self):
592	n/a	return True
593	n/a
594	n/a	def write_eof(self):
595	n/a	if self._closing:
596	n/a	return
597	n/a	assert self._pipe
598	n/a	self._closing = True
599	n/a	if not self._buffer:
600	n/a	self._loop._remove_reader(self._fileno)
601	n/a	self._loop.call_soon(self._call_connection_lost, None)
602	n/a
603	n/a	def set_protocol(self, protocol):
604	n/a	self._protocol = protocol
605	n/a
606	n/a	def get_protocol(self):
607	n/a	return self._protocol
608	n/a
609	n/a	def is_closing(self):
610	n/a	return self._closing
611	n/a
612	n/a	def close(self):
613	n/a	if self._pipe is not None and not self._closing:
614	n/a	# write_eof is all what we needed to close the write pipe
615	n/a	self.write_eof()
616	n/a
617	n/a	# On Python 3.3 and older, objects with a destructor part of a reference
618	n/a	# cycle are never destroyed. It's not more the case on Python 3.4 thanks
619	n/a	# to the PEP 442.
620	n/a	if compat.PY34:
621	n/a	def __del__(self):
622	n/a	if self._pipe is not None:
623	n/a	warnings.warn("unclosed transport %r" % self, ResourceWarning,
624	n/a	source=self)
625	n/a	self._pipe.close()
626	n/a
627	n/a	def abort(self):
628	n/a	self._close(None)
629	n/a
630	n/a	def _fatal_error(self, exc, message='Fatal error on pipe transport'):
631	n/a	# should be called by exception handler only
632	n/a	if isinstance(exc, base_events._FATAL_ERROR_IGNORE):
633	n/a	if self._loop.get_debug():
634	n/a	logger.debug("%r: %s", self, message, exc_info=True)
635	n/a	else:
636	n/a	self._loop.call_exception_handler({
637	n/a	'message': message,
638	n/a	'exception': exc,
639	n/a	'transport': self,
640	n/a	'protocol': self._protocol,
641	n/a	})
642	n/a	self._close(exc)
643	n/a
644	n/a	def _close(self, exc=None):
645	n/a	self._closing = True
646	n/a	if self._buffer:
647	n/a	self._loop._remove_writer(self._fileno)
648	n/a	self._buffer.clear()
649	n/a	self._loop._remove_reader(self._fileno)
650	n/a	self._loop.call_soon(self._call_connection_lost, exc)
651	n/a
652	n/a	def _call_connection_lost(self, exc):
653	n/a	try:
654	n/a	self._protocol.connection_lost(exc)
655	n/a	finally:
656	n/a	self._pipe.close()
657	n/a	self._pipe = None
658	n/a	self._protocol = None
659	n/a	self._loop = None
660	n/a
661	n/a
662	n/a	if hasattr(os, 'set_inheritable'):
663	n/a	# Python 3.4 and newer
664	n/a	_set_inheritable = os.set_inheritable
665	n/a	else:
666	n/a	import fcntl
667	n/a
668	n/a	def _set_inheritable(fd, inheritable):
669	n/a	cloexec_flag = getattr(fcntl, 'FD_CLOEXEC', 1)
670	n/a
671	n/a	old = fcntl.fcntl(fd, fcntl.F_GETFD)
672	n/a	if not inheritable:
673	n/a	fcntl.fcntl(fd, fcntl.F_SETFD, old \| cloexec_flag)
674	n/a	else:
675	n/a	fcntl.fcntl(fd, fcntl.F_SETFD, old & ~cloexec_flag)
676	n/a
677	n/a
678	n/a	class _UnixSubprocessTransport(base_subprocess.BaseSubprocessTransport):
679	n/a
680	n/a	def _start(self, args, shell, stdin, stdout, stderr, bufsize, **kwargs):
681	n/a	stdin_w = None
682	n/a	if stdin == subprocess.PIPE:
683	n/a	# Use a socket pair for stdin, since not all platforms
684	n/a	# support selecting read events on the write end of a
685	n/a	# socket (which we use in order to detect closing of the
686	n/a	# other end). Notably this is needed on AIX, and works
687	n/a	# just fine on other platforms.
688	n/a	stdin, stdin_w = self._loop._socketpair()
689	n/a
690	n/a	# Mark the write end of the stdin pipe as non-inheritable,
691	n/a	# needed by close_fds=False on Python 3.3 and older
692	n/a	# (Python 3.4 implements the PEP 446, socketpair returns
693	n/a	# non-inheritable sockets)
694	n/a	_set_inheritable(stdin_w.fileno(), False)
695	n/a	self._proc = subprocess.Popen(
696	n/a	args, shell=shell, stdin=stdin, stdout=stdout, stderr=stderr,
697	n/a	universal_newlines=False, bufsize=bufsize, **kwargs)
698	n/a	if stdin_w is not None:
699	n/a	stdin.close()
700	n/a	self._proc.stdin = open(stdin_w.detach(), 'wb', buffering=bufsize)
701	n/a
702	n/a
703	n/a	class AbstractChildWatcher:
704	n/a	"""Abstract base class for monitoring child processes.
705	n/a
706	n/a	Objects derived from this class monitor a collection of subprocesses and
707	n/a	report their termination or interruption by a signal.
708	n/a
709	n/a	New callbacks are registered with .add_child_handler(). Starting a new
710	n/a	process must be done within a 'with' block to allow the watcher to suspend
711	n/a	its activity until the new process if fully registered (this is needed to
712	n/a	prevent a race condition in some implementations).
713	n/a
714	n/a	Example:
715	n/a	with watcher:
716	n/a	proc = subprocess.Popen("sleep 1")
717	n/a	watcher.add_child_handler(proc.pid, callback)
718	n/a
719	n/a	Notes:
720	n/a	Implementations of this class must be thread-safe.
721	n/a
722	n/a	Since child watcher objects may catch the SIGCHLD signal and call
723	n/a	waitpid(-1), there should be only one active object per process.
724	n/a	"""
725	n/a
726	n/a	def add_child_handler(self, pid, callback, *args):
727	n/a	"""Register a new child handler.
728	n/a
729	n/a	Arrange for callback(pid, returncode, *args) to be called when
730	n/a	process 'pid' terminates. Specifying another callback for the same
731	n/a	process replaces the previous handler.
732	n/a
733	n/a	Note: callback() must be thread-safe.
734	n/a	"""
735	n/a	raise NotImplementedError()
736	n/a
737	n/a	def remove_child_handler(self, pid):
738	n/a	"""Removes the handler for process 'pid'.
739	n/a
740	n/a	The function returns True if the handler was successfully removed,
741	n/a	False if there was nothing to remove."""
742	n/a
743	n/a	raise NotImplementedError()
744	n/a
745	n/a	def attach_loop(self, loop):
746	n/a	"""Attach the watcher to an event loop.
747	n/a
748	n/a	If the watcher was previously attached to an event loop, then it is
749	n/a	first detached before attaching to the new loop.
750	n/a
751	n/a	Note: loop may be None.
752	n/a	"""
753	n/a	raise NotImplementedError()
754	n/a
755	n/a	def close(self):
756	n/a	"""Close the watcher.
757	n/a
758	n/a	This must be called to make sure that any underlying resource is freed.
759	n/a	"""
760	n/a	raise NotImplementedError()
761	n/a
762	n/a	def __enter__(self):
763	n/a	"""Enter the watcher's context and allow starting new processes
764	n/a
765	n/a	This function must return self"""
766	n/a	raise NotImplementedError()
767	n/a
768	n/a	def __exit__(self, a, b, c):
769	n/a	"""Exit the watcher's context"""
770	n/a	raise NotImplementedError()
771	n/a
772	n/a
773	n/a	class BaseChildWatcher(AbstractChildWatcher):
774	n/a
775	n/a	def __init__(self):
776	n/a	self._loop = None
777	n/a	self._callbacks = {}
778	n/a
779	n/a	def close(self):
780	n/a	self.attach_loop(None)
781	n/a
782	n/a	def _do_waitpid(self, expected_pid):
783	n/a	raise NotImplementedError()
784	n/a
785	n/a	def _do_waitpid_all(self):
786	n/a	raise NotImplementedError()
787	n/a
788	n/a	def attach_loop(self, loop):
789	n/a	assert loop is None or isinstance(loop, events.AbstractEventLoop)
790	n/a
791	n/a	if self._loop is not None and loop is None and self._callbacks:
792	n/a	warnings.warn(
793	n/a	'A loop is being detached '
794	n/a	'from a child watcher with pending handlers',
795	n/a	RuntimeWarning)
796	n/a
797	n/a	if self._loop is not None:
798	n/a	self._loop.remove_signal_handler(signal.SIGCHLD)
799	n/a
800	n/a	self._loop = loop
801	n/a	if loop is not None:
802	n/a	loop.add_signal_handler(signal.SIGCHLD, self._sig_chld)
803	n/a
804	n/a	# Prevent a race condition in case a child terminated
805	n/a	# during the switch.
806	n/a	self._do_waitpid_all()
807	n/a
808	n/a	def _sig_chld(self):
809	n/a	try:
810	n/a	self._do_waitpid_all()
811	n/a	except Exception as exc:
812	n/a	# self._loop should always be available here
813	n/a	# as '_sig_chld' is added as a signal handler
814	n/a	# in 'attach_loop'
815	n/a	self._loop.call_exception_handler({
816	n/a	'message': 'Unknown exception in SIGCHLD handler',
817	n/a	'exception': exc,
818	n/a	})
819	n/a
820	n/a	def _compute_returncode(self, status):
821	n/a	if os.WIFSIGNALED(status):
822	n/a	# The child process died because of a signal.
823	n/a	return -os.WTERMSIG(status)
824	n/a	elif os.WIFEXITED(status):
825	n/a	# The child process exited (e.g sys.exit()).
826	n/a	return os.WEXITSTATUS(status)
827	n/a	else:
828	n/a	# The child exited, but we don't understand its status.
829	n/a	# This shouldn't happen, but if it does, let's just
830	n/a	# return that status; perhaps that helps debug it.
831	n/a	return status
832	n/a
833	n/a
834	n/a	class SafeChildWatcher(BaseChildWatcher):
835	n/a	"""'Safe' child watcher implementation.
836	n/a
837	n/a	This implementation avoids disrupting other code spawning processes by
838	n/a	polling explicitly each process in the SIGCHLD handler instead of calling
839	n/a	os.waitpid(-1).
840	n/a
841	n/a	This is a safe solution but it has a significant overhead when handling a
842	n/a	big number of children (O(n) each time SIGCHLD is raised)
843	n/a	"""
844	n/a
845	n/a	def close(self):
846	n/a	self._callbacks.clear()
847	n/a	super().close()
848	n/a
849	n/a	def __enter__(self):
850	n/a	return self
851	n/a
852	n/a	def __exit__(self, a, b, c):
853	n/a	pass
854	n/a
855	n/a	def add_child_handler(self, pid, callback, *args):
856	n/a	if self._loop is None:
857	n/a	raise RuntimeError(
858	n/a	"Cannot add child handler, "
859	n/a	"the child watcher does not have a loop attached")
860	n/a
861	n/a	self._callbacks[pid] = (callback, args)
862	n/a
863	n/a	# Prevent a race condition in case the child is already terminated.
864	n/a	self._do_waitpid(pid)
865	n/a
866	n/a	def remove_child_handler(self, pid):
867	n/a	try:
868	n/a	del self._callbacks[pid]
869	n/a	return True
870	n/a	except KeyError:
871	n/a	return False
872	n/a
873	n/a	def _do_waitpid_all(self):
874	n/a
875	n/a	for pid in list(self._callbacks):
876	n/a	self._do_waitpid(pid)
877	n/a
878	n/a	def _do_waitpid(self, expected_pid):
879	n/a	assert expected_pid > 0
880	n/a
881	n/a	try:
882	n/a	pid, status = os.waitpid(expected_pid, os.WNOHANG)
883	n/a	except ChildProcessError:
884	n/a	# The child process is already reaped
885	n/a	# (may happen if waitpid() is called elsewhere).
886	n/a	pid = expected_pid
887	n/a	returncode = 255
888	n/a	logger.warning(
889	n/a	"Unknown child process pid %d, will report returncode 255",
890	n/a	pid)
891	n/a	else:
892	n/a	if pid == 0:
893	n/a	# The child process is still alive.
894	n/a	return
895	n/a
896	n/a	returncode = self._compute_returncode(status)
897	n/a	if self._loop.get_debug():
898	n/a	logger.debug('process %s exited with returncode %s',
899	n/a	expected_pid, returncode)
900	n/a
901	n/a	try:
902	n/a	callback, args = self._callbacks.pop(pid)
903	n/a	except KeyError: # pragma: no cover
904	n/a	# May happen if .remove_child_handler() is called
905	n/a	# after os.waitpid() returns.
906	n/a	if self._loop.get_debug():
907	n/a	logger.warning("Child watcher got an unexpected pid: %r",
908	n/a	pid, exc_info=True)
909	n/a	else:
910	n/a	callback(pid, returncode, *args)
911	n/a
912	n/a
913	n/a	class FastChildWatcher(BaseChildWatcher):
914	n/a	"""'Fast' child watcher implementation.
915	n/a
916	n/a	This implementation reaps every terminated processes by calling
917	n/a	os.waitpid(-1) directly, possibly breaking other code spawning processes
918	n/a	and waiting for their termination.
919	n/a
920	n/a	There is no noticeable overhead when handling a big number of children
921	n/a	(O(1) each time a child terminates).
922	n/a	"""
923	n/a	def __init__(self):
924	n/a	super().__init__()
925	n/a	self._lock = threading.Lock()
926	n/a	self._zombies = {}
927	n/a	self._forks = 0
928	n/a
929	n/a	def close(self):
930	n/a	self._callbacks.clear()
931	n/a	self._zombies.clear()
932	n/a	super().close()
933	n/a
934	n/a	def __enter__(self):
935	n/a	with self._lock:
936	n/a	self._forks += 1
937	n/a
938	n/a	return self
939	n/a
940	n/a	def __exit__(self, a, b, c):
941	n/a	with self._lock:
942	n/a	self._forks -= 1
943	n/a
944	n/a	if self._forks or not self._zombies:
945	n/a	return
946	n/a
947	n/a	collateral_victims = str(self._zombies)
948	n/a	self._zombies.clear()
949	n/a
950	n/a	logger.warning(
951	n/a	"Caught subprocesses termination from unknown pids: %s",
952	n/a	collateral_victims)
953	n/a
954	n/a	def add_child_handler(self, pid, callback, *args):
955	n/a	assert self._forks, "Must use the context manager"
956	n/a
957	n/a	if self._loop is None:
958	n/a	raise RuntimeError(
959	n/a	"Cannot add child handler, "
960	n/a	"the child watcher does not have a loop attached")
961	n/a
962	n/a	with self._lock:
963	n/a	try:
964	n/a	returncode = self._zombies.pop(pid)
965	n/a	except KeyError:
966	n/a	# The child is running.
967	n/a	self._callbacks[pid] = callback, args
968	n/a	return
969	n/a
970	n/a	# The child is dead already. We can fire the callback.
971	n/a	callback(pid, returncode, *args)
972	n/a
973	n/a	def remove_child_handler(self, pid):
974	n/a	try:
975	n/a	del self._callbacks[pid]
976	n/a	return True
977	n/a	except KeyError:
978	n/a	return False
979	n/a
980	n/a	def _do_waitpid_all(self):
981	n/a	# Because of signal coalescing, we must keep calling waitpid() as
982	n/a	# long as we're able to reap a child.
983	n/a	while True:
984	n/a	try:
985	n/a	pid, status = os.waitpid(-1, os.WNOHANG)
986	n/a	except ChildProcessError:
987	n/a	# No more child processes exist.
988	n/a	return
989	n/a	else:
990	n/a	if pid == 0:
991	n/a	# A child process is still alive.
992	n/a	return
993	n/a
994	n/a	returncode = self._compute_returncode(status)
995	n/a
996	n/a	with self._lock:
997	n/a	try:
998	n/a	callback, args = self._callbacks.pop(pid)
999	n/a	except KeyError:
1000	n/a	# unknown child
1001	n/a	if self._forks:
1002	n/a	# It may not be registered yet.
1003	n/a	self._zombies[pid] = returncode
1004	n/a	if self._loop.get_debug():
1005	n/a	logger.debug('unknown process %s exited '
1006	n/a	'with returncode %s',
1007	n/a	pid, returncode)
1008	n/a	continue
1009	n/a	callback = None
1010	n/a	else:
1011	n/a	if self._loop.get_debug():
1012	n/a	logger.debug('process %s exited with returncode %s',
1013	n/a	pid, returncode)
1014	n/a
1015	n/a	if callback is None:
1016	n/a	logger.warning(
1017	n/a	"Caught subprocess termination from unknown pid: "
1018	n/a	"%d -> %d", pid, returncode)
1019	n/a	else:
1020	n/a	callback(pid, returncode, *args)
1021	n/a
1022	n/a
1023	n/a	class _UnixDefaultEventLoopPolicy(events.BaseDefaultEventLoopPolicy):
1024	n/a	"""UNIX event loop policy with a watcher for child processes."""
1025	n/a	_loop_factory = _UnixSelectorEventLoop
1026	n/a
1027	n/a	def __init__(self):
1028	n/a	super().__init__()
1029	n/a	self._watcher = None
1030	n/a
1031	n/a	def _init_watcher(self):
1032	n/a	with events._lock:
1033	n/a	if self._watcher is None: # pragma: no branch
1034	n/a	self._watcher = SafeChildWatcher()
1035	n/a	if isinstance(threading.current_thread(),
1036	n/a	threading._MainThread):
1037	n/a	self._watcher.attach_loop(self._local._loop)
1038	n/a
1039	n/a	def set_event_loop(self, loop):
1040	n/a	"""Set the event loop.
1041	n/a
1042	n/a	As a side effect, if a child watcher was set before, then calling
1043	n/a	.set_event_loop() from the main thread will call .attach_loop(loop) on
1044	n/a	the child watcher.
1045	n/a	"""
1046	n/a
1047	n/a	super().set_event_loop(loop)
1048	n/a
1049	n/a	if self._watcher is not None and \
1050	n/a	isinstance(threading.current_thread(), threading._MainThread):
1051	n/a	self._watcher.attach_loop(loop)
1052	n/a
1053	n/a	def get_child_watcher(self):
1054	n/a	"""Get the watcher for child processes.
1055	n/a
1056	n/a	If not yet set, a SafeChildWatcher object is automatically created.
1057	n/a	"""
1058	n/a	if self._watcher is None:
1059	n/a	self._init_watcher()
1060	n/a
1061	n/a	return self._watcher
1062	n/a
1063	n/a	def set_child_watcher(self, watcher):
1064	n/a	"""Set the watcher for child processes."""
1065	n/a
1066	n/a	assert watcher is None or isinstance(watcher, AbstractChildWatcher)
1067	n/a
1068	n/a	if self._watcher is not None:
1069	n/a	self._watcher.close()
1070	n/a
1071	n/a	self._watcher = watcher
1072	n/a
1073	n/a	SelectorEventLoop = _UnixSelectorEventLoop
1074	n/a	DefaultEventLoopPolicy = _UnixDefaultEventLoopPolicy