event loop using a proactor and related classes a proactor is a notifyoncompletion multiplexer currently a proactor is only implemented on windows with iocp udp sockets may not have a peer name base class for pipe and socket transports def initself loop sock protocol waiternone extranone servernone super initextra loop self setextrasock self sock sock self setprotocolprotocol self server server self buffer none none or bytearray self readfut none self writefut none self pendingwrite 0 self connlost 0 self closing false set when close called self calledconnectionlost false self eofwritten false if self server is not none self server attach self loop callsoonself protocol connectionmade self if waiter is not none only wake up the waiter when connectionmade has been called self loop callsoonfutures setresultunlesscancelled waiter none def reprself info self class name if self sock is none info append closed elif self closing info append closing if self sock is not none info appendf fdself sock fileno if self readfut is not none info appendf readself readfut r if self writefut is not none info appendf writeself writefut r if self buffer info appendf writebufsizelenself buffer if self eofwritten info append eof written return format joininfo def setextraself sock self extra pipe sock def setprotocolself protocol self protocol protocol def getprotocolself return self protocol def isclosingself return self closing def closeself if self closing return self closing true self connlost 1 if not self buffer and self writefut is none self loop callsoonself callconnectionlost none if self readfut is not none self readfut cancel self readfut 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 pipe transport try 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 finally self forcecloseexc def forcecloseself exc if self emptywaiter is not none and not self emptywaiter done if exc is none self emptywaiter setresultnone else self emptywaiter setexceptionexc if self closing and self calledconnectionlost return self closing true self connlost 1 if self writefut self writefut cancel self writefut none if self readfut self readfut cancel self readfut none self pendingwrite 0 self buffer none self loop callsoonself callconnectionlost exc def callconnectionlostself exc if self calledconnectionlost return try self protocol connectionlostexc finally xxx if there is a pending overlapped read on the other end then it may fail with errornetnamedeleted if we just close our end first calling shutdown seems to cure it but maybe using disconnectex would be better if hasattrself sock shutdown and self sock fileno 1 self sock shutdownsocket shutrdwr self sock close self sock none server self server if server is not none server detach self server none self calledconnectionlost true def getwritebuffersizeself size self pendingwrite if self buffer is not none size lenself buffer return size class proactorreadpipetransportproactorbasepipetransport transports readtransport bpo33694 don t cancel self readfut because cancelling an overlapped wsasend loss silently data with the current proactor implementation if cancelioex fails with errornotfound it means that wsasend completed even if hasoverlappediocompleted returns 0 but overlapped cancel currently silently ignores the errornotfound error once the overlapped is ignored the iocp loop will ignores the completion io event and so not read the result of the overlapped wsarecv call the protocol method after calling loopreading since the protocol can decide to pause reading again don t call any protocol method while reading is paused the protocol will be called on resumereading deliver data later in finally clause we got endoffile so no need to reschedule a new read it s a new slice so make it immutable so protocols upstream don t have problems the future will be replaced by next proactor recv call since close has been called we ignore any read data bpo33694 bufferupdated has currently no fast path because of a data loss issue caused by overlapped wsasend cancellation reschedule a new read transport for write pipes starttlscompatible true def initself args kw super initargs kw self emptywaiter none def writeself data if not isinstancedata bytes bytearray memoryview raise typeerror fdata argument must be a byteslike object fnot typedata name if self eofwritten raise runtimeerror writeeof already called 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 observable states 1 idle writefut and buffer both none 2 writing writefut set buffer none 3 backed up writefut set buffer a bytearray we always copy the data so the caller can t modify it while we re still waiting for the io to happen if self writefut is none idle writing assert self buffer is none pass a copy except if it s already immutable self loopwritingdatabytesdata elif not self buffer writing backed up make a mutable copy which we can extend self buffer bytearraydata self maybepauseprotocol else backed up append to buffer also copies self buffer extenddata self maybepauseprotocol def loopwritingself fnone datanone try if f is not none and self writefut is none and self closing xxx most likely self forceclose has been called and it has set self writefut to none return assert f is self writefut self writefut none self pendingwrite 0 if f f result if data is none data self buffer self buffer none if not data if self closing self loop callsoonself callconnectionlost none if self eofwritten self sock shutdownsocket shutwr now that we ve reduced the buffer size tell the protocol to resume writing if it was paused note that we do this last since the callback is called immediately and it may add more data to the buffer even causing the protocol to be paused again self mayberesumeprotocol else self writefut self loop proactor sendself sock data if not self writefut done assert self pendingwrite 0 self pendingwrite lendata self writefut adddonecallbackself loopwriting self maybepauseprotocol else self writefut adddonecallbackself loopwriting if self emptywaiter is not none and self writefut is none self emptywaiter setresultnone except connectionreseterror as exc self forcecloseexc except oserror as exc self fatalerrorexc fatal write error on pipe transport def canwriteeofself return true def writeeofself self close def abortself self forceclosenone def makeemptywaiterself if self emptywaiter is not none raise runtimeerrorempty waiter is already set self emptywaiter self loop createfuture if self writefut is none self emptywaiter setresultnone return self emptywaiter def resetemptywaiterself self emptywaiter none class proactorwritepipetransportproactorbasewritepipetransport def initself args kw super initargs kw self readfut self loop proactor recvself sock 16 self readfut adddonecallbackself pipeclosed def pipeclosedself fut if fut cancelled the transport has been closed return assert fut result b if self closing assert self readfut is none return assert fut is self readfut fut self readfut self readfut none if self writefut is not none self forceclosebrokenpipeerror else self close class proactordatagramtransportproactorbasepipetransport transports datagramtransport maxsize 256 1024 def initself loop sock protocol addressnone waiternone extranone self address address self emptywaiter none self buffersize 0 we don t need to call protocol connectionmade since our base constructor does it for us super initloop sock protocol waiterwaiter extraextra the base constructor sets buffer none so we set it here self buffer collections deque self loop callsoonself loopreading def setextraself sock setsocketextraself sock def getwritebuffersizeself return self buffersize def abortself self forceclosenone def sendtoself data addrnone if not isinstancedata bytes bytearray memoryview raise typeerror data argument must be byteslike object r typedata if not data return if self address is not none and addr not in none self address raise valueerror f invalid address must be none or self address if self connlost and self address if self connlost constants logthresholdforconnlostwrites logger warning socket sendto raised exception self connlost 1 return ensure that what we buffer is immutable self buffer appendbytesdata addr self buffersize lendata if self writefut is none no current write operations are active kick one off self loopwriting else a write operation is already kicked off self maybepauseprotocol def loopwritingself futnone try if self connlost return assert fut is self writefut self writefut none if fut we are in a loopwriting done callback get the result fut result if not self buffer or self connlost and self address the connection has been closed if self closing self loop callsoonself callconnectionlost none return data addr self buffer popleft self buffersize lendata if self address is not none self writefut self loop proactor sendself sock data else self writefut self loop proactor sendtoself sock data addraddr except oserror as exc self protocol errorreceivedexc except exception as exc self fatalerrorexc fatal write error on datagram transport else self writefut adddonecallbackself loopwriting self mayberesumeprotocol def loopreadingself futnone data none try if self connlost return assert self readfut is fut or self readfut is none and self closing self readfut none if fut is not none res fut result if self closing since close has been called we ignore any read data data none return if self address is not none data addr res self address else data addr res if self connlost return if self address is not none self readfut self loop proactor recvself sock self maxsize else self readfut self loop proactor recvfromself sock self maxsize except oserror as exc self protocol errorreceivedexc except exceptions cancellederror if not self closing raise else if self readfut is not none self readfut adddonecallbackself loopreading finally if data self protocol datagramreceiveddata addr class proactorduplexpipetransportproactorreadpipetransport proactorbasewritepipetransport transports transport transport for connected sockets sendfilecompatible constants sendfilemode trynative def initself loop sock protocol waiternone extranone servernone super initloop sock protocol waiter extra server baseevents setnodelaysock def setextraself sock setsocketextraself sock def canwriteeofself return true def writeeofself if self closing or self eofwritten return self eofwritten true if self writefut is none self sock shutdownsocket shutwr class baseproactoreventloopbaseevents baseeventloop def initself proactor super init logger debug using proactor s proactor class name self proactor proactor self selector proactor convenient alias self selfreadingfuture none self acceptfutures socket file descriptor future proactor setloopself self makeselfpipe if threading currentthread is threading mainthread wakeup fd can only be installed to a file descriptor from the main thread signal setwakeupfdself csock fileno def makesockettransportself sock protocol waiternone extranone servernone return proactorsockettransportself sock protocol waiter extra server def makessltransport self rawsock protocol sslcontext waiternone serversidefalse serverhostnamenone extranone servernone sslhandshaketimeoutnone sslshutdowntimeoutnone sslprotocol sslproto sslprotocol self protocol sslcontext waiter serverside serverhostname sslhandshaketimeoutsslhandshaketimeout sslshutdowntimeoutsslshutdowntimeout proactorsockettransportself rawsock sslprotocol extraextra serverserver return sslprotocol apptransport def makedatagramtransportself sock protocol addressnone waiternone extranone return proactordatagramtransportself sock protocol address waiter extra def makeduplexpipetransportself sock protocol waiternone extranone return proactorduplexpipetransportself sock protocol waiter extra def makereadpipetransportself sock protocol waiternone extranone return proactorreadpipetransportself sock protocol waiter extra def makewritepipetransportself sock protocol waiternone extranone we want connectionlost to be called when other end closes return proactorwritepipetransportself sock protocol waiter extra def closeself if self isrunning raise runtimeerrorcannot close a running event loop if self isclosed return if threading currentthread is threading mainthread signal setwakeupfd1 call these methods before closing the event loop before calling baseeventloop close because they can schedule callbacks with callsoon which is forbidden when the event loop is closed self stopacceptfutures self closeselfpipe self proactor close self proactor none self selector none close the event loop super close async def sockrecvself sock n return await self proactor recvsock n async def sockrecvintoself sock buf return await self proactor recvintosock buf async def sockrecvfromself sock bufsize return await self proactor recvfromsock bufsize async def sockrecvfromintoself sock buf nbytes0 if not nbytes nbytes lenbuf return await self proactor recvfromintosock buf nbytes async def socksendallself sock data return await self proactor sendsock data async def socksendtoself sock data address return await self proactor sendtosock data 0 address async def sockconnectself sock address return await self proactor connectsock address async def sockacceptself sock return await self proactor acceptsock async def socksendfilenativeself sock file offset count try fileno file fileno except attributeerror io unsupportedoperation as err raise exceptions sendfilenotavailableerrornot a regular file try fsize os fstatfileno stsize except oserror raise exceptions sendfilenotavailableerrornot a regular file blocksize count if count else fsize if not blocksize return 0 empty file blocksize minblocksize 0xffffffff endpos minoffset count fsize if count else fsize offset minoffset fsize totalsent 0 try while true blocksize minendpos offset blocksize if blocksize 0 return totalsent await self proactor sendfilesock file offset blocksize offset blocksize totalsent blocksize finally if totalsent 0 file seekoffset async def sendfilenativeself transp file offset count 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 def closeselfpipeself if self selfreadingfuture is not none self selfreadingfuture cancel self selfreadingfuture none self ssock close self ssock none self csock close self csock none self internalfds 1 def makeselfpipeself a selfsocket really self ssock self csock socket socketpair self ssock setblockingfalse self csock setblockingfalse self internalfds 1 def loopselfreadingself fnone try if f is not none f result may raise if self selfreadingfuture is not f when we scheduled this future we assigned it to selfreadingfuture if it s not there now something has tried to cancel the loop while this callback was still in the queue see windowsevents proactoreventloop runforever in that case stop here instead of continuing to schedule a new iteration return f self proactor recvself ssock 4096 except exceptions cancellederror closeselfpipe has been called stop waiting for data return except systemexit keyboardinterrupt raise except baseexception as exc self callexceptionhandler message error on reading from the event loop self pipe exception exc loop self else self selfreadingfuture f f adddonecallbackself loopselfreading def writetoselfself 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 csock self csock if csock is none return try csock sendb 0 except oserror if self debug logger debugfail to write a null byte into the selfpipe socket excinfotrue def startservingself protocolfactory sock sslcontextnone servernone backlog100 sslhandshaketimeoutnone sslshutdowntimeoutnone def loopfnone try if f is not none conn addr f result if self debug logger debugr got a new connection from r r server addr conn protocol protocolfactory if sslcontext is not none self makessltransport conn protocol sslcontext serversidetrue extra peername addr serverserver sslhandshaketimeoutsslhandshaketimeout sslshutdowntimeoutsslshutdowntimeout else self makesockettransport conn protocol extra peername addr serverserver if self isclosed return f self proactor acceptsock except oserror as exc if sock fileno 1 self callexceptionhandler message accept failed on a socket exception exc socket trsock transportsocketsock sock close elif self debug logger debugaccept failed on socket r sock excinfotrue except exceptions cancellederror sock close else self acceptfuturessock fileno f f adddonecallbackloop self callsoonloop def processeventsself eventlist events are processed in the iocpproactor poll method pass def stopacceptfuturesself for future in self acceptfutures values future cancel self acceptfutures clear def stopservingself sock future self acceptfutures popsock fileno none if future future cancel self proactor stopservingsock sock close udp sockets may not have a peer name base class for pipe and socket transports none or bytearray set when close called only wake up the waiter when connection_made has been called xxx if there is a pending overlapped read on the other end then it may fail with error_netname_deleted if we just close our end first calling shutdown seems to cure it but maybe using disconnectex would be better transport for read pipes bpo 33694 don t cancel self _read_fut because cancelling an overlapped wsasend loss silently data with the current proactor implementation if cancelioex fails with error_not_found it means that wsasend completed even if hasoverlappediocompleted returns 0 but overlapped cancel currently silently ignores the error_not_found error once the overlapped is ignored the iocp loop will ignores the completion i o event and so not read the result of the overlapped wsarecv call the protocol method after calling _loop_reading since the protocol can decide to pause reading again don t call any protocol method while reading is paused the protocol will be called on resume_reading deliver data later in finally clause we got end of file so no need to reschedule a new read it s a new slice so make it immutable so protocols upstream don t have problems the future will be replaced by next proactor recv call since close has been called we ignore any read data bpo 33694 buffer_updated has currently no fast path because of a data loss issue caused by overlapped wsasend cancellation reschedule a new read transport for write pipes observable states 1 idle _write_fut and _buffer both none 2 writing _write_fut set _buffer none 3 backed up _write_fut set _buffer a bytearray we always copy the data so the caller can t modify it while we re still waiting for the i o to happen idle writing pass a copy except if it s already immutable writing backed up make a mutable copy which we can extend backed up append to buffer also copies xxx most likely self _force_close has been called and it has set self _write_fut to none now that we ve reduced the buffer size tell the protocol to resume writing if it was paused note that we do this last since the callback is called immediately and it may add more data to the buffer even causing the protocol to be paused again the transport has been closed we don t need to call _protocol connection_made since our base constructor does it for us the base constructor sets _buffer none so we set it here ensure that what we buffer is immutable no current write operations are active kick one off else a write operation is already kicked off we are in a _loop_writing done callback get the result the connection has been closed since close has been called we ignore any read data transport for duplex pipes transport for connected sockets convenient alias socket file descriptor future wakeup fd can only be installed to a file descriptor from the main thread we want connection_lost to be called when other end closes call these methods before closing the event loop before calling baseeventloop close because they can schedule callbacks with call_soon which is forbidden when the event loop is closed close the event loop empty file a self socket really may raise when we scheduled this future we assigned it to _self_reading_future if it s not there now something has tried to cancel the loop while this callback was still in the queue see windows_events proactoreventloop run_forever in that case stop here instead of continuing to schedule a new iteration _close_self_pipe has been called stop waiting for data 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 events are processed in the iocpproactor _poll method
__all__ = 'BaseProactorEventLoop',

