\c@`sfddlmZmZmZddlZddlmZddlmZddl m Z m Z ddl m Z ddlZddlmZddlZddlmZdd lmZddlZddlZddlZdd lmZmZddlZddlZddlZddlZddlZddl Z ddl!Z!ddl"Z"ddl#Z#dd l$m%Z%m&Z&m'Z'dd l(m)Z)m*Z*dd l+m,Z,ddl-m.Z.m/Z/ddl0m1Z1y,ddl2m3Z3m4Z4m5Z5m6Z6m7Z7Wn9e8k r+ddl9m3Z3m4Z4m5Z5m6Z6m7Z7nXyddl:m;Z;Wne8k rYe3Z;nXddl<m=Z=ddl>m?Z?ddl@mAZAmBZBmCZCddlDmEZEddlFmGZGddlHmIZIddlJmKZKmLZLmMZMmNZNmOZOddlPmQZQddlRmGZSmTZTmUZUmVZVmWZWmXZXddlYmZZZm[Z[m\Z\m]Z]m^Z^ddl_m`Z`ddlambZbddlcmdZddd lemfZfmgZgdd!lhmiZidd"ljmkZkdd#llmmZmdd$lnmnZndd%lompZpdd&lqmrZrmsZsmtZtmuZudd'lvmwZwmxZxmyZymzZzm{Z{m|Z|m}Z}m~Z~mZmZmZmZmZmZmZmZmZmZdd(lmZejeZe"jZdgZekgZd)Zd*Zd+Zd,eKfd-YZd.efd/YZe=jd0Zee&jed1Zd2efd3YZd4ebfd5YZd6efd7YZd8Zd9Zd:Ze=jeed;Zeed<Ze=jd=Ze=jd>Zd?efd@YZdAZedBZdCZdDZedEZdFZdGefdHYZe dIZdJZejedS(Ki(tprint_functiontdivisiontabsolute_importN(t defaultdict(tThreadPoolExecutor(tDoneAndNotDoneFuturestCancelledError(tcontextmanager(t timedelta(tpartial(tglob(tNumbertIntegral(ttokenizetnormalize_tokentcollections_to_dsk(tflattentget_dependencies(tSubgraphCallable(tapplytunicode(t ensure_dict(tfirsttgroupbytmergetvalmaptkeymap(t single_key(tgen(t TimeoutError(tEventt Conditiont Semaphore(tIOLoop(tQueuei(t BatchedSend(t WrappedKeytunpack_remotedatat pack_datatscatter_to_workerstgather_from_workers(tClientExecutor(R"tEmptytisqueuet html_escapetStopAsyncIterationtIterator(tconnecttrpctclean_exceptiontCommClosedErrort PooledRPCCall(ttime(tNode(t to_serialize(tdumpstloads(tDatasets(tPubSubClientExtension(tSecurity(tsizeof(trejoin(t dumps_taskt get_clientt get_workertsecede(tAlltsynctfuncnametignoringtqueue_to_iteratorttokeyt log_errorst str_grapht key_splitt format_bytest asciitablet thread_statet no_defaulttPeriodicCallbackt LoopRunnertparse_timedeltat shutting_downtAny(t get_versionscC`sTtttdt}x2|D]*}t|}|jdkrB|St|=qW~dS(Ntreversetclosed(tsortedtlistt_global_clientstTruetstatustNone(tLtktc((s1lib/python2.7/site-packages/distributed/client.pyt_get_global_clientks   cC`s1|dk r-|ttd>> 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 cC`s||_t|_t|}|p*t|_|jj||jj|_||jj krw|jj ||_ nt |_ |jj |<|r|jj idd6t|gd6|jj d6n|dk ry|jj|}Wntk rqX|d|ndS(Nsclient-desires-keystoptkeystclienttkey(RitFalset_clearedRGR`Rht_inc_reft generationt _generationtfuturest_statet FutureStatet_send_to_schedulertidR\t_state_handlersRc(tselfRiRhtinformtstatettkeythandler((s1lib/python2.7/site-packages/distributed/client.pyt__init__s*      cC`s|jS(N(Rh(Ru((s1lib/python2.7/site-packages/distributed/client.pytexecutorscC`s |jjS(N(RpR[(Ru((s1lib/python2.7/site-packages/distributed/client.pyR[scC`s |jjS(s Is the computation complete? (Rptdone(Ru((s1lib/python2.7/site-packages/distributed/client.pyR|scC`s|jjr%|jj|jd|S|jj|jd|dt}|jdkretj|n|jdkr}|n|SdS(s Wait until computation completes, gather result to local process. If *timeout* seconds are elapsed before returning, a ``dask.distributed.TimeoutError`` is raised. tcallback_timeouttraiseitterrort cancelledN(Rht asynchronousRCt_resultRjR[tsixtreraise(Ruttimeouttresult((s1lib/python2.7/site-packages/distributed/client.pyRs ! cc`s|jjV|jdkr`t|jj|jj}|rNtj|qtj |nh|jdkrt |j }|r|qtj |n)|j j |gV}tj |ddS(NRRi(RptwaitR[R1t exceptiont tracebackRRRtReturnRRiRht_gather(RuR~texcRR((s1lib/python2.7/site-packages/distributed/client.pyRs cc`sH|jjV|jdkr5tj|jjntjddS(NR(RpRR[RRRR\(Ru((s1lib/python2.7/site-packages/distributed/client.pyt _exceptionscK`s|jj|jd||S(s Return the exception of a failed task If *timeout* seconds are elapsed before returning, a ``dask.distributed.TimeoutError`` is raised. See Also -------- Future.traceback R}(RhRCR(RuRtkwargs((s1lib/python2.7/site-packages/distributed/client.pyRs c`st}|jdks*|jtjkrxytddd|_Wn tk retd|_nXtj|_nfd}|jj j t |t |jj |dS(s- Call callback on future when callback 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. itthread_name_prefixsDask-Callback-Threadc`s9y|Wn$tk r4tjd|nXdS(NsError in callback %s of %s:(t BaseExceptiontloggerR(tfut(tfn(s1lib/python2.7/site-packages/distributed/client.pytexecute_callback#s N(Ret _cb_executorR\t_cb_executor_pidtostgetpidRt TypeErrorRhtloopt add_callbackt done_callbackR tsubmit(RuRtclsR((Rs1lib/python2.7/site-packages/distributed/client.pytadd_done_callbacks $  cK`s|jj|g|S(sd Cancel request to run this future See Also -------- Client.cancel (Rhtcancel(RuR((s1lib/python2.7/site-packages/distributed/client.pyR-scK`s|jj|g|S(sd Retry this future if it has failed See Also -------- Client.retry (Rhtretry(RuR((s1lib/python2.7/site-packages/distributed/client.pyR6scC`s|jjdkS(s/ Returns True if the future has been cancelled R(RpR[(Ru((s1lib/python2.7/site-packages/distributed/client.pyR?scc`sH|jjV|jdkr5tj|jjntjddS(NR(RpRR[RRRR\(Ru((s1lib/python2.7/site-packages/distributed/client.pyt _tracebackCscK`s|jj|jd||S(s Return 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. 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 [...] See Also -------- Future.exception R}(RhRCR(RuRR((s1lib/python2.7/site-packages/distributed/client.pyRKscC`s |jjS(N(Rpttype(Ru((s1lib/python2.7/site-packages/distributed/client.pyRbscC`sl|j rh|jj|jkrht|_y)|jjj|jjt|j Wqht k rdqhXndS(N( RkRhRmRnRZRRt_dec_refRGRiR(Rut_in_destructor((s1lib/python2.7/site-packages/distributed/client.pytreleasefs  ) cC`s|j|jjjfS(N(RiRht schedulertaddress(Ru((s1lib/python2.7/site-packages/distributed/client.pyt __getstate__pscC`sg|\}}t|}tj||||jidd6id6t|jgd6|jd6dS(Ns update-graphRfttasksRgRh(R?ReRzRrRGRiRs(RuRwRiRR_((s1lib/python2.7/site-packages/distributed/client.pyt __setstate__ss  cC`s&y|jWntk r!nXdS(N(Rt RuntimeError(Ru((s1lib/python2.7/site-packages/distributed/client.pyt__del__s cC`sk|jrSy|jj}Wn tk r;t|j}nXd|j||jfSd|j|jfSdS(Ns's(Rt__name__tAttributeErrortstrR[Ri(Ruttyp((s1lib/python2.7/site-packages/distributed/client.pyt__repr__s   cC`sdtt|j}|di|jd6|jdkrAdndd67}|jry|jj}Wn tk rt|j}nX|d|7}n|d tt|j7}|S( NsFuture: %s sMstatus: %(status)s, R[Rtredtblacktcolors$type: %s, s!key: %s(R,RJRiR[RRRR(RuttextR((s1lib/python2.7/site-packages/distributed/client.pyt _repr_html_s !  cC`s|jjS(N(Rt __await__(Ru((s1lib/python2.7/site-packages/distributed/client.pyRsN(!Rt __module__t__doc__R\RRRZRztpropertyR{R[R|RRt coroutineRRRRRRRRRRRjRRRRRRR(((s1lib/python2.7/site-packages/distributed/client.pyRes4"         RqcB`seZdZdZdZdZdZdd Zd Z d Z d Z d Z dZ ejddZdZRS(seA Future's internal state. This is shared between all Futures with the same key and client. t_eventR[RRRcC`sd|_d|_d|_dS(Ntpending(R\RR[R(Ru((s1lib/python2.7/site-packages/distributed/client.pyRzs  cC`s,|j}|dkr(t}|_n|S(N(RR\R(Rutevent((s1lib/python2.7/site-packages/distributed/client.pyt _get_events  cC`sd|_|jjdS(NR(R[Rtset(Ru((s1lib/python2.7/site-packages/distributed/client.pyRs cC`s5d|_|jj|dk r1||_ndS(Ntfinished(R[RRR\R(RuR((s1lib/python2.7/site-packages/distributed/client.pytfinishs  cC`sd|_|jjdS(Ntlost(R[Rtclear(Ru((s1lib/python2.7/site-packages/distributed/client.pytloses cC`sd|_|jjdS(NR(R[RR(Ru((s1lib/python2.7/site-packages/distributed/client.pyRs cC`sGt||\}}}d|_||_||_|jjdS(NR(R1R[RRRR(RuRRt_((s1lib/python2.7/site-packages/distributed/client.pyt set_errors    cC`s|jdk o|jjS(N(RR\tis_set(Ru((s1lib/python2.7/site-packages/distributed/client.pyR|scC`s,d|_|jdk r(|jjndS(NR(R[RR\R(Ru((s1lib/python2.7/site-packages/distributed/client.pytresets cc`s|jj|VdS(N(RR(RuR((s1lib/python2.7/site-packages/distributed/client.pyRscC`sd|jj|jfS(Ns<%s: %s>(t __class__RR[(Ru((s1lib/python2.7/site-packages/distributed/client.pyRs(RR[RRRN(RRRt __slots__RzRRR\RRRRR|RRRRR(((s1lib/python2.7/site-packages/distributed/client.pyRqs        cc`s2x!|jdkr#|jjVqW||dS(s5 Coroutine that waits on future, then calls callback RN(R[RpR(tfuturetcallback((s1lib/python2.7/site-packages/distributed/client.pyRscC`s|jt|gS(N(RiR(tf((s1lib/python2.7/site-packages/distributed/client.pytnormalize_futurestAllExitcB`seZdZRS(s3Custom exception class to exit All(...) early. (RRR(((s1lib/python2.7/site-packages/distributed/client.pyRstClientcB`seZdZdrdreedrdredrdrdrdredrd Ze dZ e dZ dZ dZdZdZdZd Zd Zejed Zejd Zejdrd ZejdZdZdZejdZejdZdZdZdZ dZ!dZ"ejdZ#drdrdrdZ$drdZ%drdZ&drdZ'drdrdrdZ(dZ)drdZ*ejed Z+e+Z,ed!Z-d"Z.d#Z/d$Z0d%Z1d&Z2ejd'drdrd(Z3ejd)Z4d*Z5d'd+drdrd,Z6ejdredrdreed-Z7d.Z8dredred+edrd/Z9ejed0Z:dred1Z;ejd2Z<drd3Z=ejd4Z>d5Z?d6Z@d7ZAejd8ZBd9ZCejd:ZDd;ZEejd<ZFd=ZGd>ZHdrdrdrd+drdrd+drd?ZIdrdrdredrdrdrd+d@drdA ZJdBZKdCZLeedredrd+d+d@drdD ZMedrdrdrdrd+d@drdEZNejdFZOdGZPejedHZQdIZRejedJZSejdrdKZTdLZUejdrdrdMZVdrdrdNZWejdrdrdOdPZXdrdrdOdQZYdrdRZZdrdSZ[drdTZ\drdUZ]dredVZ^drdrdWZ_drdrdrdreedrdXZ`ejdrdrdrdreedrdYZadZZbd[Zced\Zddrd]Zedrdrd^Zfdred_Zgd`ZhegdaZidbZjdcZkejddZldreedrdeZmdfedrdgZne dhZoe diZpdjZqe dkZresdlZtdrdrdredmdnZuejdrdrdredmdoZvejdrdpZwdrdqZxRS(sss Connect to and drive computation on a distributed Dask cluster The Client connects users to a dask.distributed compute 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. 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()`` timeout: int Timeout duration for initial connection to the scheduler set_as_default: bool (True) Claim this scheduler as the global dask scheduler scheduler_file: string (optional) Path to a file with scheduler information if available security: (optional) Optional security information 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 direct_to_workers: bool (optional) Whether or not to connect directly to the workers, or to ask the scheduler to serve as intermediary. heartbeat_interval: int Time in milliseconds between heartbeats to scheduler **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(processes=False, threads_per_worker=1) # doctest: +SKIP See Also -------- distributed.scheduler.Scheduler: Internal scheduler distributed.deploy.local.LocalCluster: c K`s|tkr!tjjd}n|dk r?t|d}n||_t|_t d|_ g|_ |dkrtjjdd}nt |j |rd|dndttjdtj|_d|_d|_g|_i|_||_||_d|_d|_i|_tj|_t ||_!| |_"| dkrf| } n| |_#| |_$t%d |_&d|_'d|_(|pt)|_*d|_+t,|j*t)st-|d kr|j*j.d |_/n|j*j.d |_/|dkrBtjjd d}|rBt0j1d |qBnt,|t2t3frc||_n7t4|dr||_t5t6|j7}WdQXnt8|_9||_:| |_;t<d|d||_=|j=j7|_7| dkrtjjd} nt| dd} t|_>t?|j@dd|j7|j>dd<||_B|rtjjCdddd|_Dni|jEd6|jFd6|jGd6|jHd 6|jId!6|jJd"6|jKd#6|_Li|jEd$6|jFd%6|jId&6|_MtNtO|jPd'|j/d|j7d(| d)| d*|x| D]}||qHW|jQd*|dd+lRmS}||dS(,Ns!distributed.comm.timeouts.connecttscS`sdS(Ni((((s1lib/python2.7/site-packages/distributed/client.pyt\ts client-namet-t clock_seqis newly-createditworkerRhsscheduler-addresss*Config value `scheduler-address` found: %stscheduler_addressRRsdistributed.client.heartbeattdefaulttmsitio_loopsscheduler-infoit heartbeatRsdask.distributedtshuffleRs key-in-memorys lost-datas cancelled-keys task-retrieds task-erredtrestartRtmemoryRterredtconnection_argst serializerst deserializersR(tReplayExceptionClient(TRNtdasktconfigtgetR\RQt_timeouttdictRoRtrefcountt coroutinesRRRtuuidtuuid1RRRsRmR[t_pending_msg_buffert extensionstscheduler_filet_startup_kwargstclusterRt_scheduler_identityt threadingtRLockt_refcount_lockR9tdatasetst _serializerst_deserializerstdirect_to_workersR t_gather_semaphoret _gather_keyst_gather_futureR;tsecuritytscheduler_commt isinstancetAssertionErrortget_connection_argsRRtinfoR0R3thasattrRERRRjt_connecting_to_schedulert _asynchronoust_should_close_loopRPt _loop_runnert_periodic_callbacksROt_update_scheduler_infot _heartbeatt _start_argRt _set_configt_handle_key_in_memoryt_handle_lost_datat_handle_cancelled_keyt_handle_retried_keyt_handle_task_erredt_handle_restartt _handle_errort_stream_handlersRttsuperRRztstarttdistributed.recreate_exceptionsR(RuRRRtset_as_defaultRRRtnametheartbeat_intervalRRRRRtextR((s1lib/python2.7/site-packages/distributed/client.pyRzDs      F                           #             cC`stS(s@ Return global client if one exists, otherwise raise ValueError (tdefault_client(R((s1lib/python2.7/site-packages/distributed/client.pytcurrentscC`s|jo|jtjkS(sN Are we running in the event loop? This is true if the user signaled that we might be when creating the client as in the following:: client = Client(asynchronous=True) However, we override this expectation if we can definitively tell that we are running from a thread that is not the event loop. This is common when calling get_client() from within a worker task. Even though the client was originally created in asynchronous mode we may find ourselves in contexts when it is better to operate synchronously. (RRR!R(Ru((s1lib/python2.7/site-packages/distributed/client.pyRscO`s|jdd}|s3|js3ttdtr|jdd}|||}|dk r~tjtd||}n|St |j |||SdS(NRR}tseconds( tpopR\RtgetattrRMRjRt with_timeoutRRCR(RutfunctargsRRR}R((s1lib/python2.7/site-packages/distributed/client.pyRCs  cC`s|j}|jd}|rr|jdi}t|}td|jD}d|jj|||fS|jdk rd|jj|jj fSd|jjfSdS(NRtworkerscs`s|]}|dVqdS(tncoresN((t.0tw((s1lib/python2.7/site-packages/distributed/client.pys ss(<%s: scheduler=%r processes=%d cores=%d>s<%s: scheduler=%r>s<%s: not connected>( RRtlentsumtvaluesRRRR\R(RuRtaddrR&tnworkersR'((s1lib/python2.7/site-packages/distributed/client.pyRs     cC`s|jrHt|jdrH|jjrH|jjj}|jj}nj|jjr|jr|jo{|jtj k rt |j|jj}|j}nt }|j}|dk rd|j }nd}|rd|dkr|j jd\}}|dd}|dkr%d}n|jd d }tjjd }|jd |d |tj} |di| d67}n|d7}|r t|d} td|djD} td|djD} t| } d| | | f} d|| fS|SdS(NRs0

Client

  • Scheduler: %s s;

    Client

    • Scheduler: not connected tbokehtservicess://tinproct localhostt:isdistributed.dashboard.linkthosttportsF
    • Dashboard: %(web)s twebs
    R&cs`s|]}|dVqdS(R'N((R(R)((s1lib/python2.7/site-packages/distributed/client.pys 6scs`s|]}|dVqdS(t memory_limitN((R(R)((s1lib/python2.7/site-packages/distributed/client.pys 7sss

    Cluster

    • Workers: %d
    • Cores: %d
    • Memory: %s
    s
    %s %s
    (RRRtidentityR t is_startedRRR!RRCRjR\RtsplitRRRtformatRtenvironR*R+R,RK(RuRRRtprotocoltrestR5R4ttemplateRR&tcoresRttext2((s1lib/python2.7/site-packages/distributed/client.pyR sL              cK`sk|jdkrdS|jjt|d|_|jrQ|j||_nt|j|j|dS(s, Start scheduler running in separate thread s newly-createdNt connecting( R[R RRbRt_startt_startedRCR(RuR((s1lib/python2.7/site-packages/distributed/client.pyRMs    c`sEtdrjjStjfd}|jSdS(NRDc`stjdS(N(RR((Ru(s1lib/python2.7/site-packages/distributed/client.pyRas(RRDRRR(RuR((Rus1lib/python2.7/site-packages/distributed/client.pyR\s cC`s{|jdkrUy|jj|Wqwttfk rQ|jdkrRqRqwXn"|jdkrw|jj|ndS(NtrunningtclosingRBs newly-created(RERF(RBs newly-created(R[RtsendR2RRtappend(Rutmsg((s1lib/python2.7/site-packages/distributed/client.pyt_send_to_scheduler_safegs cC`sE|jdkr(|jj|j|ntd|j|fdS(NRERFRBs newly-createds<Tried sending message after closing. Status: %s Message: %s(RERFRBs newly-created(R[RRRJt Exception(RuRI((s1lib/python2.7/site-packages/distributed/client.pyRrqs c k`s|tkr|j}n|dk r6t|d}n|j}|jdk ry|jjVWn4tk rsn$tk rt j ddt nX|jj }n|j dk rWx'tjj|j stjdVqWxtdD]d}y6t|j }tj|}WdQX|d}PWqttfk rOtjdVqXqWn |jdkraddlm}y-|d |jd t |j|_|jVWncttjfk r}|j t j!krn|d d d |jd t |j|_|jVnXxF|jj" s@t#|jj$j"t#|jj"krQtjdVq W|jj }n|j$dkr|j%||_$nd|_&|j'd |Vx!|j(j)D]} | j*qW|j+|_,|j-j.|j,tj/|dS(NRs9Tried to start cluster and received an error. Proceeding.texc_infog{Gz?i Ri(t LocalClusterRRtscheduler_portiR(0RNRR\RQR RRCRRKRRRZRRRtpathtexistsRtsleeptrangetopentjsontloadt ValueErrorRctdeployRMRRtOSErrortsocketRterrnot EADDRINUSER&R*RR0Rt_ensure_connectedR R,Rt_handle_reportt_handle_scheduler_coroutineRRHR( RuRRRtiRtcfgRMtetpc((s1lib/python2.7/site-packages/distributed/client.pyRCzsl           % c c`st|jjjs"td|_d|_x!|jjD]}|j qDW|jj |j }|j j |}x|dkr|jdkry|jd|VPWqtk rtjdV||j j }qXqWtjd|j |jVWdQXdS(NRBiRg?sCFailed to reconnect to scheduler after %.2f seconds, closing client(RHRtcommRVRR[R\RoR,RRRRR4R\tEnvironmentErrorRRQRRt_close(RutstRtdeadline((s1lib/python2.7/site-packages/distributed/client.pyt _reconnects(       cc`s|jr|jj s1|js1|jdkr5dSt|_zt|jjd|d|jV}d|_ |dk rt j t d||j Vn |j V|jidd6|jd6td6VWdt|_X|dk r t j t d||jV}n |jV}t|d ks0t|d dd ksJttd d d|j}|j|||_t|d|_x|jD]}|j|qW|j2tjddS(NRRsClient->SchedulerR sregister-clientRfRhtreplyiis stream-starttintervalt10msRREs+Started scheduling coroutines. Synchronized(RRVRRR\RZR/RRRRR#RR twriteRsRjtreadR*RR#RRRbR[RRrRtdebug(RuRRcRItbcomm((s1lib/python2.7/site-packages/distributed/client.pyR\sB        $  %     cc`sO|jdkrdSy|jjV|_Wntk rJtjdnXdS(NRERBs(Not able to query scheduler for identity(RERB(R[RR8RRdRRn(Ru((s1lib/python2.7/site-packages/distributed/client.pyR s  cC`s'|jr#|jjidd6ndS(Nsheartbeat-clientRf(RRG(Ru((s1lib/python2.7/site-packages/distributed/client.pyR s cC`s |jjs|jn|S(N(R R9R(Ru((s1lib/python2.7/site-packages/distributed/client.pyt __enter__!s cc`s|jVtj|dS(N(RDRR(Ru((s1lib/python2.7/site-packages/distributed/client.pyt __aenter__&scc`s|jVdS(N(Re(RuRtvalueR((s1lib/python2.7/site-packages/distributed/client.pyt __aexit__+scC`s|jdS(N(tclose(RuRRrR((s1lib/python2.7/site-packages/distributed/client.pyt__exit__/scC`s|jdS(N(Rt(Ru((s1lib/python2.7/site-packages/distributed/client.pyR2scC`s'|j|j|cd7|jn|jdkrx|jidd6|gd6|jd6ndS(s% Release key from distributed memory sRelease key %sRVsclient-releases-keysRfRgRhN( RRnRoR!R\RR[RrRs(RuRiRf((s1lib/python2.7/site-packages/distributed/client.pyRv@s  c c`styxtr|jd kr)Pny|jjjV}WnUtk r|jdkrtj dtj dd|_|j VqqPnXt |t t fs|f}nt}x|D]}tjd|d|krd|dkrtjt|n|jd}|d ks5|d kr?t}Pny|j|}||Wqtk r|}tj|qXqW|rPqqWWntk rnXWd QXd S( s Listen to scheduler REs(Client report stream closed to schedulersReconnecting...RBsClient receives message %sR[RRfRts stream-closedN(RHRZRR\RcRmR2R[RRRhRRXttupleRjRnRRR1R!RRKRR(RutmsgstbreakoutRIRfRyRa((s1lib/python2.7/site-packages/distributed/client.pyR]KsF            cC`su|jj|}|dk rq|r[|j r[yt|}Wqatk rWd}qaXnd}|j|ndS(N(RoRR\RR8RKR(RuRiRR&Rw((s1lib/python2.7/site-packages/distributed/client.pyRxs   cC`s/|jj|}|dk r+|jndS(N(RoRR\R(RuRiRw((s1lib/python2.7/site-packages/distributed/client.pyRs cC`s/|jj|}|dk r+|jndS(N(RoRR\R(RuRiRw((s1lib/python2.7/site-packages/distributed/client.pyRs cC`s/|jj|}|dk r+|jndS(N(RoRR\R(RuRiRw((s1lib/python2.7/site-packages/distributed/client.pyRs cC`s5|jj|}|dk r1|j||ndS(N(RoRR\R(RuRiRRRw((s1lib/python2.7/site-packages/distributed/client.pyRs cC`sbtjdx!|jjD]}|jqW|jjtt|jj WdQXdS(Ns%Receive restart signal from scheduler( RRRoR,RRRERt_restart_eventR(RuRw((s1lib/python2.7/site-packages/distributed/client.pyRs    cC`stjdtj|dS(NsScheduler exception:(RtwarningR(RuR((s1lib/python2.7/site-packages/distributed/client.pyRs cc`sd|_tt|x!|jjD]}|jq-Wi|_tt|j WdQXWdQX|j t j j ddkrt j j d=n|jdkrtjn|jr |jjr |jjj r |jidd6|jidd6ntttj+tjtdd |jd tfVWdQX|jr|jjr|jjj r|jjVnx't|jD]}|jd |qW|jdkrtt|jjVWdQXn|j jd|_t!|krt"dnt#|j$}xJ|j$D]?}tt%|j&WdQX|j'r,|j(|q,q,W|j$2|stt%tjtd d t|VWdQXntt|j)j*WdQXd|_)WdQXd|_dS(s6 Send close signal and wait until scheduler completes RFNRRVs close-clientRfs close-streamt millisecondsidtquiet_exceptionsRiR i(+R[RHRdR R,tstopRRERRRRRR\RRRRcRVRrRR#RR^RRtRXRoRvR RReR0R`RbRRRRRtremoveRt close_rpc(RutfastRbRiRR((s1lib/python2.7/site-packages/distributed/client.pyResd                    ) cC`s|tkr|jd}n|jdkr/dSd|_|jru|j}|rqtjtd||}n|S|jdkrt t |j j WdQXnt|j|jdt|jdkst|jrt r|jjndS(s> Close 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. See Also -------- Client.restart iRVNRFR R(RNRR[RReRR#RR R\RERRRtRCRRZRRRRR R~(RuRR((s1lib/python2.7/site-packages/distributed/client.pyRts"     cO`stjd|j||S(s Deprecated, see close instead This was deprecated because "shutdown" was sometimes confusingly thought to refer to the cluster rather than the client s1Shutdown is deprecated. Please use close instead(twarningstwarnRt(RuR%R((s1lib/python2.7/site-packages/distributed/client.pytshutdowns cK`s t||S(sx 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 ------- An Executor object that's fully compatible with the concurrent.futures API. (R)(RuR((s1lib/python2.7/site-packages/distributed/client.pyt get_executorscO`st|stdn|jdd}|jdd}|jdd}|jdd}|jdd}|jdd } |jd t} |jd |jd t} |jd | } | ttdfkrtdn|dkrF| r#t|dt|||}qFt|dtt j }nt |} |j '| |j kr~t||dtSWdQX| r|dkrtdnt|tjtfr|g}n|dk ri|| 6}| r| gng}n i}g}|r/it|t||f| 6}ni|ft|| 6}|j|| g||did| 6d|d|ri|| 6ndd|d| d | }tjdt|||| S(sz Submit a function application to the scheduler Parameters ---------- func: callable *args: **kwargs: pure: bool (defaults to True) Whether or not the function is pure. Set ``pure=False`` for impure functions like ``np.random.random``. workers: set, iterable of sets A set of worker hostnames on which computations may be performed. Leave empty to default to all workers (common case) key: str Unique identifier for the task. Defaults to function-name and hash 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 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 Examples -------- >>> c = client.submit(add, a, b) # doctest: +SKIP Returns ------- Future See Also -------- Client.map: Submit on many arguments at once s1First input to submit must be a callable functionRiR&t resourcestretriestpriorityit fifo_timeoutt100mstallow_other_workerstactortactorstpures*allow_other_workers= must be True or FalseRRvNs/Only use allow_other_workers= if using workers=t user_prioritysSubmit %s(...), %s(tcallableRR!R\RjRZRDR RRtuuid4RGRRoReRVRRt string_typesR RRXRwt_graph_to_futuresRRn(RuR$R%RRiR&RRRRRRRtskeyt restrictionstloose_restrictionstdskRo((s1lib/python2.7/site-packages/distributed/client.pyR+s\'  ##     "  c K`st|drtj}n%t|dtr8t}n txstry#g|D]}||^qT}Wn!tk r}|j |PnX|j |||} |j | qDWdS(s% Internal function for mapping Queue iN( R+tpyQueueRRR.tnexttNotImplementedErrorRZt StopIterationtputR( Rutq_outR$tqs_inRRtqR%RaR((s1lib/python2.7/site-packages/distributed/client.pyt _threaded_maps    # c`s7tstdnttt|sFtd|Dr|jdd}td|}tjd|j ddd||fd |}t |_ |j t|dr|St |Sn|jd d!}|pt}|jd d!|jd d!}|jd d!|jdd} |jdt} |jdd} |jd|jdt} |jd| } | rd!krtdnttt|}t|tr|}n| rgt|D] }|dt||^q}nbttj}|rugtttt|D](}|d|dt|^qMng}|sfdt|t|D}nii}xe|jD]W\}}t|dkrtj|}|j |<|j!|jq||>> L = client.map(func, sequence) # doctest: +SKIP Returns ------- List, iterator, or Queue of futures, depending on the type of the inputs. See also -------- Client.submit: Submit a single function s.First input to map must be a callable functioncs`s|]}t|tVqdS(N(RR.(R(R_((s1lib/python2.7/site-packages/distributed/client.pys stmaxsizeittargetRsThreaded map()R%RRiR&RRRRRRRRRs/Only use allow_other_workers= if using workers=Rc`s&i|]\}}f||qS(((R(RiR%(R$(s1lib/python2.7/site-packages/distributed/client.pys s gj@c`s7i|]-\}}ttt|ff|qS((RRwRX(R(RiR%(R$tkwargs2(s1lib/python2.7/site-packages/distributed/client.pys s sDYou only provided %d worker restrictions for a sequence of length %dc`si|]}|qS(((R(R^(R&(s1lib/python2.7/site-packages/distributed/client.pys s s0Workers must be a list or set of workers or Nones*allow_other_workers= must be True or Falsec`si|]}|qS(((R(R^(R(s1lib/python2.7/site-packages/distributed/client.pys /s Rs map(%s, ...)N(,RRtalltmapR+R!RRtThreadRRZtdaemonRRFR\RDRjRVRXtzipRR RRRRRtminR*titemsR<Rtdelayedt_keytupdateRRR RRRRRRnRG(RuR$t iterablesRRRttRiRRRRRRRgR%tuidR_RR^tvtvvRRRRo((R$RRR&s1lib/python2.7/site-packages/distributed/client.pyRs)        3J(  !     ! traisec #`st|dt\}}g|D]}t|j^q}t} i} |dkraj}n|dkry t} Wntk rt }qX| j j j j krt}qnt j fd} x&trtjdtt>tg|D]!} | jkr| | ^qdtVWdQXd}t}t}x|D]} | jksj| j|krT|j| dkr y#j| }|j}|j}Wn0ttfk rtjtt| dq Xtjt|||ndkr0|j| d| | s R[s(Couldn't gather %s keys, rescheduling %sRgs report-keyRfRiR(RR(-R%RZRGRiRR\RR@RKRjRRRRRRnRERRBRoRR[taddRRRRcRRRRRVRRRt_gather_remoteR{R*RrRRRXR&RR(RuRoRtdirectRtunpackedt future_setRRgtbad_dataRR)RRitfailedt exceptionstbad_keysRfRRR^tresponsetlogRR((RRRus1lib/python2.7/site-packages/distributed/client.pyRCs"           .   %        1(        !(c c`s3|jjVt|j}d|_d|_z|s>|r|jjd|V}t|d|j dt V\}}}idd6|d6}|rg|D]} | |kr| ^q} |jj d| V}|ddkr|dj |qqn|jj d|V}Wd|jj Xtj|dS(s Perform 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. RgR0RttOKR[RN(RtacquireRXRR\RRtwho_hasR(R0RjtgatherRRRR( RuRRRgRtdata2t missing_keystmissing_workersRRitkeys2((s1lib/python2.7/site-packages/distributed/client.pyRs$   "%cK`sxtr|jg}x<|jrVy|j|jWqtk rRPqXqW|j||}x|D]}|j|qpWqWdS(s' Internal function for gathering Queue N(RZRtemptyRHt get_nowaitR*RR(RutqintqoutRR]tresultstitem((s1lib/python2.7/site-packages/distributed/client.pyt_threaded_gathers    ic `st|rmtd|}tjdjddd||fdid6d6}t|_|j|St|t rfd |DSt t d rt j d }nd}jj|ddd |d |SdS(sP Gather 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, iterators, queues 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. maxsize: int If the input is a queue then this produces an output queue with a maximum size. 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] >>> seq = c.gather(iter([x, x])) # support iterators # doctest: +SKIP >>> next(seq) # doctest: +SKIP 3 See Also -------- Client.scatter: Send data out to cluster RRRsThreaded gather()R%RRRc3`s*|] }j|ddVqdS(RRN(R(R(R(RRRu(s1lib/python2.7/site-packages/distributed/client.pys stexecution_stateRRRN(R+RRRRRZRRRR.RRMRR\RCR( RuRoRRRRRRR((RRRus1lib/python2.7/site-packages/distributed/client.pyRs,/      c #`s|tkrj}nt|tjtfr=|g}nt|trtd|D rjt t |||Vt j fd|Dnt|t tdrt|}nt |}t} t} t|trt|}nt|ttfr(t|}nt|tttttfsXt} |g}nt|ttfr|rg|D]#} t | jdt| ^qz} n3g|D]&} t | jdtjj^q} tt| |}nt|tsttt |} |dkr-j}n|dkry t } Wnt!k r_t}qX| j"j#j"j#krt}qnrj$d|dtj"j$dfd|Dd tt%|d j&Vntt'|}|rd}t(}xb|sg|dk r)t j)d Vnt(||krNt j*d nj"j+d |V}qW|s}t,dnt-||dtdj.V\}}}j"j$d|d |d j&Vn/j"j/d|d |d j&d|d|Vfd|Dx1| j0D]#\}}j1|j2d|qW|r|r|tkrbdn|}j3tj4d |d|Vnt5|ttttfr|fd| Dn| rt6dksttj4dnt j dS(Ncs`s$|]}t|ttfVqdS(N(RtbytesR(R(R^((s1lib/python2.7/site-packages/distributed/client.pys 1sc`s#i|]}t||qS((RG(R(R^(td(s1lib/python2.7/site-packages/distributed/client.pys 4s iRRtreportRc`si|]}jg|qS((R(R(Ri(R(s1lib/python2.7/site-packages/distributed/client.pys \s tnbytesRhg?sNo valid workers foundR&sNo valid workersR0t broadcastRc`s(i|]}t|dt|qS(Rv(ReRj(R(R^(Ru(s1lib/python2.7/site-packages/distributed/client.pys s Rtnc3`s|]}|VqdS(N((R(R^(tout(s1lib/python2.7/site-packages/distributed/client.pys si(7RNRRRRR RRt_scatterRRGRRRRRRXRjR.Rt frozensetRwRZRR RRthexRRRR\RR@RKRRt update_dataR<RsR6R4RQRR'RVR'R0tscatterRRoRt _replicateR,t issubclassR*(RuRR&RRRRthasht input_typetnamestunpacktxttypesR)RR'RRRRRiRR((RRRRus1lib/python2.7/site-packages/distributed/client.pyR!s   "  33           %    )cK`sxtrt|rf|jg}xv|jsby|j|jWq'tk r^Pq'Xq'Wn7yt|g}Wn!tk r}|j |PnX|j ||}x|D]}|j |qWqWdS(s6 Internal function for scattering Iterable/Queue data N( RZR+RRRHRR*RRRR(Rutq_or_iRRR]RaRoR((s1lib/python2.7/site-packages/distributed/client.pyt_threaded_scatters       c C`s |tkr|j}nt|s3t|trtjdtd|} tj d|j ddd|| fdi|d6|d 6} t | _ | j t|r| St| Snbttd rtjd } nd} |j|j|d|d |d |d | d|d|d|SdS(s Scatter 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, iterator, dict, Queue, 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. 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. maxsize: int (optional) Maximum size of queue if using queues, 0 implies infinite hash: bool (optional) Whether or not to hash data to determine key. If False then this uses a random key 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 Handle streaming sequences of data with iterators or queues >>> seq = c.scatter(iter([1, 2, 3])) # doctest: +SKIP >>> next(seq) # 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)] See Also -------- Client.gather: Gather data back to local process s"Starting thread for streaming dataRRRsThreaded scatter()R%RR&RRRRRRRRN(RNRR+RR.RRnRRRRRZRRRFRRMRR\RCR( RuRR&RRRRRRRRR((s1lib/python2.7/site-packages/distributed/client.pyRs8R          cc`stdt|D}|jjd|d|jd|Vx<|D]4}|jj|d}|dk rF|jqFqFWdS(NcS`sh|]}t|jqS((RGRi(R(R((s1lib/python2.7/site-packages/distributed/client.pys s RgRhtforce(RXt futures_ofRRRsRoR!R\(RuRoRRgR^Rf((s1lib/python2.7/site-packages/distributed/client.pyt_cancels #  cC`s|j|j|d|d|S(s 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 of Futures force: boolean (False) Cancel this future even if other clients desire it RR(RCR(RuRoRR((s1lib/python2.7/site-packages/distributed/client.pyR#scc`sgtdt|D}|jjd|d|jV}x%|D]}|j|}|jqBWdS(NcS`sh|]}t|jqS((RGRi(R(R((s1lib/python2.7/site-packages/distributed/client.pys 5s RgRh(RXRRRRsRo(RuRoRgRRiRf((s1lib/python2.7/site-packages/distributed/client.pyt_retry3s   cC`s|j|j|d|S(sn Retry failed futures Parameters ---------- futures: list of Futures R(RCR(RuRoR((s1lib/python2.7/site-packages/distributed/client.pyR;sc/`stgfd}|jdd}|rt|dkr[tdnt|dkrz|d}n|||nx'|jD]\}}|||qWVWdQXdS(Nc `scgt|D]}t|j^q }jjjd|d|dt|djdS(NRgRRRh(RRGRiRHRt publish_putR6Rs(RRRRg(RRu(s1lib/python2.7/site-packages/distributed/client.pytadd_coroJs( RisQIf name is provided, expecting call signature like publish_dataset(df, name='ds')i(RHR!R\R*RVR(RuR%RRRR((RRus1lib/python2.7/site-packages/distributed/client.pyt_publish_datasetEs   cO`s|j|j||S(s 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 name : optional name of the dataset to publish 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 (RCR(RuR%R((s1lib/python2.7/site-packages/distributed/client.pytpublish_datasetcs.cK`s|j|jjd||S(s^ Remove named datasets from scheduler Examples -------- >>> c.list_datasets() # doctest: +SKIP ['my_dataset'] >>> c.unpublish_datasets('my_dataset') # doctest: +SKIP >>> c.list_datasets() # doctest: +SKIP [] See Also -------- Client.publish_dataset R(RCRtpublish_delete(RuRR((s1lib/python2.7/site-packages/distributed/client.pytunpublish_datasetscK`s|j|jj|S(s List named datasets available on the scheduler See Also -------- Client.publish_dataset Client.get_dataset (RCRt publish_list(RuR((s1lib/python2.7/site-packages/distributed/client.pyt list_datasetss cc`sn|jjd|d|jV}|dkr>td|nt||d}WdQXtj|dS(NRRhsDataset '%s' not foundR(Rt publish_getRsR\Rcttemp_default_clientRR(RuRRR((s1lib/python2.7/site-packages/distributed/client.pyt _get_datasets   cK`s|j|j||S(s Get named dataset from the scheduler See Also -------- Client.publish_dataset Client.list_datasets (RCR(RuRR((s1lib/python2.7/site-packages/distributed/client.pyt get_datasets c o`s|jdt}|jjdt|dt|dt|d|V}|ddkrrtjt|ntj |ddS(NRtfunctionR%RR[RR( R!RZRt run_functionR7RRR1RR(RuRR%RRR((s1lib/python2.7/site-packages/distributed/client.pyt_run_on_schedulers  .cO`s|j|j|||S(s Run 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. 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 gen.sleep(1) >>> c.run(print_state, wait=False) # doctest: +SKIP See Also -------- Client.run: Run a function on all workers Client.start_ipython_scheduler: Start an IPython session on scheduler (RCR(RuRR%R((s1lib/python2.7/site-packages/distributed/client.pytrun_on_schedulersc o`s|jdt}|jdd}|jdt}|jjdtdddt|dt|d|d t|d|d|V}i}xa|jD]S\} } | d d kr| d || >> 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_hostname) # 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 gen.sleep(1) >>> c.run(print_state, wait=False) # doctest: +SKIP (RCR(RuRR%R((s1lib/python2.7/site-packages/distributed/client.pyR s7cO`s&tjddd|j|||S(s Spawn a coroutine on all workers. This spaws a coroutine on all currently known workers and then waits for the coroutine on each worker. The coroutines' results are returned as a dictionary keyed by worker address. Parameters ---------- function: a coroutine function (typically a function wrapped in gen.coroutine or a Python 3.5+ async function) *args: arguments for remote function **kwargs: keyword arguments for remote function wait: boolean (default True) Whether to wait for coroutines to end. workers: list Workers on which to run the function. Defaults to all known workers. scThis method has been deprecated. Instead use Client.run which detects async functions automaticallyt stackleveli(RRR(RuRR%R((s1lib/python2.7/site-packages/distributed/client.pyt run_coroutineB sc  `sj|r4j|dtj|}nr^jdtj|n| dk r| tk r| tk rtj | } nt |tt t |} fdD} fdj D} | rtjjd| ndj D}|rVt jg|jD]}|d^q=nt }d|D}td|j D|}d |j D}x2|D]*}|jk rd }t|qqW|rtt |}tt|}n|dk r$tt t |}nd |j D}xC|jD]5}x,|D]$}|jkrTt|qTqTWqGWfd D}|dkrtjjd |}tt |}nd|j D}xI|j D];\}}|rtt |j|d|B||| s c`s7h|]-\}}t|tr|kr|qS((RRe(R(R^R(tkeyset(s1lib/python2.7/site-packages/distributed/client.pys  s RgcS`s+i|]!\}}t|dt|qS(R(R%RZ(R(R^R((s1lib/python2.7/site-packages/distributed/client.pys s icS`sh|]}t|jqS((RGRi(R(R((s1lib/python2.7/site-packages/distributed/client.pys s cS`s#i|]\}}|d|qS(i((R(R^R((s1lib/python2.7/site-packages/distributed/client.pys s cS`s+i|]!\}}||k r||qS(((R(R^R((s1lib/python2.7/site-packages/distributed/client.pys s s;Inputs contain futures that were created by another client.cS`s3i|])\}}d|dDt|qS(cS`sh|]}t|jqS((RGRi(R(R((s1lib/python2.7/site-packages/distributed/client.pys s i(RG(R(R^R((s1lib/python2.7/site-packages/distributed/client.pys s c`s"i|]}t||qS((R(R(R^(R(s1lib/python2.7/site-packages/distributed/client.pys s t dependenciescS`s>i|]4\}}g|D]}t|^qt|qS((RG(R(R^tdepstdep((s1lib/python2.7/site-packages/distributed/client.pys s ic`si|]}|qS(((R(R^(R(s1lib/python2.7/site-packages/distributed/client.pys s s update-graphRfRRRRRRRitsubmitting_taskRRR(("Rt_expand_resourcest itertoolstchaint_expand_retriesR\RZRjRXt _expand_keyRRRGRRt optimizationtinlinetunionR,RIRhRVRRRoRtorderRRR RrR>R"RM(RuRRgRRRRRRRRtflatkeysRoR,RRt extra_futurest extra_keystdsk2tdsk3RRItfuture_dependenciesRRR^R((RRRRus1lib/python2.7/site-packages/distributed/client.pyR_ sz $ ;    -   t60sc K`s|j|dtt|gd|d|d|d| d| d| d| }t||}|r ttd trytt}Wqt k rt}qXnz|j |d |d |}Wd x|j D]}|j qWttd tr|rt nX|S|S( s Compute dask graph Parameters ---------- dsk: dict keys: object, or nested lists of objects restrictions: dict (optional) A mapping of {key: {set of worker hostnames}} that restricts where jobs can take place 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 sync: bool (optional) Returns Futures if False or concrete values if True (default). 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. 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 RgRRRRRRRRiRRN(RRRR&R"RMRjRARZRKRR,RR=(RuRRgRRRRCRRRRRRRRotpackedt should_rejoinRR((s1lib/python2.7/site-packages/distributed/client.pyR s40     c C`s|jnt}x`t|D]R}t||jkr|sSt}t|}nt||dt||>> 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 N(t__dask_graph__Rt __dask_keys__tredict_collection(Rut collectiontdsk_origR((s1lib/python2.7/site-packages/distributed/client.pytnormalize_collection) s   c  `swt|ttttfr$t} n|g}t} | jdt} | rdtd|D}ng|D]}tj |rk|^qk}j ||| }g|D]}dt |^q}i}xt t ||D]z\}\}}|j\}}|j}|tkrFt|dkrF| rF|d||>> 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 ttraversecs`sB|]8}t|tttttfr6tj|n|VqdS(N(RRXRRwRR.RR(R(ta((s1lib/python2.7/site-packages/distributed/client.pys ss finalize-%siic`s5i|]+\}}j|D]}||qqS((R(R(R_tpR^(Ru(s1lib/python2.7/site-packages/distributed/client.pys s RRRRRN(RRXRwRRRjRZR!Rtis_dask_collectionRR t enumerateRt__dask_postcompute__R!RR*tget_restrictionsR RRRRHRR(Rut collectionsRCtoptimize_graphR&RRRRRRRt singletonR&R't variablesRRRRR_RR$t extra_argsRgRRt futures_dictRotargR((Rus1lib/python2.7/site-packages/distributed/client.pytcomputeH s\E  (#( %     c  `st|ttttfr$t} nt} |g}ttt j |sQt j ||| } d|D} j |||\}}t|tsfd|jD}nj| | ||d|d|d|d|d| g|D]}|j^q}gt||D]=\\}}}|fdt|jD|^q$}| rwt|S|Sd S( sG Persist 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: str, list, dict Which workers can run which parts of the computation If a string a list then the output collections will run on the listed workers, but other sub-computations can run anywhere If a dict then keys should be (tuples of) collections and values should be addresses or lists. allow_other_workers: bool, list If True then all restrictions in workers= are considered loose If a list then only the keys for the listed collections are loose 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 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 cS`s/h|]%}t|jD] }|qqS((RR!(R(R_R^((s1lib/python2.7/site-packages/distributed/client.pys  s c`s5i|]+\}}j|D]}||qqS((R(R(R_R(R^(Ru(s1lib/python2.7/site-packages/distributed/client.pys  s RRRRRc`si|]}||qS(((R(R^(Ro(s1lib/python2.7/site-packages/distributed/client.pys ) s N(RRwRXRRRjRZRRRR)RRR,R RRt__dask_postpersist__RRR!R(RuR-R.R&RRRRRRRR/RRRRR_t postpersistsR$R%R((RoRus1lib/python2.7/site-packages/distributed/client.pytpersist s69   P c#`sdtjj|d|j|Vdfd}|j|dtVtjd dS(Nic `s;ddlm}ddl}ddl}|tjj|j}tjj|j}tjj |}|j |||j |}|j d|WdQXxSt tjj|d ddD],}tj|} tj|| jdBqWtjjtjj|d s-t|SWdQXdS(Ni(RHROitbint*i@(tdistributed.utilsRHtzipfiletshutilRROtjoint worker_dirt local_dirtdirnametmovetZipFilet extractallR tstattchmodtst_modeRPR( t dask_workerRHR;R<R'tbR_RRRf(R(s1lib/python2.7/site-packages/distributed/client.pytunzip7 s   ,+Ri( RROR:t_upload_large_fileR\RRZRR(RuR;RI((Rs1lib/python2.7/site-packages/distributed/client.pyt_upload_environment2 s cC`s|j|j||S(N(RCRK(RuRR;((s1lib/python2.7/site-packages/distributed/client.pytupload_environmentO scc`s|tkr|jd}n|jidd6|d6t|_y"|jj|jj|VWn$tj k rt j d|nX|j d7_ |j |jjWdQXtj|dS(NiRRfRs"Restart timed out after %f secondsi(RNRRrRRzRRR4RRRRRmRRRR(RuR((s1lib/python2.7/site-packages/distributed/client.pyt_restartR s  " cK`s|j|j|S(s Restart the distributed network This kills all active work, deletes all data on the network, and restarts the worker processes. (RCRM(RuR((s1lib/python2.7/site-packages/distributed/client.pyRc sc #`st|d}|jWdQXtjj|\}}|jjdidd6|d6td6V}td|j Drg|j D] }|dd kr|d ^q}|r|d qt j |d nt fd |j Ds t dS( NtrbRIt upload_fileRftfilenameRcs`s|]}|ddkVqdS(R[RN((R(R((s1lib/python2.7/site-packages/distributed/client.pys t sR[RRic3`s%|]}t|dkVqdS(RN(R*(R(R(R(s1lib/python2.7/site-packages/distributed/client.pys { s(RSRmRROR:RRR6tanyR,RRRR( RuRPtraise_on_errorRRRRRR((Rs1lib/python2.7/site-packages/distributed/client.pyt _upload_filek s %3 c#`sdkr%tjj|dnt|d}|jWdQX|jgV\}|j|j|Vdfd}|j |V}t fd|j Dst dS(NiRNc`sptjjs-tjj|j}n}t|d}|j|jWdQXt|jS(Ntwb( RROtisabsR=R?RSRlRR*(RGRR(Ritremote_filename(s1lib/python2.7/site-packages/distributed/client.pyt dump_to_file s c3`s!|]}t|kVqdS(N(R*(R(R(R(s1lib/python2.7/site-packages/distributed/client.pys s( R\RROR:RSRmRRiRRRR,R(Rutlocal_filenameRVRRRWR((RRiRVs1lib/python2.7/site-packages/distributed/client.pyRJ} s   cK`sA|j|j|d|j|}t|tr9|n|SdS(sF Upload local package to workers This sends a local file up to all worker nodes. This file is placed into a temporary directory on 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 Examples -------- >>> client.upload_file('mylibrary.egg') # doctest: +SKIP >>> from mylibrary import myfunc # doctest: +SKIP >>> L = c.map(myfunc, seq) # doctest: +SKIP RRN(RCRSRRRK(RuRPRR((s1lib/python2.7/site-packages/distributed/client.pyRO s  cc`s`t|Vtd|j|D}|jjd|d|V}|ddks\tdS(NcS`sh|]}t|jqS((RGRi(R(R((s1lib/python2.7/site-packages/distributed/client.pys s RgR&R[R(t_waitRXRRt rebalanceR(RuRoR&RgR((s1lib/python2.7/site-packages/distributed/client.pyt _rebalance s cK`s|j|j|||S(s Rebalance 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. This operation is generally not well tested against normal operation of the scheduler. It it 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 (RCR[(RuRoR&R((s1lib/python2.7/site-packages/distributed/client.pyRZ sic c`sZ|j|}t|Vd|D}|jjdt|d|d|d|VdS(NcS`sh|]}t|jqS((RGRi(R(R((s1lib/python2.7/site-packages/distributed/client.pys s RgRR&tbranching_factor(RRYRt replicateRX(RuRoRR&R\Rg((s1lib/python2.7/site-packages/distributed/client.pyR s   c K`s(|j|j|d|d|d||S(s Set replication of futures within network Copy data onto many workers. This helps to broadcast frequently accessed data and it helps to 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. 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 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 RR&R\(RCR(RuRoRR&R\R((s1lib/python2.7/site-packages/distributed/client.pyR] s$cK`st|tr4td|Dr4t|}n|dk ret|tttf re|g}n|j|jjd||S(s6 The 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. Examples -------- >>> c.ncores() # 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 cs`s$|]}t|ttfVqdS(N(RRRw(R(R_((s1lib/python2.7/site-packages/distributed/client.pys  sR&N( RRwRRXR\RRCRR'(RuR&R((s1lib/python2.7/site-packages/distributed/client.pyR' s % cK`s_|dk r=|j|}tttd|D}nd}|j|jjd||S(sY The workers storing each future's data Parameters ---------- futures: list (optional) A list of futures, defaults to all data 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.ncores cS`sh|]}|jqS((Ri(R(R((s1lib/python2.7/site-packages/distributed/client.pys < s RgN(R\RRXRRGRCRR(RuRoRRg((s1lib/python2.7/site-packages/distributed/client.pyR s  "cK`st|tr4td|Dr4t|}n|dk ret|tttf re|g}n|j|jjd||S(s Which 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 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.ncores Client.processing cs`s$|]}t|ttfVqdS(N(RRRw(R(R_((s1lib/python2.7/site-packages/distributed/client.pys \ sR&N( RRwRRXR\RRCRthas_what(RuR&R((s1lib/python2.7/site-packages/distributed/client.pyR^A s % cC`s~t|tr4td|Dr4t|}n|dk ret|tttf re|g}n|j|jjd|S(s The 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.ncores cs`s$|]}t|ttfVqdS(N(RRRw(R(R_((s1lib/python2.7/site-packages/distributed/client.pys z sR&N( RRwRRXR\RRCRt processing(RuR&((s1lib/python2.7/site-packages/distributed/client.pyR_c s % cK`s"|j|jjd|d||S(s  The 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 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 Rgtsummary(RCRR(RuRgR`R((s1lib/python2.7/site-packages/distributed/client.pyR scC`sl|p g}|dk rM|j|}|tttd|D7}n|j|jjd|phdS(s The 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 cS`sh|]}|jqS((Ri(R(R((s1lib/python2.7/site-packages/distributed/client.pys s RgN(R\RRXRRGRCRt call_stack(RuRoRg((s1lib/python2.7/site-packages/distributed/client.pyRa s   &cC`s_t|tjtfr%|g}n|j|jd|d|d|d|d|d|d|S(s Collect 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 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 RiR&t merge_workersRR~tplotRP(RRRR RCt_profile(RuRiRR~R&RbRcRP((s1lib/python2.7/site-packages/distributed/client.pytprofile s  c c`st|tjtfr%|g}n|jjd|d|d|d|d|V}|rbt}n|rddlm} | j|} | j | dd \} } |d kr| rd }nd d l m } | | ddd|t j || fnt j |dS(NRiR&RbRR~i(Ret sizing_modet stretch_bothtsavesdask-profile.htmli(Rhttitles Dask ProfileRP(RRRR RReRZRt plot_datat plot_figuretbokeh.plottingRhRR(RuRiRR~R&RbRcRPRwReRtfiguretsourceRh((s1lib/python2.7/site-packages/distributed/client.pyRd s(      cK`s|j|j|jS(s Basic information about the workers in the cluster 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}}} (RCR R(RuR((s1lib/python2.7/site-packages/distributed/client.pytscheduler_info sc C`s\|jrtdn ||_t|jd#}tj|j|ddWdQXdS(s Write 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 sScheduler file already setR)tindentiN(RRVRSRTtdumpRo(RuRR((s1lib/python2.7/site-packages/distributed/client.pytwrite_scheduler_file% s   cC`s@t|ttfs!|f}n|j|jjd|d|S(s Get 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 RgR(RRXRwRCRt get_metadata(RuRgR((s1lib/python2.7/site-packages/distributed/client.pyRs@ s cC`s|j|jjd|S(s  Get logs from scheduler Parameters ---------- n: int Number of logs to retrive. Maxes out at 10000 by default, confiruable in config.yaml::log-length Returns ------- Logs in reversed order (newest first) R(RCRtlogs(RuR((s1lib/python2.7/site-packages/distributed/client.pytget_scheduler_logsV s cC`s|j|jjd|d|S(s Get logs from workers Parameters ---------- n: int Number of logs to retrive. Maxes out at 10000 by default, confiruable in config.yaml::log-length workers: iterable List of worker addresses to retrive. Gets all workers by default. Returns ------- Dictionary mapping worker address to logs. Logs are returned in reversed order (newest first) RR&(RCRt worker_logs(RuRR&((s1lib/python2.7/site-packages/distributed/client.pytget_worker_logse scK`s"|j|jjd|d||S(s Retire certain workers on the scheduler See dask.distributed.Scheduler.retire_workers for the full docstring. 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&t close_workers(RCRtretire_workers(RuR&RxR((s1lib/python2.7/site-packages/distributed/client.pyRyw s  cC`s:t|ts|f}n|j|jjd|d|S(sw Set 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 RgRr(RRXRCRt set_metadata(RuRiRr((s1lib/python2.7/site-packages/distributed/client.pyRz s+ c`s+td|}y"t|j|jjd|}Wn?tk rJd}n)tk rrt|j|jj}nXt|j|jjdidd6|d6}i|d6|d6|d6}|r'd|d}d|dfg}|j fd t |j Dt t } xi|D]a\} } xR|j D]D\} } | j| d }| |krG| | j| |fqGqGWq.W| r'g}xft | j D]R\} }d|| fg}|j ||jd | td d g|fqWtddj|q'n|S(s Return version info for the scheduler, all workers and myself Parameters ---------- check : boolean, default False raise ValueError if all required & optional packages do not match packages : List[str] Extra package names to check Examples -------- >>> c.get_versions() # doctest: +SKIP >>> c.get_versions(packages=['sklearn', 'geopandas']) # doctest: +SKIP tpackagesRItversionsRfRR&RhcS`s6t|dj}tt|t|dS(NR{i(RXR,RR+R(RR]((s1lib/python2.7/site-packages/distributed/client.pyt to_packages sc3`s'|]\}}||fVqdS(N((R(R)R(R}(s1lib/python2.7/site-packages/distributed/client.pys stMISSINGs%s %sRtversionsMismatched versions found %ss N(RTRCRRR|RcR\RRtextendRWRRRXRRHRLRVR=(RutcheckR{RhRR&Rtclient_versionsR|t mismatchedRtverstpkgtcvRterrstrows((R}s1lib/python2.7/site-packages/distributed/client.pyRT s@"     )  " *cC`st|d|S(NRh(R(RuRo((s1lib/python2.7/site-packages/distributed/client.pyR scO`stddS(Ns%Method moved to start_ipython_workers(RK(RuR%R((s1lib/python2.7/site-packages/distributed/client.pyt start_ipython scc`s]|dkr|jjV}n|jjdtddd|V}tj||fdS(NRIRfRR&(R\RR'RRRR(RuR&R((s1lib/python2.7/site-packages/distributed/client.pyt_start_ipython_workerss   cC`st|tjtfr%|g}nt|j|j|\}}|rt|tjrd|krgtt|D]}|j dt |^qz}q|g}ndt j krddl m}|n|r&ddl m}x7t||D]#\} } || } || | qWn|rddl m} xU|jD]D\} } d| j dd j d d } | | d | d |qIWn|S( s Start IPython kernels on workers Parameters ---------- workers: list (optional) A list of worker addresses, defaults to all magic_names: str or list(str) (optional) If defined, register IPython magics with these names for executing code on the workers. If string has asterix then expand asterix into 0, 1, ..., n for n workers qtconsole: bool (optional) If True, launch a Jupyter QtConsole connected to the worker(s). qtconsole_args: list(str) (optional) Additional arguments to pass to the qtconsole on startup. Examples -------- >>> info = c.start_ipython_workers() # doctest: +SKIP >>> %remote info['192.168.1.101:5752'] worker.data # doctest: +SKIP {'x': 1, 'y': 100} >>> c.start_ipython_workers('192.168.1.101:5752', magic_names='w') # doctest: +SKIP >>> %w worker.data # doctest: +SKIP {'x': 1, 'y': 100} >>> c.start_ipython_workers('192.168.1.101:5752', qtconsole=True) # doctest: +SKIP Add asterix * in magic names to add one magic per worker >>> c.start_ipython_workers(magic_names='w_*') # doctest: +SKIP >>> %w_0 worker.data # doctest: +SKIP {'x': 1, 'y': 100} >>> %w_1 worker.data # doctest: +SKIP {'z': 5} Returns ------- iter_connection_info: list List of connection_info dicts containing info necessary to connect Jupyter clients to the workers. See Also -------- Client.start_ipython_scheduler: start ipython on the scheduler R9tIPythoni(tregister_remote_magic(tregister_worker_magic(tconnect_qtconsolesdask-R3Rt/RR1(RRRR RCRRRRR*treplaceRtsystmodulest_ipython_utilsRRRRR(RuR&t magic_namest qtconsoletqtconsole_argst info_dictR_RRRt magic_nametconnection_infoRR((s1lib/python2.7/site-packages/distributed/client.pytstart_ipython_workers s*3  :   "tscheduler_if_ipythonc C`st|j|jj}|dkrst}dtjkr[ddlm}t |}n|rjd}qsd }n|rddl m }|||n|rddl m }||d d d |n|S( s{ Start IPython kernel on the scheduler Parameters ---------- magic_name: str or None (optional) If defined, register IPython magic with this name for executing code on the scheduler. If not defined, register %scheduler magic if IPython is running. qtconsole: bool (optional) If True, launch a Jupyter QtConsole connected to the worker(s). qtconsole_args: list(str) (optional) Additional arguments to pass to the qtconsole on startup. Examples -------- >>> c.start_ipython_scheduler() # doctest: +SKIP >>> %scheduler scheduler.processing # doctest: +SKIP {'127.0.0.1:3595': {'inc-1', 'inc-2'}, '127.0.0.1:53589': {'inc-2', 'add-5'}} >>> c.start_ipython_scheduler(qtconsole=True) # doctest: +SKIP Returns ------- connection_info: dict connection_info dict containing info necessary to connect Jupyter clients to the scheduler. See Also -------- Client.start_ipython_workers: Start IPython on the workers RRi(t get_ipythonRi(R(RRsdask-schedulerR1N(RCRRRRjRRRRtboolR\RRR( RuRRRRt in_ipythonRRR((s1lib/python2.7/site-packages/distributed/client.pytstart_ipython_scheduler^s %   cc`sot|ts|f}nxM|D]E}tj|r\x-|jD]}t|VqDWq"t|Vq"WdS(ss Expand a user-provided task key specification, e.g. in a resources or retries dictionary. N(RRwRR)R!RG(RR^tkktkkk((s1lib/python2.7/site-packages/distributed/client.pyRs  c`sr4ttr4fdjD}n>ttr\fd|D}ntdttt|S(sp Expand the user-provided "retries" specification to a {task key: Integral} dictionary. c`s5i|]+\}}j|D]}||qqS((R(R(RiRrR(R(s1lib/python2.7/site-packages/distributed/client.pys s  c`si|]}|qS(((R(R(R(s1lib/python2.7/site-packages/distributed/client.pys s s.`retries` should be an integer or dict, got %r(RRRR RRRRG(RRRR((RRs1lib/python2.7/site-packages/distributed/client.pyRs c`st|ts(tdt|ni}i}t|}xr|jD]d\ttr|jfd|jDqM|jfd|DqMW|r|rtd|fn|p|S(s Expand the user-provided "resources" specification to a {task key: {resource name: Number}} dictionary. s$`resources` should be a dict, got %rc3`s|]}|fVqdS(N((R(R(R(s1lib/python2.7/site-packages/distributed/client.pys sc3`s"|]}|i6fVqdS(N((R(R(R^R(s1lib/python2.7/site-packages/distributed/client.pys ssFcannot have both per-key and all-key requirements in resources dict %r( RRRRRXRRRRV(RRRt per_key_reqst global_reqs((R^Rs1lib/python2.7/site-packages/distributed/client.pyR s  )$ c`s1t|tttfr.i|t|6}nt|tri}x|jD]\}ttrwgntj|rt|j }ntdt|D}|j fd|DqPWni}|t krt|}n+|r!tdt|D}ng}||fS(s1 Get restrictions from inputs to compute/persist cS`s/h|]%}t|jD] }|qqS((RR!(R(R_R^((s1lib/python2.7/site-packages/distributed/client.pys s c`si|]}|qS(((R(R^(tws(s1lib/python2.7/site-packages/distributed/client.pys s cS`s)h|]}|jD] }|qqS((R!(R(R_R^((s1lib/python2.7/site-packages/distributed/client.pys s ( RRRwRXRRRR)RR!RRZ(RR-R&RRtcollsRgR((Rs1lib/python2.7/site-packages/distributed/client.pyR,s( $ cO`st|||S(N(R(R-R%R((s1lib/python2.7/site-packages/distributed/client.pyRsstask-stream.htmlc C`s.|j|jd|d|d|d|d|S(s Get task stream data from scheduler This collects the data present in the diagnostic "Task Stream" plot on the dashboard. It includes the start, stop, transfer, and deserialization time of every task for a particular duration. Note that the task stream diagnostic does not run by default. You may wish to call this function once before you start work to ensure that things start recording, and then again after you have completed. Parameters ---------- start: Number or string When you want to start recording If a number it should be the result of calling time() If a string then it should be a time difference before now, like '60s' or '500 ms' stop: Number or string When you want to stop recording count: int The number of desired records, ignored if both start and stop are specified 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 -------- >>> 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 RR~tcountRcRP(RCt_get_task_stream(RuRR~RRcRP((s1lib/python2.7/site-packages/distributed/client.pytget_task_streams> c c`s|jjd|d|d|V}|rddlm}||}ddlm} | dd\} } | jj||d krd d lm } | | d d d|nt j || fnt j |dS(NRR~Ri(t rectangles(ttask_stream_figureRfRgRhi(RhRisDask Task StreamRP( RRtdiagnostics.task_streamRtbokeh.componentsRRRRlRhRR( RuRR~RRcRPRxRtrectsRRnRmRh((s1lib/python2.7/site-packages/distributed/client.pyREs"  cc`s|jjdt|V}i}xa|jD]S\}}|ddkr\|d|| None Function to register and run on all workers R(RCR(RuR((s1lib/python2.7/site-packages/distributed/client.pyResN(yRRRR\RNRZRjtDEFAULT_EXTENSIONSRzt classmethodRRRRCRRRRRJRrRRRCRhR\R R RpRqRsRuRRlRRvR]RRRRRRRRet _shutdownRtRRRRRRRRRRRRRRRRRRRRRRRRRRRRRRR%R4R7RKRLRMRRSRJROR[RZRR]R'RR^R_RRaReRdRoRrRsRuRwRyRzRTRRRRRRRR R,t staticmethodRRRRR(((s1lib/python2.7/site-packages/distributed/client.pyRszF |   D  I1       -     D $  e  e! Ie  m  0     9 ! ^ @  "  Y   -  ! "  %!     />   R: F tExecutorcB`seZdZdZRS(s Deprecated: see Client cO`s*tjdtt|j||dS(Ns#Executor has been renamed to Client(RRRRRz(RuR%R((s1lib/python2.7/site-packages/distributed/client.pyRz}s (RRRRz(((s1lib/python2.7/site-packages/distributed/client.pyRzscO`stddS(Ns7This has been moved to the Client.get_executor() method(RK(R%R((s1lib/python2.7/site-packages/distributed/client.pytCompatibleExecutorst ALL_COMPLETEDtFIRST_COMPLETEDc c`s/|dk r+t|t r+tdnt|}|tkrLt}n!|tkrat}n t d|d|D}|dk rt j t d||}n|Vd|Dd|D}}g|D]}|j dkr|j^q}|rt|nt jt||dS(Nstimeout= keyword received a non-numeric value. Beware that wait expects a list of values Bad: wait(x, y, z) Good: wait([x, y, z])sDOnly return_when='ALL_COMPLETED' and 'FIRST_COMPLETED' are supportedcS`sh|]}|jjqS((RpR(R(R((s1lib/python2.7/site-packages/distributed/client.pys s R cS`s%h|]}|jdkr|qS(R(R[(R(tfu((s1lib/python2.7/site-packages/distributed/client.pys s cS`s%h|]}|jdkr|qS(R(R[(R(R((s1lib/python2.7/site-packages/distributed/client.pys s R(R\RR RRRRBRRSRRR#RR[RiRRR( tfsRt return_whentwait_forRR|tnot_doneRR((s1lib/python2.7/site-packages/distributed/client.pyRYs(         +cC`s+t}|jt|d|d|}|S(s  Wait until all/any futures are finished Parameters ---------- fs: list of futures timeout: number, optional Time in seconds after which to raise a ``dask.distributed.TimeoutError`` ------- Named tuple of completed, not completed RR(RRCRY(RRRRhR((s1lib/python2.7/site-packages/distributed/client.pyRs cc`st|}td|}g|jD]}|d^q+}tjg|D]}|jj^qN}xM|js|jV||j }x"||j D]}|j |qWqoWdS(NcS`s|jS(N(Ri(R((s1lib/python2.7/site-packages/distributed/client.pyRRi( RRR,Rt WaitIteratorRpRR|Rt current_indexRit put_nowait(RtqueuetgroupsRtfirstsRt wait_iteratorR((s1lib/python2.7/site-packages/distributed/client.pyt _as_completeds #+  cc`s7t}t||V|jV}tj|dS(sK Return a single completed future See Also: _as_completed N(R"RRRR(RoRR((s1lib/python2.7/site-packages/distributed/client.pyt_first_completeds  t as_completedcB`seZdZddeedZdZej dZ dZ dZ dZ dZdZd Zd Zd Zd Zej d ZeZedZdZRS(s 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`. 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 cC`s|dkrg}ntd|_t|_tj|_|pQtj |_ t |_ tj |_ ||_ ||_|r|j|ndS(NcS`sdS(Ni((((s1lib/python2.7/site-packages/distributed/client.pyRR(R\RRoRRRtLocktlockRRRt conditiontthread_conditiont with_resultst raise_errorsR(RuRoRRR((s1lib/python2.7/site-packages/distributed/client.pyRzs      cC`s.|jj|j|jjWdQXdS(N(RtnotifyR(Ru((s1lib/python2.7/site-packages/distributed/client.pyt_notify"s  cc`syt|VWntk r"nX|jrB|jdtV}n|jn|j|cd8<|j|sy|j|=n|jr|jj||fn|jj||j WdQXdS(NR~i( RYRRRRjRRoRRR(RuRR((s1lib/python2.7/site-packages/distributed/client.pyt _track_future's      c C`sp|jaxY|D]Q}t|ts9td|n|j|cd7<|jj|j|qWWdQXdS(sp Add multiple futures to the collection. The added futures will emit from the iterator once they finishsInput must be a future, got %siN(RRReRRoRRR(RuRoR((s1lib/python2.7/site-packages/distributed/client.pyR9s   cC`s|j|fdS(sj Add a future to the collection This future will emit from the iterator once it finishes N(R(RuR((s1lib/python2.7/site-packages/distributed/client.pyRDscC`s |j S(s7Returns True if there no completed or computing futures(R(Ru((s1lib/python2.7/site-packages/distributed/client.pytis_emptyKscC`s|jj S(s6Returns True if there are completed futures available.(RR(Ru((s1lib/python2.7/site-packages/distributed/client.pyt has_readyOscC`s1|j"t|jt|jjSWdQXdS(s Return 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(RR*RoR(Ru((s1lib/python2.7/site-packages/distributed/client.pyRSs cC`s|S(N((Ru((s1lib/python2.7/site-packages/distributed/client.pyt__iter__]scC`s|S(N((Ru((s1lib/python2.7/site-packages/distributed/client.pyt __aiter__`scC`sS|jj}|jrO|\}}|jrO|jdkrOtj|qOn|S(NR(RRRRR[RR(RutresRR((s1lib/python2.7/site-packages/distributed/client.pyt_get_and_raisecs   cC`s[xN|jjrP|jr*tn|j|jjddWdQXqW|jS(NRg?(RRRRRRR(Ru((s1lib/python2.7/site-packages/distributed/client.pyt__next__ks    cc`sq|j r"|jjr"tnx3|jjrW|jsFtn|jjVq%Wtj|jdS(N( RoRRR-RRRRR(Ru((s1lib/python2.7/site-packages/distributed/client.pyt __anext__ss   cC`sN|rt|g}ng}x)|jjsI|j|jjq!W|S(sY Get 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 (RRRRHR(Rutblocktbatch((s1lib/python2.7/site-packages/distributed/client.pyt next_batchs cc`s;x4tr6y|jdtVWqtk r2dSXqWdS(s3 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] RN(RZRR(Ru((s1lib/python2.7/site-packages/distributed/client.pytbatchess   N(RRRR\RjRZRzRRRRRRRRRRRRRRRRR(((s1lib/python2.7/site-packages/distributed/client.pyRs"8          cO`stddS(NsThis has moved to as_completed(RK(R%R((s1lib/python2.7/site-packages/distributed/client.pyt AsCompletedscC`s)|p t}|r|StddS(s$ Return a client if one has started sNo clients found Start an client and point it to the scheduler address from distributed import Client client = Client('ip-addr-of-scheduler:8786') N(R`RV(R_((s1lib/python2.7/site-packages/distributed/client.pyRs cC`s!tjjddt|dS(NRsdask.distributed(RRRRb(Rh((s1lib/python2.7/site-packages/distributed/client.pytensure_default_getscC`sOddlm}t||r/||j|Stj|}||_|SdS(Ni(tDelayed(t dask.delayedRRRitcopyR(R_RRtcc((s1lib/python2.7/site-packages/distributed/client.pyR"s  cC`s1|g}t}x|r|j}t|tttfkrR|j|qt|tkrz|j|jqt|tkr|j|j jqt |t r|j |qt j|r|j|jjqqW|dk r'd|D}|r't|q'nt|S(s Future objects in a collection cS`s"h|]}|jr|qS((R(R(R((s1lib/python2.7/site-packages/distributed/client.pys s N(RR!RRwRXRRR,RRRReRRR)R R\R(toRhtstackRoRtbad((s1lib/python2.7/site-packages/distributed/client.pyRs&      cC`sRt|}x?|D]7}|jjidd6t|jgd6dd6qWdS(sn Run 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 sclient-desires-keysRfRgsfire-and-forgetRhN(RRhRrRGRi(tobjRoR((s1lib/python2.7/site-packages/distributed/client.pytfire_and_forgets   RcB`sVeZdZdeddZdZdZej dZ ej dZ RS(s 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://rawgit.com :: $ 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://rawgit.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 stask-stream.htmlcC`sSg|_||_||_d|_|p0t|_|jjdddddS(NRiR~(Rt_plott _filenameR\RmRRhR(RuRhRcRP((s1lib/python2.7/site-packages/distributed/client.pyRzFs     cC`st|_|S(N(R4R(Ru((s1lib/python2.7/site-packages/distributed/client.pyRpNs cC`sY|jjd|jd|jd|j}|jrE|\}|_n|jj|dS(NRRcRP(RhRRRRRmRR(RuRRrRR]((s1lib/python2.7/site-packages/distributed/client.pyRuRs   cC`stj|dS(N(RR(Ru((s1lib/python2.7/site-packages/distributed/client.pyRqZscc`sZ|jjd|jd|jd|jV}|jrF|\}|_n|jj|dS(NRRcRP(RhRRRRRmRR(RuRRrRR]((s1lib/python2.7/site-packages/distributed/client.pyRs^s   N( RRRR\RjRzRpRuRRRqRs(((s1lib/python2.7/site-packages/distributed/client.pyRs /  cc`s.t}t|z dVWdt|XdS(s Set the default client for the duration of the context Parameters ---------- c : Client This is what default_client() will return within the with-block. N(RRb(R_told_exec((s1lib/python2.7/site-packages/distributed/client.pyRhs   cC`s5t}|dk r1t|_|jddndS(s} Force close of global client. This cleans up when a client wasn't close explicitly, e.g. interactive sessions. RiN(R`R\RjRRt(R_((s1lib/python2.7/site-packages/distributed/client.pyt_close_global_clientys   (t __future__RRRtatexitR-Rtconcurrent.futuresRtconcurrent.futures._baseRRt contextlibRRtdatetimeRRZt functoolsR R R RTtloggingtnumbersR R RRRRRRYRtweakrefRt dask.baseR RRt dask.coreRRtdask.optimizationRtdask.compatibilityRRt dask.utilsRtcytoolzRRRRRt ImportErrorttoolzRRttornadoRt tornado.genRt tornado.locksRRR ttornado.ioloopR!ttornado.queuesR"tbatchedR#t utils_commR$R%R&R'R(t cfexecutorR)t compatibilityRR*R+R,R-R.tcoreR/R0R1R2R3tmetricsR4tnodeR5R=R6tprotocol.pickleR7R8tpublishR9tpubsubR:RR;R<tthreadpoolexecutorR=RR>R?R@RAtutilsRBRCRDRERFRGRHRIRJRKRLRMRNRORPRQRRRSR|RTt getLoggerRRtWeakValueDictionaryRYRaRR`RbRdRetobjectRqRRtregisterRRKRRRRRRR\RYRRRRRRRR"RRRRR(((s1lib/python2.7/site-packages/distributed/client.pyts               , ,  (.("v     %@ "      "S