U /e!@sddlmZddlZddlZddlZddlmZddlmZddl m Z ddl m Z m Z ddlmZmZddlmZdd lmZmZdd lmZmZeeZGd d d ZGd ddZdS)) annotationsN) defaultdict)suppress)merge)parse_timedelta stringify)ClientFuture)time) TimeoutError log_errors) get_client get_workerc@sLeZdZdZddZdddZddZdd d Zdd d Ze dd dZ dS)VariableExtensionzAn extension for the scheduler to manage Variables This adds the following routes to the scheduler * variable-set * variable-get * variable-delete cCsf||_t|_tt|_ttj|_t|_ |jj |j|j d|j |jjd<|j|jjd<dS)N) variable_set variable_getvariable-future-releasevariable_delete) schedulerdict variablesrsetwaitingasyncio Conditionwaiting_conditionsstartedhandlersupdategetfuture_releaseZstream_handlersdelete)selfrr#8/tmp/pip-unpacked-wheel-g426oqom/distributed/variable.py__init__s    zVariableExtension.__init__Nc s|dk r*d|d}|jj|gd|dn d|d}z|j|}Wntk rVYn0X|ddkr|d|krt||d|||jkr|j4IdH|jW5QIdHRX||j|<dS)Nr )typevalue variable-%skeysclientmsgpackr&r') rZclient_desires_keysrKeyErrorrZ ensure_futurereleaser notify_all)r"namekeydatar+recordoldr#r#r$r-s   zVariableExtension.setc sn|j||frH|j|4IdH|j|IdHW5QIdHRXq|jj|gd|d|j||f=dS)Nr(r))rrwaitrZclient_releases_keys)r"r1r0r#r#r$r.?s &zVariableExtension.releasec sX|j||f||j||fsT|j|4IdH|j|W5QIdHRXdSN)rremoverr/)r"r0r1tokenr+r#r#r$r Gsz VariableExtension.future_releasec s t}|jkrt|dk r(|t|}nd}|r>|dkr>tz&fdd}tj||dIdHW5jXqj|}|ddkr|d}tj } j j |} | dk r| j nd} | | d } | d kr| jj| d <| jj| d <t|| }j||f| |S) Nrcs$jIdHjIdHdSr6)racquirer5r#r"r#r$_Xsz VariableExtension.get.._)timeoutr&r r'Zlost)r8stateerred exception traceback)r rr rr.rwait_foruuiduuid4hexrZtasksrr=Zexception_blamer?r@rradd) r"r0r+r<startleftr;r3r1r8tsr=msgr#r:r$rMs0         zVariableExtension.getc sz|j|}Wntk r"Yn$X|ddkrF||d|IdHtt|j|=W5QRXtt|j|=W5QRX|jd|dS)Nr&r r'r()rr-r.rrrZ remove_client)r"r0r+r4r#r#r$r!ns   zVariableExtension.delete)NNNN)NNNN)NNN)NN) __name__ __module__ __qualname____doc__r%rr.r rr r!r#r#r#r$rs    !rc@sVeZdZdZdddZddZdd Zdd d Zdd d ZddZ ddZ ddZ dS)VariableaDistributed 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. .. warning:: This object is experimental and has known issues in Python 2 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 NrcCsJz|p t|_Wntk r0tj|_YnX|pBdtj|_dS)Nz variable-) rcurrentr+ ValueErrorrrBrCrDr0)r"r0r+maxsizer#r#r$r%s zVariable.__init__csJt|tr,|jjjt|j|jdIdHn|jjj||jdIdHdS)N)r1r0)r2r0) isinstancer r+rrrr1r0)r"r'r#r#r$_sets  z Variable._setcKs|jj|j|f|S)zSet the value of this variable Parameters ---------- value : Future or object Must be either a Future or a msgpack-encodable value )r+syncrS)r"r'kwargsr#r#r$rsz Variable.setcs|jjj||j|jjdIdH}|ddkrt|d|jd|dd}|ddkrh|j|d |d |jd |j|d|d d n|d}|S)N)r<r0r+r&r r'Tr=)Zinformr=r>r?r@rr8)opr0r1r8) r+rrr0idr _stateZ set_error_send_to_scheduler)r"r<dr'r#r#r$_gets$    z Variable._getcKs"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". r<)rr+rTr[)r"r<rUr#r#r$rs z Variable.getcCs$|jjdkr |jd|jddS)zmDelete this variable Caution, this affects all clients currently pointing to this variable. Zrunningr)rVr0N)r+statusrYr0r:r#r#r$r!s zVariable.deletecCs|j|jjjfSr6)r0r+raddressr:r#r#r$ __getstate__szVariable.__getstate__c Cs\|\}}zt|}|jj|ks"tWn$ttfk rHt|dd}YnX|j||ddS)NF)Zset_as_default)r0r+)r rr]AssertionErrorAttributeErrorrr%)r"r=r0r]r+r#r#r$ __setstate__szVariable.__setstate__)NNr)N)N) rJrKrLrMr%rSrr[rr!r^rar#r#r#r$rNs(   rN) __future__rrloggingrB collectionsr contextlibrZtlzrZ dask.utilsrrZdistributed.clientrr Zdistributed.metricsr Zdistributed.utilsr r Zdistributed.workerr r getLoggerrJloggerrrNr#r#r#r$s      j