import io
import os
import socket
import warnings
import signal
import threading
import collections

from . import base_events
from . import constants
from . import futures
from . import exceptions
from . import protocols
from . import sslproto
from . import transports
from . import trsock
from .log import logger


def _set_socket_extra(transport, sock):
    transport._extra['socket'] = trsock.TransportSocket(sock)

    try:
        transport._extra['sockname'] = sock.getsockname()
    except socket.error:
        if transport._loop.get_debug():
            logger.warning(
                "getsockname() failed on %r", sock, exc_info=True)

    if 'peername' not in transport._extra:
        try:
            transport._extra['peername'] = sock.getpeername()
        except socket.error:
            
            transport._extra['peername'] = None


class _ProactorBasePipeTransport(transports._FlowControlMixin,
                                 transports.BaseTransport):
    

    def __init__(self, loop, sock, protocol, waiter=None,
                 extra=None, server=None):
        super().__init__(extra, loop)
        self._set_extra(sock)
        self._sock = sock
        self.set_protocol(protocol)
        self._server = server
        self._buffer = None  
        self._read_fut = None
        self._write_fut = None
        self._pending_write = 0
        self._conn_lost = 0
        self._closing = False  
        self._called_connection_lost = False
        self._eof_written = False
        if self._server is not None:
            self._server._attach()
        self._loop.call_soon(self._protocol.connection_made, self)
        if waiter is not None:
            
            self._loop.call_soon(futures._set_result_unless_cancelled,
                                 waiter, None)

    def __repr__(self):
        info = [self.__class__.__name__]
        if self._sock is None:
            info.append('closed')
        elif self._closing:
            info.append('closing')
        if self._sock is not None:
            info.append(f'fd={self._sock.fileno()}')
        if self._read_fut is not None:
            info.append(f'read={self._read_fut!r}')
        if self._write_fut is not None:
            info.append(f'write={self._write_fut!r}')
        if self._buffer:
            info.append(f'write_bufsize={len(self._buffer)}')
        if self._eof_written:
            info.append('EOF written')
        return '<{}>'.format(' '.join(info))

    def _set_extra(self, sock):
        self._extra['pipe'] = sock

    def set_protocol(self, protocol):
        self._protocol = protocol

    def get_protocol(self):
        return self._protocol

    def is_closing(self):
        return self._closing

    def close(self):
        if self._closing:
            return
        self._closing = True
        self._conn_lost += 1
        if not self._buffer and self._write_fut is None:
            self._loop.call_soon(self._call_connection_lost, None)
        if self._read_fut is not None:
            self._read_fut.cancel()
            self._read_fut = 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 pipe transport'):
        try:
            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,
                })
        finally:
            self._force_close(exc)

    def _force_close(self, exc):
        if self._empty_waiter is not None and not self._empty_waiter.done():
            if exc is None:
                self._empty_waiter.set_result(None)
            else:
                self._empty_waiter.set_exception(exc)
        if self._closing and self._called_connection_lost:
            return
        self._closing = True
        self._conn_lost += 1
        if self._write_fut:
            self._write_fut.cancel()
            self._write_fut = None
        if self._read_fut:
            self._read_fut.cancel()
            self._read_fut = None
        self._pending_write = 0
        self._buffer = None
        self._loop.call_soon(self._call_connection_lost, exc)

    def _call_connection_lost(self, exc):
        if self._called_connection_lost:
            return
        try:
            self._protocol.connection_lost(exc)
        finally:
            
            
            
            
            if hasattr(self._sock, 'shutdown') and self._sock.fileno() != -1:
                self._sock.shutdown(socket.SHUT_RDWR)
            self._sock.close()
            self._sock = None
            server = self._server
            if server is not None:
                server._detach()
                self._server = None
            self._called_connection_lost = True

    def get_write_buffer_size(self):
        size = self._pending_write
        if self._buffer is not None:
            size += len(self._buffer)
        return size


