ԋg=@ddlmZddlZddlZddlZddlZddlmZmZddl m Z m Z ddl m Z ddlmZddlmZddlmZmZmZej,eZGd d ZGd d ZGd dZe dGddZe dGddZy)) annotationsN) defaultdictdeque) _deprecatedparse_timedelta)CommClosedError)time) to_serialize) TimeoutErrorsyncwait_forc>eZdZdZdZd dZd dZd dZd dZd dZ y) PubSubSchedulerExtensionz>Extend Dask's scheduler with routes to handle PubSub machineryc||_tt|_tt|_tt|_|jj jd|ji|jjj|j|j|j|jdy)Npubsub_add_publisher)pubsub-add-subscriberpubsub-remove-publisherpubsub-remove-subscriber pubsub-msg) schedulerrset publishers subscribersclient_subscribershandlersupdate add_publisherstream_handlersadd_subscriberremove_publisherremove_subscriberhandle_message)selfrs 2lib/python3.12/site-packages/distributed/pubsub.py__init__z!PubSubSchedulerExtension.__init__s"%c*&s+"-c"2 &&(>@R@R'ST &&--)-)<)<+/+@+@,0,B,B"11   Nc tjd|||j|j||j|Dcic]}|ic}||j vxrt |j |dkDdScc}w)NzAdd publisher: %s %sr)rpublish-scheduler)loggerdebugraddrrlen)r#nameworkeraddrs r$rz&PubSubSchedulerExtension.add_publisher's +T6: !!&)151A1A$1GH1GD"H1GH!%)@)@!@"7D++D12Q6  Hs Bc|rjtjd|||j|j||j|D]"}|j j |d||d$y|rjtjd|||j|D]"}|j j |d|dd$|j|j|yy)NzAdd worker subscriber: %s %sropaddressr-zAdd client subscriber: %s %spubsub-publish-schedulerTr2r-publish)r)r*rr+rr worker_sendrr#r-r.clientpubs r$rz'PubSubSchedulerExtension.add_subscriber0s  LL7v F   T " & &v .t,**2vtT-  LL7v Ft,**5tPTU-  # #D ) - -f 5r&c||j|vrptjd|||j|j||j|s+|j|s|j|=|j|=yyyy)NzRemove publisher: %s %s)rr)r*remover)r#r-r.s r$r z)PubSubSchedulerExtension.remove_publisherBs| T__T* * LL2D& A OOD ! ( ( 0##D)$//$2G$$T*OOD)3H) +r&c|rjtjd|||j|j||j|D]"}|j j |d||d$n|rtjd|||j|j||j|sA|j|=|j|D]"}|j j |d|dd$|j|sA|j|s1tjd||j|=|j|=yyy) NzRemove worker subscriber: %s %srr1zRemove client subscriber: %s %sr4Fr5zRemove PubSub topic %s)r)r*rr<rrr7rr8s r$r!z*PubSubSchedulerExtension.remove_subscriberKs0  LL:D& I   T " ) )& 1t,**5&RVW-  LL:D& I  # #D ) 0 0 8**40++D1??40CNN.."<$(',1%dood.C LL14 8  &%/D%r&cVt|j|D]/} |jj|j d||d1|r5|j|D]"}|jj|d||d$yy#t t f$r|j||YwxYw)Nrr2r-msg)r-r9) listrr client_commssendKeyErrorrr!rr7)r#r-r@r.r9csubs r$r"z'PubSubSchedulerExtension.handle_messagehsd--d34A <++A.33'cB5 ''-** d3G. o. <&&D&; Acb|jjj|jyrX)r.loop add_callbackcleanupr#s r$trigger_cleanupz%PubSubWorkerExtension.trigger_cleanup %%dll3r&ct|jjD]H\}}t|rd|d}|jj j ||j|=Jt|jjD]U\}}t|rd|d}|jj j ||j|=|j|=Wy)Nrr2r-r) dictritemsr,r.batched_streamrCrrU)r#r-sr@ps r$rdzPubSubWorkerExtension.cleanupsD,,-335GD!q67F **//4$$T* 6 DOO,224GD!q66E **//4OOD)--d3 5r&rG) rHrIrJrKr%rr!rQr"rfrdrLr&r$rNrNxs(E0",)2 4 4r&rNc*eZdZdZdZddZdZdZy)PubSubClientExtensionz=Extend Dask's Client with handlers to handle PubSub machineryc||_|jjjd|jit t j |_y)Nr)r9_stream_handlersrr"rrRrSr)r#r9s r$r%zPubSubClientExtension.__init__s=  $$++\4;N;N,OP&w7r&NcK|j|D]}|j|d{|j|s)|jjj d|dyy7?w)Nzpubsub-remove-subscribersri)rr_r9scheduler_commrCr`s r$r"z$PubSubClientExtension.handle_messagesf##D)C((3-  *% KK & & + +2DA & s'A+A)AA+cb|jjj|jyrX)r9rbrcrdres r$rfz%PubSubClientExtension.trigger_cleanuprgr&c|jjD]2\}}|r d|d}|jjj |4y)Nrri)rrkr9rtrC)r#r-rmr@s r$rdzPubSubClientExtension.cleanupsD''--/GD!7F **//40r&rG)rHrIrJrKr%r"rfrdrLr&r$rprpsG8 45r&rpz(Client.log_event() or Worker.log_event()) use_insteadc4eZdZdZddZdZdZdZdZeZ y) Puba7 Publish data with Publish-Subscribe pattern This allows clients and workers to directly communicate data between each other with a typical Publish-Subscribe pattern. This involves two components, Pub objects, into which we put data: >>> 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 degradation. Pubs and 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. worker: Worker (optional) The worker to be used for publishing data. Defaults to the value of ```get_worker()```. If given, ``client`` must be ``None``. client: Client (optional) Client used for communication with the scheduler. Defaults to the value of ``get_client()``. If given, ``worker`` must be ``None``. 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 Nc||ddlm}m} |}t |_||_||_|s|sJ|j r7|j j|_|j j|_ nB|jr6|jj|_|jj|_ ||_ d|_ g|_ |jj|j|j rm|j jd}|jj|j |j"|t%j&||j(yy#t$r |}YowxYw)Nr get_client get_workerFrP) distributedr|r} Exceptionrjrr.r9rrbr-_started_bufferrc_startrVrr+rRfinalizerf)r#r-r.r9r|r}rPs r$r%z Pub.__init__s/ >fn : &# 6   ;;![[22DN ((DI [[![[22DN ((DI    t{{+ ;;[[++H5F II " "6#4#4T#:#>#> E   T6#9#9 : ) &# &sE11FFcK|jr|jj|j|jjd{}|jj d}|j j|d|d|j|j<d|_ |jD]}|j||jdd=y7w)N)r-r.rPrr(T) r.rrr-r3rVrrrUrrput)r#resultrPr@s r$rz Pub._start=s ;;>>>>YYt{{':':?F[[++H5F    # #F=$9 :5;OP ;;'' **35({{%%h/DDTYYO **//5P [[ KK & & + +D 1r&cP|jj|j|y)z2Publish a message to all subscribers of this topicN)rbrcr_r#r@s r$rzPub.put\s tyy#.r&c"d|jdS)Nzr-res r$__repr__z Pub.__repr__` {!$$r&rG) rHrIrJrKr%rr_rr__str__rLr&r$ryrys)Qf;> 2 /%Gr&ryzClient.subscribe_topic()c`eZdZdZd dZedZd dZeZd dZ e xZ Z dZ dZ d Zd ZeZy) SubzbSubscribe to a Publish/Subscribe topic See Also -------- Pub: for full docstring Nc||ddlm}m} |}||_||_|jr|jj |_n'|j r|j j |_||_t|_ |jr|jjd}n%|j r|j jd}|j jj|j|d|jd}|jr;|j j|jjj|nQ|j r;|j j|j j j|n tt#j$||j&y#t$r |}YwxYw)Nrr{rPrri)distributed.workerr|r}rr.r9rbr-rbufferrVrcrr+rlrCrtrRrrf)r#r-r.r9r|r}rPr@s r$r%z Sub.__init__os_ >fn A &#  ;; ((DI [[ ((DI g ;;[[++H5F [[[[++H5F v11$7;;TB,dii@ ;; II " "4;;#=#=#B#BC H [[ II " "4;;#=#=#B#BC H+ v5565 &# &sGG"!G"c |jS#t$r(tj|_|jcYSwxYwrX) _conditionAttributeErrorasyncio Conditionres r$ conditionz Sub.conditions: #?? " #%//1DO?? " #s .??cKt}jsi||t|z z }|dkr td}fd} t||d{jj jsijj S7E#jj wxYww)NrcKjjd{jjd{y7'7wrX)racquirewaitresr$_zSub._get.._s<nn,,...nn))+++/+s!A A !A A A  A )r rr r rreleasepopleft)r#timeoutstarttimeout2rs` r$_getzSub._gets++""dfun5a<&.( , )qsH---&&(++"{{""$$ .&&(s5?B>BBB&B>B>BB;;B>ct|}|jr'|jj|j|S|jj t jk(r|jS|jr|jjSt|j|j|S)a6Get a single message Parameters ---------- timeout : number or string or timedelta, optional Time in seconds to wait before timing out. Instead of number of seconds, it is also possible to specify a timedelta in string format, e.g. "200ms". )r) rr9r rr. thread_id threading get_identrrrb)r#rs r$r^zSub.gets"'* ;;;;##DIIw#? ? [[ " "i&9&9&; ;99; {{{{**,, 499g> >r&c|SrXrLres r$__iter__z Sub.__iter__ r&c|SrXrLres r$ __aiter__z Sub.__aiter__rr&cK|jj||j4d{|jjdddd{y7/7#1d{7swYyxYwwrX)rrrnotifyrs r$r_zSub._putsD 3>>> NN ! ! #">>>>>>sD,A7AA7A" A7A A7 A7"A4(A+ )A40A7c"d|jdS)Nzrs"*3,$7::   8 $bbJ2424j558 CDYYEYx 34mm5mr&