code: def _test_copy_to_external(use_tensor_list, non_blocking):
      """Test whether the copy_to_external is properly synchronized before the
      output tensor is recycled.
  
      copy_to_external can work in a non-blocking mode - in this mode, the data is
      copied on a user-provided stream, but the host thread doesn't block until
      the copy finishes. However, to ensure that a tensor has been consumed before
      allowing its reuse, a synchronization is scheduled on the stream associated
      with the tensor being copied.
  
      WARNING:
      This test is crafted so that it fails when the synchronization doesn't occur.
      The timing is controlled by data sizes and number of iterations - do not
      change these values!
      """
  
      def ref_tensor(batch_size, sample_shape, start_value):
          volume = np.prod(sample_shape)
          sample0 = torch.arange(
              start_value,
              start_value + volume,
              dtype=torch.int32,
              device="cuda:0",
          ).reshape(shape)
          return torch.stack([sample0 + i for i in range(batch_size)])
  
      def check(arr, ref):
          return torch.equal(arr, ref)
  
      # get a Pytorch CUDA stream
      stream = torch.cuda.Stream(device=0)
      with torch.cuda.stream(stream):
          # allocate an empty tensor into which the pipeline output will be copied
          arr = torch.empty(
              [batch_size] + shape, dtype=torch.int32, device="cuda:0"
          )
          # create a reference tensor...
          ref = ref_tensor(batch_size, shape, 0)
          # ...and tensors which will be used to hog the GPU
          hog = [
              ref_tensor(batch_size, shape, i * batch_size)
              for i in range(20)
          ]
  
          # try 10 times
          for i in range(10):
              # create a fresh pipeline
              pipe = _test_pipe(prefetch_queue_depth=2)
              pipe.build()
              # schedule some runs ahead, so we know that the execution
              # of the next iteration starts immediately
              pipe.schedule_run()
              pipe.schedule_run()
              (out,) = pipe.share_outputs()
              # do something time-consuming on the torch stream to give DALI time
              # to clobber the buffer
              hog = [torch.sqrt(x) for x in hog]
              # copy the result asynchronously
              copy_source = out if use_tensor_list else out.as_tensor()
              feed_ndarray(
                  copy_source, arr, stream.cuda_stream, non_blocking
              )
              pipe.release_outputs()
              # drain
              (_,) = pipe.share_outputs()
              pipe.release_outputs()
              # if no appropriate synchronization is done, the array is likely
              # clobbered with the results from the second iteration
              assert check(arr, ref)
  
              # free resources to prevent OOM in the next iteration
              del pipe
  
file_path: DALI/dali/test/python/test_copy_to_external_torch.py
imports: import ctypes
  import numpy
  import nvidia.dali.fn
  import torch
  from nvidia.dali import pipeline_def
  from nvidia.dali import types
  from nvidia.dali.backend import TensorListGPU
label_desc: Use image node_id 3 for calling a global function with example usage: _test_copy_to_external(use_tensor_list, non_blocking) and returns: torch, torch