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: Scheduler handler, invoked from the Client by
  :class:`~distributed.active_memory_manager.AMMClientProxy`
docstr_tok: ["Scheduler","handler",",","invoked","from","the","Client","by",":","class",":","`","~distributed.active_memory_manager.AMMClientProxy","`"]
code: def amm_handler(self, method: str) -> Any:
      """Scheduler handler, invoked from the Client by
      :class:`~distributed.active_memory_manager.AMMClientProxy`
      """
      assert method in {"start", "stop", "run_once", "running"}
      out = getattr(self, method)
      return out() if callable(out) else out
  
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 2 for calling the ActiveMemoryManagerExtension obj's underlying member method code with example usage: obj.amm_handler(method) and returns: unknown