Python errno 模块，ENOBUFS 实例源码

def writeSomeData(self, data):
        """Connection.writeSomeData(data) -> #of bytes written | CONNECTION_LOST
        This writes as much data as possible to the socket and returns either
        the number of bytes read (which is positive) or a connection error code
        (which is negative)
        """
        try:
            # Limit length of buffer to try to send, because some OSes are too
            # stupid to do so themselves (ahem windows)
            return self.socket.send(buffer(data, 0, self.SEND_LIMIT))
        except socket.error, se:
            if se.args[0] == EINTR:
                return self.writeSomeData(data)
            elif se.args[0] in (EWOULDBLOCK, ENOBUFS):
                return 0
            else:
                return main.CONNECTION_LOST

def test_socket_busy_on_accept(self, monkeypatch):
        accept_responses = [
            lambda: MockUtils.raise_(socket.error(errno.ENOBUFS, "fatal error message")),
            lambda: MockUtils.raise_(socket.error(errno.EAGAIN, "busy, try accept again")),
            lambda: MockUtils.raise_(socket.error(errno.EPERM, "rejected, but try accept again"))
        ]
        if sys.platform != 'win32':
            monkeypatch.setattr(os, 'fork', lambda: 0)
        monkeypatch.setattr(socket, 'socket', lambda af, st: self.socket)
        self.socket.accept = lambda: accept_responses.pop()()
        try:
            RLXPort.listen(('localhost', 7000))
            assert False
        except Exception as e:
            traceback.format_exc()
            assert str(e) == '[Errno %d] fatal error message' % errno.ENOBUFS

def writeSomeData(self, data):
        """Connection.writeSomeData(data) -> #of bytes written | CONNECTION_LOST
        This writes as much data as possible to the socket and returns either
        the number of bytes read (which is positive) or a connection error code
        (which is negative)
        """
        try:
            # Limit length of buffer to try to send, because some OSes are too
            # stupid to do so themselves (ahem windows)
            return self.socket.send(buffer(data, 0, self.SEND_LIMIT))
        except socket.error, se:
            if se.args[0] == EINTR:
                return self.writeSomeData(data)
            elif se.args[0] in (EWOULDBLOCK, ENOBUFS):
                return 0
            else:
                return main.CONNECTION_LOST

def _testCongestion(self):
        # test the behavior in case of congestion
        self.data = b'fill'
        self.cli.setblocking(False)
        try:
            # try to lower the receiver's socket buffer size
            self.cli.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 16384)
        except OSError:
            pass
        with self.assertRaises(OSError) as cm:
            try:
                # fill the receiver's socket buffer
                while True:
                    self.cli.sendto(self.data, 0, (HOST, self.port))
            finally:
                # signal the receiver we're done
                self.evt.set()
        # sendto() should have failed with ENOBUFS
        self.assertEqual(cm.exception.errno, errno.ENOBUFS)
        # and we should have received a congestion notification through poll
        r, w, x = select.select([self.serv], [], [], 3.0)
        self.assertIn(self.serv, r)

def doWrite(self, sendfd=sendfd):
        """
        Transmit as many queued pending file descriptors as we can.
        """
        while self.outgoingSocketQueue:
            skt, desc = self.outgoingSocketQueue.pop(0)
            try:
                sendfd(self.outSocket.fileno(), skt.fileno(), desc)
            except SocketError, se:
                if se.errno in (EAGAIN, ENOBUFS):
                    self.outgoingSocketQueue.insert(0, (skt, desc))
                    return
                raise

            # Ready to close this socket; wait until it is acknowledged.
            self.pendingCloseSocketQueue.append(skt)

        if not self.outgoingSocketQueue:
            self.stopWriting()

def _testCongestion(self):
        # test the behavior in case of congestion
        self.data = b'fill'
        self.cli.setblocking(False)
        try:
            # try to lower the receiver's socket buffer size
            self.cli.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 16384)
        except OSError:
            pass
        with self.assertRaises(OSError) as cm:
            try:
                # fill the receiver's socket buffer
                while True:
                    self.cli.sendto(self.data, 0, (HOST, self.port))
            finally:
                # signal the receiver we're done
                self.evt.set()
        # sendto() should have failed with ENOBUFS
        self.assertEqual(cm.exception.errno, errno.ENOBUFS)
        # and we should have received a congestion notification through poll
        r, w, x = select.select([self.serv], [], [], 3.0)
        self.assertIn(self.serv, r)

