code: def test_record_stream_ndarray():
      cycles_per_ms = _get_cycles_per_ms()
  
      t = nd.array(
          np.array([1.0, 2.0, 3.0, 4.0], dtype=np.float32), ctx=nd.cpu()
      )
      t.pin_memory_()
      result = nd.empty([4], ctx=nd.gpu(0))
      stream = torch.cuda.Stream()
      ptr = [None]
  
      # Performs the CPU->GPU copy in a background stream
      def perform_copy():
          with torch.cuda.stream(stream):
              tmp = t.copyto(nd.gpu(0))
              ptr[0] = F.from_dgl_nd(tmp).data_ptr()
          torch.cuda.current_stream().wait_stream(stream)
          tmp.record_stream(
              to_dgl_stream_handle(torch.cuda.current_stream())
          )
          torch.cuda._sleep(int(50 * cycles_per_ms))  # delay the copy
          result.copyfrom(tmp)
  
      perform_copy()
      with torch.cuda.stream(stream):
          tmp2 = nd.empty([4], ctx=nd.gpu(0))
          assert (
              F.from_dgl_nd(tmp2).data_ptr() != ptr[0]
          ), "allocation re-used too soon"
  
      assert torch.equal(
          F.from_dgl_nd(result).cpu(),
          torch.tensor([1.0, 2.0, 3.0, 4.0]),
      )
  
      # Check that the block will be re-used after the main stream finishes
      torch.cuda.current_stream().synchronize()
      with torch.cuda.stream(stream):
          tmp3 = nd.empty([4], ctx=nd.gpu(0))
          assert (
              F.from_dgl_nd(tmp3).data_ptr() == ptr[0]
          ), "allocation not re-used"
  
file_path: dgl/tests/python/pytorch/test_ffi-stream.py
imports: import unittest
  from statistics import mean
  import backend
  import dgl
  import dgl.ndarray
  import dgl.ops
  import numpy
  import torch
  from dgl import rand_graph
  from dgl._ffi.streams import _dgl_get_stream, to_dgl_stream_handle
  from dgl.utils import to_dgl_context
label_desc: Use image node_id 4 for calling a global function with example usage: test_record_stream_ndarray() without return types