code: def test_multiprocessing_with_shared_memory():
      """Test if two CSCSamplingGraphs are on the same shared memory after
      spawning.
  
      For now this code only works when the sharing strategy of
      torch.multiprocessing is set to `file_system` at the beginning.
      The cause is still yet to be found.
      """
  
      total_num_nodes = 5
      total_num_edges = 10
      num_ntypes = 2
      num_etypes = 3
      (
          csc_indptr,
          indices,
          node_type_offset,
          type_per_edge,
          node_type_to_id,
          edge_type_to_id,
      ) = gbt.random_hetero_graph(
          total_num_nodes, total_num_edges, num_ntypes, num_etypes
      )
  
      csc_indptr.share_memory_()
      indices.share_memory_()
      node_type_offset.share_memory_()
      type_per_edge.share_memory_()
  
      graph = gb.fused_csc_sampling_graph(
          csc_indptr,
          indices,
          node_type_offset=node_type_offset,
          type_per_edge=type_per_edge,
          node_type_to_id=node_type_to_id,
          edge_type_to_id=edge_type_to_id,
          edge_attributes=None,
      )
  
      ctx = mp.get_context("spawn")  # Use spawn method.
  
      data_queue = ctx.Queue()  # Used for sending graph.
      flag_queue = ctx.Queue()  # Used for sending finish signal.
  
      p = ctx.Process(
          target=process_csc_sampling_graph_on_shared_memory,
          args=(graph, data_queue, flag_queue),
      )
      p.start()
      try:
          # Get data from the other process. Then check if the tensors here have
          # the same data.
          csc_indptr2 = data_queue.get()
          assert torch.equal(graph.csc_indptr, csc_indptr2)
          indices2 = data_queue.get()
          assert torch.equal(graph.indices, indices2)
          node_type_offset2 = data_queue.get()
          assert torch.equal(graph.node_type_offset, node_type_offset2)
          type_per_edge2 = data_queue.get()
          assert torch.equal(graph.type_per_edge, type_per_edge2)
      except:
          raise
      finally:
          # Send a finish signal to end sub-process.
          flag_queue.put(None)
      p.join()
  
imports: import os
  import pickle
  import re
  import tempfile
  import unittest
  import backend
  import dgl
  import dgl.graphbolt
  import pytest
  import torch
  import torch.multiprocessing
  from dgl.graphbolt.base import etype_str_to_tuple
  from scipy import sparse
  from ..None import gb_test_utils
label_desc: Use image node_id 25 for calling a global function with example usage: test_multiprocessing_with_shared_memory() without return types