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
jZG dd dee
jZG dd dee
jZG dd deZG dd deee
jZG dd deee
jZG dd dejZ 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)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 ddZ 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| jtj|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)selfloopsockprotocolwaiterextraserver) __class__ )
r$ __name__r appendr filenor r r lenr formatjoin)r infor% r% r& __repr__2 s
z#_ProactorBasePipeTransport.__repr__c C s || j d< d S )Npipe)_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 warningswarnResourceWarningr8 )r r% r% r& __del__[ s
z"_ProactorBasePipeTransport.__del__ Fatal error on pipe transportc C sV zDt |tr*| j rBtjd| |dd n| j||| | jd W d | | X d S )Nz%r: %sT)exc_info)message exceptionZ transportr )
isinstanceOSErrorr get_debugr debugcall_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_waiterdone
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| jdr,| jtj | j d | _| j}|d k rX| d | _X d S )Nshutdown)
r Zconnection_losthasattrr 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| j7 }|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 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_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,td| 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 rtd| 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|| _d S |s | d S t| jtjrryt| j| W q~ tk
rn } z| |d d S d }~X Y q~X n | j
| d S )Nz3Fatal error: protocol.buffer_updated() call failed.) rX rW r` 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 sd d }zBy`|d k r2d | _ | r*| }n| | jr@d }d S |dkrLd S | jsf| jj| j d| _ W n t
k
r } z0| js| |d n| j rt
jddd 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
r. | js* Y nX | jsF| 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 rK resultr7 r rX r _proactorrecvr ConnectionAbortedErrorrI rD r rE ConnectionResetErrorrG rC r CancelledErroradd_done_callbackrY r\ )r futr] rH r% r% r& rY s>
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 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 )N)r
r rJ )r argskw)r$ r% r& r . 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 )r] )rB bytes bytearray
memoryview TypeErrortyper* r RuntimeErrorrJ r r Z!LOG_THRESHOLD_FOR_CONNLOST_WRITESr warningr
_loop_writingr _maybe_pause_protocolextend)r r] r% r% r& write2 s*
z%_ProactorBaseWritePipeTransport.writeNc
C sT y|d k r | j d kr | jr d S d | _ d| _|r6| |d krJ| j}d | _|s| jrd| j| jd | jrx| 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 d d }~X Y n0 tk
rN } z| |d W d d }~X Y nX d S )Nr z#Fatal write error on pipe transport)r r r rb r r r r6 r r rM rO SHUT_WRZ_maybe_resume_protocolrc sendrK r- rh rt ru rJ rL rf rG rC rI )r fr] rH r% r% r& rt X s8
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 rr 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 rw rt r{ r| r} r~ r rU r% r% )r$ r& rj ( s &
)rj c s$ e Zd Z fddZdd Z ZS )_ProactorWritePipeTransportc s4 t j|| | jj| jd| _| j| j d S )N ) r
r r rc rd r r rh _pipe_closed)r rk rl )r$ r% r& r s z$_ProactorWritePipeTransport.__init__c C s@ | r d S | jrd S d | _| jd k r4| t n| d S )N)Z cancelledr r r rG BrokenPipeErrorr8 )r ri 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 kry| | j d< W n4 tjtfk
r | j rtjd|dd Y nX d S )NrO Zsocknamezgetsockname() failed on %rT)r? peernamezgetpeername() failed on %r)
r3 Z getsocknamerO errorAttributeErrorr rD r rs 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.| jtj d S )NT)r r r r rM rO rx )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 fddZd1ddZd2dddddddd Zd3d
d Zd4d d
Zd5ddZ fddZ 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 td|jj || _|| _d | _i | _ |
| | d S )NzUsing proactor: %s) r
r r rE r$ r* rc _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_sideserver_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_runningrr is_closed_stop_accept_futures_close_self_piperc r8 r r
)r )r$ r% r& r8 s
zBaseProactorEventLoop.closec s | j ||I d H S )N)rc rd )r r nr% r% r& sock_recv s z BaseProactorEventLoop.sock_recvc s | j ||I d H S )N)rc Z recv_into)r r Zbufr% r% r& sock_recv_into# s z$BaseProactorEventLoop.sock_recv_intoc s | j ||I d H S )N)rc ry )r r r] r% r% r& sock_sendall& s z"BaseProactorEventLoop.sock_sendallc s | j ||I d H S )N)rc Zconnect)r r Zaddressr% r% r& sock_connect) s z"BaseProactorEventLoop.sock_connectc s | j |I d H S )N)rc accept)r r r% r% r& sock_accept, s z!BaseProactorEventLoop.sock_acceptc
s( y | }W n2 ttjfk
r> } ztdW d d }~X Y nX yt|j}W n, t k
r| } ztdW d d }~X Y nX |r|n|}|sdS t
|d}|rt
|| |n|} t
||}d}
zLxFt
| | |}|dkr|
S | j ||||I d H ||7 }|
|7 }
qW W d |
dkr"|
| X d S )Nznot a regular filer l )r, r ioUnsupportedOperationr ZSendfileNotAvailableErrorosfstatst_sizerC minrc sendfileseek) r r fileoffsetcountr, 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 \| _| _| jd | jd | 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 ) rb rc rd r r rg r_ rF r rh r )r rz 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 ry rC _debugr rE )r r% r% r& _write_to_self| s z$BaseProactorEventLoop._write_to_selfd c s( d fdd d S )Nc
s$ y| d k rn| \}}jr,td|| }d k rXj||dd|id nj||d|id rzd S j} W n~ t k
r } zB
dkr d|d njrtjd 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? )rb r r rE r r r rc r rC r, rF r8 r rg r rh )rz 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 rc
_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 ZWriteTransportrj r Z Transportr r Z
BaseEventLoopr r% r% r% r& s: m
'