class_docstr: 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``.
class_docstr_tok: ["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","``","."]
docstr: Choose 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.
docstr_tok: ["Choose","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","."]
code: def _find_dropper(
      self,
      ts: scheduler_module.TaskState,
      candidates: set[scheduler_module.WorkerState] | None,
      pending_drop: set[scheduler_module.WorkerState],
  ) -> scheduler_module.WorkerState | None:
      """Choose 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.
      """
      if ts.who_has is None:
          return None
  
      orig_candidates = candidates
  
      def log_reject(msg: str) -> None:
          task_logger.debug(
              "(drop, %s, %s) rejected: %s", ts, orig_candidates, msg
          )
  
      if len(ts.who_has) - len(pending_drop) < 2:
          log_reject("less than 2 replicas exist")
          return None
  
      if ts.actor:
          log_reject("task is an actor")
          return None
  
      if candidates is None:
          candidates = ts.who_has.copy()
      else:
          # Don't modify orig_candidates
          candidates = candidates & ts.who_has
          if not candidates:
              log_reject(
                  "no candidates suggested by the policy own a replica"
              )
              return None
  
      candidates -= pending_drop
      if not candidates:
          log_reject("already pending drop on all candidates")
          return None
  
      # The `candidates &` bit could seem redundant with `candidates -=` immediately
      # below on first look, but beware of the second use of this variable later on!
      candidates_with_dependents_processing = candidates & {
          waiter_ts.processing_on for waiter_ts in ts.waiters or ()
      }
  
      candidates -= candidates_with_dependents_processing
      if not candidates:
          log_reject(
              "all candidates have dependent tasks queued or running on them"
          )
          return None
  
      # Select candidate with the highest memory usage.
      # Drop from workers with status paused or closing_gracefully first.
      choice = max(
          candidates,
          key=lambda ws: (
              ws.status != Status.running,
              self.workers_memory[ws],
          ),
      )
  
      # IF there is only one candidate that could drop the key
      # AND the candidate has status=running
      # AND there were candidates with status=paused or closing_gracefully, but we
      # discarded them above because they have dependent tasks running on them,
      # THEN temporarily keep the extra replica on the candidate with status=running.
      #
      # This prevents a ping-pong effect between ReduceReplicas (or any other policy
      # that yields drop commands with multiple candidates) and RetireWorker:
      # 1. RetireWorker replicates in-memory tasks from worker A (very busy and being
      #    retired) to worker B (idle)
      # 2. on the next AMM iteration 2 seconds later, ReduceReplicas drops the same
      #    tasks from B (because the replicas on A have dependants on the same worker)
      # 3. on the third AMM iteration 2 seconds later, goto 1 in an infinite loop
      #    which will last for as long as any tasks with dependencies are running on A
      if (
          len(candidates) == 1
          and choice.status == Status.running
          and candidates_with_dependents_processing
          and all(
              ws.status != Status.running
              for ws in candidates_with_dependents_processing
          )
      ):
          log_reject(
              "there is only one replica on workers that aren't paused or retiring"
          )
          return None
  
      task_logger.debug(
          "(drop, %s, %s): dropping from %s",
          ts,
          orig_candidates,
          choice,
      )
      return choice
  
imports: from __future__ import annotations
  import abc
  import logging
  from collections import defaultdict
  from collections.abc import Generator
  from typing import TYPE_CHECKING, Any, Literal, NamedTuple, Union
  import dask
  from dask.typing import Key
  from dask.utils import parse_timedelta
  from distributed import client
  from distributed import scheduler
  from distributed.compatibility import PeriodicCallback
  from distributed.core import Status
  from distributed.metrics import time
  from distributed.utils import import_term, log_errors
label_desc: Use image node_id 10 for calling the ActiveMemoryManagerExtension obj's underlying member method code with example usage: obj._find_dropper(ts, candidates, pending_drop) and returns: choice, None, None, None, None, None, None, None