code: def init_once():
      global ucp, device_array
      global ucx_create_endpoint, ucx_create_listener
      global pre_existing_cuda_context, cuda_context_created
  
      if ucp is not None:
          return
  
      # remove/process dask.ucx flags for valid ucx options
      ucx_config, ucx_environment = _prepare_ucx_config()
  
      # We ensure the CUDA context is created before initializing UCX. This can't
      # be safely handled externally because communications in Dask start before
      # preload scripts run.
      # Precedence:
      # 1. external environment
      # 2. ucx_config (high level settings passed to ucp.init)
      # 3. ucx_environment (low level settings equivalent to environment variables)
      ucx_tls = os.environ.get(
          "UCX_TLS",
          ucx_config.get("TLS", ucx_environment.get("UCX_TLS", "")),
      )
      if (
          dask.config.get("distributed.comm.ucx.create-cuda-context")
          is True
          # This is not foolproof, if UCX_TLS=all we might require CUDA
          # depending on configuration of UCX, but this is better than
          # nothing
          or ("cuda" in ucx_tls and "^cuda" not in ucx_tls)
      ):
          try:
              import numba.cuda
          except ImportError:
              raise ImportError(
                  "CUDA support with UCX requires Numba for context management"
              )
  
          cuda_visible_device = get_device_index_and_uuid(
              os.environ.get("CUDA_VISIBLE_DEVICES", "0").split(",")[0]
          )
          pre_existing_cuda_context = has_cuda_context()
          if pre_existing_cuda_context.has_context:
              _warn_existing_cuda_context(
                  pre_existing_cuda_context.device_info, os.getpid()
              )
  
          numba.cuda.current_context()
  
          cuda_context_created = has_cuda_context()
          if (
              cuda_context_created.has_context
              and cuda_context_created.device_info.uuid
              != cuda_visible_device.uuid
          ):
              _warn_cuda_context_wrong_device(
                  cuda_visible_device,
                  cuda_context_created.device_info,
                  os.getpid(),
              )
  
      import ucp as _ucp
  
      ucp = _ucp
  
      with patch.dict(os.environ, ucx_environment):
          # We carefully ensure that ucx_environment only contains things
          # that don't override ucx_config or existing slots in the
          # environment, so the user's external environment can safely
          # override things here.
          ucp.init(options=ucx_config, env_takes_precedence=True)
  
      pool_size_str = dask.config.get("distributed.rmm.pool-size")
  
      # Find the function, `cuda_array()`, to use when allocating new CUDA arrays
      try:
          import rmm
  
          def device_array(n):
              return rmm.DeviceBuffer(size=n)
  
          if pool_size_str is not None:
              pool_size = parse_bytes(pool_size_str)
              rmm.reinitialize(
                  pool_allocator=True,
                  managed_memory=False,
                  initial_pool_size=pool_size,
              )
      except ImportError:
          try:
              import numba.cuda
  
              def numba_device_array(n):
                  a = numba.cuda.device_array((n,), dtype="u1")
                  weakref.finalize(a, numba.cuda.current_context)
                  return a
  
              device_array = numba_device_array
  
          except ImportError:
  
              def device_array(n):
                  raise RuntimeError(
                      "In order to send/recv CUDA arrays, Numba or RMM is required"
                  )
  
          if pool_size_str is not None:
              logger.warning(
                  "Initial RMM pool size defined, but RMM is not available. "
                  "Please consider installing RMM or removing the pool size option."
              )
  
imports: from __future__ import annotations
  import functools
  import logging
  import os
  import struct
  import weakref
  from collections.abc import Awaitable, Callable, Collection
  from typing import TYPE_CHECKING, Any
  from unittest.mock import patch
  import dask
  from dask.utils import parse_bytes
  from distributed.comm.addressing import parse_host_port, unparse_host_port
  from distributed.comm.core import BaseListener, Comm, CommClosedError, Connector
  from distributed.comm.registry import Backend, backends
  from distributed.comm.utils import ensure_concrete_host, from_frames, to_frames
  from distributed.diagnostics.nvml import CudaDeviceInfo, get_device_index_and_uuid, has_cuda_context
  from distributed.protocol.utils import host_array
  from distributed.utils import ensure_ip, get_ip, get_ipv6, log_errors, nbytes
label_desc: Use image node_id 5 for calling a global function with example usage: init_once() and returns: rmm, a