U if<}@sTdZdZddlZddlZddlZddlZddlZddlZddlZddl m Z ddl m Z ddl m Z ddl m Z dd l mZdd l mZdd l mZdd l mZdd lmZddZGdddejejZGdddeejZGdddeejZGdddeZGdddeZGdddeeejZGdddeeejZ Gddde j!Z"dS) zEvent loop using a proactor and related classes. A proactor is a "notify-on-completion" multiplexer. Currently a proactor is only implemented on Windows with IOCP. )BaseProactorEventLoopN) base_events) constants)futures) exceptions) protocols)sslproto) transports)trsock)loggercCst||jd<z||jd<Wn0tjk rR|jrNtj d|ddYnXd|jkrz| |jd<Wn tjk rd|jd<YnXdS)NsocketZsocknamezgetsockname() failed on %rTexc_infopeername) r TransportSocket_extraZ getsocknamer error_loop get_debugr warningZ getpeername) transportsockr ) r4__name__r appendr(filenor$r%r#lenr)formatjoin)r-inforrr__repr__Hs         z#_ProactorBasePipeTransport.__repr__cCs||jd<dS)Npipe)rr-rrrrrZsz%_ProactorBasePipeTransport._set_extracCs ||_dSNr+)r-r/rrrr!]sz'_ProactorBasePipeTransport.set_protocolcCs|jSrBrCr-rrr get_protocol`sz'_ProactorBasePipeTransport.get_protocolcCs|jSrB)r(rDrrr is_closingcsz%_ProactorBasePipeTransport.is_closingcCs\|jr dSd|_|jd7_|js>|jdkr>|j|jd|jdk rX|jd|_dS)NTr) r(r'r#r%rr*_call_connection_lostr$cancelrDrrrclosefs  z _ProactorBasePipeTransport.closecCs*|jdk r&|d|t|d|dS)Nzunclosed transport )source)r ResourceWarningrI)r-Z_warnrrr__del__qs z"_ProactorBasePipeTransport.__del__Fatal error on pipe transportc CsVzDt|tr*|jrBtjd||ddn|j||||jdW5||XdS)Nz%r: %sTr)message exceptionrr/) _force_close isinstanceOSErrorrrr debugcall_exception_handlerr+)r-excrNrrr _fatal_errorvs   z'_ProactorBasePipeTransport._fatal_errorcCs|jdk r6|js6|dkr*|jdn |j||jr@dSd|_|jd7_|jrj|jd|_|jr|jd|_d|_ d|_ |j |j |dS)NTrr) _empty_waiterdone set_resultZ set_exceptionr(r'r%rHr$r&r#rr*rG)r-rUrrrrPs"   z'_ProactorBasePipeTransport._force_closec Cs^z|j |W5t|jdr,|jtj|jd|_|j}|dk rX|d|_XdS)Nshutdown) hasattrr rZr Z SHUT_RDWRrIr"Z_detachr+Zconnection_lost)r-rUr2rrrrGs  z0_ProactorBasePipeTransport._call_connection_lostcCs"|j}|jdk r|t|j7}|SrB)r&r#r;)r-sizerrrget_write_buffer_sizes z0_ProactorBasePipeTransport.get_write_buffer_size)NNN)rM)r8 __module__ __qualname____doc__rr?rr!rErFrIwarningswarnrLrVrPrGr] __classcell__rrr3rr.s   rcsTeZdZdZdfdd ZddZddZd d Zd d Zd dZ dddZ Z S)_ProactorReadPipeTransportzTransport for read pipes.Ncs:d|_d|_t|||||||j|jd|_dS)NTF) _pending_data_pausedrrrr* _loop_readingr,r3rrrs z#_ProactorReadPipeTransport.__init__cCs|j o|j SrB)rfr(rDrrr is_readingsz%_ProactorReadPipeTransport.is_readingcCs0|js |jrdSd|_|jr,td|dS)NTz%r pauses reading)r(rfrrr rSrDrrr pause_readings   z(_ProactorReadPipeTransport.pause_readingcCsn|js |jsdSd|_|jdkr0|j|jd|j}d|_|dk rT|j|j||jrjt d|dS)NFz%r resumes reading) r(rfr$rr*rgre_data_receivedrr rSr-datarrrresume_readings   z)_ProactorReadPipeTransport.resume_readingc Cs|jrtd|z|j}WnLttfk r>Yn4tk rp}z| |dWYdSd}~XYnX|s~| dS)Nz%r received EOFz1Fatal error: protocol.eof_received() call failed.) rrr rSr+Z eof_received SystemExitKeyboardInterrupt BaseExceptionrVrI)r-Z keep_openrUrrr _eof_receiveds  z(_ProactorReadPipeTransport._eof_receivedc Cs|jr||_dS|s |dSt|jtjrzt|j|Wqtt fk rZYqt k r}z| |dWYdSd}~XYqXn |j |dS)Nz3Fatal error: protocol.buffer_updated() call failed.) rfrerqrQr+rZBufferedProtocolZ_feed_data_to_buffered_protornrorprVZ data_received)r-rlrUrrrrjs"z)_ProactorReadPipeTransport._data_receivedc Cstd}zRzp|dk r2d|_|r*|}n||jrHd}WWdS|dkr\WWdS|jsv|jj |j d|_Wnt k r}z0|js| |dn|j rtjdddW5d}~XYntk r}z||W5d}~XYnftk r}z| |dW5d}~XYn8tjk r>|js:YnX|jsV|j|jW5|dk rn||XdS)Niz"Fatal read error on pipe transportz*Read error on pipe transport while closingTr)rjr$rXresultrHr(rfr _proactorrecvr ConnectionAbortedErrorrVrr rSConnectionResetErrorrPrRrCancelledErroradd_done_callbackrg)r-futrlrUrrrrgs@     z(_ProactorReadPipeTransport._loop_reading)NNN)N) r8r^r_r`rrhrirmrqrjrgrcrrr3rrds rdcs^eZdZdZdZfddZddZddd Zd d Zd d Z ddZ ddZ ddZ Z S)_ProactorBaseWritePipeTransportzTransport for write pipes.Tcstj||d|_dSrB)rrrWr-argskwr3rrrGsz(_ProactorBaseWritePipeTransport.__init__cCst|tttfs$tdt|j|jr2td|j dk rDtd|sLdS|j rz|j t j krht d|j d7_ dS|jdkr|jt|dn.|jst||_|n|j||dS)Nz/data argument must be a bytes-like object, not zwrite_eof() already calledz(unable to write; sendfile is in progresszsocket.send() raised exception.r)rl)rQbytes bytearray memoryview TypeErrortyper8r) RuntimeErrorrWr'r!LOG_THRESHOLD_FOR_CONNLOST_WRITESr rr% _loop_writingr#_maybe_pause_protocolextendrkrrrwriteKs,       z%_ProactorBaseWritePipeTransport.writeNc CsVz|dk r |jdkr |jr WdSd|_d|_|r8||dkrL|j}d|_|s|jrf|j|jd|jrz|j t j | nN|jj|j ||_|jst||_|j|j|n|j|j|jdk r|jdkr|jdWn\tk r"}z||W5d}~XYn0tk rP}z||dW5d}~XYnXdS)Nrz#Fatal write error on pipe transport)r%r(r&rsr#rr*rGr)r rZr SHUT_WR_maybe_resume_protocolrtsendrXr;ryrrrWrYrwrPrRrV)r-frlrUrrrrqs8    z-_ProactorBaseWritePipeTransport._loop_writingcCsdSNTrrDrrr can_write_eofsz-_ProactorBaseWritePipeTransport.can_write_eofcCs |dSrB)rIrDrrr write_eofsz)_ProactorBaseWritePipeTransport.write_eofcCs|ddSrBrPrDrrrabortsz%_ProactorBaseWritePipeTransport.abortcCs:|jdk rtd|j|_|jdkr4|jd|jS)NzEmpty waiter is already set)rWrrZ create_futurer%rYrDrrr_make_empty_waiters     z2_ProactorBaseWritePipeTransport._make_empty_waitercCs d|_dSrB)rWrDrrr_reset_empty_waitersz3_ProactorBaseWritePipeTransport._reset_empty_waiter)NN)r8r^r_r`Z_start_tls_compatiblerrrrrrrrrcrrr3rr{As & )r{cs$eZdZfddZddZZS)_ProactorWritePipeTransportcs4tj|||jj|jd|_|j|jdS)N) rrrrtrur r$ry _pipe_closedr|r3rrrsz$_ProactorWritePipeTransport.__init__cCs@|r dS|jrdSd|_|jdk r4|tn|dSrB)Z cancelledr(r$r%rPBrokenPipeErrorrI)r-rzrrrrs z(_ProactorWritePipeTransport._pipe_closed)r8r^r_rrrcrrr3rrs rcsXeZdZdZdfdd ZddZddZd d Zdd d Zdd dZ dddZ Z S)_ProactorDatagramTransportiNcs>||_d|_tj|||||dt|_|j|j dS)N)r0r1) _addressrWrr collectionsdequer#rr*rg)r-r.rr/addressr0r1r3rrrs  z#_ProactorDatagramTransport.__init__cCst||dSrBrrArrrrsz%_ProactorDatagramTransport._set_extracCstdd|jDS)Ncss|]\}}t|VqdSrB)r;).0rl_rrr szC_ProactorDatagramTransport.get_write_buffer_size..)sumr#rDrrrr]sz0_ProactorDatagramTransport.get_write_buffer_sizecCs|ddSrBrrDrrrrsz _ProactorDatagramTransport.abortcCst|tttfstdt||s&dS|jdk rN|d|jfkrNtd|j|jr|jr|jt j krpt d|jd7_dS|j t||f|jdkr||dS)Nz,data argument must be bytes-like object (%r)z!Invalid address: must be None or z!socket.sendto() raised exception.r)rQrrrrrr ValueErrorr'rrr rr#r9r%rr)r-rladdrrrrsendtos&     z!_ProactorDatagramTransport.sendtoc Cs z|jrWdSd|_|r ||jr2|jrN|jrN|jrH|j|jdWdS|j \}}|jdk r||jj |j ||_n|jj j |j ||d|_WnZtk r}z|j|W5d}~XYnDtk r}z||dW5d}~XYnX|j|j|dS)N)rz'Fatal write error on datagram transport)r'r%rsr#rr(rr*rGpopleftrtrr rrRr+error_received ExceptionrVryrr)r-rzrlrrUrrrrs2    z(_ProactorDatagramTransport._loop_writingc Cs4d}zz|jrWWdSd|_|dk rf|}|jrFd}WWdS|jdk r^||j}}n|\}}|jrvWWdS|jdk r|jj |j |j |_n|jj |j |j |_WnJt k r}z|j|W5d}~XYn8tjk r|jsYnX|jdk r|j|jW5|r.|j||XdSrB)r+Zdatagram_receivedr'r$rsr(rrrtrur max_sizeZrecvfromrRrrrxryrg)r-rzrlrresrUrrrrgs>         z(_ProactorDatagramTransport._loop_reading)NNN)N)N)N) r8r^r_rrrr]rrrrgrcrrr3rrs   !rc@s eZdZdZddZddZdS)_ProactorDuplexPipeTransportzTransport for duplex pipes.cCsdS)NFrrDrrrrJsz*_ProactorDuplexPipeTransport.can_write_eofcCstdSrB)NotImplementedErrorrDrrrrMsz&_ProactorDuplexPipeTransport.write_eofN)r8r^r_r`rrrrrrrEsrcsBeZdZdZejjZd fdd ZddZ ddZ d d Z Z S) _ProactorSocketTransportz Transport for connected sockets.Ncs$t||||||t|dSrB)rrrZ _set_nodelayr,r3rrrXsz!_ProactorSocketTransport.__init__cCst||dSrBrrArrrr]sz#_ProactorSocketTransport._set_extracCsdSrrrDrrrr`sz&_ProactorSocketTransport.can_write_eofcCs2|js |jrdSd|_|jdkr.|jtjdSr)r(r)r%r rZr rrDrrrrcs   z"_ProactorSocketTransport.write_eof)NNN) r8r^r_r`rZ _SendfileModeZ TRY_NATIVEZ_sendfile_compatiblerrrrrcrrr3rrQsrcseZdZfddZd3ddZd4dddddddd Zd5d d Zd6d d Zd7ddZd8ddZ fddZ ddZ ddZ ddZ ddZddZddZd d!Zd"d#Zd$d%Zd9d&d'Zd(d)Zd:d+d,Zd-d.Zd/d0Zd1d2ZZS);rcshttd|jj||_||_d|_i|_ | || t t krdt|jdS)NzUsing proactor: %s)rrr rSr4r8rt _selector_self_reading_future_accept_futuresZset_loop_make_self_pipe threadingcurrent_thread main_threadsignal set_wakeup_fd_csockr:)r-Zproactorr3rrrms  zBaseProactorEventLoop.__init__NcCst||||||SrB)r)r-rr/r0r1r2rrr_make_socket_transportzs z,BaseProactorEventLoop._make_socket_transportF) server_sideserver_hostnamer1r2ssl_handshake_timeoutc Cs0tj||||||| d} t||| ||d| jS)N)rr1r2)r Z SSLProtocolrZ_app_transport) r-Zrawsockr/ sslcontextr0rrr1r2rZ ssl_protocolrrr_make_ssl_transportsz)BaseProactorEventLoop._make_ssl_transportcCst||||||SrB)r)r-rr/rr0r1rrr_make_datagram_transports z.BaseProactorEventLoop._make_datagram_transportcCst|||||SrB)rr-rr/r0r1rrr_make_duplex_pipe_transports z1BaseProactorEventLoop._make_duplex_pipe_transportcCst|||||SrB)rdrrrr_make_read_pipe_transportsz/BaseProactorEventLoop._make_read_pipe_transportcCst|||||SrB)rrrrr_make_write_pipe_transports z0BaseProactorEventLoop._make_write_pipe_transportcsj|rtd|rdSttkr6td|| |j d|_ d|_ t dS)Nz!Cannot close a running event loop)Z is_runningr is_closedrrrrr_stop_accept_futures_close_self_pipertrIrrrDr3rrrIs  zBaseProactorEventLoop.closecs|j||IdHSrB)rtru)r-rnrrr sock_recvszBaseProactorEventLoop.sock_recvcs|j||IdHSrB)rtZ recv_into)r-rZbufrrrsock_recv_intosz$BaseProactorEventLoop.sock_recv_intocs|j||IdHSrB)rtr)r-rrlrrr sock_sendallsz"BaseProactorEventLoop.sock_sendallcs|j||IdHSrB)rtZconnect)r-rrrrr sock_connectsz"BaseProactorEventLoop.sock_connectcs|j|IdHSrB)rtacceptrArrr sock_acceptsz!BaseProactorEventLoop.sock_acceptc s(z |}Wn2ttjfk r>}ztdW5d}~XYnXzt|j}Wn,t k r|}ztdW5d}~XYnX|r|n|}|sdSt |d}|rt |||n|} t ||}d} zLt | ||}|dkr| W0S|j ||||IdH||7}| |7} qW5| dkr"| |XdS)Nznot a regular filerl)r:AttributeErrorioUnsupportedOperationrZSendfileNotAvailableErrorosfstatst_sizerRminseekrtsendfile) r-rfileoffsetcountr:errZfsizeZ blocksizeZend_posZ total_sentrrr_sock_sendfile_natives0     z+BaseProactorEventLoop._sock_sendfile_nativecsZ|}||IdHz |j|j|||ddIdHWS||rT|XdS)NF)Zfallback)rhrirrrmZ sock_sendfiler )r-Ztransprrrrmrrr_sendfile_nativesz&BaseProactorEventLoop._sendfile_nativecCsL|jdk r|jd|_|jd|_|jd|_|jd8_dS)Nr)rrH_ssockrIr _internal_fdsrDrrrrs    z&BaseProactorEventLoop._close_self_pipecCs:t\|_|_|jd|jd|jd7_dS)NFr)r Z socketpairrrZ setblockingrrDrrrrs  z%BaseProactorEventLoop._make_self_pipec Csz4|dk r||j|k r"WdS|j|jd}Wnbtjk rLYdSttfk rdYnFt k r}z| d||dW5d}~XYnX||_| |j dS)Niz.Error on reading from the event loop self pipe)rNrOr.) rsrrtrurrrxrnrorprTry_loop_self_reading)r-rrUrrrrs$ z(BaseProactorEventLoop._loop_self_readingcCsN|j}|dkrdSz|dWn(tk rH|jrDtjdddYnXdS)Nz3Fail to write a null byte into the self-pipe socketTr)rrrR_debugr rS)r-Zcsockrrr_write_to_selfsz$BaseProactorEventLoop._write_to_selfdcs(dfdd dS)Nc s,z|dk rn|\}}jr,td||}dk rXj||dd|idnj||d|idr|WdSj}Wnt k r}zH dkrʈ d|t dnjrtjd dd W5d}~XYn8tjk rYnX|j <|dS) Nz#%r got a new connection from %r: %rTr)rr1r2rrrzAccept failed on a socket)rNrOr zAccept failed on socket %rr)rsrr rSrrrrtrrRr:rTr rrIrrxrry)rZconnrr/rUr.protocol_factoryr-r2rrrrrr./s\   z2BaseProactorEventLoop._start_serving..loop)N)r*)r-rrrr2Zbacklogrrrr_start_serving+s%z$BaseProactorEventLoop._start_servingcCsdSrBr)r-Z event_listrrr_process_eventsVsz%BaseProactorEventLoop._process_eventscCs&|jD] }|q |jdSrB)rvaluesrHclear)r-futurerrrrZs z*BaseProactorEventLoop._stop_accept_futurescCs6|j|d}|r||j||dSrB)rpopr:rHrt _stop_servingrI)r-rrrrrr_s  z#BaseProactorEventLoop._stop_serving)NNN)N)NNN)NN)NN)NN)N)NNrN)r8r^r_rrrrrrrrIrrrrrrrrrrrrrrrrcrrr3rrks\            +r)#r`__all__rrr rarrrrrrrrr r r logr rZ_FlowControlMixinZ BaseTransportrZ ReadTransportrdZWriteTransportr{rrZ TransportrrZ BaseEventLooprrrrrsR           n