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: Run all policies once and asynchronously (fire and forget) enact their
  recommendations to replicate/drop tasks
docstr_tok: ["Run","all","policies","once","and","asynchronously","(","fire","and","forget",")","enact","their","recommendations","to","replicate\/drop","tasks"]
code: def run_once(self) -> None:
      """Run all policies once and asynchronously (fire and forget) enact their
      recommendations to replicate/drop tasks
      """
      ts_start = time()
      # This should never fail since this is a synchronous method
      assert not hasattr(self, "pending")
  
      self.pending = {}
      measure = self.measure
      self.workers_memory = {
          ws: getattr(ws.memory, measure)
          for ws in self.scheduler.workers.values()
      }
      try:
          # populate self.pending
          self._run_policies()
  
          if self.pending:
              self._enact_suggestions()
      finally:
          del self.workers_memory
          del self.pending
      ts_stop = time()
      logger.debug(
          "Active Memory Manager run in %.0fms",
          (ts_stop - ts_start) * 1000,
      )
  
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 7 for calling the ActiveMemoryManagerExtension obj's underlying member method code with example usage: obj.run_once() without return types