event loop using a selector and related classes a selector is a notifywhenready multiplexer for a subclass which also includes support for signal handling see the unixevents submodule fallback to send test if the selector is monitoring event events for the file descriptor fd selector event loop see events eventloop for api specification a selfsocket really this may be called from a different thread possibly after closeselfpipe 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 this method is only called once for each event loop tick where the listening socket has triggered an eventread there may be multiple connections waiting for an accept so it is called in a loop see https bugs python orgissue27906 for more details early exit because the socket accept buffer is empty there s nowhere to send the error so just log it 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 it s now up to the protocol to handle the connection this code matches selectors fileobjtofd function remove a writer callback if self isclosed return false key self selector getmap getfd if key is none return false mask reader writer key events key data remove both writer and connector mask selectors eventwrite if not mask self selector unregisterfd else self selector modifyfd mask reader none if writer is not none writer cancel return true else return false def addreaderself fd callback args remove a reader callback self ensurefdnotransportfd return self removereaderfd def addwriterself fd callback args remove a writer callback self ensurefdnotransportfd return self removewriterfd async def sockrecvself sock n baseevents checksslsocketsock if self debug and sock gettimeout 0 raise valueerrorthe socket must be nonblocking try return sock recvn except blockingioerror interruptederror pass fut self createfuture fd sock fileno self ensurefdnotransportfd handle self addreaderfd self sockrecv fut sock n fut adddonecallback functools partialself sockreaddone fd handlehandle return await fut def sockreaddoneself fd fut handlenone if handle is none or not handle cancelled self removereaderfd def sockrecvself fut sock n sockrecv can add itself as an io callback if the operation can t be done immediately don t use it directly call sockrecv if fut done return try data sock recvn except blockingioerror interruptederror return try again next time except systemexit keyboardinterrupt raise except baseexception as exc fut setexceptionexc else fut setresultdata async def sockrecvintoself sock buf baseevents checksslsocketsock if self debug and sock gettimeout 0 raise valueerrorthe socket must be nonblocking try return sock recvintobuf except blockingioerror interruptederror pass fut self createfuture fd sock fileno self ensurefdnotransportfd handle self addreaderfd self sockrecvinto fut sock buf fut adddonecallback functools partialself sockreaddone fd handlehandle return await fut def sockrecvintoself fut sock buf sockrecvinto can add itself as an io callback if the operation can t be done immediately don t use it directly call sockrecvinto if fut done return try nbytes sock recvintobuf except blockingioerror interruptederror return try again next time except systemexit keyboardinterrupt raise except baseexception as exc fut setexceptionexc else fut setresultnbytes async def sockrecvfromself sock bufsize baseevents checksslsocketsock if self debug and sock gettimeout 0 raise valueerrorthe socket must be nonblocking try return sock recvfrombufsize except blockingioerror interruptederror pass fut self createfuture fd sock fileno self ensurefdnotransportfd handle self addreaderfd self sockrecvfrom fut sock bufsize fut adddonecallback functools partialself sockreaddone fd handlehandle return await fut def sockrecvfromself fut sock bufsize sockrecvfrom can add itself as an io callback if the operation can t be done immediately don t use it directly call sockrecvfrom if fut done return try result sock recvfrombufsize except blockingioerror interruptederror return try again next time except systemexit keyboardinterrupt raise except baseexception as exc fut setexceptionexc else fut setresultresult async def sockrecvfromintoself sock buf nbytes0 baseevents checksslsocketsock if self debug and sock gettimeout 0 raise valueerrorthe socket must be nonblocking if not nbytes nbytes lenbuf try return sock recvfromintobuf nbytes except blockingioerror interruptederror pass fut self createfuture fd sock fileno self ensurefdnotransportfd handle self addreaderfd self sockrecvfrominto fut sock buf nbytes fut adddonecallback functools partialself sockreaddone fd handlehandle return await fut def sockrecvfromintoself fut sock buf bufsize sockrecvinto can add itself as an io callback if the operation can t be done immediately don t use it directly call sockrecvinto if fut done return try result sock recvfromintobuf bufsize except blockingioerror interruptederror return try again next time except systemexit keyboardinterrupt raise except baseexception as exc fut setexceptionexc else fut setresultresult async def socksendallself sock data baseevents checksslsocketsock if self debug and sock gettimeout 0 raise valueerrorthe socket must be nonblocking try n sock senddata except blockingioerror interruptederror n 0 if n lendata all data sent return fut self createfuture fd sock fileno self ensurefdnotransportfd use a trick with a list in closure to store a mutable state handle self addwriterfd self socksendall fut sock memoryviewdata n fut adddonecallback functools partialself sockwritedone fd handlehandle return await fut def socksendallself fut sock view pos if fut done future cancellation can be scheduled on previous loop iteration return start pos0 try n sock sendviewstart except blockingioerror interruptederror return except systemexit keyboardinterrupt raise except baseexception as exc fut setexceptionexc return start n if start lenview fut setresultnone else pos0 start async def socksendtoself sock data address baseevents checksslsocketsock if self debug and sock gettimeout 0 raise valueerrorthe socket must be nonblocking try return sock sendtodata address except blockingioerror interruptederror pass fut self createfuture fd sock fileno self ensurefdnotransportfd use a trick with a list in closure to store a mutable state handle self addwriterfd self socksendto fut sock data address fut adddonecallback functools partialself sockwritedone fd handlehandle return await fut def socksendtoself fut sock data address if fut done future cancellation can be scheduled on previous loop iteration return try n sock sendtodata 0 address except blockingioerror interruptederror return except systemexit keyboardinterrupt raise except baseexception as exc fut setexceptionexc else fut setresultn async def sockconnectself sock address baseevents checksslsocketsock if self debug and sock gettimeout 0 raise valueerrorthe socket must be nonblocking if sock family socket afinet or baseevents hasipv6 and sock family socket afinet6 resolved await self ensureresolved address familysock family typesock type protosock proto loopself address resolved0 fut self createfuture self sockconnectfut sock address try return await fut finally needed to break cycles when an exception occurs fut none def sockconnectself fut sock address fd sock fileno try sock connectaddress 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 ensurefdnotransportfd handle self addwriter fd self sockconnectcb fut sock address fut adddonecallback functools partialself sockwritedone fd handlehandle except systemexit keyboardinterrupt raise except baseexception as exc fut setexceptionexc else fut setresultnone finally fut none def sockwritedoneself fd fut handlenone if handle is none or not handle cancelled self removewriterfd def sockconnectcbself fut sock address if fut done return try err sock getsockoptsocket solsocket socket soerror if err 0 jump to any except clause below raise oserrorerr 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 setexceptionexc else fut setresultnone finally fut none async def sockacceptself sock baseevents checksslsocketsock if self debug and sock gettimeout 0 raise valueerrorthe socket must be nonblocking fut self createfuture self sockacceptfut sock return await fut def sockacceptself fut sock fd sock fileno try conn address sock accept conn setblockingfalse except blockingioerror interruptederror self ensurefdnotransportfd handle self addreaderfd self sockaccept fut sock fut adddonecallback functools partialself sockreaddone fd handlehandle except systemexit keyboardinterrupt raise except baseexception as exc fut setexceptionexc else fut setresultconn address async def sendfilenativeself transp file offset count del self transportstransp sockfd resumereading transp isreading transp pausereading await transp makeemptywaiter try return await self socksendfiletransp sock file offset count fallbackfalse finally transp resetemptywaiter if resumereading transp resumereading self transportstransp sockfd transp def processeventsself eventlist for key mask in eventlist fileobj reader writer key fileobj key data if mask selectors eventread and reader is not none if reader cancelled self removereaderfileobj else self addcallbackreader if mask selectors eventwrite and writer is not none if writer cancelled self removewriterfileobj else self addcallbackwriter def stopservingself sock self removereadersock fileno sock close class selectortransporttransports flowcontrolmixin transports transport maxsize 256 1024 buffer size passed to recv 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 initself loop sock protocol extranone servernone super initextra loop self extra socket trsock transportsocketsock 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 sockfd sock fileno self protocolconnected false self setprotocolprotocol self server server self buffer collections deque self connlost 0 set when call to connectionlost scheduled self closing false set when close called self paused false set when pausereading called if self server is not none self server attach loop transportsself sockfd self def reprself info self class name if self sock is none info append closed elif self closing info append closing info appendf fdself sockfd test if the transport was closed if self loop is not none and not self loop isclosed polling testselectoreventself loop selector self sockfd selectors eventread if polling info append readpolling else info append readidle polling testselectoreventself loop selector self sockfd selectors eventwrite if polling state polling else state idle bufsize self getwritebuffersize info appendf writestate bufsizebufsize return format joininfo def abortself self forceclosenone def setprotocolself protocol self protocol protocol self protocolconnected true def getprotocolself return self protocol def isclosingself return self closing def isreadingself return not self isclosing and not self paused def pausereadingself if not self isreading return self paused true self loop removereaderself sockfd if self loop getdebug logger debugr pauses reading self def resumereadingself if self closing or not self paused return self paused false self addreaderself sockfd self readready if self loop getdebug logger debugr resumes reading self def closeself if self closing return self closing true self loop removereaderself sockfd if not self buffer self connlost 1 self loop removewriterself sockfd self loop callsoonself callconnectionlost none def delself warnwarnings warn if self sock is not none warnfunclosed transport self r resourcewarning sourceself self sock close def fatalerrorself exc message fatal error on transport should be called from exception handler only if isinstanceexc oserror if self loop getdebug logger debugr s self message excinfotrue else self loop callexceptionhandler message message exception exc transport self protocol self protocol self forcecloseexc def forcecloseself exc if self connlost return if self buffer self buffer clear self loop removewriterself sockfd if not self closing self closing true self loop removereaderself sockfd self connlost 1 self loop callsoonself callconnectionlost exc def callconnectionlostself exc try if self protocolconnected self protocol connectionlostexc 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 getwritebuffersizeself return summaplen self buffer def addreaderself fd callback args if not self isreading return self loop addreaderfd callback args class selectorsockettransportselectortransport starttlscompatible true sendfilecompatible constants sendfilemode trynative def initself loop sock protocol waiternone extranone servernone self readreadycb none super initloop sock protocol extra server self eof false self emptywaiter none if hassendmsg self writeready self writesendmsg else self writeready self writesend disable the nagle algorithm small writes will be sent without waiting for the tcp ack this generally decreases the latency in some cases significantly baseevents setnodelayself sock self loop callsoonself protocol connectionmade self only start reading when connectionmade has been called self loop callsoonself addreader self sockfd self readready if waiter is not none only wake up the waiter when connectionmade has been called self loop callsoonfutures setresultunlesscancelled waiter none def setprotocolself protocol if isinstanceprotocol protocols bufferedprotocol self readreadycb self readreadygetbuffer else self readreadycb self readreadydatareceived super setprotocolprotocol def readreadyself self readreadycb def readreadygetbufferself if self connlost return try buf self protocol getbuffer1 if not lenbuf raise runtimeerror getbuffer returned an empty buffer except systemexit keyboardinterrupt raise except baseexception as exc self fatalerror exc fatal error protocol getbuffer call failed return try nbytes self sock recvintobuf except blockingioerror interruptederror return except systemexit keyboardinterrupt raise except baseexception as exc self fatalerrorexc fatal read error on socket transport return if not nbytes self readreadyoneof return try self protocol bufferupdatednbytes except systemexit keyboardinterrupt raise except baseexception as exc self fatalerror exc fatal error protocol bufferupdated call failed def readreadydatareceivedself if self connlost return try data self sock recvself maxsize except blockingioerror interruptederror return except systemexit keyboardinterrupt raise except baseexception as exc self fatalerrorexc fatal read error on socket transport return if not data self readreadyoneof return try self protocol datareceiveddata except systemexit keyboardinterrupt raise except baseexception as exc self fatalerror exc fatal error protocol datareceived call failed def readreadyoneofself if self loop getdebug logger debugr received eof self try keepopen self protocol eofreceived except systemexit keyboardinterrupt raise except baseexception as exc self fatalerror exc fatal error protocol eofreceived call failed return if keepopen 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 removereaderself sockfd else self close def writeself data if not isinstancedata bytes bytearray memoryview raise typeerrorf data argument must be a byteslike object f not typedata name r if self eof raise runtimeerror cannot call write after writeeof if self emptywaiter is not none raise runtimeerror unable to write sendfile is in progress if not data return if self connlost if self connlost constants logthresholdforconnlostwrites logger warning socket send raised exception self connlost 1 return if not self buffer optimization try to send now try n self sock senddata except blockingioerror interruptederror pass except systemexit keyboardinterrupt raise except baseexception as exc self fatalerrorexc fatal write error on socket transport return else data memoryviewdatan if not data return not all was written register write handler self loop addwriterself sockfd self writeready add it to the buffer self buffer appenddata self maybepauseprotocol def getsendmsgbufferself return itertools isliceself buffer sciovmax def writesendmsgself assert self buffer data should not be empty if self connlost return try nbytes self sock sendmsgself getsendmsgbuffer self adjustleftoverbuffernbytes except blockingioerror interruptederror pass except systemexit keyboardinterrupt raise except baseexception as exc self loop removewriterself sockfd self buffer clear self fatalerrorexc fatal write error on socket transport if self emptywaiter is not none self emptywaiter setexceptionexc else self mayberesumeprotocol may append to buffer if not self buffer self loop removewriterself sockfd if self emptywaiter is not none self emptywaiter setresultnone if self closing self callconnectionlostnone elif self eof self sock shutdownsocket shutwr def adjustleftoverbufferself nbytes int none buffer self buffer while nbytes b buffer popleft blen lenb if blen nbytes nbytes blen else buffer appendleftbnbytes break def writesendself assert self buffer data should not be empty if self connlost return try buffer self buffer popleft n self sock sendbuffer if n lenbuffer not all data was written self buffer appendleftbuffern except blockingioerror interruptederror pass except systemexit keyboardinterrupt raise except baseexception as exc self loop removewriterself sockfd self buffer clear self fatalerrorexc fatal write error on socket transport if self emptywaiter is not none self emptywaiter setexceptionexc else self mayberesumeprotocol may append to buffer if not self buffer self loop removewriterself sockfd if self emptywaiter is not none self emptywaiter setresultnone if self closing self callconnectionlostnone elif self eof self sock shutdownsocket shutwr def writeeofself if self closing or self eof return self eof true if not self buffer self sock shutdownsocket shutwr def writelinesself listofdata if self eof raise runtimeerror cannot call writelines after writeeof if self emptywaiter is not none raise runtimeerror unable to writelines sendfile is in progress if not listofdata return self buffer extendmemoryviewdata for data in listofdata self writeready if the entire buffer couldn t be written register a write handler if self buffer self loop addwriterself sockfd self writeready def canwriteeofself return true def callconnectionlostself exc super callconnectionlostexc if self emptywaiter is not none self emptywaiter setexception connectionerrorconnection is closed by peer def makeemptywaiterself if self emptywaiter is not none raise runtimeerrorempty waiter is already set self emptywaiter self loop createfuture if not self buffer self emptywaiter setresultnone return self emptywaiter def resetemptywaiterself self emptywaiter none def closeself self readreadycb none self writeready none super close class selectordatagramtransportselectortransport transports datagramtransport bufferfactory collections deque def initself loop sock protocol addressnone waiternone extranone super initloop sock protocol extra self address address self buffersize 0 self loop callsoonself protocol connectionmade self only start reading when connectionmade has been called self loop callsoonself addreader self sockfd self readready if waiter is not none only wake up the waiter when connectionmade has been called self loop callsoonfutures setresultunlesscancelled waiter none def getwritebuffersizeself return self buffersize def readreadyself if self connlost return try data addr self sock recvfromself maxsize except blockingioerror interruptederror pass except oserror as exc self protocol errorreceivedexc except systemexit keyboardinterrupt raise except baseexception as exc self fatalerrorexc fatal read error on datagram transport else self protocol datagramreceiveddata addr def sendtoself data addrnone if not isinstancedata bytes bytearray memoryview raise typeerrorf data argument must be a byteslike object f not typedata 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 connlost and self address if self connlost constants logthresholdforconnlostwrites logger warning socket send raised exception self connlost 1 return if not self buffer attempt to send it right away first try if self extra peername self sock senddata else self sock sendtodata addr return except blockingioerror interruptederror self loop addwriterself sockfd self sendtoready except oserror as exc self protocol errorreceivedexc return except systemexit keyboardinterrupt raise except baseexception as exc self fatalerror exc fatal write error on datagram transport return ensure that what we buffer is immutable self buffer appendbytesdata addr self buffersize lendata self maybepauseprotocol def sendtoreadyself while self buffer data addr self buffer popleft self buffersize lendata try if self extra peername self sock senddata else self sock sendtodata addr except blockingioerror interruptederror self buffer appendleftdata addr try again later self buffersize lendata break except oserror as exc self protocol errorreceivedexc return except systemexit keyboardinterrupt raise except baseexception as exc self fatalerror exc fatal write error on datagram transport return self mayberesumeprotocol may append to buffer if not self buffer self loop removewriterself sockfd if self closing self callconnectionlostnone pragma no cover fallback to send test if the selector is monitoring event events for the file descriptor fd selector event loop see events eventloop for api specification a self socket really 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 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 early exit because the socket accept buffer is empty there s nowhere to send the error so just log it 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 the event loop will catch log and ignore it it s now up to the protocol to handle the connection this code matches selectors _fileobj_to_fd function remove a writer callback remove both writer and connector add a reader callback remove a reader callback add a writer callback remove a writer callback 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 _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 try again next time receive data from the socket the received data is written into buf a writable buffer the return value is the number of bytes written _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 try again next time receive a datagram from a datagram socket the return value is a tuple of bytes address representing the datagram received and the address it came from the maximum amount of data to be received at once is specified by nbytes _sock_recvfrom can add itself as an i o callback if the operation can t be done immediately don t use it directly call sock_recvfrom try again next time receive data from the socket the received data is written into buf a writable buffer the return value is a tuple of number of bytes written address _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 try again next time 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 all data sent use a trick with a list in closure to store a mutable state future cancellation can be scheduled on previous loop iteration 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 use a trick with a list in closure to store a mutable state future cancellation can be scheduled on previous loop iteration connect to a remote socket at address this method is a coroutine needed to break cycles when an exception occurs 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 jump to any except clause below socket is still registered the callback will be retried later 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 buffer size passed to recv 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 set when call to connection_lost scheduled set when close called set when pause_reading called test if the transport was closed should be called from exception handler only disable the nagle algorithm small writes will be sent without waiting for the tcp ack this generally decreases the latency in some cases significantly only start reading when connection_made has been called only wake up the waiter when connection_made has been called we re keeping the connection open so the protocol can write more but we still can t receive more so remove the reader callback optimization try to send now not all was written register write handler add it to the buffer may append to buffer not all data was written may append to buffer if the entire buffer couldn t be written register a write handler only start reading when connection_made has been called only wake up the waiter when connection_made has been called attempt to send it right away first ensure that what we buffer is immutable try again later may append to buffer
__all__ = 'BaseSelectorEventLoop',

