0Fie ddlmZddlZddlZddlZddlmZddlmZddl m Z ddl m Z ddl mZddlmZdd lmZmZmZdd lmZejeZGd d ZGd dZdS)) annotationsN) defaultdict)suppress)merge)parse_timedelta)Future)time) TimeoutError log_errorswait_for) get_clientcNeZdZdZdZd dZdZd dZd dZe d dZ dS) 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|_ |jj |j|j d|j |jjd<|j|jjd<dS)N) variable_set variable_getvariable-future-releasevariable_delete) schedulerdict variablesrsetwaitingasyncio Conditionwaiting_conditionsstartedhandlersupdategetfuture_releasestream_handlersdelete)selfrs 4lib/python3.11/site-packages/distributed/variable.py__init__zVariableExtension.__init__s""3'' "-g.?"@"@(**  &&!Xtx @ @   EIDW&'@A<@K&'8999NcK|&d|d}|j|gd|znd|d} |j|}|ddkr:|d|kr.tj||d|n#t $rYnwxYw||jvrJ|j4d{V|jdddd{Vn#1d{VswxYwY||j|<dS)Nr)typevalue variable-%skeysclientmsgpackr)r*) rclient_desires_keysrr ensure_futurereleaseKeyErrorr notify_all)r$namekeydatar.recordolds r%rzVariableExtension.set-s ?&55F N . .SE-RVBV . W W W W'$77F H.&C6{h&&3w<3+>+>%dll3w<&F&FGGG     D  t~ % %| * * * * * * * * ''))) * * * * * * * * * * * * * * * * * * * * * * * * * * *%ts# B BB+C C!$C!cRK|j||frk|j|4d{V|j|d{Vdddd{Vn#1d{VswxYwY|j||fk|j|gd|z|j||f=dS)Nr+r,)rrwaitrclient_releases_keys)r$r6r5s r%r2zVariableExtension.release?sOl39% ;.t4 ; ; ; ; ; ; ; ;-d388::::::::: ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ;l39% ; ++#}t?S+TTT Ld # # #s&A A'*A'cK|j||f||j||fsY|j|4d{V|j|dddd{VdS#1d{VswxYwYdSdSN)rremoverr4)r$r5r6tokenr.s r%r!z VariableExtension.future_releaseGs( S$Y&&u---|CI& ;.t4 ; ; ; ; ; ; ; ;'-88::: ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ; ;s A:: BBcKt}|jvr||t|z z }nd}|r|dkrt fd}t||d{Vjn#jwxYw|jvj|}|ddkr|d}t jj} j j |} | | j nd} | | d} | d kr| j j| d <| j j| d <t!|| }j||f| |S) NrcKjd{Vjd{VdSr>)racquirer;r$sr%_z VariableExtension.get.._XsY,..000000000,++-----------r')timeoutr)rr*lost)r@stateerred exception traceback)r rr r rr2uuiduuid4hexrtasksr rHexception_blamerJrKrradd) r$r5r.rFstartleftrEr8r6r@tsrHmsgs ` r%r zVariableExtension.getMs$.(("$&&5.1 %q"nn$ '.....qqssD1111111111 $$&&&& $$&&&&$.(("% &>X % %/CJLL$E%))#..B "BHHFE!E22C#%#5#?K #%#5#?K 63''F Ld # ' ' . . . s $B B$cK |j|}|ddkr"||d|d{Vn#t$rYnwxYwtt5|j|=dddn #1swxYwYtt5|j|=dddn #1swxYwY|jd|zdS)Nr)rr*r+)rr2r3rrr remove_client)r$r5r.r9s r%r#zVariableExtension.deletensu 7.&C6{h&&ll3w<666666666     D  h   . .'- . . . . . . . . . . . . . . . h   % %t$ % % % % % % % % % % % % % % % $$]T%9:::::s5 A A  A $ A99A=A= B--B14B1)NNNN)NNNNN) __name__ __module__ __qualname____doc__r&rr2r!r r r#r'r%rrs H H H&&&&$$$$;;;; B ; ; ;Z ; ; ;r'rc^eZdZdZd dZedZdZdZdZ d dZ d d Z d Z d Z dS)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 NcX||_|pdtjjz|_dS)Nz variable-)_clientrLrMrNr5)r$r5r.s r%r&zVariable.__init__s' :K$*,,*:: r'ch|js% t|_n#t$rYnwxYw|jSr>)rar ValueErrorrDs r%r.zVariable.clientsF|  )||     |s  **cR|jstt|ddS)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)rDs r%_verify_runningzVariable._verify_runnings@{ ::'''   r'cKt|tr3|jj|j|jd{VdS|jj||jd{VdS)N)r6r5)r7r5) isinstancerr.rrr6r5)r$r*s r%_setz Variable._sets eV $ $ Q+'444SS S S S S S S S S S+'44%di4PP P P P P P P P P Pr'c \||jj|j|fi|S)zSet the value of this variable Parameters ---------- value : Future or object Must be either a Future or a msgpack-encodable value )rfr.syncri)r$r*kwargss r%rz Variable.sets7 t{ 5;;F;;;r'cK|jj||j|jjd{V}|ddkrt |d|jd|d}|ddkr'|j|d |d |jd |j|d|d d n|d}|S)N)rFr5r.r)rr*TrH)informrHrIrJrKrr@)opr5r6r@) r.rrr5idr_state set_error_send_to_scheduler)r$rFdr*s r%_getz Variable._gets+'44$)DKN5         V9 1W:t{4qzRRREzW$$ &&q~q~FFF K * *3 IW:wZ     gJE r'c ||t|}|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". rF)rfrr.rkru)r$rFrls r%r z Variable.getsG !'**t{ EE7EfEEEr'c||jjdkr$|jd|jddSdS)zmDelete this variable Caution, this affects all clients currently pointing to this variable. runningr)ror5N)rfr.statusrsr5rDs r%r#zVariable.deletesU  ;  * * K * *2CTY+W+W X X X X X + *r'c t|jffSr>)r_r5rDs r% __reduce__zVariable.__reduce__s$)%%r'rXr>)rYrZr[r\r&propertyr.rfrirrur r#r{r]r'r%r_r_s""H;;;;XQQQ < < <( F F F FYYY&&&&&r'r_) __future__rrloggingrL collectionsr contextlibrtlzr dask.utilsrdistributed.clientrdistributed.metricsr distributed.utilsr r r distributed.workerr getLoggerrYloggerrr_r]r'r%rsM"""""" ######&&&&&&%%%%%%$$$$$$@@@@@@@@@@))))))  8 $ $g;g;g;g;g;g;g;g;Tw&w&w&w&w&w&w&w&w&w&r'