code: def task_stream_append(lists, msg, workers):
      key = msg["key"]
      name = key_split(key)
      startstops = msg.get("startstops", [])
  
      for startstop in startstops:
          color = colors[startstop["action"]]
          if type(color) is not str:
              color = color(msg)
  
          lists["start"].append(
              (startstop["start"] + startstop["stop"]) / 2 * 1000
          )
          lists["duration"].append(
              1000 * (startstop["stop"] - startstop["start"])
          )
          lists["key"].append(key)
          lists["name"].append(prefix[startstop["action"]] + name)
          lists["color"].append(color)
          lists["alpha"].append(alphas[startstop["action"]])
          lists["worker"].append(msg["worker"])
  
          worker_thread = "%s-%d" % (msg["worker"], msg["thread"])
          lists["worker_thread"].append(worker_thread)
          if worker_thread not in workers:
              workers[worker_thread] = len(workers) / 2
          lists["y"].append(workers[worker_thread])
  
      return len(startstops)
  
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 5 for calling a global function with example usage: task_stream_append(lists, msg, workers) and returns: len