B � f5d�@sdZdZddlZddlZddlZddlZddlmZddlmZddlm Z ddlm Z dd lmZdd lmZddlm Z ddlmZGd d�de je j�ZGdd�dee j�ZGdd�dee j�ZGdd�de�ZGdd�deee j�ZGdd�deee j�ZGdd�dej�ZdS)z�Event 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. )�BaseProactorEventLoop�N�)�base_events)� constants)�events)�futures)� protocols)�sslproto)� transports)�loggercs|eZdZdZd�fdd� Zdd�Zdd�Zd d �Zdd�Zd d�Z dd�Z dd�Zddd�Zdd�Z dd�Zdd�Z�ZS)�_ProactorBasePipeTransportz*Base class for pipe and socket transports.Ncs�t��||�|�|�||_|�|�||_d|_d|_d|_d|_ d|_ d|_d|_|jdk rl|j� �|j�|jj|�|dk r�|j�tj|d�dS)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)�self�loop�sock�protocol�waiter�extra�server)� __class__�� ) r$�__name__r�appendr�filenorrr�lenr�format�join)r�infor%r%r&�__repr__2s z#_ProactorBasePipeTransport.__repr__cCs||jd<dS)N�pipe)�_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|j��d|_dS)NTr) rrrrrr�_call_connection_lostr�cancel)rr%r%r&�closePs z _ProactorBasePipeTransport.closecCs,|jdk r(tjd|��t|d�|��dS)Nzunclosed transport )�source)r�warnings�warn�ResourceWarningr8)rr%r%r&�__del__[s z"_ProactorBasePipeTransport.__del__�Fatal error on pipe transportc CsVzDt|t�r*|j��rBtjd||dd�n|j�||||jd��Wd|�|�XdS)Nz%r: %sT)�exc_info)�message� exceptionZ transportr ) � isinstance�OSErrorr� get_debugr�debug�call_exception_handlerr�_force_close)r�excr@r%r%r&�_fatal_erroras z'_ProactorBasePipeTransport._fatal_errorcCs�|jdk r6|j��s6|dkr*|j�d�n|j�|�|jr@dSd|_|jd7_|jrj|j��d|_|jr�|j��d|_d|_ d|_ |j�|j |�dS)NTrr)� _empty_waiter�done� set_resultZ set_exceptionrrrr7rrrrrr6)rrHr%r%r&rGps" z'_ProactorBasePipeTransport._force_closec Cs^z|j�|�Wdt|jd�r,|j�tj�|j��d|_|j}|dk rX|� �d|_XdS)N�shutdown) rZconnection_lost�hasattrrrM�socketZ SHUT_RDWRr8rZ_detach)rrHr#r%r%r&r6�s z0_ProactorBasePipeTransport._call_connection_lostcCs"|j}|jdk r|t|j�7}|S)N)rrr-)r�sizer%r%r&�get_write_buffer_size�s z0_ProactorBasePipeTransport.get_write_buffer_size)NNN)r>)r*� __module__�__qualname__�__doc__rr1rrr4r5r8r=rIrGr6rQ� __classcell__r%r%)r$r&rs rcsTeZdZdZd�fdd� Zdd�Zdd�Zd d �Zdd�Zd d�Z ddd�Z �ZS)�_ProactorReadPipeTransportzTransport for read pipes.Ncs:d|_d|_t��||||||�|j�|j�d|_dS)NTF)� _pending_data�_pausedr rrr� _loop_reading)rrrr r!r"r#)r$r%r&r�s z#_ProactorReadPipeTransport.__init__cCs|jo|jS)N)rXr)rr%r%r&� is_reading�sz%_ProactorReadPipeTransport.is_readingcCs0|js|jrdSd|_|j��r,t�d|�dS)NTz%r pauses reading)rrXrrDrrE)rr%r%r&� pause_reading�s z(_ProactorReadPipeTransport.pause_readingcCsn|js|jsdSd|_|jdkr0|j�|jd�|j}d|_|dk rT|j�|j|�|j��rjt � d|�dS)NFz%r resumes reading)rrXrrrrYrW�_data_receivedrDrrE)r�datar%r%r&�resume_reading�s z)_ProactorReadPipeTransport.resume_readingc Csd|j��rt�d|�y|j��}Wn.tk rR}z|�|d�dSd}~XYnX|s`|��dS)Nz%r received EOFz1Fatal error: protocol.eof_received() call failed.) rrDrrErZeof_received� ExceptionrIr8)rZ keep_openrHr%r%r&� _eof_received�s z(_ProactorReadPipeTransport._eof_receivedc Cs�|jr||_dS|s |��dSt|jtj�rryt�|j|�Wq~tk rn}z|� |d�dSd}~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�|_Wn�t k r�}z0|js�|�|d�n|j��r�t jddd�Wdd}~XYn�tk r�}z|�|�Wdd}~XYnftk �r}z|�|d�Wdd}~XYn8tjk �r.|j�s*�YnX|j�sF|j�|j�Wd|dk �r^|�|�XdS)N�i�z"Fatal read error on pipe transportz*Read error on pipe transport while closingT)r?)rrK�resultr7rrXr� _proactor�recvr�ConnectionAbortedErrorrIrDrrE�ConnectionResetErrorrGrCr�CancelledError�add_done_callbackrYr\)r�futr]rHr%r%r&rY�s> z(_ProactorReadPipeTransport._loop_reading)NNN)N)r*rRrSrTrrZr[r^r`r\rYrUr%r%)r$r&rV�srVcs^eZdZdZdZ�fdd�Zdd�Zddd �Zd d�Zdd �Z dd�Z dd�Zdd�Z�Z S)�_ProactorBaseWritePipeTransportzTransport for write pipes.Tcst�j||�d|_dS)N)r rrJ)r�args�kw)r$r%r&r.sz(_ProactorBaseWritePipeTransport.__init__cCs�t|tttf�s$tdt|�j��|jr2td��|j dk rDtd��|sLdS|j rz|j tjkrht �d�|j d7_ dS|jdkr�|jt|�d�n.|js�t|�|_|��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])rB�bytes� bytearray� memoryview� TypeError�typer*r�RuntimeErrorrJrrZ!LOG_THRESHOLD_FOR_CONNLOST_WRITESr�warningr� _loop_writingr�_maybe_pause_protocol�extend)rr]r%r%r&�write2s* z%_ProactorBaseWritePipeTransport.writeNc CsTy�|dk r|jdkr|jrdSd|_d|_|r6|��|dkrJ|j}d|_|s�|jrd|j�|jd�|jrx|j � tj�|� �nN|jj�|j |�|_|j��s�t|�|_|j�|j�|��n|j�|j�|jdk r�|jdkr�|j�d�Wn\tk �r }z|�|�Wdd}~XYn0tk �rN}z|�|d�Wdd}~XYnXdS)Nrz#Fatal write error on pipe transport)rrrrbrrrr6rrrMrO�SHUT_WRZ_maybe_resume_protocolrc�sendrKr-rhrtrurJrLrfrGrCrI)r�fr]rHr%r%r&rtXs8 z-_ProactorBaseWritePipeTransport._loop_writingcCsdS)NTr%)rr%r%r&� can_write_eof�sz-_ProactorBaseWritePipeTransport.can_write_eofcCs|��dS)N)r8)rr%r%r&� write_eof�sz)_ProactorBaseWritePipeTransport.write_eofcCs|�d�dS)N)rG)rr%r%r&�abort�sz%_ProactorBaseWritePipeTransport.abortcCs:|jdk rtd��|j��|_|jdkr4|j�d�|jS)NzEmpty waiter is already set)rJrrrZ create_futurerrL)rr%r%r&�_make_empty_waiter�s z2_ProactorBaseWritePipeTransport._make_empty_waitercCs d|_dS)N)rJ)rr%r%r&�_reset_empty_waiter�sz3_ProactorBaseWritePipeTransport._reset_empty_waiter)NN)r*rRrSrTZ_start_tls_compatiblerrwrtr{r|r}r~rrUr%r%)r$r&rj(s& )rjcs$eZdZ�fdd�Zdd�Z�ZS)�_ProactorWritePipeTransportcs4t�j||�|jj�|jd�|_|j�|j�dS)N�) r rrrcrdrrrh�_pipe_closed)rrkrl)r$r%r&r�sz$_ProactorWritePipeTransport.__init__cCs@|��rdS|jrdSd|_|jdk r4|�t��n|��dS)N)Z cancelledrrrrG�BrokenPipeErrorr8)rrir%r%r&r��s z(_ProactorWritePipeTransport._pipe_closed)r*rRrSrr�rUr%r%)r$r&r��sr�c@s eZdZdZdd�Zdd�ZdS)�_ProactorDuplexPipeTransportzTransport for duplex pipes.cCsdS)NFr%)rr%r%r&r{�sz*_ProactorDuplexPipeTransport.can_write_eofcCst�dS)N)�NotImplementedError)rr%r%r&r|�sz&_ProactorDuplexPipeTransport.write_eofN)r*rRrSrTr{r|r%r%r%r&r��sr�csBeZdZdZejjZd�fdd� Zdd�Z dd�Z d d �Z�ZS)�_ProactorSocketTransportz Transport for connected sockets.Ncs$t��||||||�t�|�dS)N)r rrZ_set_nodelay)rrrr r!r"r#)r$r%r&r�sz!_ProactorSocketTransport.__init__c Cs�||jd<y|��|jd<Wn4tjtfk rP|j��rLtjd|dd�YnXd|jkr�y|� �|jd<Wn4tjtfk r�|j��r�tjd|dd�YnXdS)NrOZsocknamezgetsockname() failed on %rT)r?�peernamezgetpeername() failed on %r) r3ZgetsocknamerO�error�AttributeErrorrrDrrsZgetpeername)rrr%r%r&r�s z#_ProactorSocketTransport._set_extracCsdS)NTr%)rr%r%r&r{�sz&_ProactorSocketTransport.can_write_eofcCs2|js|jrdSd|_|jdkr.|j�tj�dS)NT)rrrrrMrOrx)rr%r%r&r|�s z"_ProactorSocketTransport.write_eof)NNN) r*rRrSrTrZ _SendfileModeZ TRY_NATIVEZ_sendfile_compatiblerrr{r|rUr%r%)r$r&r��sr�cs�eZdZ�fdd�Zd1dd�Zd2dddddd�dd �Zd3d d�Zd4dd �Zd5dd�Z�fdd�Z dd�Z dd�Zdd�Zdd�Z dd�Zdd�Zdd�Zd d!�Zd"d#�Zd6d$d%�Zd&d'�Zd7d)d*�Zd+d,�Zd-d.�Zd/d0�Z�ZS)8rcsHt��t�d|jj�||_||_d|_i|_ |� |�|��dS)NzUsing proactor: %s)r rrrEr$r*rc� _selector�_self_reading_future�_accept_futuresZset_loop�_make_self_pipe)rZproactor)r$r%r&r�s zBaseProactorEventLoop.__init__NcCst||||||�S)N)r�)rrr r!r"r#r%r%r&�_make_socket_transport�s z,BaseProactorEventLoop._make_socket_transportF)�server_side�server_hostnamer"r#�ssl_handshake_timeoutc Cs0tj||||||| d�} t||| ||d�| jS)N)r�)r"r#)r ZSSLProtocolr�Z_app_transport)rZrawsockr � sslcontextr!r�r�r"r#r�Zssl_protocolr%r%r&�_make_ssl_transport�s z)BaseProactorEventLoop._make_ssl_transportcCst|||||�S)N)r�)rrr r!r"r%r%r&�_make_duplex_pipe_transport�sz1BaseProactorEventLoop._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|��|��|j��d|_d|_t��dS)Nz!Cannot close a running event loop) Z is_runningrr� is_closed�_stop_accept_futures�_close_self_pipercr8r�r )r)r$r%r&r8s zBaseProactorEventLoop.closec�s|j�||�IdHS)N)rcrd)rr�nr%r%r&� sock_recv szBaseProactorEventLoop.sock_recvc�s|j�||�IdHS)N)rcZ recv_into)rrZbufr%r%r&�sock_recv_into#sz$BaseProactorEventLoop.sock_recv_intoc�s|j�||�IdHS)N)rcry)rrr]r%r%r&�sock_sendall&sz"BaseProactorEventLoop.sock_sendallc�s|j�||�IdHS)N)rcZconnect)rrZaddressr%r%r&�sock_connect)sz"BaseProactorEventLoop.sock_connectc�s|j�|�IdHS)N)rc�accept)rrr%r%r&�sock_accept,sz!BaseProactorEventLoop.sock_acceptc �s(y|��}Wn2ttjfk r>}zt�d��Wdd}~XYnXyt�|�j}Wn,t k r|}zt�d��Wdd}~XYnX|r�|n|}|s�dSt |d�}|r�t |||�n|} t ||�}d} zLxFt | ||�}|dkr�| S|j�||||�IdH||7}| |7} q�WWd| dk�r"|� |�XdS)Nznot a regular filerl��)r,r��io�UnsupportedOperationrZSendfileNotAvailableError�os�fstat�st_sizerC�minrc�sendfile�seek)rr�file�offset�countr,�errZfsizeZ blocksizeZend_posZ total_sentr%r%r&�_sock_sendfile_native/s2 z+BaseProactorEventLoop._sock_sendfile_nativec�sV|��}|��|��IdHz|j|j|||dd�IdHS|��|rP|��XdS)NF)Zfallback)rZr[r~Z sock_sendfilerrr^)rZtranspr�r�r�r^r%r%r&�_sendfile_nativeLsz&BaseProactorEventLoop._sendfile_nativecCsL|jdk r|j��d|_|j��d|_|j��d|_|jd8_dS)Nr)r�r7�_ssockr8�_csock� _internal_fds)rr%r%r&r�Xs z&BaseProactorEventLoop._close_self_pipecCsFt��\|_|_|j�d�|j�d�|jd7_|�|j�dS)NFr)rOZ socketpairr�r�Zsetblockingr�r�_loop_self_reading)rr%r%r&r�bs z%BaseProactorEventLoop._make_self_pipec Cs�y$|dk r|��|j�|jd�}WnHtjk r:dStk rl}z|�d||d��Wdd}~XYnX||_|� |j �dS)Niz.Error on reading from the event loop self pipe)r@rAr)rbrcrdr�rrgr_rFr�rhr�)rrzrHr%r%r&r�jsz(BaseProactorEventLoop._loop_self_readingcCs>y|j�d�Wn(tk r8|jr4tjddd�YnXdS)N�z3Fail to write a null byte into the self-pipe socketT)r?)r�ryrC�_debugrrE)rr%r%r&�_write_to_self|sz$BaseProactorEventLoop._write_to_self�dcs(d��fdd� ��dS)Nc s$y�|dk rn|��\}}�jr,t�d�||��}�dk rX�j||�dd|i��d�n�j||d|i�d��rzdS�j��}Wn~t k r�}zB�� dkr�d|�d��n�jr�tjd �dd �Wdd}~XYn8t jk �r��YnX|�j�� <|��dS)Nz#%r got a new connection from %r: %rTr�)r�r"r#r�)r"r#��zAccept failed on a socket)r@rArOzAccept failed on socket %r)r?)rbr�rrEr�r�r�rcr�rCr,rFr8rrgr�rh)rzZconnZaddrr rH)r�protocol_factoryrr#rr�r�r%r&r�s@ z2BaseProactorEventLoop._start_serving..loop)N)r)rr�rr�r#Zbacklogr�r%)rr�rr#rr�r�r&�_start_serving�s%z$BaseProactorEventLoop._start_servingcCsdS)Nr%)rZ event_listr%r%r&�_process_events�sz%BaseProactorEventLoop._process_eventscCs*x|j��D]}|��qW|j��dS)N)r��valuesr7�clear)r�futurer%r%r&r��sz*BaseProactorEventLoop._stop_accept_futurescCs6|j�|��d�}|r|��|j�|�|��dS)N)r��popr,r7rc� _stop_servingr8)rrr�r%r%r&r��s z#BaseProactorEventLoop._stop_serving)NNN)N)NN)NN)NN)N)NNr�N)r*rRrSrr�r�r�r�r�r8r�r�r�r�r�r�r�r�r�r�r�r�r�r�r�rUr%r%)r$r&r�s< )r)rT�__all__r�r�rOr:�rrrrrr r �logrZ_FlowControlMixinZ BaseTransportrZ ReadTransportrVZWriteTransportrjr�Z Transportr�r�Z BaseEventLooprr%r%r%r&�s:m '