U /e"p@s`UdZddlmZddlZddlZddlmZddlmZddl m Z m Z m Z m Z ddlZddlmZddlmZdd lmZdd lmZdd lmZmZe rdd lmZdd lmZmZmZe e!Z"e e!dZ#GdddZ$Gddde Z%e rddl&m'Z'ee%ddfZ(de)d<Gdddej*Z+GdddZ,Gddde+Z-Gddde+Z.dS)aImplementation of the Active Memory Manager. This is a scheduler extension which sends drop/replicate suggestions to the worker. See also :mod:`distributed.worker_memory` and :mod:`distributed.spill`, which implement spill/pause/terminate mechanics on the Worker side. ) annotationsN) defaultdict) Generator) TYPE_CHECKINGAnyLiteral NamedTuple)parse_timedelta)PeriodicCallback)Status)time) import_term log_errors)Client) Scheduler TaskState WorkerStatez.tasksc @s eZdZUdZded<ded<ded<ded <d ed <d ed <d9ddddddddddddddZdddddZdddd Zddd!d"Ze ddd#d$Z d%dd&d'd(Z e ddd)d*Z ddd+d,Zd-d.d/d0d1d2d3Zd-d.d/d0d4d5d6Zddd7d8ZdS):ActiveMemoryManagerExtensiona}Scheduler extension that optimizes memory usage across the cluster. It can be either triggered by hand or automatically every few seconds; at every iteration it performs one or both of the following: - create new replicas of in-memory tasks - destroy replicas of in-memory tasks; this never destroys the last available copy. There are no 'move' operations. A move is performed in two passes: first you create a copy and, in the next iteration, you delete the original (if the copy succeeded). This extension is configured by the dask config section ``distributed.scheduler.active-memory-manager``. r schedulerzset[ActiveMemoryManagerPolicy]policiesstrmeasurefloatintervalzdict[WorkerState, int]workers_memoryz:dict[TaskState, tuple[set[WorkerState], set[WorkerState]]]pendingNT)rregisterstartrz%set[ActiveMemoryManagerPolicy] | Nonez str | Noneboolz bool | Nonez float | None)rrrrrrc Cs||_t|_|dkrTt}tjdD]*}|}t|d}| |f|q(|D]} | | qX|sxtjd}|j } ddt | D} t |tr|| krtddt| ||_|r||jd<|j|jd <|dkrttjd }||_|dkr tjd }|r|dS) Nz4distributed.scheduler.active-memory-manager.policiesclassz3distributed.scheduler.active-memory-manager.measurecSs"h|]}|ds|dkr|qS)_sum) startswith).0namer%E/tmp/pip-unpacked-wheel-g426oqom/distributed/active_memory_manager.py as z8ActiveMemoryManagerExtension.__init__..zCdistributed.scheduler.active-memory-manager.measure must be one of z, Zamm amm_handlerz4distributed.scheduler.active-memory-manager.intervalz1distributed.scheduler.active-memory-manager.start)rsetrdaskconfiggetcopyr popadd add_policymemorydir isinstancer ValueErrorjoinsortedr extensionsr(handlersr rr) selfrrrrrrkwargsclspolicyZmemZmeasure_domainr%r%r&__init__@sN        z%ActiveMemoryManagerExtension.__init__rmethodreturncCs(|dks tt||}t|r$|S|S)zyScheduler handler, invoked from the Client by :class:`~distributed.active_memory_manager.AMMClientProxy` >rstoprun_oncerunning)AssertionErrorgetattrcallable)r9r?outr%r%r&r(zs  z(ActiveMemoryManagerExtension.amm_handlerNoner@cCs>|jr dSt|j|jd}||jjdt|<|dS)zHStart executing every ``self.interval`` seconds until scheduler shutdownNg@@amm-)rCr rBrrperiodic_callbacksidrr9Zpcr%r%r&rs z"ActiveMemoryManagerExtension.startcCs*|jjdt|d}|r&|dS)zStop periodic executionrJN)rrKr.rLrArMr%r%r&rAsz!ActiveMemoryManagerExtension.stopcCsdt||jjkS)zGReturn True if the AMM is being triggered periodically; False otherwiserJ)rLrrKr9r%r%r&rCsz$ActiveMemoryManagerExtension.runningActiveMemoryManagerPolicy)r<r@cCs.t|tstd||j|||_dS)Nz(Expected ActiveMemoryManagerPolicy; got )r3rO TypeErrorrr/manager)r9r<r%r%r&r0s  z'ActiveMemoryManagerExtension.add_policycst}t|drti|_|jfdd|jjD|_z| |jrT| W5|`|`Xt}t d||ddS)zRun all policies once and asynchronously (fire and forget) enact their recommendations to replicate/drop tasks rcsi|]}|t|jqSr%)rEr1r#wsrr%r& sz9ActiveMemoryManagerExtension.run_once..z#Active Memory Manager run in %.0fmsiN) r hasattrrDrrrworkersvaluesr _run_policies_enact_suggestionsloggerdebug)r9Zts_startZts_stopr%rTr&rBs   z%ActiveMemoryManagerExtension.run_oncec CsPt|jD]>}td||}d}z||}Wntk rNYq YnXt|tsbt|}z|j |j \}}Wn0t k rt }t }||f|j |j <YnX|j dkr||j |j|}|r|||j||j j7<q(|j dkr8||j |j|}|rH||td|j||j j|j|<q(td|j q(q dS)zSequentially run ActiveMemoryManagerPolicy.run() for all registered policies, obtain replicate/drop suggestions, and use them to populate self.pending. zRunning policy: %sN replicatedroprz Unknown op: )listrr[r\runsend StopIterationr3 SuggestionrtsKeyErrorr)op_find_recipient candidatesr/rnbytes _find_droppermaxr4)r9r<Z policy_genrSZ suggestion pending_repl pending_dropr%r%r&rYsL        z*ActiveMemoryManagerExtension._run_policiesrset[WorkerState] | Nonezset[WorkerState]WorkerState | None)rdrhrlr@cs|dddfdd }jdkr8|djdSjrJ|d dS|dkr`|jj}n ||jj@}|s||d dS|j8}|s|d dS||8}|s|d dSt||jjd }t d||S)aChoose a worker to acquire a new replica of an in-memory task among a set of candidates. If candidates is None, default to all workers in the cluster. Regardless, workers that either already hold a replica or are scheduled to receive one at the end of this AMM iteration are not considered. Returns ------- The worker with the lowest memory usage (downstream of pending replications and drops), or None if no eligible candidates are available. rrHmsgr@cstd|dS)Nz (replicate, %s, %s) rejected: %s task_loggerr\rqorig_candidatesrdr%r& log_rejects z@ActiveMemoryManagerExtension._find_recipient..log_rejectr1z ts.state = Ntask is an actorzno running candidatesz$all candidates already own a replicaz-already pending replication on all candidateskeyz&(replicate, %s, %s): replicating to %s) stateactorrrCr-who_hasminr __getitem__rsr\)r9rdrhrlrwchoicer%rur&rgs<   z,ActiveMemoryManagerExtension._find_recipient)rdrhrmr@cs*|dddfdd }tjt|dkr<|ddSjrN|d dS|dkrbj}n|j@}|s||d dS||8}|s|d dS|d d jD@}||8}|s|ddSt|fddd}t|dkr|jtjkr|rt dd|Dr|ddSt d||S)aLChoose a worker to drop its replica of an in-memory task among a set of candidates. If candidates is None, default to all workers in the cluster. Regardless, workers that either do not hold a replica or are already scheduled to drop theirs at the end of this AMM iteration are not considered. This method also ensures that a key will not lose its last replica. Returns ------- The worker with the highest memory usage (downstream of pending replications and drops), or None if no eligible candidates are available. rrHrpcstd|dS)Nz(drop, %s, %s) rejected: %srrrtrur%r&rw1sz>ActiveMemoryManagerExtension._find_dropper..log_rejectzless than 2 replicas existNrxz3no candidates suggested by the policy own a replicaz&already pending drop on all candidatescSsh|] }|jqSr%Z processing_on)r#Z waiter_tsr%r%r&r'Lsz=ActiveMemoryManagerExtension._find_dropper..z=all candidates have dependent tasks queued or running on themcs|jtjkj|fSN)statusr rCr)rSrNr%r&Yz.rycss|]}|jtjkVqdSr)rr rCrRr%r%r& nsz=ActiveMemoryManagerExtension._find_dropper..zCthere is only one replica on workers that aren't paused or retiringz (drop, %s, %s): dropping from %s) lenr}r|r-waitersrkrr rCallrsr\)r9rdrhrmrwZ%candidates_with_dependents_processingrr%)rvr9rdr&rjsd      z*ActiveMemoryManagerExtension._find_dropperc Cs6tdt|j|jj}tt}tt}|jD]~\}\}}|j sHq4|rZ|j |sZt |D]&}|rt||j kstt || |j q^|D]&}|r||j kst || |j qq4dt }|D].\}} td|t| |jj|j| |dq|D]0\}} td|t| |jj|j| |dqdS)zIterate through self.pending, which was filled by self._run_policies(), and push the suggestions to the workers through bulk comms. Return immediately. z"Enacting suggestions for %d tasks:zactive_memory_manager-z- %s to acquire %d replicas) stimulus_idz- %s to drop %d replicasN)r[r\rrrvalidaterr_itemsr}rDappendrzr Zrequest_acquire_replicasaddressZrequest_remove_replicas) r9rZdrop_by_workerZrepl_by_workerrdrlrmrSrkeysr%r%r&rZ}s@ z/ActiveMemoryManagerExtension._enact_suggestions)N)__name__ __module__ __qualname____doc____annotations__r=r(rrApropertyrCr0rrBrYrgrjrZr%r%r%r&r!s4 :09_rc@s*eZdZUded<ded<dZded<dS)rczLiteral[('replicate', 'drop')]rfrrdNrnrh)rrrrrhr%r%r%r&rcs rc) TypeAliasrorSuggestionGeneratorc@s@eZdZUdZdZded<ddddZejd dd d Z d S) rOzAbstract parent class)rQrrQrrIcCs|jjdS)Nz()) __class__rrNr%r%r&__repr__sz"ActiveMemoryManagerPolicy.__repr__rcCsdS)aoThis method is invoked by the ActiveMemoryManager every few seconds, or whenever the user invokes ``client.amm.run_once``. It is an iterator that must emit :class:`~distributed.active_memory_manager.Suggestion` objects: - ``Suggestion("replicate", )`` - ``Suggestion("replicate", , {subset of potential workers to replicate to})`` - ``Suggeston("drop", )`` - ``Suggestion("drop", , {subset of potential workers to drop from})`` Each element yielded indicates the desire to create or destroy a single replica of a key. If a subset of workers is not provided, it defaults to all workers on the cluster. Either the ActiveMemoryManager or the Worker may later decide to disregard the request, e.g. because it would delete the last copy of a key or because the key is currently needed on that worker. You may optionally retrieve which worker it was decided the key will be replicated to or dropped from, as follows: .. code-block:: python choice = (yield Suggestion("replicate", ts)) ``choice`` is either a WorkerState or None; the latter is returned if the ActiveMemoryManager chose to disregard the request. The current pending (accepted) suggestions can be inspected on ``self.manager.pending``; this includes the suggestions previously yielded by this same method. The current memory usage on each worker, *downstream of all pending suggestions*, can be inspected on ``self.manager.workers_memory``. Nr%rNr%r%r&r`szActiveMemoryManagerPolicy.runN) rrrr __slots__rrabcabstractmethodr`r%r%r%r&rOs rOc@speZdZUdZded<ddddZddd d d Zdd d dZdd ddZdd ddZ dd ddZ dS)AMMClientProxyzConvenience accessors to operate the AMM from the dask client Usage: ``client.amm.start()`` etc. All methods are asynchronous if the client is asynchronous and synchronous if the client is synchronous. r_client)clientcCs ||_dSr)r)r9rr%r%r&r=szAMMClientProxy.__init__rrr>cCs|jj|jjj|dS)z8Remotely invoke ActiveMemoryManagerExtension.amm_handler)r?)rsyncrr()r9r?r%r%r&_runszAMMClientProxy._runrIcCs |dS)NrrrNr%r%r&rszAMMClientProxy.startcCs |dS)NrArrNr%r%r&rAszAMMClientProxy.stopcCs |dS)NrBrrNr%r%r&rBszAMMClientProxy.run_oncecCs |dS)NrCrrNr%r%r&rCszAMMClientProxy.runningN) rrrrrr=rrrArBrCr%r%r%r&rs rc@seZdZdZddddZdS)ReduceReplicaszrMake sure that in-memory tasks are not replicated on more workers than desired; drop the excess replicas. rrIc csd}d}|jjjD]}d}tdd|jD}t|jt||}||jjkrr|jj|\}}|t|t|7}|dkr|d7}||7}t|D]} t d|Vqq|rt d||dS)NrrcSsh|]}|jp|qSr%r)r#waiterr%r%r&r'sz%ReduceReplicas.run..r^z.)rQrrrWr,rrr)r9rSr%r%r&dones  zRetireWorker.doneN) rrrrrrr=rr`rr%r%r%r&r(s 4\r)/r __future__rrlogging collectionsrcollections.abcrtypingrrrrr*Z dask.utilsr Zdistributed.compatibilityr Zdistributed.corer Zdistributed.metricsr Zdistributed.utilsr rZdistributed.clientrZdistributed.schedulerrrr getLoggerrr[rsrrcZtyping_extensionsrrrABCrOrrrr%r%r%r&s:          2%