docstr: Compute vertex pairs that are two hops apart. The resulting pairs are
  sorted before returning.
  
  Returns
  -------
  df : cudf.DataFrame
      df[first] : cudf.Series
          the first vertex id of a pair, if an external vertex id
          is defined by only one column
      df[second] : cudf.Series
          the second vertex id of a pair, if an external vertex id
          is defined by only one column
docstr_tok: ["Compute","vertex","pairs","that","are","two","hops","apart",".","The","resulting","pairs","are","sorted","before","returning",".","Returns","--","--","--","-","df",":","cudf.DataFrame","df","[","first","]",":","cudf.Series","the","first","vertex","id","of","a","pair",",","if","an","external","vertex","id","is","defined","by","only","one","column","df","[","second","]",":","cudf.Series","the","second","vertex","id","of","a","pair",",","if","an","external","vertex","id","is","defined","by","only","one","column"]
code: def get_two_hop_neighbors(self, start_vertices=None):
      """
      Compute vertex pairs that are two hops apart. The resulting pairs are
      sorted before returning.
  
      Returns
      -------
      df : cudf.DataFrame
          df[first] : cudf.Series
              the first vertex id of a pair, if an external vertex id
              is defined by only one column
          df[second] : cudf.Series
              the second vertex id of a pair, if an external vertex id
              is defined by only one column
      """
      _client = default_client()
  
      def _call_plc_two_hop_neighbors(sID, mg_graph_x, start_vertices):
          return pylibcugraph_get_two_hop_neighbors(
              resource_handle=ResourceHandle(
                  Comms.get_handle(sID).getHandle()
              ),
              graph=mg_graph_x,
              start_vertices=start_vertices,
              do_expensive_check=False,
          )
  
      if isinstance(start_vertices, int):
          start_vertices = [start_vertices]
  
      if isinstance(start_vertices, list):
          start_vertices = cudf.Series(start_vertices)
  
      if start_vertices is not None:
          if self.renumbered:
              start_vertices = self.renumber_map.to_internal_vertex_id(
                  start_vertices
              )
              start_vertices_type = self.edgelist.edgelist_df.dtypes[0]
          else:
              start_vertices_type = self.input_df.dtypes[0]
  
          if not isinstance(start_vertices, (dask_cudf.Series)):
              start_vertices = dask_cudf.from_cudf(
                  start_vertices,
                  npartitions=min(
                      self._npartitions, len(start_vertices)
                  ),
              )
              start_vertices = start_vertices.astype(
                  start_vertices_type
              )
  
          n_workers = get_n_workers()
          start_vertices = start_vertices.repartition(
              npartitions=n_workers
          )
          start_vertices = persist_dask_df_equal_parts_per_worker(
              start_vertices, _client
          )
          start_vertices = get_persisted_df_worker_map(
              start_vertices, _client
          )
  
          result = [
              _client.submit(
                  _call_plc_two_hop_neighbors,
                  Comms.get_session_id(),
                  self._plc_graph[w],
                  start_vertices[w][0],
                  workers=[w],
                  allow_other_workers=False,
              )
              for w in start_vertices.keys()
          ]
      else:
          result = [
              _client.submit(
                  _call_plc_two_hop_neighbors,
                  Comms.get_session_id(),
                  self._plc_graph[w],
                  start_vertices,
                  workers=[w],
                  allow_other_workers=False,
              )
              for w in Comms.get_workers()
          ]
  
      wait(result)
  
      def convert_to_cudf(cp_arrays):
          """
          Creates a cudf DataFrame from cupy arrays from pylibcugraph wrapper
          """
          first, second = cp_arrays
          df = cudf.DataFrame()
          df["first"] = first
          df["second"] = second
          return df
  
      cudf_result = [
          _client.submit(convert_to_cudf, cp_arrays)
          for cp_arrays in result
      ]
  
      wait(cudf_result)
      ddf = dask_cudf.from_delayed(cudf_result).persist()
      wait(ddf)
  
      # Wait until the inactive futures are released
      wait(
          [
              (r.release(), c_r.release())
              for r, c_r in zip(result, cudf_result)
          ]
      )
  
      if self.properties.renumbered:
          ddf = self.renumber_map.unrenumber(ddf, "first")
          ddf = self.renumber_map.unrenumber(ddf, "second")
  
      return ddf
  
imports: import gc
  from typing import Union
  import warnings
  import cudf
  import cupy
  import dask
  import dask_cudf
  from dask import delayed
  from dask.distributed import wait, default_client
  import numpy
  from pylibcugraph import MGGraph, ResourceHandle, GraphProperties, get_two_hop_neighbors, select_random_vertices
  from cugraph.structure import graph_primtypes_wrapper
  from cugraph.structure.graph_primtypes_wrapper import Direction
  from cugraph.structure.number_map import NumberMap
  from cugraph.structure.symmetrize import symmetrize
  from cugraph.dask.common.part_utils import get_persisted_df_worker_map, persist_dask_df_equal_parts_per_worker
  from cugraph.dask.common.mg_utils import run_gc_on_dask_cluster
  from cugraph.dask import get_n_workers
  import cugraph.dask.comms.comms
label_desc: Use image node_id 16 for calling the simpleDistributedGraphImpl obj's underlying member method code with example usage: obj.get_two_hop_neighbors(start_vertices) and returns: ddf, pylibcugraph_get_two_hop_neighbors, df