class _ProactorReadPipeTransport(_ProactorBasePipeTransport,
                                 transports.ReadTransport):
    

    def __init__(self, loop, sock, protocol, waiter=None,
                 extra=None, server=None, buffer_size=65536):
        self._pending_data_length = -1
        self._paused = True
        super().__init__(loop, sock, protocol, waiter, extra, server)

        self._data = bytearray(buffer_size)
        self._loop.call_soon(self._loop_reading)
        self._paused = False

    def is_reading(self):
        return not self._paused and not self._closing

    def pause_reading(self):
        if self._closing or self._paused:
            return
        self._paused = True

        
        
        
        
        
        
        
        
        
        

        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
        if self._read_fut is None:
            self._loop.call_soon(self._loop_reading, None)

        length = self._pending_data_length
        self._pending_data_length = -1
        if length > -1:
            
            
            self._loop.call_soon(self._data_received, self._data[:length], length)

        if self._loop.get_debug():
            logger.debug("%r resumes reading", self)

    def _eof_received(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 not keep_open:
            self.close()

    def _data_received(self, data, length):
        if self._paused:
            
            
            assert self._pending_data_length == -1
            self._pending_data_length = length
            return

        if length == 0:
            self._eof_received()
            return

        if isinstance(self._protocol, protocols.BufferedProtocol):
            try:
                protocols._feed_data_to_buffered_proto(self._protocol, data)
            except (SystemExit, KeyboardInterrupt):
                raise
            except BaseException as exc:
                self._fatal_error(exc,
                                  'Fatal error: protocol.buffer_updated() '
                                  'call failed.')
                return
        else:
            self._protocol.data_received(data)

    def _loop_reading(self, fut=None):
        length = -1
        data = None
        try:
            if fut is not None:
                assert self._read_fut is fut or (self._read_fut is None and
                                                 self._closing)
                self._read_fut = None
                if fut.done():
                    
                    length = fut.result()
                    if length == 0:
                        
                        return

                    
                    data = bytes(memoryview(self._data)[:length])
                else:
                    
                    fut.cancel()

            if self._closing:
                
                return

            
            

            if not self._paused:
                
                self._read_fut = self._loop._proactor.recv_into(self._sock, self._data)
        except ConnectionAbortedError as exc:
            if not self._closing:
                self._fatal_error(exc, 'Fatal read error on pipe transport')
            elif self._loop.get_debug():
                logger.debug("Read error on pipe transport while closing",
                             exc_info=True)
        except ConnectionResetError as exc:
            self._force_close(exc)
        except OSError as exc:
            self._fatal_error(exc, 'Fatal read error on pipe transport')
        except exceptions.CancelledError:
            if not self._closing:
                raise
        else:
            if not self._paused:
                self._read_fut.add_done_callback(self._loop_reading)
        finally:
            if length > -1:
                self._data_received(data, length)


class _ProactorBaseWritePipeTransport(_ProactorBasePipeTransport,
                                      transports.WriteTransport):
    

    _start_tls_compatible = True

    def __init__(self, *args, **kw):
        super().__init__(*args, **kw)
        self._empty_waiter = None

    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__}")
        if self._eof_written:
            raise RuntimeError('write_eof() already called')
        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 self._write_fut is None:  
            assert self._buffer is None
            
            self._loop_writing(data=bytes(data))
        elif not self._buffer:  
            
            self._buffer = bytearray(data)
            self._maybe_pause_protocol()
        else:  
            
            self._buffer.extend(data)
            self._maybe_pause_protocol()

    def _loop_writing(self, f=None, data=None):
        try:
            if f is not None and self._write_fut is None and self._closing:
                
                
                return
            assert f is self._write_fut
            self._write_fut = None
            self._pending_write = 0
            if f:
                f.result()
            if data is None:
                data = self._buffer
                self._buffer = None
            if not data:
                if self._closing:
                    self._loop.call_soon(self._call_connection_lost, None)
                if self._eof_written:
                    self._sock.shutdown(socket.SHUT_WR)
                
                
                
                
                
                self._maybe_resume_protocol()
            else:
                self._write_fut = self._loop._proactor.send(self._sock, data)
                if not self._write_fut.done():
                    assert self._pending_write == 0
                    self._pending_write = len(data)
                    self._write_fut.add_done_callback(self._loop_writing)
                    self._maybe_pause_protocol()
                else:
                    self._write_fut.add_done_callback(self._loop_writing)
            if self._empty_waiter is not None and self._write_fut is None:
                self._empty_waiter.set_result(None)
        except ConnectionResetError as exc:
            self._force_close(exc)
        except OSError as exc:
            self._fatal_error(exc, 'Fatal write error on pipe transport')

    def can_write_eof(self):
        return True

    def write_eof(self):
        self.close()

    def abort(self):
        self._force_close(None)

    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 self._write_fut is None:
            self._empty_waiter.set_result(None)
        return self._empty_waiter

    def _reset_empty_waiter(self):
        self._empty_waiter = None


