code: def start_dask_client(
      protocol=None,
      rmm_async=False,
      rmm_pool_size=None,
      dask_worker_devices=None,
      jit_unspill=False,
      worker_class=None,
      device_memory_limit=0.8,
  ):
      """
      Creates a new dask client, and possibly also a cluster, and returns them as
      a tuple (client, cluster).
  
      If the env var SCHEDULER_FILE is set, it is assumed to contain the path to
      a JSON file generated by a running dask scheduler that can be used to
      configure the new dask client (the new client object returned will be a
      client to that scheduler), and the value of cluster will be None. If
      SCHEDULER_FILE is not set, a new LocalCUDACluster will be created and
      returned as the value of cluster.
  
      If the env var DASK_WORKER_DEVICES is set, it will be assumed to be a list
      of comma-separated GPU devices (ex. "0,1,2" for those 3 devices) for the
      LocalCUDACluster to use when setting up individual workers (1 worker per
      device). If not set, the parameter dask_worker_devices will be used the
      same way instead. If neither are set, the new LocalCUDACluster instance
      will default to one worker per device visible to this process.
  
      If the env var DASK_LOCAL_DIRECTORY is set, it will be used as the
      "local_directory" arg to LocalCUDACluster, for all temp files generated.
  
      Upon successful creation of a client (either to a LocalCUDACluster or
      otherwise), the cugraph.dask.comms.comms singleton is initialized using
      "p2p=True".
  
      Parameters
      ----------
      protocol : str or None, default None
          The "protocol" arg to LocalCUDACluster (ex. "tcp"), see docs for
          dask_cuda.LocalCUDACluster for details. This parameter is ignored if
          the env var SCHEDULER_FILE is set which implies the dask cluster has
          already been created.
  
      rmm_pool_size : int, str or None, default None
          The "rmm_pool_size" arg to LocalCUDACluster (ex. "20GB"), see docs for
          dask_cuda.LocalCUDACluster for details. This parameter is ignored if
          the env var SCHEDULER_FILE is set which implies the dask cluster has
          already been created.
  
      dask_worker_devices : str, list of int, or None, default None
          GPUs to restrict activity to. Can be a string (like ``"0,1,2,3"``),
          list (like ``[0, 1, 2, 3]``), or ``None`` to use all available GPUs.
          This parameter is overridden by the value of env var
          DASK_WORKER_DEVICES. This parameter is ignored if the env var
          SCHEDULER_FILE is set which implies the dask cluster has already been
          created.
  
      jit_unspill : bool or None, default None
          The "jit_unspill" arg to LocalCUDACluster to enable just-in-time
          spilling, see docs for dask_cuda.LocalCUDACluster for details. This
          parameter is ignored if the env var SCHEDULER_FILE is set which implies
          the dask cluster has already been created.
  
      device_memory_limit : int, float, str, or None, default 0.8
          The "device_memory_limit" arg to LocalCUDACluster to determine when
          workers start spilling to host memory, see docs for
          dask_cuda.LocalCUDACluster for details. This parameter is ignored if
          the env var SCHEDULER_FILE is set which implies the dask cluster has
          already been created.
      """
      dask_scheduler_file = os.environ.get("SCHEDULER_FILE")
      dask_local_directory = os.getenv("DASK_LOCAL_DIRECTORY")
      # Allow the DASK_WORKER_DEVICES env var to override a value passed in. If
      # neither are set, this will be None.
      dask_worker_devices = os.getenv(
          "DASK_WORKER_DEVICES", dask_worker_devices
      )
  
      cluster = None
      client = None
      tempdir_object = None
  
      if dask_scheduler_file:
          if protocol is not None:
              print(
                  f"WARNING: {protocol=} is ignored in start_dask_client() when using "
                  "dask SCHEDULER_FILE"
              )
          if rmm_pool_size is not None:
              print(
                  f"WARNING: {rmm_pool_size=} is ignored in start_dask_client() when "
                  "using dask SCHEDULER_FILE"
              )
          if dask_worker_devices is not None:
              print(
                  f"WARNING: {dask_worker_devices=} is ignored in start_dask_client() "
                  "when using dask SCHEDULER_FILE"
              )
          initialize()
          client = Client(scheduler_file=dask_scheduler_file)
          # FIXME: use proper logging, INFO or DEBUG level
          print("\nDask client created using " f"{dask_scheduler_file}")
      else:
          if dask_local_directory is None:
              # The tempdir created by tempdir_object should be cleaned up once
              # tempdir_object is deleted.
              tempdir_object = tempfile.TemporaryDirectory()
              local_directory = tempdir_object.name
          else:
              local_directory = dask_local_directory
  
          cluster = LocalCUDACluster(
              local_directory=local_directory,
              protocol=protocol,
              rmm_pool_size=rmm_pool_size,
              rmm_async=rmm_async,
              CUDA_VISIBLE_DEVICES=dask_worker_devices,
              jit_unspill=jit_unspill,
              worker_class=worker_class,
              device_memory_limit=device_memory_limit,
          )
          client = Client(cluster)
  
          if dask_worker_devices is None:
              num_workers = len(get_visible_devices())
          else:
              if isinstance(dask_worker_devices, list):
                  num_workers = len(dask_worker_devices)
              else:
                  # FIXME: this assumes a properly formatted string with commas
                  num_workers = len(dask_worker_devices.split(","))
  
          client.wait_for_workers(num_workers)
          # Add a reference to tempdir_object to the client to prevent it from
          # being deleted when this function returns. This will be deleted in
          # stop_dask_client()
          client.tempdir_object = tempdir_object
          # FIXME: use proper logging, INFO or DEBUG level
          print("\nDask client/cluster created using LocalCUDACluster")
  
      Comms.initialize(p2p=True)
  
      return (client, cluster)
  
imports: import os
  import tempfile
  from pprint import pformat
  import time
  from dask.distributed import wait, default_client
  from dask import persist
  from dask.distributed import Client
  from dask.base import is_dask_collection
  from dask_cuda import LocalCUDACluster
  from dask_cuda.initialize import initialize
  from cugraph.dask.comms import comms
  from cugraph.dask.common.mg_utils import get_visible_devices
  from cugraph.generators import rmat
  import numpy
label_desc: Use image node_id 1 for calling a global function with example usage: start_dask_client(protocol, rmm_async, rmm_pool_size, dask_worker_devices, jit_unspill, worker_class, device_memory_limit) and returns: client,  cluster