U
ifT @ s. d Z dZddlZddlZddlZddlZddlZddlZddlZz ddl Z W n e
k
rd dZ Y nX ddl m Z ddl m
Z
ddl mZ ddl mZ dd l mZ dd
l mZ dd l mZ dd l mZ dd
lmZ dd Zdd ZG dd de jZG dd dejejZ G dd de Z G dd de Z dS )zEvent 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) constants)events)futures) protocols)sslproto)
transports)trsock)loggerc C s8 z| |}W n tk
r$ Y dS X t|j|@ S d S NF)get_keyKeyErrorboolr )selectorfdZeventkey r Z!d?d@ Z"dAdB Z#dCdD Z$dEdF Z%dGdH Z&dIdJ Z'dKdL Z(dMdN Z)dOdP Z*dQdR Z+ Z,S )Wr zJSelector event loop.
See events.EventLoop for API specification.
Nc sF t |d krt }td|jj || _| t
| _ d S )NzUsing selector: %s)
super__init__ selectorsZDefaultSelectorr debug __class____name__ _selector_make_self_pipeweakrefZWeakValueDictionary _transports)selfr r r r r 6 s
z BaseSelectorEventLoop.__init__extraserverc C s t | |||||S N)_SelectorSocketTransport)r& r protocolwaiterr) r* r r r _make_socket_transport@ s
z,BaseSelectorEventLoop._make_socket_transportF) server_sideserver_hostnamer) r* ssl_handshake_timeoutc C s0 t j| |||||| d}
t| ||
||d |
jS )N)r2 r( )r Z SSLProtocolr, Z_app_transport) r& Zrawsockr-
sslcontextr. r0 r1 r) r* r2 Z ssl_protocolr r r _make_ssl_transportE s z)BaseSelectorEventLoop._make_ssl_transportc C s t | |||||S r+ )_SelectorDatagramTransport)r& r r- addressr. r) r r r _make_datagram_transportR s
z.BaseSelectorEventLoop._make_datagram_transportc sL | rtd| r d S | t | jd k rH| j d | _d S )Nz!Cannot close a running event loop)Z
is_running RuntimeError is_closed_close_self_piper closer" r& r' r r r; W s
zBaseSelectorEventLoop.closec C sB | | j | j d | _| j d | _| jd8 _d S )Nr )_remove_reader_ssockfilenor; _csock
_internal_fdsr< r r r r: b s
z&BaseSelectorEventLoop._close_self_pipec C sN t \| _| _| jd | jd | jd7 _| | j | j d S )NFr ) socketZ
socketpairr> r@ setblockingrA _add_readerr? _read_from_selfr< r r r r# j s
z%BaseSelectorEventLoop._make_self_pipec C s d S r+ r r& datar r r _process_self_datar s z(BaseSelectorEventLoop._process_self_datac C sX z"| j d}|sW qT| | W q tk
r: Y q Y q tk
rP Y qTY q X q d S )Ni )r> recvrH InterruptedErrorBlockingIOErrorrF r r r rE u s z%BaseSelectorEventLoop._read_from_selfc C sN | j }|d krd S z|d W n( tk
rH | jrDtjddd Y nX d S )N z3Fail to write a null byte into the self-pipe socketTexc_info)r@ sendOSError_debugr r )r& Zcsockr r r _write_to_self s z$BaseSelectorEventLoop._write_to_selfd c
C s" | | | j|||||| d S r+ )rD r? _accept_connection)r& protocol_factoryr r3 r* backlogr2 r r r _start_serving s z$BaseSelectorEventLoop._start_servingc
C s t |D ]}z0| \}} | jr0td|| | |d W n tttfk
rZ Y d S t k
r }
zd|
j
t
j t
j t
j
t
jfkr| d|
t|d | | | tj| j|||||| n W 5 d }
~
X Y qX d| i} | ||| |||} | | qd S )Nz#%r got a new connection from %r: %rFz&socket.accept() out of system resource)message exceptionrB peername)rangeacceptrQ r r rC rK rJ ConnectionAbortedErrorrP errnoZEMFILEZENFILEZENOBUFSZENOMEMcall_exception_handlerr TransportSocketr= r? Z
call_laterr ZACCEPT_RETRY_DELAYrW _accept_connection2Z create_task)
r& rU r r3 r* rV r2 _connaddrexcr) r\ r r r rT sV z(BaseSelectorEventLoop._accept_connectionc
s d }d }zt| }| } |r8| j|||| d|||d}n| j||| ||d}z| I d H W n tk
rx | Y nX W nt ttfk
r Y n\ tk
r }
z>| jrd|
d} |d k r|| d< |d k r|| d< | | W 5 d }
~
X Y nX d S )NT)r. r0 r) r* r2 )r. r) r* z3Error on transport creation for incoming connection)rX rY r- transport)
create_futurer4 r/
BaseExceptionr;
SystemExitKeyboardInterruptrQ r_ ) r& rU rc r) r3 r* r2 r- rf r. re contextr r r ra sP z)BaseSelectorEventLoop._accept_connection2c
C s |}t |tsJzt| }W n* tttfk
rH td|d Y nX z| j| }W n tk
rl Y n X | st d|d|d S )NzInvalid file object: zFile descriptor z is used by transport )
r intr? AttributeErrorr
ValueErrorr% r
is_closingr8 )r& r r? rf r r r _ensure_fd_no_transport s
z-BaseSelectorEventLoop._ensure_fd_no_transportc G s | t||| d }z| j|}W n* tk
rR | j|tj|d f Y n>X |j|j }\}}| j
||tjB ||f |d k r| d S r+ )
_check_closedr Handler" r r registerr
EVENT_READrG modifycancel r& r callbackargsZhandler maskreaderwriterr r r rD s
z!BaseSelectorEventLoop._add_readerc C s | r dS z| j|}W n tk
r2 Y dS X |j|j }\}}|tj M }|sd| j| n| j ||d |f |d k r|
dS dS d S NFT) r9 r" r r r rG r rt
unregisterru rv r& r r rz r{ r| r r r r= s z$BaseSelectorEventLoop._remove_readerc G s | t||| d }z| j|}W n* tk
rR | j|tjd |f Y n>X |j|j }\}}| j
||tjB ||f |d k r| d S r+ ) rq r rr r" r r rs r EVENT_WRITErG ru rv rw r r r _add_writer% s
z!BaseSelectorEventLoop._add_writerc C s | r dS z| j|}W n tk
r2 Y dS X |j|j }\}}|tj M }|sd| j| n| j |||df |dk r|
dS dS dS )Remove a writer callback.FNT) r9 r" r r r rG r r r~ ru rv r r r r _remove_writer4 s z$BaseSelectorEventLoop._remove_writerc G s | | | j||f| S )zAdd a reader callback.)rp rD r& r rx ry r r r
add_readerK s
z BaseSelectorEventLoop.add_readerc C s | | | |S )zRemove a reader callback.)rp r= r& r r r r
remove_readerP s
z#BaseSelectorEventLoop.remove_readerc G s | | | j||f| S )zAdd a writer callback..)rp r r r r r
add_writerU s
z BaseSelectorEventLoop.add_writerc C s | | | |S )r )rp r r r r r
remove_writerZ s
z#BaseSelectorEventLoop.remove_writerc s t | | jr"| dkr"tdz ||W S ttfk
rF Y nX | }| }| || j
||| | t
| j| |I dH S )zReceive 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.
r the socket must be non-blockingN)r rQ
gettimeoutrn rI rK rJ rg r? r
_sock_recvadd_done_callback functoolspartial_sock_read_done)r& r nfutr r r r sock_recv_ s z BaseSelectorEventLoop.sock_recvc C s | | d S r+ )r r& r r r r r r t s z%BaseSelectorEventLoop._sock_read_donec
C s | r d S z||}W n\ ttfk
r4 Y d S ttfk
rL Y n6 tk
rv } z|| W 5 d }~X Y n X || d S r+ ) donerI rK rJ ri rj rh
set_exception
set_result)r& r r r rG re r r r r w s z BaseSelectorEventLoop._sock_recvc s t | | jr"| dkr"tdz ||W S ttfk
rF Y nX | }| }| || j
||| | t
| j| |I dH S )zReceive data from the socket.
The received data is written into *buf* (a writable buffer).
The return value is the number of bytes written.
r r N)r rQ r rn recv_intorK rJ rg r? r _sock_recv_intor r r r )r& r bufr r r r r sock_recv_into s z$BaseSelectorEventLoop.sock_recv_intoc
C s | r d S z||}W n\ ttfk
r4 Y d S ttfk
rL Y n6 tk
rv } z|| W 5 d }~X Y n X || d S r+ ) r r rK rJ ri rj rh r r )r& r r r nbytesre r r r r s z%BaseSelectorEventLoop._sock_recv_intoc s t | | jr"| dkr"tdz||}W n ttfk
rL d}Y nX |t|kr^dS | }| }|
t | j
| | || j||t||g |I dH 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.
r r N)r rQ r rn rO rK rJ lenrg r? r r r _sock_write_doner
_sock_sendall
memoryview)r& r rG r r r r r r sock_sendall s&
z"BaseSelectorEventLoop.sock_sendallc
C s | r d S |d }z|||d }W nb ttfk
rD Y d S ttfk
r\ Y n2 tk
r } z|| W Y d S d }~X Y nX ||7 }|t|kr| d n||d< d S )Nr )
r rO rK rJ ri rj rh r r r )r& r r Zviewposstartr re r r r r s
z#BaseSelectorEventLoop._sock_sendallc s t | | jr"| dkr"tdttdr8|jtjkrf| j||j|j | dI dH }|d \}}}}}|
}| ||| |I dH S )zTConnect to a remote socket at address.
This method is a coroutine.
r r AF_UNIX)familyprotoloopN) r rQ r rn hasattrrB r r Z_ensure_resolvedr rg
_sock_connect)r& r r6 Zresolvedrb r r r r sock_connect s z"BaseSelectorEventLoop.sock_connectc
C s | }z|| W n ttfk
rV |t| j| | || j ||| Y nN t
t fk
rn Y n6 t k
r } z|
| W 5 d }~X Y n X |d d S r+ )r? ZconnectrK rJ r r r r r _sock_connect_cbri rj rh r r )r& r r r6 r re r r r r s z#BaseSelectorEventLoop._sock_connectc C s | | d S r+ )r r r r r r s z&BaseSelectorEventLoop._sock_write_donec
C s | r d S z,|tjtj}|dkr6t|d| W nZ ttfk
rP Y nN tt fk
rh Y n6 t
k
r } z| | W 5 d }~X Y n X | d d S )Nr zConnect call failed )
r Z
getsockoptrB Z
SOL_SOCKETZSO_ERRORrP rK rJ ri rj rh r r )r& r r r6 errre r r r r s z&BaseSelectorEventLoop._sock_connect_cbc sB t | | jr"| dkr"td| }| |d| |I dH S )aW 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.
r r FN)r rQ r rn rg _sock_accept)r& r r r r r sock_accept s z!BaseSelectorEventLoop.sock_acceptc
C s | }|r| | | r"d S z| \}}|d W nn ttfk
rh | || j|d| Y nR t t
fk
r Y n: t k
r } z| | W 5 d }~X Y nX |
||f d S r} )r? r r r\ rC rK rJ r r ri rj rh r r )r& r Z
registeredr r rc r6 re r r r r * s
z"BaseSelectorEventLoop._sock_acceptc sp | j |j= | }| | I d H z | j|j|||ddI d H W S | |r^| || j |j< X d S )NF)Zfallback) r% _sock_fd
is_reading
pause_reading_make_empty_waiter_reset_empty_waiterresume_readingZ
sock_sendfile_sock)r& Ztranspfileoffsetcountr r r r _sendfile_native< s
z&BaseSelectorEventLoop._sendfile_nativec C s |D ]v\}}|j |j }\}}|tj@ rL|d k rL|jrB| | n
| | |tj@ r|d k r|jrp| | q| | qd S r+ ) fileobjrG r rt Z
_cancelledr= Z
_add_callbackr r )r& Z
event_listr rz r r{ r| r r r _process_eventsJ s
z%BaseSelectorEventLoop._process_eventsc C s | | | d S r+ )r= r? r; )r& r r r r
_stop_servingX s z#BaseSelectorEventLoop._stop_serving)N)N)N)NNN)-r!
__module__ __qualname____doc__r r/ r ZSSL_HANDSHAKE_TIMEOUTr4 r7 r; r: r# rH rE rR rW rT ra rp rD r= r r r r r r r r r r r r r r r r r r r r r r
__classcell__r r r' r r 0 s~
.
) r c s e Zd ZdZeZdZd fdd Zdd Zdd Z d d
Z
d d Z d
d Z dd Z
ejfddZd ddZdd Zdd Zdd Zd d Z ZS ) _SelectorTransporti Nc s t || t|| jd< z| | jd< W n tk
rN d | jd< Y nX d| jkrz| | jd< W n tj k
r d | jd< Y nX || _
| | _ d| _
| | || _| | _d| _d| _| jd k r| j | |j| j < d S )NrB ZsocknamerZ Fr )r r r r` _extraZ getsocknamerP Z getpeernamerB errorr r? r _protocol_connected set_protocol_server_buffer_factory_buffer
_conn_lost_closingZ_attachr% )r& r r r- r) r* r' r r r i s,
z_SelectorTransport.__init__c C s | j jg}| jd kr |d n| jr0|d |d| j | jd k r| j st| jj | jt
j }|rz|d n
|d t| jj | jt
j }|rd}nd}|
}|d| d | d
d d |S )
Nclosedclosingzfd=z read=pollingz read=idlepollingZidlezwrite=z<{}> )r r! r appendr r _loopr9 r r" r rt r get_write_buffer_sizeformatjoin)r& infor statebufsizer r r __repr__ s0
z_SelectorTransport.__repr__c C s | d d S r+ ) _force_closer< r r r abort s z_SelectorTransport.abortc C s || _ d| _d S NT) _protocolr r& r- r r r r s z _SelectorTransport.set_protocolc C s | j S r+ )r r< r r r get_protocol s z _SelectorTransport.get_protocolc C s | j S r+ )r r< r r r ro s z _SelectorTransport.is_closingc C sT | j r
d S d| _ | j| j | jsP| jd7 _| j| j | j| jd d S NTr ) r r r= r r r r call_soon_call_connection_lostr< r r r r; s z_SelectorTransport.closec C s, | j d k r(|d| t| d | j d S )Nzunclosed transport )source)r ResourceWarningr; )r& Z_warnr r r __del__ s
z_SelectorTransport.__del__Fatal error on transportc C sN t |tr(| j r@tjd| |dd n| j||| | jd | | d S )Nz%r: %sTrM )rX rY rf r- ) r rP r get_debugr r r_ r r )r& re rX r r r _fatal_error s
z _SelectorTransport._fatal_errorc C sd | j r
d S | jr(| j | j| j | jsBd| _| j| j | j d7 _ | j| j | d S r )
r r clearr r r r r= r r r& re r r r r s
z _SelectorTransport._force_closec C sV z| jr| j| W 5 | j d | _ d | _d | _| j}|d k rP| d | _X d S r+ )r r; r r r Z_detachr Zconnection_lost)r& re r* r r r r s
z(_SelectorTransport._call_connection_lostc C s
t | jS r+ )r r r< r r r r s z(_SelectorTransport.get_write_buffer_sizec G s" | j r
d S | jj||f| d S r+ )r r rD r r r r rD s z _SelectorTransport._add_reader)NN)r )r! r r max_size bytearrayr r r r r r r ro r; warningswarnr r r r r rD r r r r' r r ] s
r c s e Zd ZdZejjZd# fdd Z fddZ dd Z
d d
Z d d Z d
d Z
dd Zdd Zdd Zdd Zdd Zdd Zdd Z fd d Zd d Zd!d" Z ZS )$r, TNc s~ d | _ t ||||| d| _d| _d | _t| j | j
| j j | | j
| j
| j| j |d k rz| j
tj|d d S r
)_read_ready_cbr r _eof_paused
_empty_waiterr Z _set_nodelayr r r r connection_maderD r _read_readyr _set_result_unless_cancelled)r& r r r- r. r) r* r' r r r s
z!_SelectorSocketTransport.__init__c s. t |tjr| j| _n| j| _t | d S r+ )r r ZBufferedProtocol_read_ready__get_bufferr _read_ready__data_receivedr r r r' r r r s
z%_SelectorSocketTransport.set_protocolc C s | j o| j S r+ )r r r< r r r r s z#_SelectorSocketTransport.is_readingc C s> | j s | jrd S d| _| j| j | j r:td| d S )NTz%r pauses reading)r r r r= r r r r r< r r r r s
z&_SelectorSocketTransport.pause_readingc C s@ | j s | jsd S d| _| | j| j | j r Y n4 tk
rp } z| |d W Y d S d }~X Y nX |r| j
| j n| d S )Nz%r received EOFz1Fatal error: protocol.eof_received() call failed.)
r r r r r Z eof_receivedri rj rh r r= r r; )r& Z keep_openre r r r r e s
z,_SelectorSocketTransport._read_ready__on_eofc
C s6 t |tttfs$tdt|j| jr2td| j d k rDtd|sLd S | j
rz| j
t j krht
d | j
d7 _
d S | js z| j|}W nb ttfk
r Y nb ttfk
r Y nJ tk
r } z| |d W Y d S d }~X Y nX ||d }|s d S | j| j| j | j | | d S )N/data argument must be a bytes-like object, not z%Cannot call write() after write_eof()z(unable to write; sendfile is in progress socket.send() raised exception.r %Fatal write error on socket transport) r bytesr r r typer! r r8 r r r !LOG_THRESHOLD_FOR_CONNLOST_WRITESr warningr r rO rK rJ ri rj rh r r r r _write_readyextend_maybe_pause_protocol)r& rG r re r r r writez s:
z _SelectorSocketTransport.writec
C s( | j std| jrd S z| j| j }W n ttfk
rB Y n ttfk
rZ Y n t k
r } z>| j
| j | j
| |d | jd k r| j| W 5 d }~X Y npX |r| j d |= | | j s$| j
| j | jd k r| jd | jr| d n| jr$| jtj d S )NzData should not be emptyr )r AssertionErrorr r rO rK rJ ri rj rh r r r r r r r _maybe_resume_protocolr r r r shutdownrB SHUT_WR)r& r re r r r r s4
z%_SelectorSocketTransport._write_readyc C s. | j s | jrd S d| _| js*| jtj d S r )r r r r r rB r r< r r r write_eof s
z"_SelectorSocketTransport.write_eofc C s dS r r r< r r r
can_write_eof s z&_SelectorSocketTransport.can_write_eofc s* t | | jd k r&| jtd d S )Nz Connection is closed by peer)r r r r ConnectionErrorr r' r r r s
z._SelectorSocketTransport._call_connection_lostc C s6 | j d k rtd| j | _ | js0| j d | j S )NzEmpty waiter is already set)r r8 r rg r r r< r r r r s
z+_SelectorSocketTransport._make_empty_waiterc C s
d | _ d S r+ )r r< r r r r s z,_SelectorSocketTransport._reset_empty_waiter)NNN)r! r r Z_start_tls_compatibler Z
_SendfileModeZ
TRY_NATIVEZ_sendfile_compatibler r r r r r r r r r r r r
r r r r r r r' r r, s* %' r, c sF e Zd ZejZd fdd Zdd Zdd Zd
dd Z d
d Z
Z S )r5 Nc s^ t |||| || _| j| jj| | j| j| j| j |d k rZ| jt
j |d d S r+ ) r r _addressr r r r rD r r r r )r& r r r- r6 r. r) r' r r r s
z#_SelectorDatagramTransport.__init__c C s t dd | jD S )Nc s s | ]\}}t |V qd S r+ )r ).0rG rb r r r s zC_SelectorDatagramTransport.get_write_buffer_size..)sumr r< r r r r s z0_SelectorDatagramTransport.get_write_buffer_sizec
C s | j r
d S z| j| j\}}W n ttfk
r8 Y n tk
rd } z| j| W 5 d }~X Y nT t t
fk
r| Y n< t k
r } z| |d W 5 d }~X Y nX | j
|| d S )Nz&Fatal read error on datagram transport)r r Zrecvfromr rK rJ rP r error_receivedri rj rh r Zdatagram_receivedr& rG rd re r r r r s z&_SelectorDatagramTransport._read_readyc
C s t |tttfs$tdt|j|s,d S | jrV|d | jfkrPtd| j | j}| j r| jr| j t
j krxt
d | j d7 _ d S | jslz,| jd r| j| n| j|| W d S ttfk
r | j| j| j Y n tk
r } z| j| W Y d S d }~X Y nP t t fk
r6 Y n6 t k
rj } z| |d W Y d S d }~X Y nX | j t||f | ! d S )Nr z!Invalid address: must be None or r r rZ 'Fatal write error on datagram transport)"r r r r r r r! r rn r r r r r r r r rO sendtorK rJ r r r
_sendto_readyrP r r ri rj rh r r r r r r r r sH
z!_SelectorDatagramTransport.sendtoc
C s | j r| j \}}z*| jd r.| j| n| j|| W q ttfk
rj | j ||f Y qY q t k
r } z| j
| W Y d S d }~X Y q t t
fk
r Y q tk
r } z| |d W Y d S d }~X Y q X q | | j s| j| j | jr| d d S )NrZ r )r popleftr r rO r rK rJ
appendleftrP r r ri rj rh r r r r r r r r r r r r * s2
z(_SelectorDatagramTransport._sendto_ready)NNN)N) r! r r collectionsdequer r r r r r r r r r' r r5 s
+r5 ) r __all__r r^ r r rB r r$ r ImportError r r r r r r r
r logr r r Z
BaseEventLoopr Z_FlowControlMixinZ Transportr r, r5 r r r r sF
1 o