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: Iterate through self.pending, which was filled by self._run_policies(), and
  push the suggestions to the workers through bulk comms. Return immediately.
docstr_tok: ["Iterate","through","self.pending",",","which","was","filled","by","self._run_policies","(",")",",","and","push","the","suggestions","to","the","workers","through","bulk","comms",".","Return","immediately","."]
code: def _enact_suggestions(self) -> None:
      """Iterate through self.pending, which was filled by self._run_policies(), and
      push the suggestions to the workers through bulk comms. Return immediately.
      """
      logger.debug(
          "Enacting suggestions for %d tasks:", len(self.pending)
      )
  
      validate = self.scheduler.validate
      drop_by_worker: (
          defaultdict[scheduler_module.WorkerState, list[Key]]
      ) = defaultdict(list)
      repl_by_worker: (
          defaultdict[scheduler_module.WorkerState, list[Key]]
      ) = defaultdict(list)
  
      for ts, (pending_repl, pending_drop) in self.pending.items():
          if not ts.who_has:
              continue
          if validate:
              # Never drop the last replica
              assert ts.who_has - pending_drop
  
          for ws in pending_repl:
              if validate:
                  assert ws not in (ts.who_has or ())
              repl_by_worker[ws].append(ts.key)
          for ws in pending_drop:
              if validate:
                  assert ws in (ts.who_has or ())
              drop_by_worker[ws].append(ts.key)
  
      stimulus_id = f"active_memory_manager-{time()}"
      for ws, keys in repl_by_worker.items():
          logger.debug("- %s to acquire %d replicas", ws, len(keys))
          self.scheduler.request_acquire_replicas(
              ws.address, keys, stimulus_id=stimulus_id
          )
      for ws, keys in drop_by_worker.items():
          logger.debug("- %s to drop %d replicas", ws, len(keys))
          self.scheduler.request_remove_replicas(
              ws.address, keys, stimulus_id=stimulus_id
          )
  
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 11 for calling the ActiveMemoryManagerExtension obj's underlying member method code with example usage: obj._enact_suggestions() without return types