code: def parallel_data_prefetch(
      func: callable,
      data,
      n_proc,
      target_data_type="ndarray",
      cpu_intensive=True,
      use_worker_id=False,
  ):
      if isinstance(data, np.ndarray) and target_data_type == "list":
          raise ValueError("list expected but function got ndarray.")
      elif isinstance(data, abc.Iterable):
          if isinstance(data, dict):
              logging.info(
                  f'WARNING:"data" argument passed to parallel_data_prefetch is a dict: Using only its values and disregarding keys.'
              )
              data = list(data.values())
          if target_data_type == "ndarray":
              data = np.asarray(data)
          else:
              data = list(data)
      else:
          raise TypeError(
              f"The data, that shall be processed parallel has to be either an np.ndarray or an Iterable, but is actually {type(data)}."
          )
  
      if cpu_intensive:
          Q = mp.Queue(1000)
          proc = mp.Process
      else:
          Q = Queue(1000)
          proc = Thread
      # spawn processes
      if target_data_type == "ndarray":
          arguments = [
              [func, Q, part, i, use_worker_id]
              for i, part in enumerate(np.array_split(data, n_proc))
          ]
      else:
          step = (
              int(len(data) / n_proc + 1)
              if len(data) % n_proc != 0
              else int(len(data) / n_proc)
          )
          arguments = [
              [func, Q, part, i, use_worker_id]
              for i, part in enumerate(
                  [
                      data[i : i + step]
                      for i in range(0, len(data), step)
                  ]
              )
          ]
      processes = []
      for i in range(n_proc):
          p = proc(target=_do_parallel_data_prefetch, args=arguments[i])
          processes += [p]
  
      # start processes
      logging.info(f"Start prefetching...")
      import time
  
      start = time.time()
      gather_res = [[] for _ in range(n_proc)]
      try:
          for p in processes:
              p.start()
  
          k = 0
          while k < n_proc:
              # get result
              res = Q.get()
              if res == "Done":
                  k += 1
              else:
                  gather_res[res[0]] = res[1]
  
      except Exception as e:
          logging.info("Exception: ", e)
          for p in processes:
              p.terminate()
  
          raise e
      finally:
          for p in processes:
              p.join()
          logging.info(
              f"Prefetching complete. [{time.time() - start} sec.]"
          )
  
      if target_data_type == "ndarray":
          if not isinstance(gather_res[0], np.ndarray):
              return np.concatenate(
                  [np.asarray(r) for r in gather_res], axis=0
              )
  
          # order outputs
          return np.concatenate(gather_res, axis=0)
      elif target_data_type == "list":
          out = []
          for r in gather_res:
              out.extend(r)
          return out
      else:
          return gather_res
  
imports: import importlib
  import multiprocessing
  from collections import abc
  from inspect import isfunction
  from queue import Queue
  from threading import Thread
  import numpy
  import torch
  from PIL import Image, ImageDraw
  from nemo.utils import logging
label_desc: Use image node_id 11 for calling a global function with example usage: parallel_data_prefetch(func, data, n_proc, target_data_type, cpu_intensive, use_worker_id) and returns: np, np, out, gather_res