code: def __init__(self, plugin: ShuffleWorkerPlugin) -> None:
      self.closed = False
      self._active_runs = {}
      self._runs = set()
      self._stale_run_ids = {}
      self._runs_cleanup_condition = asyncio.Condition()
      self._plugin = plugin
  
imports: from __future__ import annotations
  import asyncio
  import logging
  from collections.abc import Sequence
  from concurrent.futures import ThreadPoolExecutor
  from typing import TYPE_CHECKING, Any, overload
  from dask.context import thread_state
  from dask.typing import Key
  from dask.utils import parse_bytes
  from distributed.diagnostics.plugin import WorkerPlugin
  from distributed.protocol.serialize import ToPickle
  from distributed.shuffle._core import NDIndex, ShuffleId, ShuffleRun, ShuffleRunSpec, ShuffleSpec
  from distributed.shuffle._exceptions import ShuffleClosedError
  from distributed.shuffle._limiter import ResourceLimiter
  from distributed.utils import log_errors, sync
label_desc: Use image node_id 1 to create a new _ShuffleRunManager object  with example: obj = _ShuffleRunManager(plugin)