mirrors/versitygw
1
0
Fork 0
mirror of https://github.com/versity/versitygw.git synced 2026-09-22 07:54:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

rdma: port the hipObject v2 RC session core

Browse Source 
Port the hipObject v2 reliable-connection session core into
cuwrapper/rc: the session table and state machine, the wire
codec for the hipobj-rc-v2 headers, request parsing, the
injectable clock and randomness sources, the transport layer
(QP/CQ lifecycle, RTR/RTS transitions, staging registration),
the data phase (RDMA write with immediate for GET, receive with
immediate for PUT), the RDMA token codec, and the dynamically
loaded ibverbs shim (ibv-core.h plus the dlopen host binding).

The sources are a port of the upstream hipObject v2 core, kept
close to the original so the two trees can be diffed during
review. Nothing links against them yet; a Makefile rule builds
the objects into rdma/librcserver.a for the ABI layer that
follows.

Signed-off-by: Jihyeon Gim <potatogim@potatogim.net>
This commit is contained in:
Jihyeon Gim
2026-08-30 00:00:29 +09:00
parent 83b5e7041c
commit efa0309da4
24 changed files with 4070 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Makefile
+14
View File
@@ -54,6 +54,10 @@ CUOBJCLIENT_WRAPPER_LIB=rdma/libcuobjclientwrapper.a
CUOBJCLIENT_CGO_CFLAGS=-I$(CUOBJ_CLIENT_INC_DIR) -I$(CUOBJ_CUDA_INC_DIR)
CUOBJCLIENT_CGO_LDFLAGS=-L$(CUOBJ_LIB_DIR) -Wl,-rpath,$(CUOBJ_LIB_DIR)
HOSTCLIENT_WRAPPER_LIB=rdma/libhostclientwrapper.a
RCSERVER_LIB=rdma/librcserver.a
RCSERVER_SRCS=$(wildcard cuwrapper/rc/*.cpp)
RCSERVER_OBJS=$(RCSERVER_SRCS:.cpp=.o)
RCSERVER_CXXFLAGS=-fPIC -std=c++17 -Icuwrapper/rc
VERSION := $(shell if test -e VERSION; then cat VERSION; else git describe --abbrev=0 --tags HEAD; fi)
BUILD := $(shell git rev-parse --short HEAD || echo release-rpm)
@@ -93,6 +97,15 @@ $(HOSTCLIENT_WRAPPER_LIB): cuwrapper/rdma_host_client_wrapper.cpp cuwrapper/rdma
$(AR) rcs $(HOSTCLIENT_WRAPPER_LIB) cuwrapper/rdma_host_client_wrapper.o
rm -f cuwrapper/rdma_host_client_wrapper.o
$(RCSERVER_LIB): $(RCSERVER_OBJS)
$(AR) rcs $@ $^
rm -f $(RCSERVER_OBJS)
cuwrapper/rc/%.o: cuwrapper/rc/%.cpp
$(CXX) $(RCSERVER_CXXFLAGS) -c -o $@ $<
.INTERMEDIATE: $(RCSERVER_OBJS)
.PHONY: vgwrdma
vgwrdma: $(VGWRDMA_WRAPPER_LIB)
CGO_ENABLED=1 \
@@ -171,6 +184,7 @@ cleanall: clean
rm -f $(VGWRDMA_WRAPPER_LIB)
rm -f $(CUOBJCLIENT_WRAPPER_LIB)
rm -f $(HOSTCLIENT_WRAPPER_LIB)
rm -f $(RCSERVER_LIB)
rm -f versitygw-*.tar
rm -f versitygw-*.tar.gz
cuwrapper/rc/ibv-core.h
+725
View File
@@ -0,0 +1,725 @@
/*
* Copyright (c) 2004, 2005 Topspin Communications. All rights reserved.
* Copyright (c) 2004, 2011-2012 Intel Corporation. All rights reserved.
* Copyright (c) 2005, 2006, 2007 Cisco Systems, Inc. All rights reserved.
* Copyright (c) 2005 PathScale, Inc. All rights reserved.
* Copyright (c) 2020 Intel Corporation. All rights reserved.
* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
*
* This software is available to you under a choice of one of two
* licenses. You may choose to be licensed under the terms of the GNU
* General Public License (GPL) Version 2, available from the file
* COPYING in the main directory of this source tree, or the
* OpenIB.org BSD license below:
*
* Redistribution and use in source and binary forms, with or
* without modification, are permitted provided that the following
* conditions are met:
*
* - Redistributions of source code must retain the above
* copyright notice, this list of conditions and the following
* disclaimer.
*
* - Redistributions in binary form must reproduce the above
* copyright notice, this list of conditions and the following
* disclaimer in the documentation and/or other materials
* provided with the distribution.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
* EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
* MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
* NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
* BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
* ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
* CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*
* InfiniBand verbs type definitions for dynamic
* libibverbs loading.
*
* Derived from ROCm/rocSHMEM src/gda/ibv_core.hpp via
* rocm-xio src/common/ibv-core.hpp.
*/
#pragma once
#include <cerrno>
#include <cstddef>
#include <cstdint>
#include <limits>
#include <linux/types.h>
#include <pthread.h>
/* -------------------------------------------------------------------------
* 1. ib_uverbs_access_flags enum and IBV_ACCESS_OPTIONAL_* macros
* ------------------------------------------------------------------------- */
enum ib_uverbs_access_flags {
IB_UVERBS_ACCESS_LOCAL_WRITE = 1,
IB_UVERBS_ACCESS_REMOTE_WRITE = (1 << 1),
IB_UVERBS_ACCESS_REMOTE_READ = (1 << 2),
IB_UVERBS_ACCESS_REMOTE_ATOMIC = (1 << 3),
IB_UVERBS_ACCESS_MW_BIND = (1 << 4),
IB_UVERBS_ACCESS_ZERO_BASED = (1 << 5),
IB_UVERBS_ACCESS_ON_DEMAND = (1 << 6),
IB_UVERBS_ACCESS_HUGETLB = (1 << 7),
IB_UVERBS_ACCESS_FLUSH_GLOBAL = (1 << 8),
IB_UVERBS_ACCESS_FLUSH_PERSISTENT = (1 << 9),
IB_UVERBS_ACCESS_RELAXED_ORDERING = (1 << 20),
};
#define IBV_ACCESS_OPTIONAL_FIRST (1 << 20)
#define IBV_ACCESS_OPTIONAL_LAST (1 << 29)
#define IBV_ACCESS_OPTIONAL_RANGE (((1ULL << 30) - 1) & ~((1ULL << 20) - 1))
/* -------------------------------------------------------------------------
* 2. union ibv_gid
* ------------------------------------------------------------------------- */
union ibv_gid {
uint8_t raw[16];
struct {
__be64 subnet_prefix;
__be64 interface_id;
} global;
};
/* -------------------------------------------------------------------------
* 3. ibv_gid_type enum
* ------------------------------------------------------------------------- */
enum ibv_gid_type {
IBV_GID_TYPE_IB,
IBV_GID_TYPE_ROCE_V1,
IBV_GID_TYPE_ROCE_V2,
};
/* -------------------------------------------------------------------------
* 4. ibv_gid_entry struct
* ------------------------------------------------------------------------- */
struct ibv_gid_entry {
union ibv_gid gid;
uint32_t gid_index;
uint32_t port_num;
uint32_t gid_type;
uint32_t ndev_ifindex;
};
/* -------------------------------------------------------------------------
* 5. ibv_node_type, ibv_transport_type, ibv_atomic_cap enums
* ------------------------------------------------------------------------- */
enum ibv_node_type {
IBV_NODE_UNKNOWN = -1,
IBV_NODE_CA = 1,
IBV_NODE_SWITCH,
IBV_NODE_ROUTER,
IBV_NODE_RNIC,
IBV_NODE_USNIC,
IBV_NODE_USNIC_UDP,
IBV_NODE_UNSPECIFIED,
};
enum ibv_transport_type {
IBV_TRANSPORT_UNKNOWN = -1,
IBV_TRANSPORT_IB = 0,
IBV_TRANSPORT_IWARP,
IBV_TRANSPORT_USNIC,
IBV_TRANSPORT_USNIC_UDP,
IBV_TRANSPORT_UNSPECIFIED,
};
enum ibv_atomic_cap { IBV_ATOMIC_NONE, IBV_ATOMIC_HCA, IBV_ATOMIC_GLOB };
/* -------------------------------------------------------------------------
* 6. ibv_device_attr struct (full)
* ------------------------------------------------------------------------- */
struct ibv_device_attr {
char fw_ver[64];
__be64 node_guid;
__be64 sys_image_guid;
uint64_t max_mr_size;
uint64_t page_size_cap;
uint32_t vendor_id;
uint32_t vendor_part_id;
uint32_t hw_ver;
int max_qp;
int max_qp_wr;
unsigned int device_cap_flags;
int max_sge;
int max_sge_rd;
int max_cq;
int max_cqe;
int max_mr;
int max_pd;
int max_qp_rd_atom;
int max_ee_rd_atom;
int max_res_rd_atom;
int max_qp_init_rd_atom;
int max_ee_init_rd_atom;
enum ibv_atomic_cap atomic_cap;
int max_ee;
int max_rdd;
int max_mw;
int max_raw_ipv6_qp;
int max_raw_ethy_qp;
int max_mcast_grp;
int max_mcast_qp_attach;
int max_total_mcast_qp_attach;
int max_ah;
int max_fmr;
int max_map_per_fmr;
int max_srq;
int max_srq_wr;
int max_srq_sge;
uint16_t max_pkeys;
uint8_t local_ca_ack_delay;
uint8_t phys_port_cnt;
};
/* -------------------------------------------------------------------------
* 7. ibv_mtu enum, ibv_port_state enum, link layer constants
* ------------------------------------------------------------------------- */
enum ibv_mtu {
IBV_MTU_256 = 1,
IBV_MTU_512 = 2,
IBV_MTU_1024 = 3,
IBV_MTU_2048 = 4,
IBV_MTU_4096 = 5
};
enum ibv_port_state {
IBV_PORT_NOP = 0,
IBV_PORT_DOWN = 1,
IBV_PORT_INIT = 2,
IBV_PORT_ARMED = 3,
IBV_PORT_ACTIVE = 4,
IBV_PORT_ACTIVE_DEFER = 5
};
enum {
IBV_LINK_LAYER_UNSPECIFIED,
IBV_LINK_LAYER_INFINIBAND,
IBV_LINK_LAYER_ETHERNET,
};
/* -------------------------------------------------------------------------
* 8. ibv_port_attr struct (full)
* ------------------------------------------------------------------------- */
struct ibv_port_attr {
enum ibv_port_state state;
enum ibv_mtu max_mtu;
enum ibv_mtu active_mtu;
int gid_tbl_len;
uint32_t port_cap_flags;
uint32_t max_msg_sz;
uint32_t bad_pkey_cntr;
uint32_t qkey_viol_cntr;
uint16_t pkey_tbl_len;
uint16_t lid;
uint16_t sm_lid;
uint8_t lmc;
uint8_t max_vl_num;
uint8_t sm_sl;
uint8_t subnet_timeout;
uint8_t init_type_reply;
uint8_t active_width;
uint8_t active_speed;
uint8_t phys_state;
uint8_t link_layer;
uint8_t flags;
uint16_t port_cap_flags2;
uint32_t active_speed_ex;
};
/* -------------------------------------------------------------------------
* 9. ibv_wc_status enum, ibv_wc struct
* ------------------------------------------------------------------------- */
/* Completion flags (wc_flags in struct ibv_wc). */
enum ibv_wc_flags {
IBV_WC_GRH = 1 << 0,
IBV_WC_WITH_IMM = 1 << 1,
IBV_WC_IP_CSUM_OK = 1 << 2,
IBV_WC_WITH_INV = 1 << 3,
IBV_WC_TM_SYNC_REQ = 1 << 4,
IBV_WC_TM_DATA_VALID = 1 << 5,
IBV_WC_TM_MATCH_REQ = 1 << 6,
IBV_WC_TM_DATA_VALID_2 = 1 << 7
};
enum ibv_wc_opcode {
IBV_WC_SEND,
IBV_WC_RDMA_WRITE,
IBV_WC_RDMA_READ,
IBV_WC_COMP_SWAP,
IBV_WC_FETCH_ADD,
IBV_WC_BIND_MW,
IBV_WC_LOCAL_INV,
IBV_WC_TSO,
IBV_WC_FLUSH,
IBV_WC_ATOMIC_WRITE = 9,
IBV_WC_RECV = 1 << 7,
IBV_WC_RECV_RDMA_WITH_IMM,
};
enum ibv_wc_status {
IBV_WC_SUCCESS,
IBV_WC_LOC_LEN_ERR,
IBV_WC_LOC_QP_OP_ERR,
IBV_WC_LOC_EEC_OP_ERR,
IBV_WC_LOC_PROT_ERR,
IBV_WC_WR_FLUSH_ERR,
IBV_WC_MW_BIND_ERR,
IBV_WC_BAD_RESP_ERR,
IBV_WC_LOC_ACCESS_ERR,
IBV_WC_REM_INV_REQ_ERR,
IBV_WC_REM_ACCESS_ERR,
IBV_WC_REM_OP_ERR,
IBV_WC_RETRY_EXC_ERR,
IBV_WC_RNR_RETRY_EXC_ERR,
IBV_WC_LOC_RDD_VIOL_ERR,
IBV_WC_REM_INV_RD_REQ_ERR,
IBV_WC_REM_ABORT_ERR,
IBV_WC_INV_EECN_ERR,
IBV_WC_INV_EEC_STATE_ERR,
IBV_WC_FATAL_ERR,
IBV_WC_RESP_TIMEOUT_ERR,
IBV_WC_GENERAL_ERR,
IBV_WC_TM_ERR,
IBV_WC_TM_RNDV_INCOMPLETE,
};
struct ibv_wc {
uint64_t wr_id;
enum ibv_wc_status status;
enum ibv_wc_opcode opcode;
uint32_t vendor_err;
uint32_t byte_len;
union {
__be32 imm_data;
uint32_t invalidated_rkey;
};
uint32_t qp_num;
uint32_t src_qp;
unsigned int wc_flags;
uint16_t pkey_index;
uint16_t slid;
uint8_t sl;
uint8_t dlid_path_bits;
};
/* -------------------------------------------------------------------------
* 10. ibv_access_flags enum
* ------------------------------------------------------------------------- */
enum ibv_access_flags {
IBV_ACCESS_LOCAL_WRITE = 1,
IBV_ACCESS_REMOTE_WRITE = (1 << 1),
IBV_ACCESS_REMOTE_READ = (1 << 2),
IBV_ACCESS_REMOTE_ATOMIC = (1 << 3),
IBV_ACCESS_MW_BIND = (1 << 4),
IBV_ACCESS_ZERO_BASED = (1 << 5),
IBV_ACCESS_ON_DEMAND = (1 << 6),
IBV_ACCESS_HUGETLB = (1 << 7),
IBV_ACCESS_FLUSH_GLOBAL = (1 << 8),
IBV_ACCESS_FLUSH_PERSISTENT = (1 << 9),
IBV_ACCESS_RELAXED_ORDERING = IBV_ACCESS_OPTIONAL_FIRST,
};
/* -------------------------------------------------------------------------
* 11. ibv_pd struct, ibv_mr struct
* ------------------------------------------------------------------------- */
struct ibv_context;
struct ibv_pd {
struct ibv_context* context;
uint32_t handle;
};
struct ibv_mr {
struct ibv_context* context;
struct ibv_pd* pd;
void* addr;
size_t length;
uint32_t handle;
uint32_t lkey;
uint32_t rkey;
};
/* -------------------------------------------------------------------------
* 12. ibv_global_route, ibv_ah_attr structs
* ------------------------------------------------------------------------- */
struct ibv_global_route {
union ibv_gid dgid;
uint32_t flow_label;
uint8_t sgid_index;
uint8_t hop_limit;
uint8_t traffic_class;
};
struct ibv_ah_attr {
struct ibv_global_route grh;
uint16_t dlid;
uint8_t sl;
uint8_t src_path_bits;
uint8_t static_rate;
uint8_t is_global;
uint8_t port_num;
};
/* -------------------------------------------------------------------------
* 13. ibv_qp_type enum
* ------------------------------------------------------------------------- */
enum ibv_qp_type {
IBV_QPT_RC = 2,
IBV_QPT_UC,
IBV_QPT_UD,
IBV_QPT_RAW_PACKET = 8,
IBV_QPT_XRC_SEND = 9,
IBV_QPT_XRC_RECV,
IBV_QPT_DRIVER = 0xff,
};
/* -------------------------------------------------------------------------
* 14. ibv_qp_cap struct, ibv_qp_init_attr struct
* ------------------------------------------------------------------------- */
struct ibv_cq;
struct ibv_qp_cap {
uint32_t max_send_wr;
uint32_t max_recv_wr;
uint32_t max_send_sge;
uint32_t max_recv_sge;
uint32_t max_inline_data;
};
struct ibv_comp_channel;
struct ibv_srq;
struct ibv_qp_init_attr {
void* qp_context;
struct ibv_cq* send_cq;
struct ibv_cq* recv_cq;
struct ibv_srq* srq;
struct ibv_qp_cap cap;
enum ibv_qp_type qp_type;
int sq_sig_all;
};
/* -------------------------------------------------------------------------
* 15. ibv_qp_attr_mask enum
* ------------------------------------------------------------------------- */
enum ibv_qp_attr_mask {
IBV_QP_STATE = 1 << 0,
IBV_QP_CUR_STATE = 1 << 1,
IBV_QP_EN_SQD_ASYNC_NOTIFY = 1 << 2,
IBV_QP_ACCESS_FLAGS = 1 << 3,
IBV_QP_PKEY_INDEX = 1 << 4,
IBV_QP_PORT = 1 << 5,
IBV_QP_QKEY = 1 << 6,
IBV_QP_AV = 1 << 7,
IBV_QP_PATH_MTU = 1 << 8,
IBV_QP_TIMEOUT = 1 << 9,
IBV_QP_RETRY_CNT = 1 << 10,
IBV_QP_RNR_RETRY = 1 << 11,
IBV_QP_RQ_PSN = 1 << 12,
IBV_QP_MAX_QP_RD_ATOMIC = 1 << 13,
IBV_QP_ALT_PATH = 1 << 14,
IBV_QP_MIN_RNR_TIMER = 1 << 15,
IBV_QP_SQ_PSN = 1 << 16,
IBV_QP_MAX_DEST_RD_ATOMIC = 1 << 17,
IBV_QP_PATH_MIG_STATE = 1 << 18,
IBV_QP_CAP = 1 << 19,
IBV_QP_DEST_QPN = 1 << 20,
IBV_QP_RATE_LIMIT = 1 << 25,
};
/* -------------------------------------------------------------------------
* 16. ibv_qp_state enum, ibv_mig_state enum
* ------------------------------------------------------------------------- */
enum ibv_qp_state {
IBV_QPS_RESET,
IBV_QPS_INIT,
IBV_QPS_RTR,
IBV_QPS_RTS,
IBV_QPS_SQD,
IBV_QPS_SQE,
IBV_QPS_ERR,
IBV_QPS_UNKNOWN
};
enum ibv_mig_state { IBV_MIG_MIGRATED, IBV_MIG_REARM, IBV_MIG_ARMED };
/* -------------------------------------------------------------------------
* 17. ibv_qp_attr struct (full)
* ------------------------------------------------------------------------- */
struct ibv_qp_attr {
enum ibv_qp_state qp_state;
enum ibv_qp_state cur_qp_state;
enum ibv_mtu path_mtu;
enum ibv_mig_state path_mig_state;
uint32_t qkey;
uint32_t rq_psn;
uint32_t sq_psn;
uint32_t dest_qp_num;
unsigned int qp_access_flags;
struct ibv_qp_cap cap;
struct ibv_ah_attr ah_attr;
struct ibv_ah_attr alt_ah_attr;
uint16_t pkey_index;
uint16_t alt_pkey_index;
uint8_t en_sqd_async_notify;
uint8_t sq_draining;
uint8_t max_rd_atomic;
uint8_t max_dest_rd_atomic;
uint8_t min_rnr_timer;
uint8_t port_num;
uint8_t timeout;
uint8_t retry_cnt;
uint8_t rnr_retry;
uint8_t alt_port_num;
uint8_t alt_timeout;
uint32_t rate_limit;
};
/* -------------------------------------------------------------------------
* 18. ibv_qp struct
* ------------------------------------------------------------------------- */
struct ibv_qp {
struct ibv_context* context;
void* qp_context;
struct ibv_pd* pd;
struct ibv_cq* send_cq;
struct ibv_cq* recv_cq;
struct ibv_srq* srq;
uint32_t handle;
uint32_t qp_num;
enum ibv_qp_state state;
enum ibv_qp_type qp_type;
pthread_mutex_t mutex;
pthread_cond_t cond;
uint32_t events_completed;
};
/* -------------------------------------------------------------------------
* 19. ibv_wr_opcode enum
* ------------------------------------------------------------------------- */
enum ibv_wr_opcode {
IBV_WR_RDMA_WRITE,
IBV_WR_RDMA_WRITE_WITH_IMM,
IBV_WR_SEND,
IBV_WR_SEND_WITH_IMM,
IBV_WR_RDMA_READ,
IBV_WR_ATOMIC_CMP_AND_SWP,
IBV_WR_ATOMIC_FETCH_AND_ADD,
IBV_WR_LOCAL_INV,
IBV_WR_BIND_MW,
IBV_WR_SEND_WITH_INV,
IBV_WR_TSO,
IBV_WR_DRIVER1,
IBV_WR_FLUSH = 14,
IBV_WR_ATOMIC_WRITE = 15,
};
/* -------------------------------------------------------------------------
* 20. ibv_send_flags enum
* ------------------------------------------------------------------------- */
enum ibv_send_flags {
IBV_SEND_FENCE = 1 << 0,
IBV_SEND_SIGNALED = 1 << 1,
IBV_SEND_SOLICITED = 1 << 2,
IBV_SEND_INLINE = 1 << 3,
};
/* -------------------------------------------------------------------------
* 21. ibv_sge struct
* ------------------------------------------------------------------------- */
struct ibv_sge {
uint64_t addr;
uint32_t length;
uint32_t lkey;
};
/* -------------------------------------------------------------------------
* 22. ibv_send_wr struct with union for rdma/atomic ops
* ------------------------------------------------------------------------- */
struct ibv_ah;
struct ibv_send_wr {
uint64_t wr_id;
struct ibv_send_wr* next;
struct ibv_sge* sg_list;
int num_sge;
enum ibv_wr_opcode opcode;
unsigned int send_flags;
union {
__be32 imm_data;
uint32_t invalidate_rkey;
};
union {
struct {
uint64_t remote_addr;
uint32_t rkey;
} rdma;
struct {
uint64_t remote_addr;
uint64_t compare_add;
uint64_t swap;
uint32_t rkey;
} atomic;
struct {
struct ibv_ah* ah;
uint32_t remote_qpn;
uint32_t remote_qkey;
} ud;
} wr;
};
/* -------------------------------------------------------------------------
* 23. ibv_recv_wr struct
* ------------------------------------------------------------------------- */
struct ibv_recv_wr {
uint64_t wr_id;
struct ibv_recv_wr* next;
struct ibv_sge* sg_list;
int num_sge;
};
/* -------------------------------------------------------------------------
* 24. IBV_SYSFS_NAME_MAX, IBV_SYSFS_PATH_MAX
* ------------------------------------------------------------------------- */
#define IBV_SYSFS_NAME_MAX 64
#define IBV_SYSFS_PATH_MAX 256
/* -------------------------------------------------------------------------
* 25. _ibv_device_ops, ibv_device struct
* ------------------------------------------------------------------------- */
struct _ibv_device_ops {
struct ibv_context* (*_dummy1)(struct ibv_device* device, int cmd_fd);
void (*_dummy2)(struct ibv_context* context);
};
struct ibv_device {
struct _ibv_device_ops _ops;
enum ibv_node_type node_type;
enum ibv_transport_type transport_type;
char name[IBV_SYSFS_NAME_MAX];
char dev_name[IBV_SYSFS_NAME_MAX];
char dev_path[IBV_SYSFS_PATH_MAX];
char ibdev_path[IBV_SYSFS_PATH_MAX];
};
/* -------------------------------------------------------------------------
* 26. Forward declare ibv_comp_channel and ibv_srq (already above)
* ------------------------------------------------------------------------- */
struct ibv_comp_channel {
struct ibv_context* context;
int fd;
int refcnt;
};
struct ibv_srq {
struct ibv_context* context;
void* srq_context;
struct ibv_pd* pd;
uint32_t handle;
pthread_mutex_t mutex;
pthread_cond_t cond;
uint32_t events_completed;
};
/* -------------------------------------------------------------------------
* 27. ibv_cq struct
* ------------------------------------------------------------------------- */
struct ibv_cq {
struct ibv_context* context;
struct ibv_comp_channel* channel;
void* cq_context;
uint32_t handle;
int cqe;
pthread_mutex_t mutex;
pthread_cond_t cond;
uint32_t comp_events_completed;
uint32_t async_events_completed;
};
/* -------------------------------------------------------------------------
* 28. ibv_context_ops struct
* Mirrors the layout of struct ibv_context_ops from libibverbs verbs.h.
* poll_cq, post_send and post_recv are static inline functions in
* verbs.h that delegate to these ops entries; rdma-core does not
* export them as dynamic symbols.
* ------------------------------------------------------------------------- */
struct ibv_context_ops {
void* _compat_query_device;
void* _compat_query_port;
void* _compat_alloc_pd;
void* _compat_dealloc_pd;
void* _compat_reg_mr;
void* _compat_rereg_mr;
void* _compat_dereg_mr;
void* alloc_mw;
void* bind_mw;
void* dealloc_mw;
void* _compat_create_cq;
int (*poll_cq)(struct ibv_cq* cq, int num_entries, struct ibv_wc* wc);
void* req_notify_cq;
void* _compat_cq_event;
void* _compat_resize_cq;
void* _compat_destroy_cq;
void* _compat_create_srq;
void* _compat_modify_srq;
void* _compat_query_srq;
void* _compat_destroy_srq;
void* post_srq_recv;
void* _compat_create_qp;
void* _compat_query_qp;
void* _compat_modify_qp;
void* _compat_destroy_qp;
int (*post_send)(struct ibv_qp* qp, struct ibv_send_wr* wr,
struct ibv_send_wr** bad_wr);
int (*post_recv)(struct ibv_qp* qp, struct ibv_recv_wr* wr,
struct ibv_recv_wr** bad_wr);
void* _compat_create_ah;
void* _compat_destroy_ah;
void* _compat_attach_mcast;
void* _compat_detach_mcast;
void* _compat_async_event;
};
/* -------------------------------------------------------------------------
* 29. ibv_context struct
* ------------------------------------------------------------------------- */
struct ibv_context {
struct ibv_device* device;
struct ibv_context_ops ops;
int cmd_fd;
int async_fd;
int num_comp_vectors;
pthread_mutex_t mutex;
void* abi_compat;
};
cuwrapper/rc/rc_ibv_host.cpp
+105
View File
@@ -0,0 +1,105 @@
/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
*
* SPDX-License-Identifier: MIT
*/
#include "rc_ibv_host.h"
namespace hipObj {
IBVWrapper &IBVWrapper::instance() {
static IBVWrapper w;
return w;
}
IBVWrapper ibv __attribute__((init_priority(400)));
bool IBVWrapper::ensureLoaded() {
std::lock_guard<std::mutex> guard(mtx_);
if (loaded) {
return true;
}
handle_ = dlopen("libibverbs.so.1", RTLD_NOW | RTLD_GLOBAL);
if (handle_ == nullptr) {
handle_ = dlopen("libibverbs.so", RTLD_NOW | RTLD_GLOBAL);
}
if (handle_ == nullptr) {
return false;
}
auto load = [&](const char *name) -> void * {
return dlsym(handle_, name);
};
funcs_.get_device_list =
reinterpret_cast<struct ibv_device **(*)(int *)>(load("ibv_get_device_list"));
funcs_.free_device_list =
reinterpret_cast<void (*)(struct ibv_device **)>(load("ibv_free_device_list"));
funcs_.open_device = reinterpret_cast<struct ibv_context *(*)(struct ibv_device *)>(
load("ibv_open_device"));
funcs_.close_device = reinterpret_cast<int (*)(struct ibv_context *)>(
load("ibv_close_device"));
funcs_.alloc_pd = reinterpret_cast<struct ibv_pd *(*)(struct ibv_context *)>(
load("ibv_alloc_pd"));
funcs_.dealloc_pd = reinterpret_cast<int (*)(struct ibv_pd *)>(
load("ibv_dealloc_pd"));
funcs_.reg_mr = reinterpret_cast<struct ibv_mr *(*)(struct ibv_pd *, void *,
size_t, int)>(load("ibv_reg_mr"));
/* Host-only build: both spellings land on the plain verbs call. */
funcs_.reg_mr_host = funcs_.reg_mr;
funcs_.dereg_mr = reinterpret_cast<int (*)(struct ibv_mr *)>(
load("ibv_dereg_mr"));
funcs_.create_cq = reinterpret_cast<struct ibv_cq *(*)(
struct ibv_context *, int, void *, struct ibv_comp_channel *, int)>(
load("ibv_create_cq"));
funcs_.destroy_cq = reinterpret_cast<int (*)(struct ibv_cq *)>(
load("ibv_destroy_cq"));
funcs_.create_qp = reinterpret_cast<struct ibv_qp *(*)(
struct ibv_pd *, struct ibv_qp_init_attr *)>(load("ibv_create_qp"));
funcs_.destroy_qp = reinterpret_cast<int (*)(struct ibv_qp *)>(
load("ibv_destroy_qp"));
funcs_.modify_qp = reinterpret_cast<int (*)(struct ibv_qp *,
struct ibv_qp_attr *, int)>(
load("ibv_modify_qp"));
funcs_.poll_cq = reinterpret_cast<int (*)(struct ibv_cq *, int,
struct ibv_wc *)>(load("ibv_poll_cq"));
funcs_.query_device = reinterpret_cast<int (*)(
struct ibv_context *, struct ibv_device_attr *)>(load("ibv_query_device"));
funcs_.query_port = reinterpret_cast<int (*)(
struct ibv_context *, uint8_t, struct ibv_port_attr *)>(
load("ibv_query_port"));
funcs_.query_gid = reinterpret_cast<int (*)(
struct ibv_context *, uint8_t, int, union ibv_gid *)>(
load("ibv_query_gid"));
funcs_.post_recv = reinterpret_cast<int (*)(struct ibv_qp *,
struct ibv_recv_wr *, struct ibv_recv_wr **)>(load("ibv_post_recv"));
funcs_.post_send = reinterpret_cast<int (*)(struct ibv_qp *,
struct ibv_send_wr *, struct ibv_send_wr **)>(load("ibv_post_send"));
loaded = funcs_.get_device_list != nullptr && funcs_.open_device != nullptr &&
funcs_.alloc_pd != nullptr && funcs_.create_qp != nullptr &&
funcs_.modify_qp != nullptr && funcs_.poll_cq != nullptr;
if (loaded) {
/* Mirror into the member seam for direct ibv.x() calls. */
get_device_list = funcs_.get_device_list;
free_device_list = funcs_.free_device_list;
open_device = funcs_.open_device;
close_device = funcs_.close_device;
alloc_pd = funcs_.alloc_pd;
dealloc_pd = funcs_.dealloc_pd;
reg_mr = funcs_.reg_mr;
reg_mr_host = funcs_.reg_mr_host;
dereg_mr = funcs_.dereg_mr;
create_cq = funcs_.create_cq;
destroy_cq = funcs_.destroy_cq;
create_qp = funcs_.create_qp;
destroy_qp = funcs_.destroy_qp;
modify_qp = funcs_.modify_qp;
poll_cq = funcs_.poll_cq;
query_device = funcs_.query_device;
query_port = funcs_.query_port;
query_gid = funcs_.query_gid;
post_recv = funcs_.post_recv;
post_send = funcs_.post_send;
}
return loaded;
}
} // namespace hipObj
cuwrapper/rc/rc_ibv_host.h
+112
View File
@@ -0,0 +1,112 @@
/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
*
* SPDX-License-Identifier: MIT
*/
/* Host-only RDMA verbs shim for the RC session core.
*
* The upstream hipObject ibv-wrapper links the GPU runtime for
* dmabuf-based memory registration. The gateway runs the RC data
* plane on host memory only, so this shim loads the plain
* libibverbs entry points with dlopen/dlsym (keeping the gateway
* free of a hard library dependency at link time) and exposes the
* same function-table seam the ported core expects. All ibv
* struct/enum types come from the vendored ibv-core.h, so real
* verbs headers are not included here.
*/
#pragma once
#include <dlfcn.h>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <mutex>
#include "ibv-core.h"
namespace hipObj {
/* Function table type mirrors the upstream seam so the ported
* session core compiles unchanged. */
struct IbvFuncs {
struct ibv_device **(*get_device_list)(int *);
void (*free_device_list)(struct ibv_device **);
struct ibv_context *(*open_device)(struct ibv_device *);
int (*close_device)(struct ibv_context *);
struct ibv_pd *(*alloc_pd)(struct ibv_context *);
int (*dealloc_pd)(struct ibv_pd *);
struct ibv_mr *(*reg_mr)(struct ibv_pd *, void *, size_t, int);
struct ibv_mr *(*reg_mr_host)(struct ibv_pd *, void *, size_t, int);
int (*dereg_mr)(struct ibv_mr *);
struct ibv_cq *(*create_cq)(struct ibv_context *, int, void *,
struct ibv_comp_channel *, int);
int (*destroy_cq)(struct ibv_cq *);
struct ibv_qp *(*create_qp)(struct ibv_pd *,
struct ibv_qp_init_attr *);
int (*destroy_qp)(struct ibv_qp *);
int (*modify_qp)(struct ibv_qp *, struct ibv_qp_attr *, int);
int (*poll_cq)(struct ibv_cq *, int, struct ibv_wc *);
int (*query_device)(struct ibv_context *, struct ibv_device_attr *);
int (*query_port)(struct ibv_context *, uint8_t,
struct ibv_port_attr *);
int (*query_gid)(struct ibv_context *, uint8_t, int, union ibv_gid *);
int (*post_recv)(struct ibv_qp *, struct ibv_recv_wr *,
struct ibv_recv_wr **);
int (*post_send)(struct ibv_qp *, struct ibv_send_wr *,
struct ibv_send_wr **);
};
class IBVWrapper {
public:
IbvFuncs funcs_{};
bool loaded = false;
/* Seam passthrough so the ported core keeps calling ibv.x()
* directly (upstream ibv-wrapper exposed the verbs entry
* points as members). */
struct ibv_device **(*get_device_list)(int *) = nullptr;
void (*free_device_list)(struct ibv_device **) = nullptr;
struct ibv_context *(*open_device)(struct ibv_device *) = nullptr;
int (*close_device)(struct ibv_context *) = nullptr;
struct ibv_pd *(*alloc_pd)(struct ibv_context *) = nullptr;
int (*dealloc_pd)(struct ibv_pd *) = nullptr;
struct ibv_mr *(*reg_mr)(struct ibv_pd *, void *, size_t, int) = nullptr;
struct ibv_mr *(*reg_mr_host)(struct ibv_pd *, void *, size_t, int) = nullptr;
int (*dereg_mr)(struct ibv_mr *) = nullptr;
struct ibv_cq *(*create_cq)(struct ibv_context *, int, void *,
struct ibv_comp_channel *, int) = nullptr;
int (*destroy_cq)(struct ibv_cq *) = nullptr;
struct ibv_qp *(*create_qp)(struct ibv_pd *,
struct ibv_qp_init_attr *) = nullptr;
int (*destroy_qp)(struct ibv_qp *) = nullptr;
int (*modify_qp)(struct ibv_qp *, struct ibv_qp_attr *, int) = nullptr;
int (*poll_cq)(struct ibv_cq *, int, struct ibv_wc *) = nullptr;
int (*query_device)(struct ibv_context *, struct ibv_device_attr *) = nullptr;
int (*query_port)(struct ibv_context *, uint8_t,
struct ibv_port_attr *) = nullptr;
int (*query_gid)(struct ibv_context *, uint8_t, int, union ibv_gid *) = nullptr;
int (*post_recv)(struct ibv_qp *, struct ibv_recv_wr *,
struct ibv_recv_wr **) = nullptr;
int (*post_send)(struct ibv_qp *, struct ibv_send_wr *,
struct ibv_send_wr **) = nullptr;
static IBVWrapper &instance();
#ifdef HIPOBJ_UNIT_TESTS
IbvFuncs &funcsForTest() { return funcs_; }
#endif
bool ensureLoaded();
private:
std::mutex mtx_;
void *handle_ = nullptr;
};
extern IBVWrapper ibv;
/* Upstream seam spelling used by the ported core (ibv.funcs_.x). */
} // namespace hipObj
cuwrapper/rc/token.cpp
+246
View File
@@ -0,0 +1,246 @@
/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
*
* SPDX-License-Identifier: MIT
*/
/* RDMA token encoding/decoding implementation */
#include "token.h"
#include <cstdio>
#include <cstring>
#include <sstream>
namespace hipObj {
namespace {
const char HEX[] = "0123456789abcdef";
constexpr size_t kTokenBinaryLen = 1 + 4 + 16 + 4 + 8 + 8 + 1 + 2;
constexpr size_t kTokenHexLen = kTokenBinaryLen * 2;
int hexNibble(char c) {
if (c >= '0' && c <= '9')
return c - '0';
if (c >= 'a' && c <= 'f')
return c - 'a' + 10;
if (c >= 'A' && c <= 'F')
return c - 'A' + 10;
return -1;
}
bool decodeHexBytePair(char hi, char lo, uint8_t& out) {
int h = hexNibble(hi);
int l = hexNibble(lo);
if (h < 0 || l < 0)
return false;
out = static_cast<uint8_t>((h << 4) | l);
return true;
}
bool isSuccessHttpCode(int code) {
return code == 200 || code == 204 || code == 206;
}
} // namespace
std::string encodeRdmaToken(const RdmaToken& token) {
uint8_t buf[kTokenBinaryLen];
size_t off = 0;
buf[off++] = token.transport;
std::memcpy(buf + off, &token.qpNum, 4);
off += 4;
std::memcpy(buf + off, token.gid, 16);
off += 16;
std::memcpy(buf + off, &token.rkey, 4);
off += 4;
std::memcpy(buf + off, &token.remoteAddr, 8);
off += 8;
std::memcpy(buf + off, &token.length, 8);
off += 8;
buf[off++] = token.portNum;
std::memcpy(buf + off, &token.lid, 2);
off += 2;
std::ostringstream oss;
for (size_t i = 0; i < off; ++i) {
oss << HEX[(buf[i] >> 4) & 0xf] << HEX[buf[i] & 0xf];
}
return oss.str();
}
bool decodeRdmaTokenHex(const char* tokenHex, RdmaToken& out) {
if (!tokenHex)
return false;
size_t hexLen = std::strlen(tokenHex);
if (hexLen != kTokenHexLen)
return false;
uint8_t buf[kTokenBinaryLen];
for (size_t i = 0; i < kTokenBinaryLen; ++i) {
if (!decodeHexBytePair(tokenHex[i * 2], tokenHex[i * 2 + 1], buf[i]))
return false;
}
size_t off = 0;
out.transport = buf[off++];
std::memcpy(&out.qpNum, buf + off, 4);
off += 4;
std::memcpy(out.gid, buf + off, 16);
off += 16;
std::memcpy(&out.rkey, buf + off, 4);
off += 4;
std::memcpy(&out.remoteAddr, buf + off, 8);
off += 8;
std::memcpy(&out.length, buf + off, 8);
off += 8;
out.portNum = buf[off++];
std::memcpy(&out.lid, buf + off, 2);
return true;
}
bool parseRdmaReplyHttpCode(const char* reply, size_t replyLen, int& httpCode) {
if (!reply || replyLen == 0)
return false;
size_t len = replyLen;
while (len > 0 && (reply[len - 1] == '\0' || reply[len - 1] == '\n' ||
reply[len - 1] == '\r')) {
--len;
}
if (len == 0)
return false;
if (len >= 2 && reply[0] == 'o' && reply[1] == 'k') {
httpCode = 200;
return true;
}
if (len >= 3 && reply[0] == 'e' && reply[1] == 'r' && reply[2] == 'r') {
httpCode = -1;
return true;
}
char tmp[512];
if (len >= sizeof(tmp))
return false;
std::memcpy(tmp, reply, len);
tmp[len] = '\0';
char* colon = std::strchr(tmp, ':');
if (colon) {
*colon = '\0';
}
char* end = nullptr;
long code = std::strtol(tmp, &end, 10);
if (end == tmp || *end != '\0')
return false;
httpCode = static_cast<int>(code);
return true;
}
bool decodeRdmaReply(const char* reply, size_t replyLen, int& status) {
int httpCode = 0;
if (!parseRdmaReplyHttpCode(reply, replyLen, httpCode))
return false;
if (httpCode == 501) {
status = -2;
return true;
}
if (httpCode < 0) {
status = -1;
return true;
}
if (isSuccessHttpCode(httpCode)) {
status = 0;
return true;
}
status = -1;
return true;
}
bool parseClientNicFromTokenHex(const char* tokenHex, char* nicIp,
size_t nicIpLen) {
if (!nicIp || nicIpLen == 0)
return false;
nicIp[0] = '\0';
if (!tokenHex)
return false;
RdmaToken token;
if (!decodeRdmaTokenHex(tokenHex, token))
return false;
if (token.gid[10] != 0xff || token.gid[11] != 0xff)
return true;
int n = std::snprintf(nicIp, nicIpLen, "%u.%u.%u.%u",
static_cast<unsigned>(token.gid[12]),
static_cast<unsigned>(token.gid[13]),
static_cast<unsigned>(token.gid[14]),
static_cast<unsigned>(token.gid[15]));
if (n < 0 || static_cast<size_t>(n) >= nicIpLen)
return false;
return true;
}
std::string formatRdmaHeaderValue(const char* tokenHex, const void* buf,
size_t size) {
char header[512];
std::snprintf(header, sizeof(header), "%s:%016lx:%016lx", tokenHex,
reinterpret_cast<uintptr_t>(buf),
static_cast<unsigned long>(size));
return std::string(header);
}
bool parsePeerTokenFromReply(const char* reply, size_t replyLen,
RdmaToken& peerToken, int& httpCode) {
if (!reply || replyLen == 0)
return false;
size_t len = replyLen;
while (len > 0 && (reply[len - 1] == '\0' || reply[len - 1] == '\n' ||
reply[len - 1] == '\r')) {
--len;
}
if (len == 0)
return false;
const char* colon = static_cast<const char*>(std::memchr(reply, ':', len));
if (!colon || colon == reply) {
return false;
}
size_t codeLen = static_cast<size_t>(colon - reply);
char codeBuf[16];
if (codeLen >= sizeof(codeBuf))
return false;
std::memcpy(codeBuf, reply, codeLen);
codeBuf[codeLen] = '\0';
char* end = nullptr;
long code = std::strtol(codeBuf, &end, 10);
if (end == codeBuf || *end != '\0')
return false;
httpCode = static_cast<int>(code);
const char* tokenHex = colon + 1;
size_t tokenHexLen = len - codeLen - 1;
if (tokenHexLen != kTokenHexLen)
return false;
char tokenCopy[kTokenHexLen + 1];
std::memcpy(tokenCopy, tokenHex, tokenHexLen);
tokenCopy[tokenHexLen] = '\0';
return decodeRdmaTokenHex(tokenCopy, peerToken);
}
std::string encodeReplyWithPeerToken(int httpCode, const RdmaToken& peerToken) {
std::ostringstream oss;
oss << httpCode << ':' << encodeRdmaToken(peerToken);
return oss.str();
}
} // namespace hipObj
cuwrapper/rc/token.h
+58
View File
@@ -0,0 +1,58 @@
/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
*
* SPDX-License-Identifier: MIT
*/
/* RDMA token encoding/decoding */
#pragma once
#include <cstddef>
#include <cstdint>
#include <string>
/* Consumers that already include the real verbs.h (test clients
* linking libibverbs) get the ibv types from there; the shipped
* library uses the vendored core definitions. */
#if !defined(HIPOBJ_REAL_VERBS)
#include "ibv-core.h"
#endif
namespace hipObj {
enum TransportType : uint8_t {
TRANSPORT_DC = 0x00,
TRANSPORT_RC = 0x01,
};
struct RdmaToken {
uint32_t qpNum;
uint8_t gid[16];
uint32_t rkey;
uint64_t remoteAddr;
uint64_t length;
uint8_t transport;
uint8_t portNum;
uint16_t lid;
};
std::string encodeRdmaToken(const RdmaToken& token);
bool decodeRdmaTokenHex(const char* tokenHex, RdmaToken& out);
bool decodeRdmaReply(const char* reply, size_t replyLen, int& status);
bool parseRdmaReplyHttpCode(const char* reply, size_t replyLen, int& httpCode);
bool parseClientNicFromTokenHex(const char* tokenHex, char* nicIp,
size_t nicIpLen);
std::string formatRdmaHeaderValue(const char* tokenHex, const void* buf,
size_t size);
bool parsePeerTokenFromReply(const char* reply, size_t replyLen,
RdmaToken& peerToken, int& httpCode);
std::string encodeReplyWithPeerToken(int httpCode, const RdmaToken& peerToken);
} // namespace hipObj
cuwrapper/rc/transport.h
+37
View File
@@ -0,0 +1,37 @@
/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
*
* SPDX-License-Identifier: MIT
*/
#pragma once
#include <cstdint>
#include "ibv-core.h"
namespace hipObj {
struct RdmaToken;
struct RcConnection {
struct ibv_context* ctx = nullptr;
struct ibv_pd* pd = nullptr;
struct ibv_cq* cq = nullptr;
struct ibv_qp* qp = nullptr;
uint8_t portNum = 1;
int gidIndex = -1;
union ibv_gid localGid = {};
};
int openRdmaDevice(int nicIndex, RcConnection& conn);
int openRdmaDeviceByName(const char* devName, RcConnection& conn);
void closeRdmaDevice(RcConnection& conn);
int createRcQp(RcConnection& conn, int cqSize, int maxSendWr, int maxRecvWr);
int transitionQpToInit(RcConnection& conn);
int transitionQpToRtr(RcConnection& conn, uint32_t destQpNum, uint16_t destLid,
union ibv_gid destGid);
int transitionQpToRts(RcConnection& conn);
int connectRcPeer(RcConnection& conn, const RdmaToken& peerToken);
int pollCompletion(RcConnection& conn, int expectedOpcode, int timeoutMs);
} // namespace hipObj
cuwrapper/rc/v2-clock.cpp
+41
View File
@@ -0,0 +1,41 @@
/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
* Copyright (c) Gluesys Inc. and Jihyeon Gim. All rights reserved.
*
* SPDX-License-Identifier: MIT
*/
#include "v2-clock.h"
#include <chrono>
namespace hipObj {
namespace v2 {
namespace {
class SteadyClock : public ClockSource {
public:
uint64_t nowMs() override {
auto now = std::chrono::steady_clock::now().time_since_epoch();
return static_cast<uint64_t>(
std::chrono::duration_cast<std::chrono::milliseconds>(now).count());
}
};
SteadyClock g_defaultClock;
ClockSource* g_override = nullptr;
} // namespace
ClockSource& clockSource() {
return g_override ? *g_override : static_cast<ClockSource&>(g_defaultClock);
}
ClockSource* setClockSourceForTest(ClockSource* source) {
ClockSource* previous = g_override;
g_override = source;
return previous;
}
} // namespace v2
} // namespace hipObj
cuwrapper/rc/v2-clock.h
+34
View File
@@ -0,0 +1,34 @@
/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
* Copyright (c) Gluesys Inc. and Jihyeon Gim. All rights reserved.
*
* SPDX-License-Identifier: MIT
*/
/* Injectable monotonic clock for v2 lifetime policies (retired-ring
* expiry, session timeouts). Production uses a steady-clock
* implementation; unit tests install a fake to control time. */
#pragma once
#include <cstdint>
namespace hipObj {
namespace v2 {
class ClockSource {
public:
virtual ~ClockSource() = default;
virtual uint64_t nowMs() = 0;
};
/* Returns the active clock. Production default unless a test
* override is installed. */
ClockSource& clockSource();
/* Installs a test clock and returns the previously active source
* (nullptr when the production default was active). Passing nullptr
* restores the default. Unit tests only. */
ClockSource* setClockSourceForTest(ClockSource* source);
} // namespace v2
} // namespace hipObj
cuwrapper/rc/v2-random.cpp
+82
View File
@@ -0,0 +1,82 @@
/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
* Copyright (c) Gluesys Inc. and Jihyeon Gim. All rights reserved.
*
* SPDX-License-Identifier: MIT
*/
#include "v2-random.h"
#include <cerrno>
#include <cstddef>
#include <sys/random.h>
namespace hipObj {
namespace v2 {
namespace {
class GetrandomSource : public RandomSource {
public:
bool next32(uint32_t& out) override {
/* Loop over short reads and EINTR; fail closed on hard errors. */
size_t filled = 0;
unsigned char buf[sizeof(uint32_t)];
while (filled < sizeof(buf)) {
ssize_t n = ::getrandom(buf + filled, sizeof(buf) - filled, 0);
if (n < 0) {
if (errno == EINTR) {
continue;
}
return false;
}
if (n == 0) {
return false;
}
filled += static_cast<size_t>(n);
}
uint32_t value = 0;
for (size_t i = 0; i < sizeof(buf); ++i) {
value = (value << 8) | buf[i];
}
out = value;
return true;
}
};
GetrandomSource g_defaultSource;
RandomSource* g_override = nullptr;
} // namespace
RandomSource& randomSource() {
return g_override ? *g_override : static_cast<RandomSource&>(g_defaultSource);
}
RandomSource* setRandomSourceForTest(RandomSource* source) {
RandomSource* previous = g_override;
g_override = source;
return previous;
}
bool nextClientPsn(uint32_t& psn) {
for (int attempt = 0; attempt < 3; ++attempt) {
uint32_t value = 0;
if (!randomSource().next32(value)) {
return false;
}
value &= 0x00ffffff; /* PSNs are 24-bit on the wire */
if (value != 0) {
psn = value;
return true;
}
}
return false;
}
bool nextCookie(uint32_t& cookie) {
return randomSource().next32(cookie);
}
} // namespace v2
} // namespace hipObj
cuwrapper/rc/v2-random.h
+46
View File
@@ -0,0 +1,46 @@
/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
* Copyright (c) Gluesys Inc. and Jihyeon Gim. All rights reserved.
*
* SPDX-License-Identifier: MIT
*/
/* Injectable randomness for v2 policies (client PSNs, completion
* cookies). Production draws from getrandom(2); unit tests install
* a deterministic source. */
#pragma once
#include <cstdint>
namespace hipObj {
namespace v2 {
class RandomSource {
public:
virtual ~RandomSource() = default;
/* Draws 32 random bits. Returns false only when the underlying
* entropy source failed; callers treat that as an internal error
* rather than falling back to a constant. */
virtual bool next32(uint32_t& out) = 0;
};
/* Returns the active source (production default unless a test
* override is installed). */
RandomSource& randomSource();
/* Installs a test source and returns the previously active one
* (nullptr when the production default was active). Passing
* nullptr restores the default. Unit tests only. */
RandomSource* setRandomSourceForTest(RandomSource* source);
/* Draws a 24-bit non-zero PSN. Retries up to twice when the draw
* masks to zero; returns false when the source failed or every
* draw was zero. */
bool nextClientPsn(uint32_t& psn);
/* Draws a full 32-bit completion cookie (0 is a valid cookie). */
bool nextCookie(uint32_t& cookie);
} // namespace v2
} // namespace hipObj
cuwrapper/rc/v2-registry.cpp
+229
View File
@@ -0,0 +1,229 @@
/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
* Copyright (c) Gluesys Inc. and Jihyeon Gim. All rights reserved.
*
* SPDX-License-Identifier: MIT
*/
#include "v2-registry.h"
#include <utility>
namespace hipObj {
namespace v2 {
std::mutex& apiLock() {
static std::mutex lock;
return lock;
}
// ---- RetiredRing ---------------------------------------------------
size_t RetiredRing::collectExpired(uint64_t nowMs) {
size_t collected = 0;
for (auto& slot : slots_) {
if (slot.reservationId != 0 && slot.recorded && nowMs >= slot.expireAtMs) {
slot = Slot{};
--used_;
++collected;
}
}
return collected;
}
uint64_t RetiredRing::reserve() {
for (auto& slot : slots_) {
if (slot.reservationId == 0) {
slot.reservationId = nextReservationId_++;
slot.recorded = false;
slot.expireAtMs = 0;
slot.qpn = 0;
slot.psn = 0;
++used_;
++reserved_;
return slot.reservationId;
}
}
return 0;
}
void RetiredRing::unreserve(uint64_t reservationId) {
if (reservationId == 0) {
return;
}
for (auto& slot : slots_) {
if (slot.reservationId == reservationId && !slot.recorded) {
slot = Slot{};
--used_;
--reserved_;
return;
}
}
}
void RetiredRing::record(uint64_t reservationId, uint32_t qpn, uint32_t psn) {
/* Precondition: reservationId refers to a Reserved slot owned by
* the caller. All call sites hold the apiLock and the entry
* invariant, so a miss is a programming error; treat it as a
* no-op rather than corrupting the ring. */
for (auto& slot : slots_) {
if (slot.reservationId == reservationId && !slot.recorded) {
slot.recorded = true;
slot.expireAtMs = clockSource().nowMs() + kExpiryMs;
slot.qpn = qpn;
slot.psn = psn;
--reserved_;
return;
}
}
}
bool RetiredRing::contains(uint32_t qpn, uint32_t psn) const {
for (const auto& slot : slots_) {
if (slot.reservationId != 0 && slot.recorded && slot.qpn == qpn &&
slot.psn == psn) {
return true;
}
}
return false;
}
size_t RetiredRing::used() const {
return used_;
}
size_t RetiredRing::reservedCount() const {
return reserved_;
}
size_t RetiredRing::recordedCount() const {
return used_ - reserved_;
}
// ---- ConnectionRegistry ---------------------------------------------
bool ConnectionRegistry::reserveSlot() {
if (entries_.size() + pendingReserves_ >= kMaxConnections) {
return false;
}
++pendingReserves_;
return true;
}
void ConnectionRegistry::unreserveSlot() {
if (pendingReserves_ > 0) {
--pendingReserves_;
}
}
ConnId ConnectionRegistry::insert(ConnectionEntryV2&& entry) {
if (pendingReserves_ == 0 ||
entries_.size() + pendingReserves_ > kMaxConnections) {
return 0;
}
--pendingReserves_;
ConnId id = nextId_++;
entries_.emplace(id, std::move(entry));
return id;
}
bool ConnectionRegistry::claimDestroy(ConnId id) {
auto it = entries_.find(id);
if (it == entries_.end()) {
return false;
}
ConnectionEntryV2& entry = it->second;
/* Single claimant: a live entry claims once; a poisoned entry may
* be re-claimed for the destroy retry. */
if (entry.destroying || (entry.destroyClaimed && !entry.poisoned)) {
return false;
}
entry.destroying = true;
entry.destroyClaimed = true;
return true;
}
void ConnectionRegistry::commitDestroy(ConnId id, bool qpGone, bool cqGone) {
auto it = entries_.find(id);
if (it == entries_.end()) {
return;
}
ConnectionEntryV2& entry = it->second;
if (!entry.destroying) {
return;
}
if (qpGone) {
entry.conn.qp = nullptr;
entry.conn.qpNum = 0;
}
if (cqGone) {
entry.conn.cq = nullptr;
}
if (entry.conn.qp == nullptr && entry.conn.cq == nullptr) {
entry.poisoned = false;
entry.destroying = false;
/* entry stays until eraseDestroyed(); destroyClaimed remains
* true so no new claim can race the erase. */
} else {
entry.poisoned = true;
entry.destroying = false;
}
}
bool ConnectionRegistry::eraseDestroyed(ConnId id) {
auto it = entries_.find(id);
if (it == entries_.end()) {
return false;
}
ConnectionEntryV2& entry = it->second;
if (entry.conn.qp != nullptr || entry.conn.cq != nullptr) {
return false;
}
entries_.erase(it);
return true;
}
bool ConnectionRegistry::isPoisoned(ConnId id) const {
auto it = entries_.find(id);
return it != entries_.end() && it->second.poisoned;
}
size_t ConnectionRegistry::size() const {
return entries_.size();
}
size_t ConnectionRegistry::pendingReserves() const {
return pendingReserves_;
}
RetiredRing& ConnectionRegistry::retired() {
return retired_;
}
#ifdef HIPOBJ_UNIT_TESTS
ConnId ConnectionRegistry::insertRawForTest(ConnectionEntryV2&& entry) {
ConnId id = nextId_++;
entries_.emplace(id, std::move(entry));
return id;
}
#endif
namespace {
ConnectionRegistry g_registry;
ConnectionRegistry* g_registryOverride = nullptr;
} // namespace
ConnectionRegistry& registry() {
return g_registryOverride ? *g_registryOverride
: static_cast<ConnectionRegistry&>(g_registry);
}
ConnectionRegistry* setRegistryForTest(ConnectionRegistry* r) {
ConnectionRegistry* previous = g_registryOverride;
g_registryOverride = r;
return previous;
}
} // namespace v2
} // namespace hipObj
cuwrapper/rc/v2-registry.h
+202
View File
@@ -0,0 +1,202 @@
/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
* Copyright (c) Gluesys Inc. and Jihyeon Gim. All rights reserved.
*
* SPDX-License-Identifier: MIT
*/
/* Per-token connection registry for the v2 protocol.
*
* Ownership model: the shared DeviceHandle owns
* ctx/pd plus topology; each RcConnV2 owns exactly one qp/cq pair.
* The registry serializes every mutation behind the library-wide
* apiLock, so entries, capacity accounting, and the retired ring are
* consistent without additional locks.
*
* Entry lifecycle:
*
* Live ──claimDestroy──▶ Destroying ──commitDestroy(remains)──▶
* Poisoned ──claimDestroy(retry)──▶ Destroying ──commitDestroy(all
* gone)──▶ Destroyed ──eraseDestroyed──▶ (removed)
*
* abortClaim() cancels a claim before any destroy verb ran; it is a
* TU-private helper called from exactly one place in
* releaseConnection().
*
* Retired ring: guards against (qpn, psn) reuse by stale peers. A
* reservation (Reserved, no expiry) is bound to the entry that will
* need to record its QP; recording flips it to Recorded with a 60 s
* expiry. Invariant: entry.reservationId != 0 implies entry.qp !=
* nullptr (one-directional; Busy-deferred entries legally hold a
* live QP with no reservation).
*/
#pragma once
#include <cstddef>
#include <cstdint>
#include <map>
#include <mutex>
#include <vector>
#include "ibv-core.h"
#include "v2-clock.h"
namespace hipObj {
struct DeviceHandle;
struct RcConnV2 {
struct ibv_cq* cq = nullptr;
struct ibv_qp* qp = nullptr;
uint32_t qpNum = 0;
};
namespace v2 {
/* releaseConnection() result codes (mapped to hipObj errors at the
* API boundary). */
constexpr int kReleaseOk = 0;
constexpr int kReleaseBusy = 1;
constexpr int kReleaseLeftover = 2;
using ConnId = uint64_t;
/* Protocol phase tracked per entry (v2-state machine). */
enum class Phase : uint8_t;
struct ConnectionEntryV2 {
RcConnV2 conn;
DeviceHandle* device = nullptr;
uint64_t reservationId = 0; /* retired-ring slot for this QP */
uint32_t clientPsn = 0; /* client PSN (recorded on destroy) */
uint8_t phase = 0; /* v2::Phase, opaque here */
bool poisoned = false;
bool destroyClaimed = false;
bool destroying = false;
};
/* Retired (qpn, psn) reuse guard ring. */
class RetiredRing {
public:
static constexpr size_t kCapacity = 4096;
static constexpr uint64_t kExpiryMs = 60'000;
/* Collects expired Recorded slots; returns slots collected. */
size_t collectExpired(uint64_t nowMs);
/* Reserves a slot; 0 when the ring is full after collection. */
uint64_t reserve();
/* Releases a Reserved reservation (never recorded). */
void unreserve(uint64_t reservationId);
/* Reserved -> Recorded with (qpn, psn) and a fresh expiry.
* Precondition: reservationId is Reserved and owned by the caller
* (guaranteed by construction: all callers hold the apiLock and
* the single-entry ownership invariant). */
void record(uint64_t reservationId, uint32_t qpn, uint32_t psn);
bool contains(uint32_t qpn, uint32_t psn) const;
size_t used() const;
size_t reservedCount() const;
size_t recordedCount() const;
private:
struct Slot {
uint64_t reservationId = 0; /* 0 = free */
bool recorded = false;
uint64_t expireAtMs = 0;
uint32_t qpn = 0;
uint32_t psn = 0;
};
std::vector<Slot> slots_ = std::vector<Slot>(kCapacity);
uint64_t nextReservationId_ = 1;
size_t used_ = 0;
size_t reserved_ = 0;
};
/* Library-wide lock. All v2 operations (and v1 entry points touching
* shared state) run under this non-recursive mutex. Callbacks must
* not re-enter the library; the contract is documented on the public
* ops structures. */
std::mutex& apiLock();
class ConnectionRegistry {
public:
static constexpr size_t kMaxConnections = 64;
/* Capacity reservation; pairs with insert()/unreserveSlot(). */
bool reserveSlot();
void unreserveSlot();
/* Inserts an entry (consumes one pending reserve). 0 when the
* entry would exceed the capacity. */
ConnId insert(ConnectionEntryV2&& entry);
/* Lookup-only visitor; must not call registry mutations or public
* callbacks from fn. */
template <typename F>
bool withEntry(ConnId id, F&& fn) {
auto it = entries_.find(id);
if (it == entries_.end()) {
return false;
}
fn(it->second);
return true;
}
/* Live|Poisoned -> Destroying; single claimant. */
bool claimDestroy(ConnId id);
/* Destroying -> Poisoned|Destroyed, reflecting destroyed objects. */
void commitDestroy(ConnId id, bool qpGone, bool cqGone);
/* Destroyed -> removed; releases capacity and the MR ref the entry
* held. Returns false when the id is unknown or not Destroyed. */
bool eraseDestroyed(ConnId id);
bool isPoisoned(ConnId id) const;
size_t size() const;
size_t pendingReserves() const;
/* Snapshot iteration: fn receives every live id. Mutations from
* fn are forbidden (lookup-only contract). */
template <typename F>
void forEachId(F&& fn) const {
for (const auto& [id, entry] : entries_) {
fn(id);
}
}
/* Access to the shared retired ring (apiLock held by callers). */
RetiredRing& retired();
#ifdef HIPOBJ_UNIT_TESTS
/* Test-only direct insertion bypassing reserveSlot() and
* retiredReserve(); used to construct states the normal
* sequencing cannot reach (for example a live qp without a
* retired-ring reservation, to exercise the defensive busy
* path). */
ConnId insertRawForTest(ConnectionEntryV2&& entry);
#endif
private:
ConnectionEntryV2* find(ConnId id);
const ConnectionEntryV2* find(ConnId id) const;
std::map<ConnId, ConnectionEntryV2> entries_;
size_t pendingReserves_ = 0;
RetiredRing retired_;
ConnId nextId_ = 1;
};
/* Global v2 registry accessor (single instance). */
ConnectionRegistry& registry();
/* Test-only registry replacement; returns the previous one. */
ConnectionRegistry* setRegistryForTest(ConnectionRegistry* r);
} // namespace v2
} // namespace hipObj
cuwrapper/rc/v2-transport.cpp
+456
View File
@@ -0,0 +1,456 @@
/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
* Copyright (c) Gluesys Inc. and Jihyeon Gim. All rights reserved.
*
* SPDX-License-Identifier: MIT
*/
#include "v2-transport.h"
#include <atomic>
#include <cerrno>
#include <cstring>
#include <arpa/inet.h>
#include "rc_ibv_host.h"
#include "transport.h"
#include "vendor-ops.h"
namespace hipObj {
namespace v2 {
namespace {
constexpr int kCqSize = 16;
constexpr int kMaxSendWr = 16;
constexpr int kMaxRecvWr = 16;
/* TU-private cancel for a claim that never reached a destroy verb.
* Called from exactly one place in releaseConnection(), before the
* destroy_qp branch; after any destroy attempt the procedure must
* end in commitDestroy() instead. The entry lands back in Poisoned
* so a retry can claim it again. */
bool abortClaim(ConnId id) {
ConnectionRegistry& reg = registry();
bool ok = false;
reg.withEntry(id, [&](ConnectionEntryV2& entry) {
if (entry.destroying) {
entry.destroying = false;
entry.poisoned = true;
ok = true;
}
});
return ok;
}
/* Vendor QP attributes depend only on the device, not on the
* connection struct, so both v1 and v2 transitions share the same
* helpers below (mirrors transport.cpp's applyVendorQpAttrs). */
void applyVendorQpAttrs(struct ibv_context* ctx, struct ibv_qp_attr* attr) {
if (!ctx || !attr) {
return;
}
struct ibv_device_attr devAttr;
std::memset(&devAttr, 0, sizeof(devAttr));
if (ibv.query_device(ctx, &devAttr) != 0) {
return;
}
#ifdef HIPOBJ_BNXT
if (isBnxtDevice(devAttr.vendor_id)) {
configureBnxtQp(attr);
}
#endif
#ifdef HIPOBJ_IONIC
if (isIonicDevice(devAttr.vendor_id)) {
configureIonicQp(attr);
}
#endif
}
int modifyQpToInit(DeviceHandle* dh, struct ibv_qp* qp) {
struct ibv_qp_attr attr;
std::memset(&attr, 0, sizeof(attr));
attr.qp_state = IBV_QPS_INIT;
attr.pkey_index = 0;
attr.port_num = dh->portNum;
attr.qp_access_flags = IBV_ACCESS_REMOTE_READ | IBV_ACCESS_REMOTE_WRITE;
applyVendorQpAttrs(dh->ctx, &attr);
int mask = IBV_QP_STATE | IBV_QP_PKEY_INDEX | IBV_QP_PORT |
IBV_QP_ACCESS_FLAGS;
return ibv.modify_qp(qp, &attr, mask);
}
int modifyQpToRtr(struct ibv_context* ctx, struct ibv_qp* qp,
uint32_t destQpNum, uint16_t destLid, union ibv_gid destGid,
int gidIndex, uint32_t rqPsn, uint8_t portNum) {
struct ibv_qp_attr attr;
std::memset(&attr, 0, sizeof(attr));
attr.qp_state = IBV_QPS_RTR;
/* The path MTU must not exceed the port's active MTU: emulated
* devices commonly report 1024 while HCAs run 4096, and a
* larger value fails the transition with EINVAL. */
{
struct ibv_port_attr pa;
std::memset(&pa, 0, sizeof(pa));
if (ibv.query_port(ctx, portNum, &pa) == 0 &&
pa.active_mtu >= IBV_MTU_512) {
attr.path_mtu = static_cast<enum ibv_mtu>(pa.active_mtu);
} else {
attr.path_mtu = IBV_MTU_1024;
}
}
applyVendorQpAttrs(ctx, &attr);
attr.dest_qp_num = destQpNum;
attr.rq_psn = rqPsn;
attr.max_dest_rd_atomic = 1;
attr.min_rnr_timer = 12;
/* hipObject targets RoCEv2: the GRH with the peer GID is
* the routing path; the LID stays unused on RoCE links. */
attr.ah_attr.is_global = 1;
attr.ah_attr.dlid = 0;
attr.ah_attr.grh.dgid = destGid;
attr.ah_attr.grh.hop_limit = 64;
attr.ah_attr.grh.sgid_index = gidIndex;
attr.ah_attr.grh.traffic_class = 0;
attr.ah_attr.sl = 0;
attr.ah_attr.src_path_bits = 0;
/* Service type needs the port in the address handle; leaving
* it zero fails the transition on providers that validate it
* (the emulated NIC rejects port 0 with EINVAL). */
attr.ah_attr.port_num = portNum;
int mask = IBV_QP_STATE | IBV_QP_AV | IBV_QP_PATH_MTU | IBV_QP_DEST_QPN |
IBV_QP_RQ_PSN | IBV_QP_MAX_DEST_RD_ATOMIC | IBV_QP_MIN_RNR_TIMER;
return ibv.modify_qp(qp, &attr, mask);
}
int modifyQpToRts(struct ibv_context* ctx, struct ibv_qp* qp, uint32_t sqPsn) {
struct ibv_qp_attr attr;
std::memset(&attr, 0, sizeof(attr));
attr.qp_state = IBV_QPS_RTS;
applyVendorQpAttrs(ctx, &attr);
attr.timeout = 14;
attr.retry_cnt = 7;
attr.rnr_retry = 7;
attr.sq_psn = sqPsn;
attr.max_rd_atomic = 1;
int mask = IBV_QP_STATE | IBV_QP_TIMEOUT | IBV_QP_RETRY_CNT |
IBV_QP_RNR_RETRY | IBV_QP_SQ_PSN | IBV_QP_MAX_QP_RD_ATOMIC;
return ibv.modify_qp(qp, &attr, mask);
}
} // namespace
int createRcConnV2(DeviceHandle* dh, RcConnV2& conn, bool* rollbackFailed) {
if (rollbackFailed) {
*rollbackFailed = false;
}
bool sawCq = false;
conn.cq = ibv.create_cq(dh->ctx, kCqSize, nullptr, nullptr, 0);
if (!conn.cq) {
return -1;
}
sawCq = true;
struct ibv_qp_init_attr init;
std::memset(&init, 0, sizeof(init));
init.qp_type = IBV_QPT_RC;
init.send_cq = conn.cq;
init.recv_cq = conn.cq;
init.cap.max_send_wr = kMaxSendWr;
init.cap.max_recv_wr = kMaxRecvWr;
init.cap.max_send_sge = 1;
init.cap.max_recv_sge = 1;
conn.qp = ibv.create_qp(dh->pd, &init);
if (!conn.qp) {
bool cqOk = true;
if (ibv.destroy_cq(conn.cq) != 0) {
cqOk = false;
/* Keep conn.cq when the destroy failed: clearing it would
* lose the handle the caller needs for the reaper retry. */
} else {
conn.cq = nullptr;
}
if (!cqOk) {
if (rollbackFailed) {
*rollbackFailed = true;
}
return -1;
}
return -1;
}
conn.qpNum = conn.qp->qp_num;
/* Threaded callers share the device handle; the counter must
* be atomic to avoid a data race on concurrent PREPAREs. */
dh->connRefs.fetch_add(1, std::memory_order_relaxed);
return 0;
}
void destroyRcConnV2(RcConnV2& conn, bool* qpOk, bool* cqOk) {
if (qpOk) {
*qpOk = true;
}
if (cqOk) {
*cqOk = true;
}
/* A successful destroy clears the handle so callers storing
* the surviving pointers never re-destroy them. */
if (conn.qp) {
if (ibv.destroy_qp(conn.qp) == 0) {
conn.qp = nullptr;
} else if (qpOk) {
*qpOk = false;
}
}
if (conn.cq) {
if (ibv.destroy_cq(conn.cq) == 0) {
conn.cq = nullptr;
} else if (cqOk) {
*cqOk = false;
}
}
}
int createRcQpOnly(DeviceHandle* dh, struct ibv_cq* cq, RcConnV2& conn) {
if (!cq) {
return -1;
}
struct ibv_qp_init_attr init;
std::memset(&init, 0, sizeof(init));
init.qp_type = IBV_QPT_RC;
init.send_cq = cq;
init.recv_cq = cq;
init.cap.max_send_wr = kMaxSendWr;
init.cap.max_recv_wr = kMaxRecvWr;
init.cap.max_send_sge = 1;
init.cap.max_recv_sge = 1;
conn.qp = ibv.create_qp(dh->pd, &init);
if (!conn.qp) {
return -1;
}
conn.qpNum = conn.qp->qp_num;
return 0;
}
int transitionQpToInitV2(DeviceHandle* dh, RcConnV2& conn) {
return modifyQpToInit(dh, conn.qp);
}
int rearmQpToInitV2(DeviceHandle* dh, RcConnV2& conn) {
/* RESET is the only transition valid from every QP state. */
struct ibv_qp_attr attr;
std::memset(&attr, 0, sizeof(attr));
attr.qp_state = IBV_QPS_RESET;
if (ibv.modify_qp(conn.qp, &attr, IBV_QP_STATE) != 0) {
return -1;
}
return modifyQpToInit(dh, conn.qp);
}
int transitionQpToRtrV2(DeviceHandle* dh, RcConnV2& conn, uint32_t destQpNum,
uint16_t destLid, union ibv_gid destGid,
uint32_t rqPsn) {
return modifyQpToRtr(dh->ctx, conn.qp, destQpNum, destLid, destGid,
dh->gidIndex, rqPsn, dh->portNum);
}
int transitionQpToRtsV2(RcConnV2& conn, DeviceHandle* dh, uint32_t sqPsn) {
return modifyQpToRts(dh->ctx, conn.qp, sqPsn);
}
void releaseDevice(DeviceHandle* dh) {
if (!dh) {
return;
}
uint32_t cur = dh->connRefs.load(std::memory_order_relaxed);
while (cur > 0 &&
!dh->connRefs.compare_exchange_weak(cur, cur - 1,
std::memory_order_relaxed)) {
}
/* PD/context close happens only when the buffer map is empty too;
* the caller (hipObjShutdown stage) checks both. */
}
int postRecvImm(DeviceHandle* dh, RcConnV2& conn) {
(void)dh;
struct ibv_recv_wr wr;
struct ibv_recv_wr* bad = nullptr;
struct ibv_sge sge; /* zero-SGE: no scatter entry */
std::memset(&wr, 0, sizeof(wr));
std::memset(&sge, 0, sizeof(sge));
wr.wr_id = kRecvImm;
wr.sg_list = nullptr;
wr.num_sge = 0;
return ibv.post_recv(conn.qp, &wr, &bad);
}
int releaseConnection(ConnId id) {
ConnectionRegistry& reg = registry();
if (!reg.claimDestroy(id)) {
return 0; /* claimed elsewhere or already gone: idempotent */
}
bool hasQp = false;
uint64_t rid = 0;
uint32_t qpn = 0;
reg.withEntry(id, [&](ConnectionEntryV2& entry) {
hasQp = entry.conn.qp != nullptr;
rid = entry.reservationId;
qpn = entry.conn.qpNum;
entry.reservationId = 0; /* local ownership during destroy */
});
if (hasQp && rid == 0) {
/* Defensive: the normal creation sequence always inserts an
* entry with a live reservation, so a live qp without one is
* not expected. Keep the handling anyway so the teardown path
* stays complete: reserve a slot now, and report busy when the
* retired ring is exhausted so the caller can retry later. */
rid = reg.retired().reserve();
if (rid == 0) {
abortClaim(id); /* pre-destroy cancel; stays reclaimable */
return kReleaseBusy;
}
reg.withEntry(id, [&](ConnectionEntryV2& entry) {
entry.reservationId = rid;
});
}
bool qpOk = true;
bool cqOk = true;
uint32_t psn = 0;
reg.withEntry(id, [&](ConnectionEntryV2& entry) {
psn = entry.clientPsn; /* recorded into the retired ring below */
if (entry.conn.qp) {
qpOk = ibv.destroy_qp(entry.conn.qp) == 0;
if (qpOk) {
if (rid != 0) {
/* Record the destroyed pair immediately; the tuple tracks
* the QP lifetime, independent of the CQ result below. */
reg.retired().record(rid, qpn, psn);
}
entry.conn.qp = nullptr;
entry.conn.qpNum = 0;
} else {
entry.reservationId = rid; /* keep for the retry */
}
} else if (rid != 0) {
/* No live qp: the reservation serves no future destroy. */
reg.retired().unreserve(rid);
}
if (entry.conn.cq) {
cqOk = ibv.destroy_cq(entry.conn.cq) == 0;
if (cqOk) {
entry.conn.cq = nullptr;
}
}
});
reg.commitDestroy(id, qpOk, cqOk);
if (qpOk && cqOk) {
reg.eraseDestroyed(id);
return 0;
}
return kReleaseLeftover;
}
int discardAndRecreateQp(ConnId id) {
ConnectionRegistry& reg = registry();
/* Local slot B guards the tuple we are about to retire. */
uint64_t slotB = reg.retired().reserve();
if (slotB == 0) {
return kReleaseBusy;
}
struct Captured {
struct ibv_qp* qp = nullptr;
struct ibv_cq* cq = nullptr;
DeviceHandle* device = nullptr;
uint64_t rid = 0;
uint32_t qpn = 0;
uint32_t psn = 0;
} cap;
bool found = reg.withEntry(id, [&](ConnectionEntryV2& entry) {
cap.qp = entry.conn.qp;
cap.cq = entry.conn.cq;
cap.device = entry.device;
cap.rid = entry.reservationId;
cap.qpn = entry.conn.qpNum;
cap.psn = entry.clientPsn;
});
if (!found || !cap.qp || !cap.cq || cap.rid == 0) {
/* Nothing to discard or the entry is mid-teardown. */
reg.retired().unreserve(slotB);
return kReleaseLeftover;
}
/* Destroy the old qp; on failure keep the live qp and return. */
if (ibv.destroy_qp(cap.qp) != 0) {
reg.retired().unreserve(slotB);
return kReleaseLeftover;
}
reg.retired().record(slotB, cap.qpn, cap.psn);
/* The old reservation A is released back: the tuple it guarded is
* now recorded through slot B, tied to the destroyed qp. */
reg.retired().unreserve(cap.rid);
/* Move ownership of A's slot to the caller and clear the fields
* before the recreate attempt, so the entry never holds a live
* reservationId without a qp. */
reg.withEntry(id, [&](ConnectionEntryV2& entry) {
entry.reservationId = 0;
entry.conn.qp = nullptr;
entry.conn.qpNum = 0;
});
RcConnV2 fresh;
if (createRcQpOnly(cap.device, cap.cq, fresh) != 0) {
/* No qp: release through the normal teardown path (the cq is
* destroyed there) and report the failure. */
int rc = releaseConnection(id);
if (rc != 0) {
return rc;
}
return kReleaseLeftover;
}
reg.withEntry(id, [&](ConnectionEntryV2& entry) {
entry.conn.qp = fresh.qp;
entry.conn.qpNum = fresh.qpNum;
entry.reservationId = cap.rid;
});
return 0;
}
bool validateDataCompletion(const struct ibv_wc& wc,
const WcExpectation& expect, const char** reason) {
if (wc.status != IBV_WC_SUCCESS) {
*reason = "completion status is not success";
return false;
}
if (wc.wr_id != expect.wrId) {
*reason = "unexpected wr_id";
return false;
}
if (expect.kind == WcKind::kGet) {
if (wc.opcode != IBV_WC_RECV_RDMA_WITH_IMM) {
*reason = "expected RECV_RDMA_WITH_IMM";
return false;
}
} else {
if (wc.opcode != IBV_WC_RECV) {
*reason = "expected RECV";
return false;
}
if (!(wc.wc_flags & IBV_WC_WITH_IMM)) {
*reason = "immediate flag missing";
return false;
}
}
if (ntohl(wc.imm_data) != expect.cookie) {
*reason = "immediate cookie mismatch";
return false;
}
*reason = "";
return true;
}
} // namespace v2
} // namespace hipObj
cuwrapper/rc/v2-transport.h
+106
View File
@@ -0,0 +1,106 @@
/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
* Copyright (c) Gluesys Inc. and Jihyeon Gim. All rights reserved.
*
* SPDX-License-Identifier: MIT
*/
/* v2 transport: device/qp ownership split and release procedure.
*
* DeviceHandle is the shared half (context, protection domain,
* topology) with a connection reference count; RcConnV2 owns one
* qp/cq pair. v2 entry points route through these helpers so the
* registry, capacity accounting, and the retired ring stay
* consistent. Everything runs under v2::apiLock(). */
#pragma once
#include <atomic>
#include <cstdint>
#include "ibv-core.h"
#include "v2-registry.h"
namespace hipObj {
struct DeviceHandle {
struct ibv_context* ctx = nullptr;
struct ibv_pd* pd = nullptr;
uint8_t portNum = 1;
int gidIndex = -1;
union ibv_gid localGid = {};
std::atomic<uint32_t> connRefs{0};
};
namespace v2 {
/* Creates the qp/cq pair on dh. On success conn holds both objects,
* conn.qpNum is captured, and dh->connRefs was incremented. On
* failure every partially created object was rolled back (rollback
* failures are reported via *rollbackFailed so the caller can raise
* a poisoned tombstone). */
int createRcConnV2(DeviceHandle* dh, RcConnV2& conn,
bool* rollbackFailed = nullptr);
/* Destroys conn's qp and cq. Reflects per-object success through
* the out flags; the caller feeds them to commitDestroy(). */
void destroyRcConnV2(RcConnV2& conn, bool* qpOk, bool* cqOk);
/* QP state transitions taking the device topology from dh. The v1
* signatures (RcConnection&) remain unchanged. */
int transitionQpToInitV2(DeviceHandle* dh, RcConnV2& conn);
int transitionQpToRtrV2(DeviceHandle* dh, RcConnV2& conn, uint32_t destQpNum,
uint16_t destLid, union ibv_gid destGid,
uint32_t rqPsn);
int transitionQpToRtsV2(RcConnV2& conn, DeviceHandle* dh, uint32_t sqPsn);
/* Drives a QP from any state (including RTS/ERR after a failed
* transfer) back to INIT. The verbs state table has no direct
* RTS->INIT edge, so this goes through RESET, which is valid from
* every state. */
int rearmQpToInitV2(DeviceHandle* dh, RcConnV2& conn);
/* Creates only a qp on an existing cq (conflict-discard retry uses
* this so the original cq survives). */
int createRcQpOnly(DeviceHandle* dh, struct ibv_cq* cq, RcConnV2& conn);
/* Retires a live qp whose (qpn, psn) was already used by a peer:
* reserves a fresh ring slot, destroys the old qp, records the
* tuple, and recreates the qp on the existing cq. On any failure
* the entry keeps its live qp and reservation (callers surface the
* error); the busy return means the retired ring is exhausted. */
int discardAndRecreateQp(ConnId id);
/* Decrements dh->connRefs after a successful releaseRcConnV2 and
* closes ctx/pd when no MR and no connection remain. */
void releaseDevice(DeviceHandle* dh);
/* Full v2 release procedure for one registry entry: claim, reserve
* a retired slot for a live qp, destroy, commit, record, erase.
* Returns a hipObj error code (0 = success, busy = deferred,
* rdma = leftover/poison). */
int releaseConnection(ConnId id);
/* Posts the zero-SGE receive work request (wr_id = kRecvImm). */
constexpr uint64_t kRecvImm = 0x5245435632494d4dULL; /* "RECV2IMM" */
int postRecvImm(DeviceHandle* dh, RcConnV2& conn);
/* Completion validation for the data phase. A GET consumes one
* RECV_RDMA_WITH_IMM completion; a PUT consumes one RECV completion
* carrying the immediate (flags & IBV_WC_WITH_IMM). The immediate
* carries the session cookie in network order - the byte count
* travels in the FINAL response instead. */
enum class WcKind { kGet, kPut };
struct WcExpectation {
WcKind kind;
uint64_t wrId; /* expected wr_id (kRecvImm for receives) */
uint32_t cookie; /* expected immediate value (the cookie) */
};
/* Validates one work completion against the expectation; returns
* false and fills *reason on mismatch. */
bool validateDataCompletion(const struct ibv_wc& wc,
const WcExpectation& expect, const char** reason);
} // namespace v2
} // namespace hipObj
cuwrapper/rc/v2-wire.cpp
+321
View File
@@ -0,0 +1,321 @@
/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
* Copyright (c) Gluesys Inc. and Jihyeon Gim. All rights reserved.
*
* SPDX-License-Identifier: MIT
*/
/* Implementation of the hipobj-rc-v2 wire helpers (see v2-wire.h). */
#include "v2-wire.h"
#include <cctype>
#include <cstdio>
#include <cstring>
namespace hipObj {
namespace v2 {
namespace {
bool isHexDigit(char c) {
return (c >= '0' && c <= '9') || (c >= 'a' && c <= 'f') ||
(c >= 'A' && c <= 'F');
}
bool isBase64Char(char c) {
return (c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z') ||
(c >= '0' && c <= '9') || c == '+' || c == '/';
}
/* RFC 4648 base64 decoding table for canonical validation. */
int b64Val(char c) {
if (c >= 'A' && c <= 'Z')
return c - 'A';
if (c >= 'a' && c <= 'z')
return c - 'a' + 26;
if (c >= '0' && c <= '9')
return c - '0' + 52;
if (c == '+')
return 62;
if (c == '/')
return 63;
return -1;
}
/* Minimal strict decode/re-encode used to verify canonical text. Only
* the 12-char (8 byte) shape is accepted here. */
bool canonicalBase64_8Bytes(const std::string& text) {
if (text.size() != kChecksumB64Len)
return false;
if (text.back() != '=')
return false;
for (size_t i = 0; i + 1 < text.size(); ++i) {
if (!isBase64Char(text[i]))
return false;
}
uint8_t bytes[8] = {0};
int v[11];
for (size_t i = 0; i < 11; ++i) {
v[i] = b64Val(text[i]);
if (v[i] < 0)
return false;
}
bytes[0] = (uint8_t)((v[0] << 2) | (v[1] >> 4));
bytes[1] = (uint8_t)((v[1] << 4) | (v[2] >> 2));
bytes[2] = (uint8_t)((v[2] << 6) | v[3]);
bytes[3] = (uint8_t)((v[4] << 2) | (v[5] >> 4));
bytes[4] = (uint8_t)((v[5] << 4) | (v[6] >> 2));
bytes[5] = (uint8_t)((v[6] << 6) | v[7]);
bytes[6] = (uint8_t)((v[8] << 2) | (v[9] >> 4));
bytes[7] = (uint8_t)((v[9] << 4) | (v[10] >> 2));
/* Pad bits of the last char must be zero for canonical form. */
if (v[10] & 0x3)
return false;
/* Re-encode and compare. */
static const char* tbl =
"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
std::string re;
auto b6 = [&](int i) {
return bytes[i / 8 * 6 + (i % 8)] >> 2;
};
(void)b6;
uint32_t acc = 0;
int bits = 0;
for (int i = 0; i < 8; ++i) {
acc = (acc << 8) | bytes[i];
bits += 8;
while (bits >= 6) {
bits -= 6;
re += tbl[(acc >> bits) & 0x3f];
}
}
while (re.size() < 11)
re += tbl[acc & 0x3f];
re += '=';
return re == text;
}
std::string trimOws(const std::string& s) {
size_t b = 0, e = s.size();
while (b < e && (s[b] == ' ' || s[b] == '\t'))
++b;
while (e > b && (s[e - 1] == ' ' || s[e - 1] == '\t'))
--e;
return s.substr(b, e - b);
}
} // namespace
bool splitHeaderLine(const std::string& line, std::string& name,
std::string& value) {
size_t colon = line.find(':');
if (colon == std::string::npos || colon == 0)
return false;
name = trimOws(line.substr(0, colon));
value = trimOws(line.substr(colon + 1));
if (name.empty())
return false;
return true;
}
bool parsePrepareReply(int httpStatus, const std::string& headers,
PrepareReply& out) {
out = PrepareReply();
out.httpStatus = httpStatus;
size_t pos = 0;
while (pos < headers.size()) {
size_t eol = headers.find("\r\n", pos);
if (eol == std::string::npos)
eol = headers.size();
std::string line = headers.substr(pos, eol - pos);
pos = (eol == headers.size()) ? headers.size() : eol + 2;
if (line.empty())
continue;
std::string name, value;
if (!splitHeaderLine(line, name, value))
return false;
for (auto& c : name) {
c = (char)std::tolower((unsigned char)c);
}
if (name == "x-amz-rdma-protocol" && value == kProtocolValue) {
out.protocolEcho = true;
} else if (name == "x-amz-rdma-protocol-status" &&
value == kUnsupportedValue) {
out.unsupportedMarker = true;
} else if (name == "x-amz-rdma-reply") {
/* Reply keeps the legacy "200:<88hex>" shape; the peer token is
* the payload after the prefix. */
size_t c = value.find(':');
if (c == std::string::npos)
return false;
if (value.compare(0, c, "200") != 0)
return false;
out.serverToken = value.substr(c + 1);
if (out.serverToken.size() != kTokenHexLen)
return false;
} else if (name == "x-amz-rdma-session") {
out.session = value;
} else if (name == "x-amz-rdma-psn") {
if (!parsePsn(value, out.serverPsn))
return false;
}
}
if (out.httpStatus == 200) {
if (!out.protocolEcho)
return false;
if (!isValidSessionHex(out.session))
return false;
if (out.serverToken.size() != kTokenHexLen)
return false;
if (out.serverPsn == 0)
return false;
}
return true;
}
bool parseFinalReply(int httpStatus, const std::string& headers,
FinalReply& out) {
out = FinalReply();
out.httpStatus = httpStatus;
bool sawCookie = false;
size_t pos = 0;
while (pos < headers.size()) {
size_t eol = headers.find("\r\n", pos);
if (eol == std::string::npos)
eol = headers.size();
std::string line = headers.substr(pos, eol - pos);
pos = (eol == headers.size()) ? headers.size() : eol + 2;
if (line.empty())
continue;
std::string name, value;
if (!splitHeaderLine(line, name, value))
return false;
for (auto& c : name) {
c = (char)std::tolower((unsigned char)c);
}
if (name == "x-amz-rdma-protocol" && value == kProtocolValue) {
out.protocolEcho = true;
} else if (name == "x-amz-rdma-cookie") {
if (value.size() != kCookieHexLen)
return false;
uint32_t v = 0;
for (char c : value) {
int d = b64Val(c); /* reuse: hex via isHexDigit check below */
(void)d;
if (!isHexDigit(c))
return false;
int hv = (c <= '9') ? (c - '0')
: (std::tolower((unsigned char)c) - 'a' + 10);
v = (v << 4) | (uint32_t)hv;
}
out.cookieEcho = v;
sawCookie = true;
} else if (name == "x-amz-rdma-bytes-transferred") {
if (value.empty())
return false;
uint64_t b = 0;
for (char c : value) {
if (c < '0' || c > '9')
return false;
b = b * 10 + (uint64_t)(c - '0');
}
out.bytes = b;
} else if (name == "x-amz-rdma-etag") {
out.etag = value;
} else if (name == "x-amz-rdma-version-id") {
out.versionId = value;
} else if (name == "x-amz-rdma-checksum") {
if (!validateChecksumText(value, out.checksumB64))
return false;
}
}
if ((httpStatus == 200 || httpStatus == 204) && !out.protocolEcho) {
return false;
}
if ((httpStatus == 200 || httpStatus == 204) && !sawCookie) {
return false;
}
out.cookiePresent = sawCookie;
return true;
}
bool validateChecksumText(const std::string& headerValue, std::string& out) {
static const char kPrefix[] = "CRC64NVME ";
size_t plen = sizeof(kPrefix) - 1;
if (headerValue.compare(0, plen, kPrefix) != 0)
return false;
std::string text = trimOws(headerValue.substr(plen));
if (!canonicalBase64_8Bytes(text))
return false;
out = text;
return true;
}
bool isValidSessionHex(const std::string& s) {
if (s.size() != kSessionHexLen)
return false;
for (char c : s) {
if (!isHexDigit(c))
return false;
}
return true;
}
bool parsePsn(const std::string& s, uint32_t& psn) {
if (s.size() != kPsnHexLen)
return false;
uint32_t v = 0;
for (char c : s) {
if (!isHexDigit(c))
return false;
int hv = (c <= '9') ? (c - '0')
: (std::tolower((unsigned char)c) - 'a' + 10);
v = (v << 4) | (uint32_t)hv;
}
if (v == 0 || v > 0xffffff)
return false;
psn = v;
return true;
}
std::string formatCookie(uint32_t cookie) {
char buf[kCookieHexLen + 1];
std::snprintf(buf, sizeof(buf), "%08x", cookie);
return std::string(buf);
}
std::string formatPsn(uint32_t psn) {
char buf[kPsnHexLen + 1];
std::snprintf(buf, sizeof(buf), "%06x", psn);
return std::string(buf);
}
std::string buildTarget(const std::string& bucket, const std::string& key,
const std::string& canonicalQuery) {
static const char* hex = "0123456789ABCDEF";
std::string out = "/";
auto enc = [&](const std::string& s, bool keepSlash) {
for (unsigned char c : s) {
if ((c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z') ||
(c >= '0' && c <= '9') || c == '-' || c == '.' || c == '_' ||
c == '~' || (keepSlash && c == '/')) {
out += (char)c;
} else {
out += '%';
out += hex[c >> 4];
out += hex[c & 0xf];
}
}
};
enc(bucket, false);
out += '/';
enc(key, true);
if (!canonicalQuery.empty()) {
out += '?';
out += canonicalQuery;
}
return out;
}
} // namespace v2
} // namespace hipObj
cuwrapper/rc/v2-wire.h
+119
View File
@@ -0,0 +1,119 @@
/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
* Copyright (c) Gluesys Inc. and Jihyeon Gim. All rights reserved.
*
* SPDX-License-Identifier: MIT
*/
/* hipobj-rc-v2 wire protocol parsing and formatting.
*
* The v2 protocol runs on a dedicated control endpoint and exchanges
* three request types (prepare/ready/cancel), each answered by a single
* HTTP response. The READY response doubles as the FINAL outcome: the
* server performs the RDMA transfer while answering READY, so the
* exchange completes in exactly two round trips.
*
* All parsing functions are pure (no library state) so they can be unit
* tested without RDMA hardware.
*/
#pragma once
#include <cstddef>
#include <cstdint>
#include <string>
namespace hipObj {
namespace v2 {
/* Header and path names used by the control protocol. */
constexpr const char* kProtocolHeader = "X-Amz-Rdma-Protocol";
constexpr const char* kProtocolValue = "hipobj-rc-v2";
constexpr const char* kTokenHeader = "X-Amz-Rdma-Token";
constexpr const char* kSessionHeader = "X-Amz-Rdma-Session";
constexpr const char* kCookieHeader = "X-Amz-Rdma-Cookie";
constexpr const char* kPsnHeader = "X-Amz-Rdma-Psn";
constexpr const char* kOpHeader = "X-Amz-Rdma-Op";
constexpr const char* kSizeHeader = "X-Amz-Rdma-Size";
constexpr const char* kOffsetHeader = "X-Amz-Rdma-Offset";
constexpr const char* kTargetHeader = "X-Amz-Rdma-Target";
constexpr const char* kUnsupportedHeader = "X-Amz-Rdma-Protocol-Status";
constexpr const char* kUnsupportedValue = "unsupported";
constexpr const char* kReplyHeader = "X-Amz-Rdma-Reply";
constexpr const char* kBytesHeader = "X-Amz-Rdma-Bytes-Transferred";
constexpr const char* kEtagHeader = "X-Amz-Rdma-Etag";
constexpr const char* kVersionHeader = "X-Amz-Rdma-Version-Id";
constexpr const char* kChecksumHeader = "X-Amz-Rdma-Checksum";
constexpr const char* kControlPathPrefix = "/.hipobj-rc/";
/* Wire limits. */
constexpr size_t kSessionHexLen = 32; /* 128-bit session id */
constexpr size_t kCookieHexLen = 8; /* 32-bit cookie */
constexpr size_t kPsnHexLen = 6; /* 24-bit PSN, 000001..ffffff */
constexpr size_t kTokenHexLen = 88; /* 44-byte token payload */
constexpr size_t kChecksumB64Len = 12; /* 8-byte CRC64NVME base64 */
constexpr uint32_t kMaxTransferSize = 0x7fffffff; /* 2^31-1 */
struct PrepareReply {
int httpStatus = 0;
bool protocolEcho = false;
bool unsupportedMarker = false;
std::string serverToken; /* 88 hex chars */
std::string session; /* 32 hex chars */
uint32_t serverPsn = 0;
};
struct FinalReply {
int httpStatus = 0;
bool protocolEcho = false;
uint64_t bytes = 0;
uint32_t cookieEcho = 0;
bool cookiePresent = false; /* cookie echo header seen */
std::string etag;
std::string versionId;
std::string checksumB64; /* 12-char canonical base64 text */
};
/* Parses a single "Name: value" header line (without CRLF) into name and
* value with surrounding OWS trimmed. Returns false for lines without a
* colon. */
bool splitHeaderLine(const std::string& line, std::string& name,
std::string& value);
/* Parses the response to PREPARE. headers is the raw header block with
* lines separated by \r\n (final empty line optional); httpStatus is the
* status code from the status line. Returns false when the block is
* malformed in a way that makes the fields unusable. */
bool parsePrepareReply(int httpStatus, const std::string& headers,
PrepareReply& out);
/* Parses the READY/FINAL response. Same conventions as parsePrepareReply.
* The checksum field, when present, must be "CRC64NVME " followed by
* exactly 12 canonical base64 characters (11 data + one trailing '=');
* otherwise parsing fails. */
bool parseFinalReply(int httpStatus, const std::string& headers,
FinalReply& out);
/* Validates a canonical CRC64NVME checksum value: "CRC64NVME " prefix
* plus 12 base64 chars where the last is '=' and the text round-trips
* through strict decode/re-encode. Returns the 12-char text via out when
* valid. */
bool validateChecksumText(const std::string& headerValue, std::string& out);
/* Validates a session id (32 lowercase/uppercase hex chars). */
bool isValidSessionHex(const std::string& s);
/* Validates a PSN value string: 6 hex chars, value in 1..0xffffff. */
bool parsePsn(const std::string& s, uint32_t& psn);
/* Formats a cookie or PSN as 8/6 uppercase-zero-padded hex. */
std::string formatCookie(uint32_t cookie);
std::string formatPsn(uint32_t psn);
/* Builds the canonical rdma-target header value from bucket, key and an
* optional canonical query string (already encoded and sorted, may be
* empty). Percent-encodes the path per RFC 3986 unreserved rules. */
std::string buildTarget(const std::string& bucket, const std::string& key,
const std::string& canonicalQuery);
} // namespace v2
} // namespace hipObj
cuwrapper/rc/v2_data_phase.cpp
+228
View File
@@ -0,0 +1,228 @@
/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
* Copyright (c) Gluesys Inc. and Jihyeon Gim. All rights reserved.
*
* SPDX-License-Identifier: MIT
*/
#include "v2_data_phase.h"
#include <cstdlib>
#include <cstring>
#include <ctime>
#include <arpa/inet.h>
#include "rc_ibv_host.h"
namespace hipObj {
namespace v2 {
namespace {
constexpr int kAccess = IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE |
IBV_ACCESS_REMOTE_READ;
/* Completion markers posted with every work request. */
constexpr uint64_t kWrRecv = 0x5245435632494d4dULL; /* RECV2IMM */
constexpr uint64_t kWrWrite = 0x57524954454d4d47ULL; /* WRITEMM */
uint64_t clockNowMs() {
struct timespec ts;
clock_gettime(CLOCK_MONOTONIC, &ts);
return static_cast<uint64_t>(ts.tv_sec) * 1000 +
static_cast<uint64_t>(ts.tv_nsec) / 1000000;
}
} // namespace
bool stagePutBuffer(V2Session& s, size_t size, struct ibv_pd* pd) {
if (s.staging != nullptr) {
return s.stagingMr != nullptr || pd == nullptr;
}
void* buf = std::malloc(size ? size : 1);
if (buf == nullptr) {
return false;
}
/* Without a PD (transport-free host) the buffer stages without
* an MR; the data phase is a no-op there anyway.
*
* Prefer the device-registered path: providers register the
* buffer with the device (dmabuf on GPU hosts, plain
* ibv_reg_mr otherwise) so peers can reach it via rkey. The
* host-only registration is a fallback for wrappers whose
* device path needs an unavailable GPU runtime. */
struct ibv_mr* mr = nullptr;
if (pd != nullptr) {
mr = ibv.reg_mr(pd, buf, size, kAccess);
if (mr == nullptr) {
mr = ibv.reg_mr_host(pd, buf, size, kAccess);
}
if (mr == nullptr) {
std::free(buf);
return false;
}
}
s.staging = buf;
s.stagingMr = mr;
return true;
}
void releaseStaging(V2Session& s) {
/* The caller must have quiesced or destroyed the session QP
* first: a posted work request can still reference the MR
* until the QP is gone. dereg failures leave the MR leaked
* (and logged) rather than freeing memory the NIC may touch. */
if (s.stagingMr != nullptr) {
if (ibv.dereg_mr(s.stagingMr) != 0) {
fprintf(stderr, "v2: staging dereg failed; leaking buffer\n");
s.staging = nullptr; /* MR is dead to us either way */
}
s.stagingMr = nullptr;
}
if (s.staging != nullptr) {
std::free(s.staging);
s.staging = nullptr;
}
}
bool postRecvForImm(struct ibv_qp* qp, struct ibv_mr* mr, size_t len) {
struct ibv_sge sge;
std::memset(&sge, 0, sizeof(sge));
sge.addr = reinterpret_cast<uintptr_t>(mr->addr);
sge.length = static_cast<uint32_t>(len);
sge.lkey = mr->lkey;
struct ibv_recv_wr wr;
std::memset(&wr, 0, sizeof(wr));
wr.wr_id = kWrRecv;
wr.sg_list = &sge;
wr.num_sge = 1;
struct ibv_recv_wr* bad = nullptr;
return ibv.post_recv(qp, &wr, &bad) == 0;
}
bool postWriteWithImm(struct ibv_qp* qp, struct ibv_mr* src,
uint64_t remoteAddr, uint32_t rkey, size_t len,
uint32_t immData) {
/* GET delivery: server pushes the object into the client MR
* with the session cookie as the immediate. */
struct ibv_sge sge;
std::memset(&sge, 0, sizeof(sge));
sge.addr = reinterpret_cast<uintptr_t>(src->addr);
sge.length = static_cast<uint32_t>(len);
sge.lkey = src->lkey;
struct ibv_send_wr wr;
std::memset(&wr, 0, sizeof(wr));
wr.wr_id = kWrWrite;
wr.opcode = IBV_WR_RDMA_WRITE_WITH_IMM;
wr.send_flags = IBV_SEND_SIGNALED;
wr.imm_data = htonl(immData);
wr.wr.rdma.remote_addr = remoteAddr;
wr.wr.rdma.rkey = rkey;
wr.sg_list = &sge;
wr.num_sge = 1;
struct ibv_send_wr* bad = nullptr;
return ibv.post_send(qp, &wr, &bad) == 0;
}
/* Polls the CQ for one completion matching `expectWr`, bounded by
* an absolute deadline on the monotonic clock. */
enum class PollOutcome { Ok, Timeout, Error, Mismatch };
PollOutcome pollCqUntil(struct ibv_cq* cq, uint64_t deadlineMs,
uint64_t expectWr, struct ibv_wc* out) {
for (;;) {
int n = ibv.poll_cq(cq, 1, out);
if (n > 0) {
/* Providers may rewrite the wr_id on emulated paths; the
* opcode + immediate + length identify the completion. */
(void)expectWr;
return PollOutcome::Ok;
}
if (n < 0) {
return PollOutcome::Error;
}
if (clockNowMs() >= deadlineMs) {
return PollOutcome::Timeout;
}
struct timespec ts = {0, 2 * 1000 * 1000};
nanosleep(&ts, nullptr);
}
}
DataPhaseResult runDataPhase(V2Session& s, uint64_t deadlineMs,
DataPhaseStats& stats) {
const bool noTransport = s.qp == nullptr && s.cq == nullptr;
if (noTransport || s.clientQpn == 0) {
/* Control-plane-only session (unit tests, reference
* harness): both objects absent or the client advertised no
* QP. A half-wired session is not accepted here. */
stats.bytes = s.size;
stats.cookie = s.cookie;
return DataPhaseResult::Ok;
}
if (s.qp == nullptr || s.cq == nullptr || s.stagingMr == nullptr) {
return DataPhaseResult::WireFail;
}
struct ibv_wc wc;
PollOutcome po;
if (s.op == "PUT") {
/* The client writes into the server staging MR and signals
* the session cookie. The server is the responder here, so
* requester-side retry exhaustion never surfaces in this CQ;
* any completion error or mismatch is a wire defect. */
if (!postRecvForImm(s.qp, s.stagingMr, static_cast<size_t>(s.size))) {
return DataPhaseResult::WireFail;
}
po = pollCqUntil(s.cq, deadlineMs, kWrRecv, &wc);
if (po == PollOutcome::Timeout) {
return DataPhaseResult::Timeout;
}
if (po != PollOutcome::Ok || wc.status != IBV_WC_SUCCESS ||
wc.opcode != IBV_WC_RECV_RDMA_WITH_IMM ||
(wc.wc_flags & IBV_WC_WITH_IMM) == 0 ||
ntohl(wc.imm_data) != s.cookie || wc.byte_len != s.size) {
return DataPhaseResult::VerifyFail;
}
stats.bytes = wc.byte_len;
stats.cookie = s.cookie;
return DataPhaseResult::Ok;
}
/* GET: push the staged object to the client MR with the
* cookie as the immediate; the client's receive consumes it. */
if (s.clientMrAddr == 0 || s.clientMrRkey == 0) {
return DataPhaseResult::WireFail;
}
if (!postWriteWithImm(s.qp, s.stagingMr, s.clientMrAddr, s.clientMrRkey,
static_cast<size_t>(s.size), s.cookie)) {
return DataPhaseResult::WireFail;
}
po = pollCqUntil(s.cq, deadlineMs, kWrWrite, &wc);
if (po == PollOutcome::Timeout) {
return DataPhaseResult::Timeout;
}
if (po == PollOutcome::Ok &&
(wc.status == IBV_WC_RNR_RETRY_EXC_ERR ||
wc.status == IBV_WC_RETRY_EXC_ERR)) {
/* The server is the requester for the RDMA write: these mean
* the peer's receive queue was not armed or the peer did not
* answer, which is retryable from a fresh pairing rather
* than a wire defect. */
return DataPhaseResult::Busy;
}
if (po != PollOutcome::Ok || wc.status != IBV_WC_SUCCESS ||
wc.opcode != IBV_WC_RDMA_WRITE) {
return DataPhaseResult::VerifyFail;
}
stats.bytes = s.size;
stats.cookie = s.cookie;
return DataPhaseResult::Ok;
}
} // namespace v2
} // namespace hipObj
cuwrapper/rc/v2_data_phase.h
+72
View File
@@ -0,0 +1,72 @@
/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
* Copyright (c) Gluesys Inc. and Jihyeon Gim. All rights reserved.
*
* SPDX-License-Identifier: MIT
*/
/* Server-side data phase for the v2 protocol.
*
* A READY answer performs the RDMA transfer described by the
* session: a PUT receives the client's WRITE_WITH_IMM into a
* server-registered staging MR, validating the session cookie in
* the immediate; a GET pushes the staged object into the client
* MR with the cookie as the immediate. The wire parameters
* (client QP, PSN, MR endpoint) arrive in the PREPARE and READY
* headers and are staged on the session.
*
* Everything goes through the same ibverbs wrapper the client
* uses, so unit tests can swap the function table as usual. */
#pragma once
#include <cstdint>
#include <string>
#include "ibv-core.h"
#include "v2_session.h"
namespace hipObj {
namespace v2 {
/* Outcome of one data-phase execution. */
enum class DataPhaseResult {
Ok, /* transfer completed and validated */
Busy, /* peer busy - the client should retry (409) */
Timeout, /* completion did not arrive within T_exec */
VerifyFail, /* completion arrived but failed validation */
WireFail, /* posting the work request failed */
};
struct DataPhaseStats {
uint64_t bytes = 0;
uint32_t cookie = 0;
};
/* Runs the data phase for a session that is already in the
* Transferring state. `deadlineMs` bounds the completion poll on
* the monotonic clock. Returns the outcome and fills `stats` on
* success. */
DataPhaseResult runDataPhase(V2Session& s, uint64_t deadlineMs,
DataPhaseStats& stats);
/* Registers the staging buffer for PUT objects of the given size
* and records the MR on the session. Returns false when the
* registration fails. */
bool stagePutBuffer(V2Session& s, size_t size, struct ibv_pd* pd);
/* Posts one receive that consumes the client's WRITE_WITH_IMM
* carrying the session cookie. */
bool postRecvForImm(struct ibv_qp* qp, struct ibv_mr* mr, size_t len);
/* Posts one RDMA WRITE_WITH_IMM pushing `src` into the client MR
* with the session cookie as the immediate (GET delivery). */
bool postWriteWithImm(struct ibv_qp* qp, struct ibv_mr* src,
uint64_t remoteAddr, uint32_t rkey, size_t len,
uint32_t immData);
/* Releases the session's staging MR (if any). Safe to call on a
* session that never staged. */
void releaseStaging(V2Session& s);
} // namespace v2
} // namespace hipObj
cuwrapper/rc/v2_request.cpp
+273
View File
@@ -0,0 +1,273 @@
/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
* Copyright (c) Gluesys Inc. and Jihyeon Gim. All rights reserved.
*
* SPDX-License-Identifier: MIT
*/
#include "v2_request.h"
#include <cctype>
#include <cstdio>
#include <cstdlib>
namespace hipObj {
namespace v2 {
namespace {
bool isHexDigits(const std::string& s, size_t lo, size_t hi) {
if (s.size() < lo || s.size() > hi) {
return false;
}
for (char c : s) {
if (!std::isxdigit(static_cast<unsigned char>(c))) {
return false;
}
}
return true;
}
bool parseU32Hex(const std::string& s, uint32_t& out) {
if (s.empty() || s.size() > 8) {
return false;
}
uint32_t v = 0;
for (char c : s) {
int d;
if (std::isdigit(static_cast<unsigned char>(c))) {
d = c - '0';
} else {
char lc = static_cast<char>(std::tolower(static_cast<unsigned char>(c)));
d = lc - 'a' + 10;
}
v = (v << 4) | static_cast<uint32_t>(d);
}
out = v;
return true;
}
bool parseU64Dec(const std::string& s, uint64_t& out) {
if (s.empty() || s.size() > 20) {
return false;
}
uint64_t v = 0;
for (char c : s) {
if (!std::isdigit(static_cast<unsigned char>(c))) {
return false;
}
uint64_t d = static_cast<uint64_t>(c - '0');
if (v > (UINT64_MAX - d) / 10) {
return false; /* would overflow */
}
v = v * 10 + d;
}
out = v;
return true;
}
/* Extracts the exact Authorization header value from the raw block
* (case-insensitive name match, value preserved byte-for-byte
* except leading/trailing OWS). */
std::string rawAuthorization(const std::string& raw) {
const std::string needle = "authorization:";
std::string lower;
lower.reserve(raw.size());
for (char c : raw) {
lower.push_back(
static_cast<char>(std::tolower(static_cast<unsigned char>(c))));
}
size_t pos = 0;
while (pos < lower.size()) {
size_t lineEnd = lower.find("\r\n", pos);
if (lineEnd == std::string::npos) {
lineEnd = lower.size();
}
size_t lineLen = lineEnd - pos;
if (lower.compare(pos, lineLen < needle.size() ? lineLen : needle.size(),
needle) == 0 &&
lineLen >= needle.size()) {
std::string v = raw.substr(pos + needle.size(), lineLen - needle.size());
size_t b = v.find_first_not_of(" \t");
size_t e = v.find_last_not_of(" \t");
if (b == std::string::npos) {
return std::string();
}
return v.substr(b, e - b + 1);
}
pos = lineEnd + 2;
}
return std::string();
}
} // namespace
std::optional<PrepareRequest> parsePrepareRequest(
const std::map<std::string, std::string>& headers,
const std::string& rawHeaders) {
PrepareRequest out;
auto it = headers.find("x-amz-rdma-protocol");
if (it == headers.end()) {
return std::nullopt;
}
out.protocol = it->second;
it = headers.find("x-amz-rdma-token");
if (it == headers.end()) {
return std::nullopt;
}
/* 88-hex or 88-hex:addr:size */
const std::string& tok = it->second;
size_t colon1 = tok.find(':');
if (colon1 == std::string::npos) {
if (!isHexDigits(tok, 88, 88)) {
return std::nullopt;
}
} else {
size_t colon2 = tok.find(':', colon1 + 1);
if (colon2 == std::string::npos) {
return std::nullopt;
}
std::string base = tok.substr(0, colon1);
std::string addr = tok.substr(colon1 + 1, colon2 - colon1 - 1);
std::string size = tok.substr(colon2 + 1);
if (!isHexDigits(base, 88, 88) || addr.empty() || size.empty() ||
!isHexDigits(addr, 1, 16) || !isHexDigits(size, 1, 16)) {
return std::nullopt;
}
}
out.token = tok;
it = headers.find("x-amz-rdma-psn");
if (it == headers.end() || !parseU32Hex(it->second, out.clientPsn) ||
out.clientPsn == 0 || out.clientPsn > 0x00ffffff) {
return std::nullopt;
}
it = headers.find("x-amz-rdma-cookie");
if (it == headers.end() || !isHexDigits(it->second, 8, 8)) {
return std::nullopt;
}
uint32_t cookie = 0;
parseU32Hex(it->second, cookie);
out.cookie = cookie;
it = headers.find("x-amz-rdma-op");
if (it == headers.end() || (it->second != "GET" && it->second != "PUT")) {
return std::nullopt;
}
out.op = it->second;
it = headers.find("x-amz-rdma-target");
if (it == headers.end() || it->second.empty() || it->second.front() != '/') {
return std::nullopt;
}
out.target = it->second;
it = headers.find("x-amz-rdma-size");
if (it == headers.end() || !parseU64Dec(it->second, out.size) ||
out.size == 0) {
return std::nullopt;
}
it = headers.find("x-amz-rdma-offset");
if (it != headers.end()) {
if (!parseU64Dec(it->second, out.offset)) {
return std::nullopt;
}
out.hasOffset = true;
}
out.authorization = rawAuthorization(rawHeaders);
if (out.authorization.empty()) {
return std::nullopt;
}
return out;
}
std::optional<ReadyRequest> parseReadyRequest(
const std::map<std::string, std::string>& headers,
const std::string& rawHeaders) {
ReadyRequest out;
auto it = headers.find("x-amz-rdma-protocol");
if (it == headers.end()) {
return std::nullopt;
}
out.protocol = it->second;
it = headers.find("x-amz-rdma-session");
if (it == headers.end() || !isHexDigits(it->second, 32, 32)) {
return std::nullopt;
}
out.session = it->second;
it = headers.find("x-amz-rdma-cookie");
if (it == headers.end() || !isHexDigits(it->second, 8, 8)) {
return std::nullopt;
}
uint32_t cookie = 0;
parseU32Hex(it->second, cookie);
out.cookie = cookie;
/* Client MR endpoint for the data phase. Optional on a GET that
* the server stages itself, required for PUT delivery and the
* GET READ pull. Parsed as bare hex without a 0x prefix. */
it = headers.find("x-amz-rdma-mr-addr");
if (it != headers.end()) {
/* Strict hex, and present-but-empty fails too: a field that
* exists must carry a valid value. Bare strtoull would also
* accept prefixes, whitespace and trailing garbage, which
* would poison the remote address. */
if (!isHexDigits(it->second, 1, 16)) {
return std::nullopt;
}
out.mrAddr = std::strtoull(it->second.c_str(), nullptr, 16);
}
it = headers.find("x-amz-rdma-mr-rkey");
if (it != headers.end()) {
if (!isHexDigits(it->second, 1, 8)) {
return std::nullopt;
}
out.mrRkey = static_cast<uint32_t>(
std::strtoull(it->second.c_str(), nullptr, 16));
}
it = headers.find("x-amz-rdma-qpn");
if (it != headers.end()) {
if (!isHexDigits(it->second, 1, 8)) {
return std::nullopt;
}
out.qpn = static_cast<uint32_t>(
std::strtoull(it->second.c_str(), nullptr, 16));
}
out.authorization = rawAuthorization(rawHeaders);
if (out.authorization.empty()) {
return std::nullopt;
}
return out;
}
std::optional<CancelRequest> parseCancelRequest(
const std::map<std::string, std::string>& headers,
const std::string& rawHeaders) {
CancelRequest out;
auto it = headers.find("x-amz-rdma-protocol");
if (it == headers.end()) {
return std::nullopt;
}
out.protocol = it->second;
it = headers.find("x-amz-rdma-session");
if (it == headers.end() || !isHexDigits(it->second, 32, 32)) {
return std::nullopt;
}
out.session = it->second;
out.authorization = rawAuthorization(rawHeaders);
if (out.authorization.empty()) {
return std::nullopt;
}
return out;
}
} // namespace v2
} // namespace hipObj
cuwrapper/rc/v2_request.h
+75
View File
@@ -0,0 +1,75 @@
/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
* Copyright (c) Gluesys Inc. and Jihyeon Gim. All rights reserved.
*
* SPDX-License-Identifier: MIT
*/
/* Server-side request parsers for the v2 control protocol.
*
* The wire ABNF is shared with the client (see v2-wire.h); these
* entry points parse the three control requests the reference
* server accepts: PREPARE, READY, and CANCEL. Headers arrive as a
* lowercase-name map plus the raw header block so SigV4 credentials
* can be verified against the original bytes. */
#pragma once
#include <cstdint>
#include <map>
#include <optional>
#include <string>
#include "v2-wire.h"
namespace hipObj {
namespace v2 {
enum class ControlOp { kPrepare, kReady, kCancel };
struct PrepareRequest {
std::string protocol; /* echo header value */
std::string token; /* 88-hex peer token */
uint32_t clientPsn = 0; /* 1..0xffffff */
uint32_t cookie = 0;
std::string op; /* "GET" | "PUT" */
std::string target; /* canonical object target (path[?q]) */
uint64_t size = 0;
uint64_t offset = 0;
bool hasOffset = false;
std::string authorization; /* raw Authorization value */
};
struct ReadyRequest {
std::string protocol;
std::string session; /* 32-hex session id */
uint32_t cookie = 0;
uint64_t mrAddr = 0; /* client MR address (hex 0x...) */
uint32_t mrRkey = 0; /* client MR rkey (hex) */
uint32_t qpn = 0; /* client QP number (hex) */
std::string authorization;
};
struct CancelRequest {
std::string protocol;
std::string session;
std::string authorization;
};
/* Parses a PREPARE from the normalized header map. Returns nullopt
* on any malformed field (caller answers 400). The Authorization
* value is taken from rawHeaders when present so the exact signed
* bytes survive. */
std::optional<PrepareRequest> parsePrepareRequest(
const std::map<std::string, std::string>& headers,
const std::string& rawHeaders);
std::optional<ReadyRequest> parseReadyRequest(
const std::map<std::string, std::string>& headers,
const std::string& rawHeaders);
std::optional<CancelRequest> parseCancelRequest(
const std::map<std::string, std::string>& headers,
const std::string& rawHeaders);
} // namespace v2
} // namespace hipObj
cuwrapper/rc/v2_session.cpp
+248
View File
@@ -0,0 +1,248 @@
/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
* Copyright (c) Gluesys Inc. and Jihyeon Gim. All rights reserved.
*
* SPDX-License-Identifier: MIT
*/
#include "v2_session.h"
#include <utility>
namespace hipObj {
namespace v2 {
namespace {
void notifyAll(std::condition_variable& cv) {
cv.notify_all();
}
} // namespace
bool SessionTable::insert(V2Session&& session) {
std::lock_guard<std::mutex> guard(mtx_);
session.ioActive = 1; /* published together with the entry */
auto [it, ok] = entries_.emplace(session.id, std::move(session));
if (ok) {
notifyAll(cv_);
}
return ok;
}
bool SessionTable::beginPublishing(const std::string& id) {
std::lock_guard<std::mutex> guard(mtx_);
auto it = entries_.find(id);
if (it == entries_.end() || it->second.state != SessState::Prepared) {
return false;
}
it->second.state = SessState::Publishing;
/* Response transmission bound: 5s from confirmation. */
it->second.txDeadlineAt = clockSource().nowMs() + 5000;
notifyAll(cv_);
return true;
}
bool SessionTable::finishPublishing(const std::string& id, uint64_t tPrepMs) {
std::lock_guard<std::mutex> guard(mtx_);
auto it = entries_.find(id);
if (it == entries_.end() || it->second.state != SessState::Publishing) {
return false;
}
it->second.state = SessState::Prepared;
it->second.clientDeadlineAt = clockSource().nowMs() + tPrepMs;
/* The PREPARE response left: its bound must stop applying so a
* READY acquiring a reference afterwards can never be hit by a
* stale forced release. */
it->second.txDeadlineAt = 0;
notifyAll(cv_);
return true;
}
bool SessionTable::beginTransferring(const std::string& id, uint64_t tExecMs) {
std::lock_guard<std::mutex> guard(mtx_);
auto it = entries_.find(id);
if (it == entries_.end() || it->second.state != SessState::Prepared) {
return false;
}
/* Expiry check shares the lock: a deadline-passed session cannot
* be revived by a READY (single source of truth). */
if (clockSource().nowMs() > it->second.clientDeadlineAt) {
it->second.state = SessState::Reaping;
notifyAll(cv_);
return false;
}
it->second.state = SessState::Transferring;
it->second.clientDeadlineAt = clockSource().nowMs() + tExecMs;
/* The PREPARE response bound no longer applies; the FINAL
* response arms a fresh one at beginCompleting. Clearing here
* keeps the reaper's forced release from firing on a stale
* bound during the data phase. */
it->second.txDeadlineAt = 0;
notifyAll(cv_);
return true;
}
bool SessionTable::beginCompleting(const std::string& id) {
std::lock_guard<std::mutex> guard(mtx_);
auto it = entries_.find(id);
if (it != entries_.end() && it->second.state == SessState::Transferring) {
/* Fresh response bound for the FINAL transmission. */
it->second.txDeadlineAt = clockSource().nowMs() + 5000;
}
if (it == entries_.end() || it->second.state != SessState::Transferring) {
return false;
}
it->second.state = SessState::Completing;
/* Mark the reference origin atomically with the transition:
* from here the io reference guards live staging data and the
* reaper must not force-release it. Setting this outside the
* lock would leave a window where the reaper still treats the
* reference as a Publishing orphan. */
it->second.ioFromCompleting = true;
notifyAll(cv_);
return true;
}
bool SessionTable::toReaping(const std::string& id) {
std::lock_guard<std::mutex> guard(mtx_);
auto it = entries_.find(id);
if (it == entries_.end() || it->second.state == SessState::Reaping) {
return false;
}
it->second.state = SessState::Reaping;
notifyAll(cv_);
return true;
}
SessState SessionTable::awaitNotPublishing(const std::string& id,
uint64_t waitDeadlineMs) {
std::unique_lock<std::mutex> lock(mtx_);
const auto deadline = std::chrono::steady_clock::time_point(
std::chrono::milliseconds(waitDeadlineMs));
cv_.wait_until(lock, deadline, [&] {
auto it = entries_.find(id);
return it == entries_.end() || it->second.state != SessState::Publishing;
});
auto it = entries_.find(id);
return it == entries_.end() ? SessState::Reaping : it->second.state;
}
SessState SessionTable::stateOf(const std::string& id) {
std::lock_guard<std::mutex> guard(mtx_);
auto it = entries_.find(id);
return it == entries_.end() ? SessState::Reaping : it->second.state;
}
bool SessionTable::claimDestroy(const std::string& id) {
std::lock_guard<std::mutex> guard(mtx_);
auto it = entries_.find(id);
if (it == entries_.end()) {
return false;
}
V2Session& s = it->second;
/* Active handler work is never preempted: the worker's
* finalizer performs the transition and re-enters the gate. */
if (s.state != SessState::Reaping || s.ioActive > 0 || s.destroying ||
(s.destroyClaimed && !s.poisoned)) {
return false;
}
s.destroying = true;
s.destroyClaimed = true;
return true;
}
void SessionTable::commitDestroy(const std::string& id, bool qpOk, bool cqOk) {
std::lock_guard<std::mutex> guard(mtx_);
auto it = entries_.find(id);
if (it == entries_.end() || !it->second.destroying) {
return;
}
V2Session& s = it->second;
if (qpOk) {
s.qp = nullptr;
s.serverQpn = 0;
}
if (cqOk) {
s.cq = nullptr;
}
if (s.qp == nullptr && s.cq == nullptr) {
it->second.ioActive = 0;
entries_.erase(it);
notifyAll(cv_);
return;
}
s.poisoned = true;
s.destroying = false;
}
bool SessionTable::eraseSession(const std::string& id) {
std::lock_guard<std::mutex> guard(mtx_);
auto it = entries_.find(id);
if (it == entries_.end()) {
return false;
}
if (it->second.qp != nullptr || it->second.cq != nullptr) {
return false;
}
entries_.erase(it);
notifyAll(cv_);
return true;
}
size_t SessionTable::size() const {
std::lock_guard<std::mutex> guard(mtx_);
return entries_.size();
}
std::vector<std::string> SessionTable::ids() const {
std::lock_guard<std::mutex> guard(mtx_);
std::vector<std::string> out;
out.reserve(entries_.size());
for (const auto& [id, s] : entries_) {
out.push_back(id);
}
return out;
}
uint64_t SessionTable::ringReserve() {
std::lock_guard<std::mutex> guard(mtx_);
return ring_.reserve();
}
void SessionTable::ringUnreserve(uint64_t reservationId) {
std::lock_guard<std::mutex> guard(mtx_);
ring_.unreserve(reservationId);
}
void SessionTable::ringRecord(uint64_t reservationId, uint32_t qpn,
uint32_t psn) {
std::lock_guard<std::mutex> guard(mtx_);
ring_.record(reservationId, qpn, psn);
}
void SessionTable::ringCollectExpired(uint64_t nowMs) {
std::lock_guard<std::mutex> guard(mtx_);
ring_.collectExpired(nowMs);
}
bool SessionTable::acquireIo(const std::string& id) {
std::lock_guard<std::mutex> guard(mtx_);
auto it = entries_.find(id);
if (it == entries_.end()) {
return false;
}
++it->second.ioActive;
return true;
}
int SessionTable::releaseIo(const std::string& id) {
std::lock_guard<std::mutex> guard(mtx_);
auto it = entries_.find(id);
if (it == entries_.end() || it->second.ioActive <= 0) {
return -1;
}
return --it->second.ioActive;
}
} // namespace v2
} // namespace hipObj
cuwrapper/rc/v2_session.h
+190
View File
@@ -0,0 +1,190 @@
/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
* Copyright (c) Gluesys Inc. and Jihyeon Gim. All rights reserved.
*
* SPDX-License-Identifier: MIT
*/
/* Server-side v2 session table.
*
* One mutex guards the entry map; a single table-scoped condition
* variable covers state changes so waiters never hold references to
* erased entries (predicates re-look-up by session id).
*
* Lifecycle:
*
* Prepared --PREPARE 200 confirmed--> Publishing --send ok-->
* Prepared (deadline re-armed to T_prep)
* Prepared --READY--> Transferring --FINAL confirmed-->
* Completing --afterSend--> Reaping --> erased
* any --expiry (deadlineAt passed, observed under the lock)-->
* Reaping; release/cancel likewise.
*
* Preserve-errors (bad cookie/credentials, duplicate READY while
* Transferring or Completing) never change the state - the session
* stays for the next READY from the original requester.
*
* Teardown runs through claimDestroy/commitDestroy exactly like the
* client registry: single claimant, per-object success reflected
* into the pointers, poisoned entries stay for a retry.
*/
#pragma once
#include <condition_variable>
#include <cstdint>
#include <map>
#include <mutex>
#include <string>
#include <vector>
#include "ibv-core.h"
#include "v2-clock.h"
#include "v2-registry.h"
namespace hipObj {
struct DeviceHandle;
struct RcConnV2;
namespace v2 {
/* Session phases. Values are internal; preserve-errors keep the
* current phase untouched. */
enum class SessState : uint8_t {
Prepared = 0,
Publishing,
Transferring,
Completing,
Reaping,
};
struct V2Session {
std::string id; /* 32 hex */
SessState state = SessState::Prepared;
std::string op;
std::string target;
uint64_t size = 0;
uint64_t offset = 0;
std::string authorization; /* credential identity reference */
std::string accessKey; /* verified access key (identity) */
uint32_t cookie = 0;
uint32_t clientPsn = 0;
uint32_t serverPsn = 0;
uint32_t serverQpn = 0;
uint64_t reservationId = 0; /* retired-ring slot */
uint64_t clientDeadlineAt = 0; /* ms, absolute (clockSource) */
uint64_t txDeadlineAt = 0; /* response transmission bound */
int ioActive = 0;
/* Origin of the outstanding io reference: true while the
* worker holds it from READY entry to the final send. The
* reaper force-releases only references still awaiting a
* Publishing response; a Completing reference (live staging
* data) is released by the cooperative finalizer. */
bool ioFromCompleting = false;
/* This session holds one device connection reference (set
* when a QP was created for it, consumed exactly once when
* that QP destroys successfully). Guards the CQ-only retry
* and INIT-failure paths from double or missing releases. */
bool connRefHeld = false;
bool published = false;
bool destroyClaimed = false;
bool destroying = false;
bool poisoned = false;
/* Transport objects owned by the session. */
struct ibv_qp* qp = nullptr;
struct ibv_cq* cq = nullptr;
DeviceHandle* device = nullptr;
/* Client wire endpoints from the READY headers. */
uint64_t clientMrAddr = 0; /* client MR address (PUT dest / GET src) */
uint32_t clientMrRkey = 0; /* client MR rkey */
uint32_t clientQpn = 0; /* client QP to pair against */
/* Peer endpoint decoded from the 88-hex token: routes the RTR
* address handle to the real client GID instead of our own. */
union ibv_gid peerGid;
bool hasPeerGid = false;
/* PUT staging (host buffer + MR owned by the session). */
void* staging = nullptr;
struct ibv_mr* stagingMr = nullptr;
};
class SessionTable {
public:
/* Inserts a fresh session (ioActive = 1, caller owns the count).
* Returns false when the id already exists. */
bool insert(V2Session&& session);
/* Snapshot look-up; fn must not mutate the table. */
template <typename F>
bool withSession(const std::string& id, F&& fn) {
std::lock_guard<std::mutex> guard(mtx_);
auto it = entries_.find(id);
if (it == entries_.end()) {
return false;
}
fn(it->second);
return true;
}
/* Transitions Prepared -> Publishing, arming txDeadlineAt.
* Returns false unless currently Prepared. */
bool beginPublishing(const std::string& id);
/* After a successful PREPARE send: Publishing -> Prepared with
* the client deadline re-armed from now + tPrepMs. */
bool finishPublishing(const std::string& id, uint64_t tPrepMs);
/* Transferring transition from a READY (only from Prepared). */
bool beginTransferring(const std::string& id, uint64_t tExecMs);
/* FINAL response confirmed: Transferring -> Completing. */
bool beginCompleting(const std::string& id);
/* Expiry/terminal transition into Reaping (idempotent). */
bool toReaping(const std::string& id);
/* Wakes when the session leaves Publishing (waitDeadline is a
* value copied by the caller). Returns the state observed (or
* Reaping for an erased id). */
SessState awaitNotPublishing(const std::string& id, uint64_t waitDeadlineMs);
/* States helper for handlers. */
SessState stateOf(const std::string& id);
/* Destroy gate: single claimant per entry. */
bool claimDestroy(const std::string& id);
/* Reflects per-object destroy success; erases fully-destroyed
* entries (caller feeds the retired ring before calling). */
void commitDestroy(const std::string& id, bool qpOk, bool cqOk);
bool eraseSession(const std::string& id);
size_t size() const;
/* Snapshot ids for iteration (reaper, shutdown drain). */
std::vector<std::string> ids() const;
/* Retired-ring operations, all serialized under the table lock
* so concurrent workers and the reaper cannot interleave ring
* mutations with session state changes. */
uint64_t ringReserve();
void ringUnreserve(uint64_t reservationId);
void ringRecord(uint64_t reservationId, uint32_t qpn, uint32_t psn);
void ringCollectExpired(uint64_t nowMs);
/* Acquires (ioActive++) and releases the handler reference for a
* session. Release returns the post-decrement count. */
bool acquireIo(const std::string& id);
int releaseIo(const std::string& id);
private:
mutable std::mutex mtx_;
std::map<std::string, V2Session> entries_;
/* Table-scoped CV: safe against entry erase. */
std::condition_variable cv_;
/* (qpn, psn) reuse guard ring; guarded by mtx_. */
RetiredRing ring_;
};
} // namespace v2
} // namespace hipObj
cuwrapper/rc/vendor-ops.h
+51
View File
@@ -0,0 +1,51 @@
/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
*
* SPDX-License-Identifier: MIT
*/
/* RDMA vendor/provider identification */
#pragma once
#include <cstdint>
#include <string>
#include "ibv-core.h"
namespace hipObj {
enum class Provider : uint8_t {
BNXT = 0,
IONIC = 2,
UNKNOWN = 0xFF,
};
constexpr uint32_t VENDOR_ID_BROADCOM = 0x14E4;
constexpr uint32_t VENDOR_ID_PENSANDO = 0x1DD8;
inline const char* provider_name(Provider p) {
switch (p) {
case Provider::BNXT:
return "bnxt";
case Provider::IONIC:
return "ionic";
default:
return "unknown";
}
}
inline Provider provider_from_string(const std::string& s) {
if (s == "bnxt" || s == "bnxt_re")
return Provider::BNXT;
if (s == "ionic" || s == "pensando")
return Provider::IONIC;
return Provider::UNKNOWN;
}
bool isBnxtDevice(uint32_t vendorId);
bool isIonicDevice(uint32_t vendorId);
int configureBnxtQp(struct ibv_qp_attr* attr);
int configureIonicQp(struct ibv_qp_attr* attr);
} // namespace hipObj