def writeSomeData(self, data):
        """
        Write as much as possible of the given data to this TCP connection.

        This sends up to C{self.SEND_LIMIT} bytes from C{data}.  If the
        connection is lost, an exception is returned.  Otherwise, the number
        of bytes successfully written is returned.
        """
        # Limit length of buffer to try to send, because some OSes are too
        # stupid to do so themselves (ahem windows)
        limitedData = lazyByteSlice(data, 0, self.SEND_LIMIT)

        try:
            return untilConcludes(self.socket.send, limitedData)
        except socket.error as se:
            if se.args[0] in (EWOULDBLOCK, ENOBUFS):
                return 0
            else:
                return main.CONNECTION_LOST

def _testCongestion(self):
        # test the behavior in case of congestion
        self.data = b'fill'
        self.cli.setblocking(False)
        try:
            # try to lower the receiver's socket buffer size
            self.cli.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 16384)
        except OSError:
            pass
        with self.assertRaises(OSError) as cm:
            try:
                # fill the receiver's socket buffer
                while True:
                    self.cli.sendto(self.data, 0, (HOST, self.port))
            finally:
                # signal the receiver we're done
                self.evt.set()
        # sendto() should have failed with ENOBUFS
        self.assertEqual(cm.exception.errno, errno.ENOBUFS)
        # and we should have received a congestion notification through poll
        r, w, x = select.select([self.serv], [], [], 3.0)
        self.assertIn(self.serv, r)

def _accept_connection(self, protocol_factory, sock,
                           sslcontext=None, server=None):
        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):
            pass  # False alarm.
        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': sock,
                })
                self.remove_reader(sock.fileno())
                self.call_later(constants.ACCEPT_RETRY_DELAY,
                                self._start_serving,
                                protocol_factory, sock, sslcontext, server)
            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)
            self.create_task(accept)

def handle_accept_socket_exeption(cls, error):
        if error.errno in (errno.EWOULDBLOCK, errno.EAGAIN):
            # Try again
            return True  # continue accept loop
        elif error.errno == errno.EPERM:
            # Netfilter on Linux may have rejected the
            # connection, but we get told to try to accept()
            # anyway.
            return True  # continue accept loop
        elif error.errno in (errno.EMFILE, errno.ENOBUFS, errno.ENFILE,
                             errno.ENOMEM, errno.ECONNABORTED):
            # Linux gives EMFILE when a process is not allowed to
            # allocate any more file descriptors.  *BSD and Win32
            # give (WSA)ENOBUFS.  Linux can also give ENFILE if the
            # system is out of inodes, or ENOMEM if there is
            # insufficient memory to allocate a new dentry.
            # ECONNABORTED is documented as possible on all
            # relevant platforms (Linux, Windows, macOS, and the
            # BSDs) but occurs only on the BSDs.  It occurs when a
            # client sends a FIN or RST after the server sends a
            # SYN|ACK but before application code calls accept(2).
            # On Linux, calling accept(2) on such a listener
            # returns a connection that fails as though the it were
            # terminated after being fully established.  This
            # appears to be an implementation choice (see
            # inet_accept in inet/ipv4/af_inet.c).  On macOS X,
            # such a listener is not considered readable, so
            # accept(2) will never be called.  Calling accept(2) on
            # such a listener, however, does not return at all.
            log.error("Could not accept new connection (%s)" % error.strerror)
        return False  # break accept loop

def _accept_connection(self, protocol_factory, sock,
                           sslcontext=None, server=None, backlog=100):
        # 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': sock,
                    })
                    self._remove_reader(sock.fileno())
                    self.call_later(constants.ACCEPT_RETRY_DELAY,
                                    self._start_serving,
                                    protocol_factory, sock, sslcontext, server,
                                    backlog)
                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)
                self.create_task(accept)

def doRead(self, recvmsg=recv1msg):
        """
        Receive a status / health message and record it.
        """
        try:
            data, _ignore_flags, _ignore_ancillary = recvmsg(
                self.outSocket.fileno()
            )
        except SocketError, se:
            if se.errno not in (EAGAIN, ENOBUFS):
                raise
        else:
            closeCount = self.dispatcher.statusMessage(self, data)
            for ignored in xrange(closeCount):
                self.pendingCloseSocketQueue.pop(0).close()

