B F.\@sddlmZmZmZddlmZddlZddlZddlm Z ddl Zyddl m Z Wn e k rtddlm Z YnXddlmZmZmZddlmZdd lmZmZdd lmZeeZGd d d eZGd ddeZdS))print_functiondivisionabsolute_import) defaultdictN)gen)merge)Future_get_global_clientClient)time)tokey log_errors) get_clientc@sZeZdZdZddZdddZejddZdd d Z ejdd d Z ejdd dZ dS)VariableExtensionz An extension for the scheduler to manage queues This adds the following routes to the scheduler * variable-set * variable-get * variable-delete cCsv||_t|_tt|_ttjj|_ tj|_ |jj |j|j d|j|jjd<|j|jjd<||jjd<dS)N) variable_set variable_getzvariable-future-releasevariable_delete variables) schedulerdictrrsetwaitingtornadoZlocksZ Conditionwaiting_conditionsstartedZhandlersupdategetfuture_releaseZstream_handlersdelete extensions)selfrr"3lib/python3.7/site-packages/distributed/variable.py__init__!s    zVariableExtension.__init__NcCs|dk r*d|d}|jj|gd|dn d|d}y|j|}Wntk rVYn*X|ddkr|d|kr||d|||jkr|j||j|<dS)Nr )typevaluez variable-%s)keysclientZmsgpackr%r&)rZclient_desires_keysrKeyErrorreleaser notify_all)r!streamnamekeydatar(recordoldr"r"r#r0s    zVariableExtension.setccsJx"|j||fr"|j|VqW|jj|gd|d|j||f=dS)Nz variable-%s)r'r()rrwaitrZclient_releases_keys)r!r.r-r"r"r#r*As   zVariableExtension.releasecCs4|j||f||j||fs0|j|dS)N)rremoverr+)r!r-r.tokenr(r"r"r#rJsz VariableExtension.future_releasec cst}xN||jkrT|dk r*|t|}nd}|rB|dkrBt|jj|dVqW|j|}|ddkr|d}tj} |j j |} | dk r| j nd} | | d} | dkr| j j| d <| j j| d <t|| }|j||f| t|dS) Nr)timeoutr%r r&Zlost)r4stateerred exception traceback)r rr TimeoutErrorrr2uuiduuid4hexrZtasksrr6Zexception_blamer8r9rraddReturn) r!r,r-r(r5startleftr0r.r4Ztsr6msgr"r"r#rOs*         zVariableExtension.getc cshtXy|j|}Wntk r*Yn X|ddkrJ||d|V|j|=|j|=WdQRXdS)Nr%r r&)rrr)r*r)r!r,r-r(r1r"r"r#rhs zVariableExtension.delete)NNNNN)NNNN)NNNN)NNN) __name__ __module__ __qualname____doc__r$rr coroutiner*rrrr"r"r"r#rs   rc@sbeZdZdZdddZejddZdd Zejdd d Z dd d Z ddZ ddZ ddZ 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. .. warning:: This object is experimental and has known issues in Python 2 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 NrcCs$|pt|_|pdtj|_dS)Nz variable-)r r(r;r<r=r-)r!r-r(maxsizer"r"r#r$s zVariable.__init__ccsBt|tr(|jjjt|j|jdVn|jjj||jdVdS)N)r.r-)r/r-) isinstancer r(rrr r.r-)r!r&r"r"r#_sets   z Variable._setcKs|jj|j|f|S)z Set the value of this variable Parameters ---------- value: Future or object Must be either a Future or a msgpack-encodable value )r(syncrK)r!r&kwargsr"r"r#rsz Variable.setccs|jjj||j|jjdV}|ddkrt|d|jd|dd}|ddkrd|j|d |d |jd |j|d|d d n|d}t |dS)N)r5r-r(r%r r&Tr6)Zinformr6r7r8r9zvariable-future-releaser4)opr-r.r4) r(rrr-idr Z_stateZ set_error_send_to_schedulerrr?)r!r5dr&r"r"r#_gets   z Variable._getcKs|jj|jfd|i|S)z Get the value of this variable r5)r(rLrR)r!r5rMr"r"r#rsz Variable.getcCs$|jjdkr |jd|jddS)zn Delete this variable Caution, this affects all clients currently pointing to this variable. Zrunningr)rNr-N)r(ZstatusrPr-)r!r"r"r#rs zVariable.deletecCs|j|jjjfS)N)r-r(raddress)r!r"r"r# __getstate__szVariable.__getstate__c Cs\|\}}yt|}|jj|ks"tWn$ttfk rHt|dd}YnX|j||ddS)NF)Zset_as_default)r-r()rrrSAssertionErrorAttributeErrorr r$)r!r6r-rSr(r"r"r# __setstate__szVariable.__setstate__)NNr)N)N)rCrDrErFr$rrGrKrrRrrrTrWr"r"r"r#rHvs      rH) Z __future__rrr collectionsrZloggingr;rrZ tornado.locksZcytoolzr ImportErrorZtoolzr(r r r Zmetricsr Zutilsr rZworkerrZ getLoggerrCZloggerobjectrrHr"r"r"r#s      _