# "High-level" networking interface import errno from contextlib import contextmanager import trio from . import socket as tsocket from ._util import ConflictDetector, Final from .abc import HalfCloseableStream, Listener # XX TODO: this number was picked arbitrarily. We should do experiments to # tune it. (Or make it dynamic -- one idea is to start small and increase it # if we observe single reads filling up the whole buffer, at least within some # limits.) DEFAULT_RECEIVE_SIZE = 65536 _closed_stream_errnos = { # Unix errno.EBADF, # Windows errno.ENOTSOCK, } @contextmanager def _translate_socket_errors_to_stream_errors(): try: yield except OSError as exc: if exc.errno in _closed_stream_errnos: raise trio.ClosedResourceError("this socket was already closed") from None else: raise trio.BrokenResourceError( "socket connection broken: {}".format(exc) ) from exc class SocketStream(HalfCloseableStream, metaclass=Final): """An implementation of the :class:`trio.abc.HalfCloseableStream` interface based on a raw network socket. Args: socket: The Trio socket object to wrap. Must have type ``SOCK_STREAM``, and be connected. By default for TCP sockets, :class:`SocketStream` enables ``TCP_NODELAY``, and (on platforms where it's supported) enables ``TCP_NOTSENT_LOWAT`` with a reasonable buffer size (currently 16 KiB) – see `issue #72 `__ for discussion. You can of course override these defaults by calling :meth:`setsockopt`. Once a :class:`SocketStream` object is constructed, it implements the full :class:`trio.abc.HalfCloseableStream` interface. In addition, it provides a few extra features: .. attribute:: socket The Trio socket object that this stream wraps. """ def __init__(self, socket): if not isinstance(socket, tsocket.SocketType): raise TypeError("SocketStream requires a Trio socket object") if socket.type != tsocket.SOCK_STREAM: raise ValueError("SocketStream requires a SOCK_STREAM socket") self.socket = socket self._send_conflict_detector = ConflictDetector( "another task is currently sending data on this SocketStream" ) # Socket defaults: # Not supported on e.g. unix domain sockets try: self.setsockopt(tsocket.IPPROTO_TCP, tsocket.TCP_NODELAY, True) except OSError: pass if hasattr(tsocket, "TCP_NOTSENT_LOWAT"): try: # 16 KiB is pretty arbitrary and could probably do with some # tuning. (Apple is also setting this by default in CFNetwork # apparently -- I'm curious what value they're using, though I # couldn't find it online trivially. CFNetwork-129.20 source # has no mentions of TCP_NOTSENT_LOWAT. This presentation says # "typically 8 kilobytes": # http://devstreaming.apple.com/videos/wwdc/2015/719ui2k57m/719/719_your_app_and_next_generation_networks.pdf?dl=1 # ). The theory is that you want it to be bandwidth * # rescheduling interval. self.setsockopt(tsocket.IPPROTO_TCP, tsocket.TCP_NOTSENT_LOWAT, 2**14) except OSError: pass async def send_all(self, data): if self.socket.did_shutdown_SHUT_WR: raise trio.ClosedResourceError("can't send data after sending EOF") with self._send_conflict_detector: with _translate_socket_errors_to_stream_errors(): with memoryview(data) as data: if not data: if self.socket.fileno() == -1: raise trio.ClosedResourceError("socket was already closed") await trio.lowlevel.checkpoint() return total_sent = 0 while total_sent < len(data): with data[total_sent:] as remaining: sent = await self.socket.send(remaining) total_sent += sent async def wait_send_all_might_not_block(self): with self._send_conflict_detector: if self.socket.fileno() == -1: raise trio.ClosedResourceError with _translate_socket_errors_to_stream_errors(): await self.socket.wait_writable() async def send_eof(self): with self._send_conflict_detector: await trio.lowlevel.checkpoint() # On macOS, calling shutdown a second time raises ENOTCONN, but # send_eof needs to be idempotent. if self.socket.did_shutdown_SHUT_WR: return with _translate_socket_errors_to_stream_errors(): self.socket.shutdown(tsocket.SHUT_WR) async def receive_some(self, max_bytes=None): if max_bytes is None: max_bytes = DEFAULT_RECEIVE_SIZE if max_bytes < 1: raise ValueError("max_bytes must be >= 1") with _translate_socket_errors_to_stream_errors(): return await self.socket.recv(max_bytes) async def aclose(self): self.socket.close() await trio.lowlevel.checkpoint() # __aenter__, __aexit__ inherited from HalfCloseableStream are OK def setsockopt(self, level, option, value): """Set an option on the underlying socket. See :meth:`socket.socket.setsockopt` for details. """ return self.socket.setsockopt(level, option, value) def getsockopt(self, level, option, buffersize=0): """Check the current value of an option on the underlying socket. See :meth:`socket.socket.getsockopt` for details. """ # This is to work around # https://bitbucket.org/pypy/pypy/issues/2561 # We should be able to drop it when the next PyPy3 beta is released. if buffersize == 0: return self.socket.getsockopt(level, option) else: return self.socket.getsockopt(level, option, buffersize) ################################################################ # SocketListener ################################################################ # Accept error handling # ===================== # # Literature review # ----------------- # # Here's a list of all the possible errors that accept() can return, according # to the POSIX spec or the Linux, FreeBSD, macOS, and Windows docs: # # Can't happen with a Trio socket: # - EAGAIN/(WSA)EWOULDBLOCK # - EINTR # - WSANOTINITIALISED # - WSAEINPROGRESS: a blocking call is already in progress # - WSAEINTR: someone called WSACancelBlockingCall, but we don't make blocking # calls in the first place # # Something is wrong with our call: # - EBADF: not a file descriptor # - (WSA)EINVAL: socket isn't listening, or (Linux, BSD) bad flags # - (WSA)ENOTSOCK: not a socket # - (WSA)EOPNOTSUPP: this kind of socket doesn't support accept # - (Linux, FreeBSD, Windows) EFAULT: the sockaddr pointer points to readonly # memory # # Something is wrong with the environment: # - (WSA)EMFILE: this process hit its fd limit # - ENFILE: the system hit its fd limit # - (WSA)ENOBUFS, ENOMEM: unspecified memory problems # # Something is wrong with the connection we were going to accept. There's a # ton of variability between systems here: # - ECONNABORTED: documented everywhere, but apparently only the BSDs do this # (signals a connection was closed/reset before being accepted) # - EPROTO: unspecified protocol error # - (Linux) EPERM: firewall rule prevented connection # - (Linux) ENETDOWN, EPROTO, ENOPROTOOPT, EHOSTDOWN, ENONET, EHOSTUNREACH, # EOPNOTSUPP, ENETUNREACH, ENOSR, ESOCKTNOSUPPORT, EPROTONOSUPPORT, # ETIMEDOUT, ... or any other error that the socket could give, because # apparently if an error happens on a connection before it's accept()ed, # Linux will report that error from accept(). # - (Windows) WSAECONNRESET, WSAENETDOWN # # # Code review # ----------- # # What do other libraries do? # # Twisted on Unix or when using nonblocking I/O on Windows: # - ignores EPERM, with comment about Linux firewalls # - logs and ignores EMFILE, ENOBUFS, ENFILE, ENOMEM, ECONNABORTED # Comment notes that ECONNABORTED is a BSDism and that Linux returns the # socket before having it fail, and macOS just silently discards it. # - other errors are raised, which is logged + kills the socket # ref: src/twisted/internet/tcp.py, Port.doRead # # Twisted using IOCP on Windows: # - logs and ignores all errors # ref: src/twisted/internet/iocpreactor/tcp.py, Port.handleAccept # # Tornado: # - ignore ECONNABORTED (comments notes that it was observed on FreeBSD) # - everything else raised, but all this does (by default) is cause it to be # logged and then ignored # (ref: tornado/netutil.py, tornado/ioloop.py) # # libuv on Unix: # - ignores ECONNABORTED # - does a "trick" for EMFILE or ENFILE # - all other errors passed to the connection_cb to be handled # (ref: src/unix/stream.c:uv__server_io, uv__emfile_trick) # # libuv on Windows: # src/win/tcp.c:uv_tcp_queue_accept # this calls AcceptEx, and then arranges to call: # src/win/tcp.c:uv_process_tcp_accept_req # this gets the result from AcceptEx. If the original AcceptEx call failed, # then "we stop accepting connections and report this error to the # connection callback". I think this is for things like ENOTSOCK. If # AcceptEx successfully queues an overlapped operation, and then that # reports an error, it's just discarded. # # asyncio, selector mode: # - ignores EWOULDBLOCK, EINTR, ECONNABORTED # - on EMFILE, ENFILE, ENOBUFS, ENOMEM, logs an error and then disables the # listening loop for 1 second # - everything else raises, but then the event loop just logs and ignores it # (selector_events.py: BaseSelectorEventLoop._accept_connection) # # # What should we do? # ------------------ # # When accept() returns an error, we can either ignore it or raise it. # # We have a long list of errors that should be ignored, and a long list of # errors that should be raised. The big question is what to do with an error # that isn't on either list. On Linux apparently you can get nearly arbitrary # errors from accept() and they should be ignored, because it just indicates a # socket that crashed before it began, and there isn't really anything to be # done about this, plus on other platforms you may not get any indication at # all, so programs have to tolerate not getting any indication too. OTOH if we # get an unexpected error then it could indicate something arbitrarily bad -- # after all, it's unexpected. # # Given that we know that other libraries seem to be getting along fine with a # fairly minimal list of errors to ignore, I think we'll be OK if we write # down that list and then raise on everything else. # # The other question is what to do about the capacity problem errors: EMFILE, # ENFILE, ENOBUFS, ENOMEM. Just flat out ignoring these is clearly not optimal # -- at the very least you want to log them, and probably you want to take # some remedial action. And if we ignore them then it prevents higher levels # from doing anything clever with them. So we raise them. _ignorable_accept_errno_names = [ # Linux can do this when the a connection is denied by the firewall "EPERM", # BSDs with an early close/reset "ECONNABORTED", # All the other miscellany noted above -- may not happen in practice, but # whatever. "EPROTO", "ENETDOWN", "ENOPROTOOPT", "EHOSTDOWN", "ENONET", "EHOSTUNREACH", "EOPNOTSUPP", "ENETUNREACH", "ENOSR", "ESOCKTNOSUPPORT", "EPROTONOSUPPORT", "ETIMEDOUT", "ECONNRESET", ] # Not all errnos are defined on all platforms _ignorable_accept_errnos = set() for name in _ignorable_accept_errno_names: try: _ignorable_accept_errnos.add(getattr(errno, name)) except AttributeError: pass class SocketListener(Listener[SocketStream], metaclass=Final): """A :class:`~trio.abc.Listener` that uses a listening socket to accept incoming connections as :class:`SocketStream` objects. Args: socket: The Trio socket object to wrap. Must have type ``SOCK_STREAM``, and be listening. Note that the :class:`SocketListener` "takes ownership" of the given socket; closing the :class:`SocketListener` will also close the socket. .. attribute:: socket The Trio socket object that this stream wraps. """ def __init__(self, socket): if not isinstance(socket, tsocket.SocketType): raise TypeError("SocketListener requires a Trio socket object") if socket.type != tsocket.SOCK_STREAM: raise ValueError("SocketListener requires a SOCK_STREAM socket") try: listening = socket.getsockopt(tsocket.SOL_SOCKET, tsocket.SO_ACCEPTCONN) except OSError: # SO_ACCEPTCONN fails on macOS; we just have to trust the user. pass else: if not listening: raise ValueError("SocketListener requires a listening socket") self.socket = socket async def accept(self): """Accept an incoming connection. Returns: :class:`SocketStream` Raises: OSError: if the underlying call to ``accept`` raises an unexpected error. ClosedResourceError: if you already closed the socket. This method handles routine errors like ``ECONNABORTED``, but passes other errors on to its caller. In particular, it does *not* make any special effort to handle resource exhaustion errors like ``EMFILE``, ``ENFILE``, ``ENOBUFS``, ``ENOMEM``. """ while True: try: sock, _ = await self.socket.accept() except OSError as exc: if exc.errno in _closed_stream_errnos: raise trio.ClosedResourceError if exc.errno not in _ignorable_accept_errnos: raise else: return SocketStream(sock) async def aclose(self): """Close this listener and its underlying socket.""" self.socket.close() await trio.lowlevel.checkpoint()