class_docstr: Management of worker memory usage
  
  Parameters
  ----------
  worker
      Worker to manage
  
  For meaning of the remaining parameters, see the matching
  parameter names in :class:`~.distributed.worker.Worker`.
  
  Notes
  -----
  
  If data is a callable and has the argument ``worker_local_directory`` in its
  signature, it will be filled with the worker's attr:``local_directory``.
code: def _maybe_pause_or_unpause(
      self, worker: Worker, memory: int
  ) -> None:
      if self.memory_pause_fraction is False:
          return
  
      assert self.memory_limit
      frac = memory / self.memory_limit
      # Pause worker threads if above 80% memory use
      if frac > self.memory_pause_fraction:
          # Try to free some memory while in paused state
          self._throttled_gc.collect()
          if worker.status == Status.running:
              worker_logger.warning(
                  "Worker is at %d%% memory usage. Pausing worker.  "
                  "Process memory: %s -- Worker memory limit: %s",
                  int(frac * 100),
                  format_bytes(memory),
                  format_bytes(self.memory_limit)
                  if self.memory_limit is not None
                  else "None",
              )
              worker.status = Status.paused
      elif worker.status == Status.paused:
          worker_logger.warning(
              "Worker is at %d%% memory usage. Resuming worker. "
              "Process memory: %s -- Worker memory limit: %s",
              int(frac * 100),
              format_bytes(memory),
              format_bytes(self.memory_limit)
              if self.memory_limit is not None
              else "None",
          )
          worker.status = Status.running
  
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 WorkerMemoryManager obj's underlying member method code with example usage: obj._maybe_pause_or_unpause(worker, memory) without return types