docstr: Track worker's memory. Restart if it goes above terminate fraction.
docstr_tok: ["Track","worker","'s","memory",".","Restart","if","it","goes","above","terminate","fraction","."]
code: def memory_monitor(self, nanny: Nanny) -> None:
      """Track worker's memory. Restart if it goes above terminate fraction."""
      if (
          nanny.status != Status.running
          or nanny.process is None
          or nanny.process.process is None
          or nanny.process.process.pid is None
      ):
          return  # pragma: nocover
  
      process = nanny.process.process
      try:
          memory = psutil.Process(process.pid).memory_info().rss
      except (
          ProcessLookupError,
          psutil.NoSuchProcess,
          psutil.AccessDenied,
      ):
          return  # pragma: nocover
  
      if memory / self.memory_limit <= self.memory_terminate_fraction:
          return
  
      if self._last_terminated_pid != process.pid:
          nanny_logger.warning(
              f"Worker {nanny.worker_address} (pid={process.pid}) exceeded "
              f"{self.memory_terminate_fraction * 100:.0f}% memory budget. "
              "Restarting...",
          )
          self._last_terminated_pid = process.pid
          process.terminate()
      else:
          # We already sent SIGTERM to the worker, but the process is still alive
          # since the previous iteration of the memory_monitor - for example, some
          # user code may have tampered with signal handlers.
          # Send SIGKILL for immediate termination.
          #
          # Note that this should not be a disk-related issue. Unlike in a regular
          # worker shutdown, where the worker cleans up its own spill directory, in
          # case of SIGTERM no atexit or weakref.finalize callback is triggered
          # whatsoever; instead, the nanny cleans up the spill directory *after* the
          # worker has been shut down and before starting a new one.
          # This is important, as spill directory cleanup may potentially take tens of
          # seconds and, if the worker did it, any task that was running and leaking
          # would continue to do so for the whole duration of the cleanup, increasing
          # the risk of going beyond 100%.
          nanny_logger.warning(
              f"Worker {nanny.worker_address} (pid={process.pid}) is slow to %s",
              # On Windows, kill() is an alias to terminate()
              "terminate; trying again"
              if WINDOWS
              else "accept SIGTERM; sending SIGKILL",
          )
          process.kill()
  
imports: from __future__ import annotations
  import asyncio
  import logging
  import os
  import sys
  import warnings
  from collections.abc import Callable, Container, Hashable, MutableMapping
  from contextlib import suppress
  from functools import partial
  from typing import TYPE_CHECKING, Any, Literal, Union, cast
  import psutil
  import dask.config
  from dask.system import CPU_COUNT
  from dask.typing import Key
  from dask.utils import format_bytes, parse_bytes, parse_timedelta
  from distributed import system
  from distributed.compatibility import WINDOWS, PeriodicCallback
  from distributed.core import Status
  from distributed.metrics import context_meter, monotonic
  from distributed.spill import ManualEvictProto, SpillBuffer
  from distributed.utils import RateLimiterFilter, has_arg, log_errors
  from distributed.utils_perf import ThrottledGC
label_desc: Use image node_id 2 for calling the NannyMemoryManager obj's underlying member method code with example usage: obj.memory_monitor(nanny) without return types