class_docstr: Shuffle plugin for the scheduler
  This coordinates the individual worker plugins to ensure correctness
  and collects heartbeat messages for the dashboard.
  See Also
  --------
  ShuffleWorkerPlugin
class_docstr_tok: ["Shuffle","plugin","for","the","scheduler","This","coordinates","the","individual","worker","plugins","to","ensure","correctness","and","collects","heartbeat","messages","for","the","dashboard",".","See","Also","--","--","--","--","ShuffleWorkerPlugin"]
code: def __init__(self, scheduler: Scheduler):
      self.scheduler = scheduler
      self.scheduler.handlers.update(
          {
              "shuffle_barrier": self.barrier,
              "shuffle_get": self.get,
              "shuffle_get_or_create": self.get_or_create,
              "shuffle_restrict_task": self.restrict_task,
          }
      )
      self.heartbeats = defaultdict(lambda: defaultdict(dict))
      self.active_shuffles = {}
      self.scheduler.add_plugin(self, name="shuffle")
      self._shuffles = defaultdict(set)
      self._archived_by_stimulus = defaultdict(set)
      self._shift_counter = itertools.count()
  
imports: from __future__ import annotations
  import contextlib
  import itertools
  import logging
  from collections import defaultdict
  from typing import TYPE_CHECKING, Any
  from dask.typing import Key
  from distributed.diagnostics.plugin import SchedulerPlugin
  from distributed.metrics import time
  from distributed.protocol.pickle import dumps
  from distributed.protocol.serialize import ToPickle
  from distributed.shuffle._core import SchedulerShuffleState, ShuffleId, ShuffleRunSpec, ShuffleSpec, barrier_key, id_from_key
  from distributed.shuffle._worker_plugin import ShuffleWorkerPlugin
  from distributed.utils import log_errors
label_desc: Use image node_id 1 to create a new ShuffleSchedulerPlugin object from inherited base classes: SchedulerPlugin with example: obj = ShuffleSchedulerPlugin(scheduler)