class_docstr: Keep track of all keys, grouped by key_split
class_docstr_tok: ["Keep","track","of","all","keys",",","grouped","by","key_split"]
code: def transition(self, key, start, finish, *args, **kwargs):
      ts = self.scheduler.tasks[key]
      prefix = ts.prefix.name
      self.all[prefix].add(key)
      try:
          self.state[start][prefix].remove(key)
      except (
          KeyError
      ):  # TODO: remove me once we have a new or clean state
          pass
  
      if start == "memory" and ts.nbytes >= 0:
          # XXX why not respect DEFAULT_DATA_SIZE?
          self.nbytes[prefix] -= ts.nbytes
      if finish == "memory" and ts.nbytes >= 0:
          self.nbytes[prefix] += ts.nbytes
  
      if finish != "forgotten":
          self.state[finish][prefix].add(key)
      else:
          s = self.all[prefix]
          s.remove(key)
          if not s:
              del self.all[prefix]
              self.nbytes.pop(prefix, None)
              for v in self.state.values():
                  v.pop(prefix, None)
  
imports: from __future__ import annotations
  import asyncio
  import logging
  import warnings
  from collections import defaultdict
  from timeit import default_timer
  from typing import ClassVar
  from tlz import groupby, valmap
  from dask.base import tokenize
  from dask.utils import key_split
  from distributed.diagnostics.plugin import SchedulerPlugin
  from distributed.metrics import time
label_desc: Use image node_id 2 for calling the AllProgress obj's underlying member method code with example usage: obj.transition(key, start, finish) without return types