12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097 |
- """Event loop using a selector and related classes.
- A selector is a "notify-when-ready" multiplexer. For a subclass which
- also includes support for signal handling, see the unix_events sub-module.
- """
- __all__ = 'BaseSelectorEventLoop',
- import collections
- import errno
- import functools
- import selectors
- import socket
- import warnings
- import weakref
- try:
- import ssl
- except ImportError: # pragma: no cover
- ssl = None
- from . import base_events
- from . import constants
- from . import events
- from . import futures
- from . import protocols
- from . import sslproto
- from . import transports
- from . import trsock
- from .log import logger
- def _test_selector_event(selector, fd, event):
- # Test if the selector is monitoring 'event' events
- # for the file descriptor 'fd'.
- try:
- key = selector.get_key(fd)
- except KeyError:
- return False
- else:
- return bool(key.events & event)
- class BaseSelectorEventLoop(base_events.BaseEventLoop):
- """Selector event loop.
- See events.EventLoop for API specification.
- """
- def __init__(self, selector=None):
- super().__init__()
- if selector is None:
- selector = selectors.DefaultSelector()
- logger.debug('Using selector: %s', selector.__class__.__name__)
- self._selector = selector
- self._make_self_pipe()
- self._transports = weakref.WeakValueDictionary()
- def _make_socket_transport(self, sock, protocol, waiter=None, *,
- extra=None, server=None):
- return _SelectorSocketTransport(self, sock, protocol, waiter,
- extra, server)
- def _make_ssl_transport(
- self, rawsock, protocol, sslcontext, waiter=None,
- *, server_side=False, server_hostname=None,
- extra=None, server=None,
- ssl_handshake_timeout=constants.SSL_HANDSHAKE_TIMEOUT):
- ssl_protocol = sslproto.SSLProtocol(
- self, protocol, sslcontext, waiter,
- server_side, server_hostname,
- ssl_handshake_timeout=ssl_handshake_timeout)
- _SelectorSocketTransport(self, rawsock, ssl_protocol,
- extra=extra, server=server)
- return ssl_protocol._app_transport
- def _make_datagram_transport(self, sock, protocol,
- address=None, waiter=None, extra=None):
- return _SelectorDatagramTransport(self, sock, protocol,
- address, waiter, extra)
- def close(self):
- if self.is_running():
- raise RuntimeError("Cannot close a running event loop")
- if self.is_closed():
- return
- self._close_self_pipe()
- super().close()
- if self._selector is not None:
- self._selector.close()
- self._selector = None
- def _close_self_pipe(self):
- self._remove_reader(self._ssock.fileno())
- self._ssock.close()
- self._ssock = None
- self._csock.close()
- self._csock = None
- self._internal_fds -= 1
- def _make_self_pipe(self):
- # A self-socket, really. :-)
- self._ssock, self._csock = socket.socketpair()
- self._ssock.setblocking(False)
- self._csock.setblocking(False)
- self._internal_fds += 1
- self._add_reader(self._ssock.fileno(), self._read_from_self)
- def _process_self_data(self, data):
- pass
- def _read_from_self(self):
- while True:
- try:
- data = self._ssock.recv(4096)
- if not data:
- break
- self._process_self_data(data)
- except InterruptedError:
- continue
- except BlockingIOError:
- break
- def _write_to_self(self):
- # This may be called from a different thread, possibly after
- # _close_self_pipe() has been called or even while it is
- # running. Guard for self._csock being None or closed. When
- # a socket is closed, send() raises OSError (with errno set to
- # EBADF, but let's not rely on the exact error code).
- csock = self._csock
- if csock is None:
- return
- try:
- csock.send(b'\0')
- except OSError:
- if self._debug:
- logger.debug("Fail to write a null byte into the "
- "self-pipe socket",
- exc_info=True)
- def _start_serving(self, protocol_factory, sock,
- sslcontext=None, server=None, backlog=100,
- ssl_handshake_timeout=constants.SSL_HANDSHAKE_TIMEOUT):
- self._add_reader(sock.fileno(), self._accept_connection,
- protocol_factory, sock, sslcontext, server, backlog,
- ssl_handshake_timeout)
- def _accept_connection(
- self, protocol_factory, sock,
- sslcontext=None, server=None, backlog=100,
- ssl_handshake_timeout=constants.SSL_HANDSHAKE_TIMEOUT):
- # This method is only called once for each event loop tick where the
- # listening socket has triggered an EVENT_READ. There may be multiple
- # connections waiting for an .accept() so it is called in a loop.
- # See https://bugs.python.org/issue27906 for more details.
- for _ in range(backlog):
- try:
- conn, addr = sock.accept()
- if self._debug:
- logger.debug("%r got a new connection from %r: %r",
- server, addr, conn)
- conn.setblocking(False)
- except (BlockingIOError, InterruptedError, ConnectionAbortedError):
- # Early exit because the socket accept buffer is empty.
- return None
- except OSError as exc:
- # There's nowhere to send the error, so just log it.
- if exc.errno in (errno.EMFILE, errno.ENFILE,
- errno.ENOBUFS, errno.ENOMEM):
- # Some platforms (e.g. Linux keep reporting the FD as
- # ready, so we remove the read handler temporarily.
- # We'll try again in a while.
- self.call_exception_handler({
- 'message': 'socket.accept() out of system resource',
- 'exception': exc,
- 'socket': trsock.TransportSocket(sock),
- })
- self._remove_reader(sock.fileno())
- self.call_later(constants.ACCEPT_RETRY_DELAY,
- self._start_serving,
- protocol_factory, sock, sslcontext, server,
- backlog, ssl_handshake_timeout)
- else:
- raise # The event loop will catch, log and ignore it.
- else:
- extra = {'peername': addr}
- accept = self._accept_connection2(
- protocol_factory, conn, extra, sslcontext, server,
- ssl_handshake_timeout)
- self.create_task(accept)
- async def _accept_connection2(
- self, protocol_factory, conn, extra,
- sslcontext=None, server=None,
- ssl_handshake_timeout=constants.SSL_HANDSHAKE_TIMEOUT):
- protocol = None
- transport = None
- try:
- protocol = protocol_factory()
- waiter = self.create_future()
- if sslcontext:
- transport = self._make_ssl_transport(
- conn, protocol, sslcontext, waiter=waiter,
- server_side=True, extra=extra, server=server,
- ssl_handshake_timeout=ssl_handshake_timeout)
- else:
- transport = self._make_socket_transport(
- conn, protocol, waiter=waiter, extra=extra,
- server=server)
- try:
- await waiter
- except BaseException:
- transport.close()
- raise
- # It's now up to the protocol to handle the connection.
- except (SystemExit, KeyboardInterrupt):
- raise
- except BaseException as exc:
- if self._debug:
- context = {
- 'message':
- 'Error on transport creation for incoming connection',
- 'exception': exc,
- }
- if protocol is not None:
- context['protocol'] = protocol
- if transport is not None:
- context['transport'] = transport
- self.call_exception_handler(context)
- def _ensure_fd_no_transport(self, fd):
- fileno = fd
- if not isinstance(fileno, int):
- try:
- fileno = int(fileno.fileno())
- except (AttributeError, TypeError, ValueError):
- # This code matches selectors._fileobj_to_fd function.
- raise ValueError(f"Invalid file object: {fd!r}") from None
- try:
- transport = self._transports[fileno]
- except KeyError:
- pass
- else:
- if not transport.is_closing():
- raise RuntimeError(
- f'File descriptor {fd!r} is used by transport '
- f'{transport!r}')
- def _add_reader(self, fd, callback, *args):
- self._check_closed()
- handle = events.Handle(callback, args, self, None)
- try:
- key = self._selector.get_key(fd)
- except KeyError:
- self._selector.register(fd, selectors.EVENT_READ,
- (handle, None))
- else:
- mask, (reader, writer) = key.events, key.data
- self._selector.modify(fd, mask | selectors.EVENT_READ,
- (handle, writer))
- if reader is not None:
- reader.cancel()
- return handle
- def _remove_reader(self, fd):
- if self.is_closed():
- return False
- try:
- key = self._selector.get_key(fd)
- except KeyError:
- return False
- else:
- mask, (reader, writer) = key.events, key.data
- mask &= ~selectors.EVENT_READ
- if not mask:
- self._selector.unregister(fd)
- else:
- self._selector.modify(fd, mask, (None, writer))
- if reader is not None:
- reader.cancel()
- return True
- else:
- return False
- def _add_writer(self, fd, callback, *args):
- self._check_closed()
- handle = events.Handle(callback, args, self, None)
- try:
- key = self._selector.get_key(fd)
- except KeyError:
- self._selector.register(fd, selectors.EVENT_WRITE,
- (None, handle))
- else:
- mask, (reader, writer) = key.events, key.data
- self._selector.modify(fd, mask | selectors.EVENT_WRITE,
- (reader, handle))
- if writer is not None:
- writer.cancel()
- return handle
- def _remove_writer(self, fd):
- """Remove a writer callback."""
- if self.is_closed():
- return False
- try:
- key = self._selector.get_key(fd)
- except KeyError:
- return False
- else:
- mask, (reader, writer) = key.events, key.data
- # Remove both writer and connector.
- mask &= ~selectors.EVENT_WRITE
- if not mask:
- self._selector.unregister(fd)
- else:
- self._selector.modify(fd, mask, (reader, None))
- if writer is not None:
- writer.cancel()
- return True
- else:
- return False
- def add_reader(self, fd, callback, *args):
- """Add a reader callback."""
- self._ensure_fd_no_transport(fd)
- self._add_reader(fd, callback, *args)
- def remove_reader(self, fd):
- """Remove a reader callback."""
- self._ensure_fd_no_transport(fd)
- return self._remove_reader(fd)
- def add_writer(self, fd, callback, *args):
- """Add a writer callback.."""
- self._ensure_fd_no_transport(fd)
- self._add_writer(fd, callback, *args)
- def remove_writer(self, fd):
- """Remove a writer callback."""
- self._ensure_fd_no_transport(fd)
- return self._remove_writer(fd)
- async def sock_recv(self, sock, n):
- """Receive data from the socket.
- The return value is a bytes object representing the data received.
- The maximum amount of data to be received at once is specified by
- nbytes.
- """
- base_events._check_ssl_socket(sock)
- if self._debug and sock.gettimeout() != 0:
- raise ValueError("the socket must be non-blocking")
- try:
- return sock.recv(n)
- except (BlockingIOError, InterruptedError):
- pass
- fut = self.create_future()
- fd = sock.fileno()
- self._ensure_fd_no_transport(fd)
- handle = self._add_reader(fd, self._sock_recv, fut, sock, n)
- fut.add_done_callback(
- functools.partial(self._sock_read_done, fd, handle=handle))
- return await fut
- def _sock_read_done(self, fd, fut, handle=None):
- if handle is None or not handle.cancelled():
- self.remove_reader(fd)
- def _sock_recv(self, fut, sock, n):
- # _sock_recv() can add itself as an I/O callback if the operation can't
- # be done immediately. Don't use it directly, call sock_recv().
- if fut.done():
- return
- try:
- data = sock.recv(n)
- except (BlockingIOError, InterruptedError):
- return # try again next time
- except (SystemExit, KeyboardInterrupt):
- raise
- except BaseException as exc:
- fut.set_exception(exc)
- else:
- fut.set_result(data)
- async def sock_recv_into(self, sock, buf):
- """Receive data from the socket.
- The received data is written into *buf* (a writable buffer).
- The return value is the number of bytes written.
- """
- base_events._check_ssl_socket(sock)
- if self._debug and sock.gettimeout() != 0:
- raise ValueError("the socket must be non-blocking")
- try:
- return sock.recv_into(buf)
- except (BlockingIOError, InterruptedError):
- pass
- fut = self.create_future()
- fd = sock.fileno()
- self._ensure_fd_no_transport(fd)
- handle = self._add_reader(fd, self._sock_recv_into, fut, sock, buf)
- fut.add_done_callback(
- functools.partial(self._sock_read_done, fd, handle=handle))
- return await fut
- def _sock_recv_into(self, fut, sock, buf):
- # _sock_recv_into() can add itself as an I/O callback if the operation
- # can't be done immediately. Don't use it directly, call
- # sock_recv_into().
- if fut.done():
- return
- try:
- nbytes = sock.recv_into(buf)
- except (BlockingIOError, InterruptedError):
- return # try again next time
- except (SystemExit, KeyboardInterrupt):
- raise
- except BaseException as exc:
- fut.set_exception(exc)
- else:
- fut.set_result(nbytes)
- async def sock_sendall(self, sock, data):
- """Send data to the socket.
- The socket must be connected to a remote socket. This method continues
- to send data from data until either all data has been sent or an
- error occurs. None is returned on success. On error, an exception is
- raised, and there is no way to determine how much data, if any, was
- successfully processed by the receiving end of the connection.
- """
- base_events._check_ssl_socket(sock)
- if self._debug and sock.gettimeout() != 0:
- raise ValueError("the socket must be non-blocking")
- try:
- n = sock.send(data)
- except (BlockingIOError, InterruptedError):
- n = 0
- if n == len(data):
- # all data sent
- return
- fut = self.create_future()
- fd = sock.fileno()
- self._ensure_fd_no_transport(fd)
- # use a trick with a list in closure to store a mutable state
- handle = self._add_writer(fd, self._sock_sendall, fut, sock,
- memoryview(data), [n])
- fut.add_done_callback(
- functools.partial(self._sock_write_done, fd, handle=handle))
- return await fut
- def _sock_sendall(self, fut, sock, view, pos):
- if fut.done():
- # Future cancellation can be scheduled on previous loop iteration
- return
- start = pos[0]
- try:
- n = sock.send(view[start:])
- except (BlockingIOError, InterruptedError):
- return
- except (SystemExit, KeyboardInterrupt):
- raise
- except BaseException as exc:
- fut.set_exception(exc)
- return
- start += n
- if start == len(view):
- fut.set_result(None)
- else:
- pos[0] = start
- async def sock_connect(self, sock, address):
- """Connect to a remote socket at address.
- This method is a coroutine.
- """
- base_events._check_ssl_socket(sock)
- if self._debug and sock.gettimeout() != 0:
- raise ValueError("the socket must be non-blocking")
- if sock.family == socket.AF_INET or (
- base_events._HAS_IPv6 and sock.family == socket.AF_INET6):
- resolved = await self._ensure_resolved(
- address, family=sock.family, type=sock.type, proto=sock.proto,
- loop=self,
- )
- _, _, _, _, address = resolved[0]
- fut = self.create_future()
- self._sock_connect(fut, sock, address)
- return await fut
- def _sock_connect(self, fut, sock, address):
- fd = sock.fileno()
- try:
- sock.connect(address)
- except (BlockingIOError, InterruptedError):
- # Issue #23618: When the C function connect() fails with EINTR, the
- # connection runs in background. We have to wait until the socket
- # becomes writable to be notified when the connection succeed or
- # fails.
- self._ensure_fd_no_transport(fd)
- handle = self._add_writer(
- fd, self._sock_connect_cb, fut, sock, address)
- fut.add_done_callback(
- functools.partial(self._sock_write_done, fd, handle=handle))
- except (SystemExit, KeyboardInterrupt):
- raise
- except BaseException as exc:
- fut.set_exception(exc)
- else:
- fut.set_result(None)
- def _sock_write_done(self, fd, fut, handle=None):
- if handle is None or not handle.cancelled():
- self.remove_writer(fd)
- def _sock_connect_cb(self, fut, sock, address):
- if fut.done():
- return
- try:
- err = sock.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
- if err != 0:
- # Jump to any except clause below.
- raise OSError(err, f'Connect call failed {address}')
- except (BlockingIOError, InterruptedError):
- # socket is still registered, the callback will be retried later
- pass
- except (SystemExit, KeyboardInterrupt):
- raise
- except BaseException as exc:
- fut.set_exception(exc)
- else:
- fut.set_result(None)
- async def sock_accept(self, sock):
- """Accept a connection.
- The socket must be bound to an address and listening for connections.
- The return value is a pair (conn, address) where conn is a new socket
- object usable to send and receive data on the connection, and address
- is the address bound to the socket on the other end of the connection.
- """
- base_events._check_ssl_socket(sock)
- if self._debug and sock.gettimeout() != 0:
- raise ValueError("the socket must be non-blocking")
- fut = self.create_future()
- self._sock_accept(fut, sock)
- return await fut
- def _sock_accept(self, fut, sock):
- fd = sock.fileno()
- try:
- conn, address = sock.accept()
- conn.setblocking(False)
- except (BlockingIOError, InterruptedError):
- self._ensure_fd_no_transport(fd)
- handle = self._add_reader(fd, self._sock_accept, fut, sock)
- fut.add_done_callback(
- functools.partial(self._sock_read_done, fd, handle=handle))
- except (SystemExit, KeyboardInterrupt):
- raise
- except BaseException as exc:
- fut.set_exception(exc)
- else:
- fut.set_result((conn, address))
- async def _sendfile_native(self, transp, file, offset, count):
- del self._transports[transp._sock_fd]
- resume_reading = transp.is_reading()
- transp.pause_reading()
- await transp._make_empty_waiter()
- try:
- return await self.sock_sendfile(transp._sock, file, offset, count,
- fallback=False)
- finally:
- transp._reset_empty_waiter()
- if resume_reading:
- transp.resume_reading()
- self._transports[transp._sock_fd] = transp
- def _process_events(self, event_list):
- for key, mask in event_list:
- fileobj, (reader, writer) = key.fileobj, key.data
- if mask & selectors.EVENT_READ and reader is not None:
- if reader._cancelled:
- self._remove_reader(fileobj)
- else:
- self._add_callback(reader)
- if mask & selectors.EVENT_WRITE and writer is not None:
- if writer._cancelled:
- self._remove_writer(fileobj)
- else:
- self._add_callback(writer)
- def _stop_serving(self, sock):
- self._remove_reader(sock.fileno())
- sock.close()
- class _SelectorTransport(transports._FlowControlMixin,
- transports.Transport):
- max_size = 256 * 1024 # Buffer size passed to recv().
- _buffer_factory = bytearray # Constructs initial value for self._buffer.
- # Attribute used in the destructor: it must be set even if the constructor
- # is not called (see _SelectorSslTransport which may start by raising an
- # exception)
- _sock = None
- def __init__(self, loop, sock, protocol, extra=None, server=None):
- super().__init__(extra, loop)
- self._extra['socket'] = trsock.TransportSocket(sock)
- try:
- self._extra['sockname'] = sock.getsockname()
- except OSError:
- self._extra['sockname'] = None
- if 'peername' not in self._extra:
- try:
- self._extra['peername'] = sock.getpeername()
- except socket.error:
- self._extra['peername'] = None
- self._sock = sock
- self._sock_fd = sock.fileno()
- self._protocol_connected = False
- self.set_protocol(protocol)
- self._server = server
- self._buffer = self._buffer_factory()
- self._conn_lost = 0 # Set when call to connection_lost scheduled.
- self._closing = False # Set when close() called.
- if self._server is not None:
- self._server._attach()
- loop._transports[self._sock_fd] = self
- def __repr__(self):
- info = [self.__class__.__name__]
- if self._sock is None:
- info.append('closed')
- elif self._closing:
- info.append('closing')
- info.append(f'fd={self._sock_fd}')
- # test if the transport was closed
- if self._loop is not None and not self._loop.is_closed():
- polling = _test_selector_event(self._loop._selector,
- self._sock_fd, selectors.EVENT_READ)
- if polling:
- info.append('read=polling')
- else:
- info.append('read=idle')
- polling = _test_selector_event(self._loop._selector,
- self._sock_fd,
- selectors.EVENT_WRITE)
- if polling:
- state = 'polling'
- else:
- state = 'idle'
- bufsize = self.get_write_buffer_size()
- info.append(f'write=<{state}, bufsize={bufsize}>')
- return '<{}>'.format(' '.join(info))
- def abort(self):
- self._force_close(None)
- def set_protocol(self, protocol):
- self._protocol = protocol
- self._protocol_connected = True
- def get_protocol(self):
- return self._protocol
- def is_closing(self):
- return self._closing
- def close(self):
- if self._closing:
- return
- self._closing = True
- self._loop._remove_reader(self._sock_fd)
- if not self._buffer:
- self._conn_lost += 1
- self._loop._remove_writer(self._sock_fd)
- self._loop.call_soon(self._call_connection_lost, None)
- def __del__(self, _warn=warnings.warn):
- if self._sock is not None:
- _warn(f"unclosed transport {self!r}", ResourceWarning, source=self)
- self._sock.close()
- def _fatal_error(self, exc, message='Fatal error on transport'):
- # Should be called from exception handler only.
- if isinstance(exc, OSError):
- if self._loop.get_debug():
- logger.debug("%r: %s", self, message, exc_info=True)
- else:
- self._loop.call_exception_handler({
- 'message': message,
- 'exception': exc,
- 'transport': self,
- 'protocol': self._protocol,
- })
- self._force_close(exc)
- def _force_close(self, exc):
- if self._conn_lost:
- return
- if self._buffer:
- self._buffer.clear()
- self._loop._remove_writer(self._sock_fd)
- if not self._closing:
- self._closing = True
- self._loop._remove_reader(self._sock_fd)
- self._conn_lost += 1
- self._loop.call_soon(self._call_connection_lost, exc)
- def _call_connection_lost(self, exc):
- try:
- if self._protocol_connected:
- self._protocol.connection_lost(exc)
- finally:
- self._sock.close()
- self._sock = None
- self._protocol = None
- self._loop = None
- server = self._server
- if server is not None:
- server._detach()
- self._server = None
- def get_write_buffer_size(self):
- return len(self._buffer)
- def _add_reader(self, fd, callback, *args):
- if self._closing:
- return
- self._loop._add_reader(fd, callback, *args)
- class _SelectorSocketTransport(_SelectorTransport):
- _start_tls_compatible = True
- _sendfile_compatible = constants._SendfileMode.TRY_NATIVE
- def __init__(self, loop, sock, protocol, waiter=None,
- extra=None, server=None):
- self._read_ready_cb = None
- super().__init__(loop, sock, protocol, extra, server)
- self._eof = False
- self._paused = False
- self._empty_waiter = None
- # Disable the Nagle algorithm -- small writes will be
- # sent without waiting for the TCP ACK. This generally
- # decreases the latency (in some cases significantly.)
- base_events._set_nodelay(self._sock)
- self._loop.call_soon(self._protocol.connection_made, self)
- # only start reading when connection_made() has been called
- self._loop.call_soon(self._add_reader,
- self._sock_fd, self._read_ready)
- if waiter is not None:
- # only wake up the waiter when connection_made() has been called
- self._loop.call_soon(futures._set_result_unless_cancelled,
- waiter, None)
- def set_protocol(self, protocol):
- if isinstance(protocol, protocols.BufferedProtocol):
- self._read_ready_cb = self._read_ready__get_buffer
- else:
- self._read_ready_cb = self._read_ready__data_received
- super().set_protocol(protocol)
- def is_reading(self):
- return not self._paused and not self._closing
- def pause_reading(self):
- if self._closing or self._paused:
- return
- self._paused = True
- self._loop._remove_reader(self._sock_fd)
- if self._loop.get_debug():
- logger.debug("%r pauses reading", self)
- def resume_reading(self):
- if self._closing or not self._paused:
- return
- self._paused = False
- self._add_reader(self._sock_fd, self._read_ready)
- if self._loop.get_debug():
- logger.debug("%r resumes reading", self)
- def _read_ready(self):
- self._read_ready_cb()
- def _read_ready__get_buffer(self):
- if self._conn_lost:
- return
- try:
- buf = self._protocol.get_buffer(-1)
- if not len(buf):
- raise RuntimeError('get_buffer() returned an empty buffer')
- except (SystemExit, KeyboardInterrupt):
- raise
- except BaseException as exc:
- self._fatal_error(
- exc, 'Fatal error: protocol.get_buffer() call failed.')
- return
- try:
- nbytes = self._sock.recv_into(buf)
- except (BlockingIOError, InterruptedError):
- return
- except (SystemExit, KeyboardInterrupt):
- raise
- except BaseException as exc:
- self._fatal_error(exc, 'Fatal read error on socket transport')
- return
- if not nbytes:
- self._read_ready__on_eof()
- return
- try:
- self._protocol.buffer_updated(nbytes)
- except (SystemExit, KeyboardInterrupt):
- raise
- except BaseException as exc:
- self._fatal_error(
- exc, 'Fatal error: protocol.buffer_updated() call failed.')
- def _read_ready__data_received(self):
- if self._conn_lost:
- return
- try:
- data = self._sock.recv(self.max_size)
- except (BlockingIOError, InterruptedError):
- return
- except (SystemExit, KeyboardInterrupt):
- raise
- except BaseException as exc:
- self._fatal_error(exc, 'Fatal read error on socket transport')
- return
- if not data:
- self._read_ready__on_eof()
- return
- try:
- self._protocol.data_received(data)
- except (SystemExit, KeyboardInterrupt):
- raise
- except BaseException as exc:
- self._fatal_error(
- exc, 'Fatal error: protocol.data_received() call failed.')
- def _read_ready__on_eof(self):
- if self._loop.get_debug():
- logger.debug("%r received EOF", self)
- try:
- keep_open = self._protocol.eof_received()
- except (SystemExit, KeyboardInterrupt):
- raise
- except BaseException as exc:
- self._fatal_error(
- exc, 'Fatal error: protocol.eof_received() call failed.')
- return
- if keep_open:
- # We're keeping the connection open so the
- # protocol can write more, but we still can't
- # receive more, so remove the reader callback.
- self._loop._remove_reader(self._sock_fd)
- else:
- self.close()
- def write(self, data):
- if not isinstance(data, (bytes, bytearray, memoryview)):
- raise TypeError(f'data argument must be a bytes-like object, '
- f'not {type(data).__name__!r}')
- if self._eof:
- raise RuntimeError('Cannot call write() after write_eof()')
- if self._empty_waiter is not None:
- raise RuntimeError('unable to write; sendfile is in progress')
- if not data:
- return
- if self._conn_lost:
- if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES:
- logger.warning('socket.send() raised exception.')
- self._conn_lost += 1
- return
- if not self._buffer:
- # Optimization: try to send now.
- try:
- n = self._sock.send(data)
- except (BlockingIOError, InterruptedError):
- pass
- except (SystemExit, KeyboardInterrupt):
- raise
- except BaseException as exc:
- self._fatal_error(exc, 'Fatal write error on socket transport')
- return
- else:
- data = data[n:]
- if not data:
- return
- # Not all was written; register write handler.
- self._loop._add_writer(self._sock_fd, self._write_ready)
- # Add it to the buffer.
- self._buffer.extend(data)
- self._maybe_pause_protocol()
- def _write_ready(self):
- assert self._buffer, 'Data should not be empty'
- if self._conn_lost:
- return
- try:
- n = self._sock.send(self._buffer)
- except (BlockingIOError, InterruptedError):
- pass
- except (SystemExit, KeyboardInterrupt):
- raise
- except BaseException as exc:
- self._loop._remove_writer(self._sock_fd)
- self._buffer.clear()
- self._fatal_error(exc, 'Fatal write error on socket transport')
- if self._empty_waiter is not None:
- self._empty_waiter.set_exception(exc)
- else:
- if n:
- del self._buffer[:n]
- self._maybe_resume_protocol() # May append to buffer.
- if not self._buffer:
- self._loop._remove_writer(self._sock_fd)
- if self._empty_waiter is not None:
- self._empty_waiter.set_result(None)
- if self._closing:
- self._call_connection_lost(None)
- elif self._eof:
- self._sock.shutdown(socket.SHUT_WR)
- def write_eof(self):
- if self._closing or self._eof:
- return
- self._eof = True
- if not self._buffer:
- self._sock.shutdown(socket.SHUT_WR)
- def can_write_eof(self):
- return True
- def _call_connection_lost(self, exc):
- super()._call_connection_lost(exc)
- if self._empty_waiter is not None:
- self._empty_waiter.set_exception(
- ConnectionError("Connection is closed by peer"))
- def _make_empty_waiter(self):
- if self._empty_waiter is not None:
- raise RuntimeError("Empty waiter is already set")
- self._empty_waiter = self._loop.create_future()
- if not self._buffer:
- self._empty_waiter.set_result(None)
- return self._empty_waiter
- def _reset_empty_waiter(self):
- self._empty_waiter = None
- class _SelectorDatagramTransport(_SelectorTransport):
- _buffer_factory = collections.deque
- def __init__(self, loop, sock, protocol, address=None,
- waiter=None, extra=None):
- super().__init__(loop, sock, protocol, extra)
- self._address = address
- self._loop.call_soon(self._protocol.connection_made, self)
- # only start reading when connection_made() has been called
- self._loop.call_soon(self._add_reader,
- self._sock_fd, self._read_ready)
- if waiter is not None:
- # only wake up the waiter when connection_made() has been called
- self._loop.call_soon(futures._set_result_unless_cancelled,
- waiter, None)
- def get_write_buffer_size(self):
- return sum(len(data) for data, _ in self._buffer)
- def _read_ready(self):
- if self._conn_lost:
- return
- try:
- data, addr = self._sock.recvfrom(self.max_size)
- except (BlockingIOError, InterruptedError):
- pass
- except OSError as exc:
- self._protocol.error_received(exc)
- except (SystemExit, KeyboardInterrupt):
- raise
- except BaseException as exc:
- self._fatal_error(exc, 'Fatal read error on datagram transport')
- else:
- self._protocol.datagram_received(data, addr)
- def sendto(self, data, addr=None):
- if not isinstance(data, (bytes, bytearray, memoryview)):
- raise TypeError(f'data argument must be a bytes-like object, '
- f'not {type(data).__name__!r}')
- if not data:
- return
- if self._address:
- if addr not in (None, self._address):
- raise ValueError(
- f'Invalid address: must be None or {self._address}')
- addr = self._address
- if self._conn_lost and self._address:
- if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES:
- logger.warning('socket.send() raised exception.')
- self._conn_lost += 1
- return
- if not self._buffer:
- # Attempt to send it right away first.
- try:
- if self._extra['peername']:
- self._sock.send(data)
- else:
- self._sock.sendto(data, addr)
- return
- except (BlockingIOError, InterruptedError):
- self._loop._add_writer(self._sock_fd, self._sendto_ready)
- except OSError as exc:
- self._protocol.error_received(exc)
- return
- except (SystemExit, KeyboardInterrupt):
- raise
- except BaseException as exc:
- self._fatal_error(
- exc, 'Fatal write error on datagram transport')
- return
- # Ensure that what we buffer is immutable.
- self._buffer.append((bytes(data), addr))
- self._maybe_pause_protocol()
- def _sendto_ready(self):
- while self._buffer:
- data, addr = self._buffer.popleft()
- try:
- if self._extra['peername']:
- self._sock.send(data)
- else:
- self._sock.sendto(data, addr)
- except (BlockingIOError, InterruptedError):
- self._buffer.appendleft((data, addr)) # Try again later.
- break
- except OSError as exc:
- self._protocol.error_received(exc)
- return
- except (SystemExit, KeyboardInterrupt):
- raise
- except BaseException as exc:
- self._fatal_error(
- exc, 'Fatal write error on datagram transport')
- return
- self._maybe_resume_protocol() # May append to buffer.
- if not self._buffer:
- self._loop._remove_writer(self._sock_fd)
- if self._closing:
- self._call_connection_lost(None)
|