import collections
import errno
import functools
import itertools
import os
import selectors
import socket
import warnings
import weakref
try:
    import ssl
except ImportError:  
    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

_HAS_SENDMSG = hasattr(socket.socket, 'sendmsg')

if _HAS_SENDMSG:
    try:
        SC_IOV_MAX = os.sysconf('SC_IOV_MAX')
    except OSError:
        
        _HAS_SENDMSG = False

def _test_selector_event(selector, fd, event):
    
    
    try:
        key = selector.get_key(fd)
    except KeyError:
        return False
    else:
        return bool(key.events & event)


class BaseSelectorEventLoop(base_events.BaseEventLoop):
    

    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):
        self._ensure_fd_no_transport(sock)
        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_shutdown_timeout=constants.SSL_SHUTDOWN_TIMEOUT,
    ):
        self._ensure_fd_no_transport(rawsock)
        ssl_protocol = sslproto.SSLProtocol(
            self, protocol, sslcontext, waiter,
            server_side, server_hostname,
            ssl_handshake_timeout=ssl_handshake_timeout,
            ssl_shutdown_timeout=ssl_shutdown_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):
        self._ensure_fd_no_transport(sock)
        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):
        
        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):
        
        
        
        
        
        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,
                       ssl_shutdown_timeout=constants.SSL_SHUTDOWN_TIMEOUT):
        self._add_reader(sock.fileno(), self._accept_connection,
                         protocol_factory, sock, sslcontext, server, backlog,
                         ssl_handshake_timeout, ssl_shutdown_timeout)

    def _accept_connection(
            self, protocol_factory, sock,
            sslcontext=None, server=None, backlog=100,
            ssl_handshake_timeout=constants.SSL_HANDSHAKE_TIMEOUT,
            ssl_shutdown_timeout=constants.SSL_SHUTDOWN_TIMEOUT):
        
        
        
        
        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):
                
                return None
            except OSError as exc:
                
                if exc.errno in (errno.EMFILE, errno.ENFILE,
                                 errno.ENOBUFS, errno.ENOMEM):
                    
                    
                    
                    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,
                                    ssl_shutdown_timeout)
                else:
                    raise  
            else:
                extra = {'peername': addr}
                accept = self._accept_connection2(
                    protocol_factory, conn, extra, sslcontext, server,
                    ssl_handshake_timeout, ssl_shutdown_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,
            ssl_shutdown_timeout=constants.SSL_SHUTDOWN_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,
                    ssl_shutdown_timeout=ssl_shutdown_timeout)
            else:
                transport = self._make_socket_transport(
                    conn, protocol, waiter=waiter, extra=extra,
                    server=server)

            try:
                await waiter
            except BaseException:
                transport.close()
                raise
                

        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):
                
                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)
        key = self._selector.get_map().get(fd)
        if key is None:
            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
        key = self._selector.get_map().get(fd)
        if key is None:
            return False
        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)
        key = self._selector.get_map().get(fd)
        if key is None:
            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):
        
        if self.is_closed():
            return False
        key = self._selector.get_map().get(fd)
        if key is None:
            return False
        mask, (reader, writer) = key.events, key.data
        
        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):
        
        self._ensure_fd_no_transport(fd)
        self._add_reader(fd, callback, *args)

    def remove_reader(self, fd):
        
        self._ensure_fd_no_transport(fd)
        return self._remove_reader(fd)

    def add_writer(self, fd, callback, *args):
        
        self._ensure_fd_no_transport(fd)
        self._add_writer(fd, callback, *args)

    def remove_writer(self, fd):
        
        self._ensure_fd_no_transport(fd)
        return self._remove_writer(fd)

    async def sock_recv(self, sock, n):
        
        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):
        
        
        if fut.done():
            return
        try:
            data = sock.recv(n)
        except (BlockingIOError, InterruptedError):
            return  
        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):
        
        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):
        
        
        
        if fut.done():
            return
        try:
            nbytes = sock.recv_into(buf)
        except (BlockingIOError, InterruptedError):
            return  
        except (SystemExit, KeyboardInterrupt):
            raise
        except BaseException as exc:
            fut.set_exception(exc)
        else:
            fut.set_result(nbytes)

    async def sock_recvfrom(self, sock, bufsize):
        
        base_events._check_ssl_socket(sock)
        if self._debug and sock.gettimeout() != 0:
            raise ValueError("the socket must be non-blocking")
        try:
            return sock.recvfrom(bufsize)
        except (BlockingIOError, InterruptedError):
            pass
        fut = self.create_future()
        fd = sock.fileno()
        self._ensure_fd_no_transport(fd)
        handle = self._add_reader(fd, self._sock_recvfrom, fut, sock, bufsize)
        fut.add_done_callback(
            functools.partial(self._sock_read_done, fd, handle=handle))
        return await fut

    def _sock_recvfrom(self, fut, sock, bufsize):
        
        
        
        if fut.done():
            return
        try:
            result = sock.recvfrom(bufsize)
        except (BlockingIOError, InterruptedError):
            return  
        except (SystemExit, KeyboardInterrupt):
            raise
        except BaseException as exc:
            fut.set_exception(exc)
        else:
            fut.set_result(result)

    async def sock_recvfrom_into(self, sock, buf, nbytes=0):
        
        base_events._check_ssl_socket(sock)
        if self._debug and sock.gettimeout() != 0:
            raise ValueError("the socket must be non-blocking")
        if not nbytes:
            nbytes = len(buf)

        try:
            return sock.recvfrom_into(buf, nbytes)
        except (BlockingIOError, InterruptedError):
            pass
        fut = self.create_future()
        fd = sock.fileno()
        self._ensure_fd_no_transport(fd)
        handle = self._add_reader(fd, self._sock_recvfrom_into, fut, sock, buf,
                                  nbytes)
        fut.add_done_callback(
            functools.partial(self._sock_read_done, fd, handle=handle))
        return await fut

    def _sock_recvfrom_into(self, fut, sock, buf, bufsize):
        
        
        
        if fut.done():
            return
        try:
            result = sock.recvfrom_into(buf, bufsize)
        except (BlockingIOError, InterruptedError):
            return  
        except (SystemExit, KeyboardInterrupt):
            raise
        except BaseException as exc:
            fut.set_exception(exc)
        else:
            fut.set_result(result)

    async def sock_sendall(self, sock, data):
        
        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):
            
            return

        fut = self.create_future()
        fd = sock.fileno()
        self._ensure_fd_no_transport(fd)
        
        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():
            
            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_sendto(self, sock, data, address):
        
        base_events._check_ssl_socket(sock)
        if self._debug and sock.gettimeout() != 0:
            raise ValueError("the socket must be non-blocking")
        try:
            return sock.sendto(data, address)
        except (BlockingIOError, InterruptedError):
            pass

        fut = self.create_future()
        fd = sock.fileno()
        self._ensure_fd_no_transport(fd)
        
        handle = self._add_writer(fd, self._sock_sendto, fut, sock, data,
                                  address)
        fut.add_done_callback(
            functools.partial(self._sock_write_done, fd, handle=handle))
        return await fut

    def _sock_sendto(self, fut, sock, data, address):
        if fut.done():
            
            return
        try:
            n = sock.sendto(data, 0, address)
        except (BlockingIOError, InterruptedError):
            return
        except (SystemExit, KeyboardInterrupt):
            raise
        except BaseException as exc:
            fut.set_exception(exc)
        else:
            fut.set_result(n)

    async def sock_connect(self, sock, address):
        
        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)
        try:
            return await fut
        finally:
            
            fut = None

    def _sock_connect(self, fut, sock, address):
        fd = sock.fileno()
        try:
            sock.connect(address)
        except (BlockingIOError, InterruptedError):
            
            
            
            
            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)
        finally:
            fut = 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:
                
                raise OSError(err, f'Connect call failed {address}')
        except (BlockingIOError, InterruptedError):
            
            pass
        except (SystemExit, KeyboardInterrupt):
            raise
        except BaseException as exc:
            fut.set_exception(exc)
        else:
            fut.set_result(None)
        finally:
            fut = None

    async def sock_accept(self, sock):
        
        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  

    
    
    
    _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 = collections.deque()
        self._conn_lost = 0  
        self._closing = False  
        self._paused = False  

        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}')
        
        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 is_reading(self):
        return not self.is_closing() and not self._paused

    def pause_reading(self):
        if not self.is_reading():
            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 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'):
        
        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 sum(map(len, self._buffer))

    def _add_reader(self, fd, callback, *args):
        if not self.is_reading():
            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._empty_waiter = None
        if _HAS_SENDMSG:
            self._write_ready = self._write_sendmsg
        else:
            self._write_ready = self._write_send
        
        
        
        base_events._set_nodelay(self._sock)

        self._loop.call_soon(self._protocol.connection_made, self)
        
        self._loop.call_soon(self._add_reader,
                             self._sock_fd, self._read_ready)
        if waiter is not None:
            
            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 _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:
            
            
            
            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:
            
            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 = memoryview(data)[n:]
                if not data:
                    return
            
            self._loop._add_writer(self._sock_fd, self._write_ready)

        
        self._buffer.append(data)
        self._maybe_pause_protocol()

    def _get_sendmsg_buffer(self):
        return itertools.islice(self._buffer, SC_IOV_MAX)

    def _write_sendmsg(self):
        assert self._buffer, 'Data should not be empty'
        if self._conn_lost:
            return
        try:
            nbytes = self._sock.sendmsg(self._get_sendmsg_buffer())
            self._adjust_leftover_buffer(nbytes)
        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:
            self._maybe_resume_protocol()  
            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 _adjust_leftover_buffer(self, nbytes: int) -> None:
        buffer = self._buffer
        while nbytes:
            b = buffer.popleft()
            b_len = len(b)
            if b_len <= nbytes:
                nbytes -= b_len
            else:
                buffer.appendleft(b[nbytes:])
                break

    def _write_send(self):
        assert self._buffer, 'Data should not be empty'
        if self._conn_lost:
            return
        try:
            buffer = self._buffer.popleft()
            n = self._sock.send(buffer)
            if n != len(buffer):
                
                self._buffer.appendleft(buffer[n:])
        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:
            self._maybe_resume_protocol()  
            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 writelines(self, list_of_data):
        if self._eof:
            raise RuntimeError('Cannot call writelines() after write_eof()')
        if self._empty_waiter is not None:
            raise RuntimeError('unable to writelines; sendfile is in progress')
        if not list_of_data:
            return
        self._buffer.extend([memoryview(data) for data in list_of_data])
        self._write_ready()
        
        if self._buffer:
            self._loop._add_writer(self._sock_fd, self._write_ready)

    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

    def close(self):
        self._read_ready_cb = None
        self._write_ready = None
        super().close()


class _SelectorDatagramTransport(_SelectorTransport, transports.DatagramTransport):

    _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._buffer_size = 0
        self._loop.call_soon(self._protocol.connection_made, self)
        
        self._loop.call_soon(self._add_reader,
                             self._sock_fd, self._read_ready)
        if waiter is not None:
            
            self._loop.call_soon(futures._set_result_unless_cancelled,
                                 waiter, None)

    def get_write_buffer_size(self):
        return self._buffer_size

    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:
            
            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

        
        self._buffer.append((bytes(data), addr))
        self._buffer_size += len(data)
        self._maybe_pause_protocol()

    def _sendto_ready(self):
        while self._buffer:
            data, addr = self._buffer.popleft()
            self._buffer_size -= len(data)
            try:
                if self._extra['peername']:
                    self._sock.send(data)
                else:
                    self._sock.sendto(data, addr)
            except (BlockingIOError, InterruptedError):
                self._buffer.appendleft((data, addr))  
                self._buffer_size += len(data)
                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()  
        if not self._buffer:
            self._loop._remove_writer(self._sock_fd)
            if self._closing:
                self._call_connection_lost(None)