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: Sequentially run ActiveMemoryManagerPolicy.run() for all registered policies,
  obtain replicate/drop suggestions, and use them to populate self.pending.
docstr_tok: ["Sequentially","run","ActiveMemoryManagerPolicy.run","(",")","for","all","registered","policies",",","obtain","replicate\/drop","suggestions",",","and","use","them","to","populate","self.pending","."]
code: def _run_policies(self) -> None:
      """Sequentially run ActiveMemoryManagerPolicy.run() for all registered policies,
      obtain replicate/drop suggestions, and use them to populate self.pending.
      """
      ws: scheduler_module.WorkerState | None
  
      for policy in list(self.policies):  # a policy may remove itself
          logger.debug("Running policy: %s", policy)
          policy_gen = policy.run()
          ws = None
          while True:
              try:
                  suggestion = policy_gen.send(ws)
              except StopIteration:
                  break  # next policy
  
              if not isinstance(suggestion, Suggestion):
                  # legacy: accept plain tuples
                  suggestion = Suggestion(*suggestion)  # type: ignore[unreachable]
  
              try:
                  pending_repl, pending_drop = self.pending[
                      suggestion.ts
                  ]
              except KeyError:
                  pending_repl = set()
                  pending_drop = set()
                  self.pending[suggestion.ts] = (
                      pending_repl,
                      pending_drop,
                  )
  
              if suggestion.op == "replicate":
                  ws = self._find_recipient(
                      suggestion.ts, suggestion.candidates, pending_repl
                  )
                  if ws:
                      pending_repl.add(ws)
                      self.workers_memory[ws] += suggestion.ts.nbytes
  
              elif suggestion.op == "drop":
                  ws = self._find_dropper(
                      suggestion.ts, suggestion.candidates, pending_drop
                  )
                  if ws:
                      pending_drop.add(ws)
                      self.workers_memory[ws] = max(
                          0,
                          self.workers_memory[ws]
                          - suggestion.ts.nbytes,
                      )
  
              else:
                  raise ValueError(f"Unknown op: {suggestion.op}")
  
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 8 for calling the ActiveMemoryManagerExtension obj's underlying member method code with example usage: obj._run_policies() without return types