U i�f<}�@sPdZddlZddlZddlZddlZddlZddlZddlZddlm Z ddlm Z ddlmZddlmZddlm Z dd lmZdd lmZddlmZddlmZd d�ZGdd�dejej�ZGdd�deej�ZGdd�deej�ZGdd�de�ZGdd�de�ZGdd�deeej�ZGdd�deeej�ZGdd�de j �Z!dS))�BaseProactorEventLoop�N�)�base_events)� constants)�futures)� exceptions)� protocols)�sslproto)� transports)�trsock)�loggercCs�t�|�|jd<z|��|jd<Wn0tjk rR|j��rNtj d|dd�YnXd|jkr�z|� �|jd<Wn tjk r�d|jd<YnXdS)N�socketZsocknamezgetsockname() failed on %rT��exc_info�peername)r�TransportSocket�_extraZgetsocknamer �error�_loop� get_debugr�warningZgetpeername)� transport�sock�r�� ) r4�__name__r �appendr(�filenor$r%r#�lenr)�format�join)r-�inforrr�__repr__Hs z#_ProactorBasePipeTransport.__repr__cCs||jd<dS)N�pipe)r�r-rrrrrZsz%_ProactorBasePipeTransport._set_extracCs ||_dS�N�r+)r-r/rrrr!]sz'_ProactorBasePipeTransport.set_protocolcCs|jSrBrC�r-rrr�get_protocol`sz'_ProactorBasePipeTransport.get_protocolcCs|jSrB)r(rDrrr� is_closingcsz%_ProactorBasePipeTransport.is_closingcCs\|jr dSd|_|jd7_|js>|jdkr>|j�|jd�|jdk rX|j��d|_dS)NTr) r(r'r#r%rr*�_call_connection_lostr$�cancelrDrrr�closefs 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|t�r*|j��rBtjd||dd�n|j�||||jd��W5|�|�XdS)Nz%r: %sTr)�message� exceptionrr/) �_force_close� isinstance�OSErrorrrr�debug�call_exception_handlerr+)r-�excrNrrr�_fatal_errorvs �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_exceptionr(r'r%rHr$r&r#rr*rG)r-rUrrrrP�s" z'_ProactorBasePipeTransport._force_closec Cs^z|j� |�W5t|jd�r,|j�tj�|j��d|_|j}|dk rX|��d|_XdS)N�shutdown) �hasattrr rZr Z SHUT_RDWRrIr"Z_detachr+Zconnection_lost)r-rUr2rrrrG�s z0_ProactorBasePipeTransport._call_connection_lostcCs"|j}|jdk r|t|j�7}|SrB)r&r#r;)r-�sizerrr�get_write_buffer_size�s z0_ProactorBasePipeTransport.get_write_buffer_size)NNN)rM)r8� __module__�__qualname__rr?rr!rErFrI�warnings�warnrLrVrPrGr]� __classcell__rrr3rr.s� rcsPeZdZd�fdd� Zdd�Zdd�Zdd �Zd d�Zdd �Zddd�Z �Z S)�_ProactorReadPipeTransportNcs:d|_d|_t��||||||�|j�|j�d|_dS)NTF)� _pending_data�_pausedrrrr*� _loop_readingr,r3rrr�s z#_ProactorReadPipeTransport.__init__cCs|jo|jSrB)rer(rDrrr� is_reading�sz%_ProactorReadPipeTransport.is_readingcCs0|js|jrdSd|_|j��r,t�d|�dS)NTz%r pauses reading)r(rerrrrSrDrrr� 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)r(rer$rr*rfrd�_data_receivedrrrS�r-�datarrr�resume_reading�s z)_ProactorReadPipeTransport.resume_readingc Cs�|j��rt�d|�z|j��}WnLttfk r>�Yn4tk rp}z|� |d�WY�dSd}~XYnX|s~|� �dS)Nz%r received EOFz1Fatal error: protocol.eof_received() call failed.)rrrrSr+Zeof_received� SystemExit�KeyboardInterrupt� BaseExceptionrVrI)r-Z keep_openrUrrr� _eof_received�s �z(_ProactorReadPipeTransport._eof_receivedc Cs�|jr||_dS|s |��dSt|jtj�r�zt�|j|�Wq�tt fk rZ�Yq�t k r�}z|�|d�WY�dSd}~XYq�Xn|j�|�dS)Nz3Fatal error: protocol.buffer_updated() call failed.) rerdrprQr+rZBufferedProtocolZ_feed_data_to_buffered_protormrnrorVZ data_received)r-rkrUrrrri�s"�z)_ProactorReadPipeTransport._data_receivedc Cstd}�zRzp|dk r2d|_|��r*|��}n|��|jrHd}WW��dS|dkr\WW��dS|jsv|jj� |j d�|_Wn�tk r�}z0|js�|�|d�n|j� �r�tjddd�W5d}~XYn�tk r�}z|�|�W5d}~XYnftk �r}z|�|d�W5d}~XYn8tjk �r>|j�s:�YnX|j�sV|j�|j�W5|dk �rn|�|�XdS)N�i�z"Fatal read error on pipe transportz*Read error on pipe transport while closingTr)rir$rX�resultrHr(rer� _proactor�recvr �ConnectionAbortedErrorrVrrrS�ConnectionResetErrorrPrRr�CancelledError�add_done_callbackrf)r-�futrkrUrrrrfs@ � z(_ProactorReadPipeTransport._loop_reading)NNN)N)r8r^r_rrgrhrlrprirfrbrrr3rrc�s� rccsZeZdZdZ�fdd�Zdd�Zddd�Zd d �Zdd�Zd d�Z dd�Z dd�Z�ZS)�_ProactorBaseWritePipeTransportTcst�j||�d|_dSrB)rrrW�r-�args�kwr3rrrGsz(_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)rk)rQ�bytes� bytearray� memoryview� TypeError�typer8r)�RuntimeErrorrWr'r�!LOG_THRESHOLD_FOR_CONNLOST_WRITESrrr%� _loop_writingr#�_maybe_pause_protocol�extendrjrrr�writeKs,� z%_ProactorBaseWritePipeTransport.writeNc CsVz�|dk r |jdkr |jr WdSd|_d|_|r8|��|dkrL|j}d|_|s�|jrf|j�|jd�|jrz|j � tj�|� �nN|jj�|j |�|_|j��s�t|�|_|j�|j�|��n|j�|j�|jdk r�|jdkr�|j�d�Wn\tk �r"}z|�|�W5d}~XYn0tk �rP}z|�|d�W5d}~XYnXdS)Nrz#Fatal write error on pipe transport)r%r(r&rrr#rr*rGr)r rZr �SHUT_WR�_maybe_resume_protocolrs�sendrXr;rxr�r�rWrYrvrPrRrV)r-�frkrUrrrr�qs8 z-_ProactorBaseWritePipeTransport._loop_writingcCsdS�NTrrDrrr� can_write_eof�sz-_ProactorBaseWritePipeTransport.can_write_eofcCs|��dSrB)rIrDrrr� write_eof�sz)_ProactorBaseWritePipeTransport.write_eofcCs|�d�dSrB�rPrDrrr�abort�sz%_ProactorBaseWritePipeTransport.abortcCs:|jdk rtd��|j��|_|jdkr4|j�d�|jS)NzEmpty waiter is already set)rWr�rZ create_futurer%rYrDrrr�_make_empty_waiter�s z2_ProactorBaseWritePipeTransport._make_empty_waitercCs d|_dSrB)rWrDrrr�_reset_empty_waiter�sz3_ProactorBaseWritePipeTransport._reset_empty_waiter)NN) r8r^r_Z_start_tls_compatiblerr�r�r�r�r�r�r�rbrrr3rrzAs& )rzcs$eZdZ�fdd�Zdd�Z�ZS)�_ProactorWritePipeTransportcs4t�j||�|jj�|jd�|_|j�|j�dS)N�) rrrrsrtr r$rx�_pipe_closedr{r3rrr�sz$_ProactorWritePipeTransport.__init__cCs@|��rdS|jrdSd|_|jdk r4|�t��n|��dSrB)Z cancelledr(r$r%rP�BrokenPipeErrorrI)r-ryrrrr��s z(_ProactorWritePipeTransport._pipe_closed)r8r^r_rr�rbrrr3rr��sr�csXeZdZdZd�fdd� Zdd�Zdd�Zd d �Zddd�Zdd d�Z ddd�Z �ZS)�_ProactorDatagramTransportiNcs>||_d|_t�j|||||d�t��|_|j�|j �dS)N)r0r1) �_addressrWrr�collections�dequer#rr*rf)r-r.rr/�addressr0r1r3rrr�s z#_ProactorDatagramTransport.__init__cCst||�dSrB�rrArrrr�sz%_ProactorDatagramTransport._set_extracCstdd�|jD��S)Ncss|]\}}t|�VqdSrB)r;)�.0rk�_rrr� �szC_ProactorDatagramTransport.get_write_buffer_size..)�sumr#rDrrrr]�sz0_ProactorDatagramTransport.get_write_buffer_sizecCs|�d�dSrBr�rDrrrr��sz _ProactorDatagramTransport.abortcCs�t|tttf�stdt|��|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)rQr~rr�r�r�r�� ValueErrorr'rr�rrr#r9r%r�r�)r-rk�addrrrr�sendto�s&� � z!_ProactorDatagramTransport.sendtoc Csz�|jrWdSd|_|r |��|jr2|jrN|jrN|jrH|j�|jd�WdS|j� �\}}|jdk r||jj �|j|�|_n|jj j |j||d�|_WnZtk r�}z|j�|�W5d}~XYnDtk r�}z|�|d�W5d}~XYnX|j�|j�|��dS)N)r�z'Fatal write error on datagram transport)r'r%rrr#r�r(rr*rG�popleftrsr�r r�rRr+�error_received� ExceptionrVrxr�r�)r-ryrkr�rUrrrr��s2 ��z(_ProactorDatagramTransport._loop_writingc Cs4d}�zz�|jrWW��dSd|_|dk rf|��}|jrFd}WW��dS|jdk r^||j}}n|\}}|jrvWW��dS|jdk r�|jj� |j |j�|_n|jj�|j |j�|_WnJt k r�}z|j�|�W5d}~XYn8tjk r�|js��YnX|jdk �r|j�|j�W5|�r.|j�||�XdSrB)r+Zdatagram_receivedr'r$rrr(r�rrsrtr �max_sizeZrecvfromrRr�rrwrxrf)r-ryrkr��resrUrrrrfs> �� z(_ProactorDatagramTransport._loop_reading)NNN)N)N)N)r8r^r_r�rrr]r�r�r�rfrbrrr3rr��s� !r�c@seZdZdd�Zdd�ZdS)�_ProactorDuplexPipeTransportcCsdS)NFrrDrrrr�Jsz*_ProactorDuplexPipeTransport.can_write_eofcCst�dSrB)�NotImplementedErrorrDrrrr�Msz&_ProactorDuplexPipeTransport.write_eofN)r8r^r_r�r�rrrrr�Esr�cs>eZdZejjZd �fdd� Zdd�Zdd�Z dd �Z �ZS)�_ProactorSocketTransportNcs$t��||||||�t�|�dSrB)rrrZ_set_nodelayr,r3rrrXsz!_ProactorSocketTransport.__init__cCst||�dSrBr�rArrrr]sz#_ProactorSocketTransport._set_extracCsdSr�rrDrrrr�`sz&_ProactorSocketTransport.can_write_eofcCs2|js|jrdSd|_|jdkr.|j�tj�dSr�)r(r)r%r rZr r�rDrrrr�cs z"_ProactorSocketTransport.write_eof)NNN)r8r^r_rZ _SendfileModeZ TRY_NATIVEZ_sendfile_compatiblerrr�r�rbrrr3rr�Qs�r�cs�eZdZ�fdd�Zd3dd�Zd4dddddd�dd �Zd5d d�Zd6dd �Zd7dd�Zd8dd�Z �fdd�Z dd�Zdd�Zdd�Z dd�Zdd�Zdd�Zd d!�Zd"d#�Zd$d%�Zd9d&d'�Zd(d)�Zd:d+d,�Zd-d.�Zd/d0�Zd1d2�Z�ZS);rcsht��t�d|jj�||_||_d|_i|_ |� |�|��t� �t��krdt�|j��dS)NzUsing proactor: %s)rrrrSr4r8rs� _selector�_self_reading_future�_accept_futuresZset_loop�_make_self_pipe� threading�current_thread�main_thread�signal� set_wakeup_fd�_csockr:)r-Zproactorr3rrrms zBaseProactorEventLoop.__init__NcCst||||||�SrB)r�)r-rr/r0r1r2rrr�_make_socket_transportzs �z,BaseProactorEventLoop._make_socket_transportF)�server_side�server_hostnamer1r2�ssl_handshake_timeoutc Cs0tj||||||| d�} t||| ||d�| jS)N)r��r1r2)r ZSSLProtocolr�Z_app_transport)r-Zrawsockr/� sslcontextr0r�r�r1r2r�Zssl_protocolrrr�_make_ssl_transports��z)BaseProactorEventLoop._make_ssl_transportcCst||||||�SrB)r�)r-rr/r�r0r1rrr�_make_datagram_transport�s �z.BaseProactorEventLoop._make_datagram_transportcCst|||||�SrB)r��r-rr/r0r1rrr�_make_duplex_pipe_transport�s�z1BaseProactorEventLoop._make_duplex_pipe_transportcCst|||||�SrB)rcr�rrr�_make_read_pipe_transport�sz/BaseProactorEventLoop._make_read_pipe_transportcCst|||||�SrB)r�r�rrr�_make_write_pipe_transport�s�z0BaseProactorEventLoop._make_write_pipe_transportcsj|��rtd��|��rdSt��t��kr6t�d�|��|� �|j ��d|_ d|_t ��dS)Nz!Cannot close a running event loop��)Z is_runningr�� is_closedr�r�r�r�r��_stop_accept_futures�_close_self_pipersrIr�rrDr3rrrI�s zBaseProactorEventLoop.closec�s|j�||�IdHSrB)rsrt)r-r�nrrr� sock_recv�szBaseProactorEventLoop.sock_recvc�s|j�||�IdHSrB)rsZ recv_into)r-rZbufrrr�sock_recv_into�sz$BaseProactorEventLoop.sock_recv_intoc�s|j�||�IdHSrB)rsr�)r-rrkrrr�sock_sendall�sz"BaseProactorEventLoop.sock_sendallc�s|j�||�IdHSrB)rsZconnect)r-rr�rrr�sock_connect�sz"BaseProactorEventLoop.sock_connectc�s|j�|�IdHSrB)rs�acceptrArrr�sock_accept�sz!BaseProactorEventLoop.sock_acceptc �s(z|��}Wn2ttjfk r>}zt�d��W5d}~XYnXzt�|�j}Wn,t k r|}zt�d��W5d}~XYnX|r�|n|}|s�dSt |d�}|r�t |||�n|} t ||�}d} zLt | ||�}|dkr�| W�0S|j� ||||�IdH||7}| |7} q�W5| dk�r"|�|�XdS)Nznot a regular filerl��)r:�AttributeError�io�UnsupportedOperationrZSendfileNotAvailableError�os�fstat�st_sizerR�min�seekrs�sendfile)r-r�file�offset�countr:�errZfsizeZ blocksizeZend_posZ total_sentrrr�_sock_sendfile_native�s0 z+BaseProactorEventLoop._sock_sendfile_nativec�sZ|��}|��|��IdHz |j|j|||dd�IdHW�S|��|rT|��XdS)NF)Zfallback)rgrhr�r�rlZ sock_sendfiler )r-Ztranspr�r�r�rlrrr�_sendfile_native�s�z&BaseProactorEventLoop._sendfile_nativecCsL|jdk r|j��d|_|j��d|_|j��d|_|jd8_dS)Nr)r�rH�_ssockrIr�� _internal_fdsrDrrrr��s z&BaseProactorEventLoop._close_self_pipecCs:t��\|_|_|j�d�|j�d�|jd7_dS)NFr)r Z socketpairr�r�Zsetblockingr�rDrrrr��sz%BaseProactorEventLoop._make_self_pipec Cs�z4|dk r|��|j|k r"WdS|j�|jd�}Wnbtjk rLYdSttfk rd�YnFt k r�}z|� d||d��W5d}~XYnX||_|�|j�dS)Niz.Error on reading from the event loop self pipe)rNrOr.) rrr�rsrtr�rrwrmrnrorTrx�_loop_self_reading)r-r�rUrrrr��s$ �z(BaseProactorEventLoop._loop_self_readingcCsN|j}|dkrdSz|�d�Wn(tk rH|jrDtjddd�YnXdS)N�z3Fail to write a null byte into the self-pipe socketTr)r�r�rR�_debugrrS)r-Zcsockrrr�_write_to_selfs�z$BaseProactorEventLoop._write_to_self�dcs(d��fdd� ��dS)Nc s,z�|dk rn|��\}}�jr,t�d�||��}�dk rX�j||�dd|i��d�n�j||d|i�d��r|WdS�j��}Wn�t k r�}zH�� dkrʈ�d|t� ��d��n�jr�tjd �dd �W5d}~XYn8tjk �r��YnX|�j�� <|��dS)Nz#%r got a new connection from %r: %rTr)r�r1r2r�r�r�zAccept failed on a socket)rNrOr zAccept failed on socket %rr)rrr�rrSr�r�r�rsr�rRr:rTrrrIrrwr�rx)r�Zconnr�r/rU�r.�protocol_factoryr-r2rr�r�rrr./s\�� z2BaseProactorEventLoop._start_serving..loop)N)r*)r-r�rr�r2Zbacklogr�rr�r�_start_serving+s%z$BaseProactorEventLoop._start_servingcCsdSrBr)r-Z event_listrrr�_process_eventsVsz%BaseProactorEventLoop._process_eventscCs&|j��D]}|��q |j��dSrB)r��valuesrH�clear)r-�futurerrrr�Zs z*BaseProactorEventLoop._stop_accept_futurescCs6|j�|��d�}|r|��|j�|�|��dSrB)r��popr:rHrs� _stop_servingrI)r-rr�rrrr�_s z#BaseProactorEventLoop._stop_serving)NNN)N)NNN)NN)NN)NN)N)NNr�N)r8r^r_rr�r�r�r�r�r�rIr�r�r�r�r�r�r�r�r�r�r�r�r�r�r�rbrrr3rrks\ � �� +r)"�__all__r�r�r r`r�r�r��rrrrrr r r�logrrZ_FlowControlMixinZ BaseTransportrZ ReadTransportrcZWriteTransportrzr�r�Z Transportr�r�Z BaseEventLooprrrrr�sP��n��