From efa0309da422455dcf0aab1c33e2a2ad69bb86ef Mon Sep 17 00:00:00 2001 From: Jihyeon Gim Date: Sun, 30 Aug 2026 00:00:29 +0900 Subject: [PATCH] rdma: port the hipObject v2 RC session core 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 --- Makefile | 14 + cuwrapper/rc/ibv-core.h | 725 +++++++++++++++++++++++++++++++++ cuwrapper/rc/rc_ibv_host.cpp | 105 +++++ cuwrapper/rc/rc_ibv_host.h | 112 +++++ cuwrapper/rc/token.cpp | 246 +++++++++++ cuwrapper/rc/token.h | 58 +++ cuwrapper/rc/transport.h | 37 ++ cuwrapper/rc/v2-clock.cpp | 41 ++ cuwrapper/rc/v2-clock.h | 34 ++ cuwrapper/rc/v2-random.cpp | 82 ++++ cuwrapper/rc/v2-random.h | 46 +++ cuwrapper/rc/v2-registry.cpp | 229 +++++++++++ cuwrapper/rc/v2-registry.h | 202 +++++++++ cuwrapper/rc/v2-transport.cpp | 456 +++++++++++++++++++++ cuwrapper/rc/v2-transport.h | 106 +++++ cuwrapper/rc/v2-wire.cpp | 321 +++++++++++++++ cuwrapper/rc/v2-wire.h | 119 ++++++ cuwrapper/rc/v2_data_phase.cpp | 228 +++++++++++ cuwrapper/rc/v2_data_phase.h | 72 ++++ cuwrapper/rc/v2_request.cpp | 273 +++++++++++++ cuwrapper/rc/v2_request.h | 75 ++++ cuwrapper/rc/v2_session.cpp | 248 +++++++++++ cuwrapper/rc/v2_session.h | 190 +++++++++ cuwrapper/rc/vendor-ops.h | 51 +++ 24 files changed, 4070 insertions(+) create mode 100644 cuwrapper/rc/ibv-core.h create mode 100644 cuwrapper/rc/rc_ibv_host.cpp create mode 100644 cuwrapper/rc/rc_ibv_host.h create mode 100644 cuwrapper/rc/token.cpp create mode 100644 cuwrapper/rc/token.h create mode 100644 cuwrapper/rc/transport.h create mode 100644 cuwrapper/rc/v2-clock.cpp create mode 100644 cuwrapper/rc/v2-clock.h create mode 100644 cuwrapper/rc/v2-random.cpp create mode 100644 cuwrapper/rc/v2-random.h create mode 100644 cuwrapper/rc/v2-registry.cpp create mode 100644 cuwrapper/rc/v2-registry.h create mode 100644 cuwrapper/rc/v2-transport.cpp create mode 100644 cuwrapper/rc/v2-transport.h create mode 100644 cuwrapper/rc/v2-wire.cpp create mode 100644 cuwrapper/rc/v2-wire.h create mode 100644 cuwrapper/rc/v2_data_phase.cpp create mode 100644 cuwrapper/rc/v2_data_phase.h create mode 100644 cuwrapper/rc/v2_request.cpp create mode 100644 cuwrapper/rc/v2_request.h create mode 100644 cuwrapper/rc/v2_session.cpp create mode 100644 cuwrapper/rc/v2_session.h create mode 100644 cuwrapper/rc/vendor-ops.h diff --git a/Makefile b/Makefile index 92e9740c..78c6cebe 100644 --- a/Makefile +++ b/Makefile @@ -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 diff --git a/cuwrapper/rc/ibv-core.h b/cuwrapper/rc/ibv-core.h new file mode 100644 index 00000000..57e81aaf --- /dev/null +++ b/cuwrapper/rc/ibv-core.h @@ -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 +#include +#include +#include + +#include +#include + +/* ------------------------------------------------------------------------- + * 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; +}; diff --git a/cuwrapper/rc/rc_ibv_host.cpp b/cuwrapper/rc/rc_ibv_host.cpp new file mode 100644 index 00000000..3ebcdadd --- /dev/null +++ b/cuwrapper/rc/rc_ibv_host.cpp @@ -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 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(load("ibv_get_device_list")); + funcs_.free_device_list = + reinterpret_cast(load("ibv_free_device_list")); + funcs_.open_device = reinterpret_cast( + load("ibv_open_device")); + funcs_.close_device = reinterpret_cast( + load("ibv_close_device")); + funcs_.alloc_pd = reinterpret_cast( + load("ibv_alloc_pd")); + funcs_.dealloc_pd = reinterpret_cast( + load("ibv_dealloc_pd")); + funcs_.reg_mr = reinterpret_cast(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( + load("ibv_dereg_mr")); + funcs_.create_cq = reinterpret_cast( + load("ibv_create_cq")); + funcs_.destroy_cq = reinterpret_cast( + load("ibv_destroy_cq")); + funcs_.create_qp = reinterpret_cast(load("ibv_create_qp")); + funcs_.destroy_qp = reinterpret_cast( + load("ibv_destroy_qp")); + funcs_.modify_qp = reinterpret_cast( + load("ibv_modify_qp")); + funcs_.poll_cq = reinterpret_cast(load("ibv_poll_cq")); + funcs_.query_device = reinterpret_cast(load("ibv_query_device")); + funcs_.query_port = reinterpret_cast( + load("ibv_query_port")); + funcs_.query_gid = reinterpret_cast( + load("ibv_query_gid")); + funcs_.post_recv = reinterpret_cast(load("ibv_post_recv")); + funcs_.post_send = reinterpret_cast(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 diff --git a/cuwrapper/rc/rc_ibv_host.h b/cuwrapper/rc/rc_ibv_host.h new file mode 100644 index 00000000..0dcbc6ba --- /dev/null +++ b/cuwrapper/rc/rc_ibv_host.h @@ -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 + +#include +#include +#include +#include + +#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 diff --git a/cuwrapper/rc/token.cpp b/cuwrapper/rc/token.cpp new file mode 100644 index 00000000..eb7a41b3 --- /dev/null +++ b/cuwrapper/rc/token.cpp @@ -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 +#include +#include + +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((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(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(token.gid[12]), + static_cast(token.gid[13]), + static_cast(token.gid[14]), + static_cast(token.gid[15])); + if (n < 0 || static_cast(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(buf), + static_cast(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(std::memchr(reply, ':', len)); + if (!colon || colon == reply) { + return false; + } + + size_t codeLen = static_cast(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(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 diff --git a/cuwrapper/rc/token.h b/cuwrapper/rc/token.h new file mode 100644 index 00000000..7b24808e --- /dev/null +++ b/cuwrapper/rc/token.h @@ -0,0 +1,58 @@ +/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved. + * + * SPDX-License-Identifier: MIT + */ + +/* RDMA token encoding/decoding */ + +#pragma once + +#include +#include +#include + +/* 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 diff --git a/cuwrapper/rc/transport.h b/cuwrapper/rc/transport.h new file mode 100644 index 00000000..a4735448 --- /dev/null +++ b/cuwrapper/rc/transport.h @@ -0,0 +1,37 @@ +/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved. + * + * SPDX-License-Identifier: MIT + */ + +#pragma once + +#include + +#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 diff --git a/cuwrapper/rc/v2-clock.cpp b/cuwrapper/rc/v2-clock.cpp new file mode 100644 index 00000000..e012b2e5 --- /dev/null +++ b/cuwrapper/rc/v2-clock.cpp @@ -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 + +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( + std::chrono::duration_cast(now).count()); + } +}; + +SteadyClock g_defaultClock; +ClockSource* g_override = nullptr; + +} // namespace + +ClockSource& clockSource() { + return g_override ? *g_override : static_cast(g_defaultClock); +} + +ClockSource* setClockSourceForTest(ClockSource* source) { + ClockSource* previous = g_override; + g_override = source; + return previous; +} + +} // namespace v2 +} // namespace hipObj diff --git a/cuwrapper/rc/v2-clock.h b/cuwrapper/rc/v2-clock.h new file mode 100644 index 00000000..442ff7ca --- /dev/null +++ b/cuwrapper/rc/v2-clock.h @@ -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 + +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 diff --git a/cuwrapper/rc/v2-random.cpp b/cuwrapper/rc/v2-random.cpp new file mode 100644 index 00000000..8470231c --- /dev/null +++ b/cuwrapper/rc/v2-random.cpp @@ -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 +#include + +#include + +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(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(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 diff --git a/cuwrapper/rc/v2-random.h b/cuwrapper/rc/v2-random.h new file mode 100644 index 00000000..3c537f86 --- /dev/null +++ b/cuwrapper/rc/v2-random.h @@ -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 + +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 diff --git a/cuwrapper/rc/v2-registry.cpp b/cuwrapper/rc/v2-registry.cpp new file mode 100644 index 00000000..4562dbf8 --- /dev/null +++ b/cuwrapper/rc/v2-registry.cpp @@ -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 + +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(g_registry); +} + +ConnectionRegistry* setRegistryForTest(ConnectionRegistry* r) { + ConnectionRegistry* previous = g_registryOverride; + g_registryOverride = r; + return previous; +} + +} // namespace v2 +} // namespace hipObj diff --git a/cuwrapper/rc/v2-registry.h b/cuwrapper/rc/v2-registry.h new file mode 100644 index 00000000..6f84a98d --- /dev/null +++ b/cuwrapper/rc/v2-registry.h @@ -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 +#include +#include +#include +#include + +#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 slots_ = std::vector(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 + 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 + 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 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 diff --git a/cuwrapper/rc/v2-transport.cpp b/cuwrapper/rc/v2-transport.cpp new file mode 100644 index 00000000..002e1010 --- /dev/null +++ b/cuwrapper/rc/v2-transport.cpp @@ -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 +#include +#include + +#include + +#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(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 diff --git a/cuwrapper/rc/v2-transport.h b/cuwrapper/rc/v2-transport.h new file mode 100644 index 00000000..a52ccbdc --- /dev/null +++ b/cuwrapper/rc/v2-transport.h @@ -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 +#include + +#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 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 diff --git a/cuwrapper/rc/v2-wire.cpp b/cuwrapper/rc/v2-wire.cpp new file mode 100644 index 00000000..ab0acb54 --- /dev/null +++ b/cuwrapper/rc/v2-wire.cpp @@ -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 +#include +#include + +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 diff --git a/cuwrapper/rc/v2-wire.h b/cuwrapper/rc/v2-wire.h new file mode 100644 index 00000000..3838f743 --- /dev/null +++ b/cuwrapper/rc/v2-wire.h @@ -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 +#include +#include + +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 diff --git a/cuwrapper/rc/v2_data_phase.cpp b/cuwrapper/rc/v2_data_phase.cpp new file mode 100644 index 00000000..2a199762 --- /dev/null +++ b/cuwrapper/rc/v2_data_phase.cpp @@ -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 +#include +#include + +#include + +#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(ts.tv_sec) * 1000 + + static_cast(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(mr->addr); + sge.length = static_cast(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(src->addr); + sge.length = static_cast(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(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(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 diff --git a/cuwrapper/rc/v2_data_phase.h b/cuwrapper/rc/v2_data_phase.h new file mode 100644 index 00000000..2c2cee7a --- /dev/null +++ b/cuwrapper/rc/v2_data_phase.h @@ -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 +#include + +#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 diff --git a/cuwrapper/rc/v2_request.cpp b/cuwrapper/rc/v2_request.cpp new file mode 100644 index 00000000..a742573f --- /dev/null +++ b/cuwrapper/rc/v2_request.cpp @@ -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 +#include +#include + +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(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(c))) { + d = c - '0'; + } else { + char lc = static_cast(std::tolower(static_cast(c))); + d = lc - 'a' + 10; + } + v = (v << 4) | static_cast(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(c))) { + return false; + } + uint64_t d = static_cast(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(std::tolower(static_cast(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 parsePrepareRequest( + const std::map& 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 parseReadyRequest( + const std::map& 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( + 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( + std::strtoull(it->second.c_str(), nullptr, 16)); + } + + out.authorization = rawAuthorization(rawHeaders); + if (out.authorization.empty()) { + return std::nullopt; + } + return out; +} + +std::optional parseCancelRequest( + const std::map& 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 diff --git a/cuwrapper/rc/v2_request.h b/cuwrapper/rc/v2_request.h new file mode 100644 index 00000000..0c7c2c34 --- /dev/null +++ b/cuwrapper/rc/v2_request.h @@ -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 +#include +#include +#include + +#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 parsePrepareRequest( + const std::map& headers, + const std::string& rawHeaders); + +std::optional parseReadyRequest( + const std::map& headers, + const std::string& rawHeaders); + +std::optional parseCancelRequest( + const std::map& headers, + const std::string& rawHeaders); + +} // namespace v2 +} // namespace hipObj diff --git a/cuwrapper/rc/v2_session.cpp b/cuwrapper/rc/v2_session.cpp new file mode 100644 index 00000000..48994fd2 --- /dev/null +++ b/cuwrapper/rc/v2_session.cpp @@ -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 + +namespace hipObj { +namespace v2 { + +namespace { + +void notifyAll(std::condition_variable& cv) { + cv.notify_all(); +} + +} // namespace + +bool SessionTable::insert(V2Session&& session) { + std::lock_guard 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 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 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 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 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 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 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 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 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 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 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 guard(mtx_); + return entries_.size(); +} + +std::vector SessionTable::ids() const { + std::lock_guard guard(mtx_); + std::vector out; + out.reserve(entries_.size()); + for (const auto& [id, s] : entries_) { + out.push_back(id); + } + return out; +} + +uint64_t SessionTable::ringReserve() { + std::lock_guard guard(mtx_); + return ring_.reserve(); +} + +void SessionTable::ringUnreserve(uint64_t reservationId) { + std::lock_guard guard(mtx_); + ring_.unreserve(reservationId); +} + +void SessionTable::ringRecord(uint64_t reservationId, uint32_t qpn, + uint32_t psn) { + std::lock_guard guard(mtx_); + ring_.record(reservationId, qpn, psn); +} + +void SessionTable::ringCollectExpired(uint64_t nowMs) { + std::lock_guard guard(mtx_); + ring_.collectExpired(nowMs); +} + +bool SessionTable::acquireIo(const std::string& id) { + std::lock_guard 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 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 diff --git a/cuwrapper/rc/v2_session.h b/cuwrapper/rc/v2_session.h new file mode 100644 index 00000000..97cfcc73 --- /dev/null +++ b/cuwrapper/rc/v2_session.h @@ -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 +#include +#include +#include +#include +#include + +#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 + bool withSession(const std::string& id, F&& fn) { + std::lock_guard 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 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 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 diff --git a/cuwrapper/rc/vendor-ops.h b/cuwrapper/rc/vendor-ops.h new file mode 100644 index 00000000..88d00540 --- /dev/null +++ b/cuwrapper/rc/vendor-ops.h @@ -0,0 +1,51 @@ +/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved. + * + * SPDX-License-Identifier: MIT + */ + +/* RDMA vendor/provider identification */ + +#pragma once + +#include +#include + +#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