ó ßüÚ\c@`sddlmZmZmZddlZddlmZddlZddlZddl Z ddl Z ddl Z ddl m Z ddlmZddlmZdd lmZdd lmZdd lmZmZmZmZmZmZmZmZmZdd l m!Z!dd l"m#Z#ddl$m%Z%m&Z&m'Z'ej(e)ƒZ*defd„ƒYZ+d„Z,e j-ƒZ.ej/d„ƒZ0dS(i(tprint_functiontdivisiontabsolute_importN(t timedelta(tfactors(tgeni(tClusteri(tget_thread_identity(tCommClosedError( tget_ip_interfacetsynctignoringtAlltsilence_loggingt LoopRunnert log_errorst thread_statetparse_timedelta(tNanny(t Scheduler(tWorkertparse_memory_limitt_ncorest LocalClustercB`sjeZdZddeddddejdddddedddddd„Zd„Z d„Z e d„ƒZ d„Z d„Zejddd „ƒZejd d „ƒZd „Zejd „ƒZd„Zejdd„ƒZdd„Zejd„ƒZejd„ƒZd„Zd„Zd„Zejd„ƒZejd„ƒZe d„ƒZRS(sR 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 dashboard_address: str Address on which to listen for the Bokeh diagnostics server like 'localhost:8787' or '0.0.0.0:8787'. Defaults to ':8787'. Set to ``None`` to disable the dashboard. Use port 0 for a random port. diagnostics_port: int Deprecated. See dashboard_address. 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. worker_kwargs: dict Extra worker arguments, will be passed to the Worker constructor. blocked_handlers: List[str] A list of strings specifying a blacklist of handlers to disallow on the Scheduler, like ``['feed', 'run_function']`` service_kwargs: Dict[str, Dict] Extra keywords to hand to the running services security : Security protocol: str (optional) Protocol to use like ``tcp://``, ``tls://``, ``inproc://`` This defaults to sensible choice given other keyword arguments like ``processes`` and ``security`` interface: str (optional) Network interface to use. Defaults to lo/localhost worker_class: Worker Worker class used to instantiate workers from. Examples -------- >>> cluster = LocalCluster() # Create a local cluster with as many workers as cores # doctest: +SKIP >>> cluster # doctest: +SKIP LocalCluster("127.0.0.1:8786", workers=8, ncores=8) >>> c = Client(cluster) # connect to local cluster # doctest: +SKIP Add a new worker to the cluster >>> w = cluster.start_worker(ncores=2) # doctest: +SKIP Shut down the extra worker >>> cluster.stop_worker(w) # doctest: +SKIP Pass extra keyword arguments to Bokeh >>> LocalCluster(service_kwargs={'bokeh': {'prefix': '/foo'}}) # doctest: +SKIP is:8787c K`s€|dk r!d}t|ƒ‚n| dk rCtjdƒ| } nd|_||_|dkr»|r‰d|kr‰|jdƒd}q»|r˜d}q»|j r²| r²d}q»d}n|jdƒs×|d}n||_||_ ||_ ||_ ||_ | p i} | pi} |r7t d|ƒ|_n|dkry|dkry|rjttƒ\}}qyd }t}n|dkr§|dk r§td t|ƒ}n|rÞ|dkrÞtd ttjt|ƒƒƒ}n|r d |kr td d |ƒ|d âs(tscheduler_addresstlenRCtsum(RI((s7lib/python2.7/site-packages/distributed/deploy/local.pyt__repr__Þs cC`s |jjƒS(N(t_startedt __await__(RI((s7lib/python2.7/site-packages/distributed/deploy/local.pyRXåscC`s@|jp?ttdtƒp?t|jdƒo?|jjtƒkS(NRt_thread_identity(R0tgetattrRR:thasattrRRYR(RI((s7lib/python2.7/site-packages/distributed/deploy/local.pyRès cO`s„|jddƒs|jrj|jddƒ}|||Ž}|dk rftjtd|ƒ|ƒ}n|St|j|||ŽSdS(NRtcallback_timeouttseconds(tpopR&RRt with_timeoutRR R(RItfunctargstkwargsR\tfuture((s7lib/python2.7/site-packages/distributed/deploy/local.pyR ñs cK`sB|jjƒ|jr+|j||_n|j|j|dS(N(R9RFR0t_startRWR (RIRb((s7lib/python2.7/site-packages/distributed/deploy/local.pyRFûs  cc`sù|jdkrdS|jdkr.|j}nu|dkr^|jrUt|jƒ}q^d}nd|krs|}n |j|}|jr£|dt|jƒ7}n|jj|ƒgt |ƒD]}|j |j ^qÀVd|_t j |ƒ‚dS(s- Start all cluster services. trunningNs inproc://s 127.0.0.1s://t:(R*R.R&R1R RBtstrRARFtranget _start_workerRDRtReturn(RIR$R%taddressti((s7lib/python2.7/site-packages/distributed/deploy/local.pyRds"        * i<c k`s|jr9|jjdƒr9tjd|jddƒdS|jrOt|d>> c = LocalCluster() # doctest: +SKIP >>> c.start_worker(ncores=2) # doctest: +SKIP Returns ------- The created Worker or Nanny object. Can be discarded. (R Ri(RIRb((s7lib/python2.7/site-packages/distributed/deploy/local.pyt start_worker@scc`s1|jƒV||jkr-|jj|ƒndS(N(tcloseRCRw(RIRR((s7lib/python2.7/site-packages/distributed/deploy/local.pyt _stop_workerUs cC`s|j|j|ƒdS(sÝ 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 R{(RIRR((s7lib/python2.7/site-packages/distributed/deploy/local.pyt stop_worker[s t2sc c`sÖ|jdkrdSd|_ttjttƒ4tjtdt|ƒƒ|j j dt ƒƒVWdQXttjƒJtjtdt|ƒƒt g|j D]}|j|ƒ^qœƒƒVWdQX|j 2d|_dS(NRqtclosingR]t close_workers(R*R RRxRtOSErrorR_RRRARzRsR RCR{(RIttimeoutRR((s7lib/python2.7/site-packages/distributed/deploy/local.pyt_closefs 3ic`s§ˆjdkrdSyˆjˆjd|ƒ}Wntk rHd}nXtˆdƒrŠˆjrz|j‡fd†ƒqŠtˆj ƒnˆjs£ˆj j ƒn|S(s Close the cluster RqNR\R2c`s tˆjƒS(N(R R2(t_(RI(s7lib/python2.7/site-packages/distributed/deploy/local.pytˆt( R*R R‚t RuntimeErrorR&R[Rtadd_done_callbackR R2R9tstop(RIRtresult((RIs7lib/python2.7/site-packages/distributed/deploy/local.pyRz{s    c k`sŽtƒtj|j|ƒ}gt|t|jjƒƒD]}|j|^q<Vg|jD]}|j dkr`|^q`|_WdQXdS(sñ 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. RqN( RttoolztmergeRDRhRTRARCRiR*(RItnRbtkwargs2RlRR((s7lib/python2.7/site-packages/distributed/deploy/local.pytscale_up’s 4c#`s¡tƒ’g|jD]}|jdkr|^q|_tˆƒ‰td„ˆDƒƒrv‡fd†|jDƒ‰ngˆD]}|j|ƒ^q}VWdQXdS(sB 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. Rqcs`s|]}t|tƒVqdS(N(t isinstanceRg(RQRR((s7lib/python2.7/site-packages/distributed/deploy/local.pys µsc`s%h|]}|jˆkr|’qS((Ru(RQRR(RC(s7lib/python2.7/site-packages/distributed/deploy/local.pys ¶s N(RRCR*tsettallR{(RIRCRR((RCs7lib/python2.7/site-packages/distributed/deploy/local.pyt scale_down¥s . cC`s|jƒdS(N(Rz(RI((s7lib/python2.7/site-packages/distributed/deploy/local.pyt__del__»scC`s|S(N((RI((s7lib/python2.7/site-packages/distributed/deploy/local.pyt __enter__¾scG`s|jƒdS(N(Rz(RIRa((s7lib/python2.7/site-packages/distributed/deploy/local.pyt__exit__Áscc`s|jVtj|ƒ‚dS(N(RWRRj(RI((s7lib/python2.7/site-packages/distributed/deploy/local.pyt __aenter__Äscc`s|jƒVdS(N(R‚(RIttyptvaluet traceback((s7lib/python2.7/site-packages/distributed/deploy/local.pyt __aexit__ÉscC`s'y|jjSWntk r"dSXdS(Ns (RARkR'(RI((s7lib/python2.7/site-packages/distributed/deploy/local.pyRSÍs N( t__name__t __module__t__doc__R&RstloggingtWARNR:RPRVRXtpropertyRR RFRt coroutineRdRiRyR{R|R‚RzRŽR’R“R”R•R–RšRS(((s7lib/python2.7/site-packages/distributed/deploy/local.pyR#sVC b         c`sKˆdkrˆ}n"t‡fd†tˆƒDƒƒ}ˆ|}||fS(sB 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 ic3`s*|] }|tjˆƒkr|VqdS(N(R6tsqrt(RQtf(RŒ(s7lib/python2.7/site-packages/distributed/deploy/local.pys ìs(tminR(RŒR+tthreads((RŒs7lib/python2.7/site-packages/distributed/deploy/local.pyR3Õs   " cC`s+x$ttƒD]}|jddƒq WdS(NRi (tlistRGRz(tcluster((s7lib/python2.7/site-packages/distributed/deploy/local.pytclose_clustersôs(1t __future__RRRtatexittdatetimeRRžR6R(tweakrefRŠt dask.utilsRttornadoRR§Rt compatibilityRtcoreRtutilsR R R R R RRRRtnannyRRARtworkerRRRt getLoggerR›R>RR3tWeakSetRGtregisterR¨(((s7lib/python2.7/site-packages/distributed/deploy/local.pyts,      @ ÿ³