class _ProactorWritePipeTransport(_ProactorBaseWritePipeTransport):
    def __init__(self, *args, **kw):
        super().__init__(*args, **kw)
        self._read_fut = self._loop._proactor.recv(self._sock, 16)
        self._read_fut.add_done_callback(self._pipe_closed)

    def _pipe_closed(self, fut):
        if fut.cancelled():
            
            return
        assert fut.result() == b''
        if self._closing:
            assert self._read_fut is None
            return
        assert fut is self._read_fut, (fut, self._read_fut)
        self._read_fut = None
        if self._write_fut is not None:
            self._force_close(BrokenPipeError())
        else:
            self.close()


class _ProactorDatagramTransport(_ProactorBasePipeTransport,
                                 transports.DatagramTransport):
    max_size = 256 * 1024
    def __init__(self, loop, sock, protocol, address=None,
                 waiter=None, extra=None):
        self._address = address
        self._empty_waiter = None
        self._buffer_size = 0
        
        
        super().__init__(loop, sock, protocol, waiter=waiter, extra=extra)

        
        self._buffer = collections.deque()
        self._loop.call_soon(self._loop_reading)

    def _set_extra(self, sock):
        _set_socket_extra(self, sock)

    def get_write_buffer_size(self):
        return self._buffer_size

    def abort(self):
        self._force_close(None)

    def sendto(self, data, addr=None):
        if not isinstance(data, (bytes, bytearray, memoryview)):
            raise TypeError('data argument must be bytes-like object (%r)',
                            type(data))

        if not data:
            return

        if self._address is not None and addr not in (None, self._address):
            raise ValueError(
                f'Invalid address: must be None or {self._address}')

        if self._conn_lost and self._address:
            if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES:
                logger.warning('socket.sendto() raised exception.')
            self._conn_lost += 1
            return

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

        if self._write_fut is None:
            
            self._loop_writing()
        

        self._maybe_pause_protocol()

    def _loop_writing(self, fut=None):
        try:
            if self._conn_lost:
                return

            assert fut is self._write_fut
            self._write_fut = None
            if fut:
                
                fut.result()

            if not self._buffer or (self._conn_lost and self._address):
                
                if self._closing:
                    self._loop.call_soon(self._call_connection_lost, None)
                return

            data, addr = self._buffer.popleft()
            self._buffer_size -= len(data)
            if self._address is not None:
                self._write_fut = self._loop._proactor.send(self._sock,
                                                            data)
            else:
                self._write_fut = self._loop._proactor.sendto(self._sock,
                                                              data,
                                                              addr=addr)
        except OSError as exc:
            self._protocol.error_received(exc)
        except Exception as exc:
            self._fatal_error(exc, 'Fatal write error on datagram transport')
        else:
            self._write_fut.add_done_callback(self._loop_writing)
            self._maybe_resume_protocol()

    def _loop_reading(self, fut=None):
        data = None
        try:
            if self._conn_lost:
                return

            assert self._read_fut is fut or (self._read_fut is None and
                                             self._closing)

            self._read_fut = None
            if fut is not None:
                res = fut.result()

                if self._closing:
                    
                    data = None
                    return

                if self._address is not None:
                    data, addr = res, self._address
                else:
                    data, addr = res

            if self._conn_lost:
                return
            if self._address is not None:
                self._read_fut = self._loop._proactor.recv(self._sock,
                                                           self.max_size)
            else:
                self._read_fut = self._loop._proactor.recvfrom(self._sock,
                                                               self.max_size)
        except OSError as exc:
            self._protocol.error_received(exc)
        except exceptions.CancelledError:
            if not self._closing:
                raise
        else:
            if self._read_fut is not None:
                self._read_fut.add_done_callback(self._loop_reading)
        finally:
            if data:
                self._protocol.datagram_received(data, addr)


