U ,aT@s*dZddlZddlZddlZddlZddlZddlZddlZz ddlZWne k r`dZYnXddl m Z ddl m Z ddl m Z ddl mZddl mZdd l mZdd l mZdd l mZdd lmZd dZddZGddde jZGdddejejZGdddeZGdddeZdS))BaseSelectorEventLoopN) base_events) constants)events)futures) protocols)sslproto) transports)trsock)loggercCs8z||}Wntk r$YdSXt|j|@SdSNF)get_keyKeyErrorboolr)selectorfdZeventkeyrd?Z!d@dAZ"dBdCZ#dDdEZ$dFdGZ%dHdIZ&dJdKZ'dLdMZ(dNdOZ)dPdQZ*Z+S)VrNcsFt|dkrt}td|jj||_| t |_ dS)NzUsing selector: %s) super__init__ selectorsZDefaultSelectorr debug __class____name__ _selector_make_self_pipeweakrefZWeakValueDictionary _transports)selfrr rrr6s zBaseSelectorEventLoop.__init__extraservercCst||||||SN)_SelectorSocketTransport)r&rprotocolwaiterr)r*rrr_make_socket_transport@s z,BaseSelectorEventLoop._make_socket_transportF) server_sideserver_hostnamer)r*ssl_handshake_timeoutc Cs0tj||||||| d} t||| ||d| jS)N)r2r()r Z SSLProtocolr,Z_app_transport) r&Zrawsockr- sslcontextr.r0r1r)r*r2Z ssl_protocolrrr_make_ssl_transportEsz)BaseSelectorEventLoop._make_ssl_transportcCst||||||Sr+)_SelectorDatagramTransport)r&rr-addressr.r)rrr_make_datagram_transportRs z.BaseSelectorEventLoop._make_datagram_transportcsL|rtd|rdS|t|jdk rH|jd|_dS)Nz!Cannot close a running event loop)Z is_running RuntimeError is_closed_close_self_pipercloser"r&r'rrr;Ws   zBaseSelectorEventLoop.closecCsB||j|jd|_|jd|_|jd8_dS)Nr)_remove_reader_ssockfilenor;_csock _internal_fdsr<rrrr:bs   z&BaseSelectorEventLoop._close_self_pipecCsNt\|_|_|jd|jd|jd7_||j|jdS)NFr) socketZ socketpairr>r@ setblockingrA _add_readerr?_read_from_selfr<rrrr#js   z%BaseSelectorEventLoop._make_self_pipecCsdSr+rr&datarrr_process_self_datarsz(BaseSelectorEventLoop._process_self_datacCsXz"|jd}|sWqT||Wqtk r:YqYqtk rPYqTYqXqdS)Ni)r>recvrHInterruptedErrorBlockingIOErrorrFrrrrEus z%BaseSelectorEventLoop._read_from_selfcCsN|j}|dkrdSz|dWn(tk rH|jrDtjdddYnXdS)Nz3Fail to write a null byte into the self-pipe socketTexc_info)r@sendOSError_debugr r)r&Zcsockrrr_write_to_selfsz$BaseSelectorEventLoop._write_to_selfdc Cs"|||j||||||dSr+)rDr?_accept_connection)r&protocol_factoryrr3r*backlogr2rrr_start_servingsz$BaseSelectorEventLoop._start_servingc Cst|D]}z0|\}} |jr0td|| ||dWntttfk rZYdSt k r} zd| j t j t j t j t jfkr|d| t|d|||tj|j||||||nW5d} ~ XYqXd| i} |||| |||} || qdS)Nz#%r got a new connection from %r: %rFz&socket.accept() out of system resource)message exceptionrBpeername)rangeacceptrQr rrCrKrJConnectionAbortedErrorrPerrnoZEMFILEZENFILEZENOBUFSZENOMEMcall_exception_handlerr TransportSocketr=r?Z call_laterrZACCEPT_RETRY_DELAYrW_accept_connection2Z create_task) r&rUrr3r*rVr2_connaddrexcr)r\rrrrTsV   z(BaseSelectorEventLoop._accept_connectionc sd}d}zt|}|} |r8|j|||| d|||d}n|j||| ||d}z| IdHWntk rx|YnXWntttfk rYn\tk r} z>|jrd| d} |dk r|| d<|dk r|| d<|| W5d} ~ XYnXdS)NT)r.r0r)r*r2)r.r)r*z3Error on transport creation for incoming connection)rXrYr- transport) create_futurer4r/ BaseExceptionr; SystemExitKeyboardInterruptrQr_) r&rUrcr)r3r*r2r-rfr.recontextrrrrasP z)BaseSelectorEventLoop._accept_connection2c Cs|}t|tsJzt|}Wn*tttfk rHtd|dYnXz|j|}Wntk rlYnX|st d|d|dS)NzInvalid file object: zFile descriptor z is used by transport ) rintr?AttributeErrorr ValueErrorr%r is_closingr8)r&rr?rfrrr_ensure_fd_no_transports z-BaseSelectorEventLoop._ensure_fd_no_transportc Gs|t|||d}z|j|}Wn*tk rR|j|tj|dfYn>X|j|j }\}}|j ||tjB||f|dk r| dSr+) _check_closedrHandler"rrregisterr EVENT_READrGmodifycancel r&rcallbackargsZhandlermaskreaderwriterrrrrDs  z!BaseSelectorEventLoop._add_readercCs|r dSz|j|}Wntk r2YdSX|j|j}\}}|tjM}|sd|j|n|j ||d|f|dk r| dSdSdSNFT) r9r"rrrrGrrt unregisterrurvr&rrrzr{r|rrrr=s z$BaseSelectorEventLoop._remove_readerc Gs|t|||d}z|j|}Wn*tk rR|j|tjd|fYn>X|j|j }\}}|j ||tjB||f|dk r| dSr+) rqrrrr"rrrsr EVENT_WRITErGrurvrwrrr _add_writer%s  z!BaseSelectorEventLoop._add_writercCs|r dSz|j|}Wntk r2YdSX|j|j}\}}|tjM}|sd|j|n|j |||df|dk r| dSdSdSr}) r9r"rrrrGrrr~rurvrrrr_remove_writer4s z$BaseSelectorEventLoop._remove_writercGs|||j||f|Sr+)rprDr&rrxryrrr add_readerKs z BaseSelectorEventLoop.add_readercCs||||Sr+)rpr=r&rrrr remove_readerPs z#BaseSelectorEventLoop.remove_readercGs|||j||f|Sr+)rprrrrr add_writerUs z BaseSelectorEventLoop.add_writercCs||||Sr+)rprrrrr remove_writerZs z#BaseSelectorEventLoop.remove_writerc st||jr"|dkr"tdz ||WSttfk rFYnX|}|}| ||j |||| t |j||IdHSNrthe socket must be non-blocking)rrQ gettimeoutrnrIrKrJrgr?r _sock_recvadd_done_callback functoolspartial_sock_read_done)r&rnfutrrrr sock_recv_s  zBaseSelectorEventLoop.sock_recvcCs||dSr+)rr&rrrrrrtsz%BaseSelectorEventLoop._sock_read_donec Cs|r dSz||}Wn\ttfk r4YdSttfk rLYn6tk rv}z||W5d}~XYn X||dSr+) donerIrKrJrirjrh set_exception set_result)r&rrrrGrerrrrwsz BaseSelectorEventLoop._sock_recvc st||jr"|dkr"tdz ||WSttfk rFYnX|}|}| ||j |||| t |j||IdHSr)rrQrrn recv_intorKrJrgr?r_sock_recv_intorrrr)r&rbufrrrrrsock_recv_intos  z$BaseSelectorEventLoop.sock_recv_intoc Cs|r dSz||}Wn\ttfk r4YdSttfk rLYn6tk rv}z||W5d}~XYn X||dSr+) rrrKrJrirjrhrr)r&rrrnbytesrerrrrsz%BaseSelectorEventLoop._sock_recv_intoc st||jr"|dkr"tdz||}Wnttfk rLd}YnX|t|kr^dS|}| }| t |j ||||j||t||g|IdHSr)rrQrrnrOrKrJlenrgr?rrr_sock_write_doner _sock_sendall memoryview)r&rrGrrrrrr sock_sendalls&    z"BaseSelectorEventLoop.sock_sendallc Cs|r dS|d}z|||d}Wnbttfk rDYdSttfk r\Yn2tk r}z||WYdSd}~XYnX||7}|t|kr| dn||d<dS)Nr) rrOrKrJrirjrhrrr)r&rrZviewposstartrrerrrrs    z#BaseSelectorEventLoop._sock_sendallcst||jr"|dkr"tdttdr8|jtjkrf|j||j|j |dIdH}|d\}}}}}| }| ||||IdHS)NrrAF_UNIX)familyprotoloop) rrQrrnhasattrrBrrZ_ensure_resolvedrrg _sock_connect)r&rr6Zresolvedrbrrrr sock_connects z"BaseSelectorEventLoop.sock_connectc Cs|}z||Wnttfk rV|t|j||||j |||YnNt t fk rnYn6t k r}z| |W5d}~XYn X|ddSr+)r?ZconnectrKrJrrrrr_sock_connect_cbrirjrhrr)r&rrr6rrerrrrs z#BaseSelectorEventLoop._sock_connectcCs||dSr+)rrrrrrsz&BaseSelectorEventLoop._sock_write_donec Cs|r dSz,|tjtj}|dkr6t|d|WnZttfk rPYnNtt fk rhYn6t k r}z| |W5d}~XYn X| ddS)NrzConnect call failed ) rZ getsockoptrBZ SOL_SOCKETZSO_ERRORrPrKrJrirjrhrr)r&rrr6errrerrrrsz&BaseSelectorEventLoop._sock_connect_cbcsBt||jr"|dkr"td|}||d||IdHS)NrrF)rrQrrnrg _sock_accept)r&rrrrr sock_accepts z!BaseSelectorEventLoop.sock_acceptc Cs|}|r|||r"dSz|\}}|dWnnttfk rh|||j|d|YnRt t fk rYn:t k r}z| |W5d}~XYnX| ||fdSr})r?rrr\rCrKrJrrrirjrhrr)r&rZ registeredrrrcr6rerrrr*s  z"BaseSelectorEventLoop._sock_acceptc sp|j|j=|}||IdHz |j|j|||ddIdHWS||r^|||j|j<XdS)NF)Zfallback) r%_sock_fd is_reading pause_reading_make_empty_waiter_reset_empty_waiterresume_readingZ sock_sendfile_sock)r&Ztranspfileoffsetcountrrrr_sendfile_native<s z&BaseSelectorEventLoop._sendfile_nativecCs|D]v\}}|j|j}\}}|tj@rL|dk rL|jrB||n |||tj@r|dk r|jrp||q||qdSr+) fileobjrGrrtZ _cancelledr=Z _add_callbackrr)r&Z event_listrrzrr{r|rrr_process_eventsJs    z%BaseSelectorEventLoop._process_eventscCs|||dSr+)r=r?r;)r&rrrr _stop_servingXsz#BaseSelectorEventLoop._stop_serving)N)N)N)NNN),r! __module__ __qualname__rr/rZSSL_HANDSHAKE_TIMEOUTr4r7r;r:r#rHrErRrWrTrarprDr=rrrrrrrrrrrrrrrrrrrrrr __classcell__rrr'rr0s|        . )rcseZdZdZeZdZdfdd ZddZddZ d d Z d d Z d dZ ddZ ejfddZdddZddZddZddZddZZS) _SelectorTransportiNcst||t||jd<z||jd<Wntk rNd|jd<YnXd|jkrz||jd<Wn tj k rd|jd<YnX||_ | |_ d|_ ||||_||_d|_d|_|jdk r|j||j|j <dS)NrBZsocknamerZFr)rrr r`_extraZ getsocknamerPZ getpeernamerBerrorrr?r_protocol_connected set_protocol_server_buffer_factory_buffer _conn_lost_closingZ_attachr%)r&rrr-r)r*r'rrris,      z_SelectorTransport.__init__cCs|jjg}|jdkr |dn|jr0|d|d|j|jdk r|jst|jj |jt j }|rz|dn |dt|jj |jt j }|rd}nd}| }|d|d |d d d |S) Nclosedclosingzfd=z read=pollingz read=idlepollingZidlezwrite=z<{}> )r r!rappendrr_loopr9rr"rrtrget_write_buffer_sizeformatjoin)r&inforstatebufsizerrr__repr__s0      z_SelectorTransport.__repr__cCs|ddSr+) _force_closer<rrrabortsz_SelectorTransport.abortcCs||_d|_dSNT) _protocolrr&r-rrrrsz_SelectorTransport.set_protocolcCs|jSr+)rr<rrr get_protocolsz_SelectorTransport.get_protocolcCs|jSr+)rr<rrrrosz_SelectorTransport.is_closingcCsT|jr dSd|_|j|j|jsP|jd7_|j|j|j|jddSNTr) rrr=rrrr call_soon_call_connection_lostr<rrrr;sz_SelectorTransport.closecCs,|jdk r(|d|t|d|jdS)Nzunclosed transport )source)rResourceWarningr;)r&Z_warnrrr__del__s z_SelectorTransport.__del__Fatal error on transportcCsNt|tr(|jr@tjd||ddn|j||||jd||dS)Nz%r: %sTrM)rXrYrfr-) rrPr get_debugr rr_rr)r&rerXrrr _fatal_errors  z_SelectorTransport._fatal_errorcCsd|jr dS|jr(|j|j|j|jsBd|_|j|j|jd7_|j|j |dSr) rrclearrrrrr=rrr&rerrrrs z_SelectorTransport._force_closecCsVz|jr|j|W5|jd|_d|_d|_|j}|dk rP|d|_XdSr+)rr;rrrZ_detachrZconnection_lost)r&rer*rrrrs z(_SelectorTransport._call_connection_lostcCs t|jSr+)rrr<rrrrsz(_SelectorTransport.get_write_buffer_sizecGs"|jr dS|jj||f|dSr+)rrrDrrrrrDsz_SelectorTransport._add_reader)NN)r)r!rrmax_size bytearrayrrrrrrrror;warningswarnrrrrrrDrrrr'rr]s    rcseZdZdZejjZd#fdd ZfddZ ddZ d d Z d d Z d dZ ddZddZddZddZddZddZddZfddZdd Zd!d"ZZS)$r,TNcs~d|_t|||||d|_d|_d|_t|j|j |j j ||j |j |j|j|dk rz|j tj|ddSr )_read_ready_cbrr_eof_paused _empty_waiterrZ _set_nodelayrrrrconnection_maderDr _read_readyr_set_result_unless_cancelled)r&rrr-r.r)r*r'rrrs    z!_SelectorSocketTransport.__init__cs.t|tjr|j|_n|j|_t|dSr+)rrZBufferedProtocol_read_ready__get_bufferr_read_ready__data_receivedrrrr'rrr s  z%_SelectorSocketTransport.set_protocolcCs|j o|j Sr+)rrr<rrrrsz#_SelectorSocketTransport.is_readingcCs>|js |jrdSd|_|j|j|jr:td|dS)NTz%r pauses reading)rrrr=rrr rr<rrrrs   z&_SelectorSocketTransport.pause_readingcCs@|js |jsdSd|_||j|j|jrYn4tk rp}z| |dWYdSd}~XYnX|r|j |j n| dS)Nz%r received EOFz1Fatal error: protocol.eof_received() call failed.) rrr rrZ eof_receivedrirjrhrr=rr;)r&Z keep_openrerrrres  z,_SelectorSocketTransport._read_ready__on_eofc Cs6t|tttfs$tdt|j|jr2td|j dk rDtd|sLdS|j rz|j t j krht d|j d7_ dS|jsz|j|}Wnbttfk rYnbttfk rYnJtk r}z||dWYdSd}~XYnX||d}|s dS|j|j|j|j||dS)N/data argument must be a bytes-like object, not z%Cannot call write() after write_eof()z(unable to write; sendfile is in progresssocket.send() raised exception.r%Fatal write error on socket transport)rbytesrrrtyper!rr8rrr!LOG_THRESHOLD_FOR_CONNLOST_WRITESr warningrrrOrKrJrirjrhrrrr _write_readyextend_maybe_pause_protocol)r&rGrrerrrwritezs:      z_SelectorSocketTransport.writec Cs|jr dSz|j|j}Wnttfk r4Ynttfk rLYntk r}z>|j |j |j | |d|jdk r|j|W5d}~XYnnX|r|jd|=||js|j |j |jdk r|jd|jr|dn|jr|jtjdS)Nr)rrrOrrKrJrirjrhrrrrrrr_maybe_resume_protocolrrrrshutdownrBSHUT_WR)r&rrerrrrs2       z%_SelectorSocketTransport._write_readycCs.|js |jrdSd|_|js*|jtjdSr)rrrrrrBrr<rrr write_eofs  z"_SelectorSocketTransport.write_eofcCsdSrrr<rrr can_write_eofsz&_SelectorSocketTransport.can_write_eofcs*t||jdk r&|jtddS)NzConnection is closed by peer)rrrrConnectionErrorrr'rrrs   z._SelectorSocketTransport._call_connection_lostcCs6|jdk rtd|j|_|js0|jd|jS)NzEmpty waiter is already set)rr8rrgrrr<rrrrs    z+_SelectorSocketTransport._make_empty_waitercCs d|_dSr+)rr<rrrrsz,_SelectorSocketTransport._reset_empty_waiter)NNN)r!rrZ_start_tls_compatiblerZ _SendfileModeZ TRY_NATIVEZ_sendfile_compatiblerrrrrrrrrrrrrrrrrrrr'rr,s* %' r,csFeZdZejZd fdd ZddZddZd dd Z d d Z Z S)r5Ncs^t||||||_|j|jj||j|j|j|j |dk rZ|jt j |ddSr+) rr_addressrrrrrDrrrr)r&rrr-r6r.r)r'rrrs  z#_SelectorDatagramTransport.__init__cCstdd|jDS)Ncss|]\}}t|VqdSr+)r).0rGrbrrr szC_SelectorDatagramTransport.get_write_buffer_size..)sumrr<rrrrsz0_SelectorDatagramTransport.get_write_buffer_sizec Cs|jr dSz|j|j\}}Wnttfk r8Yntk rd}z|j|W5d}~XYnTt t fk r|Yn<t k r}z| |dW5d}~XYnX|j ||dS)Nz&Fatal read error on datagram transport)rrZrecvfromrrKrJrPrerror_receivedrirjrhrZdatagram_receivedr&rGrdrerrrrsz&_SelectorDatagramTransport._read_readyc Cst|tttfs$tdt|j|s,dS|jrV|d|jfkrPtd|j|j}|j r|jr|j t j krxt d|j d7_ dS|jslz,|jdr|j|n|j||WdSttfk r|j|j|jYntk r}z|j|WYdSd}~XYnPttfk r6Yn6tk rj}z||dWYdSd}~XYnX|j t||f|!dS)Nrz!Invalid address: must be None or rrrZ'Fatal write error on datagram transport)"rrrrrrr!r rnrrrr rrrrrOsendtorKrJrrr _sendto_readyrPrrrirjrhrrrrrrrrsH      z!_SelectorDatagramTransport.sendtoc Cs|jr|j\}}z*|jdr.|j|n|j||Wqttfk rj|j||fYqYqt k r}z|j |WYdSd}~XYqt t fk rYqtk r}z||dWYdSd}~XYqXq||js|j|j|jr|ddS)NrZr)rpopleftrrrOrrKrJ appendleftrPrrrirjrhrrrrrrrrrrrr*s2  z(_SelectorDatagramTransport._sendto_ready)NNN)N) r!rr collectionsdequerrrrrrrrrr'rr5s  +r5)__all__rr^rrrBrr$r ImportErrorrrrrrr r r logr rrZ BaseEventLooprZ_FlowControlMixinZ Transportrr,r5rrrrsD            1o