code: def bfs(
      input_graph,
      start,
      depth_limit=None,
      return_distances=True,
      check_start=True,
  ):
      """
      Find the distances and predecessors for a breadth-first traversal of a
      graph.
      The input graph must contain edge list as a dask-cudf dataframe with
      one partition per GPU.
  
      Parameters
      ----------
      input_graph : cugraph.Graph
          cuGraph graph instance, should contain the connectivity information
          as dask cudf edge list dataframe (edge weights are not used for this
          algorithm).
  
      start : Integer or list or cudf object or dask_cudf object
          The id(s) of the graph vertex from which the traversal begins
          in each component of the graph.  Only one vertex per connected
          component of the graph is allowed.
  
      depth_limit : Integer or None, optional (default=None)
          Limit the depth of the search
  
      return_distances : bool, optional (default=True)
          Indicates if distances should be returned
  
      check_start : bool, optional (default=True)
          If True, performs more extensive tests on the start vertices
          to ensure validitity, at the expense of increased run time.
  
      Returns
      -------
      df : dask_cudf.DataFrame
          df['vertex'] gives the vertex id
  
          df['distance'] gives the path distance from the
          starting vertex (Only if return_distances is True)
  
          df['predecessor'] gives the vertex it was
          reached from in the traversal
  
      Examples
      --------
      >>> import cugraph.dask as dcg
      >>> import dask_cudf
      >>> # ... Init a DASK Cluster
      >>> #    see https://docs.rapids.ai/api/cugraph/stable/dask-cugraph.html
      >>> # Download dataset from https://github.com/rapidsai/cugraph/datasets/..
      >>> chunksize = dcg.get_chunksize(datasets_path / "karate.csv")
      >>> ddf = dask_cudf.read_csv(datasets_path / "karate.csv",
      ...                          chunksize=chunksize, delimiter=" ",
      ...                          names=["src", "dst", "value"],
      ...                          dtype=["int32", "int32", "float32"])
      >>> dg = cugraph.Graph(directed=True)
      >>> dg.from_dask_cudf_edgelist(ddf, source='src', destination='dst',
      ...                            edge_attr='value')
      >>> df = dcg.bfs(dg, 0)
  
      """
  
      client = default_client()
      invalid_dtype = False
  
      if not isinstance(start, (dask_cudf.DataFrame, dask_cudf.Series)):
          if not isinstance(start, (cudf.DataFrame, cudf.Series)):
              vertex_dtype = input_graph.nodes().dtype
              start = cudf.Series(start, dtype=vertex_dtype)
          # convert into a dask_cudf
          start = dask_cudf.from_cudf(start, input_graph._npartitions)
  
      if check_start:
          if isinstance(start, dask_cudf.Series):
              vertex_dtype = input_graph.nodes().dtype
              if start.dtype is not vertex_dtype:
                  invalid_dtype = True
          else:
              # Multicolumn vertices case
              start_dtype = start.dtypes.reset_index(drop=True)
              vertex_dtype = input_graph.nodes().dtypes.reset_index(
                  drop=True
              )
              if not start_dtype.equals(vertex_dtype):
                  invalid_dtype = True
  
          if invalid_dtype:
              warning_msg = (
                  "The 'start' values dtype must match "
                  "the graph's vertices dtype."
              )
  
              warnings.warn(warning_msg, UserWarning)
              if isinstance(start, dask_cudf.Series):
                  start = start.astype(vertex_dtype)
              else:
                  start = start.astype(vertex_dtype[0])
  
          is_valid_vertex = input_graph.has_node(start)
          if not is_valid_vertex:
              raise ValueError(
                  "At least one start vertex provided was invalid"
              )
  
      if input_graph.renumbered:
          if isinstance(start, dask_cudf.DataFrame):
              tmp_col_names = start.columns
  
          elif isinstance(start, dask_cudf.Series):
              tmp_col_names = None
  
          start = input_graph.lookup_internal_vertex_id(
              start, tmp_col_names
          )
          vertex_dtype = (
              start.dtype
          )  # if the edgelist was renumbered, update
          # the vertex type accordingly
  
      data_start = persist_dask_df_equal_parts_per_worker(
          start, client, return_type="dict"
      )
  
      do_expensive_check = False
      # FIXME: Why is 'direction_optimizing' not part of the python cugraph API
      # and why is it set to 'False' by default
      direction_optimizing = False
  
      cupy_result = [
          client.submit(
              _call_plc_bfs,
              Comms.get_session_id(),
              input_graph._plc_graph[w],
              st[0] if st else cudf.Series(dtype=vertex_dtype),
              depth_limit,
              direction_optimizing,
              return_distances,
              do_expensive_check,
              workers=[w],
              allow_other_workers=False,
          )
          for w, st in data_start.items()
      ]
  
      wait(cupy_result)
  
      cudf_result = [
          client.submit(convert_to_cudf, cp_arrays)
          for cp_arrays in cupy_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(cupy_result, cudf_result)
          ]
      )
  
      if input_graph.renumbered:
          ddf = input_graph.unrenumber(ddf, "vertex")
          ddf = input_graph.unrenumber(ddf, "predecessor")
          ddf = ddf.fillna(-1)
      return ddf
  
file_path: cugraph/python/cugraph/cugraph/dask/traversal/bfs.py
imports: from pylibcugraph import ResourceHandle, bfs
  from dask.distributed import wait, default_client
  from cugraph.dask.common.part_utils import persist_dask_df_equal_parts_per_worker
  import cugraph.dask.comms.comms
  import cudf
  import dask_cudf
  import warnings
label_desc: Use image node_id 3 for calling a global function with example usage: bfs(input_graph, start, depth_limit, return_distances, check_start) and returns: ddf