code: def _test_iter_setup_zero_copy(
      use_fn_api,
      by_name,
      as_tensor,
      device,
      additional_num_keep_samples,
  ):
      batch_size = 7
      prefetch_queue_depth = 5
  
      class IterSetupPipeline(Pipeline):
          def __init__(
              self,
              iterator,
              num_threads,
              device_id,
              device,
              prefetch_queue_depth=2,
          ):
              super().__init__(
                  batch_size=iterator.batch_size,
                  num_threads=num_threads,
                  device_id=device_id,
                  prefetch_queue_depth=prefetch_queue_depth,
              )
              self.iterator = iterator
              self._device = device
  
          def define_graph(self):
              if use_fn_api:
                  self.batch_1 = fn.external_source(
                      device=self._device, name="src1", no_copy=True
                  )
                  self.batch_2 = fn.external_source(
                      device=self._device, name="src2", no_copy=True
                  )
              else:
                  input_1 = ops.ExternalSource(
                      device=self._device, no_copy=True
                  )
                  input_2 = ops.ExternalSource(
                      device=self._device, no_copy=True
                  )
                  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:
                  self.feed_input("src1", batch_1)
                  self.feed_input("src2", batch_2)
              else:
                  self.feed_input(self.batch_1, batch_1)
                  self.feed_input(self.batch_2, batch_2)
  
      iter_num = 10
      # it is enough to keep only ``prefetch_queue_depth`` or ``cpu_queue_depth * gpu_queue_depth``
      # (when they are not equal), but they are equal in this case
      num_keep_samples = (
          prefetch_queue_depth + additional_num_keep_samples
      )
      source = TestIteratorZeroCopy(
          iter_num,
          batch_size,
          [2, 3],
          as_tensor=as_tensor,
          num_keep_samples=num_keep_samples,
      )
      pipe = IterSetupPipeline(
          iter(source), 3, 0, device, prefetch_queue_depth
      )
      pipe.build()
  
      if (device == "cpu" and not cpu_input) or (
          device == "gpu" and cpu_input
      ):
          input_types = ["CPU", "GPU"]
          if device == "cpu" and not cpu_input:
              input_types.reverse()
          assert_raises(
              RuntimeError,
              pipe.run,
              glob="no_copy is supported only for the same data source device type as "
              "operator. Received: {} input for {} operator".format(
                  *input_types
              ),
          )
      elif additional_num_keep_samples < 0 and not (
          device == "gpu" and not cpu_input and not as_tensor
      ):
          # for the GPU2GPU non contiguous input DALI makes an internal copy on provided stream so no
          # data needs to be preserved by the user
          # assert_raises doesn't work here for the assertions from the test_utils.py
          if_raised = False
          try:
              # this tests bases on the race condition. Running it 5 times make this race more
              # likely to happen and tests pass in CI under high CPU load
              iterations = 5
              for _ in range(iterations):
                  run_and_check(pipe, source)
          except AssertionError:
              if_raised = True
          assert if_raised
      else:
          run_and_check(pipe, source)
  
imports: import functools
  import numpy
  import nvidia.dali.fn
  import nvidia.dali.ops
  from nvidia.dali import Pipeline
  from nose_utils import assert_raises
  from test_utils import check_output
label_desc: Use image node_id 32 for calling a global function with example usage: _test_iter_setup_zero_copy(use_fn_api, by_name, as_tensor, device, additional_num_keep_samples) and returns: list