ucp_ep.h

/**
 * Copyright (c) NVIDIA CORPORATION & AFFILIATES, 2001-2020. ALL RIGHTS RESERVED.
 * Copyright (C) Los Alamos National Security, LLC. 2019 ALL RIGHTS RESERVED.
 *
 * See file LICENSE for terms.
 */

#ifndef UCP_EP_H_
#define UCP_EP_H_

#include "ucp_types.h"

#include <ucp/proto/lane_type.h>
#include <ucp/proto/proto_select.h>
#include <ucp/wireup/ep_match.h>
#include <ucp/api/ucp.h>
#include <uct/api/uct.h>
#include <uct/api/v2/uct_v2.h>
#include <ucs/datastruct/queue.h>
#include <ucs/datastruct/ptr_map.h>
#include <ucs/datastruct/strided_alloc.h>
#include <ucs/debug/assert.h>
#include <ucs/stats/stats.h>


#define UCP_MAX_IOV                16UL


/* Endpoint flags type */
#if ENABLE_DEBUG_DATA || UCS_ENABLE_ASSERT
typedef uint32_t                   ucp_ep_flags_t;
#else
typedef uint16_t                   ucp_ep_flags_t;
#endif

#if UCS_ENABLE_ASSERT
#define UCP_EP_ASSERT_COUNTER_INC(_counter) \
    do { \
        ucs_assert(*(_counter) < UINT_MAX); \
        ++(*(_counter)); \
    } while (0)

#define UCP_EP_ASSERT_COUNTER_DEC(_counter) \
    do { \
        ucs_assert(*(_counter) > 0); \
        --(*(_counter)); \
    } while (0)
#else
#define UCP_EP_ASSERT_COUNTER_INC(_counter)
#define UCP_EP_ASSERT_COUNTER_DEC(_counter)
#endif


#define ucp_ep_refcount_add(_ep, _type) \
({ \
    ucs_assertv((_ep)->refcount < UINT8_MAX, "ep=%p", _ep); \
    ++(_ep)->refcount; \
    UCP_EP_ASSERT_COUNTER_INC(&(_ep)->refcounts._type); \
})

/* Return 1 if the endpoint was destroyed, 0 if not */
#define ucp_ep_refcount_remove(_ep, _type) \
({ \
    int __ret = 0; \
    \
    UCP_EP_ASSERT_COUNTER_DEC(&(_ep)->refcounts._type); \
    ucs_assertv((_ep)->refcount > 0, "ep=%p", _ep); \
    if (--(_ep)->refcount == 0) { \
        ucp_ep_destroy_base(_ep); \
        __ret = 1; \
    } \
    \
    (__ret); \
})

#define ucp_ep_refcount_field_assert(_ep, _refcount_field, _cmp, _val) \
    ucs_assertv((_ep)->_refcount_field _cmp (_val), "ep=%p: %s=%u vs %u", \
                (_ep), UCS_PP_MAKE_STRING(_refcount_field), \
                (_ep)->_refcount_field, _val);

#define ucp_ep_refcount_assert(_ep, _type_refcount, _cmp, _val) \
    ucp_ep_refcount_field_assert(_ep, refcounts._type_refcount, _cmp, _val)

#define UCP_SA_DATA_HEADER_VERSION_SHIFT 5


/**
 * Endpoint flags
 */
enum {
    UCP_EP_FLAG_LOCAL_CONNECTED        = UCS_BIT(0), /* All local endpoints are connected,
                                                        for CM case - local address was packed,
                                                        UCT did not report errors during
                                                        connection establishment protocol
                                                        and disconnect not called yet */
    UCP_EP_FLAG_REMOTE_CONNECTED       = UCS_BIT(1), /* All remote endpoints are connected */
    UCP_EP_FLAG_CONNECT_REQ_QUEUED     = UCS_BIT(2), /* Connection request was queued */
    UCP_EP_FLAG_FAILED                 = UCS_BIT(3), /* EP is in failed state */
    UCP_EP_FLAG_USED                   = UCS_BIT(4), /* EP is in use by the user */
    UCP_EP_FLAG_STREAM_HAS_DATA        = UCS_BIT(5), /* EP has data in the ext.stream.match_q */
    UCP_EP_FLAG_ON_MATCH_CTX           = UCS_BIT(6), /* EP is on match queue */
    UCP_EP_FLAG_REMOTE_ID              = UCS_BIT(7), /* remote ID is valid */
    UCP_EP_FLAG_BLOCK_FLUSH            = UCS_BIT(8), /* Flush ops have to be blocking
                                                        while merging pending queues */
    UCP_EP_FLAG_CONNECT_PRE_REQ_QUEUED = UCS_BIT(9), /* Pre-Connection request was queued */
    UCP_EP_FLAG_CLOSED                 = UCS_BIT(10),/* EP was closed */
    /* 11 bit is vacant for a flag */
    UCP_EP_FLAG_ERR_HANDLER_INVOKED    = UCS_BIT(12),/* error handler was called */
    UCP_EP_FLAG_INTERNAL               = UCS_BIT(13),/* the internal EP which holds
                                                        temporary wireup configuration or
                                                        mem-type EP */
    UCP_EP_FLAG_INDIRECT_ID            = UCS_BIT(14),/* protocols on this endpoint will send
                                                        indirect endpoint id instead of pointer,
                                                        can be replaced with looking at local ID */
    UCP_EP_FLAG_USER_DATA_PARAM        = UCS_BIT(15),/* EP's user_data was passed via
                                                        @ref ucp_ep_params_t::user_data */