class _ProactorDuplexPipeTransport(_ProactorReadPipeTransport,
                                   _ProactorBaseWritePipeTransport,
                                   transports.Transport):
    

    def can_write_eof(self):
        return False

    def write_eof(self):
        raise NotImplementedError


class _ProactorSocketTransport(_ProactorReadPipeTransport,
                               _ProactorBaseWritePipeTransport,
                               transports.Transport):
    

    _sendfile_compatible = constants._SendfileMode.TRY_NATIVE

    def __init__(self, loop, sock, protocol, waiter=None,
                 extra=None, server=None):
        super().__init__(loop, sock, protocol, waiter, extra, server)
        base_events._set_nodelay(sock)

    def _set_extra(self, sock):
        _set_socket_extra(self, sock)

    def can_write_eof(self):
        return True

    def write_eof(self):
        if self._closing or self._eof_written:
            return
        self._eof_written = True
        if self._write_fut is None:
            self._sock.shutdown(socket.SHUT_WR)


class BaseProactorEventLoop(base_events.BaseEventLoop):

    def __init__(self, proactor):
        super().__init__()
        logger.debug('Using proactor: %s', proactor.__class__.__name__)
        self._proactor = proactor
        self._selector = proactor   
        self._self_reading_future = None
        self._accept_futures = {}   
        proactor.set_loop(self)
        self._make_self_pipe()
        if threading.current_thread() is threading.main_thread():
            
            signal.set_wakeup_fd(self._csock.fileno())

    def _make_socket_transport(self, sock, protocol, waiter=None,
                               extra=None, server=None):
        return _ProactorSocketTransport(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=None,
            ssl_shutdown_timeout=None):
        ssl_protocol = sslproto.SSLProtocol(
                self, protocol, sslcontext, waiter,
                server_side, server_hostname,
                ssl_handshake_timeout=ssl_handshake_timeout,
                ssl_shutdown_timeout=ssl_shutdown_timeout)
        _ProactorSocketTransport(self, rawsock, ssl_protocol,
                                 extra=extra, server=server)
        return ssl_protocol._app_transport

    def _make_datagram_transport(self, sock, protocol,
                                 address=None, waiter=None, extra=None):
        return _ProactorDatagramTransport(self, sock, protocol, address,
                                          waiter, extra)

    def _make_duplex_pipe_transport(self, sock, protocol, waiter=None,
                                    extra=None):
        return _ProactorDuplexPipeTransport(self,
                                            sock, protocol, waiter, extra)

    def _make_read_pipe_transport(self, sock, protocol, waiter=None,
                                  extra=None):
        return _ProactorReadPipeTransport(self, sock, protocol, waiter, extra)

    def _make_write_pipe_transport(self, sock, protocol, waiter=None,
                                   extra=None):
        
        return _ProactorWritePipeTransport(self,
                                           sock, protocol, waiter, extra)

    def close(self):
        if self.is_running():
            raise RuntimeError("Cannot close a running event loop")
        if self.is_closed():
            return

        if threading.current_thread() is threading.main_thread():
            signal.set_wakeup_fd(-1)
        
        
        
        self._stop_accept_futures()
        self._close_self_pipe()
        self._proactor.close()
        self._proactor = None
        self._selector = None

        
        super().close()

    async def sock_recv(self, sock, n):
        return await self._proactor.recv(sock, n)

    async def sock_recv_into(self, sock, buf):
        return await self._proactor.recv_into(sock, buf)

    async def sock_recvfrom(self, sock, bufsize):
        return await self._proactor.recvfrom(sock, bufsize)

    async def sock_recvfrom_into(self, sock, buf, nbytes=0):
        if not nbytes:
            nbytes = len(buf)

        return await self._proactor.recvfrom_into(sock, buf, nbytes)

    async def sock_sendall(self, sock, data):
        return await self._proactor.send(sock, data)

    async def sock_sendto(self, sock, data, address):
        return await self._proactor.sendto(sock, data, 0, address)

    async def sock_connect(self, sock, address):
        return await self._proactor.connect(sock, address)

    async def sock_accept(self, sock):
        return await self._proactor.accept(sock)

    async def _sock_sendfile_native(self, sock, file, offset, count):
        try:
            fileno = file.fileno()
        except (AttributeError, io.UnsupportedOperation) as err:
            raise exceptions.SendfileNotAvailableError("not a regular file")
        try:
            fsize = os.fstat(fileno).st_size
        except OSError:
            raise exceptions.SendfileNotAvailableError("not a regular file")
        blocksize = count if count else fsize
        if not blocksize:
            return 0  

        blocksize = min(blocksize, 0xffff_ffff)
        end_pos = min(offset + count, fsize) if count else fsize
        offset = min(offset, fsize)
        total_sent = 0
        try:
            while True:
                blocksize = min(end_pos - offset, blocksize)
                if blocksize <= 0:
                    return total_sent
                await self._proactor.sendfile(sock, file, offset, blocksize)
                offset += blocksize
                total_sent += blocksize
        finally:
            if total_sent > 0:
                file.seek(offset)

    async def _sendfile_native(self, transp, file, offset, count):
        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()

    def _close_self_pipe(self):
        if self._self_reading_future is not None:
            self._self_reading_future.cancel()
            self._self_reading_future = None
        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

    def _loop_self_reading(self, f=None):
        try:
            if f is not None:
                f.result()  
            if self._self_reading_future is not f:
                
                
                
                
                
                
                return
            f = self._proactor.recv(self._ssock, 4096)
        except exceptions.CancelledError:
            
            return
        except (SystemExit, KeyboardInterrupt):
            raise
        except BaseException as exc:
            self.call_exception_handler({
                'message': 'Error on reading from the event loop self pipe',
                'exception': exc,
                'loop': self,
            })
        else:
            self._self_reading_future = f
            f.add_done_callback(self._loop_self_reading)

    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=None,
                       ssl_shutdown_timeout=None):

        def loop(f=None):
            try:
                if f is not None:
                    conn, addr = f.result()
                    if self._debug:
                        logger.debug("%r got a new connection from %r: %r",
                                     server, addr, conn)
                    protocol = protocol_factory()
                    if sslcontext is not None:
                        self._make_ssl_transport(
                            conn, protocol, sslcontext, server_side=True,
                            extra={'peername': addr}, server=server,
                            ssl_handshake_timeout=ssl_handshake_timeout,
                            ssl_shutdown_timeout=ssl_shutdown_timeout)
                    else:
                        self._make_socket_transport(
                            conn, protocol,
                            extra={'peername': addr}, server=server)
                if self.is_closed():
                    return
                f = self._proactor.accept(sock)
            except OSError as exc:
                if sock.fileno() != -1:
                    self.call_exception_handler({
                        'message': 'Accept failed on a socket',
                        'exception': exc,
                        'socket': trsock.TransportSocket(sock),
                    })
                    sock.close()
                elif self._debug:
                    logger.debug("Accept failed on socket %r",
                                 sock, exc_info=True)
            except exceptions.CancelledError:
                sock.close()
            else:
                self._accept_futures[sock.fileno()] = f
                f.add_done_callback(loop)

        self.call_soon(loop)

    def _process_events(self, event_list):
        
        pass

    def _stop_accept_futures(self):
        for future in self._accept_futures.values():
            future.cancel()
        self._accept_futures.clear()

    def _stop_serving(self, sock):
        future = self._accept_futures.pop(sock.fileno(), None)
        if future:
            future.cancel()
        self._proactor._stop_serving(sock)
        sock.close()