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","``","."]
code: def __init__(
      self,
      scheduler: scheduler_module.Scheduler,
      # The following parameters are exposed so that one may create, run, and throw
      # away on the fly a specialized manager, separate from the main one.
      policies: set[ActiveMemoryManagerPolicy] | None = None,
      *,
      measure: str | None = None,
      register: bool = True,
      start: bool | None = None,
      interval: float | None = None,
  ):
      self.scheduler = scheduler
      self.policies = set()
  
      if policies is None:
          # Initialize policies from config
          policies = set()
          for kwargs in dask.config.get(
              "distributed.scheduler.active-memory-manager.policies"
          ):
              kwargs = kwargs.copy()
              cls = import_term(kwargs.pop("class"))
              policies.add(cls(**kwargs))
  
      for policy in policies:
          self.add_policy(policy)
  
      if not measure:
          measure = dask.config.get(
              "distributed.scheduler.active-memory-manager.measure"
          )
      mem = scheduler.memory
      measure_domain = {
          name
          for name in dir(mem)
          if not name.startswith("_") and name != "sum"
      }
      if not isinstance(measure, str) or measure not in measure_domain:
          raise ValueError(
              "distributed.scheduler.active-memory-manager.measure "
              "must be one of " + ", ".join(sorted(measure_domain))
          )
      self.measure = measure
  
      if register:
          scheduler.handlers["amm_handler"] = self.amm_handler
  
      if interval is None:
          interval = parse_timedelta(
              dask.config.get(
                  "distributed.scheduler.active-memory-manager.interval"
              )
          )
      self.interval = interval
  
      if start is None:
          start = dask.config.get(
              "distributed.scheduler.active-memory-manager.start"
          )
      if start:
          self.start()
  
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 1 to create a new ActiveMemoryManagerExtension object  with example: obj = ActiveMemoryManagerExtension(scheduler, policies)