B /g^5d@sdZdZddlZddlZddlZddlZddlmZddlmZddlm Z ddlm Z dd lm Z dd lm Z dd lm Z dd lmZGd dde je jZGdddee jZGdddee jZGdddeZGdddeee jZGdddeee jZGdddejZdS)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)events)futures) protocols)sslproto) transports)loggercs|eZdZdZdfdd ZddZddZd d Zd d Zd dZ ddZ ddZ dddZ ddZ ddZddZZS)_ProactorBasePipeTransportz*Base class for pipe and socket transports.Ncst||||||_||||_d|_d|_d|_d|_ d|_ d|_ d|_ |jdk rl|j |j|jj||dk r|jtj|ddS)NrF)super__init__ _set_extra_sock set_protocol_server_buffer _read_fut _write_fut_pending_write _conn_lost_closing _eof_writtenZ_attach_loop call_soon _protocolZconnection_maderZ_set_result_unless_cancelled)selfloopsockprotocolwaiterextraserver) __class__//usr/lib64/python3.7/asyncio/proactor_events.pyrs$     z#_ProactorBasePipeTransport.__init__cCs|jjg}|jdkr |dn|jr0|d|jdk rP|d|j|jdk rl|d|j|jdk r|d|j|jr|dt |j|j r|dd d |S) Nclosedclosingzfd=zread=zwrite=zwrite_bufsize=z EOF writtenz<{}> ) r$__name__rappendrfilenorrrlenrformatjoin)rinfor%r%r&__repr__2s         z#_ProactorBasePipeTransport.__repr__cCs||jd<dS)Npipe)_extra)rrr%r%r&rDsz%_ProactorBasePipeTransport._set_extracCs ||_dS)N)r)rr r%r%r&rGsz'_ProactorBasePipeTransport.set_protocolcCs|jS)N)r)rr%r%r& get_protocolJsz'_ProactorBasePipeTransport.get_protocolcCs|jS)N)r)rr%r%r& is_closingMsz%_ProactorBasePipeTransport.is_closingcCs\|jr dSd|_|jd7_|js>|jdkr>|j|jd|jdk rX|jd|_dS)NTr) rrrrrr_call_connection_lostrcancel)rr%r%r&closePs  z _ProactorBasePipeTransport.closecCs,|jdk r(tjd|t|d|dS)Nzunclosed transport )source)rwarningswarnResourceWarningr8)rr%r%r&__del__[s z"_ProactorBasePipeTransport.__del__Fatal error on pipe transportc CsVzDt|tr*|jrBtjd||ddn|j||||jdWd||XdS)Nz%r: %sT)exc_info)message exceptionZ transportr ) isinstanceOSErrorr get_debugr debugcall_exception_handlerr _force_close)rexcr@r%r%r& _fatal_erroras  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_exceptionrrrr7rrrrrr6)rrHr%r%r&rGps"   z'_ProactorBasePipeTransport._force_closec Cs^z|j|Wdt|jdr,|jtj|jd|_|j}|dk rX| d|_XdS)Nshutdown) rZconnection_losthasattrrrMsocketZ SHUT_RDWRr8rZ_detach)rrHr#r%r%r&r6s  z0_ProactorBasePipeTransport._call_connection_lostcCs"|j}|jdk r|t|j7}|S)N)rrr-)rsizer%r%r&get_write_buffer_sizes z0_ProactorBasePipeTransport.get_write_buffer_size)NNN)r>)r* __module__ __qualname____doc__rr1rrr4r5r8r=rIrGr6rQ __classcell__r%r%)r$r&r s  r csTeZdZdZdfdd ZddZddZd d Zd d Zd dZ dddZ Z S)_ProactorReadPipeTransportzTransport for read pipes.Ncs:d|_d|_t|||||||j|jd|_dS)NTF) _pending_data_pausedr rrr _loop_reading)rrrr r!r"r#)r$r%r&rs z#_ProactorReadPipeTransport.__init__cCs|j o|j S)N)rXr)rr%r%r& is_readingsz%_ProactorReadPipeTransport.is_readingcCs0|js |jrdSd|_|jr,td|dS)NTz%r pauses reading)rrXrrDr rE)rr%r%r& pause_readings   z(_ProactorReadPipeTransport.pause_readingcCsn|js |jsdSd|_|jdkr0|j|jd|j}d|_|dk rT|j|j||jrjt d|dS)NFz%r resumes reading) rrXrrrrYrW_data_receivedrDr rE)rdatar%r%r&resume_readings   z)_ProactorReadPipeTransport.resume_readingc Csd|jrtd|y|j}Wn.tk rR}z||ddSd}~XYnX|s`|dS)Nz%r received EOFz1Fatal error: protocol.eof_received() call failed.) rrDr rErZ eof_received ExceptionrIr8)rZ keep_openrHr%r%r& _eof_receiveds  z(_ProactorReadPipeTransport._eof_receivedc Cs|jr||_dS|s |dSt|jtjrryt|j|Wq~tk rn}z| |ddSd}~XYq~Xn |j |dS)Nz3Fatal error: protocol.buffer_updated() call failed.) rXrWr`rBrrZBufferedProtocolZ_feed_data_to_buffered_protor_rIZ data_received)rr]rHr%r%r&r\sz)_ProactorReadPipeTransport._data_receivedc Csdd}zBy`|dk r2d|_|r*|}n||jr@d}dS|dkrLdS|jsf|jj|j d|_Wnt k r}z0|js| |dn|j rt jdddWdd}~XYntk r}z||Wdd}~XYnftk r}z| |dWdd}~XYn8tjk r.|js*YnX|jsF|j|jWd|dk r^||XdS)Niz"Fatal read error on pipe transportz*Read error on pipe transport while closingT)r?)rrKresultr7rrXr _proactorrecvrConnectionAbortedErrorrIrDr rEConnectionResetErrorrGrCrCancelledErroradd_done_callbackrYr\)rfutr]rHr%r%r&rYs>   z(_ProactorReadPipeTransport._loop_reading)NNN)N) r*rRrSrTrrZr[r^r`r\rYrUr%r%)r$r&rVsrVcs^eZdZdZdZfddZddZddd Zd d Zd d Z ddZ ddZ ddZ Z S)_ProactorBaseWritePipeTransportzTransport for write pipes.Tcstj||d|_dS)N)r rrJ)rargskw)r$r%r&r.sz(_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)r])rBbytes bytearray memoryview TypeErrortyper*r RuntimeErrorrJrrZ!LOG_THRESHOLD_FOR_CONNLOST_WRITESr warningr _loop_writingr_maybe_pause_protocolextend)rr]r%r%r&write2s*       z%_ProactorBaseWritePipeTransport.writeNc CsTy|dk r|jdkr|jrdSd|_d|_|r6||dkrJ|j}d|_|s|jrd|j|jd|jrx|j t j | nN|jj|j ||_|jst||_|j|j|n|j|j|jdk r|jdkr|jdWn\tk r }z||Wdd}~XYn0tk rN}z||dWdd}~XYnXdS)Nrz#Fatal write error on pipe transport)rrrrbrrrr6rrrMrOSHUT_WRZ_maybe_resume_protocolrcsendrKr-rhrtrurJrLrfrGrCrI)rfr]rHr%r%r&rtXs8    z-_ProactorBaseWritePipeTransport._loop_writingcCsdS)NTr%)rr%r%r& can_write_eofsz-_ProactorBaseWritePipeTransport.can_write_eofcCs |dS)N)r8)rr%r%r& write_eofsz)_ProactorBaseWritePipeTransport.write_eofcCs|ddS)N)rG)rr%r%r&abortsz%_ProactorBaseWritePipeTransport.abortcCs:|jdk rtd|j|_|jdkr4|jd|jS)NzEmpty waiter is already set)rJrrrZ create_futurerrL)rr%r%r&_make_empty_waiters     z2_ProactorBaseWritePipeTransport._make_empty_waitercCs d|_dS)N)rJ)rr%r%r&_reset_empty_waitersz3_ProactorBaseWritePipeTransport._reset_empty_waiter)NN)r*rRrSrTZ_start_tls_compatiblerrwrtr{r|r}r~rrUr%r%)r$r&rj(s & )rjcs$eZdZfddZddZZS)_ProactorWritePipeTransportcs4tj|||jj|jd|_|j|jdS)N) r rrrcrdrrrh _pipe_closed)rrkrl)r$r%r&rsz$_ProactorWritePipeTransport.__init__cCs@|r dS|jrdSd|_|jdk r4|tn|dS)N)Z cancelledrrrrGBrokenPipeErrorr8)rrir%r%r&rs z(_ProactorWritePipeTransport._pipe_closed)r*rRrSrrrUr%r%)r$r&rs rc@s eZdZdZddZddZdS)_ProactorDuplexPipeTransportzTransport for duplex pipes.cCsdS)NFr%)rr%r%r&r{sz*_ProactorDuplexPipeTransport.can_write_eofcCstdS)N)NotImplementedError)rr%r%r&r|sz&_ProactorDuplexPipeTransport.write_eofN)r*rRrSrTr{r|r%r%r%r&rsrcsBeZdZdZejjZd fdd ZddZ ddZ d d Z Z S) _ProactorSocketTransportz Transport for connected sockets.Ncs$t||||||t|dS)N)r rrZ _set_nodelay)rrrr r!r"r#)r$r%r&rsz!_ProactorSocketTransport.__init__c Cs||jd<y||jd<Wn4tjtfk rP|jrLtjd|ddYnXd|jkry| |jd<Wn4tjtfk r|jrtjd|ddYnXdS)NrOZsocknamezgetsockname() failed on %rT)r?peernamezgetpeername() failed on %r) r3Z getsocknamerOerrorAttributeErrorrrDr rsZ getpeername)rrr%r%r&rs    z#_ProactorSocketTransport._set_extracCsdS)NTr%)rr%r%r&r{sz&_ProactorSocketTransport.can_write_eofcCs2|js |jrdSd|_|jdkr.|jtjdS)NT)rrrrrMrOrx)rr%r%r&r|s   z"_ProactorSocketTransport.write_eof)NNN) r*rRrSrTrZ _SendfileModeZ TRY_NATIVEZ_sendfile_compatiblerrr{r|rUr%r%)r$r&rsrcseZdZfddZd1ddZd2dddddddd Zd3d d Zd4d d Zd5ddZfddZ ddZ ddZ ddZ ddZ ddZddZddZd d!Zd"d#Zd6d$d%Zd&d'Zd7d)d*Zd+d,Zd-d.Zd/d0ZZS)8rcsHttd|jj||_||_d|_i|_ | || dS)NzUsing proactor: %s) r rr rEr$r*rc _selector_self_reading_future_accept_futuresZset_loop_make_self_pipe)rZproactor)r$r%r&rs  zBaseProactorEventLoop.__init__NcCst||||||S)N)r)rrr r!r"r#r%r%r&_make_socket_transports z,BaseProactorEventLoop._make_socket_transportF) server_sideserver_hostnamer"r#ssl_handshake_timeoutc Cs0tj||||||| d} t||| ||d| jS)N)r)r"r#)r Z SSLProtocolrZ_app_transport) rZrawsockr sslcontextr!rrr"r#rZ ssl_protocolr%r%r&_make_ssl_transports z)BaseProactorEventLoop._make_ssl_transportcCst|||||S)N)r)rrr r!r"r%r%r&_make_duplex_pipe_transportsz1BaseProactorEventLoop._make_duplex_pipe_transportcCst|||||S)N)rV)rrr r!r"r%r%r&_make_read_pipe_transportsz/BaseProactorEventLoop._make_read_pipe_transportcCst|||||S)N)r)rrr r!r"r%r%r&_make_write_pipe_transportsz0BaseProactorEventLoop._make_write_pipe_transportcsP|rtd|rdS|||jd|_d|_tdS)Nz!Cannot close a running event loop) Z is_runningrr is_closed_stop_accept_futures_close_self_pipercr8rr )r)r$r%r&r8s zBaseProactorEventLoop.closecs|j||IdHS)N)rcrd)rrnr%r%r& sock_recv szBaseProactorEventLoop.sock_recvcs|j||IdHS)N)rcZ recv_into)rrZbufr%r%r&sock_recv_into#sz$BaseProactorEventLoop.sock_recv_intocs|j||IdHS)N)rcry)rrr]r%r%r& sock_sendall&sz"BaseProactorEventLoop.sock_sendallcs|j||IdHS)N)rcZconnect)rrZaddressr%r%r& sock_connect)sz"BaseProactorEventLoop.sock_connectcs|j|IdHS)N)rcaccept)rrr%r%r& sock_accept,sz!BaseProactorEventLoop.sock_acceptc s(y |}Wn2ttjfk r>}ztdWdd}~XYnXyt|j}Wn,t k r|}ztdWdd}~XYnX|r|n|}|sdSt |d}|rt |||n|} t ||}d} zLxFt | ||}|dkr| S|j ||||IdH||7}| |7} qWWd| dkr"| |XdS)Nznot a regular filerl)r,rioUnsupportedOperationrZSendfileNotAvailableErrorosfstatst_sizerCminrcsendfileseek) rrfileoffsetcountr,errZfsizeZ blocksizeZend_posZ total_sentr%r%r&_sock_sendfile_native/s2     z+BaseProactorEventLoop._sock_sendfile_nativecsV|}||IdHz|j|j|||ddIdHS||rP|XdS)NF)Zfallback)rZr[r~Z sock_sendfilerrr^)rZtransprrrr^r%r%r&_sendfile_nativeLsz&BaseProactorEventLoop._sendfile_nativecCsL|jdk r|jd|_|jd|_|jd|_|jd8_dS)Nr)rr7_ssockr8_csock _internal_fds)rr%r%r&rXs    z&BaseProactorEventLoop._close_self_pipecCsFt\|_|_|jd|jd|jd7_||jdS)NFr)rOZ socketpairrrZ setblockingrr_loop_self_reading)rr%r%r&rbs   z%BaseProactorEventLoop._make_self_pipec Csy$|dk r||j|jd}WnHtjk r:dStk rl}z|d||dWdd}~XYnX||_| |j dS)Niz.Error on reading from the event loop self pipe)r@rAr) rbrcrdrrrgr_rFrrhr)rrzrHr%r%r&rjsz(BaseProactorEventLoop._loop_self_readingcCs>y|jdWn(tk r8|jr4tjdddYnXdS)Nz3Fail to write a null byte into the self-pipe socketT)r?)rryrC_debugr rE)rr%r%r&_write_to_self|s z$BaseProactorEventLoop._write_to_selfdcs(dfdd dS)Nc s$y|dk rn|\}}jr,td||}dk rXj||dd|idnj||d|idrzdSj}Wn~t k r}zB dkrˆ d|d njrtjd dd Wdd}~XYn8t jk r YnX|j <|dS) Nz#%r got a new connection from %r: %rTr)rr"r#r)r"r#zAccept failed on a socket)r@rArOzAccept failed on socket %r)r?)rbrr rErrrrcrrCr,rFr8rrgrrh)rzZconnZaddrr rH)rprotocol_factoryrr#rrrr%r&rs@      z2BaseProactorEventLoop._start_serving..loop)N)r)rrrrr#Zbacklogrr%)rrrr#rrrr&_start_servings%z$BaseProactorEventLoop._start_servingcCsdS)Nr%)rZ event_listr%r%r&_process_eventssz%BaseProactorEventLoop._process_eventscCs*x|jD] }|q W|jdS)N)rvaluesr7clear)rfuturer%r%r&rs z*BaseProactorEventLoop._stop_accept_futurescCs6|j|d}|r||j||dS)N)rpopr,r7rc _stop_servingr8)rrrr%r%r&rs  z#BaseProactorEventLoop._stop_serving)NNN)N)NN)NN)NN)N)NNrN)r*rRrSrrrrrrr8rrrrrrrrrrrrrrrrUr%r%)r$r&rs<            )r)rT__all__rrrOr:rrrrrr r logr Z_FlowControlMixinZ BaseTransportr Z ReadTransportrVZWriteTransportrjrZ TransportrrZ BaseEventLooprr%r%r%r&s:          m  '