ó ßüÚ\c@`s‰ddlmZmZmZddlmZmZmZddlm Z ddl Z ddl Z ddl Z ddl mZddlmZddlmZdd lmZd d lmZd d lmZe jeƒZd efd„ƒYZdefd„ƒYZdeeƒfd„ƒYZdeeƒfd„ƒYZ deeƒfd„ƒYZ!ej"de$dd„ƒZ%e$dd„Z&dS(i(tprint_functiontdivisiontabsolute_import(tABCMetatabstractmethodtabstractproperty(t timedeltaN(twith_metaclass(tgeni(ttime(tparse_timedeltai(tregistry(t parse_addresstCommClosedErrorcB`seZRS((t__name__t __module__(((s4lib/python2.7/site-packages/distributed/comm/core.pyR stFatalCommClosedErrorcB`seZRS((RR(((s4lib/python2.7/site-packages/distributed/comm/core.pyRstCommcB`sªeZdZejƒZd„Zed d„ƒZ ed d„ƒZ ed„ƒZ ed„ƒZ ed„ƒZ ed„ƒZed„ƒZed „ƒZd „ZRS( sØ A message-oriented communication object, representing an established communication channel. There should be only one reader and one writer at a time: to manage current communications, even with a single peer, you must create distinct ``Comm`` objects. Messages are arbitrary Python objects. Concrete implementations of this class can implement different serialization mechanisms depending on the underlying transport's characteristics. cC`s|jj|ƒd|_dS(N(t _instancestaddtNonetname(tself((s4lib/python2.7/site-packages/distributed/comm/core.pyt__init__+scC`sdS(sU Read and return a message (a Python object). This method is a coroutine. Parameters ---------- deserializers : Optional[Dict[str, Tuple[Callable, Callable, bool]]] An optional dict appropriate for distributed.protocol.deserialize. See :ref:`serialization` for more. N((Rt deserializers((s4lib/python2.7/site-packages/distributed/comm/core.pytread1tcC`sdS(s( Write a message (a Python object). This method is a coroutine. Parameters ---------- msg : on_error : Optional[str] The behavior when serialization fails. See ``distributed.protocol.core.dumps`` for valid values. N((Rtmsgton_error((s4lib/python2.7/site-packages/distributed/comm/core.pytwrite?RcC`sdS(s¾ Close the communication cleanly. This will attempt to flush outgoing buffers before actually closing the underlying transport. This method is a coroutine. N((R((s4lib/python2.7/site-packages/distributed/comm/core.pytcloseNRcC`sdS(s„ Close the communication immediately and abruptly. Useful in destructors or generators' ``finally`` blocks. N((R((s4lib/python2.7/site-packages/distributed/comm/core.pytabortWRcC`sdS(s6 Return whether the stream is closed. N((R((s4lib/python2.7/site-packages/distributed/comm/core.pytclosed^RcC`sdS(sN The local address. For logging and debugging purposes only. N((R((s4lib/python2.7/site-packages/distributed/comm/core.pyt local_addressdRcC`sdS(sO The peer's address. For logging and debugging purposes only. N((R((s4lib/python2.7/site-packages/distributed/comm/core.pyt peer_addressjRcC`siS(sá Return backend-specific information about the communication, as a dict. Typically, this is information which is initialized when the communication is established and doesn't vary afterwards. ((R((s4lib/python2.7/site-packages/distributed/comm/core.pyt extra_infopscC`sJ|jj}|jƒr#d|fSd||jp5d|j|jfSdS(Ns s<%s %s local=%s remote=%s>R(t __class__RR RR!R"(Rtclsname((s4lib/python2.7/site-packages/distributed/comm/core.pyt__repr__ys    N(RRt__doc__tweakreftWeakSetRRRRRRRRR RR!R"tpropertyR#R&(((s4lib/python2.7/site-packages/distributed/comm/core.pyRs      tListenercB`sVeZed„ƒZed„ƒZed„ƒZed„ƒZd„Zd„Z RS(cC`sdS(s; Start listening for incoming connections. N((R((s4lib/python2.7/site-packages/distributed/comm/core.pytstart‡RcC`sdS(s† Stop listening. This does not shutdown already established communications, but prevents accepting new ones. N((R((s4lib/python2.7/site-packages/distributed/comm/core.pytstopRcC`sdS(s8 The listening address as a URI string. N((R((s4lib/python2.7/site-packages/distributed/comm/core.pytlisten_address”RcC`sdS(sÀ An address this listener can be contacted on. This can be different from `listen_address` if the latter is some wildcard address such as 'tcp://0.0.0.0:123'. N((R((s4lib/python2.7/site-packages/distributed/comm/core.pytcontact_addressšRcC`s|jƒ|S(N(R,(R((s4lib/python2.7/site-packages/distributed/comm/core.pyt __enter__¢s cG`s|jƒdS(N(R-(Rtexc((s4lib/python2.7/site-packages/distributed/comm/core.pyt__exit__¦s( RRRR,R-RR.R/R0R2(((s4lib/python2.7/site-packages/distributed/comm/core.pyR+†s  t ConnectorcB`seZeed„ƒZRS(cC`sdS(s  Connect to the given address and return a Comm object. This function is a coroutine. It may raise EnvironmentError if the other endpoint is unreachable or unavailable. It may raise ValueError if the address is malformed. N((Rtaddresst deserialize((s4lib/python2.7/site-packages/distributed/comm/core.pytconnect«R(RRRtTrueR6(((s4lib/python2.7/site-packages/distributed/comm/core.pyR3ªsc#`s€ˆd kr!tjjdƒ‰ntˆddƒ‰tˆƒ\}}tj|ƒ}|jƒ}t ƒ}|ˆ} d } ‡‡fd†} xßt rlyK|j |d||p¯i} t j td| t ƒƒ| dtƒV} Wn†tk rõ‚qŽtk rJ}t|ƒ} t ƒ| kr=t jdƒVtjdƒqi| | ƒqŽt jk rg| | ƒqŽXPqŽWt j| ƒ‚d S( sÄ Connect to the given address (a URI such as ``tcp://127.0.0.1:1234``) and yield a ``Comm`` object. If the connection attempt fails, it is retried until the *timeout* is expired. s!distributed.comm.timeouts.connecttdefaulttsecondsc`s/|p d}dˆˆ|f}t|ƒ‚dS(Nsconnect() didn't finish in times0Timed out trying to connect to %r after %s s: %s(tIOError(terrorR(taddrttimeout(s4lib/python2.7/site-packages/distributed/comm/core.pyt_raiseÈs   R5tquiet_exceptionsg{®Gáz„?ssleeping on connectN(RtdasktconfigtgetR R R t get_backendt get_connectorR R7R6Rt with_timeoutRtEnvironmentErrorRtstrtsleeptloggertdebugt TimeoutErrortReturn(R<R=R5tconnection_argstschemetloctbackendt connectorR,tdeadlineR;R>tfuturetcommte((R<R=s4lib/python2.7/site-packages/distributed/comm/core.pyR6µs<        cC`sŸyt|dtƒ\}}WnUtk rs|rN|jdƒrNd|}n d|}t|dtƒ\}}nXtj|ƒ}|j||||p›iS(sJ Create a listener object with the given parameters. When its ``start()`` method is called, the listener will listen on the given address (a URI such as ``tcp://0.0.0.0``) and call *handle_comm* with a ``Comm`` object for each incoming connection. *handle_comm* can be a regular function or a coroutine. tstrictt ssl_contextstls://stcp://(R R7t ValueErrorRBR RCt get_listener(R<t handle_commR5RMRNRORP((s4lib/python2.7/site-packages/distributed/comm/core.pytlistenís    ('t __future__RRRtabcRRRtdatetimeRtloggingR(R@tsixRttornadoRtmetricsR tutilsR RR t addressingR t getLoggerRRIR:R RRR+R3t coroutineRR7R6R[(((s4lib/python2.7/site-packages/distributed/comm/core.pyts(   i$ 7