    /* DEBUG bits */
    UCP_EP_FLAG_CONNECT_REQ_SENT       = UCS_BIT(16),/* DEBUG: Connection request was sent */
    UCP_EP_FLAG_CONNECT_REP_SENT       = UCS_BIT(17),/* DEBUG: Connection reply was sent */
    UCP_EP_FLAG_CONNECT_ACK_SENT       = UCS_BIT(18),/* DEBUG: Connection ACK was sent */
    UCP_EP_FLAG_CONNECT_REQ_IGNORED    = UCS_BIT(19),/* DEBUG: Connection request was ignored */
    UCP_EP_FLAG_CONNECT_PRE_REQ_SENT   = UCS_BIT(20),/* DEBUG: Connection pre-request was sent */
    UCP_EP_FLAG_FLUSH_STATE_VALID      = UCS_BIT(21),/* DEBUG: flush_state is valid */
    UCP_EP_FLAG_DISCONNECTED_CM_LANE   = UCS_BIT(22),/* DEBUG: CM lane was disconnected, i.e.
                                                        @uct_ep_disconnect was called for CM EP */
    UCP_EP_FLAG_CLIENT_CONNECT_CB      = UCS_BIT(23),/* DEBUG: Client connect callback invoked */
    UCP_EP_FLAG_SERVER_NOTIFY_CB       = UCS_BIT(24),/* DEBUG: Server notify callback invoked */
    UCP_EP_FLAG_DISCONNECT_CB_CALLED   = UCS_BIT(25),/* DEBUG: Got disconnect notification */
    UCP_EP_FLAG_CONNECT_WAIT_PRE_REQ   = UCS_BIT(26) /* DEBUG: Connection pre-request needs to be
                                                        received from a peer */
};


/**
 * UCP endpoint statistics counters
 */
enum {
    UCP_EP_STAT_TAG_TX_EAGER,
    UCP_EP_STAT_TAG_TX_EAGER_SYNC,
    UCP_EP_STAT_TAG_TX_RNDV,
    UCP_EP_STAT_LAST
};


/**
 * Endpoint init flags
 */
enum {
    UCP_EP_INIT_FLAG_MEM_TYPE          = UCS_BIT(0),  /**< Endpoint for local mem type transfers */
    UCP_EP_INIT_CREATE_AM_LANE         = UCS_BIT(1),  /**< Endpoint requires an AM lane */
    UCP_EP_INIT_CM_WIREUP_CLIENT       = UCS_BIT(2),  /**< Endpoint wireup protocol is based on CM,
                                                           client side */
    UCP_EP_INIT_CM_WIREUP_SERVER       = UCS_BIT(3),  /**< Endpoint wireup protocol is based on CM,
                                                           server side */
    UCP_EP_INIT_ERR_MODE_PEER_FAILURE  = UCS_BIT(4),  /**< Endpoint requires an
                                                           @ref UCP_ERR_HANDLING_MODE_PEER */
    UCP_EP_INIT_CM_PHASE               = UCS_BIT(5),  /**< Endpoint connection to a peer is on
                                                           CM phase */
    UCP_EP_INIT_FLAG_INTERNAL          = UCS_BIT(6),  /**< Endpoint for internal usage
                                                           (e.g. memtype, reply on keepalive) */
    UCP_EP_INIT_CONNECT_TO_IFACE_ONLY  = UCS_BIT(7),  /**< Select transports which
                                                           support CONNECT_TO_IFACE
                                                           mode only */
    UCP_EP_INIT_CREATE_AM_LANE_ONLY    = UCS_BIT(8),  /**< Endpoint requires an AM lane only */
    UCP_EP_INIT_KA_FROM_EXIST_LANES    = UCS_BIT(9),  /**< Use only existing lanes to create
                                                           keepalive lane */
    UCP_EP_INIT_ALLOW_AM_AUX_TL        = UCS_BIT(10), /**< Endpoint allows selecting of auxiliary
                                                           transports for AM lane */
    UCP_EP_INIT_ERR_MODE_FAILOVER      = UCS_BIT(11), /**< Endpoint requires an
                                                           @ref UCP_ERR_HANDLING_MODE_FAILOVER */
    UCP_EP_INIT_RECOVERY               = UCS_BIT(12),

    /**
     * For consistency with @ref UCP_SA_DATA_MASK_ERR_MODE_FAILOVER
     */
    UCP_EP_INIT_ERR_MODE_FAILOVER_MASK = UCP_EP_INIT_ERR_MODE_PEER_FAILURE |
                                         UCP_EP_INIT_ERR_MODE_FAILOVER
};


/**
 * Endpoint configuration key flags
 */
enum {
    /**
     * Endpoint is connected to own worker.
     */
    UCP_EP_CONFIG_KEY_FLAG_SELF         = UCS_BIT(0),

    /**
     * Endpoint is within same node.
     */
    UCP_EP_CONFIG_KEY_FLAG_INTRA_NODE   = UCS_BIT(1),

    /**
     * Endpoint is in an intermediate connection establishment phase,
     * some capabilities may be limited or disabled.
     */
    UCP_EP_CONFIG_KEY_FLAG_INTERMEDIATE = UCS_BIT(2)
};


