ps/rdma/raw_verbs_transport.cc
| Line | Branch | Exec | Source |
|---|---|---|---|
| 1 | #include "ps/rdma/raw_verbs_transport.h" | ||
| 2 | |||
| 3 | #include <arpa/inet.h> | ||
| 4 | |||
| 5 | #include <unistd.h> | ||
| 6 | |||
| 7 | #include <chrono> | ||
| 8 | #include <cstdio> | ||
| 9 | #include <cstdlib> | ||
| 10 | #include <cstring> | ||
| 11 | #include <fstream> | ||
| 12 | #include <limits> | ||
| 13 | #include <mutex> | ||
| 14 | #include <stdexcept> | ||
| 15 | #include <string> | ||
| 16 | #include <thread> | ||
| 17 | |||
| 18 | #include "ps/rdma/control_plane.h" | ||
| 19 | #include "base/log.h" | ||
| 20 | |||
| 21 | namespace petps { | ||
| 22 | namespace { | ||
| 23 | |||
| 24 | constexpr std::uint32_t kRawVerbsPsn = 3185; | ||
| 25 | constexpr int kRawVerbsCqDepth = 4096; | ||
| 26 | constexpr int kRawVerbsRecvDepth = 1024; | ||
| 27 | |||
| 28 | ✗ | std::string IbvError(const char* op) { return std::string(op) + " failed"; } | |
| 29 | |||
| 30 | ✗ | std::string GidString(const ibv_gid& gid) { | |
| 31 | ✗ | char text[INET6_ADDRSTRLEN] = {}; | |
| 32 | ✗ | return inet_ntop(AF_INET6, gid.raw, text, sizeof(text)) == nullptr | |
| 33 | ? "invalid" | ||
| 34 | ✗ | : text; | |
| 35 | } | ||
| 36 | |||
| 37 | ✗ | std::string QpCreateError(const RawVerbsConfig& config, int node) { | |
| 38 | return "ibv_create_qp failed: likely insufficient RDMA QP resources " | ||
| 39 | ✗ | "(global_id=" + | |
| 40 | ✗ | std::to_string(config.global_id) + | |
| 41 | ✗ | ", local_lane=" + std::to_string(config.local_lane) + | |
| 42 | ✗ | ", remote_lane=" + std::to_string(config.remote_lane) + | |
| 43 | ✗ | ", node=" + std::to_string(node) + | |
| 44 | ✗ | ", num_servers=" + std::to_string(config.num_servers) + | |
| 45 | ✗ | ", num_clients=" + std::to_string(config.num_clients) + | |
| 46 | ✗ | "). Reduce --client-count or --qps-per-client-per-shard."; | |
| 47 | } | ||
| 48 | |||
| 49 | // Opt-in RDMA wire trace for offline NIC bandwidth reconstruction. | ||
| 50 | // | ||
| 51 | // RECSTORE_RDMA_WIRE_TRACE=<path> appends one JSONL record per posted work | ||
| 52 | // request; RECSTORE_RDMA_WIRE_TRACE_TAG labels the writer (default: pid). The | ||
| 53 | // `node` field is the peer global node id, so client requests and server | ||
| 54 | // responses can be separated after the run. Disabled by default: the hot path | ||
| 55 | // costs one null check. | ||
| 56 | class RdmaWireTrace { | ||
| 57 | public: | ||
| 58 | ✗ | static RdmaWireTrace& Instance() { | |
| 59 | ✗ | static RdmaWireTrace instance; | |
| 60 | ✗ | return instance; | |
| 61 | } | ||
| 62 | |||
| 63 | ✗ | void Record(const char* op, int self, int node, std::size_t bytes) { | |
| 64 | ✗ | if (file_ == nullptr) { | |
| 65 | ✗ | return; | |
| 66 | } | ||
| 67 | ✗ | timespec now{}; | |
| 68 | ✗ | clock_gettime(CLOCK_MONOTONIC, &now); | |
| 69 | ✗ | const auto stamp = | |
| 70 | ✗ | static_cast<unsigned long long>(now.tv_sec) * 1000000000ull + | |
| 71 | ✗ | static_cast<unsigned long long>(now.tv_nsec); | |
| 72 | char line[224]; | ||
| 73 | ✗ | const int written = std::snprintf( | |
| 74 | line, | ||
| 75 | sizeof(line), | ||
| 76 | "{\"t\":%llu,\"tag\":\"%s\",\"pid\":%d,\"op\":\"%s\"," | ||
| 77 | "\"self\":%d,\"node\":%d,\"bytes\":%llu}\n", | ||
| 78 | stamp, | ||
| 79 | tag_.c_str(), | ||
| 80 | ✗ | static_cast<int>(getpid()), | |
| 81 | op, | ||
| 82 | self, | ||
| 83 | node, | ||
| 84 | static_cast<unsigned long long>(bytes)); | ||
| 85 | ✗ | if (written <= 0) { | |
| 86 | ✗ | return; | |
| 87 | } | ||
| 88 | // snprintf returns the *would-be* length; anything that did not fit in | ||
| 89 | // `line` would read past the buffer here. kMaxTagLen keeps this | ||
| 90 | // unreachable in practice; if it ever trips, drop the record instead of | ||
| 91 | // emitting a truncated (invalid) JSONL line. | ||
| 92 | ✗ | if (static_cast<std::size_t>(written) >= sizeof(line)) { | |
| 93 | ✗ | return; | |
| 94 | } | ||
| 95 | ✗ | std::lock_guard<std::mutex> guard(mutex_); | |
| 96 | ✗ | buffer_.append(line, static_cast<std::size_t>(written)); | |
| 97 | // The PS server is killed rather than shut down cleanly, so a purely | ||
| 98 | // size-triggered flush silently drops whatever sits in the buffer at | ||
| 99 | // exit. A time budget bounds that loss to kFlushIntervalNs of posts. | ||
| 100 | ✗ | if (buffer_.size() >= kFlushBytes || | |
| 101 | ✗ | stamp - last_flush_ns_ >= kFlushIntervalNs) { | |
| 102 | ✗ | FlushLocked(); | |
| 103 | ✗ | last_flush_ns_ = stamp; | |
| 104 | } | ||
| 105 | ✗ | } | |
| 106 | |||
| 107 | ✗ | ~RdmaWireTrace() { | |
| 108 | ✗ | std::lock_guard<std::mutex> guard(mutex_); | |
| 109 | ✗ | FlushLocked(); | |
| 110 | ✗ | if (file_ != nullptr) { | |
| 111 | ✗ | std::fclose(file_); | |
| 112 | } | ||
| 113 | ✗ | } | |
| 114 | |||
| 115 | private: | ||
| 116 | ✗ | RdmaWireTrace() { | |
| 117 | ✗ | const char* path = std::getenv("RECSTORE_RDMA_WIRE_TRACE"); | |
| 118 | ✗ | if (path == nullptr || *path == '\0') { | |
| 119 | ✗ | return; | |
| 120 | } | ||
| 121 | ✗ | const char* tag = std::getenv("RECSTORE_RDMA_WIRE_TRACE_TAG"); | |
| 122 | ✗ | tag_ = (tag != nullptr && *tag != '\0') ? tag : std::to_string(getpid()); | |
| 123 | // Keep every record inside `line[224]` in Record(); the fixed fields | ||
| 124 | // already take ~90 bytes at their widest. | ||
| 125 | ✗ | if (tag_.size() > kMaxTagLen) { | |
| 126 | ✗ | tag_.resize(kMaxTagLen); | |
| 127 | } | ||
| 128 | ✗ | file_ = std::fopen(path, "a"); | |
| 129 | ✗ | } | |
| 130 | |||
| 131 | ✗ | void FlushLocked() { | |
| 132 | ✗ | if (file_ != nullptr && !buffer_.empty()) { | |
| 133 | ✗ | std::fwrite(buffer_.data(), 1, buffer_.size(), file_); | |
| 134 | ✗ | std::fflush(file_); | |
| 135 | } | ||
| 136 | ✗ | buffer_.clear(); | |
| 137 | ✗ | } | |
| 138 | |||
| 139 | static constexpr std::size_t kMaxTagLen = 64; | ||
| 140 | static constexpr std::size_t kFlushBytes = 256 * 1024; | ||
| 141 | // 100ms worth of posts is negligible to lose at process kill time. | ||
| 142 | static constexpr unsigned long long kFlushIntervalNs = 100ull * 1000 * 1000; | ||
| 143 | std::FILE* file_ = nullptr; | ||
| 144 | std::string tag_; | ||
| 145 | std::string buffer_; | ||
| 146 | unsigned long long last_flush_ns_ = 0; | ||
| 147 | std::mutex mutex_; | ||
| 148 | }; | ||
| 149 | |||
| 150 | struct OpenedRawVerbsDevice { | ||
| 151 | ibv_context* context = nullptr; | ||
| 152 | int gid_index = -1; | ||
| 153 | ibv_port_attr port_attr{}; | ||
| 154 | ibv_gid gid{}; | ||
| 155 | }; | ||
| 156 | |||
| 157 | ✗ | OpenedRawVerbsDevice OpenDeviceExplicit(const RawVerbsConfig& config) { | |
| 158 | // ibv_fork_init must be called before any other libibverbs API. | ||
| 159 | // It sets MADV_DONTFORK on all mmap'd regions (CQ/QP memory) so that | ||
| 160 | // forked child processes (e.g. DataLoader workers) don't inherit them. | ||
| 161 | // Without this, child exit corrupts the parent's RDMA state. | ||
| 162 | ✗ | static const bool fork_init_done = []() { | |
| 163 | ✗ | ibv_fork_init(); | |
| 164 | ✗ | return true; | |
| 165 | ✗ | }(); | |
| 166 | (void)fork_init_done; | ||
| 167 | |||
| 168 | ✗ | int device_count = 0; | |
| 169 | ✗ | ibv_device** devices = ibv_get_device_list(&device_count); | |
| 170 | ✗ | if (devices == nullptr || device_count == 0) { | |
| 171 | ✗ | throw std::runtime_error("no RDMA devices found"); | |
| 172 | } | ||
| 173 | ✗ | for (int i = 0; i < device_count; ++i) { | |
| 174 | ✗ | if (config.device_name != ibv_get_device_name(devices[i])) | |
| 175 | ✗ | continue; | |
| 176 | ✗ | ibv_context* candidate = ibv_open_device(devices[i]); | |
| 177 | ✗ | if (candidate == nullptr) | |
| 178 | ✗ | throw std::runtime_error("ibv_open_device failed"); | |
| 179 | ✗ | ibv_port_attr port_attr{}; | |
| 180 | ✗ | if (ibv_query_port(candidate, config.port_num, &port_attr) != 0 || | |
| 181 | ✗ | port_attr.state != IBV_PORT_ACTIVE) { | |
| 182 | ✗ | ibv_close_device(candidate); | |
| 183 | ✗ | throw std::runtime_error("configured RDMA port is not active"); | |
| 184 | } | ||
| 185 | ✗ | const bool is_ib = port_attr.link_layer == IBV_LINK_LAYER_INFINIBAND; | |
| 186 | ✗ | if ((config.fabric_mode == recstore::RdmaFabricMode::kIb) != is_ib) { | |
| 187 | ✗ | ibv_close_device(candidate); | |
| 188 | ✗ | throw std::runtime_error( | |
| 189 | ✗ | "configured RDMA fabric mode does not match link layer"); | |
| 190 | } | ||
| 191 | ✗ | ibv_gid gid{}; | |
| 192 | ✗ | if (ibv_query_gid(candidate, config.port_num, config.gid_index, &gid) != | |
| 193 | 0) { | ||
| 194 | ✗ | ibv_close_device(candidate); | |
| 195 | ✗ | throw std::runtime_error("configured RDMA gid_index is unavailable"); | |
| 196 | } | ||
| 197 | ✗ | if (!is_ib) { | |
| 198 | std::ifstream type_input( | ||
| 199 | ✗ | std::string("/sys/class/infiniband/") + config.device_name + | |
| 200 | ✗ | "/ports/" + std::to_string(config.port_num) + "/gid_attrs/types/" + | |
| 201 | ✗ | std::to_string(config.gid_index)); | |
| 202 | ✗ | std::string gid_type; | |
| 203 | ✗ | std::getline(type_input, gid_type); | |
| 204 | const std::string expected = | ||
| 205 | ✗ | config.fabric_mode == recstore::RdmaFabricMode::kRocEv1 | |
| 206 | ? "RoCE v1" | ||
| 207 | ✗ | : "RoCE v2"; | |
| 208 | ✗ | if (gid_type != expected) { | |
| 209 | ✗ | ibv_close_device(candidate); | |
| 210 | ✗ | throw std::runtime_error( | |
| 211 | ✗ | "configured RDMA GID type does not match fabric mode"); | |
| 212 | } | ||
| 213 | ✗ | } | |
| 214 | ✗ | ibv_free_device_list(devices); | |
| 215 | ✗ | return {candidate, config.gid_index, port_attr, gid}; | |
| 216 | } | ||
| 217 | ✗ | ibv_free_device_list(devices); | |
| 218 | ✗ | throw std::runtime_error("configured RDMA device was not found"); | |
| 219 | } | ||
| 220 | |||
| 221 | ✗ | void ModifyQpToInit(ibv_qp* qp, std::uint8_t port) { | |
| 222 | ✗ | ibv_qp_attr attr{}; | |
| 223 | ✗ | attr.qp_state = IBV_QPS_INIT; | |
| 224 | ✗ | attr.port_num = port; | |
| 225 | ✗ | attr.pkey_index = 0; | |
| 226 | ✗ | attr.qp_access_flags = IBV_ACCESS_REMOTE_READ | IBV_ACCESS_REMOTE_WRITE | | |
| 227 | IBV_ACCESS_REMOTE_ATOMIC; | ||
| 228 | ✗ | const int flags = | |
| 229 | IBV_QP_STATE | IBV_QP_PKEY_INDEX | IBV_QP_PORT | IBV_QP_ACCESS_FLAGS; | ||
| 230 | ✗ | if (ibv_modify_qp(qp, &attr, flags) != 0) { | |
| 231 | ✗ | throw std::runtime_error(IbvError("ibv_modify_qp INIT")); | |
| 232 | } | ||
| 233 | ✗ | } | |
| 234 | |||
| 235 | ✗ | void FillAhAttr( | |
| 236 | ibv_ah_attr* ah_attr, | ||
| 237 | std::uint16_t remote_lid, | ||
| 238 | const std::uint8_t* remote_gid, | ||
| 239 | int local_gid_index, | ||
| 240 | std::uint8_t port, | ||
| 241 | recstore::RdmaFabricMode mode, | ||
| 242 | std::uint8_t hop_limit, | ||
| 243 | std::uint8_t traffic_class, | ||
| 244 | std::uint32_t flow_label) { | ||
| 245 | ✗ | std::memset(ah_attr, 0, sizeof(*ah_attr)); | |
| 246 | ✗ | ah_attr->dlid = remote_lid; | |
| 247 | ✗ | ah_attr->sl = 0; | |
| 248 | ✗ | ah_attr->src_path_bits = 0; | |
| 249 | ✗ | ah_attr->port_num = port; | |
| 250 | ✗ | if (remote_gid != nullptr && mode != recstore::RdmaFabricMode::kIb) { | |
| 251 | ✗ | ah_attr->is_global = 1; | |
| 252 | ✗ | std::memcpy(&ah_attr->grh.dgid, remote_gid, 16); | |
| 253 | ✗ | ah_attr->grh.sgid_index = local_gid_index; | |
| 254 | ✗ | ah_attr->grh.hop_limit = hop_limit; | |
| 255 | ✗ | ah_attr->grh.traffic_class = traffic_class; | |
| 256 | ✗ | ah_attr->grh.flow_label = flow_label; | |
| 257 | } | ||
| 258 | ✗ | } | |
| 259 | |||
| 260 | ✗ | void ModifyQpToRtr( | |
| 261 | ibv_qp* qp, | ||
| 262 | const RawVerbsNodeMeta& remote, | ||
| 263 | int local_gid_index, | ||
| 264 | std::uint8_t port, | ||
| 265 | recstore::RdmaFabricMode mode, | ||
| 266 | std::uint8_t hop_limit, | ||
| 267 | std::uint8_t traffic_class, | ||
| 268 | std::uint32_t flow_label) { | ||
| 269 | ✗ | ibv_port_attr port_attr{}; | |
| 270 | ✗ | if (ibv_query_port(qp->context, port, &port_attr) != 0) { | |
| 271 | ✗ | throw std::runtime_error(IbvError("ibv_query_port for active MTU")); | |
| 272 | } | ||
| 273 | ✗ | ibv_qp_attr attr{}; | |
| 274 | ✗ | attr.qp_state = IBV_QPS_RTR; | |
| 275 | ✗ | attr.path_mtu = static_cast<ibv_mtu>( | |
| 276 | ✗ | std::min(static_cast<int>(port_attr.active_mtu), | |
| 277 | ✗ | static_cast<int>(remote.active_mtu))); | |
| 278 | ✗ | attr.dest_qp_num = remote.qpn; | |
| 279 | ✗ | attr.rq_psn = remote.psn; | |
| 280 | ✗ | attr.max_dest_rd_atomic = 16; | |
| 281 | ✗ | attr.min_rnr_timer = 12; | |
| 282 | ✗ | FillAhAttr( | |
| 283 | &attr.ah_attr, | ||
| 284 | ✗ | remote.lid, | |
| 285 | ✗ | remote.gid, | |
| 286 | local_gid_index, | ||
| 287 | port, | ||
| 288 | mode, | ||
| 289 | hop_limit, | ||
| 290 | traffic_class, | ||
| 291 | flow_label); | ||
| 292 | ✗ | const int flags = | |
| 293 | IBV_QP_STATE | IBV_QP_AV | IBV_QP_PATH_MTU | IBV_QP_DEST_QPN | | ||
| 294 | IBV_QP_RQ_PSN | IBV_QP_MAX_DEST_RD_ATOMIC | IBV_QP_MIN_RNR_TIMER; | ||
| 295 | ✗ | if (ibv_modify_qp(qp, &attr, flags) != 0) { | |
| 296 | ✗ | throw std::runtime_error(IbvError("ibv_modify_qp RTR")); | |
| 297 | } | ||
| 298 | ✗ | } | |
| 299 | |||
| 300 | ✗ | void ModifyQpToRts(ibv_qp* qp) { | |
| 301 | ✗ | ibv_qp_attr attr{}; | |
| 302 | ✗ | attr.qp_state = IBV_QPS_RTS; | |
| 303 | ✗ | attr.sq_psn = kRawVerbsPsn; | |
| 304 | ✗ | attr.timeout = 14; | |
| 305 | ✗ | attr.retry_cnt = 7; | |
| 306 | ✗ | attr.rnr_retry = 7; | |
| 307 | ✗ | attr.max_rd_atomic = 16; | |
| 308 | ✗ | const int flags = | |
| 309 | IBV_QP_STATE | IBV_QP_SQ_PSN | IBV_QP_TIMEOUT | IBV_QP_RETRY_CNT | | ||
| 310 | IBV_QP_RNR_RETRY | IBV_QP_MAX_QP_RD_ATOMIC; | ||
| 311 | ✗ | if (ibv_modify_qp(qp, &attr, flags) != 0) { | |
| 312 | ✗ | throw std::runtime_error(IbvError("ibv_modify_qp RTS")); | |
| 313 | } | ||
| 314 | ✗ | } | |
| 315 | |||
| 316 | } // namespace | ||
| 317 | |||
| 318 | std::string | ||
| 319 | ✗ | DescribeRawVerbsCompletionError(const RawVerbsCompletionError& error) { | |
| 320 | ✗ | return std::string("raw verbs CQ error: ") + ibv_wc_status_str(error.status) + | |
| 321 | ✗ | " class=" + RawVerbsErrorClassToString(error.error_class) + | |
| 322 | ✗ | " wr_id=" + std::to_string(error.wr_id) + | |
| 323 | ✗ | " opcode=" + std::to_string(static_cast<int>(error.opcode)); | |
| 324 | } | ||
| 325 | |||
| 326 | ✗ | RawVerbsCompletionException::RawVerbsCompletionException( | |
| 327 | ✗ | const RawVerbsCompletionError& error, const std::string& context) | |
| 328 | ✗ | : std::runtime_error(DescribeRawVerbsCompletionError(error) + | |
| 329 | ✗ | (context.empty() ? std::string() : " " + context)), | |
| 330 | ✗ | error_(error) {} | |
| 331 | |||
| 332 | struct RawVerbsTransport::Impl { | ||
| 333 | ✗ | explicit Impl(const RawVerbsConfig& c) | |
| 334 | ✗ | : config(c), allocator(c.local_region_bytes, c.allocation_start_offset) {} | |
| 335 | |||
| 336 | RawVerbsConfig config; | ||
| 337 | ibv_context* context = nullptr; | ||
| 338 | int gid_index = -1; | ||
| 339 | ibv_pd* pd = nullptr; | ||
| 340 | ibv_cq* cq = nullptr; | ||
| 341 | ibv_mr* local_mr = nullptr; | ||
| 342 | std::vector<ibv_mr*> extra_mrs; | ||
| 343 | void* local_base = nullptr; | ||
| 344 | bool owns_local_base = false; | ||
| 345 | std::size_t local_bytes = 0; | ||
| 346 | RawVerbsRegionAllocator allocator; | ||
| 347 | std::vector<RawVerbsNodeMeta> metas; | ||
| 348 | std::vector<RawVerbsRemoteMemory> remotes; | ||
| 349 | std::vector<ibv_qp*> qps; | ||
| 350 | std::vector<std::uint32_t> max_inline_data_per_node; | ||
| 351 | // Per-peer lane generation this side publishes. Indexed by node id. | ||
| 352 | std::vector<std::uint64_t> lane_generations; | ||
| 353 | // One control-plane channel shared by Publish()/Connect()/RefreshPeerMeta(). | ||
| 354 | std::unique_ptr<RdmaControlPlaneClient> control_plane; | ||
| 355 | std::mutex cq_mutex; // Serializes CQ access across drains and wait harvests. | ||
| 356 | ibv_wc wc_batch[kRawVerbsPollBatchSize] = {}; | ||
| 357 | RawVerbsCompletionBatchCursor batch_cursor; | ||
| 358 | }; | ||
| 359 | |||
| 360 | ✗ | RawVerbsTransport::RawVerbsTransport(const RawVerbsConfig& config) | |
| 361 | ✗ | : impl_(std::make_unique<Impl>(config)) { | |
| 362 | ✗ | if (config.device_name.empty() || config.port_num == 0 || | |
| 363 | ✗ | config.gid_index < 0) { | |
| 364 | ✗ | throw std::invalid_argument( | |
| 365 | ✗ | "RDMA device_name, port_num, and gid_index are required"); | |
| 366 | } | ||
| 367 | ✗ | const auto opened = OpenDeviceExplicit(config); | |
| 368 | ✗ | impl_->context = opened.context; | |
| 369 | ✗ | impl_->gid_index = opened.gid_index; | |
| 370 | ✗ | if (config.local_lane == 0) { | |
| 371 | ✗ | const char* role = config.connect_to_servers ? "client" : "server"; | |
| 372 | ✗ | LOG(INFO) | |
| 373 | << "component=rdma_verbs event=fabric_ready" << " role=" << role | ||
| 374 | ✗ | << " node_id=" << config.global_id << " device=" << config.device_name | |
| 375 | ✗ | << " port=" << static_cast<int>(config.port_num) | |
| 376 | ✗ | << " gid_index=" << config.gid_index << " gid=" << GidString(opened.gid) | |
| 377 | << " link_layer=" | ||
| 378 | ✗ | << (opened.port_attr.link_layer == IBV_LINK_LAYER_INFINIBAND | |
| 379 | ? "ib" | ||
| 380 | : "ethernet") | ||
| 381 | ✗ | << " active_mtu_bytes=" | |
| 382 | ✗ | << (128 << static_cast<int>(opened.port_attr.active_mtu)); | |
| 383 | } | ||
| 384 | ✗ | impl_->pd = ibv_alloc_pd(impl_->context); | |
| 385 | ✗ | if (impl_->pd == nullptr) { | |
| 386 | ✗ | throw std::runtime_error("ibv_alloc_pd failed"); | |
| 387 | } | ||
| 388 | ✗ | impl_->cq = | |
| 389 | ✗ | ibv_create_cq(impl_->context, kRawVerbsCqDepth, nullptr, nullptr, 0); | |
| 390 | ✗ | if (impl_->cq == nullptr) { | |
| 391 | ✗ | throw std::runtime_error("ibv_create_cq failed"); | |
| 392 | } | ||
| 393 | |||
| 394 | ✗ | impl_->local_bytes = config.local_region_bytes; | |
| 395 | ✗ | impl_->local_base = reinterpret_cast<void*>(config.local_base_addr); | |
| 396 | ✗ | if (impl_->local_base == nullptr) { | |
| 397 | ✗ | const int rc = posix_memalign(&impl_->local_base, 4096, impl_->local_bytes); | |
| 398 | ✗ | if (rc != 0) { | |
| 399 | ✗ | throw std::runtime_error("posix_memalign failed for raw verbs region"); | |
| 400 | } | ||
| 401 | ✗ | impl_->owns_local_base = true; | |
| 402 | } | ||
| 403 | ✗ | impl_->local_mr = ibv_reg_mr( | |
| 404 | impl_->pd, | ||
| 405 | impl_->local_base, | ||
| 406 | impl_->local_bytes, | ||
| 407 | IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_READ | | ||
| 408 | IBV_ACCESS_REMOTE_WRITE | IBV_ACCESS_REMOTE_ATOMIC); | ||
| 409 | ✗ | if (impl_->local_mr == nullptr) { | |
| 410 | ✗ | throw std::runtime_error("ibv_reg_mr failed"); | |
| 411 | } | ||
| 412 | ✗ | if (config.reserved_region_bytes != 0) { | |
| 413 | ✗ | impl_->allocator.SetReservedRegion( | |
| 414 | ✗ | {config.reserved_region_offset, config.reserved_region_bytes}); | |
| 415 | } | ||
| 416 | |||
| 417 | ✗ | const int node_count = config.num_servers + config.num_clients; | |
| 418 | ✗ | impl_->qps.resize(static_cast<std::size_t>(node_count), nullptr); | |
| 419 | ✗ | impl_->max_inline_data_per_node.assign( | |
| 420 | ✗ | static_cast<std::size_t>(node_count), 0); | |
| 421 | ✗ | impl_->lane_generations.assign(static_cast<std::size_t>(node_count), 0); | |
| 422 | ✗ | for (int node = 0; node < node_count; ++node) { | |
| 423 | ✗ | if (!ShouldRawVerbsConnectToNode(config, node)) { | |
| 424 | ✗ | continue; | |
| 425 | } | ||
| 426 | ✗ | ibv_qp_init_attr init_attr{}; | |
| 427 | ✗ | init_attr.send_cq = impl_->cq; | |
| 428 | ✗ | init_attr.recv_cq = impl_->cq; | |
| 429 | ✗ | init_attr.qp_type = IBV_QPT_RC; | |
| 430 | ✗ | init_attr.cap.max_send_wr = 1024; | |
| 431 | ✗ | init_attr.cap.max_recv_wr = kRawVerbsRecvDepth; | |
| 432 | ✗ | init_attr.cap.max_send_sge = 32; | |
| 433 | ✗ | init_attr.cap.max_recv_sge = 1; | |
| 434 | ✗ | init_attr.cap.max_inline_data = config.max_inline_data; | |
| 435 | ✗ | ibv_qp* qp = ibv_create_qp(impl_->pd, &init_attr); | |
| 436 | ✗ | if (qp == nullptr) { | |
| 437 | ✗ | throw std::runtime_error(QpCreateError(config, node)); | |
| 438 | } | ||
| 439 | ✗ | impl_->qps[static_cast<std::size_t>(node)] = qp; | |
| 440 | ✗ | impl_->max_inline_data_per_node[static_cast<std::size_t>(node)] = | |
| 441 | ✗ | init_attr.cap.max_inline_data; | |
| 442 | } | ||
| 443 | ✗ | } | |
| 444 | |||
| 445 | ✗ | RawVerbsTransport::~RawVerbsTransport() { | |
| 446 | ✗ | if (!impl_) { | |
| 447 | ✗ | return; | |
| 448 | } | ||
| 449 | ✗ | for (ibv_qp* qp : impl_->qps) { | |
| 450 | ✗ | if (qp != nullptr) { | |
| 451 | ✗ | ibv_destroy_qp(qp); | |
| 452 | } | ||
| 453 | } | ||
| 454 | ✗ | if (impl_->local_mr != nullptr) { | |
| 455 | ✗ | ibv_dereg_mr(impl_->local_mr); | |
| 456 | } | ||
| 457 | ✗ | for (ibv_mr* mr : impl_->extra_mrs) { | |
| 458 | ✗ | if (mr != nullptr) { | |
| 459 | ✗ | ibv_dereg_mr(mr); | |
| 460 | } | ||
| 461 | } | ||
| 462 | ✗ | if (impl_->cq != nullptr) { | |
| 463 | ✗ | ibv_destroy_cq(impl_->cq); | |
| 464 | } | ||
| 465 | ✗ | if (impl_->pd != nullptr) { | |
| 466 | ✗ | ibv_dealloc_pd(impl_->pd); | |
| 467 | } | ||
| 468 | ✗ | if (impl_->context != nullptr) { | |
| 469 | ✗ | ibv_close_device(impl_->context); | |
| 470 | } | ||
| 471 | ✗ | if (impl_->owns_local_base && impl_->local_base != nullptr) { | |
| 472 | ✗ | free(impl_->local_base); | |
| 473 | } | ||
| 474 | ✗ | } | |
| 475 | |||
| 476 | namespace { | ||
| 477 | ✗ | bool MrContains(ibv_mr* mr, const void* ptr, std::size_t bytes) { | |
| 478 | ✗ | if (mr == nullptr) { | |
| 479 | ✗ | return false; | |
| 480 | } | ||
| 481 | ✗ | const auto begin = reinterpret_cast<std::uintptr_t>(ptr); | |
| 482 | ✗ | const auto end = begin + bytes; | |
| 483 | ✗ | const auto mr_begin = reinterpret_cast<std::uintptr_t>(mr->addr); | |
| 484 | ✗ | const auto mr_end = mr_begin + static_cast<std::uintptr_t>(mr->length); | |
| 485 | ✗ | return begin >= mr_begin && end >= begin && end <= mr_end; | |
| 486 | } | ||
| 487 | } // namespace | ||
| 488 | |||
| 489 | ibv_mr* | ||
| 490 | ✗ | RawVerbsTransport::FindLocalMr(const void* ptr, std::size_t bytes) const { | |
| 491 | ✗ | if (impl_->local_mr != nullptr && MrContains(impl_->local_mr, ptr, bytes)) { | |
| 492 | ✗ | return impl_->local_mr; | |
| 493 | } | ||
| 494 | ✗ | for (ibv_mr* mr : impl_->extra_mrs) { | |
| 495 | ✗ | if (MrContains(mr, ptr, bytes)) { | |
| 496 | ✗ | return mr; | |
| 497 | } | ||
| 498 | } | ||
| 499 | ✗ | return nullptr; | |
| 500 | } | ||
| 501 | |||
| 502 | ✗ | void RawVerbsTransport::RegisterMemoryRegion(void* base, std::size_t bytes) { | |
| 503 | ✗ | if (base == nullptr || bytes == 0) { | |
| 504 | ✗ | return; | |
| 505 | } | ||
| 506 | ✗ | if (FindLocalMr(base, bytes) != nullptr) { | |
| 507 | ✗ | return; | |
| 508 | } | ||
| 509 | ✗ | ibv_mr* mr = ibv_reg_mr( | |
| 510 | impl_->pd, | ||
| 511 | base, | ||
| 512 | bytes, | ||
| 513 | IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_READ | | ||
| 514 | IBV_ACCESS_REMOTE_WRITE | IBV_ACCESS_REMOTE_ATOMIC); | ||
| 515 | ✗ | if (mr == nullptr) { | |
| 516 | ✗ | throw std::runtime_error("ibv_reg_mr failed for extra raw verbs region"); | |
| 517 | } | ||
| 518 | ✗ | impl_->extra_mrs.push_back(mr); | |
| 519 | } | ||
| 520 | |||
| 521 | ✗ | void* RawVerbsTransport::AllocateRegistered(std::size_t bytes) { | |
| 522 | ✗ | const std::uint64_t offset = impl_->allocator.Allocate(bytes); | |
| 523 | ✗ | return static_cast<char*>(impl_->local_base) + offset; | |
| 524 | } | ||
| 525 | |||
| 526 | ✗ | std::uint64_t RawVerbsTransport::SaveAllocationState() const { | |
| 527 | ✗ | return impl_->allocator.Checkpoint(); | |
| 528 | } | ||
| 529 | |||
| 530 | ✗ | void RawVerbsTransport::RestoreAllocationState(std::uint64_t checkpoint) { | |
| 531 | ✗ | impl_->allocator.Restore(checkpoint); | |
| 532 | ✗ | } | |
| 533 | |||
| 534 | ✗ | GlobalAddress RawVerbsTransport::LocalAddress(void* ptr) const { | |
| 535 | ✗ | const auto base = reinterpret_cast<std::uintptr_t>(impl_->local_base); | |
| 536 | ✗ | const auto addr = reinterpret_cast<std::uintptr_t>(ptr); | |
| 537 | ✗ | if (addr < base || addr >= base + impl_->local_bytes) { | |
| 538 | ✗ | throw std::runtime_error("pointer outside raw verbs local region"); | |
| 539 | } | ||
| 540 | return GlobalAddress{ | ||
| 541 | ✗ | static_cast<std::uint16_t>(impl_->config.global_id), | |
| 542 | ✗ | impl_->config.region_origin_offset + | |
| 543 | ✗ | static_cast<std::uint64_t>(addr - base), | |
| 544 | ✗ | }; | |
| 545 | } | ||
| 546 | |||
| 547 | ✗ | void* RawVerbsTransport::LocalPointer(GlobalAddress address) const { | |
| 548 | ✗ | if (address.nodeID != impl_->config.global_id) { | |
| 549 | ✗ | throw std::runtime_error( | |
| 550 | ✗ | "raw verbs local pointer requested for remote node"); | |
| 551 | } | ||
| 552 | ✗ | const std::uint64_t origin = impl_->config.region_origin_offset; | |
| 553 | ✗ | if (address.offset < origin || | |
| 554 | ✗ | address.offset - origin >= impl_->local_bytes) { | |
| 555 | ✗ | throw std::runtime_error("raw verbs local offset outside region"); | |
| 556 | } | ||
| 557 | ✗ | return static_cast<char*>(impl_->local_base) + (address.offset - origin); | |
| 558 | } | ||
| 559 | |||
| 560 | ✗ | RawVerbsNodeMeta RawVerbsTransport::LocalMeta() const { | |
| 561 | ✗ | ibv_port_attr port_attr{}; | |
| 562 | ✗ | if (ibv_query_port(impl_->context, impl_->config.port_num, &port_attr) != 0) { | |
| 563 | ✗ | throw std::runtime_error("ibv_query_port failed"); | |
| 564 | } | ||
| 565 | ✗ | ibv_gid gid{}; | |
| 566 | ✗ | if (ibv_query_gid( | |
| 567 | ✗ | impl_->context, impl_->config.port_num, impl_->gid_index, &gid) != | |
| 568 | 0) { | ||
| 569 | ✗ | throw std::runtime_error("ibv_query_gid failed"); | |
| 570 | } | ||
| 571 | ✗ | RawVerbsNodeMeta meta{}; | |
| 572 | ✗ | meta.node_id = static_cast<std::uint16_t>(impl_->config.global_id); | |
| 573 | ✗ | meta.logical_id = | |
| 574 | ✗ | impl_->config.global_id < impl_->config.num_servers | |
| 575 | ✗ | ? impl_->config.global_id | |
| 576 | ✗ | : impl_->config.global_id - impl_->config.num_servers; | |
| 577 | ✗ | meta.lid = port_attr.lid; | |
| 578 | ✗ | meta.psn = kRawVerbsPsn; | |
| 579 | ✗ | meta.rkey = impl_->local_mr->rkey; | |
| 580 | // Peers address this lane with global offsets, so publish the base that | ||
| 581 | // makes `base_addr + offset` land inside the locally allocated slice. | ||
| 582 | ✗ | meta.base_addr = reinterpret_cast<std::uint64_t>(impl_->local_base) - | |
| 583 | ✗ | impl_->config.region_origin_offset; | |
| 584 | ✗ | meta.deployment_id = impl_->config.deployment_id; | |
| 585 | ✗ | meta.deployment_epoch = impl_->config.deployment_epoch; | |
| 586 | ✗ | meta.configuration_digest = impl_->config.configuration_digest; | |
| 587 | ✗ | meta.fabric_digest = impl_->config.fabric_digest; | |
| 588 | ✗ | meta.protocol_version = impl_->config.protocol_version; | |
| 589 | ✗ | meta.node_role = static_cast<std::uint8_t>( | |
| 590 | ✗ | impl_->config.global_id < impl_->config.num_servers | |
| 591 | ✗ | ? recstore::RdmaNodeRole::kServer | |
| 592 | : recstore::RdmaNodeRole::kClient); | ||
| 593 | ✗ | meta.port_num = impl_->config.port_num; | |
| 594 | ✗ | meta.link_layer = static_cast<std::uint8_t>( | |
| 595 | ✗ | port_attr.link_layer == IBV_LINK_LAYER_INFINIBAND | |
| 596 | ? IBV_LINK_LAYER_INFINIBAND | ||
| 597 | : IBV_LINK_LAYER_ETHERNET); | ||
| 598 | ✗ | meta.gid_index = impl_->config.gid_index; | |
| 599 | ✗ | meta.active_mtu = port_attr.active_mtu; | |
| 600 | ✗ | meta.fabric_mode = static_cast<std::uint8_t>(impl_->config.fabric_mode); | |
| 601 | ✗ | std::memcpy(meta.gid, &gid, sizeof(meta.gid)); | |
| 602 | ✗ | return meta; | |
| 603 | ✗ | } | |
| 604 | |||
| 605 | namespace { | ||
| 606 | RdmaControlPlaneEndpoint | ||
| 607 | ✗ | MakeControlPlaneEndpoint(const RawVerbsConfig& config) { | |
| 608 | return RdmaControlPlaneEndpoint{ | ||
| 609 | ✗ | config.control_plane_host, | |
| 610 | ✗ | config.control_plane_port, | |
| 611 | ✗ | config.control_plane_timeout_ms, | |
| 612 | ✗ | config.deployment_id, | |
| 613 | ✗ | config.deployment_epoch, | |
| 614 | ✗ | config.protocol_version, | |
| 615 | ✗ | config.configuration_digest, | |
| 616 | ✗ | config.fabric_digest, | |
| 617 | ✗ | }; | |
| 618 | } | ||
| 619 | |||
| 620 | // Fetches and validates `node`'s published metadata over the transport's shared | ||
| 621 | // control-plane channel. Shared by the initial Connect() handshake and the | ||
| 622 | // RefreshPeerMeta() re-read. | ||
| 623 | ✗ | RawVerbsNodeMeta FetchPeerMetaFromControlPlane( | |
| 624 | RdmaControlPlaneClient& control_plane, | ||
| 625 | const RawVerbsConfig& config, | ||
| 626 | int node, | ||
| 627 | int timeout_ms) { | ||
| 628 | const RawVerbsNodeMeta meta = control_plane.GetMeta( | ||
| 629 | node, | ||
| 630 | ✗ | config.remote_lane, | |
| 631 | ✗ | config.global_id, | |
| 632 | ✗ | config.local_lane, | |
| 633 | ✗ | timeout_ms > 0 ? timeout_ms : config.control_plane_timeout_ms); | |
| 634 | ✗ | ValidateRawVerbsPeerMeta(config, node, meta); | |
| 635 | ✗ | return meta; | |
| 636 | ✗ | } | |
| 637 | } // namespace | ||
| 638 | |||
| 639 | ✗ | RdmaControlPlaneClient& RawVerbsTransport::ControlPlane() { | |
| 640 | ✗ | if (impl_->control_plane == nullptr) { | |
| 641 | ✗ | impl_->control_plane = std::make_unique<RdmaControlPlaneClient>( | |
| 642 | ✗ | MakeControlPlaneEndpoint(impl_->config)); | |
| 643 | } | ||
| 644 | ✗ | return *impl_->control_plane; | |
| 645 | } | ||
| 646 | |||
| 647 | ✗ | void RawVerbsTransport::Publish() { | |
| 648 | ✗ | const int node_count = impl_->config.num_servers + impl_->config.num_clients; | |
| 649 | ✗ | const RawVerbsNodeMeta local = LocalMeta(); | |
| 650 | ✗ | RdmaControlPlaneClient& control_plane = ControlPlane(); | |
| 651 | ✗ | for (int node = 0; node < node_count; ++node) { | |
| 652 | ✗ | if (!ShouldRawVerbsConnectToNode(impl_->config, node)) { | |
| 653 | ✗ | continue; | |
| 654 | } | ||
| 655 | ✗ | RawVerbsNodeMeta peer_local = local; | |
| 656 | ✗ | peer_local.qpn = impl_->qps[static_cast<std::size_t>(node)]->qp_num; | |
| 657 | ✗ | peer_local.lane_generation = | |
| 658 | ✗ | impl_->lane_generations[static_cast<std::size_t>(node)]; | |
| 659 | ✗ | control_plane.PublishMeta( | |
| 660 | ✗ | impl_->config.global_id, | |
| 661 | ✗ | impl_->config.local_lane, | |
| 662 | node, | ||
| 663 | ✗ | impl_->config.remote_lane, | |
| 664 | peer_local); | ||
| 665 | ✗ | } | |
| 666 | ✗ | } | |
| 667 | |||
| 668 | ✗ | void RawVerbsTransport::Connect() { | |
| 669 | ✗ | const int node_count = impl_->config.num_servers + impl_->config.num_clients; | |
| 670 | ✗ | const RawVerbsNodeMeta local = LocalMeta(); | |
| 671 | ✗ | impl_->metas.assign(static_cast<std::size_t>(node_count), RawVerbsNodeMeta{}); | |
| 672 | ✗ | impl_->remotes.assign( | |
| 673 | ✗ | static_cast<std::size_t>(node_count), RawVerbsRemoteMemory{}); | |
| 674 | |||
| 675 | ✗ | for (int node = 0; node < node_count; ++node) { | |
| 676 | ✗ | if (node == impl_->config.global_id) { | |
| 677 | ✗ | impl_->metas[static_cast<std::size_t>(node)] = local; | |
| 678 | ✗ | impl_->remotes[static_cast<std::size_t>(node)] = RawVerbsRemoteMemory{ | |
| 679 | ✗ | local.node_id, | |
| 680 | ✗ | local.base_addr, | |
| 681 | ✗ | local.rkey, | |
| 682 | }; | ||
| 683 | ✗ | continue; | |
| 684 | } | ||
| 685 | ✗ | if (!ShouldRawVerbsConnectToNode(impl_->config, node)) { | |
| 686 | ✗ | continue; | |
| 687 | } | ||
| 688 | const RawVerbsNodeMeta meta = | ||
| 689 | ✗ | FetchPeerMetaFromControlPlane(ControlPlane(), impl_->config, node, -1); | |
| 690 | ✗ | impl_->metas[static_cast<std::size_t>(node)] = meta; | |
| 691 | ✗ | impl_->remotes[static_cast<std::size_t>(node)] = RawVerbsRemoteMemory{ | |
| 692 | ✗ | meta.node_id, | |
| 693 | ✗ | meta.base_addr, | |
| 694 | ✗ | meta.rkey, | |
| 695 | }; | ||
| 696 | ✗ | } | |
| 697 | |||
| 698 | ✗ | for (int node = 0; node < node_count; ++node) { | |
| 699 | ✗ | if (!ShouldRawVerbsConnectToNode(impl_->config, node)) { | |
| 700 | ✗ | continue; | |
| 701 | } | ||
| 702 | ✗ | ibv_qp* qp = impl_->qps[static_cast<std::size_t>(node)]; | |
| 703 | ✗ | ModifyQpToInit(qp, impl_->config.port_num); | |
| 704 | ✗ | ModifyQpToRtr( | |
| 705 | qp, | ||
| 706 | ✗ | impl_->metas[static_cast<std::size_t>(node)], | |
| 707 | ✗ | impl_->gid_index, | |
| 708 | ✗ | impl_->config.port_num, | |
| 709 | ✗ | impl_->config.fabric_mode, | |
| 710 | ✗ | impl_->config.hop_limit, | |
| 711 | ✗ | impl_->config.traffic_class, | |
| 712 | ✗ | impl_->config.flow_label); | |
| 713 | ✗ | ModifyQpToRts(qp); | |
| 714 | ✗ | for (int i = 0; i < kRawVerbsRecvDepth; ++i) { | |
| 715 | ✗ | ibv_recv_wr wr{}; | |
| 716 | ✗ | ibv_recv_wr* bad_wr = nullptr; | |
| 717 | ✗ | wr.wr_id = static_cast<std::uint64_t>(node); | |
| 718 | ✗ | wr.sg_list = nullptr; | |
| 719 | ✗ | wr.num_sge = 0; | |
| 720 | ✗ | if (ibv_post_recv(qp, &wr, &bad_wr) != 0) { | |
| 721 | ✗ | throw std::runtime_error("ibv_post_recv failed"); | |
| 722 | } | ||
| 723 | } | ||
| 724 | } | ||
| 725 | ✗ | } | |
| 726 | |||
| 727 | ✗ | void RawVerbsTransport::PublishAndConnect() { | |
| 728 | ✗ | Publish(); | |
| 729 | ✗ | Connect(); | |
| 730 | ✗ | } | |
| 731 | |||
| 732 | ✗ | void RawVerbsTransport::RebuildQp(std::uint16_t node_id) { | |
| 733 | ✗ | if (node_id >= impl_->qps.size() || | |
| 734 | ✗ | impl_->qps[static_cast<std::size_t>(node_id)] == nullptr || | |
| 735 | ✗ | !ShouldRawVerbsConnectToNode(impl_->config, node_id)) { | |
| 736 | ✗ | throw std::runtime_error("raw verbs rebuild: node is not connected"); | |
| 737 | } | ||
| 738 | ✗ | ibv_qp* qp = impl_->qps[static_cast<std::size_t>(node_id)]; | |
| 739 | |||
| 740 | // Serialize with CQ polling. Reset first: the transition flushes every | ||
| 741 | // outstanding WR on the QP (and its receive queue), and the resulting | ||
| 742 | // completions join the earlier failure completions already parked in the CQ. | ||
| 743 | ✗ | std::lock_guard<std::mutex> guard(impl_->cq_mutex); | |
| 744 | ✗ | impl_->batch_cursor.Reset(impl_->wc_batch, 0); | |
| 745 | ✗ | ibv_qp_attr reset_attr{}; | |
| 746 | ✗ | reset_attr.qp_state = IBV_QPS_RESET; | |
| 747 | ✗ | if (ibv_modify_qp(qp, &reset_attr, IBV_QP_STATE) != 0) { | |
| 748 | ✗ | throw std::runtime_error(IbvError("ibv_modify_qp RESET")); | |
| 749 | } | ||
| 750 | // Drop those completions so none is misattributed to a fresh submit. Callers | ||
| 751 | // only rebuild a QP whose CQ carries no other poster, so the drain cannot | ||
| 752 | // discard another peer QP's completions. | ||
| 753 | while (true) { | ||
| 754 | const int drained = | ||
| 755 | ✗ | ibv_poll_cq(impl_->cq, kRawVerbsPollBatchSize, impl_->wc_batch); | |
| 756 | ✗ | if (drained < 0) { | |
| 757 | ✗ | throw std::runtime_error("ibv_poll_cq failed during raw verbs rebuild"); | |
| 758 | } | ||
| 759 | ✗ | if (drained == 0) { | |
| 760 | ✗ | break; | |
| 761 | } | ||
| 762 | ✗ | } | |
| 763 | |||
| 764 | ✗ | ModifyQpToInit(qp, impl_->config.port_num); | |
| 765 | ✗ | ModifyQpToRtr( | |
| 766 | qp, | ||
| 767 | ✗ | impl_->metas[static_cast<std::size_t>(node_id)], | |
| 768 | ✗ | impl_->gid_index, | |
| 769 | ✗ | impl_->config.port_num, | |
| 770 | ✗ | impl_->config.fabric_mode, | |
| 771 | ✗ | impl_->config.hop_limit, | |
| 772 | ✗ | impl_->config.traffic_class, | |
| 773 | ✗ | impl_->config.flow_label); | |
| 774 | ✗ | ModifyQpToRts(qp); | |
| 775 | // Reset flushed the receive queue; re-arm it so incoming doorbells land. | ||
| 776 | ✗ | for (int i = 0; i < kRawVerbsRecvDepth; ++i) { | |
| 777 | ✗ | ibv_recv_wr wr{}; | |
| 778 | ✗ | ibv_recv_wr* bad_wr = nullptr; | |
| 779 | ✗ | wr.wr_id = static_cast<std::uint64_t>(node_id); | |
| 780 | ✗ | wr.sg_list = nullptr; | |
| 781 | ✗ | wr.num_sge = 0; | |
| 782 | ✗ | if (ibv_post_recv(qp, &wr, &bad_wr) != 0) { | |
| 783 | ✗ | throw std::runtime_error("ibv_post_recv failed during rebuild"); | |
| 784 | } | ||
| 785 | } | ||
| 786 | ✗ | } | |
| 787 | |||
| 788 | ✗ | std::uint64_t RawVerbsTransport::LaneGeneration(std::uint16_t node_id) const { | |
| 789 | ✗ | if (node_id >= impl_->lane_generations.size()) { | |
| 790 | ✗ | throw std::runtime_error("raw verbs generation: node out of range"); | |
| 791 | } | ||
| 792 | ✗ | return impl_->lane_generations[static_cast<std::size_t>(node_id)]; | |
| 793 | } | ||
| 794 | |||
| 795 | ✗ | std::uint64_t RawVerbsTransport::AdvanceLaneGeneration(std::uint16_t node_id) { | |
| 796 | ✗ | const std::uint64_t next = LaneGeneration(node_id) + 1; | |
| 797 | ✗ | impl_->lane_generations[static_cast<std::size_t>(node_id)] = next; | |
| 798 | ✗ | Publish(); | |
| 799 | ✗ | return next; | |
| 800 | } | ||
| 801 | |||
| 802 | RawVerbsNodeMeta | ||
| 803 | ✗ | RawVerbsTransport::RefreshPeerMeta(std::uint16_t node_id, int timeout_ms) { | |
| 804 | ✗ | if (node_id >= impl_->qps.size() || | |
| 805 | ✗ | !ShouldRawVerbsConnectToNode(impl_->config, node_id)) { | |
| 806 | ✗ | throw std::runtime_error("raw verbs peer meta: node is not connected"); | |
| 807 | } | ||
| 808 | const RawVerbsNodeMeta meta = FetchPeerMetaFromControlPlane( | ||
| 809 | ✗ | ControlPlane(), impl_->config, node_id, timeout_ms); | |
| 810 | ✗ | impl_->metas[static_cast<std::size_t>(node_id)] = meta; | |
| 811 | ✗ | impl_->remotes[static_cast<std::size_t>(node_id)] = RawVerbsRemoteMemory{ | |
| 812 | ✗ | meta.node_id, | |
| 813 | ✗ | meta.base_addr, | |
| 814 | ✗ | meta.rkey, | |
| 815 | }; | ||
| 816 | ✗ | return meta; | |
| 817 | ✗ | } | |
| 818 | |||
| 819 | ✗ | void RawVerbsTransport::Write( | |
| 820 | const void* local, | ||
| 821 | GlobalAddress remote, | ||
| 822 | std::size_t bytes, | ||
| 823 | std::uint64_t wr_id, | ||
| 824 | bool signaled) { | ||
| 825 | ✗ | if (remote.nodeID >= impl_->remotes.size()) { | |
| 826 | ✗ | throw std::runtime_error("raw verbs write remote node out of range"); | |
| 827 | } | ||
| 828 | ✗ | ibv_sge sge{}; | |
| 829 | ✗ | sge.addr = reinterpret_cast<std::uint64_t>(local); | |
| 830 | ✗ | sge.length = static_cast<std::uint32_t>(bytes); | |
| 831 | ✗ | ibv_mr* local_mr = FindLocalMr(local, bytes); | |
| 832 | ✗ | if (local_mr == nullptr) { | |
| 833 | ✗ | throw std::runtime_error("raw verbs write local buffer is not registered"); | |
| 834 | } | ||
| 835 | ✗ | sge.lkey = local_mr->lkey; | |
| 836 | ✗ | ibv_send_wr wr{}; | |
| 837 | ✗ | wr.wr_id = wr_id; | |
| 838 | ✗ | wr.opcode = IBV_WR_RDMA_WRITE; | |
| 839 | ✗ | wr.send_flags = signaled ? IBV_SEND_SIGNALED : 0; | |
| 840 | ✗ | if (bytes > 0 && | |
| 841 | ✗ | bytes <= impl_->max_inline_data_per_node[static_cast<std::size_t>( | |
| 842 | ✗ | remote.nodeID)]) { | |
| 843 | ✗ | wr.send_flags |= IBV_SEND_INLINE; | |
| 844 | } | ||
| 845 | ✗ | wr.sg_list = &sge; | |
| 846 | ✗ | wr.num_sge = 1; | |
| 847 | ✗ | wr.wr.rdma.remote_addr = | |
| 848 | ✗ | impl_->remotes[remote.nodeID].base_addr + remote.offset; | |
| 849 | ✗ | wr.wr.rdma.rkey = impl_->remotes[remote.nodeID].rkey; | |
| 850 | ✗ | ibv_send_wr* bad_wr = nullptr; | |
| 851 | ✗ | if (ibv_post_send(impl_->qps[remote.nodeID], &wr, &bad_wr) != 0) { | |
| 852 | ✗ | throw std::runtime_error("ibv_post_send write failed"); | |
| 853 | } | ||
| 854 | ✗ | RdmaWireTrace::Instance().Record( | |
| 855 | ✗ | "write", impl_->config.global_id, remote.nodeID, bytes); | |
| 856 | ✗ | } | |
| 857 | |||
| 858 | ✗ | void RawVerbsTransport::WriteSg( | |
| 859 | base::ConstArray<RawVerbsSge> sges, | ||
| 860 | GlobalAddress remote, | ||
| 861 | std::uint64_t wr_id, | ||
| 862 | bool signaled) { | ||
| 863 | ✗ | if (remote.nodeID >= impl_->remotes.size()) { | |
| 864 | ✗ | throw std::runtime_error("raw verbs write-sg remote node out of range"); | |
| 865 | } | ||
| 866 | ✗ | if (sges.Size() == 0) { | |
| 867 | ✗ | return; | |
| 868 | } | ||
| 869 | ✗ | std::vector<ibv_sge> verbs_sges; | |
| 870 | ✗ | verbs_sges.reserve(static_cast<std::size_t>(sges.Size())); | |
| 871 | ✗ | for (const auto& entry : sges) { | |
| 872 | ✗ | if (entry.bytes == 0) { | |
| 873 | ✗ | continue; | |
| 874 | } | ||
| 875 | ✗ | if (entry.bytes > std::numeric_limits<std::uint32_t>::max()) { | |
| 876 | ✗ | throw std::runtime_error("raw verbs write-sg entry too large"); | |
| 877 | } | ||
| 878 | ✗ | ibv_mr* local_mr = FindLocalMr(entry.data, entry.bytes); | |
| 879 | ✗ | if (local_mr == nullptr) { | |
| 880 | ✗ | throw std::runtime_error( | |
| 881 | ✗ | "raw verbs write-sg local buffer is not registered"); | |
| 882 | } | ||
| 883 | ✗ | ibv_sge sge{}; | |
| 884 | ✗ | sge.addr = reinterpret_cast<std::uint64_t>(entry.data); | |
| 885 | ✗ | sge.length = static_cast<std::uint32_t>(entry.bytes); | |
| 886 | ✗ | sge.lkey = local_mr->lkey; | |
| 887 | ✗ | verbs_sges.push_back(sge); | |
| 888 | } | ||
| 889 | ✗ | if (verbs_sges.empty()) { | |
| 890 | ✗ | return; | |
| 891 | } | ||
| 892 | ✗ | std::size_t total_sg_bytes = 0; | |
| 893 | ✗ | for (const auto& sge : verbs_sges) { | |
| 894 | ✗ | total_sg_bytes += sge.length; | |
| 895 | } | ||
| 896 | ✗ | ibv_send_wr wr{}; | |
| 897 | ✗ | wr.wr_id = wr_id; | |
| 898 | ✗ | wr.opcode = IBV_WR_RDMA_WRITE; | |
| 899 | ✗ | wr.send_flags = signaled ? IBV_SEND_SIGNALED : 0; | |
| 900 | ✗ | wr.sg_list = verbs_sges.data(); | |
| 901 | ✗ | wr.num_sge = static_cast<int>(verbs_sges.size()); | |
| 902 | ✗ | wr.wr.rdma.remote_addr = | |
| 903 | ✗ | impl_->remotes[remote.nodeID].base_addr + remote.offset; | |
| 904 | ✗ | wr.wr.rdma.rkey = impl_->remotes[remote.nodeID].rkey; | |
| 905 | ✗ | ibv_send_wr* bad_wr = nullptr; | |
| 906 | ✗ | if (ibv_post_send(impl_->qps[remote.nodeID], &wr, &bad_wr) != 0) { | |
| 907 | ✗ | throw std::runtime_error("ibv_post_send write-sg failed"); | |
| 908 | } | ||
| 909 | ✗ | RdmaWireTrace::Instance().Record( | |
| 910 | ✗ | "write_sg", impl_->config.global_id, remote.nodeID, total_sg_bytes); | |
| 911 | ✗ | } | |
| 912 | |||
| 913 | ✗ | void RawVerbsTransport::WriteWithImm( | |
| 914 | const void* local, | ||
| 915 | GlobalAddress remote, | ||
| 916 | std::size_t bytes, | ||
| 917 | std::uint32_t imm_data, | ||
| 918 | std::uint64_t wr_id, | ||
| 919 | bool signaled) { | ||
| 920 | ✗ | if (remote.nodeID >= impl_->remotes.size()) { | |
| 921 | ✗ | throw std::runtime_error( | |
| 922 | ✗ | "raw verbs write-with-imm remote node out of range"); | |
| 923 | } | ||
| 924 | ✗ | ibv_sge sge{}; | |
| 925 | ✗ | sge.addr = reinterpret_cast<std::uint64_t>(local); | |
| 926 | ✗ | sge.length = static_cast<std::uint32_t>(bytes); | |
| 927 | ✗ | sge.lkey = impl_->local_mr->lkey; | |
| 928 | ✗ | ibv_send_wr wr{}; | |
| 929 | ✗ | wr.wr_id = wr_id; | |
| 930 | ✗ | wr.opcode = IBV_WR_RDMA_WRITE_WITH_IMM; | |
| 931 | ✗ | wr.imm_data = htonl(imm_data); | |
| 932 | ✗ | wr.send_flags = signaled ? IBV_SEND_SIGNALED : 0; | |
| 933 | ✗ | if (bytes > 0 && | |
| 934 | ✗ | bytes <= impl_->max_inline_data_per_node[static_cast<std::size_t>( | |
| 935 | ✗ | remote.nodeID)]) { | |
| 936 | ✗ | wr.send_flags |= IBV_SEND_INLINE; | |
| 937 | } | ||
| 938 | ✗ | wr.sg_list = &sge; | |
| 939 | ✗ | wr.num_sge = 1; | |
| 940 | ✗ | wr.wr.rdma.remote_addr = | |
| 941 | ✗ | impl_->remotes[remote.nodeID].base_addr + remote.offset; | |
| 942 | ✗ | wr.wr.rdma.rkey = impl_->remotes[remote.nodeID].rkey; | |
| 943 | ✗ | ibv_send_wr* bad_wr = nullptr; | |
| 944 | ✗ | if (ibv_post_send(impl_->qps[remote.nodeID], &wr, &bad_wr) != 0) { | |
| 945 | ✗ | throw std::runtime_error("ibv_post_send write-with-imm failed"); | |
| 946 | } | ||
| 947 | ✗ | RdmaWireTrace::Instance().Record( | |
| 948 | ✗ | "write_imm", impl_->config.global_id, remote.nodeID, bytes); | |
| 949 | ✗ | } | |
| 950 | |||
| 951 | ✗ | void RawVerbsTransport::Read( | |
| 952 | void* local, | ||
| 953 | GlobalAddress remote, | ||
| 954 | std::size_t bytes, | ||
| 955 | std::uint64_t wr_id, | ||
| 956 | bool signaled) { | ||
| 957 | ✗ | if (remote.nodeID >= impl_->remotes.size()) { | |
| 958 | ✗ | throw std::runtime_error("raw verbs read remote node out of range"); | |
| 959 | } | ||
| 960 | ✗ | ibv_sge sge{}; | |
| 961 | ✗ | sge.addr = reinterpret_cast<std::uint64_t>(local); | |
| 962 | ✗ | sge.length = static_cast<std::uint32_t>(bytes); | |
| 963 | ✗ | sge.lkey = impl_->local_mr->lkey; | |
| 964 | ✗ | ibv_send_wr wr{}; | |
| 965 | ✗ | wr.wr_id = wr_id; | |
| 966 | ✗ | wr.opcode = IBV_WR_RDMA_READ; | |
| 967 | ✗ | wr.send_flags = signaled ? IBV_SEND_SIGNALED : 0; | |
| 968 | ✗ | wr.sg_list = &sge; | |
| 969 | ✗ | wr.num_sge = 1; | |
| 970 | ✗ | wr.wr.rdma.remote_addr = | |
| 971 | ✗ | impl_->remotes[remote.nodeID].base_addr + remote.offset; | |
| 972 | ✗ | wr.wr.rdma.rkey = impl_->remotes[remote.nodeID].rkey; | |
| 973 | ✗ | ibv_send_wr* bad_wr = nullptr; | |
| 974 | ✗ | if (ibv_post_send(impl_->qps[remote.nodeID], &wr, &bad_wr) != 0) { | |
| 975 | ✗ | throw std::runtime_error("ibv_post_send read failed"); | |
| 976 | } | ||
| 977 | ✗ | RdmaWireTrace::Instance().Record( | |
| 978 | ✗ | "read", impl_->config.global_id, remote.nodeID, bytes); | |
| 979 | ✗ | } | |
| 980 | |||
| 981 | ✗ | void RawVerbsTransport::SendDoorbell( | |
| 982 | std::uint16_t node_id, std::uint32_t imm_data, std::uint64_t wr_id) { | ||
| 983 | ✗ | if (node_id >= impl_->qps.size()) { | |
| 984 | ✗ | throw std::runtime_error("raw verbs doorbell remote node out of range"); | |
| 985 | } | ||
| 986 | ✗ | ibv_send_wr wr{}; | |
| 987 | ✗ | wr.opcode = IBV_WR_SEND_WITH_IMM; | |
| 988 | ✗ | wr.imm_data = htonl(imm_data); | |
| 989 | ✗ | wr.send_flags = IBV_SEND_SIGNALED; | |
| 990 | ✗ | wr.wr_id = wr_id; | |
| 991 | ✗ | wr.sg_list = nullptr; | |
| 992 | ✗ | wr.num_sge = 0; | |
| 993 | ✗ | ibv_send_wr* bad_wr = nullptr; | |
| 994 | ✗ | if (ibv_post_send(impl_->qps[node_id], &wr, &bad_wr) != 0) { | |
| 995 | ✗ | throw std::runtime_error("ibv_post_send doorbell failed"); | |
| 996 | } | ||
| 997 | ✗ | } | |
| 998 | |||
| 999 | ✗ | RawVerbsPollOutcome RawVerbsTransport::TryPoll(RawVerbsCompletion* completion, | |
| 1000 | RawVerbsCompletionError* error) { | ||
| 1001 | ✗ | std::lock_guard<std::mutex> guard(impl_->cq_mutex); | |
| 1002 | ✗ | if (!impl_->batch_cursor.HasCachedCompletion()) { | |
| 1003 | const int n = | ||
| 1004 | ✗ | ibv_poll_cq(impl_->cq, kRawVerbsPollBatchSize, impl_->wc_batch); | |
| 1005 | ✗ | if (n < 0) { | |
| 1006 | ✗ | throw std::runtime_error("ibv_poll_cq failed"); | |
| 1007 | } | ||
| 1008 | ✗ | if (n == 0) { | |
| 1009 | ✗ | return RawVerbsPollOutcome::kEmpty; | |
| 1010 | } | ||
| 1011 | ✗ | impl_->batch_cursor.Reset(impl_->wc_batch, n); | |
| 1012 | } | ||
| 1013 | ✗ | ibv_wc& wc = *impl_->batch_cursor.TakeCachedCompletion(); | |
| 1014 | ✗ | if (wc.status != IBV_WC_SUCCESS) { | |
| 1015 | ✗ | error->status = wc.status; | |
| 1016 | ✗ | error->wr_id = wc.wr_id; | |
| 1017 | ✗ | error->opcode = wc.opcode; | |
| 1018 | ✗ | error->error_class = ClassifyRawVerbsCompletionStatus(wc.status); | |
| 1019 | ✗ | return RawVerbsPollOutcome::kError; | |
| 1020 | } | ||
| 1021 | ✗ | if (completion != nullptr) { | |
| 1022 | ✗ | completion->wr_id = wc.wr_id; | |
| 1023 | ✗ | completion->opcode = wc.opcode; | |
| 1024 | ✗ | completion->has_imm = (wc.wc_flags & IBV_WC_WITH_IMM) != 0; | |
| 1025 | ✗ | completion->imm_data = completion->has_imm ? ntohl(wc.imm_data) : 0; | |
| 1026 | } | ||
| 1027 | ✗ | if (wc.opcode == IBV_WC_RECV || wc.opcode == IBV_WC_RECV_RDMA_WITH_IMM) { | |
| 1028 | ✗ | const std::uint16_t node_id = static_cast<std::uint16_t>(wc.wr_id); | |
| 1029 | ✗ | ibv_recv_wr wr{}; | |
| 1030 | ✗ | ibv_recv_wr* bad_wr = nullptr; | |
| 1031 | ✗ | wr.wr_id = node_id; | |
| 1032 | ✗ | wr.sg_list = nullptr; | |
| 1033 | ✗ | wr.num_sge = 0; | |
| 1034 | ✗ | if (node_id < impl_->qps.size() && | |
| 1035 | ✗ | ibv_post_recv(impl_->qps[node_id], &wr, &bad_wr) != 0) { | |
| 1036 | ✗ | throw std::runtime_error("ibv_post_recv repost failed"); | |
| 1037 | } | ||
| 1038 | } | ||
| 1039 | ✗ | return RawVerbsPollOutcome::kCompletion; | |
| 1040 | ✗ | } | |
| 1041 | |||
| 1042 | ✗ | bool RawVerbsTransport::Poll(RawVerbsCompletion* completion, int timeout_ms) { | |
| 1043 | const auto deadline = | ||
| 1044 | ✗ | timeout_ms > 0 | |
| 1045 | ✗ | ? std::chrono::steady_clock::now() + | |
| 1046 | ✗ | std::chrono::milliseconds(timeout_ms) | |
| 1047 | ✗ | : std::chrono::steady_clock::time_point::max(); | |
| 1048 | while (true) { | ||
| 1049 | ✗ | RawVerbsCompletionError error; | |
| 1050 | ✗ | const RawVerbsPollOutcome outcome = TryPoll(completion, &error); | |
| 1051 | ✗ | if (outcome == RawVerbsPollOutcome::kCompletion) { | |
| 1052 | ✗ | return true; | |
| 1053 | } | ||
| 1054 | ✗ | if (outcome == RawVerbsPollOutcome::kError) { | |
| 1055 | ✗ | throw RawVerbsCompletionException(error); | |
| 1056 | } | ||
| 1057 | ✗ | if (std::chrono::steady_clock::now() >= deadline) { | |
| 1058 | ✗ | return false; | |
| 1059 | } | ||
| 1060 | ✗ | std::this_thread::yield(); | |
| 1061 | ✗ | } | |
| 1062 | } | ||
| 1063 | |||
| 1064 | ✗ | std::uint32_t RawVerbsTransport::max_inline_data(std::uint16_t node_id) const { | |
| 1065 | ✗ | if (node_id >= impl_->max_inline_data_per_node.size()) { | |
| 1066 | ✗ | throw std::runtime_error("raw verbs inline query remote node out of range"); | |
| 1067 | } | ||
| 1068 | ✗ | return impl_->max_inline_data_per_node[static_cast<std::size_t>(node_id)]; | |
| 1069 | } | ||
| 1070 | |||
| 1071 | } // namespace petps | ||
| 1072 |