� j f�� @s�dZdgZddlZddlZddlZddlZddlZyddlZWnek rudZYnXddl m Z ddl mZddl mZddl m Z dd l mZdd l mZddl mZddl mZdd lmZddlmZdd�ZGdd�de j�ZGdd�dejej�ZGdd�de�ZGdd�de�ZGdd�de�ZdS)z�Event loop using a selector and related classes. A selector is a "notify-when-ready" multiplexer. For a subclass which also includes support for signal handling, see the unix_events sub-module. �BaseSelectorEventLoop�N�)�base_events)�compat)� constants)�events)�futures)� selectors)� transports)�sslproto)� coroutine)�loggercCsAy|j|�}Wntk r+dSYnXt|j|@�SdS)NF)�get_key�KeyError�boolr)�selector�fdZevent�key�r�d?�Z d@dA�Z!dBdC�Z"dDdE�Z#�S)FrzJSelector event loop. See events.EventLoop for API specification. NcsUt�j�|dkr(tj�}ntjd|jj�||_|j �dS)NzUsing selector: %s) �super�__init__r ZDefaultSelectorr �debug� __class__�__name__� _selector�_make_self_pipe)�selfr)rrrr0s zBaseSelectorEventLoop.__init__�extra�servercCst||||||�S)N)�_SelectorSocketTransport)r�sock�protocol�waiterrr rrr�_make_socket_transport9sz,BaseSelectorEventLoop._make_socket_transport�server_sideF�server_hostnamec Cs{tj�s:|j||||d|d|d|d|�Stj||||||�} t||| d|d|�| jS)Nr&r'rr )rZ_is_sslproto_available�_make_legacy_ssl_transportZSSLProtocolr!Z_app_transport) r�rawsockr#� sslcontextr$r&r'rr Zssl_protocolrrr�_make_ssl_transport>s z)BaseSelectorEventLoop._make_ssl_transportc Cs"t|||||||||� S)N)�_SelectorSslTransport) rr)r#r*r$r&r'rr rrrr(Msz0BaseSelectorEventLoop._make_legacy_ssl_transportcCst||||||�S)N)�_SelectorDatagramTransport)rr"r#�addressr$rrrr�_make_datagram_transportWsz.BaseSelectorEventLoop._make_datagram_transportcsn|j�rtd��n|j�r+dS|j�t�j�|jdk rj|jj�d|_ndS)Nz!Cannot close a running event loop)Z is_running�RuntimeError� is_closed�_close_self_piper�closer)r)rrrr3\s zBaseSelectorEventLoop.closecCs t�dS)N)�NotImplementedError)rrrr�_socketpairgsz!BaseSelectorEventLoop._socketpaircCsU|j|jj��|jj�d|_|jj�d|_|jd8_dS)Nr)� remove_reader�_ssock�filenor3�_csock� _internal_fds)rrrrr2js z&BaseSelectorEventLoop._close_self_pipecCsg|j�\|_|_|jjd�|jjd�|jd7_|j|jj�|j�dS)NFr)r5r7r9�setblockingr:� add_readerr8�_read_from_self)rrrrrrs z%BaseSelectorEventLoop._make_self_pipecCsdS)Nr)r�datarrr�_process_self_datazsz(BaseSelectorEventLoop._process_self_datacCsbx[y-|jjd�}|s"Pn|j|�Wqtk rGwYqtk rYPYqXqWdS)Ni)r7�recvr?�InterruptedError�BlockingIOError)rr>rrrr=}s z%BaseSelectorEventLoop._read_from_selfcCsa|j}|dk r]y|jd�Wq]tk rY|jrUtjddd�nYq]XndS)Nsz3Fail to write a null byte into the self-pipe socket�exc_infoT)r9�send�OSError�_debugr r)rZcsockrrr�_write_to_self�s z$BaseSelectorEventLoop._write_to_selfcCs)|j|j�|j||||�dS)N)r<r8�_accept_connection)r�protocol_factoryr"r*r rrr�_start_serving�sz$BaseSelectorEventLoop._start_servingc CsDyE|j�\}}|jr7tjd|||�n|jd�Wn�tttfk rbYn�tk r }z�|j t j t jt jt j fkr�|jidd6|d6|d6�|j|j��|jtj|j||||�n�WYdd}~Xn6Xi|d6}|j|||||�} |j| �dS)Nz#%r got a new connection from %r: %rFz&socket.accept() out of system resource�message� exception�socket�peername)�acceptrFr rr;rBrA�ConnectionAbortedErrorrE�errnoZEMFILEZENFILEZENOBUFSZENOMEM�call_exception_handlerr6r8Z call_laterrZACCEPT_RETRY_DELAYrJ�_accept_connection2Zcreate_task) rrIr"r*r �conn�addr�excrrOrrrrH�s0 z(BaseSelectorEventLoop._accept_connectionccs5d}d}y�|�}tjd|�}|r`|j|||d|ddd|d|�}n$|j||d|d|d|�}y |DdHWn|j��YnXWn�tk r0} zd|jridd6| d 6} |dk r�|| d �modify�cancel) rr�callback�args�handler�mask�reader�writerrrrr<�s z BaseSelectorEventLoop.add_readerc Cs�|j�rdSy|jj|�}Wntk r>dSYn{X|j|j}\}}|tjM}|s|jj|�n|jj ||d|f�|dk r�|j �dSdSdS)zRemove a reader callback.FNT)r1rrrrr>r r_� unregisterr`ra)rrrrerfrgrrrr6�s z#BaseSelectorEventLoop.remove_readerc Gs�|j�tj|||�}y|jj|�}Wn1tk rh|jj|tjd|f�YnVX|j|j }\}}|jj ||tjB||f�|dk r�|j�ndS)zAdd a writer callback..N)r\rr]rrrr^r �EVENT_WRITEr>r`ra) rrrbrcrdrrerfrgrrr� add_writers z BaseSelectorEventLoop.add_writerc Cs�|j�rdSy|jj|�}Wntk r>dSYn{X|j|j}\}}|tjM}|s|jj|�n|jj |||df�|dk r�|j �dSdSdS)zRemove a writer callback.FNT)r1rrrrr>r rirhr`ra)rrrrerfrgrrr� remove_writers z#BaseSelectorEventLoop.remove_writercCsV|jr*|j�dkr*td��ntjd|�}|j|d||�|S)z�Receive data from the socket. The return value is a bytes object representing the data received. The maximum amount of data to be received at once is specified by nbytes. This method is a coroutine. rzthe socket must be non-blockingrWF)rF� gettimeout� ValueErrorrrY� _sock_recv)rr"�n�futrrr� sock_recv2s zBaseSelectorEventLoop.sock_recvcCs�|j�}|r"|j|�n|j�r2dSy|j|�}Wnhttfk r~|j||j|d||�Yn?tk r�}z|j |�WYdd}~XnX|j |�dS)NT)r8r6� cancelledr@rBrAr<rnrZ� set_exception� set_result)rrp� registeredr"rorr>rVrrrrnAs# z BaseSelectorEventLoop._sock_recvcCsl|jr*|j�dkr*td��ntjd|�}|r[|j|d||�n |jd�|S)a�Send data to the socket. The socket must be connected to a remote socket. This method continues to send data from data until either all data has been sent or an error occurs. None is returned on success. On error, an exception is raised, and there is no way to determine how much data, if any, was successfully processed by the receiving end of the connection. This method is a coroutine. rzthe socket must be non-blockingrWFN)rFrlrmrrY� _sock_sendallrt)rr"r>rprrr�sock_sendallVs z"BaseSelectorEventLoop.sock_sendallcCs�|j�}|r"|j|�n|j�r2dSy|j|�}WnSttfk red}Yn6tk r�}z|j|�dSWYdd}~XnX|t|�kr�|j d�n8|r�||d�}n|j ||j|d||�dS)NrT)r8rkrrrDrBrArZrs�lenrtrjrv)rrprur"r>rrorVrrrrvjs" z#BaseSelectorEventLoop._sock_sendallcCs�|jr*|j�dkr*td��ntjd|�}ytj||�Wn2tk r�}z|j|�WYdd}~XnX|j|||�|S)a�Connect to a remote socket at address. The address must be already resolved to avoid the trap of hanging the entire event loop when the address requires doing a DNS lookup. For example, it must be an IP address, not an hostname, for AF_INET and AF_INET6 address families. Use getaddrinfo() to resolve the hostname asynchronously. This method is a coroutine. rzthe socket must be non-blockingrWN) rFrlrmrrYrZ_check_resolved_addressrs� _sock_connect)rr"r.rp�errrrr�sock_connect�s z"BaseSelectorEventLoop.sock_connectcCs�|j�}y|j|�Wn�ttfk ro|jtj|j|��|j||j |||�Yn?t k r�}z|j|�WYdd}~XnX|jd�dS)N) r8ZconnectrBrAZadd_done_callback� functools�partial�_sock_connect_donerj�_sock_connect_cbrZrsrt)rrpr"r.rrVrrrry�s z#BaseSelectorEventLoop._sock_connectcCs|j|�dS)N)rk)rrrprrrr~�sz(BaseSelectorEventLoop._sock_connect_donecCs�|j�rdSyA|jtjtj�}|dkrPt|d|f��nWnIttfk rkYn?tk r�}z|j |�WYdd}~XnX|j d�dS)NrzConnect call failed %s)rrZ getsockoptrMZ SOL_SOCKETZSO_ERRORrErBrArZrsrt)rrpr"r.rzrVrrrr�s z&BaseSelectorEventLoop._sock_connect_cbcCsS|jr*|j�dkr*td��ntjd|�}|j|d|�|S)a|Accept a connection. The socket must be bound to an address and listening for connections. The return value is a pair (conn, address) where conn is a new socket object usable to send and receive data on the connection, and address is the address bound to the socket on the other end of the connection. This method is a coroutine. rzthe socket must be non-blockingrWF)rFrlrmrrY�_sock_accept)rr"rprrr�sock_accept�s z!BaseSelectorEventLoop.sock_acceptcCs�|j�}|r"|j|�n|j�r2dSy#|j�\}}|jd�Wnettfk r�|j||j|d|�YnEt k r�}z|j |�WYdd}~XnX|j||f�dS)NFT)r8r6rrrOr;rBrAr<r�rZrsrt)rrprur"rrTr.rVrrrr��s z"BaseSelectorEventLoop._sock_acceptcCs�x�|D]�\}}|j|j}\}}|tj@rn|dk rn|jr^|j|�qn|j|�n|tj@r|dk r|jr�|j|�q�|j|�qqWdS)N) �fileobjr>r r_Z _cancelledr6Z _add_callbackrirk)rZ event_listrrer�rfrgrrr�_process_events�s z%BaseSelectorEventLoop._process_eventscCs!|j|j��|j�dS)N)r6r8r3)rr"rrr� _stop_serving�sz#BaseSelectorEventLoop._stop_serving)$r� __module__�__qualname__�__doc__rr%r+r(r/r3r5r2rr?r=rGrJrHrrSr<r6rjrkrqrnrwrvr{ryr~rr�r�r�r�rr)rrr*sL !#cs�eZdZdZeZdZdd�fdd�Zdd�Zdd �Z d d�Z dd �Zej r~dd�Znddd�Zdd�Zdd�Zdd�Z�S)�_SelectorTransport�iNcs�t�j||�||jd<|j�|jdz<%s>� )rrr��appendr�r��_loopr1rrr r_ri�get_write_buffer_size�join)r�infor��state�bufsizerrr�__repr__s* z_SelectorTransport.__repr__cCs|jd�dS)N)�_force_close)rrrr�abort+sz_SelectorTransport.abortcCs|jS)N)r�)rrrr� is_closing.sz_SelectorTransport.is_closingcCs^|jr dSd|_|jj|j�|jsZ|jd7_|jj|jd�ndS)NTr)r�r�r6r�r�r�� call_soon�_call_connection_lost)rrrrr31s z_SelectorTransport.closecCs7|jdk r3tjd|t�|jj�ndS)Nzunclosed transport %r)r��warnings�warn�ResourceWarningr3)rrrr�__del__>sz_SelectorTransport.__del__zFatal error on transportcCs�t|tttf�rF|jj�rutjd||dd�qun/|jji|d6|d6|d6|j d6�|j |�dS)Nz%r: %srCTrKrLrXr#)� isinstance�BrokenPipeError�ConnectionResetErrorrPr�� get_debugr rrRr�r�)rrVrKrrr�_fatal_errorCs z_SelectorTransport._fatal_errorcCs�|jr dS|jr9|jj�|jj|j�n|jsad|_|jj|j�n|jd7_|jj|j |�dS)NTr) r�r��clearr�rkr�r�r6r�r�)rrVrrrr�Rs z_SelectorTransport._force_closecCs{z |jr|jj|�nWd|jj�d|_d|_d|_|j}|dk rv|j�d|_nXdS)N)r�r�Zconnection_lostr�r3r�r�Z_detach)rrVr rrrr�^s z(_SelectorTransport._call_connection_lostcCs t|j�S)N)rxr�)rrrrr�lsz(_SelectorTransport.get_write_buffer_sizei)rr�r��max_size� bytearrayr�r�rr�r�r�r3rZPY34r�r�r�r�r�rr)rrr��s r�cseZdZddd�fdd�Zdd�Zdd�Zdd �Zd d�Zdd �Zdd�Z dd�Z �S)r!Ncs�t�j|||||�d|_d|_|jj|jj|�|jj|jj|j |j �|dk r�|jjtj|d�ndS)NF) rr�_eof�_pausedr�r�r��connection_mader<r��_read_readyr�_set_result_unless_cancelled)rrWr"r#r$rr )rrrrrs z!_SelectorSocketTransport.__init__cCsr|jrtd��n|jr0td��nd|_|jj|j�|jj�rntjd|�ndS)Nz#Cannot pause_reading() when closingzAlready pausedTz%r pauses reading) r�r0r�r�r6r�r�r r)rrrr� pause_reading�s z&_SelectorSocketTransport.pause_readingcCsm|jstd��nd|_|jr.dS|jj|j|j�|jj�ritj d|�ndS)Nz Not pausedFz%r resumes reading) r�r0r�r�r<r�r�r�r r)rrrr�resume_reading�s z'_SelectorSocketTransport.resume_readingcCs�y|jj|j�}WnLttfk r3Yn�tk rg}z|j|d�WYdd}~XnqX|r�|jj|�nW|j j �r�tjd|�n|jj �}|r�|j j|j�n |j�dS)Nz$Fatal read error on socket transportz%r received EOF)r�r@r�rBrArZr�r�� data_receivedr�r�r r�eof_receivedr6r�r3)rr>rV� keep_openrrrr��s#z$_SelectorSocketTransport._read_readycCsVt|tttf�s0tdt|��n|jrHtd��n|sRdS|jr�|jt j kr}tjd�n|jd7_dS|j s8y|jj|�}WnPttfk r�YnStk r}z|j|d�dSWYdd}~XnX||d�}|sdS|jj|j|j�n|j j|�|j�dS)Nz#data argument must be byte-ish (%r)z%Cannot call write() after write_eof()zsocket.send() raised exception.rz%Fatal write error on socket transport)r��bytesr�� memoryview� TypeError�typer�r0r�r�!LOG_THRESHOLD_FOR_CONNLOST_WRITESr �warningr�r�rDrBrArZr�r�rjr��_write_ready�extend�_maybe_pause_protocol)rr>rorVrrr�write�s4 z_SelectorSocketTransport.writecCs|jstd��y|jj|j�}Wnlttfk rHYn�tk r�}z5|jj|j �|jj �|j|d�WYdd}~Xn{X|r�|jd|�=n|j�|js|jj|j �|j r�|jd�q|jr|jjtj�qndS)NzData should not be emptyz%Fatal write error on socket transport)r��AssertionErrorr�rDrBrArZr�rkr�r�r��_maybe_resume_protocolr�r�r��shutdownrM�SHUT_WR)rrorVrrrr��s$ # z%_SelectorSocketTransport._write_readycCs9|jr dSd|_|js5|jjtj�ndS)NT)r�r�r�r�rMr�)rrrr� write_eof�s z"_SelectorSocketTransport.write_eofcCsdS)NTr)rrrr� can_write_eof�sz&_SelectorSocketTransport.can_write_eof)rr�r�rr�r�r�r�r�r�r�rr)rrr!ps #r!cs�eZdZeZddddd�fdd�Zddd�Zdd�Zd d �Zdd�Z d d�Z dd�Zdd�Zdd�Z �S)r,NFc stdkrtd��n|s6tj||�}ni|d6dd6} |rd|rd|| drVr�rrrr�zs0 #z!_SelectorSslTransport._read_readycCs�|jrMd|_|j�|jp+|jsM|jj|j|j�qMn|jr@y|jj |j�}Wn�t ttj fk r�d}Yn�tjk r�d}|jj|j�d|_YnYtk r#}z9|jj|j�|jj�|j|d�dSWYdd}~XnX|r@|jd|�=q@n|j�|js�|jj|j�|jr�|jd�q�ndS)NFrTz"Fatal write error on SSL transport)r�r�r�r�r�r<r�r�r�rDrBrAr�r�r�rkr�rZr�r�r�r�)rrorVrrrr��s4 z"_SelectorSslTransport._write_readycCs�t|tttf�s0tdt|��n|s:dS|jrx|jtjkret j d�n|jd7_dS|js�|jj |j|j�n|jj|�|j�dS)Nz#data argument must be byte-ish (%r)zsocket.send() raised exception.r)r�r�r�r�r�r�r�rr�r r�r�r�rjr�r�r�r�)rr>rrrr��s z_SelectorSslTransport.writecCsdS)NFr)rrrrr��sz#_SelectorSslTransport.can_write_eof)rr�r�r�r�rr�r�r�r�r�r�r�r�rr)rrr,�s" ? !r,csgeZdZejZddd�fdd�Zdd�Zdd�Zddd �Z d d�Z �S)r-Ncs�t�j||||�||_|jj|jj|�|jj|jj|j|j �|dk r�|jjt j|d�ndS)N)rr�_addressr�r�r�r�r<r�r�rr�)rrWr"r#r.r$r)rrrr�s z#_SelectorDatagramTransport.__init__cCstdd�|jD��S)Ncss!|]\}}t|�VqdS)N)rx)�.0r>�_rrr� �szC_SelectorDatagramTransport.get_write_buffer_size..)�sumr�)rrrrr��sz0_SelectorDatagramTransport.get_write_buffer_sizecCs�y|jj|j�\}}Wn�ttfk r9Yn|tk rm}z|jj|�WYdd}~XnHtk r�}z|j |d�WYdd}~XnX|jj ||�dS)Nz&Fatal read error on datagram transport)r�Zrecvfromr�rBrArEr��error_receivedrZr�Zdatagram_received)rr>rUrVrrrr��s"#z&_SelectorDatagramTransport._read_readycCs�t|tttf�s0tdt|��n|s:dS|jrq|d|jfkrqtd|jf��n|jr�|jr�|jt j kr�tjd�n|jd7_dS|j s�y7|jr�|jj|�n|jj||�dSWq�ttfk r+|jj|j|j�Yq�tk rc}z|jj|�dSWYdd}~Xq�tk r�}z|j|d�dSWYdd}~Xq�Xn|j jt|�|f�|j�dS)Nz#data argument must be byte-ish (%r)z#Invalid address: must be None or %szsocket.send() raised exception.rz'Fatal write error on datagram transport)r�r�r�r�r�r�r�rmr�rr�r r�r�r�rD�sendtorBrAr�rjr�� _sendto_readyrEr�r�rZr�r�r�)rr>rUrVrrrr��s< z!_SelectorDatagramTransport.sendtocCs@x�|jr�|jj�\}}y3|jr@|jj|�n|jj||�Wqttfk r�|jj||f�PYqt k r�}z|j j|�dSWYdd}~Xqtk r�}z|j |d�dSWYdd}~XqXqW|j�|js<|jj|j�|jr<|jd�qrUrVrrrr�s* z(_SelectorDatagramTransport._sendto_ready)rr�r��collections�dequer�rr�r�r�r�rr)rrr-�s 'r-)r��__all__r�rQr|rMr�r��ImportError�rrrrrr r rZ coroutinesr�logr rZ BaseEventLooprZ_FlowControlMixinZ Transportr�r!r,r-rrrr�s8 ��