def doWrite(self):
        """
        Write some data.
        """
        while self.statusQueue:
            msg = self.statusQueue.pop(0)
            try:
                send1msg(self.fd, msg, 0)
            except SocketError, se:
                if se.errno in (EAGAIN, ENOBUFS):
                    self.statusQueue.insert(0, msg)
                    return
                raise
        self.stopWriting()

def _accept_connection(self, protocol_factory, sock,
                           sslcontext=None, server=None):
        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):
            pass  # False alarm.
        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': sock,
                })
                self.remove_reader(sock.fileno())
                self.call_later(constants.ACCEPT_RETRY_DELAY,
                                self._start_serving,
                                protocol_factory, sock, sslcontext, server)
            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)
            self.create_task(accept)

def _accept_connection(self, protocol_factory, sock,
                           sslcontext=None, server=None):
        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):
            pass  # False alarm.
        except OSError as exc:
            # There's nowhere to send the error, so just log it.
            # TODO: Someone will want an error handler for this.
            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': sock,
                })
                self.remove_reader(sock.fileno())
                self.call_later(constants.ACCEPT_RETRY_DELAY,
                                self._start_serving,
                                protocol_factory, sock, sslcontext, server)
            else:
                raise  # The event loop will catch, log and ignore it.
        else:
            if sslcontext:
                self._make_ssl_transport(
                    conn, protocol_factory(), sslcontext, None,
                    server_side=True, extra={'peername': addr}, server=server)
            else:
                self._make_socket_transport(
                    conn, protocol_factory(), extra={'peername': addr},
                    server=server)
        # It's now up to the protocol to handle the connection.

def _sendCloseAlert(self):
        # Okay, *THIS* is a bit complicated.

        # Basically, the issue is, OpenSSL seems to not actually return
        # errors from SSL_shutdown. Therefore, the only way to
        # determine if the close notification has been sent is by 
        # SSL_shutdown returning "done". However, it will not claim it's
        # done until it's both sent *and* received a shutdown notification.

        # I don't actually want to wait for a received shutdown
        # notification, though, so, I have to set RECEIVED_SHUTDOWN
        # before calling shutdown. Then, it'll return True once it's
        # *SENT* the shutdown.

        # However, RECEIVED_SHUTDOWN can't be left set, because then
        # reads will fail, breaking half close.

        # Also, since shutdown doesn't report errors, an empty write call is
        # done first, to try to detect if the connection has gone away.
        # (*NOT* an SSL_write call, because that fails once you've called
        # shutdown)
        try:
            os.write(self.socket.fileno(), '')
        except OSError, se:
            if se.args[0] in (EINTR, EWOULDBLOCK, ENOBUFS):
                return 0
            # Write error, socket gone
            return main.CONNECTION_LOST

        try:
            if hasattr(self.socket, 'set_shutdown'):
                laststate = self.socket.get_shutdown()
                self.socket.set_shutdown(laststate | SSL.RECEIVED_SHUTDOWN)
                done = self.socket.shutdown()
                if not (laststate & SSL.RECEIVED_SHUTDOWN):
                    self.socket.set_shutdown(SSL.SENT_SHUTDOWN)
            else:
                #warnings.warn("SSL connection shutdown possibly unreliable, "
                #              "please upgrade to ver 0.XX", category=UserWarning)
                self.socket.shutdown()
                done = True
        except SSL.Error, e:
            return e

        if done:
            self.stopWriting()
            # Note that this is tested for by identity below.
            return main.CONNECTION_DONE
        else:
            self.startWriting()
            return None

