code: def _test_memory_consumption(device, test_case):
      batch_size = 32
      num_iters = 128
  
      if device == "cpu":
          import numpy as np
  
          fw = np
      else:
          fw = cp
  
      def no_copy_sample():
          batch = [
              fw.full((1024, 1024, 4), i, dtype=fw.int32)
              for i in range(batch_size)
          ]
  
          def cb(sample_info):
              return batch[sample_info.idx_in_batch]
  
          return cb
  
      def copy_sample():
          def cb(sample_info):
              return fw.full(
                  (1024, 1024, 4),
                  sample_info.idx_in_batch,
                  dtype=fw.int32,
              )
  
          return cb
  
      def copy_batch():
          def cb():
              return fw.full(
                  (batch_size, 1024, 1024, 4), 42, dtype=fw.int32
              )
  
          return cb
  
      cases = {
          "no_copy_sample": (no_copy_sample, True, False),
          "copy_sample": (copy_sample, False, False),
          "copy_batch": (copy_batch, False, True),
      }
  
      cb, no_copy, batch_mode = cases[test_case]
  
      @pipeline_def
      def pipeline():
          return fn.external_source(
              source=cb(),
              device=device,
              batch=batch_mode,
              no_copy=no_copy,
          )
  
      pipe = pipeline(batch_size=batch_size, num_threads=4, device_id=0)
      pipe.build()
      for _ in range(num_iters):
          pipe.run()
  
file_path: DALI/dali/test/python/test_external_source_cupy.py
imports: from nose.plugins.attrib import attr
  from test_external_source_impl import *
  from test_external_source_impl import use_cupy
  from test_utils import check_output, check_output_pattern
  import nvidia.dali
  from nvidia.dali import Pipeline, pipeline_def
  import nvidia.dali.fn
  from nvidia.dali.tensors import TensorGPU
  import numpy
  import cupy
label_desc: Use image node_id 5 for calling a global function with example usage: _test_memory_consumption(device, test_case) and returns: cb, cb, cb, fn, batch, fw, fw