class_docstr: Comm object using UCP.
  
  Parameters
  ----------
  ep : ucp.Endpoint
      The UCP endpoint.
  address : str
      The address, prefixed with `ucx://` to use.
  deserialize : bool, default True
      Whether to deserialize data in :meth:`distributed.protocol.loads`
  
  Notes
  -----
  The read-write cycle uses the following pattern:
  
  Each msg is serialized into a number of "data" frames. We prepend these
  real frames with two additional frames
  
      1. is_gpus: Boolean indicator for whether the frame should be
         received into GPU memory. Packed in '?' format. Unpack with
         ``<n_frames>?`` format.
      2. frame_size : Unsigned int describing the size of frame (in bytes)
         to receive. Packed in 'Q' format, so a length-0 frame is equivalent
         to an unsized frame. Unpacked with ``<n_frames>Q``.
  
  The expected read cycle is
  
  1. Read the frame describing if connection is closing and number of frames
  2. Read the frame describing whether each data frame is gpu-bound
  3. Read the frame describing whether each data frame is sized
  4. Read all the data frames.
class_docstr_tok: ["Comm","object","using","UCP",".","Parameters","--","--","--","--","--","ep",":","ucp.Endpoint","The","UCP","endpoint",".","address",":","str","The","address",",","prefixed","with","`","ucx",":","\/\/","`","to","use",".","deserialize",":","bool",",","default","True","Whether","to","deserialize","data","in",":","meth",":","`","distributed.protocol.loads","`","Notes","--","--","-","The","read-write","cycle","uses","the","following","pattern",":","Each","msg","is","serialized","into","a","number","of","``","data","''","frames",".","We","prepend","these","real","frames","with","two","additional","frames","1.","is_gpus",":","Boolean","indicator","for","whether","the","frame","should","be","received","into","GPU","memory",".","Packed","in","'","?","'","format",".","Unpack","with","``","<","n_frames",">","?","``","format",".","2.","frame_size",":","Unsigned","int","describing","the","size","of","frame","(","in","bytes",")","to","receive",".","Packed","in","'","Q","'","format",",","so","a","length-0","frame","is","equivalent","to","an","unsized","frame",".","Unpacked","with","``","<","n_frames",">","Q","``",".","The","expected","read","cycle","is","1",".","Read","the","frame","describing","if","connection","is","closing","and","number","of","frames","2",".","Read","the","frame","describing","whether","each","data","frame","is","gpu-bound","3",".","Read","the","frame","describing","whether","each","data","frame","is","sized","4",".","Read","all","the","data","frames","."]
code: def ep(self):
      if self._ep is not None:
          return self._ep
      else:
          raise CommClosedError("UCX Endpoint is closed")
  
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 6 for calling the UCX obj's underlying member method code with example usage: obj.ep() and returns: self