\c@`sddlmZmZmZddlmZmZddlmZddl m Z ddl Z ddl Z ddl Z ddlZddlZddlZddl mZddlmZddlmZdd lmZdd lmZd d lmZd d lmZmZmZm Z m!Z!m"Z"d dl#m$Z$d dl%m&Z&d dl'm(Z(d dl)m*Z*m+Z+m,Z,m-Z-m.Z.m/Z/m0Z0d dl%m1Z1de2fdYZ3e j4e5Z6dZ7dZ8e7Z9e/ej:j;dddZ<ej:j;dZ=de>fdYZ?dZ@ejAeBeCeCdZDeCeCeCdZEd e>fd!YZFd"e>fd#YZGd$e>fd%YZHd&ZId'd(ZJd)ZKdS(*i(tprint_functiontdivisiontabsolute_import(t defaultdicttdeque(tCancelledError(tpartialN(t string_types(tmerge(tgen(tIOLoop(tEventi(tget_thread_identity(tconnecttlistentCommClosedErrortnormalize_addresstunparse_host_porttget_address_host_port(ttime(tprofile(t SystemMonitor(t get_tracebackttruncate_exceptiontignoringt shutting_downtPeriodicCallbacktparse_timedeltat has_keyword(tprotocolt RPCClosedcB`seZRS((t__name__t __module__(((s/lib/python2.7/site-packages/distributed/core.pyR+scC`s:y!ddl}|jjdSWntk r5dSXdS(NiigeA(tpsutiltvirtual_memoryttotalt ImportError(R!((s/lib/python2.7/site-packages/distributed/core.pytget_total_physical_memory2s   c`sfd}|S(Nc`s dS(N((targstkwargs(texc(s/lib/python2.7/site-packages/distributed/core.pyt_raise<s((R(R)((R(s/lib/python2.7/site-packages/distributed/core.pyt raise_later;ssdistributed.admin.tick.limittdefaulttmssdistributed.admin.pdb-on-errtServercB`seZdZdZdZdddeddZdZdZ dZ dZ e d Z e d Ze d Zdd Zddd ZejedZejdgdZejdZRS(sO Distributed TCP Server Superclass for endpoints in a distributed cluster, such as Worker and Scheduler objects. **Handlers** Servers define operations with a ``handlers`` dict mapping operation names to functions. The first argument of a handler function will be a ``Comm`` for the communication established with the client. Other arguments will receive inputs from the keys of the incoming message which will always be a dictionary. >>> def pingpong(comm): ... return b'pong' >>> def add(comm, x, y): ... return x + y >>> handlers = {'ping': pingpong, 'add': add} >>> server = Server(handlers) # doctest: +SKIP >>> server.listen('tcp://0.0.0.0:8000') # doctest: +SKIP **Message Format** The server expects messages to be dictionaries with a special key, `'op'` that corresponds to the name of the operation, and other key-value pairs as required by the function. So in the example above the following would be good messages. * ``{'op': 'ping'}`` * ``{'op': 'add', 'x': 10, 'y': 20}`` tiic `s<ijd6jd6_jj||dkrdtjjdtj j g}n|_ i_ j j|pitj dt tj_d_d_d_i_|_t_d_d_d_d_tj_d_|p6t j!_"j"_#t$j"dstj%j"t$j"drfd}nfd}t&j'd d!d tjjd dtjjdd|j"_&nt(t)3ddl*m+}t,t-|dj"_WdQXddl*m.} t,t-| dj"_t,d_t,d_t/_0t1jjddj"} | j0dR.cS`sdS(Ni((((s/lib/python2.7/site-packages/distributed/core.pyR>R.itio_looptmonitorsdistributed.admin.tick.intervalR+R,ittickic`st_dS(N(R t thread_id((tself(s/lib/python2.7/site-packages/distributed/core.pytset_thread_idents(s profile.pys selectors.py(<R/t handle_streamthandlerstupdateR3tdasktconfigtgetttypeRtlowertblocked_handlerststream_handlerststrtuuidtuuid4tidt_addresst_listen_addresst_portt_commst deserializeRR@tcounterstdigeststeventst event_countstweakreftWeakSett_ongoing_coroutinestlistenerR tcurrentR?R4thasattrR5RtwatchRR$tcounterR;RRR<tdicttperiodic_callbacksRRt _last_tickt _measure_tickRRBR t coroutinet add_callbacktFalset_Server__stopped( RCRFRMRNtconnection_limitRWR?R6R;R<tpcRD((R5RCs/lib/python2.7/site-packages/distributed/core.pyt__init__ssr    "  &              $      c`s/t_fd}jj|dS(s Start Periodic Callbacks consistently This starts all PeriodicCallbacks stored in self.periodic_callbacks if they are not yet running. It does this safely on the IOLoop. c`s7x0jjD]}|js|jqqWdS(N(Retvaluest is_runningtstart(Rm(RC(s/lib/python2.7/site-packages/distributed/core.pyt start_pcss N(RRfR?Ri(RCRr((RCs/lib/python2.7/site-packages/distributed/core.pytstart_periodic_callbackss cC`sA|js=t|_|jdk r=|jj|jjq=ndS(N(RktTrueR_R3R?RiR6(RC((s/lib/python2.7/site-packages/distributed/core.pyR6s  cC`stt}||j}||_|tkrJtjdt|j|n|jdk rp|jdj |ndS(NsEvent loop was unresponsive in %s for %.2fs. This is often caused by long-running GIL-holding functions or moving large chunks of data. This can cause timeouts and instability.s tick-duration( RRfttick_maximum_delaytloggertinfoRKRRYR3tadd(RCtnowtdiff((s/lib/python2.7/site-packages/distributed/core.pyRgs      cC`st|d The address this Server can be contacted on. s(cannot get address of non-running ServerN(RSR_R3t ValueErrortcontact_address(RC((s/lib/python2.7/site-packages/distributed/core.pytaddress s  cC`s@|js9|jdkr'tdn|jj|_n|jS(s The address this Server is listening on. This may be a wildcard address such as `tcp://0.0.0.0:1234`. s/cannot get listen address of non-running ServerN(RTR_R3Rtlisten_address(RC((s/lib/python2.7/site-packages/distributed/core.pyRs  cC`s+|js$t|j\}|_n|jS(s The port number this Server is listening on. This will raise ValueError if the Server is listening on a non-IP based protocol. (RURR(RCt_((s/lib/python2.7/site-packages/distributed/core.pytport#s cC`sit|jd6|jd6S(NRKRR(RKRRR(RCtcomm((s/lib/python2.7/site-packages/distributed/core.pyR//scC`s|dkr|j}nt|tr<t|j|}n9t|trZt|}n|}t|tsutt ||j d|j d||_ |j j dS(NRWtconnection_args(R3t default_portR{tintRt default_ipttupleRtAssertionErrorRt handle_commRWR_Rq(RCt port_or_addrt listen_argstaddr((s/lib/python2.7/site-packages/distributed/core.pyR2s     cc`sg|jr|jdS|j}d}tjd|t|j||j|sg{Gz?( R_R6trangeRVR tsleepRR^tcanceltall(RCtiR((Rs/lib/python2.7/site-packages/distributed/core.pyRs  N(RR t__doc__RRR3RtRnRsR6RgRtpropertyRRRR/RR RhRRRER(((s/lib/python2.7/site-packages/distributed/core.pyR-Ks,# \      {)cC`sdS(Ntpong((R((s/lib/python2.7/site-packages/distributed/core.pytpingpongsc k`sH|}||d<|jd}t}|dkr:|}n|dk rS||d>> remote = rpc(address) # doctest: +SKIP >>> response = yield remote.add(x=10, y=20) # doctest: +SKIP One rpc object can be reused for several interactions. Additionally, this object creates and destroys many comms as necessary and so is safe to use in multiple overlapping communications. When done, close comms explicitly. >>> remote.close_comms() # doctest: +SKIP cC`si|_t||_||_d|_||_||_|dk rN|n||_||_ t j |_ t jj|dS(NR(tcommstcoerce_to_addressRttimeoutRRWRR3RRR\R]t_createdRtactiveRx(RCtargRRWRRRR((s/lib/python2.7/site-packages/distributed/core.pyRnGs      cc`s|jdkrtdnt}t}xC|jjD]2\}}|jre|j|n|r=Pq=q=Wx|D]}|j|=qzW| s|jrt|j |j d|j d|j V}d|_ nt|j|(RRR(RC((s/lib/python2.7/site-packages/distributed/core.pyt__repr__s(N(RR RR\R]RRR3RRtRnR RhRRRRRRRR(((s/lib/python2.7/site-packages/distributed/core.pyR4s&   )      t PooledRPCCallcB`sYeZdZdddZedZdZdZdZ dZ dZ RS( sW The result of ConnectionPool()('host:port') See Also: ConnectionPool cC`s:||_||_||_|dk r-|n||_dS(N(RtpoolRR3R(RCRRRR((s/lib/python2.7/site-packages/distributed/core.pyRns   cC`s|jS(N(R(RC((s/lib/python2.7/site-packages/distributed/core.pyRsc`stjfd}|S(Nc;`sjdk r4|jddkr4j|d(R(RC((s/lib/python2.7/site-packages/distributed/core.pyRsN( RR RR3RnRRRRRRR(((s/lib/python2.7/site-packages/distributed/core.pyRs    tConnectionPoolcB`seZdZejZded d d d d dZe dZ e dZ dZ d d d dZ ejd dZdZd Zd Zd ZRS( s` A maximum sized pool of Comm objects. This provides a connect method that mirrors the normal distributed.connect method, but provides connection sharing and tracks connection limits. This object provides an ``rpc`` like interface:: >>> rpc = ConnectionPool(limit=512) >>> scheduler = rpc('127.0.0.1:8786') >>> workers = [rpc(address) for address ...] >>> info = yield scheduler.identity() It creates enough comms to satisfy concurrent connections to any particular address:: >>> a, b = yield [scheduler.who_has(), scheduler.has_what()] It reuses existing comms so that we don't have to continuously reconnect. It also maintains a comm limit to avoid "too many open file handle" issues. Whenever this maximum is reached we clear out all idling comms. If that doesn't do the trick then we wait until one of the occupied comms closes. Parameters ---------- limit: int The number of open comms to maintain at once deserialize: bool Whether or not to deserialize data by default or pass it through icC`s||_tt|_tt|_||_||_|dk rK|n||_||_ ||_ t |_ |rt j|nd|_t j|_|jj|dS(N(tlimitRRt availabletoccupiedRWRR3RRRR teventR\R5tserverR]Rt _instancesRx(RCRRWRRRRR((s/lib/python2.7/site-packages/distributed/core.pyRns      cC`sttt|jjS(N(tsumtmapRRRo(RC((s/lib/python2.7/site-packages/distributed/core.pyR7scC`s#|jttt|jjS(N(RRRRRRo(RC((s/lib/python2.7/site-packages/distributed/core.pyR;scC`sd|j|jfS(Ns$(RR(RC((s/lib/python2.7/site-packages/distributed/core.pyR?scC`s:td|d|d|}t||d|jd|jS(s Cached rpc objects RRRRR(RRRR(RCRRR((s/lib/python2.7/site-packages/distributed/core.pyt__call__Bscc`sN|j|}|j|}|rZ|j}|jsZ|j|tj|qZnx;|j|jkr|j j |j |j j Vq]Wy]t |d|p|jd|jd|jV}d|_tj||_|jj|Wntk r nX|j||j|jkr;|j j ntj|dS(sE Get a Comm to the given address. For internal use. RRWRRN(RRRRRxR RRRRRtcollecttwaitR RRWRR~R\R5t_poolRR(RCRRRRR((s/lib/python2.7/site-packages/distributed/core.pyR Is4            cC`suy|j|j|Wntk r+nFX|jr]|j|jkrq|jjqqn|j|j |dS(sV Reuse an open communication to the given address. For internal use. N( RtremoveRRRRRRRRx(RCRR((s/lib/python2.7/site-packages/distributed/core.pyRns  cC`stjd|j|jxB|jjD]1\}}x|D]}|jq<W|jq)W|j|jkr|j j ndS(sV Collect open but unused communications, to allow opening other ones. s.Collecting unused comms. open: %d, active: %dN( RvRwRRRRRRRRR(RCRRR((s/lib/python2.7/site-packages/distributed/core.pyR}s cC`stjd|||jkrO|jj|}x|D]}|jq8Wn||jkr|jj|}x|D]}|jqwWn|j|jkr|jj ndS(s6 Remove all Comms to a given address. sRemoving comms to %sN( RvRwRRRRRRRR(RCRRR((s/lib/python2.7/site-packages/distributed/core.pyRs  cC`sx2|jjD]!}x|D]}|jqWqWx2|jjD]!}x|D]}|jqRWqEWx'|jD]}tjj|jqtWdS(s4 Close all communications abruptly. N(RRoRRRR R`Ri(RCRR((s/lib/python2.7/site-packages/distributed/core.pyRs  N(RR RR\R]RRtR3RnRRRRRR RhR RRRR(((s/lib/python2.7/site-packages/distributed/core.pyRs&    $   cC`s.t|ttfr$t|}nt|S(N(R{R|RRR(to((s/lib/python2.7/site-packages/distributed/core.pyRsRcC`s t}t|d}y&tjj|}tjj|Wn#tk rctt|}nXtj|}ytjj|}Wn-tk rdj t j |}}nXt |dkrd}ntj|}i|d6|d6|d6t|d6S( s Produce message to send back given an exception has occurred This does the following: 1. Gets the traceback 2. Truncates the exception and the traceback 3. Serializes the exception and traceback or 4. If they can't be serialized send string versions 5. Format a message and return See Also -------- clean_exception: deserialize and unpack message into exception/traceback six.reraise: raise exception/traceback iR.i'RRt tracebackRN(RRRtpickletdumpstloadsRROt to_serializetjoinRt format_tbRR3(RRttbte2te3te4ttb2t tb_result((s/lib/python2.7/site-packages/distributed/core.pyRs      cK`st|tst|trWytjj|}Wqutk rSt|}quXnt|trut|}nt|trytjj|}Wqtt fk rd}qXnt|t rd}nt |||fS(s Reraise exception and traceback. Deserialize if necessary See Also -------- error_message: create and serialize errors into message N( R{tbytest bytearrayRRRRRORtAttributeErrorR3RRK(RRR'((s/lib/python2.7/site-packages/distributed/core.pyRs   (Lt __future__RRRt collectionsRRtconcurrent.futuresRt functoolsRtloggingRRRPR\RHRttoolzRttornadoR ttornado.ioloopR t tornado.locksR t compatibilityR RR RRRRRtmetricsRR.Rtsystem_monitorRtutilsRRRRRRRRtIOErrorRt getLoggerRRvR%R*tMAX_BUFFER_SIZERIRJRuRtobjectR-RRhRtR3RRRRRRRR(((s/lib/python2.7/site-packages/distributed/core.pytsR      .4    ( 2  %