ԋgdUddlmZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl Z ddl Z ddl Z ddlZddlZddlZddlZddlmZddlmZmZmZmZmZddlmZddlmZddlmZm Z m!Z!ddl"m#Z#dd l$m%Z%m&Z&dd l'm(Z(m)Z)dd l*m+Z+dd l,m-Z.dd l/m0Z0m1Z1m2Z2m3Z3m4Z4m5Z5m6Z6m7Z7ddl8m9Z:ddl;mZ>m?Z?m@Z@ddlAZAddlBmCZCddlDmEZEmFZFddlGmHZHddlImJZJddlKmLZLddlMmNZNddlOmPZPmQZQmRZRmSZSddlTmUZUmVZVmWZWmXZXmYZYmZZZm[Z[ddl\m]Z]ddl^m_Z_ddl`maZaddlbmcZcmdZd ddlemfZfddlhmiZiddljmkZkddllmmZmmnZnmoZompZpmqZqmrZrddlbZsdd lsmtZtmuZudd!lsmvZwdd"lxmyZydd#lzm{Z{dd$l|m}Z}dd%l^m~Z~mZmZmZmZmZmZdd&lmZmZmZmZmZmZmZdd'lmZdd(lmZmZmZdd)lmZdd*lmZmZdd+lmZdd,lmZdd-lmZdd.lmZdd/lmZdd0lmZdd1lbmZmZmZmZmZmZmZmZmZmZmZmZdd2lmZmZmZmZmZdd3lmZmZmZe0rdd4lmZejxeZej~Zd5ed6<dgZe#d7d8Zd9ed7<d:eiZd;ZGd<d=eZGd>d?eZGd@dAZGdBdCe5ZdudDZdvdEZdwdGZGdHdIeqZGdJdKZdLZGdMdNeЫZdOZdPZdQZGdRdSe6ZeePenfZdTedU<dVZGdWdXeJZGdYdFeZdxdZZGd[d\eZd]Zd^Zd_Zdefd`ZdefdaZdbZdcZGdddeZdfZdydgZdhZdiZdydjZdkZGdldmZGdndoZGdpdqZe drZdsZdtZejey#eg$rez0FuturesCancelledError.__init__..s C ,=rT)rreverse)sortedr)rrs rrzFuturesCancelledError.__init__s" =t rc ttd|j}|d|dkDrdndd}dj|dg|jDcgc] }t |c}zScc}w) Nc,t|jSrrrs rrz/FuturesCancelledError.__str__..s c%,,&7r Futuresz cancelled:rzReasons:)summaprrr)rcountrrs rrzFuturesCancelledError.__str__sqC79J9JKL7'#!;;Gyy Z D>U{{m<'89IMSr2"TF < 4sBN)rrrr)rrrrtuple __slots__rrrrrrrs# && Ko&I  rrc6eZdZUded<ded<ded<ded<y) SourceCodercodeint lineno_framelineno_relativefilenameNrrrrrrrrrs IMrrctj}|r|Stttd}|D]%}t|}|j dk7r|cSt|='y)NT)rclosed)r~getrlistr}status)cLks r_get_global_clientrs[AtO$d3A  A  88x H"  rcZ|)d|_|ttd<tdxxdz cc<yy)NTrr)_set_as_defaultr}_global_client_indexrs r_set_global_clientrs4} 34,Q/0Q1$rClientcpttD]} t||urt|=y#t$rY$wxYwr)rr}KeyError)rrs r_del_global_clientrs@ / " q!Q&#A&#   s ) 55ceZdZUdZeej Zded<dZ dZ e jZ ejj Zd"dZedZdZdZd Zed Zed Zd Zd#d Zd$dZdZd#dZdZd%dZ dZ!dZ"dZ#d#dZ$edZ%dZ&d&dZ'dZ(dZ)dZ*dZ+dZ,dZ-d Z.d!Z/y)'FutureaA remotely running computation A Future is a local proxy to a result running on a remote worker. A user manages future objects in the local Python process to determine what happens in the larger cluster. .. note:: Users should not instantiate futures manually. This can lead to state corruption and deadlocking clusters. Parameters ---------- key: str, or tuple Key of remote data to which this future refers client: Client Client that should own this future. Defaults to _get_global_client() inform: bool Do we inform the scheduler that we need an update on this future state: FutureState The state of the future Examples -------- Futures typically emerge from Client computations >>> my_future = client.submit(add, 1, 2) # doctest: +SKIP We can track the progress and results of a future >>> my_future # doctest: +SKIP We can get the result or the exception and traceback from the future >>> my_future.result() # doctest: +SKIP See Also -------- Client: Creates futures staticmethod[[], bool]_is_finalizingNc||_d|_||_|xs)tjt tj f|_||_d|_ |jy)NF) r_cleared_clientr_uidnext_counter_id _input_state_state _bind_late)rrclientstaters rrzFuture.__init__0sN  >6;;V__(=>!  rc:|j|jSr)rrrs rrz Future.client9s ||rc2||_|jyr)rr)rrs r bind_clientzFuture.bind_client>s  rcr|jr|js |jj|j|jj|_|j|jj vr)|jj |j|_n=t|jx|_|jj |j<|j8 |jj|j}||jyyyy#t$rYywxYw)Nr) rr_inc_refr generation _generationfutures FutureStater_state_handlersr)rhandlers rrzFuture._bind_lateBs << LL ! !$(( +#||66D xx4<<///"ll22488< ?J488?TT dll22488<  ,*"ll::4;L;LMG) -!,< s0#D** D65D6cb|jr |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.)rr RuntimeErrortypers r_verify_initializedzFuture._verify_initializedTs4{{$++:,'' #.rc|jS)z|Returns the executor, which is the client. Returns ------- Client The executor rrs rexecutorzFuture.executor\s{{rc.|jjS)z_Returns the status Returns ------- str The status rrrs rrz Future.statusgs{{!!!rc6|jjS)zReturns whether or not the computation completed. Returns ------- bool True if the computation is complete, otherwise False )rdoners rrz Future.doners{{!!rc|jt5|jj|j|cdddS#1swYyxYw)a=Wait until computation completes, gather result to local process. Parameters ---------- timeout : number, optional Time in seconds after which to raise a ``dask.distributed.TimeoutError`` Raises ------ dask.distributed.TimeoutError If *timeout* seconds are elapsed before returning, a ``dask.distributed.TimeoutError`` is raised. Returns ------- result The result of the computation. Or a coroutine if the client is asynchronous. )callback_timeoutN)rr6rrr_resultrtimeouts rrz Future.result|s=(   "  ;;##DLL7#K! s 'A  AcK|jjd{|jdk(rOt|jj|jj }|r|\}}}|j ||S|jr<|jsJ|jj}t|tsJ|r||S|jj|gd{}|dS77 w)Nrr) rwaitrrQ exception tracebackwith_traceback cancelled isinstancerhr_gather)rraiseitexctyptbrrs rr zFuture._resultskk   ;;' !!$++"7"79N9NOC" S"((,, ^^ ;; ; --Ii8 88  ;;..v66F!9 % !"7s"C<C8C C<.C:/ C<:C<cK|jjd{|jdk(r|jjSy7*wNr)rr rrrs r _exceptionzFuture._exception@kk   ;;' !;;(( ( !A A +A c t|j|jj|jfd|i|S)a<Return the exception of a failed task Parameters ---------- timeout : number, optional Time in seconds after which to raise a ``dask.distributed.TimeoutError`` **kwargs : dict Optional keyword arguments for the function Returns ------- Exception The exception that was raised If *timeout* seconds are elapsed before returning, a ``dask.distributed.TimeoutError`` is raised. See Also -------- Future.traceback r)rrrrrrr kwargss rrzFuture.exceptions5,   "t{{T'TVTTrc|jt}|j!|jt j k7r, t dd|_t j |_fd}|jjjt|t|jj|y#t$rt d|_YwxYw)aCall callback on future when future has finished The callback ``fn`` should take the future as its only argument. This will be called regardless of if the future completes successfully, errs, or is cancelled The callback is executed in a separate thread. Parameters ---------- fn : callable The method or function to be called NrzDask-Callback-Thread)thread_name_prefixcb |y#t$rtjd|wxYw)NzError in callback %s of %s:) BaseExceptionloggerr)futfns rexecute_callbackz2Future.add_done_callback..execute_callbacks4 3    !>CH s ".)rr _cb_executor_cb_executor_pidosgetpidr TypeErrorrloop add_callback done_callbackrsubmit)rr&clsr's ` radd_done_callbackzFuture.add_done_callbacks   "    #s';';ryy{'J 9#5*@$ $&99;C   %% 4)9)9)@)@BR!S  9#5a#8  9sCCCc d|j|jj|gf||d|S)zgCancel the request to run this future See Also -------- Client.cancel rr)rrcancel)rrrrs rr5z Future.cancels5   "!t{{!!4&KSKFKKrc ^|j|jj|gfi|S)zcRetry this future if it has failed See Also -------- Client.retry )rrretryrrs rr7z Future.retrys.   " t{{  $2622rc4|jjdk(S)zReturns True if the future has been cancelled Returns ------- bool True if the future was 'cancelled', otherwise False rrrs rrzFuture.cancelleds{{!![00rcK|jjd{|jdk(r|jjSy7*wr)rr rrrs r _tracebackzFuture._traceback rrc t|j|jj|jfd|i|S)aReturn the traceback of a failed task This returns a traceback object. You can inspect this object using the ``traceback`` module. Alternatively if you call ``future.result()`` this traceback will accompany the raised exception. Parameters ---------- timeout : number, optional Time in seconds after which to raise a ``dask.distributed.TimeoutError`` If *timeout* seconds are elapsed before returning, a ``dask.distributed.TimeoutError`` is raised. Examples -------- >>> import traceback # doctest: +SKIP >>> tb = future.traceback() # doctest: +SKIP >>> traceback.format_tb(tb) # doctest: +SKIP [...] Returns ------- traceback The traceback object. Or a coroutine if the client is asynchronous. See Also -------- Future.exception r)rrrrr;rs rrzFuture.tracebacks5>   "t{{T'TVTTrc.|jjS)zReturns the type)rrrs rrz Future.type4s{{rc<|j|jsq|jj|jk(rMd|_ |jj j |jj|jyyy#t$rYywxYw)z Notes ----- This method can be called from different threads (see e.g. Client.get() or Future.__del__()) TN) rrrrrr-r._dec_refrr,rs rreleasezFuture.release9s~   "}}!7!74;K;K!K DM    --dkk.B.BDHHM"L}  sAB BBc(t|jffSr)rrrs rrzFuture.__reduce__Hs{""rcZt|j|j|jfSr)rrrrrs r__dask_tokenize__zFuture.__dask_tokenize__Ks!T ##TXXtxx88rcz |jy#t$r|jsYyt$rYywxYwr)r@AttributeErrorrrrs r__del__zFuture.__del__Ns@  LLN &&()   s:::ct|Sr)reprrs rrzFuture.__str__Ys Dzrc|jr3d|jdt|jd|jdSd|jd|jdS)Nz )rrr7rrs r__repr__zFuture.__repr__\sX 99DKK=$))1D0EWTXXJVWX t{{m7488*A> >rctdjt|jt |j |j S)Nzfuture.html.j2)rrr)r8renderrrr7rrrs r _repr_html_zFuture._repr_html_ds>,-44DHH $))$;;5  rc>|jjSr)r __await__rs rrPzFuture.__await__ks{{}&&((rc,t|jSr)hashrrs r__hash__zFuture.__hash__nsDHH~rc ||uSrr)rothers r__eq__z Future.__eq__qs u}rNNNrTNN)rzstr | tuple[Any, ...])0rrr__doc__ staticmethodsys is_finalizingrrr(r) itertoolsrruuiduuid4hexrrpropertyrrrrrrrrr rrr2r5r7rr;rrr@rrCrFrrKrNrPrSrVrrrrrs(T.:#:K:K-LN*LLy H 4::<  D*$"""L0*U2" HL31 UD   #9 ? )rrc^eZdZdZdZddZdZddZddZdZ d Z d Z d Z d Z dd ZdZy)rzeA Future's internal state. This is shared between all Futures with the same key and client. )_eventrrrrrcXd|_||_d|_d|_d|_d|_y)Npending)rdrrrrrrrs rrzFutureState.__init__}s,   rcX|j}|tjx}|_|Sr)rdasyncioEvent)revents r _get_eventzFutureState._get_events)  =")--/ 1EDK rNcd|_t|j|||_|j j y)zCancels the operationrrN)rrrrrlset)rrrs rr5zFutureState.cancels0! -$((6sS rcbd|_|jj|||_yy)zSets the status to 'finished' and sets the event Parameters ---------- type : any The type finishedN)rrlrnr)rrs rfinishzFutureState.finishs0!    DI rcNd|_|jjy)z.Sets the status to 'lost' and clears the eventlostNrrlclearrs rlosezFutureState.loses  !rcNd|_|jjyz1Sets the status to 'pending' and clears the eventrfNrtrs rr7zFutureState.retrys  !rct||\}}}d|_||_||_|j j y)aSets the error data Sets the status to 'error'. Sets the exception, the traceback, and the event Parameters ---------- exception: Exception The exception traceback: Exception The traceback rN)rQrrrrlrn)rrr_s r set_errorzFutureState.set_errors?#2)Y"G9i "" rcV|jduxr|jjS)z>**+2dkk]!<swwu~DPWHXrc|\}}t|tstj|yd|vr t dd|vrt j |d}nd}tjt j |d|y)NmessagezS_handle_warn: client received a warn event missing the required "message" argument.category)r)rrwarningswarnr,picklera)rkrzrrs r _handle_warnr s FAs c4   c C &   ||C O4HH LLY ( rctjjd}|r t|}|d|iSy#t$r}t d|d|d}~wwxYw)Nz"distributed.client.security-loaderzFailed to import `zP` configured at `distributed.client.security-loader` - is this module installed?address)daskconfigrro Exception ImportError)rsecurity_loader_termsecurity_loaderrs r_maybe_call_security_loaderr$su;;??+OP )*>?O 7344  $%9$:;   s 9 AAAc,eZdZUded<ded<ded<y) VersionsDictzdict[str, dict[str, Any]] schedulerz$dict[str, dict[str, dict[str, Any]]]workersrNrrrrrr3s(( 11 %%rrr|_T_LowLevelGraphcz|D]6}t|tst|tr%t|tr6yy)NTF)rrrbytes)iterableitems r _is_nestedr<s2 tX &tS)tU+  rceZdZUded<ded<ded<ded<d ed < d dfd Zdd Zedd ZeddZddZ dZ ddZ ddZ ddZ ddZddZxZS) _MapLayerrfunc Iterable[Any] iterablesstr | Iterable[str] | Nonerboolpuredict[str, Any] | Nonerc "||_|Dcgc]}ttt|c}|_||_||_|jD cic]\}} |t| c} }|_t |)|ycc}wcc} }w)Nr) rrrrErrritemsrsuperr) rrrrrrrrrvrs rrz_MapLayer.__init__Ns# LUVI%K :;IV/2 5;\\^D^TQq+a.(^D  [1 WEs BB c^t|jdt|jdS)Nz )rrr3rrs rrKz_MapLayer.__repr__^s*t*%%&hx /B.C2FFrc||t|dr |jS|j}||_|jSN _cached_dict)hasattrr_construct_graph)rdsks r_dictz_MapLayer._dictas?  4 ($$ $'')C #D    rct|dr |jSt|jtr7t|jt s|j|_|jS|j rZt|j|j}t|jDcgc]}|jdzt||z }}nwt tj}|jrLttt!t"|jDcgc]}|jd|d|c}ng}||_|Scc}wcc}w)N _cached_keys-)rrrrrrrr,rrziprr_r`rangeminrr)rtokrruidis r_keysz_MapLayer._keysms) 4 ($$ $$((H-j36O3788!xx99"499dkk:C%($8$8D3#t)<<$8 djjl+C  >>&+3s3/G+H%I%I $xxj#as3%I  %)! s *#E'Ec,t|jSr)rnrrs rget_output_keysz_MapLayer.get_output_keys4::rc,t|jSr)rrrs rget_ordered_keysz_MapLayer.get_ordered_keysDJJrct|dSr)rrs ris_materializedz_MapLayer.is_materializedst^,,rc |j|Sr)rrgs r __getitem__z_MapLayer.__getitem__szz#rc,t|jSr)iterrrs r__iter__z_MapLayer.__iter__rrc,t|jSr)rrrs r__len__z_MapLayer.__len__rrc li}|jsPt|jt|jDcic]\}}|t ||j g|}}}|Si}i}|jj D]\}}t|dkDr/t||}|j||<|||j<n|||<|jt|jt|jDcic] \}}|t ||j g|i|"c}}|Scc}}wcc}}w)Ngj@) rrrrrCrrrer@refrupdate)rrrrkwargs2rrvvs rrz_MapLayer._construct_graphs2 "{{"%TZZdnn1E!F!FICT#tyy0400!F , !GC ))+1!9s?!!QB!#GAJ"$CK!"GAJ *-TZZdnn9M)N)NICT#tyyC4C7CC)N, -"s "D*9%D0 )NTN) rrrrrrrrrrr)rr)rz Iterable[Key])rzset[Key])rr)rr-rrA)rz Iterator[Key])rr)rrrrrrKrbrrrrrrrrr __classcell__)rs@rrrGs N ## J&& +/-1 22!2( 2  2 + 2 G ! !8 - rrcjeZdZUdZeej Zded<e jZ ded<e e dZded<d Zd ed <d d ed d d d d d d d ed dfdZeddZej*ddZeddZej*ddZedZeddZedZdZdZdZdZdZ dZ!dZ"efdZ#e$dZ%dd Z&d!Z' d dd"Z(ddd#Z)d$Z*d%Z+d&Z,d'Z-d(Z.d)Z/d*Z0d+Z1d,Z2e$d-Z3dd.Z4dd/Z5d0Z6dd1Z7dd2Z8d3Z9dd4Z:e;dd5Zd8Z?d9Z@d:ZAd d d d d;dZBd d d d d;d dfdZddZy )raConnect to and submit computation to a Dask cluster The Client connects users to a Dask cluster. It provides an asynchronous user interface around functions and futures. This class resembles executors in ``concurrent.futures`` but also allows ``Future`` objects within ``submit/map`` calls. When a Client is instantiated it takes over all ``dask.compute`` and ``dask.persist`` calls by default. It is also common to create a Client without specifying the scheduler address , like ``Client()``. In this case the Client creates a :class:`LocalCluster` in the background and connects to that. Any extra keywords are passed from Client to LocalCluster in this case. See the LocalCluster documentation for more information. Parameters ---------- address: string, or Cluster This can be the address of a ``Scheduler`` server like a string ``'127.0.0.1:8786'`` or a cluster object like ``LocalCluster()`` loop The event loop timeout: int (defaults to configuration ``distributed.comm.timeouts.connect``) Timeout duration for initial connection to the scheduler set_as_default: bool (True) Use this Client as the global dask scheduler scheduler_file: string (optional) Path to a file with scheduler information if available security: Security or bool, optional Optional security information. If creating a local cluster can also pass in ``True``, in which case temporary self-signed credentials will be created automatically. asynchronous: bool (False by default) Set to True if using this client within async/await functions or within Tornado gen.coroutines. Otherwise this should remain False for normal use. name: string (optional) Gives the client a name that will be included in logs generated on the scheduler for matters relating to this client heartbeat_interval: int (optional) Time in milliseconds between heartbeats to scheduler serializers Iterable of approaches to use when serializing the object. See :ref:`serialization` for more. deserializers Iterable of approaches to use when deserializing the object. See :ref:`serialization` for more. extensions : list The extensions direct_to_workers: bool (optional) Whether or not to connect directly to the workers, or to ask the scheduler to serve as intermediary. connection_limit : int The number of open comms to maintain at once in the connection pool **kwargs: If you do not pass a scheduler address, Client will create a ``LocalCluster`` object, passing any extra keyword arguments. Examples -------- Provide cluster's scheduler node address on initialization: >>> client = Client('127.0.0.1:8786') # doctest: +SKIP Use ``submit`` method to send individual computations to the cluster >>> a = client.submit(add, 1, 2) # doctest: +SKIP >>> b = client.submit(add, 10, 20) # doctest: +SKIP Continue using submit or map on results to build up larger computations >>> c = client.submit(add, a, b) # doctest: +SKIP Gather results with the ``gather`` method. >>> client.gather(c) # doctest: +SKIP 33 You can also call Client with no arguments in order to create your own local cluster. >>> client = Client() # makes your own local "cluster" # doctest: +SKIP Extra keywords will be passed directly to LocalCluster >>> client = Client(n_workers=2, threads_per_worker=4) # doctest: +SKIP See Also -------- distributed.scheduler.Scheduler: Internal scheduler distributed.LocalCluster: rrz!ClassVar[weakref.WeakSet[Client]] _instances)rrzpreloading.PreloadManagerpreloadsN IOLoop | None _Client__loopTFic  |turtjjd}t |d}| t d||_t|_tt|_ d|_ | tjjdd}t|j|rd|zdzndztt!j"t%j&z|_d|_d|_g|_i|_||_||_d|_d|_i|_t=j>|_ tC||_"| |_#| | } | |_$| |_%d|_&d|_'|8tjjd d}|rtPjSd |||rt d tU|tW|tXtZfr||_ntWt]|d dtr||_|j6j,}|t^j`t^jbfvrted |j6dtgth5|jj}ddd|Mt]|j6dd}n5|3tW|ts#tmdjot||tW|tr tq|}| ts}n_tW|tr tsd'i|}nC|dur$tsjt}||j4d<ntW|trs tmd||_;|dk(r!|jvjyd|_=n |jvjyd|_=||_>t|||_@d|_Ad|_Bd|_C| tjjd} t | d} t tjjdd}t|_Dt|j|dz|jd<t|j| dz|jd<||_H||_I|r%tjjd|_Ki|_L|j|j|j|j|j|j|j|jd |_U|j|j|jd!|_Vt|| | d|jz||"|_,| jDcic]\}}|||c}}|_tjjd#}tjjd$}tj||||_[|j|%tjj|dd&l`ma}||y#1swYxYwcc}}w)(Nz!distributed.comm.timeouts.connectrz2None is an invalid value for global client timeoutz client-namer) clock_seqr newly-createdzscheduler-addressz*Config value `scheduler-address` found: %szUnexpected keyword arguments: scheduler_addressz:Trying to connect to an already closed or closing Cluster rsecurityz@Scheduler address must be a string or a Cluster instance, got {}Tz"security must be a Security objectworkerrr- asynchronousFzdistributed.client.heartbeatmsrz*distributed.client.scheduler-info-intervalizscheduler-info heartbeatdask.distributedr)z key-in-memoryz lost-datazcancelled-keysz task-retriedz task-erredrestartrrk)memoryrserred)limit serializers deserializers deserializeconnection_argsr serverzdistributed.client.preloadzdistributed.client.preload-argvr )ReplayTaskClientr)br0rrrr5 ValueError_timeoutrrrrrefcount_handle_report_taskrrrr_uuid1r*r+idrr_pending_msg_buffer extensionsscheduler_file_startup_kwargsclusterr_scheduler_identity threadingRLock_refcount_lockrbdatasets _serializers_deserializersdirect_to_workers_previous_as_currentscheduler_commr$inforrrSrOgetattrrPrclosingrrrEr-r,formatrrd temporaryrget_connection_argsr _asynchronousri _loop_runner_connecting_to_scheduler _gather_keys_gather_future_periodic_callbacksrL_update_scheduler_info _heartbeat _start_argrrn _set_config_event_handlers_handle_key_in_memory_handle_lost_data_handle_cancelled_keys_handle_retried_key_handle_task_erred_handle_restart _handle_error _handle_event_stream_handlersrrNrrGprocess_preloadsrstartrradddistributed.recreate_tasksr)rrr-r set_as_defaultrrrnameheartbeat_intervalrrrrconnection_limitrrscheduler_info_interval extensionpreload preload_argvrs rrzClient.__init__ s$ j kkoo&IJG!'3/ ?QR R v #C( #' <;;??=$7D J  #'sTzCS 2$**ryy{34 5  % #% ,% #% (oo/  '  'M+!2$(!# ?kkoo&94@G H'R  6=fVn=MNO O g]3 4$DN )?{{'HI "33D'<P  7 #d#?M*)r s1 X;Y;YcRtjdtd|jSNz"The io_loop property is deprecatedr stacklevelrrDeprecationWarningr-rs rio_loopzClient.io_loops# 02DQR yyrcJtjdtd||_yr:r=rvalues rr?zClient.io_loops  02DQR  rc\|j}||jjx|_}|Sr)rrr-)rr-s rr-z Client.loops/{{ < "&!2!2!7!7 7DK$ rcJtjdtd||_y)Nz'setting the loop property is deprecatedrr;)rrr>rrAs rr-z Client.loops  57IVW  rc# Ktj|}tjjd5 dtj | dddy#tj |wxYw#1swYyxYww)zThread-local, Task-local context manager that causes the Client.current class method to return self. Any Future objects deserialized inside this context manager will be automatically attached to this Client. rrN)r~rnrrr)rrs r as_currentzClient.as_currentsc !!$' [[__'9_ : +%%c* ; : %%c* ; :s26BA7AA7 BA44A77B<Bc`tj}|r|S|r tStd)aWhen running within the context of `as_client`, return the context-local current client. Otherwise, return the latest initialised Client. If no Client instances exist, raise ValueError. If allow_global is set to False, raise ValueError if running outside of the `as_client` context manager. Parameters ---------- allow_global : bool If True returns the default client Returns ------- Client The current client Raises ------ ValueError If there is no client set, a ValueError is raised See also -------- default_client z3Not running inside the `as_current` context manager)r~rdefault_clientr)r1 allow_globalouts rcurrentzClient.currents26!!# J !# #NOOrc |jjS#t$rh|j\}}|Yy|jj d\}}|dd}|dk(rd}n|j dd}t ||cYSwxYw) amLink to the scheduler's dashboard. Returns ------- str Dashboard URL. Examples -------- Opening the dashboard in your default web browser: >>> import webbrowser >>> from distributed import Client >>> client = Client() >>> webbrowser.open(client.dashboard_link) N://services dashboardinproc localhost:r)rdashboard_linkrE_get_scheduler_inforsplitrm)rrrprotocolrestporthosts rrSzClient.dashboard_link%s& 5<<.. . 5"668OIt !*!2!2!8!8!?$ #K0D8#"zz#q)(t4 4 5sB AB B c$ddlm}|jrqt|jdr[t |jj |r;|jj j }|jj }n|jjrO|j rC|js7t|j|j j }|j }n|j}|j }|t|fS)Nr) Schedulerr)distributed.schedulerr[rrrridentityr is_startedrrrr-r r])rr[rrs rrTzClient._get_scheduler_infoIs3 LL k24<<119=<<))224D ..I    ( ( *t~~dFWFW 4>>#:#:;DI++DI----rcV|j}|jd}|r|jdi}t|}td|j D}d|j j |||fz}|j Dcgc]}|d }}t|r|dtt|zz }|dz }|S|j:dj|j j |jjSd |j j d Scc}w) Nrrc3&K|] }|d yw)nthreadsNr).0ws r z"Client.__repr__..esC2BQ1Z=2Bz<%s: %r processes=%d threads=%d memory_limitz , memory=rJz<{}: scheduler={!r}>rz: No scheduler connected>) r rrrvaluesrrallr2rrr) rraddrrnworkersratextrcrs rrKzClient.__repr__^s''xx " hhy"-G7|HC'..2BCCH4'' 8D 291AB1AAa'1AFB6{ l3v;&??? CKDK ^^ ')00''&&  t~~..//HI ICs D&c * ttd}|tdkrdnd}|j\}}t dj |j |||j|j|j|S#t$rd}YiwxYw)Nzdask-labextensionz6.0.0FTzclient.html.j2)rrrrrrS jupyterlab) parse_versionrrrTr8rMrrrrS)r dle_version JUPYTERLABrrs rrNzClient._repr_html_zs '0C(DEK"- g0F"FDJ224 4,-44wwLL....!5  $ J s&B BBc 6|jdk7ry|jj|jr t ||j r+t j|jdi||_ yt|j|jfi|y)z*Start scheduler running in separate threadrNr) rrr.rrrri ensure_future_start_startedrrr-r8s rr.z Client.startst ;;/ )  !    t $   #11+$++2G2GHDM DKK 26 2rctdrjjSfd}|jS)NrtcKSwrrrsrrzzClient.__await__.._s  s)rrtrP)rrzs` rrPzClient.__await__s5 4 $==**, , 3==? "rc|jdvr |jj|y|jdvr|j j |yy#ttf$r|jdk(rYywxYw)N)runningrrx) connectingr)rrsendrMrErappendrrs r_send_to_scheduler_safezClient._send_to_scheduler_safesz ;;0 0 ##((-[[; ;  $ $ + +C 0<$^4 ;;)+, sAA98A9c|jdvr'|jj|j|yt d|jd|)N)rxrryrz.Tried sending message after closing. Status: z Message: )rr-r.r}rr|s r_send_to_schedulerzClient._send_to_schedulersD ;;O O II " "4#?#? E!%c3 rcKd|_|jjd{|tur |j}t |d}|j }|j- |jd{|jj}n%|jtjj|jsGt!j"dd{tjj|jsGt%dD]<} t'|j5}t)j*|}dddd}n`n^|j Rd d lm}|d|j4|j6d |j8d{|_|jj}t!j:d |_|j>|j||_d|_ |jC| d{|jJjMD]}|jtNjPjSD]\} } |jU| | |jVjd{t!jX|j[|_.|S77V#t$rtjddYwwxYw7#1swYxYw#t,t.f$r"t!j"dd{7YwxYw77,#tDtFf$r|jId{7wxYw7ҭw)Nryrz9Tried to start cluster and received an error. Proceeding.T)exc_infog{Gz? rr) LocalClusterrrr)/rrSr.r0rr5r!rrr$rrrr*pathexistsrisleepropenjsonloadrrdistributed.deployrr-rr Semaphore_gather_semaphorerr_ensure_connectedOSErrorr_closerrgr_default_event_handlersrsubscribe_topicr create_task_handle_reportr) rr rrrzfcfgrpctopicrs rrsz Client._starts" hhnn j mmG!'3/// << # ll"" ll44G  ,ggnnT%8%89mmD)))ggnnT%8%892Y.d112a"iil3!)nG __ $ 7!-"YY!//"&&"DL ll44G!(!2!21!5 >> !!XXg.DN" (((9 9 9 **113B HHJ4%<<BBDNE7   0Emm!!####*#6#6t7J7J7L#M  y #  O! *32#H-.!-----.  :% ++-     $s%M-KL;L>BM-M+/M- K K1-M-0K11M-7L <L'L5+L.,L51M-4L55M-;L>>"M( M#!M((M-cK|jjjsJd|_d|_|jj D]}|j dd|jj|j}t|z}|dkDr+|jdk(r |j|d{yt j#d|j|jd{y7>#t$r.tjdd{7|tz }Yn't$r|jd{7YywxYw|dkDs|jdk(rʌ7{w) Nryzscheduler-connection-lostz_Client lost the connection to the scheduler. Please check your connection and re-run your work.r4rr皙?zCFailed to reconnect to scheduler after %.2f seconds, closing client)rcommrrrrgr5rurr[rrrirrrr$r)rstr deadlines r _reconnectzClient._reconnectsM""''..000" ",,%%'B II2I  ( --6G#kdkk\9 ,,W,=== LL*   ++-  !> ,mmC((("TV+ kkm## kdkk\9$ sxB.E=1DDD 4E=?E;E=D!E!(D+)E!;E==E!EE!E= E!!E=*E=;E=cK|jr|jjr|js |jyd|_ t |jj fd|i|j d{}d|_|#t|j|d{n|jd{|jd|jdtjdd{ d|_|#t|j!|d{}n|j!d{}t#|dk(sJ|d d d k(sJ|d j%d rt'|d d |d j%d r.t)j*tj,|d d t/d|j0}|j3|||_d|_|j4D]}|j7||j4dd=t8j;dy7777T#t$r|jdk(r Yd|_ywxYw#d|_wxYw7]7Gw)NTr zClient->Schedulerzregister-clientF)oprreplyrHrrrrz stream-startrwarning10ms)intervalr-rxz+Started scheduling coroutines. Synchronized)rrrrrRrrr2r<rwriterversion_module get_versionsrrreadrrrrrVersionMismatchWarningrIr-r.rrr$debug)rr rrbcomms rrzClient._ensure_connected sL   ''..0,,~~% (,% 2 &&07;?;O;OD,DI"t::9I7I 8I=&J#J$J<J=DJIII II1&I4'J/I11I44 I==JJcK|jdvs |jy t|jjd{|_y7#t $rt jdYywxYww)N)rxryz(Not able to query scheduler for identity)rrr]r]r rr$rrs rrzClient._update_scheduler_infoZsb ;;7 74>>;Q  E'44>>;R;R;T5T'UD $5T E LLC D Es:A6"AA AA6AA30A62A33A6cK|jjd{}t||_|rt t |z}nd}d}|||kr|r$t |kDrt d||||fztjdd{|jjd{}t||_|||kryy77F7&w)Nct|djDcgc]$}|dtjjk(r|&c}Scc}w)Nrr)rrgrPrxr2)rwss rrunning_workersz1Client._wait_for_workers..running_workerslsT#9o4466(|v~~':'::6 s)A z#Only %d/%d workers arrived after %sr) rr]r]r r[r5rlrir)r n_workersr rrrs r_wait_for_workerszClient._wait_for_workersbs^^,,..#0#6 v 88HH d#i/DFX-"9&t,iAB--$ $ $0022D'4T':D $d#i/!/, %2s:C'C!A;C'C#!C'>C%?C'C'#C'%C'ct|tr|dkrtd|d|jr2t |jdr|jj ||S|j |j||S)a#Blocking call to wait for n workers before continuing Parameters ---------- n_workers : int The number of workers timeout : number, optional Time in seconds after which to raise a ``dask.distributed.TimeoutError`` rz4`n_workers` must be a positive integer. Instead got rwait_for_workersr)rrrrrrrrr)rrr s rrzClient.wait_for_workerssw)S)Y]FykQRS  <%? @"X . +rc|jjs|j|jrtj ||_|Sr)rr^r.rr~rnrrs r __enter__zClient.__enter__s>  ++- JJL   (7(;(;D(AD % rcrK|d{|jrtj||_|S7,wr)rr~rnrrs r __aenter__zClient.__aenter__s3   (7(;(;D(AD %  s 75-7cBK|jr tj|j|j|dud{y#t$rE}|jdj dst jdtdYd}~fd}~wwxYw7Vw)Nr# was created in a different ContextzRIt is deprecated to enter and exit the Client context manager from different tasksrr;fast) rr~rrrendswithrrr>rrexc_type exc_valueres r __aexit__zClient.__aexit__s  $ $ %%d&?&?@kk$&    vvay))*OP 3&   s9 BA BBB B;BBBBc|jr tj|j|jy#t$rE}|jdj dst jdtdYd}~Zd}~wwxYw)NrrzTIt is deprecated to enter and exit the Client context manager from different threadsrr;) rr~rrrrrrr>closers r__exit__zClient.__exit__sw  $ $ %%d&?&?@  vvay))*OP 5&  s> B ;BB c>|j|jyyr)rrrs rrFzClient.__del__s ;; " JJL #rct|j5|j|xxdz cc<dddy#1swYyxYwNr)r rrgs rrzClient._inc_refs+  MM# ! # ! s.7c|j5|j|xxdzcc<|j|dk(r|j|=|j|dddy#1swYyxYw)Nrr)r r _release_keyrgs rr?zClient._dec_refsV  MM# ! # }}S!Q&MM#&!!#& ! s AAA'ctjd||jj|d}||j |j dk7r!|j d|g|jdyy)z#Release key from distributed memoryzRelease key %sNrzclient-releases-keysrrr)r$rrpopr5rrr)rrrs rrzClient._release_keysd %s+ \\  c4 ( > IIK ;;( "  # #-uP  #rcfK |jy |jjjd{}t|tt fs|f}d}|D]}tj#d|j$|d |vr'd |d vr t'di|\}}}|j)||j+d }|d k(s|d k(rd}n9 |j,|}|di|} t/j0| r | d{|ry7#t$r|j rYy|j dk(r|j rQ|j j tjtjfvr|jd{7Yytjdtjdd|_|jd{7YYywxYw7#t2$r } tj5| Yd} ~ d} ~ wwxYw#t6t8j6f$rYywxYww)zListen to schedulerTNrxz(Client report stream closed to schedulerzReconnecting...ryFzClient %s receives message %srrrrz stream-closedr)rrrrMrrrrPrrrr$rrrrrrrrQrrr,inspect isawaitablerrrhri) rmsgsbreakoutrrrrrrrrs rrzClient._handle_reports4 &&.!%!4!4!9!9!>!>!@@D("$u 6 7D CLL!@$''3O37c(m+C'6'='= S"!0044BW}o(=#',"&"7"7";!(3"..v6"(LL# (cA&**,{{i/<s/   %   OOIy 1 rcftjd|jjD]}|j dd|jj |xj dz c_|j5|jj dddy#1swYyxYw)Nz%Receive restart signal from schedulerzscheduler-restartz1Scheduler has restarted. Please re-run your work.r4r) r$rrrgr5rurr r)rrs rr)zClient._handle_restartCs ;<\\((*E LL*G  +  1  MM   !! s B''B0cXtjdtj|y)NzScheduler exception:)r$rr)rrs rr*zClient._handle_errorOs-.#rcKtj}|j}|duxr||u}|rNttjt 5t tj|dd{dddd|r@tt tj5t ||rdndd{dddyy7T#1swYSxYw7#1swYyxYww)Nrrr)ri current_taskrrrhrlr<shield)rrrhandle_report_task should_waits r_wait_for_handle_report_taskz#Client._wait_for_handle_report_taskSs++- !55 d * U/A/U  '00,?w~~.@A3GGG@  ,(>(>?11!DDD@?  H@?E@?sZA C! #C0C1C5-C!"C6C7C; C!CC C!CCC!cK|jdk(ryd|_|jjd{tt5|j j D]}|j dddt|i|_ |jr8ts.tt5|j5 dddddd|jtjjddk(rtjjd=|j r`|j j"rJ|j j"j%s&|j'ddi|j'ddi|j)|4d{|j r\|j j"rF|j j"j%s"|j j+d{t-|j.D]}|j1| |j2:tt5|j4j+d{ddd|j6j+d{d|_t|ur t9ddddd{tt5|j:j=d{dddd|_d|_y7#1swYxYw#1swYdxYw#1swYixYw77=7#1swYxYw77#1d{7swYxYw7r#1swYqxYww) zSend close signal and wait until scheduler completes If fast is True, the client will close forcefully, by cancelling tasks the background _handle_report_task. rNrrrz close-clientz close-streamrr)rrteardownrrErrgstoprr rrr"rrrrrrrrrrrrr!rrSrr close_rpc)rrrrs rrz Client._closefs ;;( "  mm$$&&& n %..557 8& 4 #%   (:(<.)%%&* 88t{{ud3 3 ""5)   ##(('',,335  # #T>$: ;  # #T>$: ;44$4??##'',,++00779))//111DLL)!!c!**&n-,,,,....((.." " ""DK!#t+"4()@?,n %..**, , ,& c ' % %&%*)@ 2/.- #@???. -& %s5NLN 0L N M MM MNM0$M' %M0,N3M55M>:NcH|tur|jdz}|jdvr|jr t Syd|_t t 5|jjD]}|j ddd|jr |j}|r t||}|St|j|jd||jdk(sJ|js|jjyy#1swYxYw)aClose this client Clients will also close automatically when your Python session ends If you started a client without arguments like ``Client()`` then this will also close the local cluster that was started at the same time. Parameters ---------- timeout : number Time in seconds after which to raise a ``dask.distributed.TimeoutError`` See Also -------- Client.restart r)rrNrT)rrr)r0rrrrjrrErrgrrr<rrr-rr)rr rcoros rrz Client.closes& j mma'G ;;5 5  $&  n %..557 8&   ;;=Dg.K TYY $I{{h&&&""$    " " $%& %s 0DD!cLKtjdd|_|jj D]}|j |j 4d{|jr#|jjd{n:tt5|jjd{ddddddd{|jd{y77j79#1swY8xYw7/#1d{7swY?xYw7.w)Nz#Shutting down scheduler from Clientr)r$rrrrgrrrrrrMr terminater)rrs r _shutdownzClient._shutdowns 9: **113B GGI44466||ll((***o...22444/ 76kkm7*5/. 7666 sA D$"C9#D$&*D C;D %C?C=C?D  D$D D$3D"4D$;D =C??D D  D$ DD DD$c8|j|jS)zShut down the connected scheduler and workers Note, this may disrupt other clients that may be using the same scheduler and workers. See Also -------- Client.close : close only this client )rrrrs rshutdownzClient.shutdownsyy((rc t|fi|S)a Return a concurrent.futures Executor for submitting tasks on this Client Parameters ---------- **kwargs Any submit()- or map()- compatible arguments, such as `workers` or `resources`. Returns ------- ClientExecutor An Executor object that's fully compatible with the concurrent.futures API. rJr8s r get_executorzClient.get_executors"d-f--rrz100 ms) rr resourcesretriespriority fifo_timeoutallow_other_workersactoractorsrc  t|s td| xs| } | r| } |dvr td|K| rt|dzt|| g| z}n,t|dzt t j z}|j5||jvrt||cdddS ddd|r | tdt|ttfr|g}|t||gd| Di| jDcic]\}}|t|c}}i}|j!||g|||di||||| t#d d ig  }t$j'd t||||S#1swYxYwcc}}w)aT Submit a function application to the scheduler Parameters ---------- func : callable Callable to be scheduled as ``func(*args **kwargs)``. If ``func`` returns a coroutine, it will be run on the main event loop of a worker. Otherwise ``func`` will be run in a worker's task executor pool (see ``Worker.executors`` for more information.) *args : tuple Optional positional arguments key : str Unique identifier for the task. Defaults to function-name and hash workers : string or iterable of strings A set of worker addresses or hostnames on which computations may be performed. Leave empty to default to all workers (common case) resources : dict (defaults to {}) Defines the ``resources`` each instance of this mapped task requires on the worker; e.g. ``{'GPU': 2}``. See :doc:`worker resources ` for details on defining resources. retries : int (default to 0) Number of allowed automatic retries if the task fails priority : Number Optional prioritization of task. Zero is default. Higher priorities take precedence fifo_timeout : str timedelta (default '100ms') Allowed amount of time between calls to consider the same priority allow_other_workers : bool (defaults to False) Used with ``workers``. Indicates whether or not the computations may be performed on workers that are not in the `workers` set(s). actor : bool (default False) Whether this task should exist on the worker as a stateful actor. See :doc:`actors` for additional details. actors : bool (default False) Alias for `actor` pure : bool (defaults to True) Whether or not the function is pure. Set ``pure=False`` for impure functions like ``np.random.random``. Note that if both ``actor`` and ``pure`` kwargs are set to True, then the value of ``pure`` will be reverted to False, since an actor is stateful. See :ref:`pure functions` for more details. **kwargs Examples -------- >>> c = client.submit(add, a, b) # doctest: +SKIP Notes ----- The current implementation of a task graph resolution searches for occurrences of ``key`` and replaces it with a corresponding ``Future`` result. That can lead to unwanted substitution of strings passed as arguments to a task if these strings match some ``key`` that already exists on a cluster. To avoid these situations it is required to use unique values if a ``key`` is set manually. See https://github.com/dask/dask/issues/9969 to track progress on resolving this issue. Returns ------- Future If running in asynchronous mode, returns the future. Otherwise returns the concrete value Raises ------ TypeError If 'func' is not callable, a TypeError is raised ValueError If 'allow_other_workers'is True and 'workers' is None, a ValueError is raised See Also -------- Client.map : Submit on many arguments at once z1First input to submit must be a callable functionTFNz*allow_other_workers= must be True or FalseNr/Only use allow_other_workers= if using workers=c32K|]}t|ywr)rErbas rrdz Client.submit..{s/$Q+a.$rrr collections) rrinternal_priority user_priorityrrrr span_metadatazSubmit %s(...), %s)callabler,r3r,rr_r`r rrrrrrCrrE_graph_to_futuresrfr$r)rrrrrrrrrrrrrrrrrrs rr0z Client.submitsv~OP P 9D &9 9HI I ;tns*XdF-JT-JJtns*S->>  dll"c4(! "! 7?NO O gV} -iG 0$/28@A1k!n$@  ((  E 3"Ah"%&VX4F3GH)    )8D>3?s|C! As E4F4E=) rrrrrrrrrr batch_sizec l ts tdtd| Dstd| Dr tdtd| D} r dkDr| kDrt t fd| D}t |tr#t |Dcgc] }t |}}n!tt|Dcgc]}|}}t  f dt ||DgS|xs t} xs r r  td t| f| d }|j}t ttfrg!t tt fs td t#t |tt|}j%||| t'd d ig }|j)rJdt*j-dt|Dcgc]}|| c}Scc}wcc}wcc}w)aMap a function on a sequence of arguments Arguments can be normal objects or Futures Parameters ---------- func : callable Callable to be scheduled for execution. If ``func`` returns a coroutine, it will be run on the main event loop of a worker. Otherwise ``func`` will be run in a worker's task executor pool (see ``Worker.executors`` for more information.) iterables : Iterables List-like objects to map over. They should have the same length. key : str, list Prefix for task names if string. Explicit names if list. workers : string or iterable of strings A set of worker hostnames on which computations may be performed. Leave empty to default to all workers (common case) retries : int (default to 0) Number of allowed automatic retries if a task fails resources : dict (defaults to {}) Defines the `resources` each instance of this mapped task requires on the worker; e.g. ``{'GPU': 2}``. See :doc:`worker resources ` for details on defining resources. priority : Number Optional prioritization of task. Zero is default. Higher priorities take precedence allow_other_workers : bool (defaults to False) Used with `workers`. Indicates whether or not the computations may be performed on workers that are not in the `workers` set(s). fifo_timeout : str timedelta (default '100ms') Allowed amount of time between calls to consider the same priority actor : bool (default False) Whether these tasks should exist on the worker as stateful actors. See :doc:`actors` for additional details. actors : bool (default False) Alias for `actor` pure : bool (defaults to True) Whether or not the function is pure. Set ``pure=False`` for impure functions like ``np.random.random``. Note that if both ``actor`` and ``pure`` kwargs are set to True, then the value of ``pure`` will be reverted to False, since an actor is stateful. See :ref:`pure functions` for more details. batch_size : int, optional (default: just one batch whose size is the entire iterable) Submit tasks to the scheduler in batches of (at most) ``batch_size``. The tradeoff in batch size is that large batches avoid more per-batch overhead, but batches that are too big can take a long time to submit and unreasonably delay the cluster from starting its processing. **kwargs : dict Extra keyword arguments to send to the function. Large values will be included explicitly in the task graph. Examples -------- >>> L = client.map(func, sequence) # doctest: +SKIP Notes ----- The current implementation of a task graph resolution searches for occurrences of ``key`` and replaces it with a corresponding ``Future`` result. That can lead to unwanted substitution of strings passed as arguments to a task if these strings match some ``key`` that already exists on a cluster. To avoid these situations it is required to use unique values if a ``key`` is set manually. See https://github.com/dask/dask/issues/9969 to track progress on resolving this issue. Returns ------- List, iterator, or Queue of futures, depending on the type of the inputs. See Also -------- Client.submit : Submit a single function z.First input to map must be a callable functionc3<K|]}t|tywr)rpyQueue)rbits rrdzClient.map..s;2z"g&c3<K|]}t|tywr)rr rbrs rrdzClient.map..sC -6Jq( #Yr kDask no longer supports mapping over Iterators or Queues.Consider using a normal for loop and Client.submitc32K|]}t|ywrr)rbxs rrdzClient.map..s59a3q69rrc36K|]}t|ywr)r#)rbrrs rrdzClient.map..sT)hmJ9)sc3f K|](\}} jg||   d *yw)) rrrrrrrrrrN)r)rbrbatchrrrrrrrrrrrrs rrdzClient.map.. sf '9 UDHH ' '!),?%1"+#%!!'9s.1r)rrz0Workers must be a list or set of workers or Nonerrr) rrrrrrrrrzGraph must be non-materializedz map(%s, ...))rr,rhrrrrr#rrr3rrrrrrnrrrfrr$r)rrrrrrrrrrrrrrr total_lengthbatcheselementrrzrrrrs`` `````````` ` rrz Client.maps[z~LM M ;; ;sC -6C @ E 5955 *q.\J-FT)TUG#t$5B:s5ST5S'W 5ST%*3w<%89%8%89 '*$&8!$' ,#Xd^ 9D 7?NO O       ##% gV} -iG  z'D#;'GNO O T5T+;!<=((   3/"%&VX4F3GH)  &&(J*JJ( ^Xd^4$()Dq D))CU9~*s)H' H, H1raisec Kt|d\}}|Dcgc]}|jus|}}|rVtdt|dt|djd|Dcic]}||jjc}|D cgc]} | j } } t } i} | j}|; t} | jjjjk(rd} fd} tjd tt5t j"j%| Dcgc]}|j&vs||c}t d{dddd }t)}t)}|D]} | j }|j&vs| j*|vs,|j-|d k(r5| j.}|j0}|j2}|j5|d k(r|j-|d| |<tdz| Dcgc]}||vs|| vs |} }|rP| j7| Dcic]!}||j8vs||j8|#c}| Dcgc] }|| vs| } }j:r3xj<t)| zc_j:d{}nZt)| _t?j@jC||} j<d_n| _| d{}|ddk(rd k(rtjDntj}|dt|d|d|dD]}jGd|d|dD] } j&|jI"nn| r)d k(r$tM|tNr|Dcgc] }|| vs| }}| j7|dtQ|tS| | }|Scc}wcc}wcc} w#t$rd}YFwxYwcc}w7#1swYxYwcc}wcc}wcc}w774#tJ$rYwxYwcc}ww)NT) byte_keysz?Cannot gather Futures created by another client. These are the z (out of z:) mismatched Futures and their client IDs (this client is z): FcK j|}|jd{|jdk7rdk(r tyy7$#t$r twxYww)z3Want to stop the All(...) early if we find an errorNrpr)rr rrr)rrrrs rr zClient._gather..wait[ sf \\!_ggiyyJ&6W+<i,=&  i s%A%AA%A %A%A""A%z)Waiting on futures to clear before gather)quiet_exceptions)rrrskipzBad value, `errors=%s`rrz(Couldn't gather %s keys, rescheduling %srz report-key)rrdata)*rxrrrrrrrrzrrrr$rrr distributedutilsAllrrnrr/rrrrrrrrrirr_gather_remoterrrrrrrur")rrrdirect local_workerunpacked future_setrmismatched_futuresrrbad_datarrcr rfailed exceptionsbad_keysrrrrresponselogrs` ` rrzClient._gatherB s~0DI*)3LAqxxt7KaL !!$%7!8 93w<.QKKO77)SV-?@-?Q ^-?@BD  *44v 46 >++F > "L;;&&$..*@*@@!F  LLD E'"!''++*.F$3#2ET#Y$F%,,# ,FJuH$jjdll*fmmv.ENN3'(#]]$&LL $&LL '66yAA' S)(, ()AF)JKK% $Kt!q'8Qd]AtDK 6:UdaTQ ))!,,dU$(94a1D=49""!!SY.!!%!4!44$'I! ..'' =$$,*.D'*0D'!'<!W,(.'(9fnnv||>()V$ $F+C++<,LM,#F+C S)//1, EH &(Z$-G#+A8aq/@8HA HV$%8U4%:; WM A4  ,G#"2LV9 5($ Bs'RP$P$3R$P) RP."R P3 AR&Q Q  Q "Q 2Q 3Q 7A RA:R= QQ QR#Q7Q R Q$Q$";RQ)AR8Q,9A0R*Q/!R( Q>2Q>6=R3 Q>RQRQ  QR,R/ Q;8R:Q;;Rc|K|j4d{t|j}d|_d|_|s|rt |j j |d{}t||jd{\}}}}d|d} |s|rrt |j j||zd{} | ddk(r>| dj|n)t |j j|d{} dddd{| S7777c77#1d{7swY SxYww)zPerform gather with workers or scheduler This method exists to limit and batch many concurrent gathers into a few. In controls access using a Tornado semaphore, and picks up keys from other requests made recently. Nr)rSOK)rrrr) rrrrrvrwho_hasrtrSgatherr) rr!r"rr/r missing_keys failed_keysrzr*s rr zClient._gather_remote s))))))*D $D "&D  /0F0FT RR;N<62lK7;D+I;%4--L;4N& H )T1 (//5"11F1FT!RR'*)*+* S6  S'*)))*sD<DD<A D&#D$D&D9D&<D =AD&D"D& D<D$D<D&D& D&"D&$D<&D9,D/ -D94D<c 2t|tr tdt|tr fd|DS t }t 5jj|||cdddS#t $rd}YBwxYw#1swYyxYw)aGather futures from distributed memory Accepts a future, nested container of futures, iterator, or queue. The return type will match the input type. Parameters ---------- futures : Collection of futures This can be a possibly nested collection of Future objects. Collections can be lists, sets, or dictionaries errors : string Either 'raise' or 'skip' if we should raise if a future has erred or skip its inclusion in the output collection direct : boolean Whether or not to connect directly to the workers, or to ask the scheduler to serve as intermediary. This can also be set when creating the Client. asynchronous: bool If True the client is in asynchronous mode Returns ------- results: a collection of the same type as the input, but now with gathered results rather than futures Examples -------- >>> from operator import add # doctest: +SKIP >>> c = Client('127.0.0.1:8787') # doctest: +SKIP >>> x = c.submit(add, 1, 2) # doctest: +SKIP >>> c.gather(x) # doctest: +SKIP 3 >>> c.gather([x, [x], x]) # support lists and dicts # doctest: +SKIP [3, [3], 3] See Also -------- Client.scatter : Send data out to cluster zvDask no longer supports gathering over Iterators and Queues. Consider using a normal for loop and Client.submit/gatherc3FK|]}j|yw))rr!N)r0)rbrr!rrs rrdz Client.gather.. s!R'QDKK&K@'s!N)rr!r"r) rrr,r rzrr6rrr)rrrr!rr"s` `` rr0z Client.gather sP gw 'L  gx (R'R R %dd{t;||zkDr tAd|j,jC| d{}|sa|s tEdt|jG}tI|||jJd{\}}}|j,j3|||j6d{n2|j,jM|||j6|| d{|Dcic]}|tO||c}| jQD]$\}}|jR|jU| &|r=|r;|durdn|}|jWtjY|| d{t[|ttttfr|fd | D| r,t]dk(sJtjYdScc} wcc} w#t0$rd}YwxYwcc}w7G7 777W7'cc}w7w)NrFTr)r)r/rqrrzNo valid workers foundr)rrr broadcastr )r)rnc3(K|] }| ywrr)rbrrJs rrdz"Client._scatter..p s3USVUsr)/r0rrrrrrrr rn frozensetrrrr,r_r`rarr$rrzrrr update_datarerr_r[rirrlncores_runningrrrwrSscatterrrrrq _replicaterg issubclassr)rrrr7r!r"r rR input_typenamesunpackrtypesrcrdata2rar.rzr/rqrrr8rJs @r_scatterzClient._scatter s- j mmG gV} -iG dDqN +:D$Z  dH %:D dS), -:D$tUC CDF6D dT5M *GKLt!a))C/(1+=tLLPQDqa))C/$**,2B2BBDQE4()D$%%%tT" >++F > "L;;&&$..*@*@@!F   $ $$ $ /..,,@DE|3344Efd+ww-  <.E"+%mmC000v/*+CDD%)^^%B%B7%B%SSH # $%=>>x}}/+=gudhh+W%W"7Fnn00#F4771nn,,#77'# -,004aq&D/!40 HC LL  $ $# $ .& i!T)yA//$szz|"4g/K K K j4Y"? @3U33C s8q= =szz|$Q'C SMQ  F 1 T &X1 LsC(R&++Q,R&>Q1A R&% Q6/AR& R!%R&R AR&R>R&RR&AR&R3R&R2R&5R6R&>RA4R&R$A/R&6 RR&R R&R&R&R&R&R&R&c |tur |j}t|tst|tr t d t }|j|j|||||||| S#t$rd}Y1wxYw)aScatter data into distributed memory This moves data from the local client process into the workers of the distributed scheduler. Note that it is often better to submit jobs to your workers to have them load the data rather than loading data locally and then scattering it out to them. Parameters ---------- data : list, dict, or object Data to scatter out to workers. Output type matches input type. workers : list of tuples (optional) Optionally constrain locations of data. Specify workers as hostname/port pairs, e.g. ``('127.0.0.1', 8787)``. broadcast : bool (defaults to False) Whether to send each data element to all workers. By default we round-robin based on number of cores. .. note:: Setting this flag to True is incompatible with the Active Memory Manager's :ref:`ReduceReplicas` policy. If you wish to use it, you must first disable the policy or disable the AMM entirely. direct : bool (defaults to automatically check) Whether or not to connect directly to the workers, or to ask the scheduler to serve as intermediary. This can also be set when creating the Client. hash : bool (optional) Whether or not to hash data to determine key. If False then this uses a random key timeout : number, optional Time in seconds after which to raise a ``dask.distributed.TimeoutError`` asynchronous: bool If True the client is in asynchronous mode Returns ------- List, dict, iterator, or queue of futures matching the type of input. Examples -------- >>> c = Client('127.0.0.1:8787') # doctest: +SKIP >>> c.scatter(1) # doctest: +SKIP >>> c.scatter([1, 2, 3]) # doctest: +SKIP [, , ] >>> c.scatter({'x': 1, 'y': 2, 'z': 3}) # doctest: +SKIP {'x': , 'y': , 'z': } Constrain location of data to subset of workers >>> c.scatter([1, 2, 3], workers=[('hostname', 8788)]) # doctest: +SKIP Broadcast data to all workers >>> [future] = c.scatter([element], broadcast=True) # doctest: +SKIP Send scattered data to parallelized function using client futures interface >>> data = c.scatter(data, broadcast=True) # doctest: +SKIP >>> res = [c.submit(func, data, i) for i in range(100)] Notes ----- Scattering a dictionary uses ``dict`` keys to create ``Future`` keys. The current implementation of a task graph resolution searches for occurrences of ``key`` and replaces it with a corresponding ``Future`` result. That can lead to unwanted substitution of strings passed as arguments to a task if these strings match some ``key`` that already exists on a cluster. To avoid these situations it is required to use unique values if a ``key`` is set manually. See https://github.com/dask/dask/issues/9969 to track progress on resolving this issue. See Also -------- Client.gather : Gather data back to local process r N)rr7r!r"r rrR) r0rrrr r,rzrrrrE) rrrr7r!rRr rr"s rr=zClient.scatterw s| j mmG dG $ 4(BE   %?B(BcB|j|j||||S)a Cancel running futures This stops future tasks from being scheduled if they have not yet run and deletes them if they have already run. After calling, this result and all dependent results will no longer be accessible Parameters ---------- futures : List[Future] The list of Futures asynchronous: bool If True the client is in asynchronous mode force : boolean (False) Cancel this future even if other clients desire it reason: str Reason for cancelling the futures msg : str Message that will be attached to the cancelled future )rrHr)rrrJ)rrrrHrrs rr5z Client.cancel s)(yy LL' Es  rcKtt|Dchc]}|jc}}|jj ||j d{}|D]!}|j |}|j #ycc}w70w)N)rr)rrIrrr7rr)rrrrr*rrs r_retryz Client._retry ssJw$78$7qQUU$789--4-HHCc"B HHJ9HsB B1B B 1B c>|j|j||S)z Retry failed futures Parameters ---------- futures : list of Futures The list of Futures asynchronous: bool If True the client is in asynchronous mode )r)rrrM)rrrs rr7z Client.retry syygLyIIr)r2overridectKgtjjjddfd}|r5t |dk(r t dt |dk(r|d}||||j D]\}}|||tjd{y7w)Npublish_flush_batched_send)rrctDcgc]}|jc}fd}j|ycc}w)NcKjjd{jjtjd{y7@7w)N)r)rr2rrOr)rpublish_wait_flush publish_putr_r)rrr2rOrrsrrzz4Client._publish_dataset..add_coro.._+ sgnn77C7@@@nn00%d+%77 1As! A(A$:A(A&A(&A()rIrr{) r2rrrzr coroutinesrOrrs `` @radd_coroz)Client._publish_dataset..add_coro( sC#-d#34#3aAEE#34D     ac "5sA rzQIf name is provided, expecting call signature like publish_dataset(df, name='ds')r) r_r`rarrrrrir0) rr2rOrrrWrrVrs ` ` @@r_publish_datasetzClient._publish_dataset" s jjl 'CC PQ # 4yA~ 6 TaAw T4  ,,.JD$ T4 )nnj)))sB*B80B61B8cB|j|jg|i|S)a Publish named datasets to scheduler This stores a named reference to a dask collection or list of futures on the scheduler. These references are available to other Clients which can download the collection or futures with ``get_dataset``. Datasets are not immediately computed. You may wish to call ``Client.persist`` prior to publishing a dataset. Parameters ---------- args : list of objects to publish as name kwargs : dict named collections to publish on the scheduler Examples -------- Publishing client: >>> df = dd.read_csv('s3://...') # doctest: +SKIP >>> df = c.persist(df) # doctest: +SKIP >>> c.publish_dataset(my_dataset=df) # doctest: +SKIP Alternative invocation >>> c.publish_dataset(df, name='my_dataset') Receiving client: >>> c.list_datasets() # doctest: +SKIP ['my_dataset'] >>> df2 = c.get_dataset('my_dataset') # doctest: +SKIP Returns ------- None See Also -------- Client.list_datasets Client.get_dataset Client.unpublish_dataset Client.persist )rrrX)rrrs rpublish_datasetzClient.publish_datasetG s&Ztyy..@@@@rc T|j|jjfd|i|S)a Remove named datasets from scheduler Parameters ---------- name : str The name of the dataset to unpublish Examples -------- >>> c.list_datasets() # doctest: +SKIP ['my_dataset'] >>> c.unpublish_dataset('my_dataset') # doctest: +SKIP >>> c.list_datasets() # doctest: +SKIP [] See Also -------- Client.publish_dataset r2)rrrpublish_delete)rr2rs runpublish_datasetzClient.unpublish_datasetv s)*tyy66LTLVLLrc P|j|jjfi|S)z List named datasets available on the scheduler See Also -------- Client.publish_dataset Client.get_dataset )rrr publish_listr8s r list_datasetszClient.list_datasets s$tyy44???rc^K|j5|jj||jd{}ddd|turt d|d|St |dD]}|j||j|dS7`#1swY_xYww)N)r2rz Dataset 'z ' not foundr) rFr publish_getrr0rrIr_inform_scheduler_of_futures)rr2rrJr%s r _get_datasetzClient._get_dataset s __ 22TWW2MMC ;*$4& <==c&k*C OOD !+ ))+6{N s-B-+B!BB!AB-B!!B*&B-c B|j|j|fd|i|S)a Get named dataset from the scheduler if present. Return the default or raise a KeyError if not present. Parameters ---------- name : str name of the dataset to retrieve default : str optional, not set by default If set, do not raise a KeyError if the name is not present but return this default kwargs : dict additional keyword arguments to _get_dataset Returns ------- The dataset from the scheduler, if present See Also -------- Client.publish_dataset Client.list_datasets r)rrrd)rr2rrs r get_datasetzClient.get_dataset s'2tyy**DL'LVLLr)r cK|jjt|t|t||d{}|ddk(r tdi|\}}}|j ||dS71w)N)functionrrr rrrr)r run_functionr`rQr) rrhr rrr*rrrs r_run_on_schedulerzClient._run_on_scheduler s448_t= 5   H  (*6X6LCb$$R( (H% % s>A4A22A4cD|j|j|g|i|S)aeRun a function on the scheduler process This is typically used for live debugging. The function should take a keyword argument ``dask_scheduler=``, which will be given the scheduler object itself. Parameters ---------- function : callable The function to run on the scheduler process *args : tuple Optional arguments for the function **kwargs : dict Optional keyword arguments for the function Examples -------- >>> def get_number_of_tasks(dask_scheduler=None): ... return len(dask_scheduler.tasks) >>> client.run_on_scheduler(get_number_of_tasks) # doctest: +SKIP 100 Run asynchronous functions in the background: >>> async def print_state(dask_scheduler): # doctest: +SKIP ... while True: ... print(dask_scheduler.status) ... await asyncio.sleep(1) >>> c.run(print_state, wait=False) # doctest: +SKIP See Also -------- Client.run : Run a function on all workers )rrrj)rrhrrs rrun_on_schedulerzClient.run_on_scheduler s(Jtyy//KDKFKKr)nannyrr on_errorc K|jjtdt|t||t|||dd{}i} |j D]\} } t | t rt| } t | ts>J| ddk(r!td i| \} } }| j|} n| ddk(sJ| d| | <o|d k(r| |d k(r| | | <|d k7std ||r| Sy7w)Nrun)rrhrr r return_pickle)rrrmrnrrr.rrrignorez5on_error must be 'raise', 'return', or 'ignore'; got r) rr7rr`rrrrarrQrr)rrhrmrr rnrr responsesresultsrresprrzrs r_runz Client._run s;..22x4[V} $3   "*IC$&Dk!#y111h7*,4t4 3((,H~---#H~ 7" X%" X% #,('+0 N K  sA DDBD,Drr rmrncN|j|j|g|||||d|S)a% Run a function on all workers outside of task scheduling system This calls a function on all currently known workers immediately, blocks until those results come back, and returns the results asynchronously as a dictionary keyed by worker address. This method is generally used for side effects such as collecting diagnostic information or installing libraries. If your function takes an input argument named ``dask_worker`` then that variable will be populated with the worker itself. Parameters ---------- function : callable The function to run *args : tuple Optional arguments for the remote function **kwargs : dict Optional keyword arguments for the remote function workers : list Workers on which to run the function. Defaults to all known workers. wait : boolean (optional) If the function is asynchronous whether or not to wait until that function finishes. nanny : bool, default False Whether to run ``function`` on the nanny. By default, the function is run on the worker process. If specified, the addresses in ``workers`` should still be the worker addresses, not the nanny addresses. on_error: "raise" | "return" | "ignore" If the function raises an error on a worker: raise (default) Re-raise the exception on the client. The output from other workers will be lost. return Return the Exception object instead of the function output for the worker ignore Ignore the exception and remove the worker from the result dict Examples -------- >>> c.run(os.getpid) # doctest: +SKIP {'192.168.0.100:9000': 1234, '192.168.0.101:9000': 4321, '192.168.0.102:9000': 5555} Restrict computation to particular workers with the ``workers=`` keyword argument. >>> c.run(os.getpid, workers=['192.168.0.100:9000', ... '192.168.0.101:9000']) # doctest: +SKIP {'192.168.0.100:9000': 1234, '192.168.0.101:9000': 4321} >>> def get_status(dask_worker): ... return dask_worker.status >>> c.run(get_status) # doctest: +SKIP {'192.168.0.100:9000': 'running', '192.168.0.101:9000': 'running} Run asynchronous functions in the background: >>> async def print_state(dask_worker): # doctest: +SKIP ... while True: ... print(dask_worker.status) ... await asyncio.sleep(1) >>> c.run(print_state, wait=False) # doctest: +SKIP rw)rrrv)rrhrr rmrnrrs rrpz Client.run& sIftyy II         rc |dkrytjjd}t|tst dt |d|dtjjd}t|tst dt |d|dd }d }|l|r6tjd j|Dcgc]}d |d c}}|r;tjd d jd |Dzdz}n |dkDr|nd}g}ttjtjjdD]c\}\} } t!||k\rnL|||k7r#| j"j$dvr<|r,|j'| j(jddrj|r&|j'| j"j*rt-| | | j"j.z | j"j*} |j1t3ddt5j6| i| tE| jds| jj(d}|dk(rn4 tjF|jH}|jMdsdntOtQ|Scc}w#t8$rT ddlm} | } | 0|j1t3dd| j>j@i| n#tB$rYnwxYwYuwxYw#tJ$rYwxYw)aWalk up the stack to the user code and extract the code surrounding the compute/submit/persist call. All modules encountered which are ignored through the option `distributed.diagnostics.computations.ignore-modules` will be ignored. This can be used to exclude commonly used libraries which wrap dask/distributed compute calls. ``stacklevel`` may be used to explicitly indicate from which frame on the stack to get the source code. rrz3distributed.diagnostics.computations.ignore-modulesz-Ignored modules must be a list. Instead got (z, )z1distributed.diagnostics.computations.ignore-filesz+Ignored files must be a list. Instead got (N|z(?:z.*[\\/](c3 K|]}|ywrr)rbmods rrdz/Client._get_computation_code.. s*G,33,s z )([\\/]|$)r)z z rr)rrrr) get_ipython f_globals__channelexec__interactiveshell))rrrrrr,rrecompiler enumerater walk_stackr\ _getframef_backrf_codeco_namematchr co_filenamerco_firstlinenor{rr getsourcerIPythonr~history_manager_i00rrmodulesrrrrreversed)r<nframesignore_modules ignore_files mod_pattern fname_patternr}rrfrrrr~ip module_names r_get_computation_codezClient._get_computation_code s a< A .$/()N+;1> {{ ? ,-&'r,q:  *. +/   jjHHnEnsC5lnEF  " #((*G,*G"GG-W! (2A~1J!#%.  !7!7 8!& !A!L4yG#%!z/yy  $@@{001A1A*b1QR!4!4RYY5J5J!K) ,ryy/G/G G..F  JLG,=,=b,ALVLMryy+. ii11*= "33"%++k":"C"CK ''(:;]& `Xd^$$wFB 3$B~ J$VB4F4F4K4K$Vv$VW" & sN K /KL3 L0?LL0 L)&L0(L))L0/L03 M?McR|jdt|jdy)Nclient-desires-keys)rr)rrrrs rrcz#Client._inform_scheduler_of_futures s$ +T]]+  rc J|j5| "| dur| durt|j| } t|ts!t j t ||d}i} |r|| d<|r#t|tttfs|g}|| d<| r| | d<|dvr td |r|| d <|r|| d <ttj| } t|} | D] }t|| Dcic]}|t||}}d d lm}d dlm}|||d\}}t't)t*|g|z}|t-tj.j1dkDr"t3j4dt7|d|j9tj.j1d}|j;d||t||t=t>dd| | |||| ||d |cdddScc}w#1swYyxYw)NTFr dependenciesrrrrz1allow_other_workers= must be True, False, or Nonerrr) serialize)ToPicklerrnz/distributed.admin.large-graph-warning-thresholdzSending large graph of size a . This may cause some slowdown. Consider loading the data with Dask directly or using futures or delayed objects to embed the data into the graph without repetition. See also https://docs.dask.org/en/stable/best-practices.html#load-data-with-dask for more information.z,distributed.diagnostics.computations.nframesrz update-graphr) r graph_header graph_framesrrsubmitting_taskrrrrr) r r _expand_keyrr)from_collectionsrrrnr,r"rget_annotationsr(rdistributed.protocolrdistributed.protocol.serializerrrrqr4rrrrr2rrrrs)rrrrrrrrrrrrrkeysetrrrrheaderframes pickled_size computationss rrzClient._graph_to_futures s9 !fD&8V5=Pd..v67c>2$55bgsQSTK*7 J'!'D%+=>&iG)0 I&)0 I&"*== STT"5H 12+4 K(  4 4 6 DKYFS!:@@#sF3--G@ 6 ?&x}wGNFFs6F8f+<=>Lk  QR 2< 3M2NO}} 55 (VW6L  # #($*$*I):'.|UD'I$0$$\2#+K#8%-m%<  O! DAE! sC'H4HDHHH"c  :|j|tt|g|||| | | | tddig }t ||}|rwt t ddr td} |j|||}|jD]}|jt t ddr r t|S|S#t$rd}YhwxYw#|jD]}|jt t ddrr twwwxYw) a Compute dask graph Parameters ---------- dsk : dict keys : object, or nested lists of objects workers : string or iterable of strings A set of worker addresses or hostnames on which computations may be performed. Leave empty to default to all workers (common case) allow_other_workers : bool (defaults to False) Used with ``workers``. Indicates whether or not the computations may be performed on workers that are not in the `workers` set(s). resources : dict (defaults to {}) Defines the ``resources`` each instance of this mapped task requires on the worker; e.g. ``{'GPU': 2}``. See :doc:`worker resources ` for details on defining resources. sync : bool (optional) Returns Futures if False or concrete values if True (default). asynchronous: bool If True the client is in asynchronous mode direct : bool Whether or not to connect directly to the workers, or to ask the scheduler to serve as intermediary. This can also be set when creating the Client. retries : int (default to 0) Number of allowed automatic retries if computing a result fails priority : Number Optional prioritization of task. Zero is default. Higher priorities take precedence fifo_timeout : timedelta str (defaults to '60s') Allowed amount of time between calls to consider the same priority actors : bool or dict (default None) Whether these tasks should exist on the worker as stateful actors. Specified on a global (True/False) or per-task (``{'x': True, 'y': False}``) basis. See :doc:`actors` for additional details. Returns ------- results If 'sync' is True, returns the results. Otherwise, returns the known data packed If 'sync' is False, returns the known data. Otherwise, returns the results Examples -------- >>> from operator import add # doctest: +SKIP >>> c = Client('127.0.0.1:8787') # doctest: +SKIP >>> c.get({'x': (add, 1, 2)}, 'x') # doctest: +SKIP 3 See Also -------- Client.compute : Compute asynchronous collections rzlow-level-graphr) rrrrrrrrrrFT)rr!) rrnr'rfrurrsr{rr0rgr@rg)rrrrrrrrrr!rrrrrrpacked should_rejoinrtrs rrz Client.getG sR(( WdV_% 3%"&V=N4O3PQ)  4) |UE2*H$(M ++f>> len(x.__dask_graph__()) # x is a dask collection with 100 tasks # doctest: +SKIP 100 >>> set(client.futures).intersection(x.__dask_graph__()) # some overlap exists # doctest: +SKIP 10 >>> x = client.normalize_collection(x) # doctest: +SKIP >>> len(x.__dask_graph__()) # smaller computational graph # doctest: +SKIP 20 See Also -------- Client.persist : trigger computation of collection's tasks )__dask_graph__r __dask_keys__redict_collection)r collectiondsk_origrs rnormalize_collectionzClient.normalize_collection sHB,,.++Hj6N6N6PQ (? $Z5 5rc t|ttttfrd} n|g}d} | rtd|D}|Dcgc]}t j |s|}}t|Dcgc] }t|c}}|j||fi| }|Dcgc]}dt|z}}i}tt||D]r\}\}}|j\}}|j}|turt!|dk(r |s |d||<Mt#||t%|g|}|||j&<tt|t(s!t)j*t-||d }t|}||i}|j/|j0|t|j0j3i}|j/|j4t)||}|j7|||||||| | | }d}g}|D]E} t j | r|j9||||dz }5|j9| G|r|j;|}!n|}!| r t=|!S|!Scc}wcc}wcc}w) ab Compute dask collections on cluster Parameters ---------- collections : iterable of dask objects or single dask object Collections like dask.array or dataframe or dask.value objects sync : bool (optional) Returns Futures if False (default) or concrete values if True optimize_graph : bool Whether or not to optimize the underlying graphs workers : string or iterable of strings A set of worker hostnames on which computations may be performed. Leave empty to default to all workers (common case) allow_other_workers : bool (defaults to False) Used with `workers`. Indicates whether or not the computations may be performed on workers that are not in the `workers` set(s). retries : int (default to 0) Number of allowed automatic retries if computing a result fails priority : Number Optional prioritization of task. Zero is default. Higher priorities take precedence fifo_timeout : timedelta str (defaults to '60s') Allowed amount of time between calls to consider the same priority traverse : bool (defaults to True) By default dask traverses builtin python collections looking for dask objects passed to ``compute``. For large collections this can be expensive. If none of the arguments contain any dask objects, set ``traverse=False`` to avoid doing this traversal. resources : dict (defaults to {}) Defines the `resources` each instance of this mapped task requires on the worker; e.g. ``{'GPU': 2}``. See :doc:`worker resources ` for details on defining resources. actors : bool or dict (default None) Whether these tasks should exist on the worker as stateful actors. Specified on a global (True/False) or per-task (``{'x': True, 'y': False}``) basis. See :doc:`actors` for additional details. **kwargs Options to pass to the graph optimize calls Returns ------- List of Futures if input is a sequence, or a single future otherwise Examples -------- >>> from dask import delayed >>> from operator import add >>> x = delayed(add)(1, 2) >>> y = delayed(add)(x, x) >>> xx, yy = client.compute([x, y]) # doctest: +SKIP >>> xx # doctest: +SKIP >>> xx.result() # doctest: +SKIP 3 >>> yy.result() # doctest: +SKIP 6 Also support single arguments >>> xx = client.compute(x) # doctest: +SKIP See Also -------- Client.get : Normal synchronous dask.get function FTc 3K|]@}t|ttttt frt j|n|Bywr)rrrnrrr rdelayedrs rrdz!Client.compute..GsC  %A"!dCh%GHLLO%sAArz finalize-%srrrrrrrrrrrr)rrrrnr:ris_dask_collectionrfget_collections_metadatar&r,rr__dask_postcompute__rr=rrC_convert_dask_keysrr)rrrlayersrrrr{r0r )"rrrroptimize_graphrrrrrrrtraverser singletonr variablesrmetadatarrAdsk2rr2r extra_argsrt finalize_namerr futures_dictrargrs" rcomputezClient.compute sb kD%i#@ AI&-KI   %  K!,J 1t/F/Fq/IQ J>GHi1!4iH &d%%iJ6J6?@i!,i@%c%&; C,,-V\2--   3"%".   C&&s+|E!H56Qs#  [[)FF = MuKHAsI5I5+I:I?c  t|ttttfrd} nd} |g}t t tj|sJt|D cgc] } t| c} } |j||fi| }|Dchc]"}t|jD]}|$}}}|j||||||||| |  }|Dcgc]}|j}}t!||Dcgc]:\\}}}|t|jDcic]}||| c}g|<}}}}}| r t#|S|Scc} wcc}}wcc}wcc}wcc}}}}w)aPersist dask collections on cluster Starts computation of the collection on the cluster in the background. Provides a new dask collection that is semantically identical to the previous one, but now based off of futures currently in execution. Parameters ---------- collections : sequence or single dask object Collections like dask.array or dataframe or dask.value objects optimize_graph : bool Whether or not to optimize the underlying graphs workers : string or iterable of strings A set of worker hostnames on which computations may be performed. Leave empty to default to all workers (common case) allow_other_workers : bool (defaults to False) Used with `workers`. Indicates whether or not the computations may be performed on workers that are not in the `workers` set(s). retries : int (default to 0) Number of allowed automatic retries if computing a result fails priority : Number Optional prioritization of task. Zero is default. Higher priorities take precedence fifo_timeout : timedelta str (defaults to '60s') Allowed amount of time between calls to consider the same priority resources : dict (defaults to {}) Defines the `resources` each instance of this mapped task requires on the worker; e.g. ``{'GPU': 2}``. See :doc:`worker resources ` for details on defining resources. actors : bool or dict (default None) Whether these tasks should exist on the worker as stateful actors. Specified on a global (True/False) or per-task (``{'x': True, 'y': False}``) basis. See :doc:`actors` for additional details. **kwargs Options to pass to the graph optimize calls Returns ------- List of collections, or single collection, depending on type of input. Examples -------- >>> xx = client.persist(x) # doctest: +SKIP >>> xx, yy = client.persist([x, y]) # doctest: +SKIP See Also -------- Client.compute FTrr)rrrrnr:rhrrrrfrr&r'rr__dask_postpersist__rr )rrrrrrrrrrrrrrrrrrAr postpersistsrrrs rpersistzClient.persists~ kE4i#@ AII&-K3t.. <===>IJk1!4kJ &d%%k>LVL'LKq9J1KA1KKL((   3"%")  ;FF+Q..0+ F$'|[#A #A ta 1B)CD)CA!WQZ-)CD Lt L#A  = M9KMGD s*E 'EE=(E% % E 2 E% E% cK|tur|jdz}ttd|d}|jj ||dt d{y7w)N str|int|floatrzclient-restart-)r r stimulus_id)r0rr5rrrr[rr rs r_restartzClient._restartsd j mma'G!$"@#Fnn$$-)$&2%   sAA$A"A$c>|j|j||S)a% Restart all workers. Reset local state. Optionally wait for workers to return. Workers without nannies are shut down, hoping an external deployment system will restart them. Therefore, if not using nannies and your deployment system does not automatically restart workers, ``restart`` will just shut down all workers, then time out! After ``restart``, all connected workers are new, regardless of whether ``TimeoutError`` was raised. Any workers that failed to shut down in time are removed, and may or may not shut down on their own in the future. Parameters ---------- timeout: How long to wait for workers to shut down and come back, if ``wait_for_workers`` is True, otherwise just how long to wait for workers to shut down. Raises ``asyncio.TimeoutError`` if this is exceeded. wait_for_workers: Whether to wait for all workers to reconnect, or just for them to shut down (default True). Use ``restart(wait_for_workers=False)`` combined with :meth:`Client.wait_for_workers` for granular control over how many workers to wait for. See also -------- Scheduler.restart Client.restart_workers )r r)rrrrs rrzClient.restarts'Dyy MM7=M  rc@K|tur|jdz}ttd|d}|j }|dj Dcic] \}}|d| }}}|Dcgc]}|j ||} }|jj| ||rdnddt d{} t|| D cic] \}} || |  } }} |r'td | jDsJ| | Scc}}wcc}w7\cc} }ww) Nrrrrr2rrzclient-restart-workers-)rr rnrc3&K|] }|dk( yw)r.Nr)rbrs rrdz*Client._restart_workers..=s7,QqDy,re) r0rr5rscheduler_inforrrrestart_workersr[rrhrg) rrr raise_for_errorrrimeta name_to_addrrc worker_addrsrJw_addrs r_restart_workerszClient._restart_workers$s2 j mma'G!$"@#F""$=A)_=R=R=TU=TztTV d*=T U8?@1 ((A. @..00$$35dfX> 1  037L/IJ/I)!Vq#f+~/IJ 7#**,77 < <7 V@ Ks<ADD &D-D3D9D:D D:DDc|j}|djD]1\}}||vs|d|vs|dtd|d|d|dd|j|j|||S) aRestart a specified set of workers .. note:: Only workers being monitored by a :class:`distributed.Nanny` can be restarted. See ``Nanny.restart`` for more details. Parameters ---------- workers : list[str] Workers to restart. This can be a list of worker addresses, names, or a both. timeout : int | float | None Number of seconds to wait raise_for_error: bool (default True) Whether to raise a :py:class:`TimeoutError` if restarting worker(s) doesn't finish within ``timeout``, or another exception caused from restarting worker(s). Returns ------- dict[str, "OK" | "removed" | "timed out"] Mapping of worker and restart status, the keys will match the original values passed in via ``workers``. Notes ----- This method differs from :meth:`Client.restart` in that this method simply restarts the specified set of workers, while ``Client.restart`` will restart all workers and also reset local state on the cluster (e.g. all keys are released). Additionally, this method does not gracefully handle tasks that are being executed when a worker is restarted. These tasks may fail or have their suspicious count incremented. Examples -------- You can get information about active workers using the following: >>> workers = client.scheduler_info()['workers'] From that list you may want to select some workers to restart >>> client.restart_workers(workers=['tcp://address:port', ...]) See Also -------- Client.restart rr2rmz7Restarting workers requires a nanny to be used. Worker z has type rr)rr r)rrrrrr)rrr rrrrs rrzClient.restart_workers@sn""$ O113LFD'!T&\W%<$w-BW Mhji)@)H(IL4 yy  ! !+   rcK"tjj|dt|d5}|j ddd|j gd{\}|j |j|d{dfd }|j|d{}tfd|jDsJy#1swYxYw7|7W79w)NrrbcJtjjs+tjj|j}n}t |d5}|j |jdddt|jS#1swY!xYw)Nwb) r*risabsrlocal_directoryrrrr) dask_workerr&rrremote_filenames r dump_to_filez/Client._upload_large_file..dump_to_filesy77==1WW\\+"="=O$b$1 ((-. {'',- - s BB"c3:K|]}t|k(ywrr)rbrrs rrdz,Client._upload_large_file..s=++rvrhrg) rlocal_filenamerrrrr*rrs ` @@r_upload_large_filezClient._upload_large_files  " ggmmN;A>O .$ '1668D(v..jjoof%%% .<00=8??+<===='( '/%1sL1C'CC'$C!%&C' C# C'+C%,)C'CC'#C'%C'c|ttjzfd}j|S)aUpload local package to scheduler and workers This sends a local file up to the scheduler and all worker nodes. This file is placed into the working directory of each node, see config option ``temporary-directory`` (defaults to :py:func:`tempfile.gettempdir`). This directory will be added to the Python's system path so any ``.py``, ``.egg`` or ``.zip`` files will be importable. Parameters ---------- filename : string Filename of ``.py``, ``.egg``, or ``.zip`` file to send to workers load : bool, optional Whether or not to import the module as part of the upload process. Defaults to ``True``. Examples -------- >>> client.upload_file('mylibrary.egg') # doctest: +SKIP >>> from mylibrary import myfunc # doctest: +SKIP >>> L = client.map(myfunc, seq) # doctest: +SKIP >>> >>> # Where did that file go? Use `dask_worker.local_directory`. >>> def where_is_mylibrary(dask_worker): >>> path = pathlib.Path(dask_worker.local_directory) / 'mylibrary.egg' >>> assert path.exists() >>> return str(path) >>> >>> client.run(where_is_mylibrary) # doctest: +SKIP c Ktjjtjt d{}|dS7 w)N)rr2r)rir0register_pluginrWrX)rtrrr2rs rrzzClient.upload_file.._sf#NN$$'t<4%$$Zt%D4$P G1: sAA!A A!)rr_r`rr)rrrrzr2s``` @r upload_filezClient.upload_files0@#djjl++ yy|rcLK|Ft|d{t|j|Dchc]}|jc}}nd}|jj ||d{}|ddk(rt d|d|ddk(sJ|y7cc}w74w)N)rrrz partial-failzCould not rebalance keys: rr.)_waitrrIrr rebalancer)rrrrrrs r _rebalancezClient._rebalances  . (@A(@1(@ABDD~~//T7/KK ( ~ -7v7GHI Ih4'//' !ALs,B$BB$B*B$-B"..B$B$c @|j|j||fi|S)aRebalance data within network Move data between workers to roughly balance memory burden. This either affects a subset of the keys/workers or the entire network, depending on keyword arguments. For details on the algorithm and configuration options, refer to the matching scheduler-side method :meth:`~distributed.scheduler.Scheduler.rebalance`. .. warning:: This operation is generally not well tested against normal operation of the scheduler. It is not recommended to use it while waiting on computations. Parameters ---------- futures : list, optional A list of futures to balance, defaults all data workers : list, optional A list of workers on which to balance, defaults to all workers **kwargs : dict Optional keyword arguments for the function )rrr)rrrrs rrzClient.rebalances".tyy'7EfEErcK|j|}t|d{|Dchc]}|j}}|jj t ||||d{y7Ncc}w7 w)N)rr8rbranching_factor)rIrrr replicater)rrr8rrrrs rr>zClient._replicatesu//'*Gn&'w!w'nn&&dq'DT'    ' s+ A:A1A:A3-A:+A8,A:3A:c F|j|j|f|||d|S)aSet replication of futures within network Copy data onto many workers. This helps to broadcast frequently accessed data and can improve resilience. This performs a tree copy of the data throughout the network individually on each piece of data. This operation blocks until complete. It does not guarantee replication of data to future workers. .. note:: This method is incompatible with the Active Memory Manager's :ref:`ReduceReplicas` policy. If you wish to use it, you must first disable the policy or disable the AMM entirely. Parameters ---------- futures : list of futures Futures we wish to replicate n : int, optional Number of processes on the cluster on which to replicate the data. Defaults to all. workers : list of worker addresses Workers on which we want to restrict the replication. Defaults to all. branching_factor : int, optional The number of workers that can copy data in each generation **kwargs : dict Optional keyword arguments for the remote function Examples -------- >>> x = c.submit(func, *args) # doctest: +SKIP >>> c.replicate([x]) # send to all workers # doctest: +SKIP >>> c.replicate([x], n=3) # send to three workers # doctest: +SKIP >>> c.replicate([x], workers=['alice', 'bob']) # send to specific # doctest: +SKIP >>> c.replicate([x], n=1, workers=['alice', 'bob']) # send to one of specific workers # doctest: +SKIP >>> c.replicate([x], n=1) # reduce replications # doctest: +SKIP See Also -------- Client.rebalance )r8rr)rrr>)rrr8rrrs rrzClient.replicates<Vtyy OO  -     rc t|trtd|Dr t|}|t|tttfs|g}|j |j jfd|i|S)aThe number of threads/cores available on each worker node Parameters ---------- workers : list (optional) A list of workers that we care about specifically. Leave empty to receive information about all workers. **kwargs : dict Optional keyword arguments for the remote function Examples -------- >>> c.nthreads() # doctest: +SKIP {'192.168.1.141:46784': 8, '192.167.1.142:47548': 8, '192.167.1.143:47329': 8, '192.167.1.144:37297': 8} See Also -------- Client.who_has Client.has_what c3HK|]}t|ttfywrrrrr s rrdz"Client.nthreads..E. 18AJq3, ' "r)rrrhrrnrrrncores)rrrs rrazClient.nthreads-so0 gu %#. 18. + 7mG  z'E4;M'NiGtyy..JJ6JJrc |5j|}t|Dchc]}|jc}ndfd}j|Scc}w)aThe workers storing each future's data Parameters ---------- futures : list (optional) A list of futures, defaults to all data **kwargs : dict Optional keyword arguments for the remote function Examples -------- >>> x, y, z = c.map(inc, [1, 2, 3]) # doctest: +SKIP >>> wait([x, y, z]) # doctest: +SKIP >>> c.who_has() # doctest: +SKIP {'inc-1c8dd6be1c21646c71f76c16d09304ea': ['192.168.1.141:46784'], 'inc-1e297fc27658d7b67b3a758f16bcf47a': ['192.168.1.141:46784'], 'inc-fd65c238a7ea60f6a01bf4c8a5fcf44b': ['192.168.1.141:46784']} >>> c.who_has([x, y]) # doctest: +SKIP {'inc-1c8dd6be1c21646c71f76c16d09304ea': ['192.168.1.141:46784'], 'inc-1e297fc27658d7b67b3a758f16bcf47a': ['192.168.1.141:46784']} See Also -------- Client.has_what Client.nthreads NcnKtjjddid{S7w)Nrr)r^rr/)rrrsrrzzClient.who_has.._qs2 6 6 6 KD KF KKL LK '53 5)rIrrrr)rrrrrzrs` ` @rr/zClient.who_hasOsV8  oog.G0101DD Myy|1sAc ttrtdDr tttttfsgfd}j |S)aaWhich keys are held by which workers This returns the keys of the data that are held in each worker's memory. Parameters ---------- workers : list (optional) A list of worker addresses, defaults to all **kwargs : dict Optional keyword arguments for the remote function Examples -------- >>> x, y, z = c.map(inc, [1, 2, 3]) # doctest: +SKIP >>> wait([x, y, z]) # doctest: +SKIP >>> c.has_what() # doctest: +SKIP {'192.168.1.141:46784': ['inc-1c8dd6be1c21646c71f76c16d09304ea', 'inc-fd65c238a7ea60f6a01bf4c8a5fcf44b', 'inc-1e297fc27658d7b67b3a758f16bcf47a']} See Also -------- Client.who_has Client.nthreads Client.processing c3HK|]}t|ttfywrrr s rrdz"Client.has_what..r r cnKtjjddid{S7w)Nrr)r\rhas_what)rrrsrrzzClient.has_what.._s2!8!8!8!S!SF!SST TSr)rrrhrrnrr)rrrrzs``` rrzClient.has_whatvsa8 gu %#. 18. + 7mG  z'E4;M'NiG Uyy|rct|trtd|Dr t|}|t|tttfs|g}|j |j j|S)aThe tasks currently running on each worker Parameters ---------- workers : list (optional) A list of worker addresses, defaults to all Examples -------- >>> x, y, z = c.map(inc, [1, 2, 3]) # doctest: +SKIP >>> c.processing() # doctest: +SKIP {'192.168.1.141:46784': ['inc-1c8dd6be1c21646c71f76c16d09304ea', 'inc-fd65c238a7ea60f6a01bf4c8a5fcf44b', 'inc-1e297fc27658d7b67b3a758f16bcf47a']} See Also -------- Client.who_has Client.has_what Client.nthreads c3HK|]}t|ttfywrrr s rrdz$Client.processing..r r r6)rrrhrrnrrr processing)rrs rrzClient.processingsg, gu %#. 18. + 7mG  z'E4;M'NiGyy22GyDDrc V|j|jjf||d|S)axThe bytes taken up by each key on the cluster This is as measured by ``sys.getsizeof`` which may not accurately reflect the true cost. Parameters ---------- keys : list (optional) A list of keys, defaults to all keys summary : boolean, (optional) Summarize keys into key types **kwargs : dict Optional keyword arguments for the remote function Examples -------- >>> x, y, z = c.map(inc, [1, 2, 3]) # doctest: +SKIP >>> c.nbytes(summary=False) # doctest: +SKIP {'inc-1c8dd6be1c21646c71f76c16d09304ea': 28, 'inc-1e297fc27658d7b67b3a758f16bcf47a': 28, 'inc-fd65c238a7ea60f6a01bf4c8a5fcf44b': 28} >>> c.nbytes(summary=True) # doctest: +SKIP {'inc': 84} See Also -------- Client.who_has )rsummary)rrrrq)rrrrs rrqz Client.nbytess+<tyy..UT7UfUUrc|xsg}|7|j|}|t|Dchc]}|jc}z }|j|jj |xsdScc}w)aThe actively running call stack of all relevant keys You can specify data of interest either by providing futures or collections in the ``futures=`` keyword or a list of explicit keys in the ``keys=`` keyword. If neither are provided then all call stacks will be returned. Parameters ---------- futures : list (optional) List of futures, defaults to all data keys : list (optional) List of key names, defaults to all data Examples -------- >>> df = dd.read_parquet(...).persist() # doctest: +SKIP >>> client.call_stack(df) # call on collections >>> client.call_stack() # Or call with no arguments for all activity # doctest: +SKIP Nr-)rIrrrrr call_stack)rrrrs rrzClient.call_stacksi,zr  oog.G D1A!%%12 2Dyy22yFF2sA+ c L|j|j|||||||||  S)akCollect statistical profiling information about recent work Parameters ---------- key : str Key prefix to select, this is typically a function name like 'inc' Leave as None to collect all data start : time stop : time workers : list List of workers to restrict profile information server : bool If true, return the profile of the worker's administrative thread rather than the worker threads. This is useful when profiling Dask itself, rather than user code. scheduler : bool If true, return the profile information from the scheduler's administrative thread rather than the workers. This is useful when profiling Dask's scheduling itself. plot : boolean or string Whether or not to return a plot object filename : str Filename to save the plot Examples -------- >>> client.profile() # call on collections >>> client.profile(filename='dask-profile.html') # save to html file ) rr merge_workersr.rplotrrr)rr_profile) rrr.rrrrrrrs rprofilezClient.profiles=Ryy MM'  rc TKt|ttfr|g}|jj ||||||| d{} |rd}|rXddlm} | j | } | j| d\} }|dk(r|sd}|rdd lm }m }||d || | | | fS| S7dw) N)rrrr.rrrTr)r stretch_both sizing_modesavezdask-profile.html output_filer#z Dask Profilertitle)r) rrrrrr plot_data plot_figurebokeh.plottingr%r#)rrr.rrrrrrrrrrfiguresourcer%r#s rrzClient._profile.s gV} -iGnn,,'-   D  +$$U+D$00>0RNFFv~h.<X^DVh/6? "L= s?B(B&A%B(c h|js|j|j|jS)aBasic information about the workers in the cluster Parameters ---------- **kwargs : dict Optional keyword arguments for the remote function Examples -------- >>> c.scheduler_info() # doctest: +SKIP {'id': '2de2b6da-69ee-11e6-ab6a-e82aea155996', 'services': {}, 'type': 'Scheduler', 'workers': {'127.0.0.1:40575': {'active': 0, 'last-seen': 1472038237.4845693, 'name': '127.0.0.1:40575', 'services': {}, 'stored': 0, 'time-delay': 0.0061032772064208984}}} )rrrrr r8s rrzClient.scheduler_info]s+*  IId11 2'''rdask-cluster-dumpc F|j|jf||||d|S)a Extract a dump of the entire cluster state and persist to disk or a URL. This is intended for debugging purposes only. Warning: Memory usage on the scheduler (and client, if writing the dump locally) can be large. On a large or long-running cluster, this can take several minutes. The scheduler may be unresponsive while the dump is processed. Results will be stored in a dict:: { "scheduler": {...}, # scheduler state "workers": { worker_addr: {...}, # worker state ... } "versions": { "scheduler": {...}, "workers": { worker_addr: {...}, ... } } } Parameters ---------- filename: The path or URL to write to. The appropriate file suffix (``.msgpack.gz`` or ``.yaml``) will be appended automatically. Must be a path supported by :func:`fsspec.open` (like ``s3://my-bucket/cluster-dump``, or ``cluster-dumps/dump``). See ``write_from_scheduler`` to control whether the dump is written directly to ``filename`` from the scheduler, or sent back to the client over the network, then written locally. write_from_scheduler: If None (default), infer based on whether ``filename`` looks like a URL or a local path: True if the filename contains ``://`` (like ``s3://my-bucket/cluster-dump``), False otherwise (like ``local_dir/cluster-dump``). If True, write cluster state directly to ``filename`` from the scheduler. If ``filename`` is a local path, the dump will be written to that path on the *scheduler's* filesystem, so be careful if the scheduler is running on ephemeral hardware. Useful when the scheduler is attached to a network filesystem or persistent disk, or for writing to buckets. If False, transfer cluster state from the scheduler back to the client over the network, then write it to ``filename``. This is much less efficient for large dumps, but useful when the scheduler doesn't have access to any persistent storage. exclude: A collection of attribute names which are supposed to be excluded from the dump, e.g. to exclude code, tracebacks, logs, etc. Defaults to exclude ``run_spec``, which is the serialized user code. This is typically not required for debugging. To allow serialization of this, pass an empty tuple. format: Either ``"msgpack"`` or ``"yaml"``. If msgpack is used (default), the output will be stored in a gzipped file as msgpack. To read:: import gzip, msgpack with gzip.open("filename") as fd: state = msgpack.unpack(fd) or:: import yaml try: from yaml import CLoader as Loader except ImportError: from yaml import Loader with open("filename") as fd: state = yaml.load(fd, Loader=Loader) **storage_options: Any additional arguments to :func:`fsspec.open` when writing to a URL. )rwrite_from_schedulerexcluder)rr_dump_cluster_staterrr0r1rstorage_optionss rdump_cluster_statezClient.dump_cluster_statevs<ltyy  $ $ !5     rc Kt|}|d|v}|r)|jjd|||d|d{ytjt |jj |||fi|d{y7E7w)NrM)urlr1r)r1r)rrdump_cluster_state_to_urlrF write_staterget_cluster_stater3s rr2zClient._dump_cluster_statesx=  '#(H#4 :$..::"   **88'J"     s!7BA??B9B:BBc|jr td||_t|jd5}tj|j |ddddy#1swYyxYw)aWrite the scheduler information to a json file. This facilitates easy sharing of scheduler information using a file system. The scheduler file can be used to instantiate a second Client using the same scheduler. Parameters ---------- scheduler_file : str Path to a write the scheduler file. Examples -------- >>> client = Client() # doctest: +SKIP >>> client.write_scheduler_file('scheduler.json') # doctest: +SKIP # connect to previous client's scheduler >>> client2 = Client(scheduler_file='scheduler.json') # doctest: +SKIP zScheduler file already setrcr)indentN)rrrrdumpr)rrrs rwrite_scheduler_filezClient.write_scheduler_filesX&   9: :"0D  $%%s +q IId))+Qq 9, + +s 'A%%A.ct|ttfs|f}|j|jj ||S)aGet arbitrary metadata from scheduler See set_metadata for the full docstring with examples Parameters ---------- keys : key or list Key to access. If a list then gets within a nested collection default : optional If the key does not exist then return this value instead. If not provided then this raises a KeyError if the key is not present See Also -------- Client.set_metadata )rr)rrrrrr get_metadata)rrrs rr@zClient.get_metadata s8$$u .7Dyy444yQQrcP|j|jj|S)aQGet logs from scheduler Parameters ---------- n : int Number of logs to retrieve. Maxes out at 10000 by default, configurable via the ``distributed.admin.log-length`` configuration value. Returns ------- Logs in reversed order (newest first) )r8)rrrlogs)rr8s rget_scheduler_logszClient.get_scheduler_logs!s!yy,,y22rcT|j|jj|||S)aGet logs from workers Parameters ---------- n : int Number of logs to retrieve. Maxes out at 10000 by default, configurable via the ``distributed.admin.log-length`` configuration value. workers : iterable List of worker addresses to retrieve. Gets all workers by default. nanny : bool, default False Whether to get the logs from the workers (False) or the nannies (True). If specified, the addresses in `workers` should still be the worker addresses, not the nanny addresses. Returns ------- Dictionary mapping worker address to logs. Logs are returned in reversed order (newest first) )r8rrm)rrr worker_logs)rr8rrms rget_worker_logszClient.get_worker_logs1s&*yy33q'QVyWWrcL|j|jjS)ay Run a benchmark on the workers for memory, disk, and network bandwidths Returns ------- result: dict A dictionary mapping the names "disk", "memory", and "network" to dictionaries mapping sizes to bandwidths. These bandwidths are averaged over many workers running computations across the cluster. )rrrbenchmark_hardwarers rrHzClient.benchmark_hardwareHsyy::;;rct|stdt|d|j|jj ||S)aLog an event under a given topic Parameters ---------- topic : str, list[str] Name of the topic under which to log an event. To log the same event under multiple topics, pass a list of topic names. msg Event message to log. Note this must be msgpack serializable. Examples -------- >>> from time import time >>> client.log_event("current-time", time()) z4Message must be msgpack serializable. Got type(msg)=z instead.)rr)r:r,rrrr log_eventrrrs rrJzClient.log_eventUsJ C GT#YL R yy11CyHHrcP|j|jj|S)zRetrieve structured topic logs Parameters ---------- topic : str, optional Name of topic log to retrieve events for. If no ``topic`` is provided, then logs for all topics will be returned. )r)rrrevents)rrs r get_eventszClient.get_eventsks!yy..ey<>> import logging >>> logger = logging.getLogger("myLogger") # Log config not shown >>> client.subscribe_topic("topic-name", lambda: logger.info) See Also -------- dask.distributed.Client.unsubscribe_topic dask.distributed.Client.get_events dask.distributed.Client.log_event z(Handler for %s already set. Overwriting.zsubscribe-topicrrrN)r#r$rrr)rrrrs rrzClient.subscribe_topicsO2 D(( ( KKBE J&-U#&$''J $rc||jvr"d||jd}|j|ytd|d)zUnsubscribe from a topic and remove event handler See Also -------- dask.distributed.Client.subscribe_topic dask.distributed.Client.get_events dask.distributed.Client.log_event zunsubscribe-topicrSz!No event handler known for topic rN)r#rrrrKs rrPzClient.unsubscribe_topicsE D(( (,uPC  # #C (@qIJ Jrc V|j|jjf||d|S)aRetire certain workers on the scheduler See :meth:`distributed.Scheduler.retire_workers` for the full docstring. Parameters ---------- workers close_workers **kwargs : dict Optional keyword arguments for the remote function Examples -------- You can get information about active workers using the following: >>> workers = client.scheduler_info()['workers'] From that list you may want to select some workers to close >>> client.retire_workers(workers=['tcp://address:port', ...]) See Also -------- dask.distributed.Scheduler.retire_workers )r close_workers)rrrretire_workers)rrrVrs rrWzClient.retire_workerss98tyy NN ) ) '    rcxt|ts|f}|j|jj||S)avSet arbitrary metadata in the scheduler This allows you to store small amounts of data on the central scheduler process for administrative purposes. Data should be msgpack serializable (ints, strings, lists, dicts) If the key corresponds to a task then that key will be cleaned up when the task is forgotten by the scheduler. If the key is a list then it will be assumed that you want to index into a nested dictionary structure using those keys. For example if you call the following:: >>> client.set_metadata(['a', 'b', 'c'], 123) Then this is the same as setting >>> scheduler.task_metadata['a']['b']['c'] = 123 The lower level dictionaries will be created on demand. Examples -------- >>> client.set_metadata('x', 123) # doctest: +SKIP >>> client.get_metadata('x') # doctest: +SKIP 123 >>> client.set_metadata(['x', 'y'], 123) # doctest: +SKIP >>> client.get_metadata('x') # doctest: +SKIP {'y': 123} >>> client.set_metadata(['x', 'w', 'z'], 456) # doctest: +SKIP >>> client.get_metadata('x') # doctest: +SKIP {'y': 123, 'w': {'z': 456}} >>> client.get_metadata(['x', 'w']) # doctest: +SKIP {'z': 456} See Also -------- get_metadata )rrB)rrrrr set_metadata)rrrBs rrYzClient.set_metadatas5V#t$&Cyy443eyLLrcF|j|j||xsgS)aReturn version info for the scheduler, all workers and myself Parameters ---------- check raise ValueError if all required & optional packages do not match packages Extra package names to check Examples -------- >>> c.get_versions() # doctest: +SKIP >>> c.get_versions(packages=['sklearn', 'geopandas']) # doctest: +SKIP )checkpackages)rr _get_versions)rr[r\s rrzClient.get_versionss#&yy++58>ryRRrcK|xsg}tj|}|jj|d{}|jj d|ddd{}t |||}|rGtj |||}|drtj|d|drt|d|S77_w) N)r\rH)rr\rrrrn)rrrrr) rrrrHr7r error_messagerrr)rr[r\rrrrrs rr]zClient._get_versionss>r,,h?..1181DD 00!x81   76R  ..y'6JC9~ c)n-7| W.. E s".[sB'3#'s)rrgrrr7rextend)rrnrf flat_storiesrrsstoriesrs @r_storyz Client._storyFs .... I!%!9!9 /":"L>> client.get_task_stream() # prime plugin if not already connected >>> x.compute() # do some work >>> client.get_task_stream() [{'task': ..., 'type': ..., 'thread': ..., ...}] Pass the ``plot=True`` or ``plot='save'`` keywords to get back a Bokeh figure >>> data, figure = client.get_task_stream(plot='save', filename='myfile.html') Alternatively consider the context manager >>> from dask.distributed import get_task_stream >>> with get_task_stream() as ts: ... x.compute() >>> ts.data [...] Returns ------- L: List[Dict] See Also -------- get_task_stream : a context manager version of this method )r.rrrrbokeh_resources)rr_get_task_stream)rr.rrrrrps rget_task_streamzClient.get_task_streambs6Lyy  ! !+  rc$K|jj|||d{}|raddlm}||} ddlm} | d\} } | j j| |dk(rddlm } m }| |d || || || fS|S7iw) N)r.rrr) rectangles)task_stream_figurer r!r#r$zDask Task Streamr&)rr) rrr#distributed.diagnostics.task_streamrt*distributed.dashboard.components.schedulerrurrr*r%r#)rr.rrrrrprrtrectsrur,r+r%r#s rrqzClient._get_task_streams^^33%dRW3XX  Ft$E U/NKNFF KK  u %v~<X5GHVh/J&> !K!Ys"BBA*Bc| t|}|sJ|tjdtdn t |dd}t |t sJ|j|||S)a$Register a plugin. See https://distributed.readthedocs.io/en/latest/plugins.html Parameters ---------- plugin : A nanny, scheduler, or worker plugin to register. name : Name for the plugin; if None, a name is taken from the plugin instance or automatically generated if not present. idempotent : Do not re-register if a plugin of the given name already exists. If None, ``plugin.idempotent`` is taken if defined, False otherwise. zThe `idempotent` argument is deprecated and will be removed in a future version. Please mark your plugin as idempotent by setting its `.idempotent` attribute to `True`.rr; idempotentF)rZrr FutureWarningrrr_register_pluginrpluginr2rzs rrzClient.register_pluginsn* <#F+D t  ! MM5  !u=J*d+++$$VT:>>rct|tr tdtd|jj Dr tdtd)Nz3Please provide an instance of a plugin, not a type.c36K|]}dt|vyw)z#dask.distributed.diagnostics.pluginN)r)rbrs rrdz*Client._register_plugin..s! / 2SV ;/szImporting plugin base classes from `dask.distributed.diagnostics.plugin` is not supported. Please import directly from `distributed.diagnostics.plugin` instead.zRegistering duck-typed plugins is not allowed. Please inherit from NannyPlugin, WorkerPlugin, or SchedulerPlugin to create a plugin.)rrr,anyr __bases__r}s rr|zClient._register_pluginse fd #QR R  %%//  X  P  rc@|j|j|||SNr~r2rz)rr_register_scheduler_pluginr}s rrzzClient._s*yy  + +!   rc@|j|j|||Sr)rr_register_nanny_pluginr}s rrzzClient._s,yy  ' '!   rc@|j|j|||Sr)rr_register_worker_pluginr}s rrzzClient._s*yy  ( (!   rclK|jjt|||d{S7wr)rregister_scheduler_pluginr`r}s rrz!Client._register_scheduler_plugin%s;^^===!>    s +424c|tjdtdtt|j |||S)al Register a scheduler plugin. .. deprecated:: 2023.9.2 Use :meth:`Client.register_plugin` instead. See https://distributed.readthedocs.io/en/latest/plugins.html#scheduler-plugins Parameters ---------- plugin : SchedulerPlugin SchedulerPlugin instance to pass to the scheduler. name : str Name for the plugin; if None, a name is taken from the plugin instance or automatically generated if not present. idempotent : bool Do not re-register if a plugin of the given name already exists. z_`Client.register_scheduler_plugin` has been deprecated; please `Client.register_plugin` insteadrr;)rrr>rr9rr}s rrz Client.register_scheduler_plugin.s90   6   It33FD*MNNrcVK|jj|d{S7wNr)runregister_scheduler_pluginrr2s r_unregister_scheduler_pluginz#Client._unregister_scheduler_pluginNs$^^??T?JJJJs )')c<|j|j|S)a1Unregisters a scheduler plugin See https://distributed.readthedocs.io/en/latest/plugins.html#scheduler-plugins Parameters ---------- name : str Name of the plugin to unregister. See the :meth:`Client.register_scheduler_plugin` docstring for more information. Examples -------- >>> class MyPlugin(SchedulerPlugin): ... def __init__(self, *args, **kwargs): ... pass # the constructor is up to you ... async def start(self, scheduler: Scheduler) -> None: ... pass ... async def before_close(self) -> None: ... pass ... async def close(self) -> None: ... pass ... def restart(self, scheduler: Scheduler) -> None: ... pass >>> plugin = MyPlugin(1, 2, 3) >>> client.register_plugin(plugin, name='foo') >>> client.unregister_scheduler_plugin(name='foo') See Also -------- register_scheduler_plugin r)rrrrs rrz"Client.unregister_scheduler_pluginQsByy::yFFrc6|jt|S)a Registers a setup callback function for all current and future workers. This registers a new setup function for workers in this cluster. The function will run immediately on all currently connected workers. It will also be run upon connection by any workers that are added in the future. Multiple setup functions can be registered - these will be called in the order they were added. If the function takes an input argument named ``dask_worker`` then that variable will be populated with the worker itself. Parameters ---------- setup : callable(dask_worker: Worker) -> None Function to register and run on all workers )r_WorkerSetupPluginrsetups rregister_worker_callbacksz Client.register_worker_callbacksts$##$6u$=>>rc6K|jjt|||d{}|jD]5}|ddk(s t |d|d\}}}|sJ|j |t tttf|S7iwNrrrrr) rregister_worker_pluginr`rgrQrrrrr9 rr~r2rzrsr*rzrrs rrzClient._register_worker_plugins..??=t @  "((*H!W,,[)8K+@ 3 s((,, +Di()44 +BBBA Bc6K|jjt|||d{}|jD]5}|ddk(s t |d|d\}}}|sJ|j |t tttf|S7iwr) rregister_nanny_pluginr`rgrQrrrrr9rs rrzClient._register_nanny_plugins..>>=t ?  "((*H!W,,[)8K+@ 3 s((,, +Di()44 rctjdtd| t|}|sJt |t r-|j }|durtjdtdnt |tr-|j}|durtjdtdnt |trH|rtjdtdntjd td|j}n9tjd td|dur |j}n |j }|j|||d S) a Registers a lifecycle worker plugin for all current and future workers. .. deprecated:: 2023.9.2 Use :meth:`Client.register_plugin` instead. This registers a new object to handle setup, task state transitions and teardown for workers in this cluster. The plugin will instantiate itself on all currently connected workers. It will also be run on any worker that connects in the future. The plugin may include methods ``setup``, ``teardown``, ``transition``, and ``release_key``. See the ``dask.distributed.WorkerPlugin`` class or the examples below for the interface and docstrings. It must be serializable with the pickle or cloudpickle modules. If the plugin has a ``name`` attribute, or if the ``name=`` keyword is used then that will control idempotency. If a plugin with that name has already been registered, then it will be removed and replaced by the new one. For alternatives to plugins, you may also wish to look into preload scripts. Parameters ---------- plugin : WorkerPlugin or NannyPlugin WorkerPlugin or NannyPlugin instance to register. name : str, optional A name for the plugin. Registering a plugin with the same name will have no effect. If plugin has no name attribute a random name is used. nanny : bool, optional Whether to register the plugin with workers or nannies. Examples -------- >>> class MyPlugin(WorkerPlugin): ... def __init__(self, *args, **kwargs): ... pass # the constructor is up to you ... def setup(self, worker: dask.distributed.Worker): ... pass ... def teardown(self, worker: dask.distributed.Worker): ... pass ... def transition(self, key: str, start: str, finish: str, ... **kwargs): ... pass ... def release_key(self, key: str, state: str, cause: str | None, reason: None, report: bool): ... pass >>> plugin = MyPlugin(1, 2, 3) >>> client.register_plugin(plugin) You can get access to the plugin with the ``get_worker`` function >>> client.register_plugin(other_plugin, name='my-plugin') >>> def f(): ... worker = get_worker() ... plugin = worker.plugins['my-plugin'] ... return plugin.my_state >>> future = client.run(f) See Also -------- distributed.WorkerPlugin unregister_worker_plugin z``Client.register_worker_plugin` has been deprecated; please use `Client.register_plugin` insteadrr;TzRegistering a `WorkerPlugin` as a nanny plugin is not allowed, registering as a worker plugin instead. To register as a nanny plugin, inherit from `NannyPlugin`.FzRegistering a `NannyPlugin` as a worker plugin is not allowed, registering as a nanny plugin instead. To register as a worker plugin, inherit from `WorkerPlugin`.zRegistering a `SchedulerPlugin` as a nanny plugin is not allowed, registering as a scheduler plugin instead. To register as a nanny plugin, inherit from `NannyPlugin`.zRegistering a `SchedulerPlugin` as a worker plugin is not allowed, registering as a scheduler plugin instead. To register as a worker plugin, inherit from `WorkerPlugin`.zRegistering duck-typed plugins has been deprecated. Please make sure your plugin inherits from `NannyPlugin` or `WorkerPlugin`.r) rrr>rZrrYr UserWarningrUrrVrrr)rr~r2rmmethods rrzClient.register_worker_pluginsET   :   <#F+D t fl +11F} Q   ,00F~ S   0 Q   S  44F MM%#  }4455yyTeyLLrc&K|r%|jj|d{}n$|jj|d{}|jD]*}|ddk(s t di|\}}}|j ||S7h7Ew)Nrrrr)runregister_nanny_pluginunregister_worker_pluginrgrQr)rr2rmrsr*rzrrs r_unregister_worker_pluginz Client._unregister_worker_plugin1s "nnDD$DOOI"nnEE4EPPI!((*H!W,,8x8 3((,,+PPs'"BB $B B B*$BBc>|j|j||S)aUnregisters a lifecycle worker plugin This unregisters an existing worker plugin. As part of the unregistration process the plugin's ``teardown`` method will be called. Parameters ---------- name : str Name of the plugin to unregister. See the :meth:`Client.register_plugin` docstring for more information. Examples -------- >>> class MyPlugin(WorkerPlugin): ... def __init__(self, *args, **kwargs): ... pass # the constructor is up to you ... def setup(self, worker: dask.distributed.Worker): ... pass ... def teardown(self, worker: dask.distributed.Worker): ... pass ... def transition(self, key: str, start: str, finish: str, **kwargs): ... pass ... def release_key(self, key: str, state: str, cause: str | None, reason: None, report: bool): ... pass >>> plugin = MyPlugin(1, 2, 3) >>> client.register_plugin(plugin, name='foo') >>> client.unregister_worker_plugin(name='foo') See Also -------- register_plugin )r2rm)rrr)rr2rms rrzClient.unregister_worker_plugin=s Dyy77d%yPPrcddlm}||S)z:Convenience accessors for the :doc:`active_memory_manager`r)AMMClientProxy)!distributed.active_memory_managerr)rrs rammz Client.ammas Ed##rc|\}}tj|}tj|j}|j |yr)logging makeLogRecord getLoggerr2handle)rrkrz record_attrsrecord dest_loggers r_handle_forwarded_log_recordz#Client._handle_forwarded_log_recordhs=<&&|4'' 4 6"rcd|xsd}td|}|j||j|jt ||||S)a Begin forwarding the given logger (by default the root) and all loggers under it from worker tasks to the client process. Whenever the named logger handles a LogRecord on the worker-side, the record will be serialized, sent to the client, and handled by the logger with the same name on the client-side. Note that worker-side loggers will only handle LogRecords if their level is set appropriately, and the client-side logger will only emit the forwarded LogRecord if its own level is likewise set appropriately. For example, if your submitted task logs a DEBUG message to logger "foo", then in order for ``forward_logging()`` to cause that message to be emitted in your client session, you must ensure that the logger "foo" have its level set to DEBUG (or lower) in the worker process *and* in the client process. Parameters ---------- logger_name : str, optional The name of the logger to begin forwarding. The usual rules of the ``logging`` module's hierarchical naming system apply. For example, if ``name`` is ``"foo"``, then not only ``"foo"``, but also ``"foo.bar"``, ``"foo.baz"``, etc. will be forwarded. If ``name`` is ``None``, this indicates the root logger, and so *all* loggers will be forwarded. Note that a logger will only forward a given LogRecord if the logger's level is sufficient for the LogRecord to be handled at all. level : str | int, optional Optionally restrict forwarding to LogRecords of this level or higher, even if the forwarded logger's own level is lower. Examples -------- For purposes of the examples, suppose we configure client-side logging as a user might: with a single StreamHandler attached to the root logger with an output level of INFO and a simple output format:: import logging import distributed import io, yaml TYPICAL_LOGGING_CONFIG = ''' version: 1 handlers: console: class : logging.StreamHandler formatter: default level : INFO formatters: default: format: '%(asctime)s %(levelname)-8s [worker %(worker)s] %(name)-15s %(message)s' datefmt: '%Y-%m-%d %H:%M:%S' root: handlers: - console ''' config = yaml.safe_load(io.StringIO(TYPICAL_LOGGING_CONFIG)) logging.config.dictConfig(config) Now create a client and begin forwarding the root logger from workers back to our local client process. >>> client = distributed.Client() >>> client.forward_logging() # forward the root logger at any handled level Then submit a task that does some error logging on a worker. We see output from the client-side StreamHandler. >>> def do_error(): ... logging.getLogger("user.module").error("Hello error") ... return 42 >>> client.submit(do_error).result() 2022-11-09 03:43:25 ERROR [worker tcp://127.0.0.1:34783] user.module Hello error 42 Note how an attribute ``"worker"`` is also added by dask to the forwarded LogRecord, which our custom formatter uses. This is useful for identifying exactly which worker logged the error. One nuance worth highlighting: even though our client-side root logger is configured with a level of INFO, the worker-side root loggers still have their default level of ERROR because we haven't done any explicit logging configuration on the workers. Therefore worker-side INFO logs will *not* be forwarded because they never even get handled in the first place. >>> def do_info_1(): ... # no output on the client side ... logging.getLogger("user.module").info("Hello info the first time") ... return 84 >>> client.submit(do_info_1).result() 84 It is necessary to set the client-side logger's level to INFO before the info message will be handled and forwarded to the client. In other words, the "effective" level of the client-side forwarded logging is the maximum of each logger's client-side and worker-side levels. >>> def do_info_2(): ... logger = logging.getLogger("user.module") ... logger.setLevel(logging.INFO) ... # now produces output on the client side ... logger.info("Hello info the second time") ... return 84 >>> client.submit(do_info_2).result() 2022-11-09 03:57:39 INFO [worker tcp://127.0.0.1:42815] user.module Hello info the second time 84 forward-logging-r)!TOPIC_PREFIX_FORWARDED_LOG_RECORDrrrrT)r logger_namelevel plugin_namers rforward_loggingzClient.forward_loggingnsa`))@(AB 45Q{mD UD$E$EF ## eU ;[  rcnd|xsd}td|}|j||j|S)zr Stop forwarding the given logger (default root) from worker tasks to the client process. rrr)rrPr)rrrrs runforward_loggingzClient.unforward_loggingsF ))@(AB 45Q{mD u%,,[99r)rr)rBr?rNonerXr)rrr z float | NonerrrW)F)rrrr)rrrrrzstr | list | Nonerrr int | Nonerrrrrrrrrrrrrr)rNN)r!rr"rrzdict[str, Any])NNF)NFNN)rmrrlist[str] | Noner rrn$Literal['raise', 'return', 'ignore'])rrr rrmrrnrr)r<rrrrztuple[SourceCode, ...])NNNrNNrN) NNNTNNNr60sN) FTNFNrrrNT)TNNNNrrN)r str | int | float | NoDefaultrrrr)r rrr)r list[str]r rrrrz0dict[str, Literal['OK', 'removed', 'timed out']])rrr rrr)rrrY)NNr)NT) NNNNTFNFF)r.Nrmsgpack)rrr0 bool | Noner1zCollection[str]rzLiteral['msgpack', 'yaml'])rr)rzstr | Collection[str]rr)rr)rrrVr)FN)r[rr\Sequence[str] | Nonerz0VersionsDict | Coroutine[Any, Any, VersionsDict])r[rr\rrr)rfr)NNNFtask-stream.htmlN)r~,NannyPlugin | SchedulerPlugin | WorkerPluginr2rrzr)r~rr2rrzr)r~rVr2rrzr)r~rUr2rrzrrdict[str, OKMessage])r~rYr2rrzr)r~rVr2rrzr)r~rYr2rrzrrr)r~zNannyPlugin | WorkerPluginr2rrmr)rrrrZr[r\r]rrweakrefWeakSetrrrrrr0DEFAULT_EXTENSIONSrrbr?setterr-rrF classmethodrKrSrTrKrNr.rPr}rrsrprrrrrr rrrrrFrr?rrr$r%r&r'r(r)r*r rrrrrrr0rrr r0rEr=rJr5rMr7rXrZr]r`rdrfrjrlrvrprrcrrrrrrrrrrrrrrr>rrar r/rrrqrrrrr5rFDEFAULT_CLUSTER_DUMP_EXCLUDEDEFAULT_CLUSTER_DUMP_FORMATr2r>r@rCrFrHrJrNr+rrPrWrYrr]rIrr&rlrnrrrqrrr|registerrzrrrrrrrrrrrrrNOTSETrrrrrrrs [z.:#:K:K-LN*L4CGOO4EJ1E(5|L'' FM   %tl  ^^  [[  + +PPB!5!5F.*J8 & 3#1$.?B$ $ L8DtE7;;;'3; ;:M*A  &  $' 66p    2 "$EE$??B'*%X ).. ! Ld"&.2"+/$)$o*o*o* o* , o*  o*)o*o*"o*o*o*o*o*bm^<>F fV t l2 0 J15"*"*H-A^M. @0: )3M6=A &%LV$(9@0 0 " 0  070l%)9@\ " \  \  \ 7\ |67a%a%03a% a%a%F  Uv  eN0'6X![@ cJ  4  HL    22< $ D D /D  D L>2,\ 0F2 2 hK@F%N&PE<V@G<   4 p   -^(6,,0#%-6 ] ] *] ! ] + ] B,,0#/#L#L-9-U-U *!  + 6:6*4R,3 X. <I, =%> K GK! '! ?C! F-M`EISS-AS 9S,EI-A *0  @@      % -0 >B  "& OOO O@K!GF?( 5" 5*- 5;? 5  5 5! 5), 5:> 5  5$ ! IM*IMIM IMV "QH$$ # +/gnnz x:rct|tsJg}|D]G}t|tr|jt|.|jt |It |Sr)rrr{rrDrB)rnew_keysrs rrrsY dD !! !%'H c4 OO.s3 4 OOGCL )  ?rceZdZdZdZdZy)rz:This is used to support older setup functions as callbacksc||_yr)_setuprs rrz_WorkerSetupPlugin.__init__s  rcrt|jdr|j|S|jS)Nr)r)rnr)rrs rrz_WorkerSetupPlugin.setups- t{{M 2;;6;2 2;;= rN)rrrrZrrrrrrrsD!rrctd)Nz7This has been moved to the Client.get_executor() methodrrrs rCompatibleExecutorr s M NNr ALL_COMPLETEDFIRST_COMPLETEDcK|t|ts tdt|}|tk(rFt j j|Dchc]}|jjc}}nZ|tk(rFt j j|Dchc]}|jjc}}n td| t||}|d{|Dchc]}|jdk7s|c}|Dchc]}|jdk(s|c}}}tt }|D]Y}|j#s|jj$} t| t&sJ|| j(j+| [|r8|j-D cgc]\} } t/| | } } } t1| t3||Scc}wcc}w7cc}wcc}wcc} } ww)Nztimeout= keyword received a non-numeric value. Beware that wait expects a list of values Bad: wait(x, y, z) Good: wait([x, y, z])zDOnly return_when='ALL_COMPLETED' and 'FIRST_COMPLETED' are supportedrfr)rrr,rIrrrrrr rrNotImplementedErrorr<rrrrrrrr{rrrr ) fsr  return_whenrrfurnot_donecancelled_errorsrrrgroupss rrrs:gv#> &  BBm#""&&'DA 'DE  '""&&'DA 'DE! R  &'* LL3bBII2b33bBII2b3 D#4( {{} HH&& )%9:::))*11)<  #3"8"8": ": " ?":  $F++ x 00=(E'D 43 saAG<!G2.G< !G$&G<'G)(G<0G,G, G<G1$G1(BG<.G6%G<,G<c|#t|ttfr t|d}t }|j t |||}|S)aWait until all/any futures are finished Parameters ---------- fs : List[Future] timeout : number, string, optional Time after which to raise a ``dask.distributed.TimeoutError``. Can be a string like ``"10 minutes"`` or a number of seconds to wait. return_when : str, optional One of `ALL_COMPLETED` or `FIRST_COMPLETED` Returns ------- Named tuple of completed, not completed rr)r r)rrrr5rHrrr)rr rrrs rr r @sF z'FC=A!'37  F [[G[ MF Mrc Kt|}td|}|jDcgc]}|d }}tjt t j|Dcgc]}|jjc}}|js^|jd{||j}||jD]}|j||js]yycc}wcc}w7Uw)Nc|jSrr)rs rrz_as_completed..Ysquurr)rIr!rgr> WaitIteratorrrirrrr rr current_indexr put_nowait)rqueuerrfirstsr wait_iteratorrs r _as_completedrWs BB _b )F"MMO ,OqadOF ,$$ W " "f$EfQXX]]_f$E FM  "  """ 334 #A   Q $   " -$E #s.+D C:)D"!C?+D.D/AD8 DcKtj}t||d{|jd{}|S77w)zJReturn a single completed future See Also: _as_completed N)rirrr)rqrs r_first_completedrgs>  A  ###557]F M$ s$A AA AA A ceZdZdZ ddddZedZdZdZdZ d Z d Z d Z d Z d ZdZdZdZdZdZeZddZdZdZy) as_completeda` Return futures in the order in which they complete This returns an iterator that yields the input future objects in the order in which they complete. Calling ``next`` on the iterator will block until the next future completes, irrespective of order. Additionally, you can also add more futures to this object during computation with the ``.add`` method Parameters ---------- futures: Collection of futures A list of Future objects to be iterated over in the order in which they complete with_results: bool (False) Whether to wait and include results of futures as well; in this case ``as_completed`` yields a tuple of (future, result) raise_errors: bool (True) Whether we should raise when the result of a future raises an exception; only affects behavior when ``with_results=True``. timeout: int (optional) The returned iterator raises a ``dask.distributed.TimeoutError`` if ``__next__()`` or ``__anext__()`` is called and the result isn't available after timeout seconds from the original call to ``as_completed()``. If timeout is not specified or ``None``, there is no limit to the wait time. Examples -------- >>> x, y, z = client.map(inc, [1, 2, 3]) # doctest: +SKIP >>> for future in as_completed([x, y, z]): # doctest: +SKIP ... print(future.result()) # doctest: +SKIP 3 2 4 Add more futures during computation >>> x, y, z = client.map(inc, [1, 2, 3]) # doctest: +SKIP >>> ac = as_completed([x, y, z]) # doctest: +SKIP >>> for future in ac: # doctest: +SKIP ... print(future.result()) # doctest: +SKIP ... if random.random() < 0.5: # doctest: +SKIP ... ac.add(c.submit(double, future)) # doctest: +SKIP 4 2 8 3 6 12 24 Optionally wait until the result has been gathered as well >>> ac = as_completed([x, y, z], with_results=True) # doctest: +SKIP >>> for future, result in ac: # doctest: +SKIP ... print(result) # doctest: +SKIP 2 4 3 Nrcz|g}tt|_t|_t j |_|xstj|_ t j|_ ||_ ||_ tjt!||_|r|j%|yyr)rrrrrr LocklockrHr- Conditionthread_condition with_results raise_errorsr;afterr5 _deadliner)rrr-rrr s rrzas_completed.__init__s ?G"3' Y NN$ 1N,11 ) 3 3 5((!(@A  KK  rc |jS#t$r(tj|_|jcYSwxYwr) _conditionrErirrs r conditionzas_completed.conditions: #?? " #%//1DO?? " #s .??cTK t|d{|jr |jdd{}|j5||j vr|j |xxdzcc<|j |s |j |=|jr|j j|fn|j j||j4d{|jjdddd{|j5|jjddddddy76#t$rY@wxYw7"#t$r }|}Yd}~1d}~wwxYw77m#1d{7swY}xYw#1swYZxYw#1swYyxYww)NF)rr) rrhrr rrrrrnotifyr)rrrrs r _track_futurezas_completed._track_futuresD -      %~~e~<<YY% V$)$||F+ V,$$JJ))66*:;JJ))&1>>>NN))+*>**))002+Y    =!  *>>>>**YsF(E EE F(EEE F( BF!E7"F%E; F E9 FF7F? F(E EF(EF(E E4'E/)F(/E44F(7F9F;F FF FF FF%!F(cddlm}|j5|D]_}t|t|fst d|z|j |xxdz cc<|jj|j|a dddy#1swYyxYw)zoAdd multiple futures to the collection. The added futures will emit from the iterator once they finishr)BaseActorFuturezInput must be a future, got %srN) distributed.actorrrrrr,rr-r.r)rrrrs rrzas_completed.updatesn 6 YY!!fo%>?#$Dq$HII Q1$ &&t'9'91= YYs A%BB c(|j|fy)ziAdd a future to the collection This future will emit from the iterator once it finishes N)r)rrs rr/zas_completed.adds VIrc$|j S)z7Returns True if there no completed or computing futures)rrs ris_emptyzas_completed.is_emptys::<rc8|jj S)z6Returns True if there are completed futures available.)remptyrs r has_readyzas_completed.has_readys::##%%%rc|j5t|jt|jjzcdddS#1swYyxYw)zReturn the number of futures yet to be returned This includes both the number of futures still computing, as well as those that are finished, but have not yet been returned from this iterator. N)rrrrrs rrzas_completed.counts5YYt||$s4::+;+;'<<YYs 5A  Acdjt|jt|jjS)Nz")rrrrrs rrKzas_completed.__repr__s23::  s4::#3#34  rc|Srrrs rrzas_completed.__iter__ rc|Srrrs r __aiter__zas_completed.__aiter__rrc|jj}|jr7|\}}|jr&|jdk(r|\}}}|j ||Sr)rrrrrr)rresrrrrrs r_get_and_raisezas_completed._get_and_raisesYjjnn    NFF  V]]g%=% S"((,, rcz|jjr|jjr t |j r t |j5|jjdddd|jjr|jS#1swY4xYw)Nrr) rrrexpiredrlr  StopIterationrr rrs r__next__zas_completed.__next__$sjj ~~%%"n$}}#o%&&%%**5*9' jj ""$$'&s !B11B:cK|jjs|jd{St|j|jjd{S7:7wr)rexpires_anextr< remainingrs r __anext__zas_completed.__anext__.sK~~%%& &dkkmT^^-E-EFFF'Fs!*A+A'5A+"A)#A+)A+cK|js |jjrt|jjrt|jst|j4d{|jj d{dddd{|jjrt|j S7a7A73#1d{7swYCxYwwr)rrrStopAsyncIterationrr rrs rrzas_completed._anext3s|| 0 0 2$ $jj <<((~~~nn))+++&~jj ""$$&+&~~~s`A)C(+C ,C(/C CC C(CC(=C(CC(C%C C%!C(c|r t|g}ng}|jjsD|j|jj |jjsD|S)aYGet the next batch of completed futures. Parameters ---------- block : bool, optional If True then wait until we have some result, otherwise return immediately, even with an empty list. Defaults to True. Examples -------- >>> ac = as_completed(futures) # doctest: +SKIP >>> client.gather(ac.next_batch()) # doctest: +SKIP [4, 1, 3] >>> client.gather(ac.next_batch(block=False)) # doctest: +SKIP [] Returns ------- List of futures or (future, result) tuples )rrrr{r)rblockrs r next_batchzas_completed.next_batch@sU, $ZLEE**""$ LL) ***""$ rc#VK |jd#t$rYywxYww)a3 Yield all finished futures at once rather than one-by-one This returns an iterator of lists of futures or lists of (future, result) tuples rather than individual futures or individual (future, result) tuples. It will yield these as soon as possible without waiting. Examples -------- >>> for batch in as_completed(futures).batches(): # doctest: +SKIP ... results = client.gather(batch) ... print(results) [4, 2] [1, 3, 7] [5] [6] T)r%N)r&rrs rrzas_completed.batches^s9& ooDo11!  s)) &)&)c|j5|jj|jj s5|jj |jj s5dddy#1swYyxYw)zClear out all submitted futuresN)rrrurrrrs rruzas_completed.clearwsS YY LL   jj&&(  jj&&(YYs A)BB )NNFTrX)rrrrZrrbrrrr/r rrrKrrrrr!rrr&rrurrrrrss=B  !!.##30 > &= %G % D<2!rrctd)NzThis has moved to as_completedrrs r AsCompletedr*s 4 55rc<|xs t}|r|Std)a!Return a client if one has started Parameters ---------- c : Client The client to return. If None, the default client is returned. Returns ------- c : Client The client, if one has started See also -------- Client.current (alias) zNo clients found Start a client and point it to the scheduler address from distributed import Client client = Client('ip-addr-of-scheduler:8786') )rrrs rrHrHs," !  !A ?  rct|y)zEnsures the client passed as argument is set as the default Parameters ---------- client : Client The client N)rrs rensure_default_clientr-s vrcddlm}t||r||j|St j|}||_|S)aChange the dictionary in the collection Parameters ---------- c : collection The collection dsk : dict The dictionary Returns ------- c : Delayed If the collection is a 'Delayed' object the collection is returned cc : collection If the collection is not a 'Delayed' object a copy of the collection with xthe new dictionary is returned r)Delayed) dask.delayedr/rrcopyr)rrr/ccs rrrs;&%!Wquuc"" YYq\ rc|g}t}t}|r'|j}t|tttfvr|j |nt|t ur |j |jnt|tur*|j |jjnyt|tr'||vre|j||j|nBtj|r-|j |j!j|r'|t#t}|D]Y}|j%s|j&j(}t|t*sJ||j,j|[|r8|j/D cgc]\} } t1| | } } } t3| |dddScc} } w)aFuture objects in a collection Parameters ---------- o : collection A possibly nested collection of Dask objects client : Client, optional The client Examples -------- >>> futures_of(my_dask_dataframe) [, ] Raises ------ CancelledError If one of the futures is cancelled a CancelledError is raised Returns ------- futures : List[Future] A list of futures held by those collections Nr)rnrrrrrirrgr+rrrDr/r{rrrrrrrrrrrr) orstackseenrrrrrrrrs rrIrIs4CE 5DfG  IIK 7uc4( ( LLO !W_ LL $ !W( ( LL ( 7 #} q!  $ $Q ' LL))+224 5 &t,A;;=**Ii)=> >> Y-- . 5 5i @  '7&<&<&>&>NFF&VFC&> (/ / 4R4= sG3c|t|}|D],}|jjd|jgdd.y)anRun tasks at least once, even if we release the futures Under normal operation Dask will not run any tasks for which there is not an active future (this avoids unnecessary work in many situations). However sometimes you want to just fire off a task, not track its future, and expect it to finish eventually. You can use this function on a future or collection of futures to ask Dask to complete the task even if no active client is tracking it. The results will not be kept in memory after the task completes (unless there is an active future) so this is only useful for tasks that depend on side effects. Parameters ---------- obj : Future, list, dict, dask collection The futures that you want to run at least once Examples -------- >>> fire_and_forget(client.submit(func, *args)) # doctest: +SKIP rzfire-and-forgetrN)rIrrr)objrrs rfire_and_forgetr:s=.oG ((+ +  rc0eZdZdZddZdZdZdZdZy) rra Collect task stream within a context block This provides diagnostic information about every task that was run during the time when this block was active. This must be used as a context manager. Parameters ---------- plot: boolean, str If true then also return a Bokeh figure If plot == 'save' then save the figure to a file filename: str (optional) The filename to save to if you set ``plot='save'`` Examples -------- >>> with get_task_stream() as ts: ... x.compute() >>> ts.data [...] Get back a Bokeh figure and optionally save to a file >>> with get_task_stream(plot='save', filename='task-stream.html') as ts: ... x.compute() >>> ts.figure To share this file with others you may wish to upload and serve it online. A common way to do this is to upload the file as a gist, and then serve it on https://raw.githack.com :: $ python -m pip install gist $ gist task-stream.html https://gist.github.com/8a5b3c74b10b413f612bb5e250856ceb You can then navigate to that site, click the "Raw" button to the right of the ``task-stream.html`` file, and then provide that URL to https://raw.githack.com . This process should provide a sharable link that others can use to see your task stream plot. See Also -------- Client.get_task_stream: Function version of this context manager Ncg|_||_||_d|_|xs t |_|j j ddy)Nrr.r)r_plot _filenamer+rHrrr)rrrrs rrzget_task_stream.__init__XsD  ! 0 0  ##!!#4rc$t|_|Sr)r[r.rs rrzget_task_stream.__enter__`sV  rc|jj|j|j|j}|jr |\}|_|j j|yN)r.rrrrrr.r>r?r+rrirrrrrs rrzget_task_stream.__exit__dsT KK ' '**4:: (  ::NAt{ rcK|Swrrrs rrzget_task_stream.__aenter__ls  scK|jj|j|j|jd{}|jr |\}|_|j j|y76wrBrCrDs rrzget_task_stream.__aexit__osb++--**4::.   ::NAt{   sAA;A97A;)NFr) rrrrZrrrrrrrrrrrr's!.`5rrrc4eZdZdZ ddZdZd dZdZdZy) performance_reportaGather performance report This creates a static HTML file that includes many of the same plots of the dashboard for later viewing. The resulting file uses JavaScript, and so must be viewed with a web browser. Locally we recommend using ``python -m http.server`` or hosting the file live online. Parameters ---------- filename: str, optional The filename to save the performance report locally stacklevel: int, optional The code execution frame utilized for populating the Calling Code section of the report. Defaults to `1` which is the frame calling ``performance_report`` mode: str, optional Mode parameter to pass to :func:`bokeh.io.output.output_file`. Defaults to ``None``. storage_options: dict, optional Any additional arguments to :func:`fsspec.open` when writing to a URL. Examples -------- >>> with performance_report(filename="myfile.html", stacklevel=1): ... x.compute() NcR||_|dkDr|nd|_||_|xsi|_y)Nrr)r _stacklevelmoder4)rrr<rKr4s rrzperformance_report.__init__s/! )3a:Q .4"rcKt|_tjdd{|_tj ddd{y7-7w)Nc.|jjSr)monitorr)dask_schedulers rrz/performance_report.__aenter__..s>#9#9#?#?rrr=)r[r.ryrl last_countrrrs rrzperformance_report.__aenter__sSV * = = ?!  l***;;;  )r.rPrrKrcinfer)rK compression)fsspecryrrJrrrHr.rPrKrrr4r) rrrrrrUrrrrs rrzperformance_report.__aexit__s <11$2B2BQ2FPQ1RF%+6!9>>1GD%%88**t$))9  V[[ MM # <@.>.BA-N!'vay~~-C DNNHi Nr)zdask-report.htmlrNNr rrrrZrrrrrrrrrHrHxs(>UY5< +OrrHc.eZdZdZdZdZdZdZdZy)get_task_metadataawCollect task metadata within a context block This gathers ``TaskState`` metadata and final state from the scheduler for tasks which are submitted and finished within the scope of this context manager. Examples -------- >>> with get_task_metadata() as tasks: ... x.compute() >>> tasks.metadata {...} >>> tasks.state {...} cdtjj|_t |_d|_d|_y)Nztask-metadata-)r_r`rar2rnrrrrs rrzget_task_metadata.__init__s4$TZZ\%5%5$67 E   rc~Ktjj|jd{|S7wr)ryrstart_task_metadatar2rs rrzget_task_metadata.__aenter__s4l$$88dii8HHH  Is 2=;=cKtjj|jd{}|d|_|d|_y7w)Nrrr)ryrstop_task_metadatar2rr)rrrrr*s rrzget_task_metadata.__aexit__sF#//BB BRR , g& Ss2AAAcHtj|jSrrWrs rrzget_task_metadata.__enter__s|  11rcNtj|j|||Sr)ryrrr)rrrrs rrzget_task_metadata.__exit__s|  9iPPrNrYrrrr[r[s!  ' 2Qrr[c#Kt}t| |j5ddddt|y#1swYxYw#t|wxYww)aSet the default client for the duration of the context .. note:: This function should be used exclusively for unit testing the default client functionality. In all other cases, please use ``Client.as_current`` instead. .. note:: Unlike ``Client.as_current``, this context manager is neither thread-local nor task-local. Parameters ---------- c : Client This is what default_client() will return within the with-block. N)rHrrF)rold_execs rtemp_default_clientresJ$Hq% \\^  8$^ 8$s1AAAA AA A AAc t}|kd|_ttt5|j r(|j j|jdn|jddddyy#1swYyxYw)z} Force close of global client. This cleans up when a client wasn't close explicitly, e.g. interactive sessions. NFr) r_should_close_looprrlrrr-r.rrs r_close_global_clientrisg A}$ lL 1~~##AGGQ#7" 2 1 1 1s AA99Bc0dt|jiS)Nr)rr)rs rrrsZ )) r)r Client | None)rrkrr)rrrr)rr.rrBr) __future__rriatexitr1rr^rrr*rrr\r rr_rrrrcollections.abcrrrr r concurrent.futuresr concurrent.futures._baser contextlibr rr contextvarsr functoolsrrimportlib.metadatarrnumbersrrrrtypingrrrrrrrrpackaging.versionrrntlzr r!r"r#r$r dask.baser& dask.corer'r(dask.highlevelgraphr) dask.layersr*dask.optimizationr+ dask.tokenizer, dask.typingr-r.r/r0 dask.utilsr1r2r3r4r5r6r7 dask.widgetsr8distributed.corer9rr:distributed.utilsr;r<r0r=rtornador>tornado.ioloopr?dask._task_specr@rArBrCrDrErrFrGrHrdistributed.batchedrIdistributed.cfexecutorrKdistributed.compatibilityrLrMrNrOrPrQrRrSdistributed.diagnostics.pluginrTrUrVrWrXrYrZdistributed.metricsr[distributed.objectsr\r]r^rr_distributed.protocol.pickler`radistributed.publishrbdistributed.pubsubrcdistributed.securityrddistributed.sizeofredistributed.spansrfdistributed.threadpoolexecutorrgrhrirjrkrlrmrnrorprqrrrsdistributed.utils_commrtrurvrwrxdistributed.workerryrzr{typing_extensionsr|rrr$WeakValueDictionaryr}rrr~rrrrrrrrrrrr/rrrrrrrrrrrrrrrrrr rrrr*rHr-rrIr:rrrHr[rerirrrrrrsO"    #OO1:DD"3<"   5<< (+..">>&&70'!QQ02+16%>>-4(4)%*1    >=+   8 $ G!9s-78ISW-X*X #%;!A>A(  N    : %uWup `=`=F 9i9@ 2 &9& #3 >2)2oodyH:_yH:xQ ! !O #!m(1V}.  I!I!X6 :::z DNNbEOEOP$Q$QN%%4 # $%gFJsM""M-,M-