code: def _test_large_sample(start_method):
      batch_size = 2
  
      @pipeline_def
      def create_pipeline():
          large = fn.external_source(
              large_sample_cb,
              batch=False,
              parallel=True,
              prefetch_queue_depth=1,
          )
          # iteration over array in Python is too slow, so reduce the number of elements
          # to iterate over
          reduced = fn.reductions.sum(large, axes=(1, 2))
          return reduced
  
      pipe = create_pipeline(
          batch_size=batch_size,
          py_num_workers=2,
          py_start_method=start_method,
          prefetch_queue_depth=1,
          num_threads=2,
          device_id=0,
      )
      pipe.build()
      capture_processes(pipe._py_pool)
      for batch_idx in range(8):
          (out,) = pipe.run()
          for idx_in_batch in range(batch_size):
              idx_in_epoch = batch_size * batch_idx + idx_in_batch
              expected_val = idx_in_epoch * 1024 * 1024
              a = np.array(out[idx_in_batch])
              assert a.shape == (
                  512,
              ), "Expected shape (512,) but got {}".format(a.shape)
              for val in a.flat:
                  assert val == expected_val, (
                      f"Unexpected value in batch: got {val}, expected {expected_val}, "
                      f"for batch {batch_idx}, sample {idx_in_batch}"
                  )
  
imports: import numpy
  from nose.tools import with_setup
  from nvidia.dali import pipeline_def
  import nvidia.dali.fn
  from test_external_source_parallel_utils import setup_function, teardown_function, capture_processes
label_desc: Use image node_id 2 for calling a global function with example usage: _test_large_sample(start_method) and returns: reduced