B
©Ø1ɨ› ã @ s d Z dZddlZddlZddlZddlZddlmZ ddlmZ ddlm Z ddlm
Z
dd lm Z dd
lm Z dd lm
Z
dd lmZ G d
d„ de
je
jƒZG dd„ dee
jƒZG dd„ dee
jƒZG dd„ deƒZG dd„ deee
jƒZG dd„ deee
jƒZG dd„ dejƒZ dS )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)Úloggerc s| e Zd ZdZd ‡ fdd„ Zdd„ Zdd„ Zd d
„ Zd d „ Zd
d„ Z dd„ Z
dd„ Z d dd„Z dd„ Z
dd„ Zdd„ Z‡ ZS ) Ú_ProactorBasePipeTransportz*Base class for pipe and socket transports.Nc sœ t ƒ ||¡ | |¡ || _| |¡ || _d | _d | _d | _d| _ d| _
d| _ d| _ | jd k rl| j
¡ | j | jj| ¡ |d k r˜| j tj|d ¡ d S )Nr F)Ú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_mader Z _set_result_unless_cancelled)ÚselfÚloopÚsockÚprotocolÚwaiterÚextraÚserver)Ú __class__© úú )
r$ Ú__name__r Úappendr Úfilenor r r Úlenr ÚformatÚjoin)r Úinfor% r% r& Ú__repr__2 s
z#_ProactorBasePipeTransport.__repr__c C s || j d< d S )NÚpipe)Ú_extra)r r r% r% r& r D s z%_ProactorBasePipeTransport._set_extrac C s
|| _ d S )N)r )r r r% r% r& r G s z'_ProactorBasePipeTransport.set_protocolc C s | j S )N)r )r r% r% r& Ú get_protocolJ s z'_ProactorBasePipeTransport.get_protocolc C s | j S )N)r )r r% r% r& Ú
is_closingM 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 Úcancel)r r% r% r& ÚcloseP s
z _ProactorBasePipeTransport.closec C s, | j d k r(tjd| ›t| d | ¡ d S )Nzunclosed transport )Úsource)r ÚwarningsÚwarnÚResourceWarningr8 )r r% r% r& Ú__del__[ s
z"_ProactorBasePipeTransport.__del__ú Fatal error on pipe transportc C sV zDt |tƒr*| j ¡ rBtjd| |dd n| j ||| | jdœ¡ W d | |¡ X d S )Nz%r: %sT)Úexc_info)ÚmessageÚ exceptionZ transportr ) Ú
isinstanceÚOSErrorr Ú get_debugr ÚdebugÚcall_exception_handlerr Ú _force_close)r Úexcr@ r% r% r& Ú _fatal_errora 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_waiterÚdoneÚ
set_resultZ
set_exceptionr r r r7 r r r r r r6 )r rH r% r% r& rG p s"
z'_ProactorBasePipeTransport._force_closec C s^ z| j |¡ W d t| jdƒr,| j tj¡ | j ¡ d | _| j}|d k rX| ¡ d | _X d S )NÚshutdown)
r Zconnection_lostÚhasattrr rM ÚsocketZ SHUT_RDWRr8 r Z_detach)r rH r# r% r% r& r6 „ s
z0_ProactorBasePipeTransport._call_connection_lostc C s" | j }| jd k r |t| jƒ7 }|S )N)r r r- )r Úsizer% r% r& Úget_write_buffer_size• s
z0_ProactorBasePipeTransport.get_write_buffer_size)NNN)r> )r* Ú
__module__Ú __qualname__Ú__doc__r r1 r r r4 r5 r8 r= rI rG r6 rQ Ú
__classcell__r% r% )r$ r& r s
r c sT e Zd ZdZd‡ fdd„ Zdd„ Zdd„ Zd d
„ Zd d „ Zd
d„ Z ddd„Z
‡ Z S )Ú_ProactorReadPipeTransportzTransport for read pipes.Nc s: d | _ d| _tƒ ||||||¡ | j | j¡ d| _d S )NTF)Ú
_pending_dataÚ_pausedr
r r r Ú
_loop_reading)r r r r r! r" r# )r$ r% r& r s
z#_ProactorReadPipeTransport.__init__c C s | j o| j S )N)rX r )r r% r% r& Ú
is_reading© s z%_ProactorReadPipeTransport.is_readingc C s0 | j s | jrd S d| _| j ¡ r,t d| ¡ d S )NTz%r pauses reading)r rX r rD r rE )r 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 rX r r r rY rW Ú_data_receivedrD r rE )r Údatar% r% r& Úresume_reading¿ s
z)_ProactorReadPipeTransport.resume_readingc
C sd | j ¡ rt d| ¡ y| j ¡ }W n. tk
rR } z| |d¡ d S d }~X Y nX |s`| ¡ d S )Nz%r received EOFz1Fatal error: protocol.eof_received() call failed.) r rD r rE r Z eof_receivedÚ ExceptionrI r8 )r Z keep_openrH r% r% r& Ú
_eof_receivedÑ s
z(_ProactorReadPipeTransport._eof_receivedc
C s | j r | jd kst‚|| _d S |s.| ¡ d S t| jtjƒr€yt | j|¡ W qŒ t k
r| } z|
|d¡ d S d }~X Y qŒX n | j |¡ d S )Nz3Fatal error: protocol.buffer_updated() call failed.) rX rW ÚAssertionErrorr` rB r r ZBufferedProtocolZ _feed_data_to_buffered_protor_ rI Z
data_received)r r] rH r% r% r& r\ ß s z)_ProactorReadPipeTransport._data_receivedc
C s‚ d }z`y~|d k rP| j |ks0| j d kr,| js0t‚d | _ | ¡ rH| ¡ }n| ¡ | jr^d }d S |dkrjd S | js„| jj | j
d¡| _ W nÆ t k
rÔ } z0| js¬| |d¡ n| j
¡ rÄtjddd W d d }~X Y n tk
rþ } z| |¡ W d d }~X Y nf tk
r, } z| |d¡ W d d }~X Y n8 tjk
rL | jsH‚ Y nX | jsd| j | j¡ W d |d k r|| |¡ X d S )Nó i € z"Fatal read error on pipe transportz*Read error on pipe transport while closingT)r? )r r ra rK Úresultr7 rX r Ú _proactorÚrecvr ÚConnectionAbortedErrorrI rD r rE ÚConnectionResetErrorrG rC r ÚCancelledErrorÚadd_done_callbackrY r\ )r Úfutr] rH r% r% r& rY ö sB
z(_ProactorReadPipeTransport._loop_reading)NNN)N) r* rR rS rT r rZ r[ r^ r` r\ rY rU r% r% )r$ r& rV œ s rV c s^ e Zd ZdZdZ‡ fdd„Zdd„ 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 )N)r
r rJ )r ÚargsÚkw)r$ r% r& r . s z(_ProactorBaseWritePipeTransport.__init__c C sÖ t |tttfƒs$tdt|ƒj› ƒ‚| jr2tdƒ‚| j d k rDtdƒ‚|sLd S | j
rz| j
t j krht
d¡ | j
d7 _
d S | jd kr¤| jd ks’t‚| jt|ƒd n.| js¾t|ƒ| _| ¡ 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 )r] )rB ÚbytesÚ bytearrayÚ
memoryviewÚ TypeErrorÚtyper* r Ú RuntimeErrorrJ r r Z!LOG_THRESHOLD_FOR_CONNLOST_WRITESr Úwarningr r ra Ú
_loop_writingÚ_maybe_pause_protocolÚextend)r r] r% r% r& Úwrite2 s,
z%_ProactorBaseWritePipeTransport.writeNc
C sv y|d k r | j d kr | jr d S || j ks.t‚d | _ d| _|rF| ¡ |d krZ| j}d | _|s’| jrt| j | jd ¡ | j rˆ| j
t j
¡ | ¡ n\| jj | j
|¡| _ | j ¡ sà| jdks¾t‚t|ƒ| _| j | j¡ | ¡ n| j | j¡ | jd k r| j d kr| j d ¡ W n\ tk
rB } z| |¡ W d d }~X Y n0 tk
rp } z| |d¡ W d d }~X Y nX d S )Nr z#Fatal write error on pipe transport) r r ra r rc r r r r6 r r rM rO ÚSHUT_WRZ_maybe_resume_protocolrd ÚsendrK r- ri ru rv rJ rL rg rG rC rI )r Úfr] rH r% r% r& ru X s<
z-_ProactorBaseWritePipeTransport._loop_writingc C s dS )NTr% )r r% r% r& Ú
can_write_eof s z-_ProactorBaseWritePipeTransport.can_write_eofc C s | ¡ d S )N)r8 )r r% r% r& Ú write_eof„ s z)_ProactorBaseWritePipeTransport.write_eofc C s | d ¡ d S )N)rG )r 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)rJ rs r Z
create_futurer rL )r r% r% r& Ú_make_empty_waiterŠ s
z2_ProactorBaseWritePipeTransport._make_empty_waiterc C s
d | _ d S )N)rJ )r r% r% r& Ú_reset_empty_waiter’ s z3_ProactorBaseWritePipeTransport._reset_empty_waiter)NN)r* rR rS rT Z_start_tls_compatibler rx ru r| r} r~ r r€ rU r% r% )r$ r& rk ( s &
)rk c s$ e Zd Z‡ fdd„Zdd„ Z‡ ZS )Ú_ProactorWritePipeTransportc s4 t ƒ j||Ž | jj | jd¡| _| j | j¡ d S )Né ) r
r r rd re r r ri Ú _pipe_closed)r rl rm )r$ r% r& r — s z$_ProactorWritePipeTransport.__init__c C sv | ¡ r d S | ¡ dks t‚| jr4| jd ks0t‚d S || jksLt|| jfƒ‚d | _| jd k rj| tƒ ¡ n| ¡ d S )Nrb ) Z cancelledrc ra r r r rG ÚBrokenPipeErrorr8 )r rj r% r% r& rƒ œ s
z(_ProactorWritePipeTransport._pipe_closed)r* rR rS r rƒ rU r% r% )r$ r& r – s r c @ s e Zd ZdZdd„ Zdd„ ZdS )Ú _ProactorDuplexPipeTransportzTransport for duplex pipes.c C s dS )NFr% )r r% r% r& r| ± s z*_ProactorDuplexPipeTransport.can_write_eofc C s t ‚d S )N)ÚNotImplementedError)r r% r% r& r} ´ s z&_ProactorDuplexPipeTransport.write_eofN)r* rR rS rT r| r} r% r% r% r& r… ¬ 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 )N)r
r r Z _set_nodelay)r r r r r! r" r# )r$ r% r& r ¿ s z!_ProactorSocketTransport.__init__c C s¨ || j d< y| ¡ | j d< W n4 tjtfk
rP | j ¡ rLtjd|dd Y nX d| j kr¤y| ¡ | j d< W n4 tjtfk
r¢ | j ¡ ržtjd|dd Y nX d S )NrO Zsocknamezgetsockname() failed on %rT)r? Úpeernamezgetpeername() failed on %r)
r3 Z getsocknamerO ÚerrorÚAttributeErrorr rD r rt Z getpeername)r r r% r% r& r Ä s
z#_ProactorSocketTransport._set_extrac C s dS )NTr% )r r% r% r& r| Ö s z&_ProactorSocketTransport.can_write_eofc C s2 | j s | jrd S d| _| jd kr.| j tj¡ d S )NT)r r r r rM rO ry )r r% r% r& r} Ù s
z"_ProactorSocketTransport.write_eof)NNN)
r* rR rS rT r Z
_SendfileModeZ
TRY_NATIVEZ_sendfile_compatibler r r| r} rU r% r% )r$ r& r‡ ¸ s r‡ c sä e Zd Z‡ fdd„Zd1dd„Zd2ddddddœdd „Zd3d
d „Zd4d d
„Zd5dd„Z‡ fdd„Z dd„ Z
dd„ Z dd„ Z dd„ Z
dd„ Zd d „ Zd d „ Zd d!„ Zd"d#„ Zd6d$d%„Zd&d'„ Zd7d)d*„Zd+d,„ Zd-d.„ Zd/d0„ Z‡ ZS )8r c sH t ƒ ¡ t d|jj¡ || _|| _d | _i | _ |
| ¡ | ¡ d S )NzUsing proactor: %s) r
r r rE r$ r* rd Ú _selectorÚ_self_reading_futureÚ_accept_futuresZset_loopÚ_make_self_pipe)r Zproactor)r$ r% r& r ã s
z BaseProactorEventLoop.__init__Nc C s t | |||||ƒS )N)r‡ )r r r r! r" r# r% r% r& Ú_make_socket_transportí s
z,BaseProactorEventLoop._make_socket_transportF)Ú server_sideÚserver_hostnamer" r# Ússl_handshake_timeoutc C s0 t j| |||||| d}
t| ||
||d |
jS )N)r’ )r" r# )r Z SSLProtocolr‡ Z_app_transport) r Zrawsockr Ú
sslcontextr! r r‘ r" r# r’ Z ssl_protocolr% r% r& Ú_make_ssl_transportò s
z)BaseProactorEventLoop._make_ssl_transportc C s t | ||||ƒS )N)r… )r r r r! r" r% r% r& Ú_make_duplex_pipe_transportÿ s z1BaseProactorEventLoop._make_duplex_pipe_transportc C s t | ||||ƒS )N)rV )r r r r! r" r% r% r& Ú_make_read_pipe_transport s z/BaseProactorEventLoop._make_read_pipe_transportc C s t | ||||ƒS )N)r )r r r r! r" r% r% r& Ú_make_write_pipe_transport s z0BaseProactorEventLoop._make_write_pipe_transportc sP | ¡ rtdƒ‚| ¡ r d S | ¡ | ¡ | j ¡ d | _d | _tƒ ¡ d S )Nz!Cannot close a running event loop) Z
is_runningrs Ú is_closedÚ_stop_accept_futuresÚ_close_self_piperd r8 r‹ r
)r )r$ r% r& r8 s
zBaseProactorEventLoop.closec à s | j ||¡I d H S )N)rd re )r r Únr% r% r& Ú sock_recv s z BaseProactorEventLoop.sock_recvc à s | j ||¡I d H S )N)rd Z recv_into)r r Zbufr% r% r& Úsock_recv_into# s z$BaseProactorEventLoop.sock_recv_intoc à s | j ||¡I d H S )N)rd rz )r r r] r% r% r& Ú sock_sendall& s z"BaseProactorEventLoop.sock_sendallc à s | j ||¡I d H S )N)rd Zconnect)r r Zaddressr% r% r& Ú sock_connect) s z"BaseProactorEventLoop.sock_connectc à s | j |¡I d H S )N)rd Úaccept)r r r% r% r& Ú sock_accept, s z!BaseProactorEventLoop.sock_acceptc
à s( y | ¡ }W n2 ttjfk
r> } zt d¡‚W d d }~X Y nX yt |¡j}W n, t k
r| } zt d¡‚W d d }~X Y nX |r†|n|}|s’dS t
|dƒ}|r®t
|| |ƒn|} t
||ƒ}d}
zLxFt
| | |ƒ}|dkrÞ|
S | j ||||¡I d H ||7 }|
|7 }
qÄW W d |
dkr"|
|¡ X d S )Nznot a regular filer l ÿÿ )r, rŠ ÚioÚUnsupportedOperationr ZSendfileNotAvailableErrorÚosÚfstatÚst_sizerC Úminrd ÚsendfileÚseek) r r ÚfileÚoffsetÚcountr, ÚerrZfsizeZ blocksizeZend_posZ
total_sentr% r% r& Ú_sock_sendfile_native/ s2
z+BaseProactorEventLoop._sock_sendfile_nativec à sV | ¡ }| ¡ | ¡ I d H z | j|j|||ddI d H S | ¡ |rP| ¡ X d S )NF)Zfallback)rZ r[ r Z
sock_sendfiler r€ r^ )r Ztransprª r« r¬ r^ r% r% r& Ú_sendfile_nativeL s z&BaseProactorEventLoop._sendfile_nativec C sL | j d k r| j ¡ d | _ | j ¡ d | _| j ¡ d | _| jd8 _d S )Nr )rŒ r7 Ú_ssockr8 Ú_csockÚ
_internal_fds)r r% r% r& rš X s
z&BaseProactorEventLoop._close_self_pipec C sF t ¡ \| _| _| j d¡ | j d¡ | jd7 _| | j¡ d S )NFr )rO Z
socketpairr° r± Z setblockingr² r Ú_loop_self_reading)r r% r% r& rŽ b s
z%BaseProactorEventLoop._make_self_pipec
C s„ y$|d k r| ¡ | j | jd¡}W nH tjk
r: d S tk
rl } z| d|| dœ¡ W d d }~X Y nX || _| | j
¡ d S )Ni z.Error on reading from the event loop self pipe)r@ rA r ) rc rd re r° r rh r_ rF rŒ ri r³ )r r{ rH r% r% r& r³ j s z(BaseProactorEventLoop._loop_self_readingc C s> y| j d¡ W n( tk
r8 | jr4tjddd Y nX d S )Nó z3Fail to write a null byte into the self-pipe socketT)r? )r± rz rC Ú_debugr rE )r r% r% r& Ú_write_to_self| s z$BaseProactorEventLoop._write_to_selféd c s( d‡ ‡‡‡‡‡‡fdd„ ‰ ˆ ˆ ¡ d S )Nc
s$ yˆ| d k rn| ¡ \}}ˆjr,t dˆ||¡ ˆƒ }ˆd k rXˆj||ˆdd|iˆˆd nˆj||d|iˆd ˆ ¡ rzd S ˆj ˆ¡} W n~ t k
rè } zBˆ
¡ dkrˆ d|ˆdœ¡ ˆ ¡ nˆjrØtjd ˆdd
W d d }~X Y n8 t
jk
r ˆ ¡ Y nX | ˆjˆ
¡ < | ˆ ¡ d S ) Nz#%r got a new connection from %r: %rTrˆ )r r" r# r’ )r" r# éÿÿÿÿzAccept failed on a socket)r@ rA rO zAccept failed on socket %r)r? )rc rµ r rE r” r r˜ rd r rC r, rF r8 r rh r ri )r{ ZconnZaddrr rH )r Úprotocol_factoryr r# r r’ r“ r% r& r ‰ s@
z2BaseProactorEventLoop._start_serving..loop)N)r )r r¹ r r“ r# Zbacklogr’ r% )r r¹ r r# r r’ r“ r& Ú_start_serving… s %z$BaseProactorEventLoop._start_servingc C s d S )Nr% )r Z
event_listr% r% r& Ú_process_events° s z%BaseProactorEventLoop._process_eventsc C s* x| j ¡ D ] }| ¡ q W | j ¡ d S )N)r Úvaluesr7 Úclear)r Úfuturer% r% r& r™ ´ s z*BaseProactorEventLoop._stop_accept_futuresc C s6 | j | ¡ d ¡}|r | ¡ | j |¡ | ¡ d S )N)r Úpopr, r7 rd Ú
_stop_servingr8 )r r r¾ r% r% r& rÀ ¹ s
z#BaseProactorEventLoop._stop_serving)NNN)N)NN)NN)NN)N)NNr· N)r* rR rS r r r” r• r– r— r8 rœ r rž rŸ r¡ r® r¯ rš rŽ r³ r¶ rº r» r™ rÀ rU r% r% )r$ r& r á s<
)r ) rT Ú__all__r¢ r¤ rO r: Ú r r r r r r r
Úlogr Z_FlowControlMixinZ
BaseTransportr Z
ReadTransportrV ZWriteTransportrk r Z Transportr… r‡ Z
BaseEventLoopr r% r% r% r& Ú s: m
'