Python code coverage for Lib/asyncio/selector_events.py

#	count	content
1	n/a	"""Event loop using a selector and related classes.
2	n/a
3	n/a	A selector is a "notify-when-ready" multiplexer. For a subclass which
4	n/a	also includes support for signal handling, see the unix_events sub-module.
5	n/a	"""
6	n/a
7	n/a	__all__ = ['BaseSelectorEventLoop']
8	n/a
9	n/a	import collections
10	n/a	import errno
11	n/a	import functools
12	n/a	import socket
13	n/a	import warnings
14	n/a	import weakref
15	n/a	try:
16	n/a	import ssl
17	n/a	except ImportError: # pragma: no cover
18	n/a	ssl = None
19	n/a
20	n/a	from . import base_events
21	n/a	from . import compat
22	n/a	from . import constants
23	n/a	from . import events
24	n/a	from . import futures
25	n/a	from . import selectors
26	n/a	from . import transports
27	n/a	from . import sslproto
28	n/a	from .coroutines import coroutine
29	n/a	from .log import logger
30	n/a
31	n/a
32	n/a	def _test_selector_event(selector, fd, event):
33	n/a	# Test if the selector is monitoring 'event' events
34	n/a	# for the file descriptor 'fd'.
35	n/a	try:
36	n/a	key = selector.get_key(fd)
37	n/a	except KeyError:
38	n/a	return False
39	n/a	else:
40	n/a	return bool(key.events & event)
41	n/a
42	n/a
43	n/a	if hasattr(socket, 'TCP_NODELAY'):
44	n/a	def _set_nodelay(sock):
45	n/a	if (sock.family in {socket.AF_INET, socket.AF_INET6} and
46	n/a	sock.type == socket.SOCK_STREAM and
47	n/a	sock.proto == socket.IPPROTO_TCP):
48	n/a	sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
49	n/a	else:
50	n/a	def _set_nodelay(sock):
51	n/a	pass
52	n/a
53	n/a
54	n/a	class BaseSelectorEventLoop(base_events.BaseEventLoop):
55	n/a	"""Selector event loop.
56	n/a
57	n/a	See events.EventLoop for API specification.
58	n/a	"""
59	n/a
60	n/a	def __init__(self, selector=None):
61	n/a	super().__init__()
62	n/a
63	n/a	if selector is None:
64	n/a	selector = selectors.DefaultSelector()
65	n/a	logger.debug('Using selector: %s', selector.__class__.__name__)
66	n/a	self._selector = selector
67	n/a	self._make_self_pipe()
68	n/a	self._transports = weakref.WeakValueDictionary()
69	n/a
70	n/a	def _make_socket_transport(self, sock, protocol, waiter=None, *,
71	n/a	extra=None, server=None):
72	n/a	return _SelectorSocketTransport(self, sock, protocol, waiter,
73	n/a	extra, server)
74	n/a
75	n/a	def _make_ssl_transport(self, rawsock, protocol, sslcontext, waiter=None,
76	n/a	*, server_side=False, server_hostname=None,
77	n/a	extra=None, server=None):
78	n/a	if not sslproto._is_sslproto_available():
79	n/a	return self._make_legacy_ssl_transport(
80	n/a	rawsock, protocol, sslcontext, waiter,
81	n/a	server_side=server_side, server_hostname=server_hostname,
82	n/a	extra=extra, server=server)
83	n/a
84	n/a	ssl_protocol = sslproto.SSLProtocol(self, protocol, sslcontext, waiter,
85	n/a	server_side, server_hostname)
86	n/a	_SelectorSocketTransport(self, rawsock, ssl_protocol,
87	n/a	extra=extra, server=server)
88	n/a	return ssl_protocol._app_transport
89	n/a
90	n/a	def _make_legacy_ssl_transport(self, rawsock, protocol, sslcontext,
91	n/a	waiter, *,
92	n/a	server_side=False, server_hostname=None,
93	n/a	extra=None, server=None):
94	n/a	# Use the legacy API: SSL_write, SSL_read, etc. The legacy API is used
95	n/a	# on Python 3.4 and older, when ssl.MemoryBIO is not available.
96	n/a	return _SelectorSslTransport(
97	n/a	self, rawsock, protocol, sslcontext, waiter,
98	n/a	server_side, server_hostname, extra, server)
99	n/a
100	n/a	def _make_datagram_transport(self, sock, protocol,
101	n/a	address=None, waiter=None, extra=None):
102	n/a	return _SelectorDatagramTransport(self, sock, protocol,
103	n/a	address, waiter, extra)
104	n/a
105	n/a	def close(self):
106	n/a	if self.is_running():
107	n/a	raise RuntimeError("Cannot close a running event loop")
108	n/a	if self.is_closed():
109	n/a	return
110	n/a	self._close_self_pipe()
111	n/a	super().close()
112	n/a	if self._selector is not None:
113	n/a	self._selector.close()
114	n/a	self._selector = None
115	n/a
116	n/a	def _socketpair(self):
117	n/a	raise NotImplementedError
118	n/a
119	n/a	def _close_self_pipe(self):
120	n/a	self._remove_reader(self._ssock.fileno())
121	n/a	self._ssock.close()
122	n/a	self._ssock = None
123	n/a	self._csock.close()
124	n/a	self._csock = None
125	n/a	self._internal_fds -= 1
126	n/a
127	n/a	def _make_self_pipe(self):
128	n/a	# A self-socket, really. :-)
129	n/a	self._ssock, self._csock = self._socketpair()
130	n/a	self._ssock.setblocking(False)
131	n/a	self._csock.setblocking(False)
132	n/a	self._internal_fds += 1
133	n/a	self._add_reader(self._ssock.fileno(), self._read_from_self)
134	n/a
135	n/a	def _process_self_data(self, data):
136	n/a	pass
137	n/a
138	n/a	def _read_from_self(self):
139	n/a	while True:
140	n/a	try:
141	n/a	data = self._ssock.recv(4096)
142	n/a	if not data:
143	n/a	break
144	n/a	self._process_self_data(data)
145	n/a	except InterruptedError:
146	n/a	continue
147	n/a	except BlockingIOError:
148	n/a	break
149	n/a
150	n/a	def _write_to_self(self):
151	n/a	# This may be called from a different thread, possibly after
152	n/a	# _close_self_pipe() has been called or even while it is
153	n/a	# running. Guard for self._csock being None or closed. When
154	n/a	# a socket is closed, send() raises OSError (with errno set to
155	n/a	# EBADF, but let's not rely on the exact error code).
156	n/a	csock = self._csock
157	n/a	if csock is not None:
158	n/a	try:
159	n/a	csock.send(b'\0')
160	n/a	except OSError:
161	n/a	if self._debug:
162	n/a	logger.debug("Fail to write a null byte into the "
163	n/a	"self-pipe socket",
164	n/a	exc_info=True)
165	n/a
166	n/a	def _start_serving(self, protocol_factory, sock,
167	n/a	sslcontext=None, server=None, backlog=100):
168	n/a	self._add_reader(sock.fileno(), self._accept_connection,
169	n/a	protocol_factory, sock, sslcontext, server, backlog)
170	n/a
171	n/a	def _accept_connection(self, protocol_factory, sock,
172	n/a	sslcontext=None, server=None, backlog=100):
173	n/a	# This method is only called once for each event loop tick where the
174	n/a	# listening socket has triggered an EVENT_READ. There may be multiple
175	n/a	# connections waiting for an .accept() so it is called in a loop.
176	n/a	# See https://bugs.python.org/issue27906 for more details.
177	n/a	for _ in range(backlog):
178	n/a	try:
179	n/a	conn, addr = sock.accept()
180	n/a	if self._debug:
181	n/a	logger.debug("%r got a new connection from %r: %r",
182	n/a	server, addr, conn)
183	n/a	conn.setblocking(False)
184	n/a	except (BlockingIOError, InterruptedError, ConnectionAbortedError):
185	n/a	# Early exit because the socket accept buffer is empty.
186	n/a	return None
187	n/a	except OSError as exc:
188	n/a	# There's nowhere to send the error, so just log it.
189	n/a	if exc.errno in (errno.EMFILE, errno.ENFILE,
190	n/a	errno.ENOBUFS, errno.ENOMEM):
191	n/a	# Some platforms (e.g. Linux keep reporting the FD as
192	n/a	# ready, so we remove the read handler temporarily.
193	n/a	# We'll try again in a while.
194	n/a	self.call_exception_handler({
195	n/a	'message': 'socket.accept() out of system resource',
196	n/a	'exception': exc,
197	n/a	'socket': sock,
198	n/a	})
199	n/a	self._remove_reader(sock.fileno())
200	n/a	self.call_later(constants.ACCEPT_RETRY_DELAY,
201	n/a	self._start_serving,
202	n/a	protocol_factory, sock, sslcontext, server,
203	n/a	backlog)
204	n/a	else:
205	n/a	raise # The event loop will catch, log and ignore it.
206	n/a	else:
207	n/a	extra = {'peername': addr}
208	n/a	accept = self._accept_connection2(protocol_factory, conn, extra,
209	n/a	sslcontext, server)
210	n/a	self.create_task(accept)
211	n/a
212	n/a	@coroutine
213	n/a	def _accept_connection2(self, protocol_factory, conn, extra,
214	n/a	sslcontext=None, server=None):
215	n/a	protocol = None
216	n/a	transport = None
217	n/a	try:
218	n/a	protocol = protocol_factory()
219	n/a	waiter = self.create_future()
220	n/a	if sslcontext:
221	n/a	transport = self._make_ssl_transport(
222	n/a	conn, protocol, sslcontext, waiter=waiter,
223	n/a	server_side=True, extra=extra, server=server)
224	n/a	else:
225	n/a	transport = self._make_socket_transport(
226	n/a	conn, protocol, waiter=waiter, extra=extra,
227	n/a	server=server)
228	n/a
229	n/a	try:
230	n/a	yield from waiter
231	n/a	except:
232	n/a	transport.close()
233	n/a	raise
234	n/a
235	n/a	# It's now up to the protocol to handle the connection.
236	n/a	except Exception as exc:
237	n/a	if self._debug:
238	n/a	context = {
239	n/a	'message': ('Error on transport creation '
240	n/a	'for incoming connection'),
241	n/a	'exception': exc,
242	n/a	}
243	n/a	if protocol is not None:
244	n/a	context['protocol'] = protocol
245	n/a	if transport is not None:
246	n/a	context['transport'] = transport
247	n/a	self.call_exception_handler(context)
248	n/a
249	n/a	def _ensure_fd_no_transport(self, fd):
250	n/a	try:
251	n/a	transport = self._transports[fd]
252	n/a	except KeyError:
253	n/a	pass
254	n/a	else:
255	n/a	if not transport.is_closing():
256	n/a	raise RuntimeError(
257	n/a	'File descriptor {!r} is used by transport {!r}'.format(
258	n/a	fd, transport))
259	n/a
260	n/a	def _add_reader(self, fd, callback, *args):
261	n/a	self._check_closed()
262	n/a	handle = events.Handle(callback, args, self)
263	n/a	try:
264	n/a	key = self._selector.get_key(fd)
265	n/a	except KeyError:
266	n/a	self._selector.register(fd, selectors.EVENT_READ,
267	n/a	(handle, None))
268	n/a	else:
269	n/a	mask, (reader, writer) = key.events, key.data
270	n/a	self._selector.modify(fd, mask \| selectors.EVENT_READ,
271	n/a	(handle, writer))
272	n/a	if reader is not None:
273	n/a	reader.cancel()
274	n/a
275	n/a	def _remove_reader(self, fd):
276	n/a	if self.is_closed():
277	n/a	return False
278	n/a	try:
279	n/a	key = self._selector.get_key(fd)
280	n/a	except KeyError:
281	n/a	return False
282	n/a	else:
283	n/a	mask, (reader, writer) = key.events, key.data
284	n/a	mask &= ~selectors.EVENT_READ
285	n/a	if not mask:
286	n/a	self._selector.unregister(fd)
287	n/a	else:
288	n/a	self._selector.modify(fd, mask, (None, writer))
289	n/a
290	n/a	if reader is not None:
291	n/a	reader.cancel()
292	n/a	return True
293	n/a	else:
294	n/a	return False
295	n/a
296	n/a	def _add_writer(self, fd, callback, *args):
297	n/a	self._check_closed()
298	n/a	handle = events.Handle(callback, args, self)
299	n/a	try:
300	n/a	key = self._selector.get_key(fd)
301	n/a	except KeyError:
302	n/a	self._selector.register(fd, selectors.EVENT_WRITE,
303	n/a	(None, handle))
304	n/a	else:
305	n/a	mask, (reader, writer) = key.events, key.data
306	n/a	self._selector.modify(fd, mask \| selectors.EVENT_WRITE,
307	n/a	(reader, handle))
308	n/a	if writer is not None:
309	n/a	writer.cancel()
310	n/a
311	n/a	def _remove_writer(self, fd):
312	n/a	"""Remove a writer callback."""
313	n/a	if self.is_closed():
314	n/a	return False
315	n/a	try:
316	n/a	key = self._selector.get_key(fd)
317	n/a	except KeyError:
318	n/a	return False
319	n/a	else:
320	n/a	mask, (reader, writer) = key.events, key.data
321	n/a	# Remove both writer and connector.
322	n/a	mask &= ~selectors.EVENT_WRITE
323	n/a	if not mask:
324	n/a	self._selector.unregister(fd)
325	n/a	else:
326	n/a	self._selector.modify(fd, mask, (reader, None))
327	n/a
328	n/a	if writer is not None:
329	n/a	writer.cancel()
330	n/a	return True
331	n/a	else:
332	n/a	return False
333	n/a
334	n/a	def add_reader(self, fd, callback, *args):
335	n/a	"""Add a reader callback."""
336	n/a	self._ensure_fd_no_transport(fd)
337	n/a	return self._add_reader(fd, callback, *args)
338	n/a
339	n/a	def remove_reader(self, fd):
340	n/a	"""Remove a reader callback."""
341	n/a	self._ensure_fd_no_transport(fd)
342	n/a	return self._remove_reader(fd)
343	n/a
344	n/a	def add_writer(self, fd, callback, *args):
345	n/a	"""Add a writer callback.."""
346	n/a	self._ensure_fd_no_transport(fd)
347	n/a	return self._add_writer(fd, callback, *args)
348	n/a
349	n/a	def remove_writer(self, fd):
350	n/a	"""Remove a writer callback."""
351	n/a	self._ensure_fd_no_transport(fd)
352	n/a	return self._remove_writer(fd)
353	n/a
354	n/a	def sock_recv(self, sock, n):
355	n/a	"""Receive data from the socket.
356	n/a
357	n/a	The return value is a bytes object representing the data received.
358	n/a	The maximum amount of data to be received at once is specified by
359	n/a	nbytes.
360	n/a
361	n/a	This method is a coroutine.
362	n/a	"""
363	n/a	if self._debug and sock.gettimeout() != 0:
364	n/a	raise ValueError("the socket must be non-blocking")
365	n/a	fut = self.create_future()
366	n/a	self._sock_recv(fut, False, sock, n)
367	n/a	return fut
368	n/a
369	n/a	def _sock_recv(self, fut, registered, sock, n):
370	n/a	# _sock_recv() can add itself as an I/O callback if the operation can't
371	n/a	# be done immediately. Don't use it directly, call sock_recv().
372	n/a	fd = sock.fileno()
373	n/a	if registered:
374	n/a	# Remove the callback early. It should be rare that the
375	n/a	# selector says the fd is ready but the call still returns
376	n/a	# EAGAIN, and I am willing to take a hit in that case in
377	n/a	# order to simplify the common case.
378	n/a	self.remove_reader(fd)
379	n/a	if fut.cancelled():
380	n/a	return
381	n/a	try:
382	n/a	data = sock.recv(n)
383	n/a	except (BlockingIOError, InterruptedError):
384	n/a	self.add_reader(fd, self._sock_recv, fut, True, sock, n)
385	n/a	except Exception as exc:
386	n/a	fut.set_exception(exc)
387	n/a	else:
388	n/a	fut.set_result(data)
389	n/a
390	n/a	def sock_sendall(self, sock, data):
391	n/a	"""Send data to the socket.
392	n/a
393	n/a	The socket must be connected to a remote socket. This method continues
394	n/a	to send data from data until either all data has been sent or an
395	n/a	error occurs. None is returned on success. On error, an exception is
396	n/a	raised, and there is no way to determine how much data, if any, was
397	n/a	successfully processed by the receiving end of the connection.
398	n/a
399	n/a	This method is a coroutine.
400	n/a	"""
401	n/a	if self._debug and sock.gettimeout() != 0:
402	n/a	raise ValueError("the socket must be non-blocking")
403	n/a	fut = self.create_future()
404	n/a	if data:
405	n/a	self._sock_sendall(fut, False, sock, data)
406	n/a	else:
407	n/a	fut.set_result(None)
408	n/a	return fut
409	n/a
410	n/a	def _sock_sendall(self, fut, registered, sock, data):
411	n/a	fd = sock.fileno()
412	n/a
413	n/a	if registered:
414	n/a	self.remove_writer(fd)
415	n/a	if fut.cancelled():
416	n/a	return
417	n/a
418	n/a	try:
419	n/a	n = sock.send(data)
420	n/a	except (BlockingIOError, InterruptedError):
421	n/a	n = 0
422	n/a	except Exception as exc:
423	n/a	fut.set_exception(exc)
424	n/a	return
425	n/a
426	n/a	if n == len(data):
427	n/a	fut.set_result(None)
428	n/a	else:
429	n/a	if n:
430	n/a	data = data[n:]
431	n/a	self.add_writer(fd, self._sock_sendall, fut, True, sock, data)
432	n/a
433	n/a	@coroutine
434	n/a	def sock_connect(self, sock, address):
435	n/a	"""Connect to a remote socket at address.
436	n/a
437	n/a	This method is a coroutine.
438	n/a	"""
439	n/a	if self._debug and sock.gettimeout() != 0:
440	n/a	raise ValueError("the socket must be non-blocking")
441	n/a
442	n/a	if not hasattr(socket, 'AF_UNIX') or sock.family != socket.AF_UNIX:
443	n/a	resolved = base_events._ensure_resolved(
444	n/a	address, family=sock.family, proto=sock.proto, loop=self)
445	n/a	if not resolved.done():
446	n/a	yield from resolved
447	n/a	_, _, _, _, address = resolved.result()[0]
448	n/a
449	n/a	fut = self.create_future()
450	n/a	self._sock_connect(fut, sock, address)
451	n/a	return (yield from fut)
452	n/a
453	n/a	def _sock_connect(self, fut, sock, address):
454	n/a	fd = sock.fileno()
455	n/a	try:
456	n/a	sock.connect(address)
457	n/a	except (BlockingIOError, InterruptedError):
458	n/a	# Issue #23618: When the C function connect() fails with EINTR, the
459	n/a	# connection runs in background. We have to wait until the socket
460	n/a	# becomes writable to be notified when the connection succeed or
461	n/a	# fails.
462	n/a	fut.add_done_callback(
463	n/a	functools.partial(self._sock_connect_done, fd))
464	n/a	self.add_writer(fd, self._sock_connect_cb, fut, sock, address)
465	n/a	except Exception as exc:
466	n/a	fut.set_exception(exc)
467	n/a	else:
468	n/a	fut.set_result(None)
469	n/a
470	n/a	def _sock_connect_done(self, fd, fut):
471	n/a	self.remove_writer(fd)
472	n/a
473	n/a	def _sock_connect_cb(self, fut, sock, address):
474	n/a	if fut.cancelled():
475	n/a	return
476	n/a
477	n/a	try:
478	n/a	err = sock.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
479	n/a	if err != 0:
480	n/a	# Jump to any except clause below.
481	n/a	raise OSError(err, 'Connect call failed %s' % (address,))
482	n/a	except (BlockingIOError, InterruptedError):
483	n/a	# socket is still registered, the callback will be retried later
484	n/a	pass
485	n/a	except Exception as exc:
486	n/a	fut.set_exception(exc)
487	n/a	else:
488	n/a	fut.set_result(None)
489	n/a
490	n/a	def sock_accept(self, sock):
491	n/a	"""Accept a connection.
492	n/a
493	n/a	The socket must be bound to an address and listening for connections.
494	n/a	The return value is a pair (conn, address) where conn is a new socket
495	n/a	object usable to send and receive data on the connection, and address
496	n/a	is the address bound to the socket on the other end of the connection.
497	n/a
498	n/a	This method is a coroutine.
499	n/a	"""
500	n/a	if self._debug and sock.gettimeout() != 0:
501	n/a	raise ValueError("the socket must be non-blocking")
502	n/a	fut = self.create_future()
503	n/a	self._sock_accept(fut, False, sock)
504	n/a	return fut
505	n/a
506	n/a	def _sock_accept(self, fut, registered, sock):
507	n/a	fd = sock.fileno()
508	n/a	if registered:
509	n/a	self.remove_reader(fd)
510	n/a	if fut.cancelled():
511	n/a	return
512	n/a	try:
513	n/a	conn, address = sock.accept()
514	n/a	conn.setblocking(False)
515	n/a	except (BlockingIOError, InterruptedError):
516	n/a	self.add_reader(fd, self._sock_accept, fut, True, sock)
517	n/a	except Exception as exc:
518	n/a	fut.set_exception(exc)
519	n/a	else:
520	n/a	fut.set_result((conn, address))
521	n/a
522	n/a	def _process_events(self, event_list):
523	n/a	for key, mask in event_list:
524	n/a	fileobj, (reader, writer) = key.fileobj, key.data
525	n/a	if mask & selectors.EVENT_READ and reader is not None:
526	n/a	if reader._cancelled:
527	n/a	self._remove_reader(fileobj)
528	n/a	else:
529	n/a	self._add_callback(reader)
530	n/a	if mask & selectors.EVENT_WRITE and writer is not None:
531	n/a	if writer._cancelled:
532	n/a	self._remove_writer(fileobj)
533	n/a	else:
534	n/a	self._add_callback(writer)
535	n/a
536	n/a	def _stop_serving(self, sock):
537	n/a	self._remove_reader(sock.fileno())
538	n/a	sock.close()
539	n/a
540	n/a
541	n/a	class _SelectorTransport(transports._FlowControlMixin,
542	n/a	transports.Transport):
543	n/a
544	n/a	max_size = 256 * 1024 # Buffer size passed to recv().
545	n/a
546	n/a	_buffer_factory = bytearray # Constructs initial value for self._buffer.
547	n/a
548	n/a	# Attribute used in the destructor: it must be set even if the constructor
549	n/a	# is not called (see _SelectorSslTransport which may start by raising an
550	n/a	# exception)
551	n/a	_sock = None
552	n/a
553	n/a	def __init__(self, loop, sock, protocol, extra=None, server=None):
554	n/a	super().__init__(extra, loop)
555	n/a	self._extra['socket'] = sock
556	n/a	self._extra['sockname'] = sock.getsockname()
557	n/a	if 'peername' not in self._extra:
558	n/a	try:
559	n/a	self._extra['peername'] = sock.getpeername()
560	n/a	except socket.error:
561	n/a	self._extra['peername'] = None
562	n/a	self._sock = sock
563	n/a	self._sock_fd = sock.fileno()
564	n/a	self._protocol = protocol
565	n/a	self._protocol_connected = True
566	n/a	self._server = server
567	n/a	self._buffer = self._buffer_factory()
568	n/a	self._conn_lost = 0 # Set when call to connection_lost scheduled.
569	n/a	self._closing = False # Set when close() called.
570	n/a	if self._server is not None:
571	n/a	self._server._attach()
572	n/a	loop._transports[self._sock_fd] = self
573	n/a
574	n/a	def __repr__(self):
575	n/a	info = [self.__class__.__name__]
576	n/a	if self._sock is None:
577	n/a	info.append('closed')
578	n/a	elif self._closing:
579	n/a	info.append('closing')
580	n/a	info.append('fd=%s' % self._sock_fd)
581	n/a	# test if the transport was closed
582	n/a	if self._loop is not None and not self._loop.is_closed():
583	n/a	polling = _test_selector_event(self._loop._selector,
584	n/a	self._sock_fd, selectors.EVENT_READ)
585	n/a	if polling:
586	n/a	info.append('read=polling')
587	n/a	else:
588	n/a	info.append('read=idle')
589	n/a
590	n/a	polling = _test_selector_event(self._loop._selector,
591	n/a	self._sock_fd,
592	n/a	selectors.EVENT_WRITE)
593	n/a	if polling:
594	n/a	state = 'polling'
595	n/a	else:
596	n/a	state = 'idle'
597	n/a
598	n/a	bufsize = self.get_write_buffer_size()
599	n/a	info.append('write=<%s, bufsize=%s>' % (state, bufsize))
600	n/a	return '<%s>' % ' '.join(info)
601	n/a
602	n/a	def abort(self):
603	n/a	self._force_close(None)
604	n/a
605	n/a	def set_protocol(self, protocol):
606	n/a	self._protocol = protocol
607	n/a
608	n/a	def get_protocol(self):
609	n/a	return self._protocol
610	n/a
611	n/a	def is_closing(self):
612	n/a	return self._closing
613	n/a
614	n/a	def close(self):
615	n/a	if self._closing:
616	n/a	return
617	n/a	self._closing = True
618	n/a	self._loop._remove_reader(self._sock_fd)
619	n/a	if not self._buffer:
620	n/a	self._conn_lost += 1
621	n/a	self._loop._remove_writer(self._sock_fd)
622	n/a	self._loop.call_soon(self._call_connection_lost, None)
623	n/a
624	n/a	# On Python 3.3 and older, objects with a destructor part of a reference
625	n/a	# cycle are never destroyed. It's not more the case on Python 3.4 thanks
626	n/a	# to the PEP 442.
627	n/a	if compat.PY34:
628	n/a	def __del__(self):
629	n/a	if self._sock is not None:
630	n/a	warnings.warn("unclosed transport %r" % self, ResourceWarning,
631	n/a	source=self)
632	n/a	self._sock.close()
633	n/a
634	n/a	def _fatal_error(self, exc, message='Fatal error on transport'):
635	n/a	# Should be called from exception handler only.
636	n/a	if isinstance(exc, base_events._FATAL_ERROR_IGNORE):
637	n/a	if self._loop.get_debug():
638	n/a	logger.debug("%r: %s", self, message, exc_info=True)
639	n/a	else:
640	n/a	self._loop.call_exception_handler({
641	n/a	'message': message,
642	n/a	'exception': exc,
643	n/a	'transport': self,
644	n/a	'protocol': self._protocol,
645	n/a	})
646	n/a	self._force_close(exc)
647	n/a
648	n/a	def _force_close(self, exc):
649	n/a	if self._conn_lost:
650	n/a	return
651	n/a	if self._buffer:
652	n/a	self._buffer.clear()
653	n/a	self._loop._remove_writer(self._sock_fd)
654	n/a	if not self._closing:
655	n/a	self._closing = True
656	n/a	self._loop._remove_reader(self._sock_fd)
657	n/a	self._conn_lost += 1
658	n/a	self._loop.call_soon(self._call_connection_lost, exc)
659	n/a
660	n/a	def _call_connection_lost(self, exc):
661	n/a	try:
662	n/a	if self._protocol_connected:
663	n/a	self._protocol.connection_lost(exc)
664	n/a	finally:
665	n/a	self._sock.close()
666	n/a	self._sock = None
667	n/a	self._protocol = None
668	n/a	self._loop = None
669	n/a	server = self._server
670	n/a	if server is not None:
671	n/a	server._detach()
672	n/a	self._server = None
673	n/a
674	n/a	def get_write_buffer_size(self):
675	n/a	return len(self._buffer)
676	n/a
677	n/a
678	n/a	class _SelectorSocketTransport(_SelectorTransport):
679	n/a
680	n/a	def __init__(self, loop, sock, protocol, waiter=None,
681	n/a	extra=None, server=None):
682	n/a	super().__init__(loop, sock, protocol, extra, server)
683	n/a	self._eof = False
684	n/a	self._paused = False
685	n/a
686	n/a	# Disable the Nagle algorithm -- small writes will be
687	n/a	# sent without waiting for the TCP ACK. This generally
688	n/a	# decreases the latency (in some cases significantly.)
689	n/a	_set_nodelay(self._sock)
690	n/a
691	n/a	self._loop.call_soon(self._protocol.connection_made, self)
692	n/a	# only start reading when connection_made() has been called
693	n/a	self._loop.call_soon(self._loop._add_reader,
694	n/a	self._sock_fd, self._read_ready)
695	n/a	if waiter is not None:
696	n/a	# only wake up the waiter when connection_made() has been called
697	n/a	self._loop.call_soon(futures._set_result_unless_cancelled,
698	n/a	waiter, None)
699	n/a
700	n/a	def pause_reading(self):
701	n/a	if self._closing:
702	n/a	raise RuntimeError('Cannot pause_reading() when closing')
703	n/a	if self._paused:
704	n/a	raise RuntimeError('Already paused')
705	n/a	self._paused = True
706	n/a	self._loop._remove_reader(self._sock_fd)
707	n/a	if self._loop.get_debug():
708	n/a	logger.debug("%r pauses reading", self)
709	n/a
710	n/a	def resume_reading(self):
711	n/a	if not self._paused:
712	n/a	raise RuntimeError('Not paused')
713	n/a	self._paused = False
714	n/a	if self._closing:
715	n/a	return
716	n/a	self._loop._add_reader(self._sock_fd, self._read_ready)
717	n/a	if self._loop.get_debug():
718	n/a	logger.debug("%r resumes reading", self)
719	n/a
720	n/a	def _read_ready(self):
721	n/a	if self._conn_lost:
722	n/a	return
723	n/a	try:
724	n/a	data = self._sock.recv(self.max_size)
725	n/a	except (BlockingIOError, InterruptedError):
726	n/a	pass
727	n/a	except Exception as exc:
728	n/a	self._fatal_error(exc, 'Fatal read error on socket transport')
729	n/a	else:
730	n/a	if data:
731	n/a	self._protocol.data_received(data)
732	n/a	else:
733	n/a	if self._loop.get_debug():
734	n/a	logger.debug("%r received EOF", self)
735	n/a	keep_open = self._protocol.eof_received()
736	n/a	if keep_open:
737	n/a	# We're keeping the connection open so the
738	n/a	# protocol can write more, but we still can't
739	n/a	# receive more, so remove the reader callback.
740	n/a	self._loop._remove_reader(self._sock_fd)
741	n/a	else:
742	n/a	self.close()
743	n/a
744	n/a	def write(self, data):
745	n/a	if not isinstance(data, (bytes, bytearray, memoryview)):
746	n/a	raise TypeError('data argument must be a bytes-like object, '
747	n/a	'not %r' % type(data).__name__)
748	n/a	if self._eof:
749	n/a	raise RuntimeError('Cannot call write() after write_eof()')
750	n/a	if not data:
751	n/a	return
752	n/a
753	n/a	if self._conn_lost:
754	n/a	if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES:
755	n/a	logger.warning('socket.send() raised exception.')
756	n/a	self._conn_lost += 1
757	n/a	return
758	n/a
759	n/a	if not self._buffer:
760	n/a	# Optimization: try to send now.
761	n/a	try:
762	n/a	n = self._sock.send(data)
763	n/a	except (BlockingIOError, InterruptedError):
764	n/a	pass
765	n/a	except Exception as exc:
766	n/a	self._fatal_error(exc, 'Fatal write error on socket transport')
767	n/a	return
768	n/a	else:
769	n/a	data = data[n:]
770	n/a	if not data:
771	n/a	return
772	n/a	# Not all was written; register write handler.
773	n/a	self._loop._add_writer(self._sock_fd, self._write_ready)
774	n/a
775	n/a	# Add it to the buffer.
776	n/a	self._buffer.extend(data)
777	n/a	self._maybe_pause_protocol()
778	n/a
779	n/a	def _write_ready(self):
780	n/a	assert self._buffer, 'Data should not be empty'
781	n/a
782	n/a	if self._conn_lost:
783	n/a	return
784	n/a	try:
785	n/a	n = self._sock.send(self._buffer)
786	n/a	except (BlockingIOError, InterruptedError):
787	n/a	pass
788	n/a	except Exception as exc:
789	n/a	self._loop._remove_writer(self._sock_fd)
790	n/a	self._buffer.clear()
791	n/a	self._fatal_error(exc, 'Fatal write error on socket transport')
792	n/a	else:
793	n/a	if n:
794	n/a	del self._buffer[:n]
795	n/a	self._maybe_resume_protocol() # May append to buffer.
796	n/a	if not self._buffer:
797	n/a	self._loop._remove_writer(self._sock_fd)
798	n/a	if self._closing:
799	n/a	self._call_connection_lost(None)
800	n/a	elif self._eof:
801	n/a	self._sock.shutdown(socket.SHUT_WR)
802	n/a
803	n/a	def write_eof(self):
804	n/a	if self._eof:
805	n/a	return
806	n/a	self._eof = True
807	n/a	if not self._buffer:
808	n/a	self._sock.shutdown(socket.SHUT_WR)
809	n/a
810	n/a	def can_write_eof(self):
811	n/a	return True
812	n/a
813	n/a
814	n/a	class _SelectorSslTransport(_SelectorTransport):
815	n/a
816	n/a	_buffer_factory = bytearray
817	n/a
818	n/a	def __init__(self, loop, rawsock, protocol, sslcontext, waiter=None,
819	n/a	server_side=False, server_hostname=None,
820	n/a	extra=None, server=None):
821	n/a	if ssl is None:
822	n/a	raise RuntimeError('stdlib ssl module not available')
823	n/a
824	n/a	if not sslcontext:
825	n/a	sslcontext = sslproto._create_transport_context(server_side, server_hostname)
826	n/a
827	n/a	wrap_kwargs = {
828	n/a	'server_side': server_side,
829	n/a	'do_handshake_on_connect': False,
830	n/a	}
831	n/a	if server_hostname and not server_side:
832	n/a	wrap_kwargs['server_hostname'] = server_hostname
833	n/a	sslsock = sslcontext.wrap_socket(rawsock, **wrap_kwargs)
834	n/a
835	n/a	super().__init__(loop, sslsock, protocol, extra, server)
836	n/a	# the protocol connection is only made after the SSL handshake
837	n/a	self._protocol_connected = False
838	n/a
839	n/a	self._server_hostname = server_hostname
840	n/a	self._waiter = waiter
841	n/a	self._sslcontext = sslcontext
842	n/a	self._paused = False
843	n/a
844	n/a	# SSL-specific extra info. (peercert is set later)
845	n/a	self._extra.update(sslcontext=sslcontext)
846	n/a
847	n/a	if self._loop.get_debug():
848	n/a	logger.debug("%r starts SSL handshake", self)
849	n/a	start_time = self._loop.time()
850	n/a	else:
851	n/a	start_time = None
852	n/a	self._on_handshake(start_time)
853	n/a
854	n/a	def _wakeup_waiter(self, exc=None):
855	n/a	if self._waiter is None:
856	n/a	return
857	n/a	if not self._waiter.cancelled():
858	n/a	if exc is not None:
859	n/a	self._waiter.set_exception(exc)
860	n/a	else:
861	n/a	self._waiter.set_result(None)
862	n/a	self._waiter = None
863	n/a
864	n/a	def _on_handshake(self, start_time):
865	n/a	try:
866	n/a	self._sock.do_handshake()
867	n/a	except ssl.SSLWantReadError:
868	n/a	self._loop._add_reader(self._sock_fd,
869	n/a	self._on_handshake, start_time)
870	n/a	return
871	n/a	except ssl.SSLWantWriteError:
872	n/a	self._loop._add_writer(self._sock_fd,
873	n/a	self._on_handshake, start_time)
874	n/a	return
875	n/a	except BaseException as exc:
876	n/a	if self._loop.get_debug():
877	n/a	logger.warning("%r: SSL handshake failed",
878	n/a	self, exc_info=True)
879	n/a	self._loop._remove_reader(self._sock_fd)
880	n/a	self._loop._remove_writer(self._sock_fd)
881	n/a	self._sock.close()
882	n/a	self._wakeup_waiter(exc)
883	n/a	if isinstance(exc, Exception):
884	n/a	return
885	n/a	else:
886	n/a	raise
887	n/a
888	n/a	self._loop._remove_reader(self._sock_fd)
889	n/a	self._loop._remove_writer(self._sock_fd)
890	n/a
891	n/a	peercert = self._sock.getpeercert()
892	n/a	if not hasattr(self._sslcontext, 'check_hostname'):
893	n/a	# Verify hostname if requested, Python 3.4+ uses check_hostname
894	n/a	# and checks the hostname in do_handshake()
895	n/a	if (self._server_hostname and
896	n/a	self._sslcontext.verify_mode != ssl.CERT_NONE):
897	n/a	try:
898	n/a	ssl.match_hostname(peercert, self._server_hostname)
899	n/a	except Exception as exc:
900	n/a	if self._loop.get_debug():
901	n/a	logger.warning("%r: SSL handshake failed "
902	n/a	"on matching the hostname",
903	n/a	self, exc_info=True)
904	n/a	self._sock.close()
905	n/a	self._wakeup_waiter(exc)
906	n/a	return
907	n/a
908	n/a	# Add extra info that becomes available after handshake.
909	n/a	self._extra.update(peercert=peercert,
910	n/a	cipher=self._sock.cipher(),
911	n/a	compression=self._sock.compression(),
912	n/a	ssl_object=self._sock,
913	n/a	)
914	n/a
915	n/a	self._read_wants_write = False
916	n/a	self._write_wants_read = False
917	n/a	self._loop._add_reader(self._sock_fd, self._read_ready)
918	n/a	self._protocol_connected = True
919	n/a	self._loop.call_soon(self._protocol.connection_made, self)
920	n/a	# only wake up the waiter when connection_made() has been called
921	n/a	self._loop.call_soon(self._wakeup_waiter)
922	n/a
923	n/a	if self._loop.get_debug():
924	n/a	dt = self._loop.time() - start_time
925	n/a	logger.debug("%r: SSL handshake took %.1f ms", self, dt * 1e3)
926	n/a
927	n/a	def pause_reading(self):
928	n/a	# XXX This is a bit icky, given the comment at the top of
929	n/a	# _read_ready(). Is it possible to evoke a deadlock? I don't
930	n/a	# know, although it doesn't look like it; write() will still
931	n/a	# accept more data for the buffer and eventually the app will
932	n/a	# call resume_reading() again, and things will flow again.
933	n/a
934	n/a	if self._closing:
935	n/a	raise RuntimeError('Cannot pause_reading() when closing')
936	n/a	if self._paused:
937	n/a	raise RuntimeError('Already paused')
938	n/a	self._paused = True
939	n/a	self._loop._remove_reader(self._sock_fd)
940	n/a	if self._loop.get_debug():
941	n/a	logger.debug("%r pauses reading", self)
942	n/a
943	n/a	def resume_reading(self):
944	n/a	if not self._paused:
945	n/a	raise RuntimeError('Not paused')
946	n/a	self._paused = False
947	n/a	if self._closing:
948	n/a	return
949	n/a	self._loop._add_reader(self._sock_fd, self._read_ready)
950	n/a	if self._loop.get_debug():
951	n/a	logger.debug("%r resumes reading", self)
952	n/a
953	n/a	def _read_ready(self):
954	n/a	if self._conn_lost:
955	n/a	return
956	n/a	if self._write_wants_read:
957	n/a	self._write_wants_read = False
958	n/a	self._write_ready()
959	n/a
960	n/a	if self._buffer:
961	n/a	self._loop._add_writer(self._sock_fd, self._write_ready)
962	n/a
963	n/a	try:
964	n/a	data = self._sock.recv(self.max_size)
965	n/a	except (BlockingIOError, InterruptedError, ssl.SSLWantReadError):
966	n/a	pass
967	n/a	except ssl.SSLWantWriteError:
968	n/a	self._read_wants_write = True
969	n/a	self._loop._remove_reader(self._sock_fd)
970	n/a	self._loop._add_writer(self._sock_fd, self._write_ready)
971	n/a	except Exception as exc:
972	n/a	self._fatal_error(exc, 'Fatal read error on SSL transport')
973	n/a	else:
974	n/a	if data:
975	n/a	self._protocol.data_received(data)
976	n/a	else:
977	n/a	try:
978	n/a	if self._loop.get_debug():
979	n/a	logger.debug("%r received EOF", self)
980	n/a	keep_open = self._protocol.eof_received()
981	n/a	if keep_open:
982	n/a	logger.warning('returning true from eof_received() '
983	n/a	'has no effect when using ssl')
984	n/a	finally:
985	n/a	self.close()
986	n/a
987	n/a	def _write_ready(self):
988	n/a	if self._conn_lost:
989	n/a	return
990	n/a	if self._read_wants_write:
991	n/a	self._read_wants_write = False
992	n/a	self._read_ready()
993	n/a
994	n/a	if not (self._paused or self._closing):
995	n/a	self._loop._add_reader(self._sock_fd, self._read_ready)
996	n/a
997	n/a	if self._buffer:
998	n/a	try:
999	n/a	n = self._sock.send(self._buffer)
1000	n/a	except (BlockingIOError, InterruptedError, ssl.SSLWantWriteError):
1001	n/a	n = 0
1002	n/a	except ssl.SSLWantReadError:
1003	n/a	n = 0
1004	n/a	self._loop._remove_writer(self._sock_fd)
1005	n/a	self._write_wants_read = True
1006	n/a	except Exception as exc:
1007	n/a	self._loop._remove_writer(self._sock_fd)
1008	n/a	self._buffer.clear()
1009	n/a	self._fatal_error(exc, 'Fatal write error on SSL transport')
1010	n/a	return
1011	n/a
1012	n/a	if n:
1013	n/a	del self._buffer[:n]
1014	n/a
1015	n/a	self._maybe_resume_protocol() # May append to buffer.
1016	n/a
1017	n/a	if not self._buffer:
1018	n/a	self._loop._remove_writer(self._sock_fd)
1019	n/a	if self._closing:
1020	n/a	self._call_connection_lost(None)
1021	n/a
1022	n/a	def write(self, data):
1023	n/a	if not isinstance(data, (bytes, bytearray, memoryview)):
1024	n/a	raise TypeError('data argument must be a bytes-like object, '
1025	n/a	'not %r' % type(data).__name__)
1026	n/a	if not data:
1027	n/a	return
1028	n/a
1029	n/a	if self._conn_lost:
1030	n/a	if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES:
1031	n/a	logger.warning('socket.send() raised exception.')
1032	n/a	self._conn_lost += 1
1033	n/a	return
1034	n/a
1035	n/a	if not self._buffer:
1036	n/a	self._loop._add_writer(self._sock_fd, self._write_ready)
1037	n/a
1038	n/a	# Add it to the buffer.
1039	n/a	self._buffer.extend(data)
1040	n/a	self._maybe_pause_protocol()
1041	n/a
1042	n/a	def can_write_eof(self):
1043	n/a	return False
1044	n/a
1045	n/a
1046	n/a	class _SelectorDatagramTransport(_SelectorTransport):
1047	n/a
1048	n/a	_buffer_factory = collections.deque
1049	n/a
1050	n/a	def __init__(self, loop, sock, protocol, address=None,
1051	n/a	waiter=None, extra=None):
1052	n/a	super().__init__(loop, sock, protocol, extra)
1053	n/a	self._address = address
1054	n/a	self._loop.call_soon(self._protocol.connection_made, self)
1055	n/a	# only start reading when connection_made() has been called
1056	n/a	self._loop.call_soon(self._loop._add_reader,
1057	n/a	self._sock_fd, self._read_ready)
1058	n/a	if waiter is not None:
1059	n/a	# only wake up the waiter when connection_made() has been called
1060	n/a	self._loop.call_soon(futures._set_result_unless_cancelled,
1061	n/a	waiter, None)
1062	n/a
1063	n/a	def get_write_buffer_size(self):
1064	n/a	return sum(len(data) for data, _ in self._buffer)
1065	n/a
1066	n/a	def _read_ready(self):
1067	n/a	if self._conn_lost:
1068	n/a	return
1069	n/a	try:
1070	n/a	data, addr = self._sock.recvfrom(self.max_size)
1071	n/a	except (BlockingIOError, InterruptedError):
1072	n/a	pass
1073	n/a	except OSError as exc:
1074	n/a	self._protocol.error_received(exc)
1075	n/a	except Exception as exc:
1076	n/a	self._fatal_error(exc, 'Fatal read error on datagram transport')
1077	n/a	else:
1078	n/a	self._protocol.datagram_received(data, addr)
1079	n/a
1080	n/a	def sendto(self, data, addr=None):
1081	n/a	if not isinstance(data, (bytes, bytearray, memoryview)):
1082	n/a	raise TypeError('data argument must be a bytes-like object, '
1083	n/a	'not %r' % type(data).__name__)
1084	n/a	if not data:
1085	n/a	return
1086	n/a
1087	n/a	if self._address and addr not in (None, self._address):
1088	n/a	raise ValueError('Invalid address: must be None or %s' %
1089	n/a	(self._address,))
1090	n/a
1091	n/a	if self._conn_lost and self._address:
1092	n/a	if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES:
1093	n/a	logger.warning('socket.send() raised exception.')
1094	n/a	self._conn_lost += 1
1095	n/a	return
1096	n/a
1097	n/a	if not self._buffer:
1098	n/a	# Attempt to send it right away first.
1099	n/a	try:
1100	n/a	if self._address:
1101	n/a	self._sock.send(data)
1102	n/a	else:
1103	n/a	self._sock.sendto(data, addr)
1104	n/a	return
1105	n/a	except (BlockingIOError, InterruptedError):
1106	n/a	self._loop._add_writer(self._sock_fd, self._sendto_ready)
1107	n/a	except OSError as exc:
1108	n/a	self._protocol.error_received(exc)
1109	n/a	return
1110	n/a	except Exception as exc:
1111	n/a	self._fatal_error(exc,
1112	n/a	'Fatal write error on datagram transport')
1113	n/a	return
1114	n/a
1115	n/a	# Ensure that what we buffer is immutable.
1116	n/a	self._buffer.append((bytes(data), addr))
1117	n/a	self._maybe_pause_protocol()
1118	n/a
1119	n/a	def _sendto_ready(self):
1120	n/a	while self._buffer:
1121	n/a	data, addr = self._buffer.popleft()
1122	n/a	try:
1123	n/a	if self._address:
1124	n/a	self._sock.send(data)
1125	n/a	else:
1126	n/a	self._sock.sendto(data, addr)
1127	n/a	except (BlockingIOError, InterruptedError):
1128	n/a	self._buffer.appendleft((data, addr)) # Try again later.
1129	n/a	break
1130	n/a	except OSError as exc:
1131	n/a	self._protocol.error_received(exc)
1132	n/a	return
1133	n/a	except Exception as exc:
1134	n/a	self._fatal_error(exc,
1135	n/a	'Fatal write error on datagram transport')
1136	n/a	return
1137	n/a
1138	n/a	self._maybe_resume_protocol() # May append to buffer.
1139	n/a	if not self._buffer:
1140	n/a	self._loop._remove_writer(self._sock_fd)
1141	n/a	if self._closing:
1142	n/a	self._call_connection_lost(None)