code: def progress_quads(msg, nrows=8, ncols=3):
      """
      >>> msg = {'all': {'inc': 5, 'dec': 1, 'add': 4},
      ...        'memory': {'inc': 2, 'dec': 0, 'add': 1},
      ...        'erred': {'inc': 0, 'dec': 1, 'add': 0},
      ...        'released': {'inc': 1, 'dec': 0, 'add': 1},
      ...        'processing': {'inc': 1, 'dec': 0, 'add': 2},
      ...        'queued': {'inc': 1, 'dec': 0, 'add': 2}}
  
      >>> progress_quads(msg, nrows=2)  # doctest: +SKIP
      {'all': [5, 4, 1],
      'memory': [2, 1, 0],
      'erred': [0, 0, 1],
      'released': [1, 1, 0],
      'processing': [1, 2, 0],
      'queued': [1, 2, 0],
      'name': ['inc', 'add', 'dec'],
      'show-name': ['inc', 'add', 'dec'],
      'left': [0, 0, 1],
      'right': [0.9, 0.9, 1.9],
      'top': [0, -1, 0],
      'bottom': [-0.8, -1.8, -0.8],
      'color': ['#45BF6F', '#2E6C8E', '#440154'],
      'released-loc': [0.18, 0.225, 1.0],
      'memory-loc': [0.54, 0.45, 1.0],
      'erred-loc': [0.54, 0.45, 1.9],
      'processing-loc': [0.72, 0.9, 1.9],
      'queued-loc': [0.9, 1.35, 1.9],
      'done': ['3 / 5', '2 / 4', '1 / 1']}
      """
      width = 0.9
      names = sorted(msg["all"], key=msg["all"].get, reverse=True)
      names = names[: nrows * ncols]
      n = len(names)
      d = {
          k: [v.get(name, 0) for name in names] for k, v in msg.items()
      }
  
      d["name"] = names
      d["show-name"] = [
          name if len(name) <= 15 else name[:12] + "..."
          for name in names
      ]
      d["left"] = [i // nrows for i in range(n)]
      d["right"] = [i // nrows + width for i in range(n)]
      d["top"] = [-(i % nrows) for i in range(n)]
      d["bottom"] = [-(i % nrows) - 0.8 for i in range(n)]
      d["color"] = [color_of(name) for name in names]
  
      d["released-loc"] = []
      d["memory-loc"] = []
      d["erred-loc"] = []
      d["processing-loc"] = []
      d["queued-loc"] = []
      d["no-worker-loc"] = []
      d["done"] = []
      for r, m, e, p, q, nw, a, l in zip(
          d["released"],
          d["memory"],
          d["erred"],
          d["processing"],
          d["queued"],
          d.get("no_worker", [0] * n),
          d["all"],
          d["left"],
      ):
          rl = width * r / a + l
          ml = width * (r + m) / a + l
          el = width * (r + m + e) / a + l
          pl = width * (p + r + m + e) / a + l
          ql = width * (p + r + m + e + q) / a + l
          nwl = width * (p + r + m + e + q + nw) / a + l
          done = "%d / %d" % (r + m + e, a)
          d["released-loc"].append(rl)
          d["memory-loc"].append(ml)
          d["erred-loc"].append(el)
          d["processing-loc"].append(pl)
          d["queued-loc"].append(ql)
          d["no-worker-loc"].append(nwl)
          d["done"].append(done)
  
      return d
  
file_path: distributed/distributed/diagnostics/progress_stream.py
imports: from __future__ import annotations
  import logging
  from tlz import merge, valmap
  from dask.utils import key_split
  from distributed.core import coerce_to_address, connect
  from distributed.diagnostics.progress import AllProgress
  from distributed.utils import color_of
  from distributed.worker import dumps_function
label_desc: Use image node_id 3 for calling a global function with example usage: progress_quads(msg, nrows, ncols) and returns: d