B ^\ 5@sddlmZmZmZddlZddlmZddlZddlZddl Z ddl Z ddl Z ddl m Z ddlmZddlmZdd lmZdd lmZdd lmZmZmZmZmZmZmZmZdd lm Z dd l!m"Z"ddl#m$Z$m%Z%e&e'Z(GdddeZ)ddZ*e +Z,ej-ddZ.dS))print_functiondivisionabsolute_importN) timedelta)factors)gen)Cluster)get_thread_identity)CommClosedError)syncignoringAllsilence_logging LoopRunner log_errors thread_stateparse_timedelta)Nanny) Scheduler)Worker_ncoresc@seZdZdZdddddddejddddddfddZd d Zd d Ze d dZ ddZ ddZ e jd4ddZe jd5ddZddZe jddZddZe jd6dd Zd7d"d#Ze jd$d%Ze jd&d'Zd(d)Zd*d+Zd,d-Ze jd.d/Ze jd0d1Ze d2d3ZdS)8 LocalClustera Create local Scheduler and Workers This creates a "cluster" of a scheduler and workers running on the local machine. Parameters ---------- n_workers: int Number of workers to start processes: bool Whether to use processes (True) or threads (False). Defaults to True threads_per_worker: int Number of threads per each worker scheduler_port: int Port of the scheduler. 8786 by default, use 0 to choose a random port silence_logs: logging level Level of logs to print out to stdout. ``logging.WARN`` by default. Use a falsey value like False or None for no change. ip: string IP address on which the scheduler will listen, defaults to only localhost diagnostics_port: int Port on which the :doc:`web` will be provided. 8787 by default, use 0 to choose a random port, ``None`` to disable it, or an :samp:`({ip}:{port})` tuple to listen on a different IP address than the scheduler. asynchronous: bool (False by default) Set to True if using this cluster within async/await functions or within Tornado gen.coroutines. This should remain False for normal use. kwargs: dict Extra worker arguments, will be passed to the Worker constructor. service_kwargs: Dict[str, Dict] Extra keywords to hand to the running services security : Security Examples -------- >>> c = LocalCluster() # Create a local cluster with as many workers as cores # doctest: +SKIP >>> c # doctest: +SKIP LocalCluster("127.0.0.1:8786", workers=8, ncores=8) >>> c = Client(c) # connect to local cluster # doctest: +SKIP Add a new worker to the cluster >>> w = c.start_worker(ncores=2) # doctest: +SKIP Shut down the extra worker >>> c.stop_worker(w) # doctest: +SKIP Pass extra keyword arguments to Bokeh >>> LocalCluster(service_kwargs={'bokeh': {'prefix': '/foo'}}) # doctest: +SKIP NTriS"FcKs|dk rd}t|d|_||_||_| |_||_| p8i} | p@i} |rRt|d|_|dkr||dkr||rttt \}}nd}t }|dkr|dk rt dt |}|r|dkrt dt t t |}||| dt|| d|_|jj|_| dk r^| dk r^yddlm}dd lm}Wn tk r6td Yn(X|| pBid if| d | f<|| d <t|j| |d |_||_g|_||_|r||jd<|j||dt !|dS)NzThe start= parameter is deprecated. LocalCluster always starts. For asynchronous operation use the following: cluster = yield LocalCluster(asynchronous=True))levelr)ncoresservices)loop asynchronousFr)BokehScheduler) BokehWorkerz4To start diagnostics web server please install Bokehbokeh)r!r)rrsecurityr")ip n_workers)" ValueErrorstatus processes silence_logs _asynchronousr"r_old_logging_levelnprocesses_nthreadsrmaxintmathZceilupdater _loop_runnerrZdistributed.bokeh.schedulerrZdistributed.bokeh.workerr ImportErrorloggerdebuggetr schedulerscheduler_portworkers worker_kwargsstartclusters_to_closeadd)selfr$Zthreads_per_workerr'rr9r#r6r(Zdiagnostics_portrZworker_servicesZservice_kwargsrr"r8msgrr r>7lib/python3.7/site-packages/distributed/deploy/local.py__init__QsX       zLocalCluster.__init__cCs&d|jt|jtdd|jDfS)Nz'LocalCluster(%r, workers=%d, ncores=%d)css|] }|jVqdS)N)r).0wr>r>r? sz(LocalCluster.__repr__..)scheduler_addresslenr7sum)r<r>r>r?__repr__s zLocalCluster.__repr__cCs |jS)N)_started __await__)r<r>r>r?rIszLocalCluster.__await__cCs,|jp*ttddp*t|jdo*|jjtkS)NrF_thread_identity)r)getattrrhasattrrrJr )r<r>r>r?rs zLocalCluster.asynchronouscOs^|dds|jrF|dd}|||}|dk rBtt|d|}|St|j|f||SdS)Nrcallback_timeout)seconds)poprr with_timeoutrr r)r<funcargskwargsrMZfuturer>r>r?r s   zLocalCluster.synccKs4|j|jr |jf||_n|j|jf|dS)N)r0r9r)_startrHr )r<rSr>r>r?r9s zLocalCluster.startc#sjdkrdS|dkr(js(js(d}n8|dk rJ|drJd|jf}n|dkrVd}|jf}j|fddt|DVd_tdS) z- Start all cluster services. runningNz inproc://ztls://z%s:%dz 127.0.0.1csg|]}jfjqSr>) _start_workerr8)rAi)r<r>r? sz'LocalCluster._start..) r&r6r' startswithr5r9rangerReturn)r<r#r$rDr>)r<r?rTs   zLocalCluster._start<cks|jr&|jdr&td|jdS|jr:t}d|d<nt}||jjf|j ||j d|}| V|j |x(|jdkr|j|jj krtdVqvW|jdkr|jjdkr|j |td t|dS) NZclosz*Tried to start a worker while status=='%s'Tquiet)r death_timeoutr(closedg{Gz?rUzWorker failed to start)r&rYwarningswarnr'rrr5addressrr(rTr7appendworker_addressrZsleepremove TimeoutErrorr[)r<r^rSWrBr>r>r?rVs$     zLocalCluster._start_workercKs|j|jf|S)a Add a new worker to the running cluster Parameters ---------- port: int (optional) Port on which to serve the worker, defaults to 0 or random ncores: int (optional) Number of threads to use. Defaults to number of logical cores Examples -------- >>> c = LocalCluster() # doctest: +SKIP >>> c.start_worker(ncores=2) # doctest: +SKIP Returns ------- The created Worker or Nanny object. Can be discarded. )r rV)r<rSr>r>r? start_workerszLocalCluster.start_workerccs$|V||jkr |j|dS)N)_closer7re)r<rBr>r>r? _stop_workers  zLocalCluster._stop_workercCs||j|dS)z Stop a running worker Examples -------- >>> c = LocalCluster() # doctest: +SKIP >>> w = c.start_worker(ncores=2) # doctest: +SKIP >>> c.stop_worker(w) # doctest: +SKIP N)r rj)r<rBr>r>r? stop_workers zLocalCluster.stop_worker2sc #sjdkrdSd_jttj2ttt|dt fddj DVWdQRXj dd=z:ttjt t jj ddVWdQRXj dd=Wdd_XdS)Nr_closing)rNcsg|]}|qSr>)rj)rArB)r<r>r?rXsz'LocalCluster._close..T)Zfast)r&r5Zclear_task_staterrrfrPrrrr7r OSErrorclose)r<timeoutr>)r<r?ris    & zLocalCluster._closecs~jdkrdSyjj|d}Wntk r:d}YnXtdrjjr`|fddn tjjszj |S)z Close the cluster r_N)rMr*cs tjS)N)rr*)_)r<r>r?2sz$LocalCluster.close..) r&r ri RuntimeErrorrLrZadd_done_callbackrr*r0stop)r<rpresultr>)r<r?ro&s     zLocalCluster.closec +s\tLtj|fddt|tjjDVddjD_WdQRXdS)z Bring the total count of workers up to ``n`` This function/coroutine should bring the total number of workers up to the number ``n``. This can be implemented either as a function or as a Tornado coroutine. csg|]}jfqSr>)rV)rArW)kwargs2r<r>r?rXFsz)LocalCluster.scale_up..cSsg|]}|jdkr|qS)r_)r&)rArBr>r>r?rXJsN)rtoolzmerger8rZrEr5r7)r<nrSr>)rwr<r?scale_up;s  zLocalCluster.scale_upc #sjtZddjD_ttddDrHfddjDfddDVWdQRXdS) aB Remove ``workers`` from the cluster Given a list of worker addresses this function should remove those workers from the cluster. This may require tracking which jobs are associated to which worker address. This can be implemented either as a function or as a Tornado coroutine. cSsg|]}|jdkr|qS)r_)r&)rArBr>r>r?rXXsz+LocalCluster.scale_down..css|]}t|tVqdS)N) isinstancestr)rArBr>r>r?rC\sz*LocalCluster.scale_down..csh|]}|jkr|qSr>)rd)rArB)r7r>r? ]sz*LocalCluster.scale_down..csg|]}|qSr>)rj)rArB)r<r>r?rX`sN)rr7setall)r<r7r>)r<r7r? scale_downLs zLocalCluster.scale_downcCs |dS)N)ro)r<r>r>r?__del__bszLocalCluster.__del__cCs|S)Nr>)r<r>r>r? __enter__eszLocalCluster.__enter__cGs |dS)N)ro)r<rRr>r>r?__exit__hszLocalCluster.__exit__ccs|jVt|dS)N)rHrr[)r<r>r>r? __aenter__kszLocalCluster.__aenter__ccs|VdS)N)ri)r<typvalue tracebackr>r>r? __aexit__pszLocalCluster.__aexit__cCs"y|jjStk rdSXdS)Nz )r5rbr%)r<r>r>r?rDtszLocalCluster.scheduler_address)Nr)r\)rl)rq)__name__ __module__ __qualname____doc__loggingZWARNr@rGrIpropertyrr r9r coroutinerTrVrhrjrkriror{rrrrrrrDr>r>r>r?rs:6=       rcs8dkr}ntfddtD}|}||fS)aB The default breakdown of processes and threads for a given number of cores Parameters ---------- n: int Number of available cores Examples -------- >>> nprocesses_nthreads(4) (4, 1) >>> nprocesses_nthreads(32) (8, 4) Returns ------- nprocesses, nthreads c3s |]}|tkr|VqdS)N)r.Zsqrt)rAf)rzr>r?rCsz&nprocesses_nthreads..)minr)rzr'Zthreadsr>)rzr?r+|s r+cCs"xttD]}|jddq WdS)N )rp)listr:ro)clusterr>r>r?close_clusterssr)/Z __future__rrratexitZdatetimerrr.r`weakrefrxZ dask.utilsrZtornadorrr Z compatibilityr Zcorer Zutilsr rrrrrrrZnannyrr5rZworkerrrZ getLoggerrr2rr+WeakSetr:registerrr>r>r>r?s.      (   d