code: def _test_iter_setup(use_fn_api, by_name, src_device, gen_device):
      batch_size = 7
  
      class IterSetupPipeline(Pipeline):
          def __init__(
              self, iterator, num_threads, device_id, src_device
          ):
              super().__init__(
                  batch_size=iterator.batch_size,
                  num_threads=num_threads,
                  device_id=device_id,
              )
  
              self.iterator = iterator
              self._device = src_device
  
          def define_graph(self):
              if use_fn_api:
                  # pass a Torch stream where data is generated
                  self.batch_1 = fn.external_source(
                      device=self._device,
                      name="src1",
                      cuda_stream=torch.cuda.default_stream(),
                  )
                  self.batch_2 = fn.external_source(
                      device=self._device,
                      name="src2",
                      cuda_stream=torch.cuda.default_stream(),
                  )
              else:
                  input_1 = ops.ExternalSource(device=self._device)
                  input_2 = ops.ExternalSource(device=self._device)
                  self.batch_1 = input_1(name="src1")
                  self.batch_2 = input_2(name="src2")
              return [self.batch_1, self.batch_2]
  
          def iter_setup(self):
              batch_1, batch_2 = next(self.iterator)
              if by_name:
                  # pass a Torch stream where data is generated
                  self.feed_input(
                      "src1",
                      batch_1,
                      cuda_stream=torch.cuda.default_stream(),
                  )
                  self.feed_input(
                      "src2",
                      batch_2,
                      cuda_stream=torch.cuda.default_stream(),
                  )
              else:
                  # pass a Torch stream where data is generated
                  self.feed_input(
                      self.batch_1,
                      batch_1,
                      cuda_stream=torch.cuda.default_stream(),
                  )
                  self.feed_input(
                      self.batch_2,
                      batch_2,
                      cuda_stream=torch.cuda.default_stream(),
                  )
  
      iter_num = 5
      source = TestIterator(
          n=iter_num,
          batch_size=batch_size,
          dims=[2, 3],
          device=gen_device,
      )
      pipe = IterSetupPipeline(iter(source), 3, 0, src_device)
      pipe.build()
  
      run_and_check(pipe, source)
  
file_path: DALI/dali/test/python/test_external_source_pytorch_dlpack.py
imports: import numpy
  import nvidia.dali.fn
  import nvidia.dali.ops
  import torch
  from nvidia.dali.backend_impl import *
  from nvidia.dali import Pipeline
  from torch.utils.dlpack import to_dlpack, from_dlpack
  from test_utils import check_output
label_desc: Use image node_id 3 for calling a global function with example usage: _test_iter_setup(use_fn_api, by_name, src_device, gen_device) and returns: list