U
if<} @ sT d Z dZddlZddlZddlZddlZddlZddlZddlZddl m
Z
ddl m Z ddl m Z ddl m
Z
dd l mZ dd
l mZ dd l mZ dd l mZ dd
lmZ dd ZG dd dejejZG dd deejZG dd deejZG dd deZ G dd deZ G dd deeej Z G d d d eeej Z G d d d e
j!Z"dS ) 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.
)BaseProactorEventLoop N ) base_events) constants)futures)
exceptions) protocols)sslproto)
transports)trsock)loggerc C s t || jd< z| | jd< W n0 tjk
rR | j rNtj d|dd Y nX d| jkrz|
| jd< W n tjk
r d | jd< Y nX d S )NsocketZsocknamezgetsockname() failed on %rTexc_infopeername) r TransportSocket_extraZ getsocknamer
error_loop get_debugr warningZ getpeername) transportsock r )
r4 __name__r appendr( filenor$ r% r# lenr) formatjoin)r- infor r r __repr__H s
z#_ProactorBasePipeTransport.__repr__c C s || j d< d S )Npipe)r r- r r r r r Z s z%_ProactorBasePipeTransport._set_extrac C s
|| _ d S Nr+ )r- r/ r r r r! ] s z'_ProactorBasePipeTransport.set_protocolc C s | j S rB rC r- r r r get_protocol` s z'_ProactorBasePipeTransport.get_protocolc C s | j S rB )r( rD r r r
is_closingc s z%_ProactorBasePipeTransport.is_closingc C s\ | j r
d S d| _ | jd7 _| js>| jd kr>| j| jd | jd k rX| j d | _d S )NTr ) r( r' r# r% r r* _call_connection_lostr$ cancelrD r r r closef s
z _ProactorBasePipeTransport.closec C s* | j d k r&|d| t| d | d S )Nzunclosed transport )source)r ResourceWarningrI )r- Z_warnr r r __del__q s
z"_ProactorBasePipeTransport.__del__ Fatal error on pipe transportc C sV zDt|tr*| j rBtjd| |dd n| j||| | jd W 5 | | X d S )Nz%r: %sTr )message exceptionr r/ ) _force_close
isinstanceOSErrorr r r debugcall_exception_handlerr+ )r- excrN r r r _fatal_errorv s
z'_ProactorBasePipeTransport._fatal_errorc C s | j d k r6| j s6|d kr*| j d n | j | | jr@d S d| _| jd7 _| jrj| j d | _| jr| j d | _d| _ d | _
| j | j
| d S )NTr r )
_empty_waiterdone
set_resultZ
set_exceptionr( r' r% rH r$ r& r# r r* rG )r- rU r r r rP s"
z'_ProactorBasePipeTransport._force_closec C s^ z| j | W 5 t | jdr,| jtj | j d | _| j}|d k rX| d | _X d S )Nshutdown)
hasattrr rZ r
Z SHUT_RDWRrI r" Z_detachr+ Zconnection_lost)r- rU r2 r r r rG s
z0_ProactorBasePipeTransport._call_connection_lostc C s" | j }| jd k r |t| j7 }|S rB )r& r# r; )r- sizer r r get_write_buffer_size s
z0_ProactorBasePipeTransport.get_write_buffer_size)NNN)rM )r8
__module__ __qualname____doc__r r? r r! rE rF rI warningswarnrL rV rP rG r]
__classcell__r r r3 r r . s
r c sT e Zd ZdZd fdd Zdd Zdd Zd d
Zd d Zd
d Z dddZ
Z S )_ProactorReadPipeTransportzTransport for read pipes.Nc s: d | _ d| _t |||||| | j| j d| _d S )NTF)
_pending_data_pausedr r r r*
_loop_readingr, r3 r r r s
z#_ProactorReadPipeTransport.__init__c C s | j o| j S rB )rf r( rD r r r
is_reading s z%_ProactorReadPipeTransport.is_readingc C s0 | j s | jrd S d| _| j r,td| d S )NTz%r pauses reading)r( rf r r r rS rD r r r
pause_reading s
z(_ProactorReadPipeTransport.pause_readingc C sn | j s | jsd S d| _| jd kr0| j| jd | j}d | _|d k rT| j| j| | j rjt
d| d S )NFz%r resumes reading) r( rf r$ r r* rg re _data_receivedr r rS r- datar r r resume_reading s
z)_ProactorReadPipeTransport.resume_readingc
C s | j rtd| z| j }W nL ttfk
r> Y n4 tk
rp } z| |d W Y d S d }~X Y nX |s~|
d S )Nz%r received EOFz1Fatal error: protocol.eof_received() call failed.) r r r rS r+ Z eof_received
SystemExitKeyboardInterrupt
BaseExceptionrV rI )r- Z keep_openrU r r r
_eof_received s
z(_ProactorReadPipeTransport._eof_receivedc
C s | j r|| _d S |s | d S t| jtjrzt| j| W q tt fk
rZ Y q t
k
r } z| |d W Y d S d }~X Y qX n | j | d S )Nz3Fatal error: protocol.buffer_updated() call failed.)
rf re rq rQ r+ r ZBufferedProtocolZ _feed_data_to_buffered_protorn ro rp rV Z
data_received)r- rl rU r r r rj s" z)_ProactorReadPipeTransport._data_receivedc
C st d }zRzp|d k r2d | _| r*| }n| | jrHd }W W d S |dkr\W W d S | jsv| jj | j
d| _W n t k
r } z0| js| |d n| j
rtjddd W 5 d }~X Y n tk
r } z| | W 5 d }~X Y nf tk
r } z| |d W 5 d }~X Y n8 tjk
r> | js: Y nX | jsV| j| j W 5 |d k rn| | X d S )N i z"Fatal read error on pipe transportz*Read error on pipe transport while closingTr )rj r$ rX resultrH r( rf r _proactorrecvr ConnectionAbortedErrorrV r r rS ConnectionResetErrorrP rR r CancelledErroradd_done_callbackrg )r- futrl rU r r r rg s@
z(_ProactorReadPipeTransport._loop_reading)NNN)N) r8 r^ r_ r` r rh ri rm rq rj rg rc r r r3 r rd s rd c s^ e Zd ZdZdZ fddZdd Zddd Zd
d Zd d
Z dd Z
dd Z dd Z Z
S ) _ProactorBaseWritePipeTransportzTransport for write pipes.Tc s t j|| d | _d S rB )r r rW r- argskwr3 r r r G s z(_ProactorBaseWritePipeTransport.__init__c C s t |tttfs$tdt|j | jr2td| j d k rDtd|sLd S | j
rz| j
t j krht
d | j
d7 _
d S | jd kr| jt|d n.| jst|| _| n| j| | d S )Nz/data argument must be a bytes-like object, not zwrite_eof() already calledz(unable to write; sendfile is in progressz socket.send() raised exception.r )rl )rQ bytes bytearray
memoryview TypeErrortyper8 r) RuntimeErrorrW r' r !LOG_THRESHOLD_FOR_CONNLOST_WRITESr r r%
_loop_writingr# _maybe_pause_protocolextendrk r r r writeK s,
z%_ProactorBaseWritePipeTransport.writeNc
C sV z|d k r | j d kr | jr W d S d | _ d| _|r8| |d krL| j}d | _|s| jrf| j| jd | jrz| j
t j |
nN| jj| j || _ | j st|| _| j | j | n| j | j | jd k r| j d kr| jd W n\ tk
r" } z| | W 5 d }~X Y n0 tk
rP } z| |d W 5 d }~X Y nX d S )Nr z#Fatal write error on pipe transport)r% r( r& rs r# r r* rG r) r rZ r
SHUT_WR_maybe_resume_protocolrt sendrX r; ry r r rW rY rw rP rR rV )r- frl rU r r r r q s8
z-_ProactorBaseWritePipeTransport._loop_writingc C s dS NTr rD r r r
can_write_eof s z-_ProactorBaseWritePipeTransport.can_write_eofc C s | d S rB )rI rD r r r write_eof s z)_ProactorBaseWritePipeTransport.write_eofc C s | d d S rB rP rD r r r abort s z%_ProactorBaseWritePipeTransport.abortc C s: | j d k rtd| j | _ | jd kr4| j d | j S )NzEmpty waiter is already set)rW r r Z
create_futurer% rY rD r r r _make_empty_waiter s
z2_ProactorBaseWritePipeTransport._make_empty_waiterc C s
d | _ d S rB )rW rD r r r _reset_empty_waiter s z3_ProactorBaseWritePipeTransport._reset_empty_waiter)NN)r8 r^ r_ r` Z_start_tls_compatibler r r r r r r r rc r r r3 r r{ A s &
)r{ c s$ e Zd Z fddZdd Z ZS )_ProactorWritePipeTransportc s4 t j|| | jj| jd| _| j| j d S )N ) r r r rt ru r r$ ry _pipe_closedr| r3 r r r s z$_ProactorWritePipeTransport.__init__c C s@ | r d S | jrd S d | _| jd k r4| t n| d S rB )Z cancelledr( r$ r% rP BrokenPipeErrorrI )r- rz r r r r s
z(_ProactorWritePipeTransport._pipe_closed)r8 r^ r_ r r rc r r r3 r r s r c sX e Zd ZdZd fdd Zdd Zdd Zd d
Zdd d Zdd
dZ dddZ
Z S )_ProactorDatagramTransporti Nc s> || _ d | _t j|||||d t | _| j| j d S )N)r0 r1 )
_addressrW r r collectionsdequer# r r* rg )r- r. r r/ addressr0 r1 r3 r r r s
z#_ProactorDatagramTransport.__init__c C s t | | d S rB r rA r r r r s z%_ProactorDatagramTransport._set_extrac C s t dd | jD S )Nc s s | ]\}}t |V qd S rB )r; ).0rl _r r r s zC_ProactorDatagramTransport.get_write_buffer_size..)sumr# rD r r r r] s z0_ProactorDatagramTransport.get_write_buffer_sizec C s | d d S rB r rD r r r r s z _ProactorDatagramTransport.abortc C s t |tttfs tdt||s&d S | jd k rN|d | jfkrNtd| j | jr| jr| jt j
krpt d | jd7 _d S | j
t||f | jd kr| | d S )Nz,data argument must be bytes-like object (%r)z!Invalid address: must be None or z!socket.sendto() raised exception.r )rQ r r r r r r
ValueErrorr' r r r r r# r9 r% r r )r- rl addrr r r sendto s&
z!_ProactorDatagramTransport.sendtoc
C s z| j rW d S d | _|r | | jr2| j rN| jrN| jrH| j| jd W d S | j \}}| jd k r|| jj
| j || _n| jj
j
| j ||d| _W nZ tk
r } z| j| W 5 d }~X Y nD tk
r } z| |d W 5 d }~X Y nX | j| j | d S )N)r z'Fatal write error on datagram transport)r' r% rs r# r r( r r* rG popleftrt r r r rR r+ error_received ExceptionrV ry r r )r- rz rl r rU r r r r s2
z(_ProactorDatagramTransport._loop_writingc
C s4 d }zz| jr W W d S d | _|d k rf| }| jrFd }W W d S | jd k r^|| j }}n|\}}| jrvW W d S | jd k r| jj | j
| j | _n| jj | j
| j | _W nJ t
k
r } z| j | W 5 d }~X Y n8 tjk
r | js Y n X | jd k r| j| j W 5 |r.| j || X d S rB )r+ Zdatagram_receivedr' r$ rs r( r r rt ru r max_sizeZrecvfromrR r r rx ry rg )r- rz rl r resrU r r r rg s>
z(_ProactorDatagramTransport._loop_reading)NNN)N)N)N) r8 r^ r_ r r r r] r r r rg rc r r r3 r r s
!r c @ s e Zd ZdZdd Zdd ZdS ) _ProactorDuplexPipeTransportzTransport for duplex pipes.c C s dS )NFr rD r r r r J s z*_ProactorDuplexPipeTransport.can_write_eofc C s t d S rB )NotImplementedErrorrD r r r r M s z&_ProactorDuplexPipeTransport.write_eofN)r8 r^ r_ r` r r r r r r r E s r c sB e Zd ZdZejjZd fdd Zdd Z dd Z
d d
Z Z S ) _ProactorSocketTransportz Transport for connected sockets.Nc s$ t |||||| t| d S rB )r r r Z _set_nodelayr, r3 r r r X s z!_ProactorSocketTransport.__init__c C s t | | d S rB r rA r r r r ] s z#_ProactorSocketTransport._set_extrac C s dS r r rD r r r r ` s z&_ProactorSocketTransport.can_write_eofc C s2 | j s | jrd S d| _| jd kr.| jtj d S r )r( r) r% r rZ r
r rD r r r r c s
z"_ProactorSocketTransport.write_eof)NNN)
r8 r^ r_ r` r Z
_SendfileModeZ
TRY_NATIVEZ_sendfile_compatibler r r r rc r r r3 r r Q s r c s e Zd Z fddZd3ddZd4dddddddd Zd5d
d Zd6d d
Zd7ddZd8ddZ fddZ
dd Z dd Z dd Z
dd Zd d Zd d Zd d! Zd"d# Zd$d% Zd9d&d'Zd(d) Zd:d+d,Zd-d. Zd/d0 Zd1d2 Z ZS );r c sh t td|jj || _|| _d | _i | _ |
| | t
t krdt| j d S )NzUsing proactor: %s)r r r rS r4 r8 rt _selector_self_reading_future_accept_futuresZset_loop_make_self_pipe threadingcurrent_thread main_threadsignal
set_wakeup_fd_csockr: )r- Zproactorr3 r r r m s
z BaseProactorEventLoop.__init__Nc C s t | |||||S rB )r )r- r r/ r0 r1 r2 r r r _make_socket_transportz s
z,BaseProactorEventLoop._make_socket_transportF) server_sideserver_hostnamer1 r2 ssl_handshake_timeoutc C s0 t j| |||||| d}
t| ||
||d |
jS )N)r r1 r2 )r Z SSLProtocolr Z_app_transport) r- Zrawsockr/
sslcontextr0 r r r1 r2 r Z ssl_protocolr r r _make_ssl_transport s z)BaseProactorEventLoop._make_ssl_transportc C s t | |||||S rB )r )r- r r/ r r0 r1 r r r _make_datagram_transport s
z.BaseProactorEventLoop._make_datagram_transportc C s t | ||||S rB )r r- r r/ r0 r1 r r r _make_duplex_pipe_transport s z1BaseProactorEventLoop._make_duplex_pipe_transportc C s t | ||||S rB )rd r r r r _make_read_pipe_transport s z/BaseProactorEventLoop._make_read_pipe_transportc C s t | ||||S rB )r r r r r _make_write_pipe_transport s z0BaseProactorEventLoop._make_write_pipe_transportc sj | rtd| r d S t t kr6td | | | j
d | _
d | _ t
d S )Nz!Cannot close a running event loop)Z
is_runningr is_closedr r r r r _stop_accept_futures_close_self_pipert rI r r rD r3 r r rI s
zBaseProactorEventLoop.closec s | j ||I d H S rB )rt ru )r- r nr r r sock_recv s z BaseProactorEventLoop.sock_recvc s | j ||I d H S rB )rt Z recv_into)r- r Zbufr r r sock_recv_into s z$BaseProactorEventLoop.sock_recv_intoc s | j ||I d H S rB )rt r )r- r rl r r r sock_sendall s z"BaseProactorEventLoop.sock_sendallc s | j ||I d H S rB )rt Zconnect)r- r r r r r sock_connect s z"BaseProactorEventLoop.sock_connectc s | j |I d H S rB )rt acceptrA r r r sock_accept s z!BaseProactorEventLoop.sock_acceptc
s( z | }W n2 ttjfk
r> } ztdW 5 d }~X Y nX zt|j}W n, t k
r| } ztdW 5 d }~X Y nX |r|n|}|sdS t
|d}|rt
|| |n|} t
||}d}
zLt
| | |}|dkr|
W 0S | j
||||I d H ||7 }|
|7 }
qW 5 |
dkr"| | X d S )Nznot a regular filer l )r: AttributeErrorioUnsupportedOperationr ZSendfileNotAvailableErrorosfstatst_sizerR minseekrt sendfile) r- r fileoffsetcountr: errZfsizeZ blocksizeZend_posZ
total_sentr r r _sock_sendfile_native s0
z+BaseProactorEventLoop._sock_sendfile_nativec sZ | }| | I d H z | j|j|||ddI d H W S | |rT| X d S )NF)Zfallback)rh ri r r rm Z
sock_sendfiler )r- Ztranspr r r rm r r r _sendfile_native s z&BaseProactorEventLoop._sendfile_nativec C sL | j d k r| j d | _ | j d | _| j d | _| jd8 _d S )Nr )r rH _ssockrI r
_internal_fdsrD r r r r s
z&BaseProactorEventLoop._close_self_pipec C s: t \| _| _| jd | jd | jd7 _d S )NFr )r
Z
socketpairr r Z setblockingr rD r r r r s z%BaseProactorEventLoop._make_self_pipec
C s z4|d k r| | j|k r"W d S | j| jd}W nb tjk
rL Y d S ttfk
rd Y nF t k
r } z|
d|| d W 5 d }~X Y nX || _| | j d S )Ni z.Error on reading from the event loop self pipe)rN rO r. )
rs r rt ru r r rx rn ro rp rT ry _loop_self_reading)r- r rU r r r r s$
z(BaseProactorEventLoop._loop_self_readingc 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 socketTr )r r rR _debugr rS )r- Zcsockr r r _write_to_self s z$BaseProactorEventLoop._write_to_selfd c s( d fdd d S )Nc
s, z| d k rn| \}}jr,td|| }d k rXj||dd|id nj||d|id r|W d S j} W n t k
r } zH
dkrʈ d|t
d njrtjd dd
W 5 d }~X Y n8 tjk
r Y nX | j
< | d S ) Nz#%r got a new connection from %r: %rTr )r r1 r2 r r r zAccept failed on a socket)rN rO r
zAccept failed on socket %rr )rs r r rS r r r rt r rR r: rT r r rI r rx r ry )r Zconnr r/ rU r. protocol_factoryr- r2 r r r r r r. / s\
z2BaseProactorEventLoop._start_serving..loop)N)r* )r- r r r r2 Zbacklogr r r r _start_serving+ s %z$BaseProactorEventLoop._start_servingc C s d S rB r )r- Z
event_listr r r _process_eventsV s z%BaseProactorEventLoop._process_eventsc C s& | j D ] }| q
| j d S rB )r valuesrH clear)r- futurer r r r Z s
z*BaseProactorEventLoop._stop_accept_futuresc C s6 | j | d }|r | | j| | d S rB )r popr: rH rt
_stop_servingrI )r- r r r r r r _ s
z#BaseProactorEventLoop._stop_serving)NNN)N)NNN)NN)NN)NN)N)NNr N)r8 r^ r_ r r r r r r r rI r r r r r r r r r r r r r r r rc r r r3 r r k s\
+r )#r` __all__r r r
ra r r r r r r r r r r
r logr r Z_FlowControlMixinZ
BaseTransportr Z
ReadTransportrd ZWriteTransportr{ r r Z Transportr r Z
BaseEventLoopr r r r r sR n