ԋg"ddlmZddlZddlZddlZddlmZddlmZddl m Z ddl m Z ddl mZddlmZdd lmZmZmZmZdd lmZej0eZGd d ZGd dZy)) annotationsN) defaultdict)suppress)merge)parse_timedelta)Future)time)Deadline TimeoutError log_errorswait_for) get_clientcHeZdZdZdZd dZdZ d dZd dZe d dZ y) VariableExtensionzAn extension for the scheduler to manage Variables This adds the following routes to the scheduler * variable-set * variable-get * variable-delete c||_t|_tt|_tt j|_t j|_ |jjj|j|jd|j|jjd<|j|jjd<y)N) variable_set variable_get variable-future-received-confirmvariable_delete) schedulerdict variablesrsetwaitingasyncio Conditionwaiting_conditionsstartedhandlersupdategetfuture_received_confirmstream_handlersdelete)selfrs 4lib/python3.12/site-packages/distributed/variable.py__init__zVariableExtension.__init__s""3' "-g.?.?"@((*  &&!XXtxx @  ( ( &&'IJ=AKK&&'89NcKtjt|}|d|d}||jjvrQt j dd{|jrtd|d||jjvrQ|jj|gd|znd|d} |j|}|d dk(r1|d |k7r)t j|j|d | ||jvr?|j4d{|jjdddd{||j|<y7#t$rYkwxYw7P7(#1d{7swY8xYww) Nr)typevalueg{Gz?zTask z unknown to scheduler. variable-%skeysclientmsgpackr*r+)r afterrrtasksrsleepexpiredr client_desires_keysr ensure_futurereleaseKeyErrorr notify_all) r%namekeydatar/timeoutdeadlinerecordolds r&rzVariableExtension.set/sU>>/'":; ?&5FT^^111mmD)))##&se3I'JKKT^^111 NN . .SE-RVBV . W'$7F H..&C6{h&3w<3+>%%dll3w<&FG t~~ %||| '')$|%t!*    $||||sAF E7F 'F 9E!AF E0 F #E4> F  E2 F ! E-*F ,E--F 2F 4F:E= ;FF cK|j||fr_|j|4d{|j|jd{dddd{|j||fr_|jj |gd|z|j||f=y7|7Y7K#1d{7swY[xYww)Nr,r-)rrwaitrclient_releases_keys)r%r;r:s r&r7zVariableExtension.releaseFsll39%..t44--d388:::54ll39% ++#}t?S+T LLd # 5:5444s\%B?B$B?!B* B& B* B?B(B?32B?&B*(B?*B<0B3 1B<8B?cfK|jj|g||j||fj||j||fsF|j|4d{|j|j dddd{yy737#1d{7swYyxYwwN)rr5rremoverr9)r%r:r;tokenr/s r&r"z)VariableExtension.future_received_confirmNs **C5&9 S$Y&&u-||CI&..t44''-88:544'44444sHA"B1$B%B1(B B1BB1B1B."B% #B.*B1cKt}|jvro||t|z z }nd}|r|dkr t fd}t||d{jj |jvroj|}|ddk(r|d}t jj} jjj|} | | jnd} | | d} | d k(r2| jj| d <| jj| d <t!|| }j"||fj%| |S7#jj wxYww) NrcKjjd{jjd{y7'7wrE)racquirerBr%sr&_z VariableExtension.get.._bs<,,..000,,++---1-s!A A !A A A  A )r=r*rr+lost)rGstateerred exception traceback)r rr r rr7uuiduuid4hexrr2r!rNexception_blamerPrQrradd) r%r:r/r=startleftrLr?r;rGtsrNmsgs ` r&r!zVariableExtension.getWsa$..("$&5.1q"n$ '.qsD111 $$&$..("% &>X %/CJJL$$E%%))#.B "BHHFE!E2C#%#5#5#?#?K #%#5#5#?#?K 63'F LLd # ' ' . !2 $$&s6?FE'E$E'"(F CF$E''FFcK |j|}|ddk(r|j|d|d{ tt5|j|=dddtt5|j|=ddd|j j d|zy7n#t$rYvwxYw#1swY^xYw#1swYExYww)Nr*rr+r,)rr7r8rrr remove_client)r%r:r/r@s r&r$zVariableExtension.deletexs 7..&C6{h&ll3w<666 h ''- h t$  $$]T%9: 7        saC B$ C B"C B3C -B?;(C $ B0-C /B00C 3B<8C ?CC )NNNNN)NNNN)NNNNN) __name__ __module__ __qualname____doc__r'rr7r"r!r r$r(r&rrs;H &.$7;;B ; ;r(rcXeZdZdZd dZedZdZdZd dZ ddZ dd Z d Z d Z y)Variablea-Distributed Global Variable This allows multiple clients to share futures and data between each other with a single mutable variable. All metadata is sequentialized through the scheduler. Race conditions can occur. Values must be either Futures or msgpack-encodable data (ints, lists, strings, etc..) All data will be kept and sent through the scheduler, so it is wise not to send too much. If you want to share a large amount of data then ``scatter`` it and share the future instead. Parameters ---------- name: string (optional) Name used by other clients and the scheduler to identify the variable. If not given, a random name will be generated. client: Client (optional) Client used for communication with the scheduler. If not given, the default global client will be used. Examples -------- >>> from dask.distributed import Client, Variable # doctest: +SKIP >>> client = Client() # doctest: +SKIP >>> x = Variable('x') # doctest: +SKIP >>> x.set(123) # docttest: +SKIP >>> x.get() # docttest: +SKIP 123 >>> future = client.submit(f, x) # doctest: +SKIP >>> x.set(future) # doctest: +SKIP See Also -------- Queue: shared multi-producer/multi-consumer queue between clients Ncf||_|xs!dtjjz|_y)Nz variable-)_clientrRrSrTr:)r%r:r/s r&r'zVariable.__init__s& :K$**,*:*:: r(c|js t|_|jS|jS#t$rY|jSwxYwrE)rfr ValueErrorrKs r&r/zVariable.clientsJ|| )| ||t|| || s5 A  A cJ|jstt|dy)Nz object not properly initialized. This can happen if the object is being deserialized outside of the context of a Client or Worker.)r/ RuntimeErrorr*rKs r&_verify_runningzVariable._verify_runnings.{{:,'' r(c2Kt|trE|jjj |j |j |d{y|jjj ||j |d{y7@7w)N)r;r:r=)r<r:r=) isinstancerr/rrr;r:)r%r+r=s r&_setz Variable._sets eV $++''44IIDIIw5  ++''44G5     s$ABB:B BBBc v|j|jj|j|fd|i|S)zSet the value of this variable Parameters ---------- value : Future or object Must be either a Future or a msgpack-encodable value r=)rkr/syncrn)r%r+r=kwargss r&rz Variable.sets7 t{{ 5L'LVLLr(cK|jjj||j|jjd{}|ddk(rzt |d|j|d}|ddk(r"|j j|d|d |jjd |j|d|d d |S|d}|S7w) N)r=r:r/r*rr+rN)rNrOrPrQrrG)opr:r;rG) r/rrr:idr_state set_error_send_to_scheduler)r%r=dr+s r&_getz Variable._gets++''44$))DKKNN5   V9 1W:t{{!G*EEzW$ &&q~q~F KK * *< IIW:wZ   gJE # sA C C BCc |jt|}|jj|jfd|i|S)a@Get the value of this variable 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=)rkrr/rpry)r%r=rqs r&r!z Variable.gets? !'*t{{ E7EfEEr(c|j|jjdk(r)|jjd|jdyy)zmDelete this variable Caution, this affects all clients currently pointing to this variable. runningr)rsr:N)rkr/statusrwr:rKs r&r$zVariable.deletesC  ;;   * KK * *2CTYY+W X +r(c(t|jffSrE)rdr:rKs r& __reduce__zVariable.__reduce__s$))%%r(r])z30 srE)r^r_r`rar'propertyr/rkrnrryr!r$rrbr(r&rdrdsG"H; M( FY&r(rd) __future__rrloggingrR collectionsr contextlibrtlzr dask.utilsrdistributed.clientrdistributed.metricsr distributed.utilsr r r r distributed.workerr getLoggerr^loggerrrdrbr(r&rsW" #&%$JJ)   8 $q;q;h{&{&r(