­ ­ ­ ­ ­ ­ ­ ­ ­ ­ ­ ­ ­ ­ ­ ­ ­ ­ 3 \ @s8dgZddlZddlZddlZddlZddlZddlZy ddlZWnek rZdZYnXddl m Z ddl m Z ddl m Z ddl m Z ddl mZdd l mZdd l mZdd l mZdd lmZdd lmZddZGddde jZGdddejejZGdddeZGdddeZGdddeZdS)BaseSelectorEventLoopN) base_events)compat) constants)events)futures) selectors) transports)sslproto) coroutine)loggerc Cs6y|j|}Wntk r"dSXt|j|@SdS)NF)get_keyKeyErrorboolr)selectorfdZeventkeyr//usr/lib64/python3.6/asyncio/selector_events.py_test_selector_event s rcsneZdZdNfdd ZdOdddddZdPdddddd d Zdddddd d ZdQd dZfddZddZ ddZ ddZ ddZ ddZ ddZdRddZdSd d!ZedTd"d#Zd$d%Zd&d'Zd(d)Zd*d+Zd,d-Zd.d/Zd0d1Zd2d3Zd4d5Zd6d7Zd8d9Zd:d;Zdd?Z d@dAZ!dBdCZ"dDdEZ#dFdGZ$dHdIZ%dJdKZ&dLdMZ'Z(S)UrNcsFtj|dkrtj}tjd|jj||_|j t j |_ dS)NzUsing selector: %s) super__init__r ZDefaultSelectorr debug __class____name__ _selector_make_self_pipeweakrefWeakValueDictionary _transports)selfr)rrrr1s zBaseSelectorEventLoop.__init__)extraservercCst||||||S)N)_SelectorSocketTransport)r!sockprotocolwaiterr"r#rrr_make_socket_transport;s z,BaseSelectorEventLoop._make_socket_transportF) server_sideserver_hostnamer"r#c CsNtjs"|j||||||||dStj||||||} t||| ||d| jS)N)r)r*r"r#)r"r#)r Z_is_sslproto_available_make_legacy_ssl_transportZ SSLProtocolr$Z_app_transport) r!rawsockr& sslcontextr'r)r*r"r#Z ssl_protocolrrr_make_ssl_transport@s   z)BaseSelectorEventLoop._make_ssl_transportc Cst||||||||| S)N)_SelectorSslTransport) r!r,r&r-r'r)r*r"r#rrrr+Os z0BaseSelectorEventLoop._make_legacy_ssl_transportcCst||||||S)N)_SelectorDatagramTransport)r!r%r&addressr'r"rrr_make_datagram_transportYsz.BaseSelectorEventLoop._make_datagram_transportcsL|jrtd|jrdS|jtj|jdk rH|jjd|_dS)Nz!Cannot close a running event loop)Z is_running RuntimeError is_closed_close_self_pipercloser)r!)rrrr6^s   zBaseSelectorEventLoop.closecCstdS)N)NotImplementedError)r!rrr _socketpairisz!BaseSelectorEventLoop._socketpaircCsB|j|jj|jjd|_|jjd|_|jd8_dS)Nr)_remove_reader_ssockfilenor6_csock _internal_fds)r!rrrr5ls   z&BaseSelectorEventLoop._close_self_pipecCsN|j\|_|_|jjd|jjd|jd7_|j|jj|jdS)NFr)r8r:r< setblockingr= _add_readerr;_read_from_self)r!rrrrts   z%BaseSelectorEventLoop._make_self_pipecCsdS)Nr)r!datarrr_process_self_data|sz(BaseSelectorEventLoop._process_self_datac CsVxPy |jjd}|sP|j|Wqtk r8wYqtk rLPYqXqWdS)Ni)r:recvrBInterruptedErrorBlockingIOError)r!rArrrr@s z%BaseSelectorEventLoop._read_from_selfc CsJ|j}|dk rFy|jdWn(tk rD|jr@tjdddYnXdS)Nz3Fail to write a null byte into the self-pipe socketT)exc_info)r<sendOSError_debugr r)r!Zcsockrrr_write_to_selfsz$BaseSelectorEventLoop._write_to_selfdcCs |j|j|j|||||dS)N)r?r;_accept_connection)r!protocol_factoryr%r-r#backlogrrr_start_servingsz$BaseSelectorEventLoop._start_servingc Csxt|D]}y0|j\}}|jr2tjd||||jdWntttfk rXdSt k r} z^| j t j t j t j t jfkr|jd| |d|j|j|jtj|j|||||nWYdd} ~ Xq Xd|i} |j||| ||} |j| q WdS)Nz#%r got a new connection from %r: %rFz&socket.accept() out of system resource)message exceptionsocketpeername)rangeacceptrJr rr>rErDConnectionAbortedErrorrIerrnoZEMFILEZENFILEZENOBUFSZENOMEMcall_exception_handlerr9r;Z call_laterrZACCEPT_RETRY_DELAYrP_accept_connection2Z create_task) r!rNr%r-r#rO_connaddrexcr"rVrrrrMs4     z(BaseSelectorEventLoop._accept_connectionc csd}d}yj|}|j}|r6|j||||d||d}n|j|||||d}y|EdHWn|jYnXWn\tk r} z@|jrd| d} |dk r|| d<|dk r|| d<|j| WYdd} ~ XnXdS)NT)r'r)r"r#)r'r"r#z3Error on transport creation for incoming connection)rQrRr& transport) create_futurer.r(r6 ExceptionrJrY) r!rNr\r"r-r#r&r_r'r^contextrrrrZs4 z)BaseSelectorEventLoop._accept_connection2c Cs@y|j|}Wntk r"YnX|jsX|j|j }\}}|jj ||tjB||f|dk r|j dS)N) _check_closedrHandlerrrregisterr EVENT_READrAmodifycancel) r!rcallbackargshandlermaskreaderwriterrrrr?s  z!BaseSelectorEventLoop._add_readerc Cs|jr dSy|jj|}Wntk r0dSX|j|j}\}}|tjM}|sb|jj|n|jj ||d|f|dk r|j dSdSdS)NFT) r4rrrrrAr ri unregisterrjrk)r!rrrorprqrrrr9s z$BaseSelectorEventLoop._remove_readerc Gs|jtj|||}y|jj|}Wn*tk rP|jj|tjd|fYn>X|j|j }\}}|jj ||tjB||f|dk r|j dS)N) rfrrgrrrrhr EVENT_WRITErArjrk) r!rrlrmrnrrorprqrrr _add_writers  z!BaseSelectorEventLoop._add_writerc Cs|jr dSy|jj|}Wntk r0dSX|j|j}\}}|tjM}|sb|jj|n|jj |||df|dk r|j dSdSdS)NFT) r4rrrrrAr rsrrrjrk)r!rrrorprqrrr_remove_writer,s z$BaseSelectorEventLoop._remove_writercGs|j||j||f|S)N)rer?)r!rrlrmrrr add_readerCs z BaseSelectorEventLoop.add_readercCs|j||j|S)N)rer9)r!rrrr remove_readerHs z#BaseSelectorEventLoop.remove_readercGs|j||j||f|S)N)rert)r!rrlrmrrr add_writerMs z BaseSelectorEventLoop.add_writercCs|j||j|S)N)reru)r!rrrr remove_writerRs z#BaseSelectorEventLoop.remove_writercCs6|jr|jdkrtd|j}|j|d|||S)Nrzthe socket must be non-blocking)rJ gettimeout ValueErrorr` _sock_recv)r!r%nfutrrr sock_recvWs zBaseSelectorEventLoop.sock_recvcCs|dk r|j||jrdSy|j|}Wn`ttfk rb|j}|j||j||||Yn6tk r}z|j |WYdd}~Xn X|j |dS)N) rw cancelledrCrErDr;rvr|ra set_exception set_result)r!r~ registered_fdr%r}rArr^rrrr|fs z BaseSelectorEventLoop._sock_recvcCsF|jr|jdkrtd|j}|r8|j|d||n |jd|S)Nrzthe socket must be non-blocking)rJrzr{r` _sock_sendallr)r!r%rAr~rrr sock_sendall{s  z"BaseSelectorEventLoop.sock_sendallcCs|dk r|j||jrdSy|j|}WnDttfk rHd}Yn*tk rp}z|j|dSd}~XnX|t|kr|jdn.|r||d}|j }|j ||j ||||dS)Nr) ryrrHrErDrarlenrr;rxr)r!r~rr%rAr}r^rrrrrs"     z#BaseSelectorEventLoop._sock_sendallccs|jr|jdkrtdttd s2|jtjkrptj||j|j |d}|j sZ|EdH|j d\}}}}}|j }|j ||||EdHS)Nrzthe socket must be non-blockingAF_UNIX)familyprotoloop)rJrzr{hasattrrSrrrZ_ensure_resolvedrdoneresultr` _sock_connect)r!r%r1Zresolvedr[r~rrr sock_connects z"BaseSelectorEventLoop.sock_connectcCs|j}y|j|Wnjttfk rV|jtj|j||j||j |||Yn6t k r}z|j |WYdd}~Xn X|j ddS)N) r;ZconnectrErDZadd_done_callback functoolspartial_sock_connect_donerx_sock_connect_cbrarr)r!r~r%r1rr^rrrrsz#BaseSelectorEventLoop._sock_connectcCs|j|dS)N)ry)r!rr~rrrrsz(BaseSelectorEventLoop._sock_connect_donecCs|jr dSy,|jtjtj}|dkr6t|d|fWnBttfk rPYn6tk rz}z|j |WYdd}~Xn X|j ddS)NrzConnect call failed %s) rZ getsockoptrSZ SOL_SOCKETZSO_ERRORrIrErDrarr)r!r~r%r1errr^rrrrsz&BaseSelectorEventLoop._sock_connect_cbcCs4|jr|jdkrtd|j}|j|d||S)Nrzthe socket must be non-blockingF)rJrzr{r` _sock_accept)r!r%r~rrr sock_accepts z!BaseSelectorEventLoop.sock_acceptcCs|j}|r|j||jr"dSy|j\}}|jdWnVttfk rh|j||j|d|Yn:t k r}z|j |WYdd}~XnX|j ||fdS)NFT) r;rwrrVr>rErDrvrrarr)r!r~Z registeredr%rr\r1r^rrrrs  z"BaseSelectorEventLoop._sock_acceptcCsx~|D]v\}}|j|j}\}}|tj@rN|dk rN|jrD|j|n |j||tj@r|dk r|jrr|j|q|j|qWdS)N) fileobjrAr riZ _cancelledr9Z _add_callbackrsru)r!Z event_listrrorrprqrrr_process_eventss   z%BaseSelectorEventLoop._process_eventscCs|j|j|jdS)N)r9r;r6)r!r%rrr _stop_serving sz#BaseSelectorEventLoop._stop_serving)N)N)N)NNN)NNrL)NNrL)NN))r __module__ __qualname__rr(r.r+r2r6r8r5rrBr@rKrPrMr rZrer?r9rtrurvrwrxryrr|rrrrrrrrrr __classcell__rr)rrr+sR      ( #  cseZdZdZeZdZd fdd ZddZdd Z d d Z d d Z ddZ ddZ ejr`ddZd!ddZddZddZddZddZZS)"_SelectorTransportiNc stj||||jd<|j|jd<d|jkrdy|j|jd<Wn tjk rbd|jd<YnX||_|j|_ ||_ d|_ ||_ |j |_d|_d|_|j dk r|j j||j|j <dS)NrSZsocknamerTTrF)rr_extraZ getsocknameZ getpeernamerSerror_sockr;_sock_fd _protocol_protocol_connected_server_buffer_factory_buffer _conn_lost_closingZ_attachr )r!rr%r&r"r#)rrrrs&      z_SelectorTransport.__init__cCs|jjg}|jdkr |jdn|jr0|jd|jd|j|jdk r|jj rt|jj |jt j }|rz|jdn |jdt|jj |jt j }|rd}nd}|j }|jd||fd d j|S) Nclosedclosingzfd=%sz read=pollingz read=idlepollingZidlezwrite=<%s, bufsize=%s>z<%s> )rrrappendrr_loopr4rrr rirsget_write_buffer_sizejoin)r!inforstatebufsizerrr__repr__2s*       z_SelectorTransport.__repr__cCs|jddS)N) _force_close)r!rrrabortNsz_SelectorTransport.abortcCs ||_dS)N)r)r!r&rrr set_protocolQsz_SelectorTransport.set_protocolcCs|jS)N)r)r!rrr get_protocolTsz_SelectorTransport.get_protocolcCs|jS)N)r)r!rrrrcWsz_SelectorTransport.is_closingcCsT|jr dSd|_|jj|j|jsP|jd7_|jj|j|jj|jddS)NTr) rrr9rrrru call_soon_call_connection_lost)r!rrrr6Zsz_SelectorTransport.closecCs,|jdk r(tjd|t|d|jjdS)Nzunclosed transport %r)source)rwarningswarnResourceWarningr6)r!rrr__del__hs  z_SelectorTransport.__del__Fatal error on transportcCsPt|tjr*|jjrBtjd||ddn|jj||||jd|j |dS)Nz%r: %sT)rG)rQrRr_r&) isinstancerZ_FATAL_ERROR_IGNOREr get_debugr rrYrr)r!r^rQrrr _fatal_errorns   z_SelectorTransport._fatal_errorcCsd|jr dS|jr(|jj|jj|j|jsBd|_|jj|j|jd7_|jj|j |dS)NTr) rrclearrrurrr9rr)r!r^rrrr|s z_SelectorTransport._force_closec CsVz|jr|jj|Wd|jjd|_d|_d|_|j}|dk rP|jd|_XdS)N)rrZconnection_lostrr6rrZ_detach)r!r^r#rrrrs z(_SelectorTransport._call_connection_lostcCs t|jS)N)rr)r!rrrrsz(_SelectorTransport.get_write_buffer_sizecGs"|jr dS|jj||f|dS)N)rrr?)r!rrlrmrrrr?sz_SelectorTransport._add_readeri)NN)r)rrrmax_size bytearrayrrrrrrrrcr6rZPY34rrrrrr?rrr)rrrs"   rcsVeZdZdfdd ZddZddZdd Zd d Zd d ZddZ ddZ Z S)r$Ncsrtj|||||d|_d|_tj|j|jj|j j ||jj|j |j |j |dk rn|jjtj|ddS)NF)rr_eof_pausedrZ _set_nodelayrrrrconnection_mader?r _read_readyr_set_result_unless_cancelled)r!rr%r&r'r"r#)rrrrs    z!_SelectorSocketTransport.__init__cCs>|js |jrdSd|_|jj|j|jjr:tjd|dS)NTz%r pauses reading)rrrr9rrr r)r!rrr pause_readings   z&_SelectorSocketTransport.pause_readingcCsB|js|j rdSd|_|j|j|j|jjr>tjd|dS)NFz%r resumes reading) rrr?rrrrr r)r!rrrresume_readings  z'_SelectorSocketTransport.resume_readingcCs|jr dSy|jj|j}WnDttfk r4Yn|tk r`}z|j|dWYdd}~XnPX|rt|jj |n<|j j rt j d||jj}|r|j j|jn|jdS)Nz$Fatal read error on socket transportz%r received EOF)rrrCrrErDrarr data_receivedrrr r eof_receivedr9rr6)r!rAr^ keep_openrrrrs    z$_SelectorSocketTransport._read_readycCst|tttfs"tdt|j|jr0td|s8dS|j rf|j t j krTt j d|j d7_ dS|jsy|jj|}WnBttfk rYn@tk r}z|j|ddSd}~XnX||d}|sdS|jj|j|j|jj||jdS)Nz1data argument must be a bytes-like object, not %rz%Cannot call write() after write_eof()zsocket.send() raised exception.rz%Fatal write error on socket transport)rbytesr memoryview TypeErrortyperrr3rr!LOG_THRESHOLD_FOR_CONNLOST_WRITESr warningrrrHrErDrarrrtr _write_readyextend_maybe_pause_protocol)r!rAr}r^rrrwrites4     z_SelectorSocketTransport.writecCs|jr dSy|jj|j}Wn\ttfk r4Yntk rx}z*|jj|j |jj |j |dWYdd}~XnTX|r|jd|=|j |js|jj|j |j r|jdn|jr|jjtjdS)Nz%Fatal write error on socket transport)rrrHrrErDrarrurrr_maybe_resume_protocolrrrshutdownrSSHUT_WR)r!r}r^rrrrs&   z%_SelectorSocketTransport._write_readycCs.|js |jrdSd|_|js*|jjtjdS)NT)rrrrrrSr)r!rrr write_eofs  z"_SelectorSocketTransport.write_eofcCsdS)NTr)r!rrr can_write_eof sz&_SelectorSocketTransport.can_write_eof)NNN) rrrrrrrrrrrrrr)rrr$s#r$csdeZdZeZdfdd ZdddZddZd d Zd d Z d dZ ddZ ddZ ddZ ZS)r/NFc stdkrtd|s tj||}|dd} |r<| r<|| d<|j|f| } tj|| ||| d|_||_||_ ||_ d|_ |j j |d|jjrtjd||jj} nd} |j| dS)Nzstdlib ssl module not availableF)r)Zdo_handshake_on_connectr*)r-z%r starts SSL handshake)sslr3r Z_create_transport_contextZ wrap_socketrrr_server_hostname_waiter _sslcontextrrupdaterrr rtime _on_handshake) r!rr,r&r-r'r)r*r"r#Z wrap_kwargsZsslsock start_time)rrrr(s*     z_SelectorSslTransport.__init__cCsD|jdkrdS|jjs:|dk r.|jj|n |jjdd|_dS)N)rrrr)r!r^rrr_wakeup_waiterLs   z$_SelectorSslTransport._wakeup_waiterc"Cs$y|jjWntjk r8|jj|j|j|dStjk r`|jj |j|j|dSt k r}z`|jj rt j d|dd|jj|j|jj|j|jj|j|t|trdSWYdd}~XnX|jj|j|jj|j|jj}t|jds|jr|jjtjkrytj||jWnRtk r}z4|jj rjt j d|dd|jj|j|dSd}~XnX|jj||jj|jj|jdd|_d|_ |jj|j|j!d|_"|jj#|j$j%||jj#|j|jj r |jj&|}t j'd||d dS) Nz%r: SSL handshake failedT)rGZcheck_hostnamez1%r: SSL handshake failed on matching the hostname)peercertcipher compressionZ ssl_objectFz%r: SSL handshake took %.1f msg@@)(rZ do_handshakerSSLWantReadErrorrr?rrSSLWantWriteErrorrt BaseExceptionrr rr9rur6rrraZ getpeercertrrrZ verify_modeZ CERT_NONEZmatch_hostnamerrrr_read_wants_write_write_wants_readrrrrrrr)r!rr^rZdtrrrrVsb                z#_SelectorSslTransport._on_handshakecCsJ|jrtd|jrtdd|_|jj|j|jjrFtjd|dS)Nz#Cannot pause_reading() when closingzAlready pausedTz%r pauses reading) rr3rrr9rrr r)r!rrrrs z#_SelectorSslTransport.pause_readingcCsJ|jstdd|_|jrdS|jj|j|j|jjrFtj d|dS)Nz Not pausedFz%r resumes reading) rr3rrr?rrrr r)r!rrrrs z$_SelectorSslTransport.resume_readingcCs"|jr dS|jr6d|_|j|jr6|jj|j|jy|jj|j }Wnt t t j fk rdYnt jk rd|_|jj|j|jj|j|jYntk r}z|j|dWYdd}~XnTX|r|jj|n@z4|jjrtjd||jj}|rtjdWd|jXdS)NFTz!Fatal read error on SSL transportz%r received EOFz?returning true from eof_received() has no effect when using ssl)rrrrrrtrrrCrrErDrrrrr9rarrrrr rrrr6)r!rAr^rrrrrs4   z!_SelectorSslTransport._read_readycCs(|jr dS|jrszC_SelectorDatagramTransport.get_write_buffer_size..)sumr)r!rrrrsz0_SelectorDatagramTransport.get_write_buffer_sizecCs|jr dSy|jj|j\}}Wnpttfk r8Ynhtk rd}z|jj|WYdd}~Xn<t k r}z|j |dWYdd}~XnX|jj ||dS)Nz&Fatal read error on datagram transport) rrZrecvfromrrErDrIrerror_receivedrarZdatagram_received)r!rAr]r^rrrr sz&_SelectorDatagramTransport._read_readycCsTt|tttfs"tdt|j|s*dS|jrN|d|jfkrNtd|jf|j r|jr|j t j krpt j d|j d7_ dS|js4y&|jr|jj|n|jj||dSttfk r|jj|j|jYnZtk r}z|jj|dSd}~Xn.tk r2}z|j|ddSd}~XnX|jjt||f|jdS)Nz1data argument must be a bytes-like object, not %rz#Invalid address: must be None or %szsocket.send() raised exception.rz'Fatal write error on datagram transport)rrrrrrrrr{rrrr rrrrHsendtorErDrrtr _sendto_readyrIrrrarrr)r!rAr]r^rrrr.s<     z!_SelectorDatagramTransport.sendtocCsx|jr|jj\}}y&|jr,|jj|n|jj||Wqttfk rf|jj||fPYqt k r}z|j j |dSd}~Xqt k r}z|j |ddSd}~XqXqW|j|js|jj|j|jr|jddS)Nz'Fatal write error on datagram transport)rpopleftrrrHrrErD appendleftrIrrrarrrrurrr)r!rAr]r^rrrrUs* z(_SelectorDatagramTransport._sendto_ready)NNN)N) rrr collectionsdequerrrrrrrrr)rrr0 s  'r0)__all__rrXrrSrrr ImportErrorrrrrrr r r Z coroutinesr logr rZ BaseEventLooprZ_FlowControlMixinZ Transportrr$r/r0rrrrsB             ii