code: def run_dali(
      reduce_fn,
      batch_fn,
      keep_dims,
      axes,
      output_type,
      add_mean_input=False,
      ddof=0,
      layout=None,
  ):
      batch_size = batch_fn.batch_size()
  
      # Needed due to how ExternalSource API works. It fails on methods, partials.
      def get_batch():
          return batch_fn()
  
      result_cpu = []
      result_gpu = []
  
      pipe = Pipeline(batch_size=batch_size, num_threads=4, device_id=0)
  
      args = {"keep_dims": keep_dims, "axes": axes}
      if output_type is not None:
          args["dtype"] = np_type_to_dali(output_type)
  
      with pipe:
          input = fn.external_source(source=get_batch, layout=layout)
          if not add_mean_input:
              reduced_cpu = reduce_fn(input, **args)
              reduced_gpu = reduce_fn(input.gpu(), **args)
          else:
              mean = fn.reductions.mean(input, **args)
              args["ddof"] = ddof
              reduced_cpu = reduce_fn(input, mean, **args)
              reduced_gpu = reduce_fn(input.gpu(), mean.gpu(), **args)
          pipe.set_outputs(reduced_cpu, reduced_gpu)
  
      pipe.build()
  
      for _ in range(batch_fn.num_iter()):
          output = pipe.run()
          check_layout(output[0], layout, axes, keep_dims)
          check_layout(output[1], layout, axes, keep_dims)
          reduced_cpu = output[0].as_array()
          reduced_gpu = output[1].as_cpu().as_array()
          result_cpu.append(reduced_cpu)
          result_gpu.append(reduced_gpu)
  
      return result_cpu, result_gpu
  
imports: import nvidia.dali.fn
  from nvidia.dali.pipeline import Pipeline
  from nose.tools import nottest
  import numpy
  from test_utils import np_type_to_dali
  from nose_utils import assert_raises
label_desc: Use image node_id 3 for calling a global function with example usage: run_dali(reduce_fn, batch_fn, keep_dims, axes, output_type, add_mean_input, ddof, layout) and returns: result_cpu,  result_gpu, batch_fn