#define UCP_EP_STAT_TAG_OP(_ep, _op) \
    UCS_STATS_UPDATE_COUNTER((_ep)->stats, UCP_EP_STAT_TAG_TX_##_op, 1);


typedef struct ucp_ep_config_key_lane {
    ucp_rsc_index_t      rsc_index; /* Resource index */
    ucp_md_index_t       dst_md_index; /* Destination memory domain index */
    ucs_sys_device_t     dst_sys_dev; /* Destination system device */
    uint8_t              path_index; /* Device path index */
    ucp_lane_type_mask_t lane_types; /* Which types of operations this lane
                                        was selected for */
    uint8_t              port_speed; /* Quantized port speed */
    size_t               seg_size; /* Maximal fragment size which can be
                                      received by the peer */
} ucp_ep_config_key_lane_t;


/*
 * Endpoint configuration key.
 * This is filled by the transport selection logic, according to the local
 * resources and set of remote addresses.
 */
struct ucp_ep_config_key {

    ucp_lane_index_t         num_lanes;       /* Number of active lanes */
    ucp_ep_config_key_lane_t lanes[UCP_MAX_LANES]; /* Active lanes */

    ucp_lane_index_t         am_lane;         /* Lane for AM (can be NULL) */
    ucp_lane_index_t         tag_lane;        /* Lane for tag matching offload (can be NULL) */
    ucp_lane_index_t         wireup_msg_lane; /* Lane for wireup messages (can be NULL) */
    ucp_lane_index_t         cm_lane;         /* Lane for holding a CM connection (can be NULL) */
    ucp_lane_index_t         keepalive_lane;  /* Lane for checking a connection state (can be NULL) */

    /* Lanes for remote memory access, sorted by priority, highest first */
    ucp_lane_index_t         rma_lanes[UCP_MAX_LANES];

    /* Lanes for high-bw memory access, sorted by priority, highest first */
    ucp_lane_index_t         rma_bw_lanes[UCP_MAX_LANES];

    /* Lane for obtaining remote memory pointer */
    ucp_lane_index_t         rkey_ptr_lane;

    /* Lanes for atomic operations, sorted by priority, highest first */
    ucp_lane_index_t         amo_lanes[UCP_MAX_LANES];

    /* Lanes for high-bw active messages, sorted by priority, highest first */
    ucp_lane_index_t         am_bw_lanes[UCP_MAX_LANES];

    /* Local memory domains to send remote keys used by high-bw rma protocols
     * NOTE: potentially it can be different than what is imposed by rma_bw_lanes,
     * since these are the MDs used by remote side for accessing our memory. */
    ucp_md_map_t             rma_bw_md_map;

    /* Local memory domains to use for RMA protocol. */
    ucp_md_map_t             rma_md_map;

    /* Bitmap of remote mds which are reachable from this endpoint (with any set
     * of transports which could be selected in the future).
     */
    ucp_md_map_t             reachable_md_map;

    /* Array with popcount(reachable_md_map) elements, each entry holds the local
     * component index to be used for unpacking remote key from each set bit in
     * reachable_md_map */
    ucp_rsc_index_t          *dst_md_cmpts;

    /* Error handling mode */
    ucp_err_handling_mode_t  err_mode;

    /* Additional flags */
    unsigned                 flags;

    /* Indicates peer release version */
    unsigned                 dst_version;
};


/*
 * Configuration for AM and tag offload protocols
 */
typedef struct ucp_ep_msg_config {
        ssize_t            max_short;
        size_t             max_bcopy;
        size_t             max_zcopy;
        size_t             max_hdr;
        size_t             max_iov;

        /* zero-copy threshold for operations which do not have to wait for remote side */
        size_t             zcopy_thresh[UCP_MAX_IOV];

        /* zero-copy threshold for mem type buffers */
        size_t             mem_type_zcopy_thresh[UCS_MEMORY_TYPE_LAST];

        /* zero-copy threshold for operations which anyways have to wait for remote side */
        size_t             sync_zcopy_thresh[UCP_MAX_IOV];
        uint8_t            zcopy_auto_thresh; /* if != 0 the zcopy enabled */
} ucp_ep_msg_config_t;


/*
 * Thresholds with and without non-host memory
 */
typedef struct ucp_memtype_thresh {
        ssize_t            memtype_on;
        ssize_t            memtype_off;
} ucp_memtype_thresh_t;


/*
 * Rendezvous thresholds
 */
typedef struct ucp_rndv_thresh {
    /* threshold calculated assuming faster remote completion */
    size_t            remote;
    /* threshold calculated assuming faster local completion, for instance
     * when UCP_OP_ATTR_FLAG_FAST_CMP flag is provided to send operation
     * parameters */
    size_t            local;
} ucp_rndv_thresh_t;


/*
 * Rendezvous Zcopy configuration
 */
typedef struct ucp_rndv_zcopy {
    /* Maximal total size of Zcopy operation */
    size_t           max;
    /* Minimal size of Zcopy operation */
    size_t           min;
    /* Can the message which are > maximal size be split to the segments which are
     * >= minimal size */
    int              split;
    /* Lanes for Zcopy operation */
    ucp_lane_index_t lanes[UCP_MAX_LANES];
    /* BW based scale factor for zcopy lanes */
    double           scale[UCP_MAX_LANES];
} ucp_ep_rndv_zcopy_config_t;


/*
 * Element in ep peer memory hash. The element represents remote peer shared
 * memory segment. Having it hashed helps to avoid expensive rkey unpacking
 * and md registration procedures. Unpacking is expensive, because for shared
 * memory segments it assumes attach/mmap calls. Registration is needed for
 * better performance of CPU<->GPU memory transfers and is typically quite
 * expensive on memtype ep mds, such as cuda copy.
 */
typedef struct {
    /* Unpacked rkey with the only MD supporting RKEY_PTR */
    ucp_rkey_h       rkey;
    /* Size of the buffer corresponding to the unpacked rkey */
    size_t           size;
    /* MD index corresponding to memtype ep */
    ucp_md_index_t   md_index;
    /* Memory handle holding registration of the remote buffer on memtype
     * ep MD */
    uct_mem_h        uct_memh;
    /* Local pointer to the remote memory described by rkey */
    void             *local_ptr;
} ucp_ep_peer_mem_data_t;


KHASH_DECLARE(ucp_ep_peer_mem_hash, uint64_t, ucp_ep_peer_mem_data_t);


typedef enum {
    /* Protocol initialization was done */
    UCP_EP_PROTO_INITIALIZED = UCS_BIT(0),
} ucp_ep_init_flags_t;


struct ucp_ep_config {

    /* A key which uniquely defines the configuration, and all other fields of
     * configuration (in the current worker) and defined only by it.
     */
    ucp_ep_config_key_t     key;

    /* Bitmap of which lanes are p2p; affects the behavior of connection
     * establishment protocols.
     */
    ucp_lane_map_t          p2p_lanes;

    /* Flags which has to be used @ref uct_md_mkey_pack_v2 */
    unsigned                uct_rkey_pack_flags;

    /* Configuration for AM lane */
    ucp_ep_msg_config_t     am;

    /* MD index of each lane */
    ucp_md_index_t          md_index[UCP_MAX_LANES];

    struct {
        /* RNDV GET Zcopy configuration */
        ucp_ep_rndv_zcopy_config_t get_zcopy;
        /* RNDV PUT Zcopy configuration */
        ucp_ep_rndv_zcopy_config_t put_zcopy;
        /* Threshold for switching from eager to RMA based rendezvous */
        ucp_rndv_thresh_t          rma_thresh;
        /* Threshold for switching from eager to AM based rendezvous */
        ucp_rndv_thresh_t          am_thresh;
        /* Total size of packed rkey, according to high-bw md_map */
        size_t                     rkey_size;
        /* Remote memory domains which support rkey_ptr, excluding
         * rkey_ptr_lane. Used by 2-stage pipeline rendezvous protocol. */
        ucp_md_map_t               proto_rndv_rkey_skip_mds;
        /* Remote memory domain corresponding to rkey_ptr_lane */
        ucp_md_map_t               rkey_ptr_lane_dst_mds;
    } rndv;

    struct {
        /* Protocols used for tag matching operations
         * (can be AM based or tag offload). */
        const ucp_request_send_proto_t   *proto;
        const ucp_request_send_proto_t   *sync_proto;

        /* Lane used for tag matching operations. */
        ucp_lane_index_t     lane;

        /* Maximal size for eager short. */
        ucp_memtype_thresh_t max_eager_short;

        /* Configuration of the lane used for eager protocols
         * (can be AM or tag offload). */
        ucp_ep_msg_config_t  eager;

        /* Threshold for switching from eager to rendezvous. Can be different
         * from AM thresholds if tag offload is enabled and tag offload lane is
         * not the same as AM lane. */
        struct {
            ucp_rndv_thresh_t    rma_thresh;
            ucp_rndv_thresh_t    am_thresh;
        } rndv;

        struct {
            /* Maximal size for eager short. */
            ucp_memtype_thresh_t max_eager_short;

            /* Maximal iov count for RNDV offload */
            size_t          max_rndv_iov;
            /* Maximal total size for RNDV offload */
            size_t          max_rndv_zcopy;
        } offload;
    } tag;

    struct {
        /* Protocols used for stream operations
         * (currently it's only AM based). */
        const ucp_request_send_proto_t   *proto;
    } stream;

    struct {
        /* Protocols used for am operations */
        const ucp_request_send_proto_t   *proto;
        const ucp_request_send_proto_t   *reply_proto;

        /* Maximal size for eager short */
        ucp_memtype_thresh_t             max_eager_short;

        /* Maximal size for eager short with reply protocol */
        ucp_memtype_thresh_t             max_reply_eager_short;
    } am_u;

    /* Protocol selection data */
    ucp_proto_select_t            proto_select;

    /* Bitmap of preregistration for am_bw lanes */
    ucp_md_map_t                  am_bw_prereg_md_map;

    /* Bitmap of lanes selected by the protocols */
    ucp_lane_map_t                proto_lane_map;

    /* EP initialization flags from @ref ucp_ep_init_flags_t */
    unsigned                      proto_init_flags;

    /* Number of endpoints using this configuration */
    unsigned                      ep_count;
};


/**
 * Status of protocol-level remote completions
 */
typedef struct {
    ucs_hlist_head_t reqs; /* Queue of flush requests which
                              are waiting for remote completion */
    uint32_t         send_sn; /* Sequence number of sent operations */
    uint32_t         cmpl_sn; /* Sequence number of completions */
    uint32_t         mem_in_progress; /* Track ongoing memory flushes for this endpoint */
} ucp_ep_flush_state_t;


typedef struct ucp_ep_recovery_probe {
    uct_completion_t  comp;
    ucp_ep_h          ep;
    ucp_lane_index_t  lane;
} ucp_ep_recovery_probe_t;


enum {
    UCP_EP_RECOVERY_STATE_IDLE,
    UCP_EP_RECOVERY_STATE_WAIT_REPLY,
    UCP_EP_RECOVERY_STATE_PROBING,
    UCP_EP_RECOVERY_STATE_PROBE_OK
};


/* Per-EP recovery retry state. */
typedef struct ucp_ep_recovery_arg {
    /* number of retries left before giving up */
    unsigned                retries_left;
    uint8_t                 state;
    ucp_ep_recovery_probe_t probe[UCP_MAX_LANES];
} ucp_ep_recovery_arg_t;


/**
 * Endpoint extension
 */
typedef struct ucp_ep_ext {
    ucp_ep_h                      ep;            /* Back pointer to endpoint */
    void                          *user_data;    /* User data associated with ep */
    ucs_list_link_t               ep_list;       /* List entry in worker's all eps list */
    ucp_rsc_index_t               cm_idx;        /* CM index */
    ucs_ptr_map_key_t             local_ep_id;   /* Local EP ID */
    ucs_ptr_map_key_t             remote_ep_id;  /* Remote EP ID */
    ucp_err_handler_cb_t          err_cb;        /* Error handler */
    union {
        ucp_request_t             *close_req;    /* Close protocol request */
        ucp_ep_recovery_arg_t     *recovery_arg; /* Lanes recovery state object.
                                                    United with close request since:
                                                    1) recovery is not supported for connected to sockaddr EPs
                                                    2) it does not make sense to recover lanes during close protocol */
    };
    khash_t(ucp_ep_peer_mem_hash) *peer_mem;     /* Hash of remote memory segments
                                                    used by 2-stage ppln rndv proto */
    /* List of requests which are waiting for remote completion */
    ucs_hlist_head_t              proto_reqs;
#if UCS_ENABLE_ASSERT
    ucs_time_t                    ka_last_round; /* Time of last KA round done */
#endif

    /* Endpoint match context and remote completion status are mutually exclusive,
     * since remote completions are counted only after the endpoint is already
     * matched to a remote peer.
     */
    union {
        ucp_ep_match_elem_t       ep_match;      /* Matching with remote endpoints */
        ucp_ep_flush_state_t      flush_state;   /* Remote completion status */
    };

    struct {
        ucs_list_link_t           ready_list;     /* List entry in worker's EP list */
        ucs_queue_head_t          match_q;        /* Queue of receive data or requests,
                                                     depends on UCP_EP_FLAG_STREAM_HAS_DATA */
    } stream;

    struct {
        ucs_list_link_t           started_ams;
        ucs_queue_head_t          mid_rdesc_q;    /* Queue of middle fragments, which
                                                     arrived before the first one */
        uint64_t                  psn;
    } am;

    ucp_lane_map_t                unflushed_lanes; /* Bitmap of lanes which have
                                                      unflushed operations */
    uint64_t                      fence_seq;       /* Sequence number for fence
                                                      detection */

    /**
     * UCT endpoints for every slow-path lane that has no room in the base endpoint
     * structure. TODO allocate this array dynamically.
     */
    uct_ep_h                     *uct_eps;


    /**
     * Map of system devices that require a flush operation
     */
    ucp_sys_dev_map_t             flush_sys_dev_map;
} ucp_ep_ext_t;


/**
 * Protocol layer endpoint, represents a connection to a remote worker
 */
typedef struct ucp_ep {
    ucp_worker_h                  worker;        /* Worker this endpoint belongs to */

    ucp_worker_cfg_index_t        cfg_index;     /* Configuration index */
    ucp_ep_match_conn_sn_t        conn_sn;       /* Sequence number for remote connection */
    ucp_ep_flags_t                flags;         /* Endpoint flags */
    uint8_t                       refcount;      /* Reference counter: 0 - it is
                                                    allowed to destroy EP */
    ucp_lane_index_t              am_lane;       /* Cached value */
    /* Transports for every lane */
    uct_ep_h                      uct_eps[UCP_MAX_FAST_PATH_LANES];
    ucp_ep_ext_t                  *ext;                   /* Endpoint extension */

#if ENABLE_DEBUG_DATA
    char                          peer_name[UCP_WORKER_ADDRESS_NAME_MAX];
    /* Endpoint name for tracing and analysis */
    char                          name[UCP_ENTITY_NAME_MAX];
#endif

#if UCS_ENABLE_ASSERT
    struct {
        /* How many times the EP create was done */
        unsigned                      create;
        /* How many Worker flush operations are in-progress where the EP is the
         * next EP for flushing */
        unsigned                      flush;
        /* How many UCT EP discarding operations are in-progress scheduled for
         * the EP */
        unsigned                      discard;
        /* How many recovery aux probes are in-progress on the EP */
        unsigned                      probe;
    } refcounts;
#endif

    UCS_STATS_NODE_DECLARE(stats)
} ucp_ep_t;


enum {
    UCP_WIREUP_SA_DATA_CM_ADDR   = UCS_BIT(1)  /* Sockaddr client data contains address
                                                  for CM based wireup: there is only
                                                  iface and ep address of transport
                                                  lanes, remote device address is
                                                  provided by CM and has to be added to
                                                  unpacked UCP address locally. */
};


/* Sockaddr data flags that are packed to the header field in
 * ucp_wireup_sockaddr_data_base_t structure.
 */
enum {
    /* Indicates support of @ref UCP_ERR_HANDLING_MODE_PEER error mode. */
    UCP_SA_DATA_FLAG_ERR_MODE_PEER     = UCS_BIT(0),
    /* Indicates support of @ref UCP_ERR_HANDLING_MODE_FAILOVER error mode.
     * NOTE: use @ref UCP_SA_DATA_MASK_ERR_MODE_FAILOVER for backward
     *       compatibility to fallback peer failure mode to
     *       @ref UCP_ERR_HANDLING_MODE_PEER */
    UCP_SA_DATA_FLAG_ERR_MODE_FAILOVER = UCS_BIT(1),

    /**
     * Backward compatibility mask
     */
    UCP_SA_DATA_MASK_ERR_MODE_FAILOVER = UCP_SA_DATA_FLAG_ERR_MODE_PEER |
                                         UCP_SA_DATA_FLAG_ERR_MODE_FAILOVER
};


/* Basic sockaddr data. Version 1 uses some additional fields which are not
 * really needed and removed in version 2.
 */
typedef struct ucp_wireup_sockaddr_data_base {
    uint64_t                  ep_id; /**< Endpoint ID */

    /* This field has different meaning for sa_data v1 and other versions:
     * v1:           it is error handling mode
     * v2 and newer: it is sa_data header with the following format:
     *   +---+-----+
     *   | 3 |  5  |
     *   +---+-----+
     *     v    |
     * version  |
     *          v
     *        flags
     *
     * It is safe to keep version in 3 MSB, because it will always be zeros
     * (i.e. UCP_OBJECT_VERSION_V1) in sa_data v1 (err_mode value is small).
     */
    uint8_t                   header;
    /* packed worker address (or sa_data v1) follows */
} UCS_S_PACKED ucp_wireup_sockaddr_data_base_t;


typedef struct ucp_wireup_sockaddr_data_v1 {
    ucp_wireup_sockaddr_data_base_t super;
    uint8_t                         addr_mode; /**< The attached address format
                                                    defined by
                                                    UCP_WIREUP_SA_DATA_xx */
    uint8_t                         dev_index; /**< Device address index used to
                                                    build remote address in
                                                    UCP_WIREUP_SA_DATA_CM_ADDR
                                                    mode */
    /* packed worker address follows */
} UCS_S_PACKED ucp_wireup_sockaddr_data_v1_t;


typedef struct ucp_conn_request {
    ucp_listener_h              listener;
    uct_listener_h              uct_listener;
    uct_conn_request_h          uct_req;
    ucp_rsc_index_t             cm_idx;
    char                        dev_name[UCT_DEVICE_NAME_MAX];
    uct_device_addr_t           *remote_dev_addr;
    struct sockaddr_storage     client_address;
    ucp_ep_h                    ep; /* valid only if request is handled internally */
    /* sa_data and packed worker address follow */
} ucp_conn_request_t;


int ucp_is_uct_ep_failed(uct_ep_h uct_ep);

uct_iface_h ucp_failed_tl_iface_get(void);

void ucp_ep_config_key_reset(ucp_ep_config_key_t *key);

void ucp_ep_config_cm_lane_info_str(ucp_worker_h worker,
                                    const ucp_ep_config_key_t *key,
                                    ucp_lane_index_t lane,
                                    ucp_rsc_index_t cm_index,
                                    ucs_string_buffer_t *buf);

void ucp_ep_config_lane_info_str(ucp_worker_h worker,
                                 const ucp_ep_config_key_t *key,
                                 const unsigned *addr_indices,
                                 ucp_lane_index_t lane,
                                 ucp_rsc_index_t aux_rsc_index,
                                 ucs_string_buffer_t *buf);

ucs_status_t ucp_ep_create_base(ucp_worker_h worker, unsigned ep_init_flags,
                                const char *peer_name, const char *message,
                                ucp_ep_h *ep_p);

void ucp_ep_destroy_base(ucp_ep_h ep);

void ucp_ep_delete(ucp_ep_h ep);

void ucp_ep_flush_state_reset(ucp_ep_h ep);

void ucp_ep_flush_state_invalidate(ucp_ep_h ep);

void ucp_ep_release_id(ucp_ep_h ep);

ucs_status_t
ucp_ep_config_err_mode_check_mismatch(ucp_ep_h ep,
                                      ucp_err_handling_mode_t err_mode);

ucs_status_t
ucp_ep_create_to_worker_addr(ucp_worker_h worker,
                             const ucp_tl_bitmap_t *local_tl_bitmap,
                             const ucp_unpacked_address_t *remote_address,
                             unsigned ep_init_flags, const char *message,
                             unsigned *addr_indices, ucp_ep_h *ep_p);

ucs_status_t ucp_ep_create_server_accept(ucp_worker_h worker,
                                         const ucp_conn_request_h conn_request,
                                         ucp_ep_h *ep_p);

ucs_status_ptr_t ucp_ep_flush_internal(ucp_ep_h ep, unsigned req_flags,
                                       const ucp_request_param_t *param,
                                       ucp_request_t *worker_req,
                                       ucp_request_callback_t flushed_cb,
                                       const char *debug_name,
                                       unsigned uct_flags);

void ucp_ep_config_key_set_err_mode(ucp_ep_config_key_t *key,
                                    unsigned ep_init_flags);

void ucp_ep_config_key_init_flags(ucp_ep_config_key_t *key,
                                  unsigned ep_init_flags);

void ucp_ep_err_pending_purge(uct_pending_req_t *self, void *arg);

void ucp_destroyed_ep_pending_purge(uct_pending_req_t *self, void *arg);

void ucp_ep_disconnected(ucp_ep_h ep, int force);

void ucp_ep_destroy_internal(ucp_ep_h ep);

void ucp_ep_set_lanes_failed(ucp_ep_h ucp_ep, ucp_lane_map_t lanes,
                             ucs_status_t status);

void ucp_ep_set_lanes_failed_schedule(ucp_ep_h ucp_ep, ucp_lane_map_t lanes,
                                      ucs_status_t status);

void ucp_ep_unprogress_uct_ep(ucp_ep_h ep, uct_ep_h uct_ep,
                              ucp_rsc_index_t rsc_index);

void ucp_ep_cleanup_lanes(ucp_ep_h ep);

ucs_status_t ucp_ep_config_init(ucp_worker_h worker, ucp_ep_config_t *config,
                                const ucp_ep_config_key_t *key);

void ucp_ep_config_cleanup(ucp_worker_h worker, ucp_ep_config_t *config);

int ucp_ep_config_lane_is_peer_match(const ucp_ep_config_key_t *key1,
                                     ucp_lane_index_t lane1,
                                     const ucp_ep_config_key_t *key2,
                                     ucp_lane_index_t lane2);

ucp_lane_index_t
ucp_ep_config_find_match_lane(const ucp_ep_config_key_t *old_key,
                              ucp_lane_index_t old_lane,
                              const ucp_ep_config_key_t *new_key);

void ucp_ep_config_lanes_intersect(const ucp_ep_config_key_t *old_key,
                                   const ucp_ep_config_key_t *new_key,
                                   const ucp_ep_h ep,
                                   const ucp_unpacked_address_t *remote_address,
                                   const unsigned *addr_indices,
                                   ucp_lane_index_t *lane_map);

int ucp_ep_config_lane_is_equal(const ucp_ep_config_key_t *key1,
                                const ucp_ep_config_key_t *key2,
                                ucp_lane_index_t lane);

/**
 * @brief Compare two endpoint configurations.
 *
 * @param [in] key1  First config key to compare.
 * @param [in] key2  Second config key to compare.
 *
 * @return Whether the configurations are equal.
 */
int ucp_ep_config_is_equal(const ucp_ep_config_key_t *key1,
                           const ucp_ep_config_key_t *key2);

void ucp_ep_config_name(ucp_worker_h worker, ucp_worker_cfg_index_t cfg_index,
                        ucs_string_buffer_t *strb);

int ucp_ep_config_get_multi_lane_prio(const ucp_lane_index_t *lanes,
                                      ucp_lane_index_t lane);

size_t ucp_ep_config_get_zcopy_auto_thresh(size_t iovcnt,
                                           const ucs_linear_func_t *reg_cost,
                                           const ucp_context_h context,
                                           double bandwidth);

ucs_status_t ucp_worker_mem_type_eps_create(ucp_worker_h worker);

void ucp_worker_mem_type_eps_destroy(ucp_worker_h worker);

void ucp_worker_mem_type_eps_print_info(ucp_worker_h worker,
                                              FILE *stream);

ucp_wireup_ep_t * ucp_ep_get_cm_wireup_ep(ucp_ep_h ep);

void ucp_ep_get_tl_bitmap(const ucp_ep_config_key_t *key,
                          ucp_tl_bitmap_t *tl_bitmap);

uct_ep_h ucp_ep_get_cm_uct_ep(ucp_ep_h ep);

int ucp_ep_is_cm_local_connected(ucp_ep_h ep);

int ucp_ep_is_local_connected(ucp_ep_h ep);

unsigned ucp_ep_local_disconnect_progress(void *arg);

size_t ucp_ep_tag_offload_min_rndv_thresh(ucp_context_h context,
                                          const ucp_ep_config_key_t *key);

void ucp_ep_config_rndv_zcopy_commit(ucp_lane_index_t lanes_count,
                                     ucp_ep_rndv_zcopy_config_t *rndv_zcopy);

void ucp_ep_get_lane_info_str(ucp_ep_h ucp_ep, ucp_lane_index_t lane,
                              ucs_string_buffer_t *lane_info_strb);

void ucp_ep_config_rndv_zcopy_commit(ucp_lane_index_t lanes_count,
                                     ucp_ep_rndv_zcopy_config_t *rndv_zcopy);

void ucp_ep_invoke_err_cb(ucp_ep_h ep, ucs_status_t status);

ucs_status_t ucp_ep_flush_progress_pending(uct_pending_req_t *self);

void ucp_ep_flush_completion(uct_completion_t *self);

void ucp_ep_flush_request_ff(ucp_request_t *req, ucs_status_t status);

void
ucp_ep_purge_lanes(ucp_ep_h ep, uct_pending_purge_callback_t purge_cb,
                   void *purge_arg);

void ucp_ep_register_disconnect_progress(ucp_request_t *req);

ucp_lane_index_t ucp_ep_lookup_lane(ucp_ep_h ucp_ep, uct_ep_h uct_ep);

void ucp_ep_peer_mem_destroy(ucp_context_h context,
                             ucp_ep_peer_mem_data_t *data);

ucp_ep_peer_mem_data_t*
ucp_ep_peer_mem_get(ucp_context_h context, ucp_ep_h ep, uint64_t address,
                    size_t size, const void *rkey_buf,
                    ucs_memory_type_t local_mem_type,
                    ucp_md_index_t rkey_ptr_md_index);

/**
 * @brief Indicates AM-based keepalive necessity.
 * 
 * @param [in] ep      UCP endpoint to check.
 * @param [in] rsc_idx Resource index to check.
 * @param [in] is_p2p  Flag that indicates whether UCT EP was created as p2p
 *                     (i.e. CONNECT_TO_EP) or not.
 *
 * @return Whether AM-based keepalive is required or not.
 */
int ucp_ep_is_am_keepalive(ucp_ep_h ep, ucp_rsc_index_t rsc_idx, int is_p2p);

/**
 * @brief Do AM-based keepalive operation for a specific UCT EP.
 *
 * @param [in] ucp_ep  UCP Endpoint object to operate keepalive.
 * @param [in] uct_ep  UCT Endpoint object to do keepalive on.
 * @param [in] rsc_idx Resource index to check.
 *
 * @return Status of keepalive operation.
 */
ucs_status_t ucp_ep_do_uct_ep_am_keepalive(ucp_ep_h ucp_ep, uct_ep_h uct_ep,
                                           ucp_rsc_index_t rsc_idx);


/**
 * @brief Purge the protocol request scheduled on a given UCP endpoint.
 *
 * @param [in]     ucp_ep           Endpoint object on which the request should
 *                                  be purged.
 * @param [in]     req              The request to purge.
 * @param [in]     status           Completion status.
 * @param [in]     recursive        Indicates if the function was called from
 *                                  the @ref ucp_ep_req_purge recursively.
 */
void ucp_ep_req_purge(ucp_ep_h ucp_ep, ucp_request_t *req,
                      ucs_status_t status, int recursive);


/**
 * @brief Purge flush and protocol requests scheduled on a given UCP endpoint.
 *
 * @param [in]     ucp_ep           Endpoint object on which requests should be
 *                                  purged.
 * @param [in]     status           Completion status.
 */
void ucp_ep_reqs_purge(ucp_ep_h ucp_ep, ucs_status_t status);


/**
 * @brief Query local and/or remote socket address of endpoint @a ucp_ep.
 *
 * @param [in]     ucp_ep           Endpoint object to query.
 * @param [inout]  attr             Filled with attributes containing socket
 *                                  address of the endpoint.
 *
 * @return Error code as defined by @ref ucs_status_t
 */
ucs_status_t ucp_ep_query_sockaddr(ucp_ep_h ucp_ep, ucp_ep_attr_t *attr);

/**
 * @brief Realloc slow lanes according to current number of lanes.
 *        Slow lanes number will be (new_num_lanes - MAX_FAST_PATH_LANES).
 *        If new slow lanes are added, fill them with NULL.
 *
 * @param [in] ucp_ep           Endpoint object.
 * @param [in] new_num_lanes    Number of total lanes required
 *                              (including fast lanes).
 *
 * @return Error code as defined by @ref ucs_status_t
 */
ucs_status_t ucp_ep_realloc_lanes(ucp_ep_h ep, unsigned new_num_lanes);


/**
 * @brief Set configuration index to the endpoint.
 * 
 * Changing of the configuration index deactivates UCP worker interfaces
 * corresponding to the previous endpoint configuration and activates interfaces
 * of the new configuration.
 *
 * @param [in] ep         Endpoint object.
 * @param [in] cfg_index  Endpoint configuration index.
 * @param [in] reactivate Flag indicating whether to reactivate worker
 *                        interfaces.
 */
void ucp_ep_set_cfg_index(ucp_ep_h ep, ucp_worker_cfg_index_t cfg_index,
                          int reactivate);


/**
 * @brief Progress function for memory specific remote flushing.
 *
 * This call starts and progresses all memory specific remote flushes.
 *
 * @param[in] self        Pending request tracking the flush.
 *
 * @return Error code as defined by @ref ucs_status_t
 */
ucs_status_t ucp_ep_flush_mem_progress(uct_pending_req_t *self);


/**
 * @brief Get the failed lanes from the endpoint configuration.
 *
 * @param [in] key        Endpoint configuration key.
 *
 * @return Bitmask of failed lanes.
 */
ucp_lane_map_t ucp_ep_config_get_failed_lanes(const ucp_ep_config_key_t *key);


/**
 * @brief Clear UCP_LANE_TYPE_FAILED for a subset of previously failed lanes
 *        that have just been recovered and reconfigure the endpoint.
 *
 * @param [in] ep     Endpoint object.
 * @param [in] lanes  Lanes to clear UCP_LANE_TYPE_FAILED for. Must be a
 *                    subset of ucp_ep_get_failed_lanes(ep); the caller is
 *                    responsible for intersecting with the currently-failed
 *                    set. Passing 0 is a no-op.
 *
 * @return Error code as defined by @ref ucs_status_t
 */
ucs_status_t ucp_ep_reconfig_clear_failed_lanes(ucp_ep_h ep,
                                                ucp_lane_map_t lanes);


/**
 * Arm (or re-arm) failed-lane recovery for an endpoint.
 */
ucs_status_t ucp_ep_recovery_arm(ucp_ep_h ep);


/**
 * Notify recovery progress that a lanes-address reply was received.
 */
void ucp_ep_recovery_on_reply_received(ucp_ep_h ep);


/**
 * Progress function for failed lanes recovery.
 *
 * @return 0 if the @a ep is recovered, continue normal keep alive checks,
 *         otherwise non-zero - the @a ep is under recovery, fully failed or
 *         other reason to skip keepalive round.
 */
int ucp_ep_recovery_progress(ucp_ep_h ep);


/**
 * Rebuild UCT endpoints for the given set of currently-failed lanes using the
 * peer addresses in @a remote_address. Returns the bitmap of lanes rebuilt.
 */
ucp_lane_map_t
ucp_ep_recovery_rebuild_lanes(ucp_ep_h ep, ucp_lane_map_t lanes_to_rebuild,
                              const ucp_unpacked_address_t *remote_address);


/**
 * Mark a subset of lanes as UCP_LANE_TYPE_FAILED and arm recovery.
 */
ucs_status_t ucp_ep_failover_reconfig(ucp_ep_h ucp_ep,
                                      ucp_lane_map_t failed_lanes,
                                      ucs_status_t discard_status);


/**
 * @brief Update EP configuration and rkey configuration according to the latest
 * interfaces state.
 *
 * @param [in] ep      Endpoint object.
 * @param [in] rkey    Rkey object.
 *
 * @return Error code as defined by @ref ucs_status_t
 */
ucs_status_t ucp_ep_update_rkey_config(ucp_ep_h ep, ucp_rkey_h rkey);

ucs_status_t ucp_stub_iface_open(ucs_status_t status, uct_iface_h *iface_p);

#endif