def test_SocketListener_accept_errors():
    class FakeSocket(tsocket.SocketType):
        def __init__(self, events):
            self._events = iter(events)

        type = tsocket.SOCK_STREAM

        # Fool the check for SO_ACCEPTCONN in SocketListener.__init__
        def getsockopt(self, level, opt):
            return True

        def setsockopt(self, level, opt, value):
            pass

        # Fool the check for connection in SocketStream.__init__
        def getpeername(self):
            pass

        async def accept(self):
            await _core.checkpoint()
            event = next(self._events)
            if isinstance(event, BaseException):
                raise event
            else:
                return event, None

    fake_server_sock = FakeSocket([])

    fake_listen_sock = FakeSocket(
        [
            OSError(errno.ECONNABORTED, "Connection aborted"),
            OSError(errno.EPERM, "Permission denied"),
            OSError(errno.EPROTO, "Bad protocol"),
            fake_server_sock,
            OSError(errno.EMFILE, "Out of file descriptors"),
            OSError(errno.EFAULT, "attempt to write to read-only memory"),
            OSError(errno.ENOBUFS, "out of buffers"),
            fake_server_sock,
        ]
    )

    l = SocketListener(fake_listen_sock)

    with assert_checkpoints():
        s = await l.accept()
        assert s.socket is fake_server_sock

    for code in [errno.EMFILE, errno.EFAULT, errno.ENOBUFS]:
        with assert_checkpoints():
            with pytest.raises(OSError) as excinfo:
                await l.accept()
            assert excinfo.value.errno == code

    with assert_checkpoints():
        s = await l.accept()
        assert s.socket is fake_server_sock

def _sendCloseAlert(self):
        # Okay, *THIS* is a bit complicated.

        # Basically, the issue is, OpenSSL seems to not actually return
        # errors from SSL_shutdown. Therefore, the only way to
        # determine if the close notification has been sent is by 
        # SSL_shutdown returning "done". However, it will not claim it's
        # done until it's both sent *and* received a shutdown notification.

        # I don't actually want to wait for a received shutdown
        # notification, though, so, I have to set RECEIVED_SHUTDOWN
        # before calling shutdown. Then, it'll return True once it's
        # *SENT* the shutdown.

        # However, RECEIVED_SHUTDOWN can't be left set, because then
        # reads will fail, breaking half close.

        # Also, since shutdown doesn't report errors, an empty write call is
        # done first, to try to detect if the connection has gone away.
        # (*NOT* an SSL_write call, because that fails once you've called
        # shutdown)
        try:
            os.write(self.socket.fileno(), '')
        except OSError, se:
            if se.args[0] in (EINTR, EWOULDBLOCK, ENOBUFS):
                return 0
            # Write error, socket gone
            return main.CONNECTION_LOST

        try:
            if hasattr(self.socket, 'set_shutdown'):
                laststate = self.socket.get_shutdown()
                self.socket.set_shutdown(laststate | SSL.RECEIVED_SHUTDOWN)
                done = self.socket.shutdown()
                if not (laststate & SSL.RECEIVED_SHUTDOWN):
                    self.socket.set_shutdown(SSL.SENT_SHUTDOWN)
            else:
                #warnings.warn("SSL connection shutdown possibly unreliable, "
                #              "please upgrade to ver 0.XX", category=UserWarning)
                self.socket.shutdown()
                done = True
        except SSL.Error, e:
            return e

        if done:
            self.stopWriting()
            # Note that this is tested for by identity below.
            return main.CONNECTION_DONE
        else:
            self.startWriting()
            return None

def writeSomeData(self, data):
        """
        Send as much of C{data} as possible.  Also send any pending file
        descriptors.
        """
        # Make it a programming error to send more file descriptors than you
        # send regular bytes.  Otherwise, due to the limitation mentioned
        # below, we could end up with file descriptors left, but no bytes to
        # send with them, therefore no way to send those file descriptors.
        if len(self._sendmsgQueue) > len(data):
            return error.FileDescriptorOverrun()

        # If there are file descriptors to send, try sending them first, using
        # a little bit of data from the stream-oriented write buffer too.  It
        # is not possible to send a file descriptor without sending some
        # regular data.
        index = 0
        try:
            while index < len(self._sendmsgQueue):
                fd = self._sendmsgQueue[index]
                try:
                    untilConcludes(
                        sendmsg.sendmsg, self.socket, data[index:index+1],
                        _ancillaryDescriptor(fd))
                except socket.error as se:
                    if se.args[0] in (EWOULDBLOCK, ENOBUFS):
                        return index
                    else:
                        return main.CONNECTION_LOST
                else:
                    index += 1
        finally:
            del self._sendmsgQueue[:index]

        # Hand the remaining data to the base implementation.  Avoid slicing in
        # favor of a buffer, in case that happens to be any faster.
        limitedData = lazyByteSlice(data, index)
        result = self._writeSomeDataBase.writeSomeData(self, limitedData)
        try:
            return index + result
        except TypeError:
            return result