σ ίόΪ\c@`s΅ddlmZmZmZddlmZddlZddlZddlZddl m Z ddl m Z ddl mZmZmZejeƒZdefd „ƒYZdS( i(tprint_functiontdivisiontabsolute_import(tdequeN(tgeni(ttime(t log_errorstPeriodicCallbacktparse_timedeltatAdaptivec B`s§eZdZddddddddd„d„ Zd„Zd „Zd „Zd „Zd „Z e j dd „ƒZ d„Z dd„Ze j d„ƒZd„ZRS(sF Adaptively allocate workers based on scheduler load. A superclass. Contains logic to dynamically resize a Dask cluster based on current use. This class needs to be paired with a system that can create and destroy Dask workers using a cluster resource manager. Typically it is built into already existing solutions, rather than used directly by users. It is most commonly used from the ``.adapt(...)`` method of various Dask cluster classes. Parameters ---------- scheduler: distributed.Scheduler cluster: object Must have scale_up and scale_down methods/coroutines startup_cost : timedelta or str, default "1s" Estimate of the number of seconds for nnFactor representing how costly it is to start an additional worker. Affects quickly to adapt to high tasks per worker loads interval : timedelta or str, default "1000 ms" Milliseconds between checks wait_count: int, default 3 Number of consecutive times that a worker should be suggested for removal before we remove it. scale_factor : int, default 2 Factor to scale by when it's determined additional workers are needed target_duration: timedelta or str, default "5s" Amount of time we want a computation to take. This affects how aggressively we scale up. worker_key: Callable[WorkerState] Function to group workers together when scaling down See Scheduler.workers_to_close for more information minimum: int Minimum number of workers to keep around maximum: int Maximum number of workers to keep around **kwargs: Extra parameters to pass to Scheduler.workers_to_close Examples -------- This is commonly used from existing Dask classes, like KubeCluster >>> from dask_kubernetes import KubeCluster >>> cluster = KubeCluster() >>> cluster.adapt(minimum=10, maximum=100) Alternatively you can use it from your own Cluster class by subclassing from Dask's Cluster superclass >>> from distributed.deploy import Cluster >>> class MyCluster(Cluster): ... def scale_up(self, n): ... """ Bring worker count up to n """ ... def scale_down(self, workers): ... """ Remove worker addresses from cluster """ >>> cluster = MyCluster() >>> cluster.adapt(minimum=10, maximum=100) Notes ----- Subclasses can override :meth:`Adaptive.should_scale_up` and :meth:`Adaptive.workers_to_close` to control when the cluster should be resized. The default implementation checks if there are too many tasks per worker or too little memory available (see :meth:`Adaptive.needs_cpu` and :meth:`Adaptive.needs_memory`). :meth:`Adaptive.get_scale_up_kwargs` method controls the arguments passed to the cluster's ``scale_up`` method. t1siiit5scC`s|S(N((tx((s:lib/python2.7/site-packages/distributed/deploy/adaptive.pytdtc K`st|ddƒ}| |_||_||_t|ddƒ|_||_|jr’t|j|dd|jƒ|_ |jjj |j j ƒnt |_ | |_||_||_tddƒ|_i|_||_t| ƒ|_|j|jjdis,CPU limit exceeded [%d occupancy / %d cores]is:pending tasks exceed number of cores [%d tasks / %d cores]( Rttotal_occupancyt total_ncoresRtloggertinfotworkerstvaluestlent processingtTrueR(R*R0t total_coresttasks_processingtw((s:lib/python2.7/site-packages/distributed/deploy/adaptive.pyt needs_cpuƒs"    cC`s‘d„|jjjƒDƒ}g|jjjƒD]}|j^q/}t|jƒƒ}t|ƒ}|d|kr‰tjd||ƒtSt SdS(sθ Check if the cluster is RAM constrained Notes ----- Returns ``True`` if the required bytes in distributed memory is some factor larger than the actual distributed memory available. cS`s"i|]\}}|j|“qS((t memory_limit(t.0taddrtws((s:lib/python2.7/site-packages/distributed/deploy/adaptive.pys ±s g333333γ?sRam limit exceeded [%d/%d]N( RR4titemsR5tnbytestsumR2R3R8R(R*t limit_bytesR@t worker_bytestlimitttotal((s:lib/python2.7/site-packages/distributed/deploy/adaptive.pyt needs_memory§s ( cC`sͺtƒ›t|jjƒ|jkr)tS|jdk rWt|jjƒ|jkrWtS|jj rt|jj rttS|j ƒ}|j ƒ}|s˜|rœtStSWdQXdS(sΰ Determine whether additional workers should be added to the cluster Returns ------- scale_up : bool Notes ---- Additional workers are added whenever 1. There are unrunnable tasks and no workers 2. The cluster is CPU constrained 3. The cluster is RAM constrained 4. There are fewer workers than our minimum See Also -------- needs_cpu needs_memory N( RR6RR4R"R8R#R/Rt unrunnableR<RH(R*R<RH((s:lib/python2.7/site-packages/distributed/deploy/adaptive.pytshould_scale_up½s *   cK`sΰt|jjƒ|jkrgSt|jƒ}|j|ƒ|jdk r…t|jjƒ|jkr…t|jjƒ|j|ds