code: def rectangles(msgs, workers=None, start_boundary=0):
      if workers is None:
          workers = {}
  
      L_start = []
      L_duration = []
      L_duration_text = []
      L_key = []
      L_name = []
      L_color = []
      L_alpha = []
      L_worker = []
      L_worker_thread = []
      L_y = []
  
      for msg in msgs:
          key = msg["key"]
          name = key_split(key)
          startstops = msg.get("startstops", [])
          try:
              worker_thread = "%s-%d" % (msg["worker"], msg["thread"])
          except Exception:
              continue
              logger.warning(
                  "Message contained bad information: %s",
                  msg,
                  exc_info=True,
              )
              worker_thread = ""
  
          if worker_thread not in workers:
              workers[worker_thread] = len(workers) / 2
  
          for startstop in startstops:
              if startstop["start"] < start_boundary:
                  continue
              color = colors[startstop["action"]]
              if type(color) is not str:
                  color = color(msg)
  
              L_start.append(
                  (startstop["start"] + startstop["stop"]) / 2 * 1000
              )
              L_duration.append(
                  1000 * (startstop["stop"] - startstop["start"])
              )
              L_duration_text.append(
                  format_time(startstop["stop"] - startstop["start"])
              )
              L_key.append(key)
              L_name.append(prefix[startstop["action"]] + name)
              L_color.append(color)
              L_alpha.append(alphas[startstop["action"]])
              L_worker.append(msg["worker"])
              L_worker_thread.append(worker_thread)
              L_y.append(workers[worker_thread])
  
      return {
          "start": L_start,
          "duration": L_duration,
          "duration_text": L_duration_text,
          "key": L_key,
          "name": L_name,
          "color": L_color,
          "alpha": L_alpha,
          "worker": L_worker,
          "worker_thread": L_worker_thread,
          "y": L_y,
      }
  
file_path: distributed/distributed/diagnostics/task_stream.py
imports: from __future__ import annotations
  import logging
  from collections import deque
  import dask
  from dask.utils import format_time, key_split, parse_timedelta
  from distributed.diagnostics.plugin import SchedulerPlugin
  from distributed.diagnostics.progress_stream import color_of
  from distributed.metrics import time
label_desc: Use image node_id 1 for calling a global function with example usage: rectangles(msgs, workers, start_boundary) and returns: dict