code: def build_dynamic_dataset_iter(
      opt,
      transforms_cls,
      vocabs,
      copy=False,
      task=CorpusTask.TRAIN,
      stride=1,
      offset=0,
      src=None,
      tgt=None,
      align=None,
      device_id=-1,
  ):
      """
      Build `DynamicDatasetIter` from opt.
      if src, tgt,align are passed then dataset is built from those lists
      instead of opt.[src, tgt, align]
      Typically this function is called for CorpusTask.[TRAIN,VALID,INFER]
      from the main tain / translate scripts
      We disable automatic batching in the DataLoader.
      The custom optimized batching is performed by the
      custom class DynamicDatasetIter inherited from IterableDataset
      (and not by a custom collate function).
      We load opt.bucket_size examples, sort them and yield
      mini-batchs of size opt.batch_size.
      The bucket_size must be large enough to ensure homogeneous batches.
      Each worker will load opt.prefetch_factor mini-batches in
      advance to avoid the GPU waiting during the refilling of the bucket.
      """
      transforms = make_transforms(opt, transforms_cls, vocabs)
      corpora = get_corpora(opt, task, src=src, tgt=tgt, align=align)
      if corpora is None:
          assert (
              task != CorpusTask.TRAIN
          ), "only valid corpus is ignorable."
          return None
      device = (
          torch.device(device_id)
          if device_id >= 0
          else torch.device("cpu")
      )
      num_workers = (
          opt.num_workers if hasattr(opt, "num_workers") else 0
      )
      if num_workers == 0 or task == CorpusTask.INFER:
          # single thread - create batch directly on GPU if device is gpu
          data_iter = DynamicDatasetIter.from_opt(
              corpora,
              transforms,
              vocabs,
              opt,
              task,
              copy=copy,
              stride=stride,
              offset=offset,
              device=device,
          )
          data_iter.num_workers = num_workers
          data_iter._init_datasets(
              0
          )  # when workers=0 init_fn not called
          return data_iter
      else:
          # multithread faster to create batch on CPU in each thread and then move it to gpu
          data_iter = DynamicDatasetIter.from_opt(
              corpora,
              transforms,
              vocabs,
              opt,
              task,
              copy=copy,
              stride=stride,
              offset=offset,
              device=torch.device("cpu"),
          )
          data_iter.num_workers = num_workers
          data_loader = DataLoader(
              data_iter,
              batch_size=None,
              pin_memory=True,
              multiprocessing_context="spawn",
              num_workers=data_iter.num_workers,
              worker_init_fn=data_iter._init_datasets,
              prefetch_factor=opt.prefetch_factor,
          )
          # Move tensor_batch from cpu to device
          return OnDeviceDatasetIter(data_loader, device)
  
imports: import torch
  from itertools import cycle
  from onmt.constants import CorpusTask, ModelTask
  from onmt.inputters.text_corpus import get_corpora, build_corpora_iters
  from onmt.inputters.text_utils import text_sort_key, process, numericalize, tensorify, _addcopykeys
  from onmt.transforms import make_transforms
  from onmt.utils.logging import init_logger, logger
  from onmt.utils.misc import RandomShuffler
  from torch.utils.data import DataLoader
label_desc: Use image node_id 1 for calling a global function with example usage: build_dynamic_dataset_iter(opt, transforms_cls, vocabs, copy, task, stride, offset, src, tgt, align, device_id) and returns: None, data_iter, OnDeviceDatasetIter