\[\@sdZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl Z ddl mZddl mZddl mZddl mZddlmZdd lmZd d gZd Zd ZdZddZddZGdddeZddZddZGdd d ejZ Gdd d ej!Z"dS)aBase implementation of event loop. The event loop can be broken up into a multiplexer (the part responsible for notifying us of I/O events) and the event loop proper, which wraps a multiplexer with functionality for scheduling callbacks, immediately or at a given time in the future. Whenever a public API takes a callback, subsequent positional arguments will be passed to the callback if/when it is called. This avoids the proliferation of trivial lambdas implementing closures. Keyword arguments for the callback are not supported; this is a conscious design decision, leaving the door open for keyword arguments to modify the meaning of the API call itself. N) coroutines)events)futures)tasks) coroutine)logger BaseEventLoopServerdg?cCsH|j}tj|r:t|jtjr:t|jSt|SdS)N) _callbackinspectZismethod isinstance__self__rTaskreprstr)handlecbr)/usr/lib/python3.4/asyncio/base_events.py_format_handle3s $ rcCs4|tjkrdS|tjkr&dSt|SdS)Nzz) subprocessPIPESTDOUTr)fdrrr _format_pipe<s rc@seZdZdZdS) _StopErrorzRaised to stop the event loop.N)__name__ __module__ __qualname____doc__rrrrrEs rcCs|j}|tjkr'|\}}n,|tjkrO|dd\}}ndSd}ttdrx|tjO}nttdr|tjO}ny:tj||d|d|j|@d|j dtj Wn>tj k r}zt d ||fWYdd}~XnXdS) Nr SOCK_NONBLOCK SOCK_CLOEXECfamilytypeprotoflagsz1address must be resolved (IP address), got %r: %s) r&socketZAF_INETAF_INET6hasattrr$r% getaddrinfor'r(ZAI_NUMERICHOSTZgaierror ValueError)sockaddressr&hostportZ type_maskerrrrr_check_resolved_addressIs(  r4cGs tdS)N)r)argsrrr_raise_stop_erroresr6c@sjeZdZddZddZddZddZd d Zd d Ze d dZ dS)r cCs(||_||_d|_g|_dS)Nr)_loopsockets _active_count_waiters)selfloopr8rrr__init__ks   zServer.__init__cCsd|jj|jfS)Nz<%s sockets=%r>) __class__rr8)r;rrr__repr__qszServer.__repr__cCs(|jdk st|jd7_dS)Nr)r8AssertionErrorr9)r;rrr_attachtszServer._attachcCsS|jdkst|jd8_|jdkrO|jdkrO|jndS)Nrr)r9r@r8_wakeup)r;rrr_detachxszServer._detachcCsc|j}|dkrdSd|_x|D]}|jj|q)W|jdkr_|jndS)Nr)r8r7Z _stop_servingr9rB)r;r8r/rrrclose~s    z Server.closecCsC|j}d|_x*|D]"}|js|j|qqWdS)N)r:done set_result)r;waiterswaiterrrrrBs     zServer._wakeupccsT|jdks|jdkr"dStjd|j}|jj||DdHdS)Nr<)r8r:rFuturer7append)r;rHrrr wait_closeds zServer.wait_closedN) rr r!r=r?rArCrDrBrrKrrrrr is      c@seZdZddZddZddZdddd dd d Zd d ddddd dddZdddddZddddZ ddddZ e dddZ ddZ ddZddZdd Zd!d"Zd#d$Zd%d&Zd'd(Zd)d*Zd+d,Zd-d.Zd/d0Zd1d2Zd3d4Zd5d6Zd7d8Zd9d:Zd;d<Zd=d>Zd?d@dAd@dBd@dCd@dDdEZ d@dFdGZ!e dddHdd?d@dBd@dCd@dIddJddddKdLZ"e dMdNZ#e ddd?d@dBd@dCd@dOdPZ$e ddd?e%j&dCe%j'dIddQdRdHddSddTdUZ(e dVdWZ)e dXdYZ*dZd[Z+e d\e,j-d]e,j-d^e,j-d_d d`dadbd@dcddZ.e d\e,j-d]e,j-d^e,j-d_d d`d dbd@dedfZ/dgdhZ0didjZ1dkdlZ2dmdnZ3dodpZ4dqdrZ5dsdtZ6dudvZ7dwdxZ8dS)yr cCsd|_d|_tj|_g|_d|_d|_d|_t j dj |_ d|_ tjj ottjjd|_d|_dS)NrF monotonicZPYTHONASYNCIODEBUGg?)_timer_cancelled_count_closed collectionsdeque_ready _scheduled_default_executorZ _internal_fds_runningtimeZget_clock_infoZ resolution_clock_resolution_exception_handlersysr)ignore_environmentboolosenvironget_debugslow_callback_duration)r;rrrr=s        zBaseEventLoop.__init__cCs,d|jj|j|j|jfS)Nz"<%s running=%s closed=%s debug=%s>)r>r is_running is_closed get_debug)r;rrrr?szBaseEventLoop.__repr__cCs/tj|d|}|jr+|jd=n|S)zDSchedule a coroutine object. Return a task object. r<r)rr_source_traceback)r;ZcoroZtaskrrr create_tasks  zBaseEventLoop.create_taskNextraservercCs tdS)zCreate socket transport.N)NotImplementedError)r;r/protocolrHrfrgrrr_make_socket_transportsz$BaseEventLoop._make_socket_transport server_sideFserver_hostnamec Cs tdS)zCreate SSL transport.N)rh) r;Zrawsockri sslcontextrHrkrlrfrgrrr_make_ssl_transportsz!BaseEventLoop._make_ssl_transportcCs tdS)zCreate datagram transport.N)rh)r;r/rir0rHrfrrr_make_datagram_transportsz&BaseEventLoop._make_datagram_transportcCs tdS)zCreate read pipe transport.N)rh)r;piperirHrfrrr_make_read_pipe_transportsz'BaseEventLoop._make_read_pipe_transportcCs tdS)zCreate write pipe transport.N)rh)r;rprirHrfrrr_make_write_pipe_transportsz(BaseEventLoop._make_write_pipe_transportc Ks tdS)zCreate subprocess transport.N)rh) r;rir5shellstdinstdoutstderrbufsizerfkwargsrrr_make_subprocess_transportsz(BaseEventLoop._make_subprocess_transportcCs tdS)zWrite a byte to self-pipe, to wake up the event loop. This may be called from a different thread. The subclass is responsible for implementing the self-pipe. N)rh)r;rrr_write_to_selfszBaseEventLoop._write_to_selfcCs tdS)zProcess selector events.N)rh)r; event_listrrr_process_eventsszBaseEventLoop._process_eventscCs|jrtdndS)NzEvent loop is closed)rN RuntimeError)r;rrr _check_closeds zBaseEventLoop._check_closedcCsj|j|jr"tdnd|_z.x'y|jWq1tk rTPYq1Xq1Wdd|_XdS)zRun until stop() is called.zEvent loop is running.TNF)r~rTr} _run_oncer)r;rrr run_forevers     zBaseEventLoop.run_forevercCs|jt|tj }tj|d|}|rDd|_n|jt|j |j t|j st dn|j S)a\Run until the Future is done. If the argument is a coroutine, it is wrapped in a Task. WARNING: It would be disastrous to call run_until_complete() with the same coroutine twice -- it would wrap it in two different Tasks and that can't be good. Return the Future's result, or raise its exception. r<Fz+Event loop stopped before Future completed.)r~rrrIrZasyncZ_log_destroy_pendingZadd_done_callbackr6rZremove_done_callbackrEr}result)r;ZfutureZnew_taskrrrrun_until_completes      z BaseEventLoop.run_until_completecCs|jtdS)zStop running the event loop. Every callback scheduled before stop() is called will run. Callbacks scheduled after stop() is called will not run. However, those callbacks will run if run_forever is called again later. N) call_soonr6)r;rrrstopszBaseEventLoop.stopcCs|jrtdn|jr%dS|jrAtjd|nd|_|jj|jj|j }|dk rd|_ |j ddndS)zClose the event loop. This clears the queues and shuts down the executor, but does not wait for the executor to finish. The event loop must not be running. z!Cannot close a running event loopNzClose %rTwaitF) rTr}rNr^rdebugrQclearrRrSZshutdown)r;executorrrrrDs         zBaseEventLoop.closecCs|jS)z*Returns True if the event loop was closed.)rN)r;rrrra5szBaseEventLoop.is_closedcCs|jS)z*Returns True if the event loop is running.)rT)r;rrrr`9szBaseEventLoop.is_runningcCs tjS)zReturn the time according to the event loop's clock. This is a float expressed in seconds since an epoch, but the epoch, precision, accuracy and drift are unspecified and may differ per event loop. )rUrL)r;rrrrU=szBaseEventLoop.timecGs9|j|j|||}|jr5|jd=n|S)a8Arrange for a callback to be called at a given time. Return a Handle: an opaque object with a cancel() method that can be used to cancel the call. The delay can be an int or float, expressed in seconds. It is always relative to the current time. Each callback will be called exactly once. If two callbacks are scheduled for exactly the same time, it undefined which will be called first. Any positional arguments after the callback will be passed to the callback when it is called. rrc)call_atrUrd)r;Zdelaycallbackr5timerrrr call_laterFs  zBaseEventLoop.call_latercGstj|rtdn|jr4|jntj||||}|jrb|jd=ntj |j |d|_ |S)z|Like call_later(), but uses an absolute time. Absolute time corresponds to the event loop's time() method. z(coroutines cannot be used with call_at()rTrc) riscoroutinefunction TypeErrorr^_assert_is_current_event_loopr TimerHandlerdheapqheappushrR)r;whenrr5rrrrr[s     zBaseEventLoop.call_atcGs2|j||dd}|jr.|jd=n|S)aTArrange for a callback to be called as soon as possible. This operates as a FIFO queue: callbacks are called in the order in which they are registered. Each callback will be called exactly once. Any positional arguments after the callback will be passed to the callback when it is called. check_loopTrrc) _call_soonrd)r;rr5rrrrrks   zBaseEventLoop.call_sooncCsytj|rtdn|jr:|r:|jntj|||}|jre|jd=n|jj ||S)Nz*coroutines cannot be used with call_soon()rrc) rrrr^rrHandlerdrQrJ)r;rr5rrrrrrzs   zBaseEventLoop._call_soonc CsHytj}Wntk r(dSYnX||k rDtdndS)acAsserts that this event loop is the current event loop. Non-thread-safe methods of this class make this assumption and will likely behave incorrectly when the assumption is violated. Should only be called when (self._debug == True). The caller is responsible for checking this condition for performance reasons. NzMNon-thread-safe operation invoked on an event loop other than the current one)rZget_event_loopr@r})r;Zcurrentrrrrs    z+BaseEventLoop._assert_is_current_event_loopcGs<|j||dd}|jr.|jd=n|j|S)z"Like call_soon(), but thread-safe.rFrrc)rrdrz)r;rr5rrrrcall_soon_threadsafes    z"BaseEventLoop.call_soon_threadsafecGstj|rtdnt|tjr| s=tt|tj sVt|jrt j d|}|j d|S|j |j }}n|dkr|j}|dkrtj jt}||_qnt j|j||d|S)Nz0Coroutines cannot be used with run_in_executor()r<)rrrrrrr@r _cancelledrrIrFr Z_argsrS concurrentZThreadPoolExecutor _MAX_WORKERSZ wrap_futureZsubmit)r;rrr5frrrrun_in_executors       zBaseEventLoop.run_in_executorcCs ||_dS)N)rS)r;rrrrset_default_executorsz"BaseEventLoop.set_default_executorc Csd||fg}|r-|jd|n|rG|jd|n|ra|jd|n|r{|jd|ndj|}tjd||j}tj||||||} |j|} d|| d | f}| |jkr tj|n tj|| S) Nz%s:%rz family=%rztype=%rzproto=%rzflags=%rz, zGet address info %sz(Getting address info %s took %.3f ms: %rg@@) rJjoinrrrUr*r-r_info) r;r1r2r&r'r(r)msgt0Zaddrinfodtrrr_getaddrinfo_debugs(  z BaseEventLoop._getaddrinfo_debugr&rr'r(r)c CsW|jr.|jd|j||||||S|jdtj||||||SdS)N)r^rrr*r-)r;r1r2r&r'r(r)rrrr-s  zBaseEventLoop.getaddrinfocCs|jdtj||S)N)rr* getnameinfo)r;Zsockaddrr)rrrrszBaseEventLoop.getnameinfosslr/ local_addrc#s0| dk r"| r"tdn| dkrR|rR|sItdn|} n|dk sj|dk r|dk rtdn|j||d|dtjd|d|} | g} | dk r|jd|dtjd|d|| } | j| nd} tj| d |DdH| j}|s@td n| dk rp| j}|sptd qpng}xF|D]\}}}}}ytjd|d|d|}|j d | dk rpx|D]\}}}}}y|j |PWqtk rU}z9t|j d j ||j j}|j|WYdd}~XqXqW|jd}w}n|jrtjd ||n|j||DdHWnntk r}z+|dk r|jn|j|WYdd}~Xq}|dk r|jnYq}XPq}Wt|dkr:|dqt|dtfdd|Drv|dntdj djdd|Dn|dkrtdn|j d |j|||| DdH\}}|jr&|jd}tjd|||||n||fS)aConnect to a TCP server. Create a streaming transport connection to a given Internet host and port: socket family AF_INET or socket.AF_INET6 depending on host (or family if specified), socket type SOCK_STREAM. protocol_factory must be a callable returning a protocol instance. This method is a coroutine which will try to establish the connection in the background. When successful, the coroutine returns a (transport, protocol) pair. Nz+server_hostname is only meaningful with sslz:You must set server_hostname when using ssl without a hostz8host/port and sock can not be specified at the same timer&r'r(r)r<z!getaddrinfo() returned empty listFz2error while attempting to bind on address {!r}: {}zconnect %r to %rrrc3s!|]}t|kVqdS)N)r).0exc)modelrr >sz2BaseEventLoop.create_connection..zMultiple exceptions: {}z, css|]}t|VqdS)N)r)rrrrrrCsz5host and port was not specified and no sock specifiedr*z%r connected to %s:%r: (%r, %r))r.r-r* SOCK_STREAMrJrrrOSError setblockingbinderrnoformatstrerrorlowerrDr^rr sock_connectlenrallr_create_connection_transportZget_extra_info)r;protocol_factoryr1r2rr&r(r)r/rrlf1Zfsf2infosZ laddr_infos exceptionsr'Zcnamer0_Zladdrr transportrir)rrcreate_connections             $         %     zBaseEventLoop.create_connectionc cs|}tjd|}|rct|tr6dn|}|j||||ddd|}n|j|||}|DdH||fS)Nr<rkFrl)rrIrrZrnrj) r;r/rrrlrirHrmrrrrrUs  z*BaseEventLoop._create_connection_transportc#sDp s?|dkr'tdn||fdff}nEtj}xdfdffD]\} } | dk rdt| trt| dkstd|jd|dtj d |d || DdH} | st d nxY| D]N\} } }} }| |f}||kr4ddg||s z:BaseEventLoop.create_datagram_endpoint..zcan not get address informationFr<z@Datagram endpoint local_addr=%r remote_addr=%r created: (%r, %r)z2Datagram endpoint remote_addr=%r created: (%r, %r))NN)r.rO OrderedDictrtuplerr@r-r*Z SOCK_DGRAMritems setsockopt SOL_SOCKET SO_REUSEADDRrrrrDrJrrIror^rrr)r;rrrr&r(r)Zaddr_pairs_infoZ addr_infosidxZaddrrZfamrZpror0rrZ local_addressZremote_addressr/Zr_addrrrirHrr)rrrcreate_datagram_endpointesv   % $    "              z&BaseEventLoop.create_datagram_endpointbacklogr reuse_addressc cst|trtdn|dk s6|dk rs|dk rQtdnttdd} | dkrtjdkotj dk} ng} |dkrd}n|j ||d |d tj d dd |DdH} | st d nd} zPxC| D];}|\}}}}}ytj|||}Wn@tj k rx|jrqtjd|||ddnwYnX| j|| r|jtjtjdn|| krttdr|jtjtjdny|j|Wqt k r<}z*t |jd||jjfWYdd}~XqXqWd} Wd| sox| D]}|jqXWnXn$|dkrtdn|g} t|| }x>| D]6}|j||jd|j ||||qW|jrtj!d|n|S)zCreate a TCP server bound to host and port. Return a Server object which can be used to stop the service. This method is a coroutine. z*ssl argument must be an SSLContext or NoneNz8host/port and sock can not be specified at the same timer+rposixcygwinr&r'r(r)z!getaddrinfo() returned empty listFz:create_server() failed to create socket.socket(%r, %r, %r)exc_infoT IPPROTO_IPV6z0error while attempting to bind on address %r: %sz)Neither host/port nor sock were specifiedz %r is serving)"rrZrr.getattrr*r[namerXplatformr-rrerrorr^rwarningrJrrrr,rZ IPV6_V6ONLYrrrrrDr ZlistenrZ_start_servingr)r;rr1r2r&r)r/rrrr+r8rZ completedZresZafZsocktyper(Z canonnameZsar3rgrrr create_serversp   !          0        zBaseEventLoop.create_serverccsk|}tjd|}|j|||}|DdH|jratjd|j||n||fS)Nr<z Read pipe %r connected: (%r, %r))rrIrqr^rrfileno)r;rrprirHrrrrconnect_read_pipe s    zBaseEventLoop.connect_read_pipeccsk|}tjd|}|j|||}|DdH|jratjd|j||n||fS)Nr<z!Write pipe %r connected: (%r, %r))rrIrrr^rrr)r;rrprirHrrrrconnect_write_pipes    z BaseEventLoop.connect_write_pipecCs|g}|dk r/|jdt|n|dk rd|tjkrd|jdt|nL|dk r|jdt|n|dk r|jdt|ntjdj|dS)Nzstdin=%szstdout=stderr=%sz stdout=%sz stderr=%s )rJrrrrrr)r;rrtrurvrrrr_log_subprocess#s    zBaseEventLoop._log_subprocessrtrurvuniversal_newlinesrsTrwc kst|ttfs$tdn|r9tdn|sNtdn|dkritdn|} |jrd|} |j| |||n|j| |d||||| DdH} |jrtjd| | fn| | fS) Nzcmd must be a stringz universal_newlines must be Falsezshell must be Truerzbufsize must be 0zrun shell command %rTz%s: %r) rbytesrr.r^rryrr) r;rcmdrtrurvrrsrwrxri debug_logrrrrsubprocess_shell0s"    # zBaseEventLoop.subprocess_shellc os|rtdn|r*tdn|dkrEtdn|f| } x?| D]7} t| ttfsYtdt| jqYqYW|} |jrd|}|j||||n|j | | d||||| DdH}|jrt j d||fn|| fS) Nz universal_newlines must be Falsezshell must be Falserzbufsize must be 0z8program arguments must be a bytes or text string, not %szexecute program %rFz%s: %r) r.rrrrr'rr^rryrr)r;rZprogramrtrurvrrsrwr5rxZ popen_argsargrirrrrrsubprocess_execIs*       zBaseEventLoop.subprocess_execcCs>|dk r1t| r1tdj|n||_dS)aSet handler as the new event loop exception handler. If handler is None, the default exception handler will be set. If handler is a callable object, it should have a signature matching '(loop, context)', where 'loop' will be a reference to the active event loop, 'context' will be a dict object (see `call_exception_handler()` documentation for details about context). Nz/A callable object or None is expected, got {!r})callablerrrW)r;Zhandlerrrrset_exception_handlergs z#BaseEventLoop.set_exception_handlerc Cs|jd}|sd}n|jd}|dk rTt|||jf}nd}|g}xt|D]}|d krqpn||}|dkrdjtj|}d}||j7}n t|}|j d j ||qpWt j d j|d |dS) a@Default exception handler. This is called when an exception occurs and no exception handler is set, and can be called by a custom exception handler that wants to defer to the default behavior. The context parameter has the same meaning as in `call_exception_handler()`. messagez!Unhandled exception in event loop exceptionNFZsource_tracebackrz+Object created at (most recent call last): z{}: {} r>message exception) r]r' __traceback__sortedr traceback format_listrstriprrJrrr) r;contextrrrZ log_linesrvaluetbrrrdefault_exception_handlerxs&        z'BaseEventLoop.default_exception_handlercCs|jdkrKy|j|Wqtk rGtjdddYqXny|j||Wnstk r}zSy&|jidd6|d6|d6Wn%tk rtjd ddYnXWYdd}~XnXdS) aCall the current event loop's exception handler. The context argument is a dict containing the following keys: - 'message': Error message; - 'exception' (optional): Exception object; - 'future' (optional): Future instance; - 'handle' (optional): Handle instance; - 'protocol' (optional): Protocol instance; - 'transport' (optional): Transport instance; - 'socket' (optional): Socket instance. New keys maybe introduced in the future. Note: do not overload this method in an event loop subclass. For custom exception handling, use the `set_exception_handler()` method. Nz&Exception in default exception handlerrTz$Unhandled error in exception handlerrrrzeException in default exception handler while handling an unexpected error in custom exception handler)rWr Exceptionrr)r;rrrrrcall_exception_handlers"     z$BaseEventLoop.call_exception_handlercCsXt|tjstd|jr+dSt|tj sDt|jj|dS)z3Add a Handle to _scheduled (TimerHandle) or _ready.zA Handle is required hereN)rrrr@rrrQrJ)r;rrrr _add_callbacks  zBaseEventLoop._add_callbackcCs|j||jdS)z6Like _add_callback() but called from a signal handler.N)rrz)r;rrrr_add_callback_signalsafes z&BaseEventLoop._add_callback_signalsafecCs|jr|jd7_ndS)z3Notification that a TimerHandle has been cancelled.rN)rRrM)r;rrrr_timer_handle_cancelleds z%BaseEventLoop._timer_handle_cancelledcCsXt|j}|tkr|j|tkrg}x3|jD](}|jrYd|_q>|j|q>Wtj|||_d|_nJxG|jr|jdjr|jd8_tj |j}d|_qWd}|j rd}n5|jr#|jdj }t d||j }n|jr|dkr|j }|jj|}|j |}|dkr~tj} n tj} t|} |dkrtj| d|d| q(| rtj| d|d|d| q(|dkr(tj| d |d|dq(n|jj|}|j||j |j} xX|jr|jd}|j | krtPntj |j}d|_|j j|qKWt|j } xt| D]} |j j}|jrqn|jr@|j }|j|j |}||jkrJtjd t||qJq|jqWd}dS) zRun one full iteration of the event loop. This calls all currently ready callbacks, polls for I/O, schedules the resulting callbacks, and finally schedules 'call_later' callbacks. FrrNg?zpoll took %.3f ms: %s eventsg@@z$poll %.3f ms took %.3f ms: %s eventsz"poll %.3f ms took %.3f ms: timeoutzExecuting %s took %.3f seconds)rrR_MIN_SCHEDULED_TIMER_HANDLESrM%_MIN_CANCELLED_TIMER_HANDLES_FRACTIONrrJrheapifyheappoprQZ_whenmaxrUr^Z _selectorZselectloggingINFODEBUGrlogr|rVrangepopleftZ_runr_rr)r;Z sched_countZ new_scheduledrZtimeoutrrr{rlevelZneventZend_timeZntodoirrrrs                               zBaseEventLoop._run_oncecCs|jS)N)r^)r;rrrrb<szBaseEventLoop.get_debugcCs ||_dS)N)r^)r;Zenabledrrr set_debug?szBaseEventLoop.set_debug)9rr r!r=r?rerjrnrorqrrrryrzr|r~rrrrDrar`rUrrrrrrrrrr-rrrrr*Z AF_UNSPECZ AI_PASSIVErrrrrrrrrrrrrrrrbrrrrrr s                    ! y Q   L     # 0    _ )#r"rOZconcurrent.futuresrrrrr[r*rrUrrXrrrrrrrr__all__rrrrr BaseExceptionrr4r6ZAbstractServerr ZAbstractEventLoopr rrrrs8               /