ó ßüÚ\c@sddlmZmZddlZddlZddlZddlZddlmZddl m Z m Z ddl m Z ddlmZddlmZejeƒZd efd „ƒYZd efd „ƒYZd efd„ƒYZdefd„ƒYZdefd„ƒYZdS(iÿÿÿÿ(t defaultdicttdequeN(tgeni(tfinalizetget_thread_identity(tCommClosedError(tsync(t to_serializetPubSubSchedulerExtensioncBszeZdZd„Zdddd„Zddddd„Zdddd„Zddddd„Zddddd„Z RS(s@ Extend Dask's scheduler with routes to handle PubSub machinery cCs¢||_ttƒ|_ttƒ|_ttƒ|_|jjji|jd6ƒ|jj ji|j d6|j d6|j d6|j d6ƒ||jjd+s Rispublish-scheduler(tloggertdebugR taddRRtlen(Rtcommtnametworker((s1lib/python2.7/site-packages/distributed/pubsub.pyR's cCsä|rptjd||ƒ|j|j|ƒx°|j|D].}|jj|idd6|d6|d6ƒq;Wnp|ràtjd||ƒx=|j|D].}|jj|idd6|d6td6ƒq—W|j|j|ƒndS( NsAdd worker subscriber: %s %sspubsub-add-subscribertoptaddressR"sAdd client subscriber: %s %sspubsub-publish-schedulertpublish( RRRRR R t worker_sendtTrueR(RR!R"R#tclienttpub((s1lib/python2.7/site-packages/distributed/pubsub.pyR0s #  cCst||j|krptjd||ƒ|j|j|ƒ|j| rp|j| rp|j|=|j|=qpndS(NsRemove publisher: %s %s(R RRtremoveR(RR!R"R#((s1lib/python2.7/site-packages/distributed/pubsub.pyRBs  cCsA|rptjd||ƒ|j|j|ƒxÊ|j|D].}|jj|idd6|d6|d6ƒq;WnŠ|rútjd||ƒ|j|j|ƒ|j|sú|j|=x@|j|D].}|jj|idd6|d6td6ƒqÂWqún|j| r=|j| r=tjd |ƒ|j|=|j|=ndS( NsAdd worker subscriber: %s %sspubsub-remove-subscriberR$R%R"sAdd client subscriber: %s %sspubsub-publish-schedulerR&sRemove PubSub topic %s( RRRR+R R R'RtFalse(RR!R"R#R)R*((s1lib/python2.7/site-packages/distributed/pubsub.pyRKs. #    cCsËx{t|j|ƒD]f}y0|jj|jidd6|d6|d6ƒWqttfk ry|jd|d|ƒqXqW|rÇx@|j|D].}|jj |idd6|d6|d6ƒq’WndS(Ns pubsub-msgR$R"tmsgR)( tlistRR t client_commstsendtKeyErrorRRRR'(RR"R-R#R)tctsub((s1lib/python2.7/site-packages/distributed/pubsub.pyRhs  N( t__name__t __module__t__doc__RtNoneRRRRR(((s1lib/python2.7/site-packages/distributed/pubsub.pyRs   tPubSubWorkerExtensioncBseeZdZd„Zddd„Zddd„Zddd„Zddd„Zd„Z d„Z RS( s= Extend Dask's Worker with routes to handle PubSub machinery cCsŽ||_|jjji|jd6|jd6|jd6|jd6ƒttj ƒ|_ ttj ƒ|_ td„ƒ|_ ||jj dˆtR (R#RRRRRtpublish_schedulerRtweakreftWeakSetRR tpublish_to_schedulerR(RR#((s1lib/python2.7/site-packages/distributed/pubsub.pyR{s     cKs)x"|j|D]}||j|(RR"R&((s1lib/python2.7/site-packages/distributed/pubsub.pyR;”scCs1x*|jj|gƒD]}|j|ƒqWdS(N(Rtgett_put(RR"R-R3((s1lib/python2.7/site-packages/distributed/pubsub.pyR—scCs|jjj|jƒdS(N(R#tloopt add_callbacktcleanup(R((s1lib/python2.7/site-packages/distributed/pubsub.pyttrigger_cleanup›scCsÚxct|jƒjƒD]L\}}t|ƒsidd6|d6}|jjj|ƒ|j|=qqWxmt|jƒjƒD]V\}}t|ƒs|idd6|d6}|jjj|ƒ|j|=|j|=q|q|WdS(Nspubsub-remove-subscriberR$R"spubsub-remove-publisher( tdictRtitemsR R#tbatched_streamR0R R>(RR"tsR-tp((s1lib/python2.7/site-packages/distributed/pubsub.pyRDžs" "  N( R4R5R6RR7RRR;RRERD(((s1lib/python2.7/site-packages/distributed/pubsub.pyR8xs  tPubSubClientExtensioncBs8eZdZd„Zddd„Zd„Zd„ZRS(s? Extend Dask's Client with handlers to handle PubSub machinery cCsL||_|jjji|jd6ƒttjƒ|_||jjd>> pub = Pub('my-topic') >>> pub.put(123) And Sub objects, from which we collect data: >>> sub = Sub('my-topic') >>> sub.get() 123 Many Pub and Sub objects can exist for the same topic. All data sent from any Pub will be sent to all Sub objects on that topic that are currently connected. Pub's and Sub's find each other using the scheduler, but they communicate directly with each other without coordination from the scheduler. Pubs and Subs use the central scheduler to find each other, but not to mediate the communication. This means that there is very little additional latency or overhead, and they are appropriate for very frequent data transfers. For context, most data transfer first checks with the scheduler to find which workers should participate, and then does direct worker-to-worker transfers. This checking in with the scheduler provides some stability guarantees, but also adds in a few extra network hops. PubSub doesn't do this, and so is faster, but also can easily drop messages if Pubs or Subs disappear without notice. When using a Pub or Sub from a Client all communications will be routed through the scheduler. This can cause some performance degredation. Pubs an Subs only operate at top-speed when they are both on workers. Parameters ---------- name: object (msgpack serializable) The name of the group of Pubs and Subs on which to participate Examples -------- >>> pub = Pub('my-topic') >>> sub = Sub('my-topic') >>> pub.put([1, 2, 3]) >>> sub.get() [1, 2, 3] You can also use sub within a for loop: >>> for msg in sub: # doctest: +SKIP ... print(msg) or an async for loop >>> async for msg in sub: # doctest: +SKIP ... print(msg) Similarly the ``.get`` method will return an awaitable if used by an async client or within the IOLoop thread of a worker >>> await sub.get() # doctest: +SKIP You can see the set of connected worker subscribers by looking at the ``.subscribers`` attribute: >>> pub.subscribers {'tcp://...': {}, 'tcp://...': {}} See Also -------- Sub cCsZ|dkr[|dkr[ddlm}m}y |ƒ}Wq[tk rW|ƒ}q[Xntƒ|_||_||_|s‹|s‹t ‚|jrµ|jj |_ |jj |_ n*|jrß|jj |_ |jj |_ n||_ t |_g|_|j j|jƒ|jrV|jjd}|j j|j|j|ƒt||jƒndS(Niÿÿÿÿ(t get_workert get_clientR (R7t distributedRORPt ExceptionRFRR#R)tAssertionErrorR RBR"R,t_startedt_bufferRCt_startRR RRRE(RR"R#R)RORPR ((s1lib/python2.7/site-packages/distributed/pubsub.pyRs0           ccsž|jri|jjd|jd|jjƒV}|jjd}|jj|dƒ|d|j|jR(RTRUtput(RtresultR R-((s1lib/python2.7/site-packages/distributed/pubsub.pyRV7s   cCsÇ|js|jj|ƒdSidd6|jd6t|ƒd6}|jr¤x$|jD]}|jj||ƒqTW|jjdj |jrÃ|jj j |ƒqÃn|j rÃ|j j j |ƒndS(Ns pubsub-msgR$R"R-R (RTRUtappendR"RR#Rtsend_to_workerRR>RHR0R)RM(RR-tdataR3((s1lib/python2.7/site-packages/distributed/pubsub.pyRAGs $  cCs|jj|j|ƒdS(s4 Publish a message to all subscribers of this topic N(RBRCRA(RR-((s1lib/python2.7/site-packages/distributed/pubsub.pyRWWsN( R4R5R6R7RRt coroutineRVRARW(((s1lib/python2.7/site-packages/distributed/pubsub.pyRNÊs L tSubcBsieZdZddd„Zejdd„ƒZeZdd„Z e Z Z d„Z d„Z d„ZRS(sc Subscribe to a Publish/Subscribe topic See Also -------- Pub: for full docstring cCs£|dkr[|dkr[ddlm}m}y |ƒ}Wq[tk rW|ƒ}q[Xn||_||_|jrˆ|jj|_n|jr£|jj|_n||_t ƒ|_ t j j ƒ|_|jræ|jjd}n|jr|jjd}n|jj|j|j|ƒidd6|jd6}|jr^|jj|jjj|ƒn1|jr†|jj|jjj|ƒn tƒ‚t||jƒdS(Niÿÿÿÿ(RORPR spubsub-add-subscriberR$R"(R7tdistributed.workerRORPRRR#R)RBR"Rtbufferttornadotlockst Conditiont conditionRRCRRRHR0RMRRE(RR"R#R)RORPR R-((s1lib/python2.7/site-packages/distributed/pubsub.pyRds6             ccs½|dk r!tjd|ƒ}ntjjƒ}xn|js |dk rƒ|tjjƒ|}|jƒdkr‰tjƒ‚q‰nd}|jj d|ƒVq3Wtj |jj ƒƒ‚dS(Ntsecondsittimeout( R7tdatetimet timedeltatnowR_t total_secondsRt TimeoutErrorRctwaittReturntpopleft(RRetstartttimeout2((s1lib/python2.7/site-packages/distributed/pubsub.pyt_get‡s   cCst|jr"|jj|jd|ƒS|jjtƒkrA|jƒS|jrW|jjƒSt|j|jd|ƒSdS(s Get a single message ReN( R)RRpR#t thread_idRR_RmRB(RRe((s1lib/python2.7/site-packages/distributed/pubsub.pyR@™s    cCs|S(N((R((s1lib/python2.7/site-packages/distributed/pubsub.pyt__iter__¦scCs|S(N((R((s1lib/python2.7/site-packages/distributed/pubsub.pyt __aiter__©scCs!|jj|ƒ|jjƒdS(N(R_RYRctnotify(RR-((s1lib/python2.7/site-packages/distributed/pubsub.pyRA¬sN(R4R5R6R7RRR\Rpt __anext__R@tnextt__next__RrRsRA(((s1lib/python2.7/site-packages/distributed/pubsub.pyR]\s#   (t collectionsRRRftloggingR<t tornado.locksR`Rt compatibilityRRtcoreRtutilsRtprotocol.serializeRt getLoggerR4RtobjectRR8RKRNR](((s1lib/python2.7/site-packages/distributed/pubsub.pyts    g5’