Python code coverage for Lib/asyncio/proactor_events.py

#	count	content
1	n/a	"""Event loop using a proactor and related classes.
2	n/a
3	n/a	A proactor is a "notify-on-completion" multiplexer. Currently a
4	n/a	proactor is only implemented on Windows with IOCP.
5	n/a	"""
6	n/a
7	n/a	__all__ = ['BaseProactorEventLoop']
8	n/a
9	n/a	import socket
10	n/a	import warnings
11	n/a
12	n/a	from . import base_events
13	n/a	from . import compat
14	n/a	from . import constants
15	n/a	from . import futures
16	n/a	from . import sslproto
17	n/a	from . import transports
18	n/a	from .log import logger
19	n/a
20	n/a
21	n/a	class _ProactorBasePipeTransport(transports._FlowControlMixin,
22	n/a	transports.BaseTransport):
23	n/a	"""Base class for pipe and socket transports."""
24	n/a
25	n/a	def __init__(self, loop, sock, protocol, waiter=None,
26	n/a	extra=None, server=None):
27	n/a	super().__init__(extra, loop)
28	n/a	self._set_extra(sock)
29	n/a	self._sock = sock
30	n/a	self._protocol = protocol
31	n/a	self._server = server
32	n/a	self._buffer = None # None or bytearray.
33	n/a	self._read_fut = None
34	n/a	self._write_fut = None
35	n/a	self._pending_write = 0
36	n/a	self._conn_lost = 0
37	n/a	self._closing = False # Set when close() called.
38	n/a	self._eof_written = False
39	n/a	if self._server is not None:
40	n/a	self._server._attach()
41	n/a	self._loop.call_soon(self._protocol.connection_made, self)
42	n/a	if waiter is not None:
43	n/a	# only wake up the waiter when connection_made() has been called
44	n/a	self._loop.call_soon(futures._set_result_unless_cancelled,
45	n/a	waiter, None)
46	n/a
47	n/a	def __repr__(self):
48	n/a	info = [self.__class__.__name__]
49	n/a	if self._sock is None:
50	n/a	info.append('closed')
51	n/a	elif self._closing:
52	n/a	info.append('closing')
53	n/a	if self._sock is not None:
54	n/a	info.append('fd=%s' % self._sock.fileno())
55	n/a	if self._read_fut is not None:
56	n/a	info.append('read=%s' % self._read_fut)
57	n/a	if self._write_fut is not None:
58	n/a	info.append("write=%r" % self._write_fut)
59	n/a	if self._buffer:
60	n/a	bufsize = len(self._buffer)
61	n/a	info.append('write_bufsize=%s' % bufsize)
62	n/a	if self._eof_written:
63	n/a	info.append('EOF written')
64	n/a	return '<%s>' % ' '.join(info)
65	n/a
66	n/a	def _set_extra(self, sock):
67	n/a	self._extra['pipe'] = sock
68	n/a
69	n/a	def set_protocol(self, protocol):
70	n/a	self._protocol = protocol
71	n/a
72	n/a	def get_protocol(self):
73	n/a	return self._protocol
74	n/a
75	n/a	def is_closing(self):
76	n/a	return self._closing
77	n/a
78	n/a	def close(self):
79	n/a	if self._closing:
80	n/a	return
81	n/a	self._closing = True
82	n/a	self._conn_lost += 1
83	n/a	if not self._buffer and self._write_fut is None:
84	n/a	self._loop.call_soon(self._call_connection_lost, None)
85	n/a	if self._read_fut is not None:
86	n/a	self._read_fut.cancel()
87	n/a	self._read_fut = None
88	n/a
89	n/a	# On Python 3.3 and older, objects with a destructor part of a reference
90	n/a	# cycle are never destroyed. It's not more the case on Python 3.4 thanks
91	n/a	# to the PEP 442.
92	n/a	if compat.PY34:
93	n/a	def __del__(self):
94	n/a	if self._sock is not None:
95	n/a	warnings.warn("unclosed transport %r" % self, ResourceWarning,
96	n/a	source=self)
97	n/a	self.close()
98	n/a
99	n/a	def _fatal_error(self, exc, message='Fatal error on pipe transport'):
100	n/a	if isinstance(exc, base_events._FATAL_ERROR_IGNORE):
101	n/a	if self._loop.get_debug():
102	n/a	logger.debug("%r: %s", self, message, exc_info=True)
103	n/a	else:
104	n/a	self._loop.call_exception_handler({
105	n/a	'message': message,
106	n/a	'exception': exc,
107	n/a	'transport': self,
108	n/a	'protocol': self._protocol,
109	n/a	})
110	n/a	self._force_close(exc)
111	n/a
112	n/a	def _force_close(self, exc):
113	n/a	if self._closing:
114	n/a	return
115	n/a	self._closing = True
116	n/a	self._conn_lost += 1
117	n/a	if self._write_fut:
118	n/a	self._write_fut.cancel()
119	n/a	self._write_fut = None
120	n/a	if self._read_fut:
121	n/a	self._read_fut.cancel()
122	n/a	self._read_fut = None
123	n/a	self._pending_write = 0
124	n/a	self._buffer = None
125	n/a	self._loop.call_soon(self._call_connection_lost, exc)
126	n/a
127	n/a	def _call_connection_lost(self, exc):
128	n/a	try:
129	n/a	self._protocol.connection_lost(exc)
130	n/a	finally:
131	n/a	# XXX If there is a pending overlapped read on the other
132	n/a	# end then it may fail with ERROR_NETNAME_DELETED if we
133	n/a	# just close our end. First calling shutdown() seems to
134	n/a	# cure it, but maybe using DisconnectEx() would be better.
135	n/a	if hasattr(self._sock, 'shutdown'):
136	n/a	self._sock.shutdown(socket.SHUT_RDWR)
137	n/a	self._sock.close()
138	n/a	self._sock = None
139	n/a	server = self._server
140	n/a	if server is not None:
141	n/a	server._detach()
142	n/a	self._server = None
143	n/a
144	n/a	def get_write_buffer_size(self):
145	n/a	size = self._pending_write
146	n/a	if self._buffer is not None:
147	n/a	size += len(self._buffer)
148	n/a	return size
149	n/a
150	n/a
151	n/a	class _ProactorReadPipeTransport(_ProactorBasePipeTransport,
152	n/a	transports.ReadTransport):
153	n/a	"""Transport for read pipes."""
154	n/a
155	n/a	def __init__(self, loop, sock, protocol, waiter=None,
156	n/a	extra=None, server=None):
157	n/a	super().__init__(loop, sock, protocol, waiter, extra, server)
158	n/a	self._paused = False
159	n/a	self._loop.call_soon(self._loop_reading)
160	n/a
161	n/a	def pause_reading(self):
162	n/a	if self._closing:
163	n/a	raise RuntimeError('Cannot pause_reading() when closing')
164	n/a	if self._paused:
165	n/a	raise RuntimeError('Already paused')
166	n/a	self._paused = True
167	n/a	if self._loop.get_debug():
168	n/a	logger.debug("%r pauses reading", self)
169	n/a
170	n/a	def resume_reading(self):
171	n/a	if not self._paused:
172	n/a	raise RuntimeError('Not paused')
173	n/a	self._paused = False
174	n/a	if self._closing:
175	n/a	return
176	n/a	self._loop.call_soon(self._loop_reading, self._read_fut)
177	n/a	if self._loop.get_debug():
178	n/a	logger.debug("%r resumes reading", self)
179	n/a
180	n/a	def _loop_reading(self, fut=None):
181	n/a	if self._paused:
182	n/a	return
183	n/a	data = None
184	n/a
185	n/a	try:
186	n/a	if fut is not None:
187	n/a	assert self._read_fut is fut or (self._read_fut is None and
188	n/a	self._closing)
189	n/a	self._read_fut = None
190	n/a	data = fut.result() # deliver data later in "finally" clause
191	n/a
192	n/a	if self._closing:
193	n/a	# since close() has been called we ignore any read data
194	n/a	data = None
195	n/a	return
196	n/a
197	n/a	if data == b'':
198	n/a	# we got end-of-file so no need to reschedule a new read
199	n/a	return
200	n/a
201	n/a	# reschedule a new read
202	n/a	self._read_fut = self._loop._proactor.recv(self._sock, 4096)
203	n/a	except ConnectionAbortedError as exc:
204	n/a	if not self._closing:
205	n/a	self._fatal_error(exc, 'Fatal read error on pipe transport')
206	n/a	elif self._loop.get_debug():
207	n/a	logger.debug("Read error on pipe transport while closing",
208	n/a	exc_info=True)
209	n/a	except ConnectionResetError as exc:
210	n/a	self._force_close(exc)
211	n/a	except OSError as exc:
212	n/a	self._fatal_error(exc, 'Fatal read error on pipe transport')
213	n/a	except futures.CancelledError:
214	n/a	if not self._closing:
215	n/a	raise
216	n/a	else:
217	n/a	self._read_fut.add_done_callback(self._loop_reading)
218	n/a	finally:
219	n/a	if data:
220	n/a	self._protocol.data_received(data)
221	n/a	elif data is not None:
222	n/a	if self._loop.get_debug():
223	n/a	logger.debug("%r received EOF", self)
224	n/a	keep_open = self._protocol.eof_received()
225	n/a	if not keep_open:
226	n/a	self.close()
227	n/a
228	n/a
229	n/a	class _ProactorBaseWritePipeTransport(_ProactorBasePipeTransport,
230	n/a	transports.WriteTransport):
231	n/a	"""Transport for write pipes."""
232	n/a
233	n/a	def write(self, data):
234	n/a	if not isinstance(data, (bytes, bytearray, memoryview)):
235	n/a	raise TypeError('data argument must be byte-ish (%r)',
236	n/a	type(data))
237	n/a	if self._eof_written:
238	n/a	raise RuntimeError('write_eof() already called')
239	n/a
240	n/a	if not data:
241	n/a	return
242	n/a
243	n/a	if self._conn_lost:
244	n/a	if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES:
245	n/a	logger.warning('socket.send() raised exception.')
246	n/a	self._conn_lost += 1
247	n/a	return
248	n/a
249	n/a	# Observable states:
250	n/a	# 1. IDLE: _write_fut and _buffer both None
251	n/a	# 2. WRITING: _write_fut set; _buffer None
252	n/a	# 3. BACKED UP: _write_fut set; _buffer a bytearray
253	n/a	# We always copy the data, so the caller can't modify it
254	n/a	# while we're still waiting for the I/O to happen.
255	n/a	if self._write_fut is None: # IDLE -> WRITING
256	n/a	assert self._buffer is None
257	n/a	# Pass a copy, except if it's already immutable.
258	n/a	self._loop_writing(data=bytes(data))
259	n/a	elif not self._buffer: # WRITING -> BACKED UP
260	n/a	# Make a mutable copy which we can extend.
261	n/a	self._buffer = bytearray(data)
262	n/a	self._maybe_pause_protocol()
263	n/a	else: # BACKED UP
264	n/a	# Append to buffer (also copies).
265	n/a	self._buffer.extend(data)
266	n/a	self._maybe_pause_protocol()
267	n/a
268	n/a	def _loop_writing(self, f=None, data=None):
269	n/a	try:
270	n/a	assert f is self._write_fut
271	n/a	self._write_fut = None
272	n/a	self._pending_write = 0
273	n/a	if f:
274	n/a	f.result()
275	n/a	if data is None:
276	n/a	data = self._buffer
277	n/a	self._buffer = None
278	n/a	if not data:
279	n/a	if self._closing:
280	n/a	self._loop.call_soon(self._call_connection_lost, None)
281	n/a	if self._eof_written:
282	n/a	self._sock.shutdown(socket.SHUT_WR)
283	n/a	# Now that we've reduced the buffer size, tell the
284	n/a	# protocol to resume writing if it was paused. Note that
285	n/a	# we do this last since the callback is called immediately
286	n/a	# and it may add more data to the buffer (even causing the
287	n/a	# protocol to be paused again).
288	n/a	self._maybe_resume_protocol()
289	n/a	else:
290	n/a	self._write_fut = self._loop._proactor.send(self._sock, data)
291	n/a	if not self._write_fut.done():
292	n/a	assert self._pending_write == 0
293	n/a	self._pending_write = len(data)
294	n/a	self._write_fut.add_done_callback(self._loop_writing)
295	n/a	self._maybe_pause_protocol()
296	n/a	else:
297	n/a	self._write_fut.add_done_callback(self._loop_writing)
298	n/a	except ConnectionResetError as exc:
299	n/a	self._force_close(exc)
300	n/a	except OSError as exc:
301	n/a	self._fatal_error(exc, 'Fatal write error on pipe transport')
302	n/a
303	n/a	def can_write_eof(self):
304	n/a	return True
305	n/a
306	n/a	def write_eof(self):
307	n/a	self.close()
308	n/a
309	n/a	def abort(self):
310	n/a	self._force_close(None)
311	n/a
312	n/a
313	n/a	class _ProactorWritePipeTransport(_ProactorBaseWritePipeTransport):
314	n/a	def __init__(self, args, *kw):
315	n/a	super().__init__(args, *kw)
316	n/a	self._read_fut = self._loop._proactor.recv(self._sock, 16)
317	n/a	self._read_fut.add_done_callback(self._pipe_closed)
318	n/a
319	n/a	def _pipe_closed(self, fut):
320	n/a	if fut.cancelled():
321	n/a	# the transport has been closed
322	n/a	return
323	n/a	assert fut.result() == b''
324	n/a	if self._closing:
325	n/a	assert self._read_fut is None
326	n/a	return
327	n/a	assert fut is self._read_fut, (fut, self._read_fut)
328	n/a	self._read_fut = None
329	n/a	if self._write_fut is not None:
330	n/a	self._force_close(BrokenPipeError())
331	n/a	else:
332	n/a	self.close()
333	n/a
334	n/a
335	n/a	class _ProactorDuplexPipeTransport(_ProactorReadPipeTransport,
336	n/a	_ProactorBaseWritePipeTransport,
337	n/a	transports.Transport):
338	n/a	"""Transport for duplex pipes."""
339	n/a
340	n/a	def can_write_eof(self):
341	n/a	return False
342	n/a
343	n/a	def write_eof(self):
344	n/a	raise NotImplementedError
345	n/a
346	n/a
347	n/a	class _ProactorSocketTransport(_ProactorReadPipeTransport,
348	n/a	_ProactorBaseWritePipeTransport,
349	n/a	transports.Transport):
350	n/a	"""Transport for connected sockets."""
351	n/a
352	n/a	def _set_extra(self, sock):
353	n/a	self._extra['socket'] = sock
354	n/a	try:
355	n/a	self._extra['sockname'] = sock.getsockname()
356	n/a	except (socket.error, AttributeError):
357	n/a	if self._loop.get_debug():
358	n/a	logger.warning("getsockname() failed on %r",
359	n/a	sock, exc_info=True)
360	n/a	if 'peername' not in self._extra:
361	n/a	try:
362	n/a	self._extra['peername'] = sock.getpeername()
363	n/a	except (socket.error, AttributeError):
364	n/a	if self._loop.get_debug():
365	n/a	logger.warning("getpeername() failed on %r",
366	n/a	sock, exc_info=True)
367	n/a
368	n/a	def can_write_eof(self):
369	n/a	return True
370	n/a
371	n/a	def write_eof(self):
372	n/a	if self._closing or self._eof_written:
373	n/a	return
374	n/a	self._eof_written = True
375	n/a	if self._write_fut is None:
376	n/a	self._sock.shutdown(socket.SHUT_WR)
377	n/a
378	n/a
379	n/a	class BaseProactorEventLoop(base_events.BaseEventLoop):
380	n/a
381	n/a	def __init__(self, proactor):
382	n/a	super().__init__()
383	n/a	logger.debug('Using proactor: %s', proactor.__class__.__name__)
384	n/a	self._proactor = proactor
385	n/a	self._selector = proactor # convenient alias
386	n/a	self._self_reading_future = None
387	n/a	self._accept_futures = {} # socket file descriptor => Future
388	n/a	proactor.set_loop(self)
389	n/a	self._make_self_pipe()
390	n/a
391	n/a	def _make_socket_transport(self, sock, protocol, waiter=None,
392	n/a	extra=None, server=None):
393	n/a	return _ProactorSocketTransport(self, sock, protocol, waiter,
394	n/a	extra, server)
395	n/a
396	n/a	def _make_ssl_transport(self, rawsock, protocol, sslcontext, waiter=None,
397	n/a	*, server_side=False, server_hostname=None,
398	n/a	extra=None, server=None):
399	n/a	if not sslproto._is_sslproto_available():
400	n/a	raise NotImplementedError("Proactor event loop requires Python 3.5"
401	n/a	" or newer (ssl.MemoryBIO) to support "
402	n/a	"SSL")
403	n/a
404	n/a	ssl_protocol = sslproto.SSLProtocol(self, protocol, sslcontext, waiter,
405	n/a	server_side, server_hostname)
406	n/a	_ProactorSocketTransport(self, rawsock, ssl_protocol,
407	n/a	extra=extra, server=server)
408	n/a	return ssl_protocol._app_transport
409	n/a
410	n/a	def _make_duplex_pipe_transport(self, sock, protocol, waiter=None,
411	n/a	extra=None):
412	n/a	return _ProactorDuplexPipeTransport(self,
413	n/a	sock, protocol, waiter, extra)
414	n/a
415	n/a	def _make_read_pipe_transport(self, sock, protocol, waiter=None,
416	n/a	extra=None):
417	n/a	return _ProactorReadPipeTransport(self, sock, protocol, waiter, extra)
418	n/a
419	n/a	def _make_write_pipe_transport(self, sock, protocol, waiter=None,
420	n/a	extra=None):
421	n/a	# We want connection_lost() to be called when other end closes
422	n/a	return _ProactorWritePipeTransport(self,
423	n/a	sock, protocol, waiter, extra)
424	n/a
425	n/a	def close(self):
426	n/a	if self.is_running():
427	n/a	raise RuntimeError("Cannot close a running event loop")
428	n/a	if self.is_closed():
429	n/a	return
430	n/a
431	n/a	# Call these methods before closing the event loop (before calling
432	n/a	# BaseEventLoop.close), because they can schedule callbacks with
433	n/a	# call_soon(), which is forbidden when the event loop is closed.
434	n/a	self._stop_accept_futures()
435	n/a	self._close_self_pipe()
436	n/a	self._proactor.close()
437	n/a	self._proactor = None
438	n/a	self._selector = None
439	n/a
440	n/a	# Close the event loop
441	n/a	super().close()
442	n/a
443	n/a	def sock_recv(self, sock, n):
444	n/a	return self._proactor.recv(sock, n)
445	n/a
446	n/a	def sock_sendall(self, sock, data):
447	n/a	return self._proactor.send(sock, data)
448	n/a
449	n/a	def sock_connect(self, sock, address):
450	n/a	return self._proactor.connect(sock, address)
451	n/a
452	n/a	def sock_accept(self, sock):
453	n/a	return self._proactor.accept(sock)
454	n/a
455	n/a	def _socketpair(self):
456	n/a	raise NotImplementedError
457	n/a
458	n/a	def _close_self_pipe(self):
459	n/a	if self._self_reading_future is not None:
460	n/a	self._self_reading_future.cancel()
461	n/a	self._self_reading_future = None
462	n/a	self._ssock.close()
463	n/a	self._ssock = None
464	n/a	self._csock.close()
465	n/a	self._csock = None
466	n/a	self._internal_fds -= 1
467	n/a
468	n/a	def _make_self_pipe(self):
469	n/a	# A self-socket, really. :-)
470	n/a	self._ssock, self._csock = self._socketpair()
471	n/a	self._ssock.setblocking(False)
472	n/a	self._csock.setblocking(False)
473	n/a	self._internal_fds += 1
474	n/a	self.call_soon(self._loop_self_reading)
475	n/a
476	n/a	def _loop_self_reading(self, f=None):
477	n/a	try:
478	n/a	if f is not None:
479	n/a	f.result() # may raise
480	n/a	f = self._proactor.recv(self._ssock, 4096)
481	n/a	except futures.CancelledError:
482	n/a	# _close_self_pipe() has been called, stop waiting for data
483	n/a	return
484	n/a	except Exception as exc:
485	n/a	self.call_exception_handler({
486	n/a	'message': 'Error on reading from the event loop self pipe',
487	n/a	'exception': exc,
488	n/a	'loop': self,
489	n/a	})
490	n/a	else:
491	n/a	self._self_reading_future = f
492	n/a	f.add_done_callback(self._loop_self_reading)
493	n/a
494	n/a	def _write_to_self(self):
495	n/a	self._csock.send(b'\0')
496	n/a
497	n/a	def _start_serving(self, protocol_factory, sock,
498	n/a	sslcontext=None, server=None, backlog=100):
499	n/a
500	n/a	def loop(f=None):
501	n/a	try:
502	n/a	if f is not None:
503	n/a	conn, addr = f.result()
504	n/a	if self._debug:
505	n/a	logger.debug("%r got a new connection from %r: %r",
506	n/a	server, addr, conn)
507	n/a	protocol = protocol_factory()
508	n/a	if sslcontext is not None:
509	n/a	self._make_ssl_transport(
510	n/a	conn, protocol, sslcontext, server_side=True,
511	n/a	extra={'peername': addr}, server=server)
512	n/a	else:
513	n/a	self._make_socket_transport(
514	n/a	conn, protocol,
515	n/a	extra={'peername': addr}, server=server)
516	n/a	if self.is_closed():
517	n/a	return
518	n/a	f = self._proactor.accept(sock)
519	n/a	except OSError as exc:
520	n/a	if sock.fileno() != -1:
521	n/a	self.call_exception_handler({
522	n/a	'message': 'Accept failed on a socket',
523	n/a	'exception': exc,
524	n/a	'socket': sock,
525	n/a	})
526	n/a	sock.close()
527	n/a	elif self._debug:
528	n/a	logger.debug("Accept failed on socket %r",
529	n/a	sock, exc_info=True)
530	n/a	except futures.CancelledError:
531	n/a	sock.close()
532	n/a	else:
533	n/a	self._accept_futures[sock.fileno()] = f
534	n/a	f.add_done_callback(loop)
535	n/a
536	n/a	self.call_soon(loop)
537	n/a
538	n/a	def _process_events(self, event_list):
539	n/a	# Events are processed in the IocpProactor._poll() method
540	n/a	pass
541	n/a
542	n/a	def _stop_accept_futures(self):
543	n/a	for future in self._accept_futures.values():
544	n/a	future.cancel()
545	n/a	self._accept_futures.clear()
546	n/a
547	n/a	def _stop_serving(self, sock):
548	n/a	self._stop_accept_futures()
549	n/a	self._proactor._stop_serving(sock)
550	n/a	sock.close()