code: def add_to_broadcast_source(source, benchmark):
      scheduler_type = benchmark["scheduler_type"].value
  
      if scheduler_type not in source:
          source[scheduler_type] = get_value_dict(
              bytes_or_tasks="bytes"
          )
  
      for duration, bytes, engines in [
          (
              result.duration_in_ms,
              result.number_of_bytes,
              result.number_of_engines,
          )
          for result in benchmark["results"]
          if not result.failed
      ]:
          source[scheduler_type]["Duration in ms"].append(duration)
          source[scheduler_type]["Number of bytes"].append(bytes)
          source[scheduler_type]["Number of engines"].append(engines)
  
imports: import json
  import os
  import pickle
  import re
  from datetime import datetime
  from enum import Enum
  from itertools import product
  from utils import seconds_to_ms
  from benchmarks.constants import DEFAULT_NUMBER_OF_ENGINES
label_desc: Use image node_id 9 for calling a global function with example usage: add_to_broadcast_source(source, benchmark) without return types