code: def core_number(input_graph, degree_type="bidirectional"):
      """
      Compute the core numbers for the nodes of the graph G. A k-core of a graph
      is a maximal subgraph that contains nodes of degree k or more.
      A node has a core number of k if it belongs a k-core but not to k+1-core.
      This call does not support a graph with self-loops and parallel
      edges.
  
      Parameters
      ----------
      input_graph : cugraph.graph
          cuGraph graph descriptor, should contain the connectivity information,
          (edge weights are not used in this algorithm).
          The current implementation only supports undirected graphs.
  
      degree_type: str, (default="bidirectional")
          This option determines if the core number computation should be based
          on input, output, or both directed edges, with valid values being
          "incoming", "outgoing", and "bidirectional" respectively.
  
  
      Returns
      -------
      result : dask_cudf.DataFrame
          GPU distributed data frame containing 2 dask_cudf.Series
  
          ddf['vertex']: dask_cudf.Series
              Contains the core number vertices
          ddf['core_number']: dask_cudf.Series
              Contains the core number of vertices
      """
  
      if input_graph.is_directed():
          raise ValueError("input graph must be undirected")
  
      if degree_type not in ["incoming", "outgoing", "bidirectional"]:
          raise ValueError(
              f"'degree_type' must be either incoming, "
              f"outgoing or bidirectional, got: {degree_type}"
          )
  
      # Initialize dask client
      client = default_client()
  
      do_expensive_check = False
  
      result = [
          client.submit(
              _call_plc_core_number,
              Comms.get_session_id(),
              input_graph._plc_graph[w],
              degree_type,
              do_expensive_check,
              workers=[w],
              allow_other_workers=False,
          )
          for w in Comms.get_workers()
      ]
  
      wait(result)
  
      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 input_graph.renumbered:
          ddf = input_graph.unrenumber(ddf, "vertex")
  
      return ddf
  
imports: from dask.distributed import wait, default_client
  import cugraph.dask.comms.comms
  import dask_cudf
  import cudf
  from pylibcugraph import ResourceHandle, core_number
label_desc: Use image node_id 3 for calling a global function with example usage: core_number(input_graph, degree_type) and returns: ddf