code: def _replace_and_get_dt(complex_iter):
      """_replace_and_get_dt.
  
      FIXME: this function may cause infinite loop when the complex data-structure contains loop-reference
  
      Parameters
      ----------
      complex_iter :
          complex_iter
      """
      if isinstance(complex_iter, DelayedTask):
          dt = complex_iter
          return dt, [dt]
      elif is_delayed_tuple(complex_iter):
          dt = DelayedTuple(complex_iter)
          return dt, [dt]
      elif isinstance(complex_iter, (list, tuple)):
          new_ci = []
          dt_all = []
          for item in complex_iter:
              new_item, dt_list = _replace_and_get_dt(item)
              new_ci.append(new_item)
              dt_all += dt_list
          return new_ci, dt_all
      elif isinstance(complex_iter, dict):
          new_ci = {}
          dt_all = []
          for key, item in complex_iter.items():
              new_item, dt_list = _replace_and_get_dt(item)
              new_ci[key] = new_item
              dt_all += dt_list
          return new_ci, dt_all
      else:
          return complex_iter, []
  
imports: from functools import partial
  from threading import Thread
  from typing import Callable, Text, Union
  from joblib import Parallel, delayed
  from joblib._parallel_backends import MultiprocessingBackend
  import pandas
  from queue import Queue
  import concurrent
  from qlib.config import C, QlibConfig
label_desc: Use image node_id 3 for calling a global function with example usage: _replace_and_get_dt(complex_iter) and returns: dt,  list, dt,  list, new_ci,  dt_all, new_ci,  dt_all, complex_iter,  list