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 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.
docstr_tok: ["Choose","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","."]
code: def _find_recipient(
      self,
      ts: scheduler_module.TaskState,
      candidates: set[scheduler_module.WorkerState] | None,
      pending_repl: set[scheduler_module.WorkerState],
  ) -> scheduler_module.WorkerState | None:
      """Choose 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.
      """
      orig_candidates = candidates
  
      def log_reject(msg: str) -> None:
          task_logger.debug(
              "(replicate, %s, %s) rejected: %s",
              ts,
              orig_candidates,
              msg,
          )
  
      if ts.state != "memory":
          log_reject(f"ts.state = {ts.state}")
          return None
      assert ts.who_has
      if ts.actor:
          log_reject("task is an actor")
          return None
  
      if candidates is None:
          candidates = self.scheduler.running.copy()
      else:
          # Don't modify orig_candidates
          candidates = candidates & self.scheduler.running
      if not candidates:
          log_reject("no running candidates")
          return None
  
      candidates -= ts.who_has
      if not candidates:
          log_reject("all candidates already own a replica")
          return None
  
      candidates -= pending_repl
      if not candidates:
          log_reject("already pending replication on all candidates")
          return None
  
      # Select candidate with the lowest memory usage
      choice = min(candidates, key=self.workers_memory.__getitem__)
      task_logger.debug(
          "(replicate, %s, %s): replicating to %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 9 for calling the ActiveMemoryManagerExtension obj's underlying member method code with example usage: obj._find_recipient(ts, candidates, pending_repl) and returns: choice, None, None, None, None, None