ps/rdma/rc_transport.cc
| Line | Branch | Exec | Source |
|---|---|---|---|
| 1 | #include "ps/rdma/rc_transport.h" | ||
| 2 | |||
| 3 | #include <algorithm> | ||
| 4 | #include <atomic> | ||
| 5 | #include <chrono> | ||
| 6 | #include <cstring> | ||
| 7 | #include <iostream> | ||
| 8 | #include <stdexcept> | ||
| 9 | #include <string> | ||
| 10 | #include <thread> | ||
| 11 | |||
| 12 | #include <folly/portability/GFlags.h> | ||
| 13 | |||
| 14 | #include "base/log.h" | ||
| 15 | #include "ps/rdma/control_plane.h" | ||
| 16 | #include "ps/rdma/rdma_common.h" | ||
| 17 | #include "ps/rdma/rc_options.h" | ||
| 18 | |||
| 19 | namespace petps { | ||
| 20 | namespace { | ||
| 21 | |||
| 22 | using petps::Exchange; | ||
| 23 | using petps::NowNs; | ||
| 24 | |||
| 25 | enum class TrackedWrKind : std::uint64_t { | ||
| 26 | kSubmitCommit = 1, | ||
| 27 | kResponseStatus = 2, | ||
| 28 | }; | ||
| 29 | |||
| 30 | struct RcTransportProfileCounters { | ||
| 31 | std::atomic<std::uint64_t> submit_request_count{0}; | ||
| 32 | std::atomic<std::uint64_t> submit_descriptor_write_count{0}; | ||
| 33 | std::atomic<std::uint64_t> submit_commit_write_count{0}; | ||
| 34 | std::atomic<std::uint64_t> submit_request_ns{0}; | ||
| 35 | std::atomic<std::uint64_t> drain_pending_submit_count{0}; | ||
| 36 | std::atomic<std::uint64_t> drain_pending_submit_ns{0}; | ||
| 37 | |||
| 38 | std::atomic<std::uint64_t> complete_response_count{0}; | ||
| 39 | std::atomic<std::uint64_t> response_payload_write_count{0}; | ||
| 40 | std::atomic<std::uint64_t> response_payload_sg_write_count{0}; | ||
| 41 | std::atomic<std::uint64_t> response_payload_sg_wr_count{0}; | ||
| 42 | std::atomic<std::uint64_t> response_status_write_count{0}; | ||
| 43 | std::atomic<std::uint64_t> response_payload_bytes{0}; | ||
| 44 | std::atomic<std::uint64_t> complete_response_ns{0}; | ||
| 45 | std::atomic<std::uint64_t> drain_pending_response_count{0}; | ||
| 46 | std::atomic<std::uint64_t> drain_pending_response_ns{0}; | ||
| 47 | std::atomic<std::uint64_t> next_report_ns{0}; | ||
| 48 | }; | ||
| 49 | |||
| 50 | ✗ | RcTransportProfileCounters& TransportProfile() { | |
| 51 | static RcTransportProfileCounters counters; | ||
| 52 | ✗ | return counters; | |
| 53 | } | ||
| 54 | |||
| 55 | ✗ | void MaybeReportTransportProfile(const RcTransportConfig& config, | |
| 56 | const char* role) { | ||
| 57 | ✗ | if (FLAGS_rdma_rc_profile_interval_ms <= 0) { | |
| 58 | ✗ | return; | |
| 59 | } | ||
| 60 | ✗ | auto& counters = TransportProfile(); | |
| 61 | ✗ | const std::uint64_t now = NowNs(); | |
| 62 | ✗ | const std::uint64_t interval = | |
| 63 | ✗ | static_cast<std::uint64_t>(FLAGS_rdma_rc_profile_interval_ms) * 1000000; | |
| 64 | std::uint64_t expected = | ||
| 65 | ✗ | counters.next_report_ns.load(std::memory_order_relaxed); | |
| 66 | ✗ | if (expected == 0) { | |
| 67 | ✗ | counters.next_report_ns.compare_exchange_strong( | |
| 68 | expected, now + interval, std::memory_order_relaxed); | ||
| 69 | ✗ | return; | |
| 70 | } | ||
| 71 | ✗ | if (now < expected || | |
| 72 | ✗ | !counters.next_report_ns.compare_exchange_strong( | |
| 73 | expected, now + interval, std::memory_order_relaxed)) { | ||
| 74 | ✗ | return; | |
| 75 | } | ||
| 76 | |||
| 77 | ✗ | const std::uint64_t submit_count = Exchange(&counters.submit_request_count); | |
| 78 | ✗ | const std::uint64_t submit_ns = Exchange(&counters.submit_request_ns); | |
| 79 | const std::uint64_t complete_count = | ||
| 80 | ✗ | Exchange(&counters.complete_response_count); | |
| 81 | ✗ | const std::uint64_t complete_ns = Exchange(&counters.complete_response_ns); | |
| 82 | const std::uint64_t drain_submit_count = | ||
| 83 | ✗ | Exchange(&counters.drain_pending_submit_count); | |
| 84 | const std::uint64_t drain_submit_ns = | ||
| 85 | ✗ | Exchange(&counters.drain_pending_submit_ns); | |
| 86 | const std::uint64_t drain_response_count = | ||
| 87 | ✗ | Exchange(&counters.drain_pending_response_count); | |
| 88 | const std::uint64_t drain_response_ns = | ||
| 89 | ✗ | Exchange(&counters.drain_pending_response_ns); | |
| 90 | std::cout | ||
| 91 | << "component=rdma_rc_transport_profile role=" << role | ||
| 92 | ✗ | << " shard=" << config.shard_id << " client_id=" << config.client_id | |
| 93 | ✗ | << " submit_count=" << submit_count << " submit_descriptor_writes=" | |
| 94 | ✗ | << Exchange(&counters.submit_descriptor_write_count) | |
| 95 | ✗ | << " submit_commit_writes=" | |
| 96 | ✗ | << Exchange(&counters.submit_commit_write_count) | |
| 97 | ✗ | << " submit_avg_ns=" << (submit_count == 0 ? 0 : submit_ns / submit_count) | |
| 98 | ✗ | << " drain_submit_count=" << drain_submit_count << " drain_submit_avg_ns=" | |
| 99 | ✗ | << (drain_submit_count == 0 ? 0 : drain_submit_ns / drain_submit_count) | |
| 100 | ✗ | << " complete_count=" << complete_count << " response_payload_writes=" | |
| 101 | ✗ | << Exchange(&counters.response_payload_write_count) | |
| 102 | ✗ | << " response_payload_sg_writes=" | |
| 103 | ✗ | << Exchange(&counters.response_payload_sg_write_count) | |
| 104 | ✗ | << " response_payload_sg_wrs=" | |
| 105 | ✗ | << Exchange(&counters.response_payload_sg_wr_count) | |
| 106 | ✗ | << " response_status_writes=" | |
| 107 | ✗ | << Exchange(&counters.response_status_write_count) | |
| 108 | ✗ | << " response_payload_bytes=" | |
| 109 | ✗ | << Exchange(&counters.response_payload_bytes) << " complete_avg_ns=" | |
| 110 | ✗ | << (complete_count == 0 ? 0 : complete_ns / complete_count) | |
| 111 | ✗ | << " drain_response_count=" << drain_response_count | |
| 112 | ✗ | << " drain_response_avg_ns=" | |
| 113 | << (drain_response_count == 0 ? 0 | ||
| 114 | ✗ | : drain_response_ns / drain_response_count) | |
| 115 | ✗ | << std::endl; | |
| 116 | } | ||
| 117 | |||
| 118 | ✗ | std::size_t TotalClientSlotsPerShard(const RcTransportConfig& config) { | |
| 119 | ✗ | return static_cast<std::size_t>(config.qps_per_client_per_shard) * | |
| 120 | ✗ | static_cast<std::size_t>(config.slots_per_qp); | |
| 121 | } | ||
| 122 | |||
| 123 | ✗ | std::size_t ClientSlotBytes(const RcTransportConfig& config) { | |
| 124 | ✗ | return config.response_slot_bytes + config.request_slot_bytes; | |
| 125 | } | ||
| 126 | |||
| 127 | ✗ | std::size_t ClientLaneBytes(const RcTransportConfig& config) { | |
| 128 | ✗ | return static_cast<std::size_t>(config.slots_per_qp) * | |
| 129 | ✗ | ClientSlotBytes(config); | |
| 130 | } | ||
| 131 | |||
| 132 | ✗ | int LogicalClientsPerProcess(const RcTransportConfig& config) { | |
| 133 | ✗ | if (config.num_os_clients <= 0) { | |
| 134 | ✗ | throw std::runtime_error("num_os_clients must be positive"); | |
| 135 | } | ||
| 136 | ✗ | if (config.num_clients < config.num_os_clients) { | |
| 137 | ✗ | throw std::runtime_error( | |
| 138 | ✗ | "logical client count smaller than OS client process count"); | |
| 139 | } | ||
| 140 | ✗ | if (config.num_clients % config.num_os_clients != 0) { | |
| 141 | ✗ | throw std::runtime_error( | |
| 142 | ✗ | "logical client count must be divisible by OS client process count"); | |
| 143 | } | ||
| 144 | ✗ | return config.num_clients / config.num_os_clients; | |
| 145 | } | ||
| 146 | |||
| 147 | ✗ | int OsClientIndexForLogicalClient(const RcTransportConfig& config, | |
| 148 | int client_id) { | ||
| 149 | ✗ | return client_id / LogicalClientsPerProcess(config); | |
| 150 | } | ||
| 151 | |||
| 152 | ✗ | int LocalLogicalClientIndex(const RcTransportConfig& config, int client_id) { | |
| 153 | ✗ | return client_id % LogicalClientsPerProcess(config); | |
| 154 | } | ||
| 155 | |||
| 156 | ✗ | int RawLaneForLogicalClient( | |
| 157 | const RcTransportConfig& config, int client_id, int qp_index) { | ||
| 158 | ✗ | return LocalLogicalClientIndex(config, client_id) * | |
| 159 | ✗ | config.qps_per_client_per_shard + | |
| 160 | ✗ | qp_index; | |
| 161 | } | ||
| 162 | |||
| 163 | ✗ | int RawLanesPerOsClient(const RcTransportConfig& config) { | |
| 164 | ✗ | return LogicalClientsPerProcess(config) * config.qps_per_client_per_shard; | |
| 165 | } | ||
| 166 | |||
| 167 | ✗ | std::size_t ServerLaneBytes(const RcTransportConfig& config) { | |
| 168 | ✗ | return static_cast<std::size_t>(config.num_clients) * | |
| 169 | ✗ | static_cast<std::size_t>(config.slots_per_qp) * | |
| 170 | ✗ | config.request_slot_bytes; | |
| 171 | } | ||
| 172 | |||
| 173 | std::size_t | ||
| 174 | ✗ | ClientShardLaneOffset(const RcTransportConfig& config, int raw_lane) { | |
| 175 | ✗ | return (static_cast<std::size_t>(config.shard_id) * | |
| 176 | ✗ | static_cast<std::size_t>(RawLanesPerOsClient(config)) + | |
| 177 | ✗ | static_cast<std::size_t>(raw_lane)) * | |
| 178 | ✗ | ClientLaneBytes(config); | |
| 179 | } | ||
| 180 | |||
| 181 | ✗ | std::size_t ClientSlotOffset(const RcTransportConfig& config, int slot_in_qp) { | |
| 182 | ✗ | return static_cast<std::size_t>(slot_in_qp) * ClientSlotBytes(config); | |
| 183 | } | ||
| 184 | |||
| 185 | ✗ | std::size_t ServerRequestOffset( | |
| 186 | const RcTransportConfig& config, int client_id, int slot_in_qp) { | ||
| 187 | ✗ | return (static_cast<std::size_t>(client_id) * | |
| 188 | ✗ | static_cast<std::size_t>(config.slots_per_qp) + | |
| 189 | ✗ | static_cast<std::size_t>(slot_in_qp)) * | |
| 190 | ✗ | config.request_slot_bytes; | |
| 191 | } | ||
| 192 | |||
| 193 | ✗ | std::size_t ClientResponseOffsetForRawLane( | |
| 194 | const RcTransportConfig& config, int raw_lane, int slot_in_qp) { | ||
| 195 | ✗ | return ClientShardLaneOffset(config, raw_lane) + | |
| 196 | ✗ | ClientSlotOffset(config, slot_in_qp); | |
| 197 | } | ||
| 198 | |||
| 199 | std::size_t | ||
| 200 | ✗ | ClientResponseOffset(const RcTransportConfig& config, int slot_in_qp) { | |
| 201 | ✗ | return ClientResponseOffsetForRawLane( | |
| 202 | ✗ | config, RawLaneForLogicalClient(config, config.client_id, 0), slot_in_qp); | |
| 203 | } | ||
| 204 | |||
| 205 | std::size_t | ||
| 206 | ✗ | ClientRequestStagingOffset(const RcTransportConfig& config, int slot_in_qp) { | |
| 207 | ✗ | return ClientResponseOffset(config, slot_in_qp) + config.response_slot_bytes; | |
| 208 | } | ||
| 209 | |||
| 210 | ✗ | std::uint64_t RequestCommitOffset(const RcTransportConfig& config) { | |
| 211 | ✗ | return config.request_slot_bytes - Align64(sizeof(CommitWord)); | |
| 212 | } | ||
| 213 | |||
| 214 | ✗ | std::uint64_t ResponseStatusOffset(const RcTransportConfig& config) { | |
| 215 | ✗ | return config.response_slot_bytes - Align64(sizeof(StatusWord)); | |
| 216 | } | ||
| 217 | |||
| 218 | ✗ | int GlobalSlotIndex(const RcTransportConfig& config, | |
| 219 | int client_id, | ||
| 220 | int qp_index, | ||
| 221 | int slot_in_qp) { | ||
| 222 | return static_cast<int>( | ||
| 223 | ✗ | (static_cast<std::size_t>(client_id) * TotalClientSlotsPerShard(config)) + | |
| 224 | ✗ | (static_cast<std::size_t>(qp_index) * | |
| 225 | ✗ | static_cast<std::size_t>(config.slots_per_qp)) + | |
| 226 | ✗ | static_cast<std::size_t>(slot_in_qp)); | |
| 227 | } | ||
| 228 | |||
| 229 | ✗ | void DecodeGlobalSlotIndex( | |
| 230 | const RcTransportConfig& config, | ||
| 231 | int slot_index, | ||
| 232 | int* client_id, | ||
| 233 | int* qp_index, | ||
| 234 | int* slot_in_qp) { | ||
| 235 | ✗ | if (slot_index < 0) { | |
| 236 | ✗ | throw std::runtime_error("slot_index out of range"); | |
| 237 | } | ||
| 238 | ✗ | const std::size_t slots_per_client = TotalClientSlotsPerShard(config); | |
| 239 | ✗ | if (slots_per_client == 0) { | |
| 240 | ✗ | throw std::runtime_error("slots_per_client is zero"); | |
| 241 | } | ||
| 242 | ✗ | const std::size_t slot = static_cast<std::size_t>(slot_index); | |
| 243 | ✗ | if (client_id != nullptr) { | |
| 244 | ✗ | *client_id = static_cast<int>(slot / slots_per_client); | |
| 245 | } | ||
| 246 | ✗ | const std::size_t slot_in_client = slot % slots_per_client; | |
| 247 | ✗ | if (qp_index != nullptr) { | |
| 248 | ✗ | *qp_index = static_cast<int>( | |
| 249 | ✗ | slot_in_client / static_cast<std::size_t>(config.slots_per_qp)); | |
| 250 | } | ||
| 251 | ✗ | if (slot_in_qp != nullptr) { | |
| 252 | ✗ | *slot_in_qp = static_cast<int>( | |
| 253 | ✗ | slot_in_client % static_cast<std::size_t>(config.slots_per_qp)); | |
| 254 | } | ||
| 255 | ✗ | } | |
| 256 | |||
| 257 | ✗ | int ResponseSlotOrdinal( | |
| 258 | const RcTransportConfig& config, int client_id, int slot_in_qp) { | ||
| 259 | ✗ | return client_id * config.slots_per_qp + slot_in_qp; | |
| 260 | } | ||
| 261 | |||
| 262 | ✗ | std::uint64_t MakeTrackedWrId(TrackedWrKind kind, int slot_ordinal) { | |
| 263 | ✗ | return (static_cast<std::uint64_t>(kind) << 32) | | |
| 264 | ✗ | static_cast<std::uint32_t>(slot_ordinal); | |
| 265 | } | ||
| 266 | |||
| 267 | ✗ | int DecodeTrackedWrId( | |
| 268 | TrackedWrKind kind, std::uint64_t wr_id, std::size_t expected_slots) { | ||
| 269 | ✗ | const std::uint64_t kind_bits = wr_id >> 32; | |
| 270 | ✗ | if (kind_bits != static_cast<std::uint64_t>(kind)) { | |
| 271 | ✗ | throw std::runtime_error("unexpected tracked RC WR kind"); | |
| 272 | } | ||
| 273 | ✗ | const std::uint32_t slot_ordinal = static_cast<std::uint32_t>(wr_id); | |
| 274 | ✗ | if (slot_ordinal >= expected_slots) { | |
| 275 | ✗ | throw std::runtime_error("tracked RC WR slot ordinal out of range"); | |
| 276 | } | ||
| 277 | ✗ | return static_cast<int>(slot_ordinal); | |
| 278 | } | ||
| 279 | |||
| 280 | ✗ | void WaitForTrackedCompletion( | |
| 281 | RawVerbsTransport* verbs, | ||
| 282 | std::vector<std::uint8_t>* ready, | ||
| 283 | TrackedWrKind kind, | ||
| 284 | int slot_ordinal, | ||
| 285 | const std::string& context) { | ||
| 286 | ✗ | if (verbs == nullptr || ready == nullptr) { | |
| 287 | ✗ | throw std::runtime_error("tracked completion state is null"); | |
| 288 | } | ||
| 289 | ✗ | auto& ready_flags = *ready; | |
| 290 | ✗ | if (slot_ordinal < 0 || | |
| 291 | ✗ | static_cast<std::size_t>(slot_ordinal) >= ready_flags.size()) { | |
| 292 | ✗ | throw std::runtime_error("tracked completion slot ordinal out of range"); | |
| 293 | } | ||
| 294 | ✗ | if (ready_flags[static_cast<std::size_t>(slot_ordinal)] != 0) { | |
| 295 | ✗ | ready_flags[static_cast<std::size_t>(slot_ordinal)] = 0; | |
| 296 | ✗ | return; | |
| 297 | } | ||
| 298 | |||
| 299 | while (true) { | ||
| 300 | ✗ | RawVerbsCompletion completion; | |
| 301 | ✗ | if (!verbs->Poll(&completion, FLAGS_rdma_wait_timeout_ms)) { | |
| 302 | ✗ | throw RcTrackedCompletionTimeout( | |
| 303 | ✗ | "RC verbs write completion timeout " + context + " expected_wr_id=" + | |
| 304 | ✗ | std::to_string(MakeTrackedWrId(kind, slot_ordinal))); | |
| 305 | } | ||
| 306 | const int completed_slot = | ||
| 307 | ✗ | DecodeTrackedWrId(kind, completion.wr_id, ready_flags.size()); | |
| 308 | ✗ | if (completed_slot == slot_ordinal) { | |
| 309 | ✗ | return; | |
| 310 | } | ||
| 311 | ✗ | ready_flags[static_cast<std::size_t>(completed_slot)] = 1; | |
| 312 | ✗ | } | |
| 313 | } | ||
| 314 | |||
| 315 | ✗ | void DrainTrackedPendingWrite( | |
| 316 | RawVerbsTransport* verbs, | ||
| 317 | std::vector<std::uint8_t>* pending, | ||
| 318 | std::vector<std::uint8_t>* ready, | ||
| 319 | TrackedWrKind kind, | ||
| 320 | int slot_ordinal, | ||
| 321 | const std::string& context) { | ||
| 322 | ✗ | if (pending == nullptr || ready == nullptr) { | |
| 323 | ✗ | return; | |
| 324 | } | ||
| 325 | ✗ | auto& pending_flags = *pending; | |
| 326 | ✗ | if (slot_ordinal < 0 || | |
| 327 | ✗ | static_cast<std::size_t>(slot_ordinal) >= pending_flags.size()) { | |
| 328 | ✗ | throw std::runtime_error("tracked pending slot ordinal out of range"); | |
| 329 | } | ||
| 330 | ✗ | if (pending_flags[static_cast<std::size_t>(slot_ordinal)] == 0) { | |
| 331 | ✗ | return; | |
| 332 | } | ||
| 333 | ✗ | WaitForTrackedCompletion(verbs, ready, kind, slot_ordinal, context); | |
| 334 | ✗ | pending_flags[static_cast<std::size_t>(slot_ordinal)] = 0; | |
| 335 | } | ||
| 336 | |||
| 337 | ✗ | void DrainTrackedPendingWrite( | |
| 338 | RawVerbsTransport* verbs, | ||
| 339 | std::vector<std::uint8_t>* pending, | ||
| 340 | std::vector<std::uint8_t>* ready, | ||
| 341 | TrackedWrKind kind, | ||
| 342 | int slot_ordinal, | ||
| 343 | const std::string& context, | ||
| 344 | bool profile_enabled, | ||
| 345 | std::atomic<std::uint64_t>* drain_count, | ||
| 346 | std::atomic<std::uint64_t>* drain_ns) { | ||
| 347 | ✗ | if (pending == nullptr || ready == nullptr || drain_count == nullptr || | |
| 348 | drain_ns == nullptr) { | ||
| 349 | ✗ | return; | |
| 350 | } | ||
| 351 | ✗ | if (slot_ordinal < 0 || | |
| 352 | ✗ | static_cast<std::size_t>(slot_ordinal) >= pending->size() || | |
| 353 | ✗ | pending->at(static_cast<std::size_t>(slot_ordinal)) == 0) { | |
| 354 | ✗ | return; | |
| 355 | } | ||
| 356 | ✗ | if (!profile_enabled) { | |
| 357 | ✗ | DrainTrackedPendingWrite( | |
| 358 | verbs, pending, ready, kind, slot_ordinal, context); | ||
| 359 | ✗ | return; | |
| 360 | } | ||
| 361 | |||
| 362 | ✗ | const std::uint64_t drain_start_ns = NowNs(); | |
| 363 | ✗ | DrainTrackedPendingWrite(verbs, pending, ready, kind, slot_ordinal, context); | |
| 364 | ✗ | drain_count->fetch_add(1, std::memory_order_relaxed); | |
| 365 | ✗ | drain_ns->fetch_add(NowNs() - drain_start_ns, std::memory_order_relaxed); | |
| 366 | } | ||
| 367 | |||
| 368 | ✗ | std::string WriteContext( | |
| 369 | const RcTransportConfig& config, | ||
| 370 | int client_id, | ||
| 371 | int qp_index, | ||
| 372 | int slot_in_qp, | ||
| 373 | std::uint64_t seq, | ||
| 374 | std::uint64_t remote_offset, | ||
| 375 | int remote_node, | ||
| 376 | const char* phase) { | ||
| 377 | ✗ | return "phase=" + std::string(phase) + | |
| 378 | ✗ | " shard=" + std::to_string(config.shard_id) + " client_id=" + | |
| 379 | ✗ | std::to_string(client_id) + " qp=" + std::to_string(qp_index) + | |
| 380 | ✗ | " slot_in_qp=" + std::to_string(slot_in_qp) + " seq=" + | |
| 381 | ✗ | std::to_string(seq) + " remote_node=" + std::to_string(remote_node) + | |
| 382 | ✗ | " remote_offset=" + std::to_string(remote_offset); | |
| 383 | } | ||
| 384 | |||
| 385 | ✗ | void ValidateClientId(const RcTransportConfig& config, int client_id) { | |
| 386 | ✗ | if (client_id < 0 || client_id >= config.num_clients) { | |
| 387 | ✗ | throw std::runtime_error("client_id out of range"); | |
| 388 | } | ||
| 389 | ✗ | } | |
| 390 | |||
| 391 | ✗ | void ValidateSlotInQp(const RcTransportConfig& config, int slot_in_qp) { | |
| 392 | ✗ | if (slot_in_qp < 0 || slot_in_qp >= config.slots_per_qp) { | |
| 393 | ✗ | throw std::runtime_error("slot_in_qp out of range"); | |
| 394 | } | ||
| 395 | ✗ | } | |
| 396 | |||
| 397 | ✗ | void ValidateTransportConfig(const RcTransportConfig& config) { | |
| 398 | ✗ | if (config.node_id < 0 || config.num_servers <= 0 || | |
| 399 | ✗ | config.num_os_clients <= 0 || config.num_clients <= 0) { | |
| 400 | ✗ | throw std::runtime_error( | |
| 401 | ✗ | "RDMA transport deployment identity is incomplete"); | |
| 402 | } | ||
| 403 | ✗ | if (config.request_slot_bytes <= | |
| 404 | ✗ | Align64(sizeof(RequestDescriptor)) + Align64(sizeof(CommitWord)) || | |
| 405 | ✗ | config.response_slot_bytes <= Align64(sizeof(StatusWord))) { | |
| 406 | ✗ | throw std::runtime_error("RDMA request/response slot bytes are too small"); | |
| 407 | } | ||
| 408 | ✗ | } | |
| 409 | |||
| 410 | ✗ | RawVerbsConfig MakeRawConfig( | |
| 411 | const RcTransportConfig& config, | ||
| 412 | int local_lane, | ||
| 413 | std::size_t local_region_bytes, | ||
| 414 | bool is_client, | ||
| 415 | int only_node_id) { | ||
| 416 | ✗ | RawVerbsConfig raw; | |
| 417 | ✗ | raw.global_id = config.node_id; | |
| 418 | ✗ | raw.local_lane = local_lane; | |
| 419 | ✗ | raw.remote_lane = local_lane; | |
| 420 | ✗ | raw.only_node_id = only_node_id; | |
| 421 | ✗ | raw.num_servers = config.num_servers; | |
| 422 | ✗ | raw.num_clients = config.num_os_clients; | |
| 423 | ✗ | raw.device_name = config.fabric.device; | |
| 424 | ✗ | raw.port_num = static_cast<std::uint8_t>(config.fabric.port); | |
| 425 | ✗ | raw.fabric_mode = config.fabric.mode; | |
| 426 | ✗ | raw.gid_index = config.fabric.gid_index; | |
| 427 | ✗ | raw.hop_limit = static_cast<std::uint8_t>(config.fabric.hop_limit); | |
| 428 | ✗ | raw.traffic_class = static_cast<std::uint8_t>(config.fabric.traffic_class); | |
| 429 | ✗ | raw.flow_label = config.fabric.flow_label; | |
| 430 | ✗ | raw.deployment_id = config.deployment_id; | |
| 431 | ✗ | raw.deployment_epoch = config.deployment_epoch; | |
| 432 | ✗ | raw.protocol_version = config.protocol_version; | |
| 433 | ✗ | raw.configuration_digest = config.configuration_digest; | |
| 434 | ✗ | raw.fabric_digest = config.fabric_digest; | |
| 435 | ✗ | raw.max_inline_data = | |
| 436 | ✗ | static_cast<std::uint32_t>(std::max(0, FLAGS_rdma_rc_inline_bytes)); | |
| 437 | ✗ | raw.connect_to_servers = is_client; | |
| 438 | ✗ | raw.connect_to_clients = !is_client; | |
| 439 | ✗ | raw.local_region_bytes = local_region_bytes; | |
| 440 | ✗ | raw.control_plane_host = config.control_plane_host; | |
| 441 | ✗ | raw.control_plane_port = config.control_plane_port; | |
| 442 | ✗ | raw.control_plane_timeout_ms = config.control_plane_timeout_ms; | |
| 443 | ✗ | return raw; | |
| 444 | ✗ | } | |
| 445 | |||
| 446 | } // namespace | ||
| 447 | |||
| 448 | ✗ | RcShardClientTransport::RcShardClientTransport(const RcTransportConfig& config) | |
| 449 | ✗ | : config_(config), server_node_id_(config.shard_id) { | |
| 450 | ✗ | ValidateTransportConfig(config_); | |
| 451 | ✗ | ValidateClientId(config_, config_.client_id); | |
| 452 | ✗ | if (config_.slots_per_qp <= 0) { | |
| 453 | ✗ | throw std::runtime_error("slots_per_qp must be positive"); | |
| 454 | } | ||
| 455 | ✗ | if (server_node_id_ < 0 || server_node_id_ >= config_.num_servers) { | |
| 456 | ✗ | throw std::runtime_error("server shard id out of global node range"); | |
| 457 | } | ||
| 458 | ✗ | lanes_.reserve(static_cast<std::size_t>(config_.qps_per_client_per_shard)); | |
| 459 | ✗ | for (int qp = 0; qp < config_.qps_per_client_per_shard; ++qp) { | |
| 460 | ✗ | Lane lane; | |
| 461 | const int raw_lane = | ||
| 462 | ✗ | RawLaneForLogicalClient(config_, config_.client_id, qp); | |
| 463 | // Each lane owns only its own slot slice; peers keep addressing it with | ||
| 464 | // global offsets, which region_origin_offset maps back into the slice. | ||
| 465 | ✗ | const std::uint64_t lane_origin = ClientShardLaneOffset(config_, raw_lane); | |
| 466 | RawVerbsConfig raw = MakeRawConfig( | ||
| 467 | ✗ | config_, raw_lane, ClientLaneBytes(config_), true, server_node_id_); | |
| 468 | ✗ | raw.region_origin_offset = lane_origin; | |
| 469 | // The whole slice is reserved for slots; nothing is bump-allocated here. | ||
| 470 | ✗ | raw.reserved_region_offset = 0; | |
| 471 | ✗ | raw.reserved_region_bytes = ClientLaneBytes(config_); | |
| 472 | ✗ | lane.verbs = std::make_unique<RawVerbsTransport>(raw); | |
| 473 | ✗ | lane.lane_base = lane.verbs->LocalPointer(GlobalAddress{ | |
| 474 | ✗ | static_cast<std::uint16_t>(config_.node_id), | |
| 475 | ✗ | lane_origin, | |
| 476 | }); | ||
| 477 | ✗ | std::memset(lane.lane_base, 0, ClientLaneBytes(config_)); | |
| 478 | ✗ | lane.submit_completion_pending.assign( | |
| 479 | ✗ | static_cast<std::size_t>(config_.slots_per_qp), 0); | |
| 480 | ✗ | lane.submit_completion_ready.assign( | |
| 481 | ✗ | static_cast<std::size_t>(config_.slots_per_qp), 0); | |
| 482 | ✗ | lane.verbs->PublishAndConnect(); | |
| 483 | ✗ | lanes_.push_back(std::move(lane)); | |
| 484 | ✗ | } | |
| 485 | // Each lane registers only its own slot slice, so the total is | ||
| 486 | // lanes * ClientLaneBytes regardless of the shard count. | ||
| 487 | ✗ | LOG(INFO) << "component=rdma_verbs event=client_region_lanes" | |
| 488 | ✗ | << " shard=" << server_node_id_ | |
| 489 | ✗ | << " client_id=" << config_.client_id << " lanes=" << lanes_.size() | |
| 490 | ✗ | << " lane_bytes=" << ClientLaneBytes(config_) | |
| 491 | ✗ | << " total_bytes=" << lanes_.size() * ClientLaneBytes(config_); | |
| 492 | ✗ | } | |
| 493 | |||
| 494 | ✗ | RcShardClientTransport::~RcShardClientTransport() { | |
| 495 | try { | ||
| 496 | ✗ | for (std::size_t qp = 0; qp < lanes_.size(); ++qp) { | |
| 497 | ✗ | Lane& lane = lanes_[qp]; | |
| 498 | ✗ | if (!lane.verbs) { | |
| 499 | ✗ | continue; | |
| 500 | } | ||
| 501 | ✗ | for (int slot_in_qp = 0; slot_in_qp < config_.slots_per_qp; | |
| 502 | ++slot_in_qp) { | ||
| 503 | ✗ | DrainTrackedPendingWrite( | |
| 504 | lane.verbs.get(), | ||
| 505 | &lane.submit_completion_pending, | ||
| 506 | &lane.submit_completion_ready, | ||
| 507 | TrackedWrKind::kSubmitCommit, | ||
| 508 | slot_in_qp, | ||
| 509 | ✗ | WriteContext( | |
| 510 | ✗ | config_, | |
| 511 | config_.client_id, | ||
| 512 | static_cast<int>(qp), | ||
| 513 | slot_in_qp, | ||
| 514 | 0, | ||
| 515 | ✗ | ServerRequestOffset(config_, config_.client_id, slot_in_qp) + | |
| 516 | ✗ | RequestCommitOffset(config_), | |
| 517 | server_node_id_, | ||
| 518 | "shutdown_submit_commit")); | ||
| 519 | } | ||
| 520 | } | ||
| 521 | ✗ | } catch (...) { | |
| 522 | ✗ | } | |
| 523 | ✗ | } | |
| 524 | |||
| 525 | ✗ | RcShardClientTransport::Lane& RcShardClientTransport::LaneAt(int qp_index) { | |
| 526 | ✗ | if (qp_index < 0 || qp_index >= config_.qps_per_client_per_shard) { | |
| 527 | ✗ | throw std::runtime_error("qp_index out of range"); | |
| 528 | } | ||
| 529 | ✗ | return lanes_.at(static_cast<std::size_t>(qp_index)); | |
| 530 | } | ||
| 531 | |||
| 532 | const RcShardClientTransport::Lane& | ||
| 533 | ✗ | RcShardClientTransport::LaneAt(int qp_index) const { | |
| 534 | ✗ | if (qp_index < 0 || qp_index >= config_.qps_per_client_per_shard) { | |
| 535 | ✗ | throw std::runtime_error("qp_index out of range"); | |
| 536 | } | ||
| 537 | ✗ | return lanes_.at(static_cast<std::size_t>(qp_index)); | |
| 538 | } | ||
| 539 | |||
| 540 | ✗ | RcClientQpView RcShardClientTransport::OpenQp(int qp_index) { | |
| 541 | ✗ | return OpenSlot(qp_index, 0); | |
| 542 | } | ||
| 543 | |||
| 544 | ✗ | RcClientQpView RcShardClientTransport::OpenSlot(int qp_index, int slot_in_qp) { | |
| 545 | ✗ | ValidateSlotInQp(config_, slot_in_qp); | |
| 546 | ✗ | const Lane& lane = LaneAt(qp_index); | |
| 547 | ✗ | auto* slot_base = static_cast<char*>(lane.lane_base) + | |
| 548 | ✗ | ClientSlotOffset(config_, slot_in_qp); | |
| 549 | ✗ | auto* response_slot = static_cast<void*>(slot_base); | |
| 550 | ✗ | auto* response_payload = slot_base; | |
| 551 | auto* status = reinterpret_cast<StatusWord*>( | ||
| 552 | ✗ | response_payload + config_.response_slot_bytes - | |
| 553 | ✗ | Align64(sizeof(StatusWord))); | |
| 554 | |||
| 555 | ✗ | auto* request_slot = slot_base + config_.response_slot_bytes; | |
| 556 | ✗ | auto* descriptor = reinterpret_cast<RequestDescriptor*>(request_slot); | |
| 557 | ✗ | auto* payload = request_slot + Align64(sizeof(RequestDescriptor)); | |
| 558 | auto* commit = reinterpret_cast<CommitWord*>( | ||
| 559 | ✗ | request_slot + config_.request_slot_bytes - Align64(sizeof(CommitWord))); | |
| 560 | const int slot_index = | ||
| 561 | ✗ | GlobalSlotIndex(config_, config_.client_id, qp_index, slot_in_qp); | |
| 562 | |||
| 563 | return RcClientQpView{ | ||
| 564 | qp_index, | ||
| 565 | slot_in_qp, | ||
| 566 | slot_index, | ||
| 567 | request_slot, | ||
| 568 | descriptor, | ||
| 569 | payload, | ||
| 570 | commit, | ||
| 571 | response_slot, | ||
| 572 | response_payload, | ||
| 573 | status, | ||
| 574 | ✗ | }; | |
| 575 | } | ||
| 576 | |||
| 577 | ✗ | void RcShardClientTransport::SubmitRequest( | |
| 578 | const RcClientQpView& view, | ||
| 579 | const RequestDescriptor& descriptor, | ||
| 580 | const void* payload, | ||
| 581 | std::size_t payload_bytes) { | ||
| 582 | ✗ | const bool profile_enabled = FLAGS_rdma_rc_profile_interval_ms > 0; | |
| 583 | ✗ | const std::uint64_t start_ns = profile_enabled ? NowNs() : 0; | |
| 584 | ✗ | ValidateSlotInQp(config_, view.slot_in_qp); | |
| 585 | ✗ | Lane& lane = LaneAt(view.qp_index); | |
| 586 | const std::uint64_t remote_request_offset = | ||
| 587 | ✗ | ServerRequestOffset(config_, config_.client_id, view.slot_in_qp); | |
| 588 | ✗ | auto& counters = TransportProfile(); | |
| 589 | ✗ | DrainTrackedPendingWrite( | |
| 590 | lane.verbs.get(), | ||
| 591 | &lane.submit_completion_pending, | ||
| 592 | &lane.submit_completion_ready, | ||
| 593 | TrackedWrKind::kSubmitCommit, | ||
| 594 | ✗ | view.slot_in_qp, | |
| 595 | ✗ | WriteContext( | |
| 596 | ✗ | config_, | |
| 597 | config_.client_id, | ||
| 598 | ✗ | view.qp_index, | |
| 599 | ✗ | view.slot_in_qp, | |
| 600 | ✗ | descriptor.seq - 1, | |
| 601 | ✗ | remote_request_offset + RequestCommitOffset(config_), | |
| 602 | server_node_id_, | ||
| 603 | "previous_submit_commit"), | ||
| 604 | profile_enabled, | ||
| 605 | &counters.drain_pending_submit_count, | ||
| 606 | &counters.drain_pending_submit_ns); | ||
| 607 | |||
| 608 | ✗ | auto* request_slot = static_cast<char*>(view.request_slot); | |
| 609 | ✗ | auto* local_descriptor = reinterpret_cast<RequestDescriptor*>(request_slot); | |
| 610 | ✗ | auto* local_payload = request_slot + Align64(sizeof(RequestDescriptor)); | |
| 611 | auto* local_commit = reinterpret_cast<CommitWord*>( | ||
| 612 | ✗ | request_slot + RequestCommitOffset(config_)); | |
| 613 | ✗ | *local_descriptor = descriptor; | |
| 614 | ✗ | if (payload_bytes > 0 && payload != local_payload) { | |
| 615 | ✗ | std::memcpy(local_payload, payload, payload_bytes); | |
| 616 | } | ||
| 617 | ✗ | local_commit->seq.store(descriptor.seq, std::memory_order_release); | |
| 618 | ✗ | local_commit->state.store(kRcSlotReady, std::memory_order_release); | |
| 619 | |||
| 620 | ✗ | lane.verbs->Write( | |
| 621 | request_slot, | ||
| 622 | GlobalAddress{ | ||
| 623 | ✗ | static_cast<std::uint16_t>(server_node_id_), | |
| 624 | ✗ | remote_request_offset, | |
| 625 | }, | ||
| 626 | ✗ | Align64(sizeof(RequestDescriptor)) + payload_bytes, | |
| 627 | /*wr_id=*/0, | ||
| 628 | false); | ||
| 629 | ✗ | if (profile_enabled) { | |
| 630 | ✗ | TransportProfile().submit_descriptor_write_count.fetch_add( | |
| 631 | 1, std::memory_order_relaxed); | ||
| 632 | } | ||
| 633 | |||
| 634 | ✗ | lane.verbs->Write( | |
| 635 | local_commit, | ||
| 636 | GlobalAddress{ | ||
| 637 | ✗ | static_cast<std::uint16_t>(server_node_id_), | |
| 638 | ✗ | remote_request_offset + RequestCommitOffset(config_), | |
| 639 | }, | ||
| 640 | sizeof(CommitWord), | ||
| 641 | ✗ | MakeTrackedWrId(TrackedWrKind::kSubmitCommit, view.slot_in_qp), | |
| 642 | true); | ||
| 643 | ✗ | lane.submit_completion_pending[static_cast<std::size_t>(view.slot_in_qp)] = 1; | |
| 644 | ✗ | if (profile_enabled) { | |
| 645 | ✗ | auto& profile = TransportProfile(); | |
| 646 | ✗ | profile.submit_request_count.fetch_add(1, std::memory_order_relaxed); | |
| 647 | ✗ | profile.submit_commit_write_count.fetch_add(1, std::memory_order_relaxed); | |
| 648 | ✗ | profile.submit_request_ns.fetch_add( | |
| 649 | ✗ | NowNs() - start_ns, std::memory_order_relaxed); | |
| 650 | ✗ | MaybeReportTransportProfile(config_, "client"); | |
| 651 | } | ||
| 652 | ✗ | } | |
| 653 | |||
| 654 | ✗ | void RcShardClientTransport::ClearRequestSlot(const RcClientQpView& view) { | |
| 655 | ✗ | auto* commit = view.commit; | |
| 656 | ✗ | commit->state.store(0, std::memory_order_release); | |
| 657 | ✗ | } | |
| 658 | |||
| 659 | ✗ | bool RcShardClientTransport::HarvestLaneCompletion( | |
| 660 | int qp_index, RawVerbsCompletionError* error) { | ||
| 661 | ✗ | Lane& lane = LaneAt(qp_index); | |
| 662 | // Drain up to one poll batch so a busy lane does not delay a completion error | ||
| 663 | // behind a queue of parked successes. | ||
| 664 | ✗ | for (int drained = 0; drained < kRawVerbsPollBatchSize; ++drained) { | |
| 665 | ✗ | RawVerbsCompletion completion; | |
| 666 | ✗ | const RawVerbsPollOutcome outcome = lane.verbs->TryPoll(&completion, error); | |
| 667 | ✗ | if (outcome == RawVerbsPollOutcome::kEmpty) { | |
| 668 | ✗ | return false; | |
| 669 | } | ||
| 670 | ✗ | if (outcome == RawVerbsPollOutcome::kError) { | |
| 671 | ✗ | return true; | |
| 672 | } | ||
| 673 | ✗ | const bool is_recv = completion.opcode == IBV_WC_RECV || | |
| 674 | ✗ | completion.opcode == IBV_WC_RECV_RDMA_WITH_IMM; | |
| 675 | ✗ | const std::uint64_t kind_bits = completion.wr_id >> 32; | |
| 676 | ✗ | const std::uint32_t slot_ordinal = | |
| 677 | ✗ | static_cast<std::uint32_t>(completion.wr_id); | |
| 678 | ✗ | if (!is_recv && | |
| 679 | ✗ | kind_bits == static_cast<std::uint64_t>(TrackedWrKind::kSubmitCommit) && | |
| 680 | ✗ | slot_ordinal < lane.submit_completion_ready.size()) { | |
| 681 | // Park the completion for the next drain of that slot, do not lose it. | ||
| 682 | ✗ | lane.submit_completion_ready[slot_ordinal] = 1; | |
| 683 | ✗ | continue; | |
| 684 | } | ||
| 685 | ✗ | if (!is_recv) { | |
| 686 | ✗ | LOG(WARNING) << "component=rdma_rc_client event=unexpected_completion" | |
| 687 | ✗ | << " shard=" << config_.shard_id << " qp=" << qp_index | |
| 688 | ✗ | << " wr_id=" << completion.wr_id; | |
| 689 | } | ||
| 690 | } | ||
| 691 | ✗ | return false; | |
| 692 | } | ||
| 693 | |||
| 694 | ✗ | bool RcShardClientTransport::RebuildLane(int qp_index) { | |
| 695 | ✗ | if (config_.num_os_clients != 1) { | |
| 696 | // The server multiplexes every OS client QP of this lane onto one CQ, so it | ||
| 697 | // cannot reset/drain it safely. Refuse the rebuild and keep the lane | ||
| 698 | // terminally dirty rather than desynchronizing the two ends. | ||
| 699 | ✗ | VLOG(1) << "component=rdma_rc_client event=lane_rebuild_skipped" | |
| 700 | ✗ | << " shard=" << config_.shard_id << " qp=" << qp_index | |
| 701 | ✗ | << " reason=shared_server_cq"; | |
| 702 | ✗ | return false; | |
| 703 | } | ||
| 704 | ✗ | Lane& lane = LaneAt(qp_index); | |
| 705 | ✗ | const auto server_node = static_cast<std::uint16_t>(server_node_id_); | |
| 706 | // Re-read the peer's current addressing first: if the shard process | ||
| 707 | // restarted, its QP number / rkey / base changed, and re-pointing at the | ||
| 708 | // stale meta would leave the lane just as broken. Its generation is our ack | ||
| 709 | // baseline: the two sides count rebuilds independently, so an ack is "the | ||
| 710 | // peer rebuilt at least once since we took this snapshot", not "the peer | ||
| 711 | // caught up to our counter". | ||
| 712 | const std::uint64_t server_baseline = | ||
| 713 | lane.verbs | ||
| 714 | ✗ | ->RefreshPeerMeta( | |
| 715 | ✗ | server_node, std::max(1, FLAGS_rdma_lane_rebuild_wait_ms)) | |
| 716 | ✗ | .lane_generation; | |
| 717 | // Bring both ends down before any new traffic: reset our QP, then publish a | ||
| 718 | // bumped generation so the server knows to rebuild its side too. | ||
| 719 | ✗ | lane.verbs->RebuildQp(server_node); | |
| 720 | // The rebuild drained the CQ and reset the QP, so no submit completion can | ||
| 721 | // still be parked for this lane; drop the stale bookkeeping. | ||
| 722 | ✗ | std::fill(lane.submit_completion_pending.begin(), | |
| 723 | lane.submit_completion_pending.end(), | ||
| 724 | ✗ | std::uint8_t{0}); | |
| 725 | ✗ | std::fill(lane.submit_completion_ready.begin(), | |
| 726 | lane.submit_completion_ready.end(), | ||
| 727 | ✗ | std::uint8_t{0}); | |
| 728 | // Response status/payload from the dead generation must not look done. | ||
| 729 | ✗ | std::memset(lane.lane_base, 0, ClientLaneBytes(config_)); | |
| 730 | ✗ | lane.generation = lane.verbs->AdvanceLaneGeneration(server_node); | |
| 731 | |||
| 732 | ✗ | const int wait_ms = std::max(0, FLAGS_rdma_lane_rebuild_wait_ms); | |
| 733 | const auto deadline = | ||
| 734 | ✗ | std::chrono::steady_clock::now() + std::chrono::milliseconds(wait_ms); | |
| 735 | while (true) { | ||
| 736 | const auto remaining_ms = | ||
| 737 | ✗ | std::chrono::duration_cast<std::chrono::milliseconds>( | |
| 738 | ✗ | deadline - std::chrono::steady_clock::now()) | |
| 739 | ✗ | .count(); | |
| 740 | ✗ | if (remaining_ms <= 0) { | |
| 741 | ✗ | break; | |
| 742 | } | ||
| 743 | try { | ||
| 744 | // The server republishes once it has rebuilt, so an ack is "the peer | ||
| 745 | // rebuilt after our baseline snapshot". Keep the RPC inside the handshake | ||
| 746 | // budget: the default control-plane deadline is much longer than | ||
| 747 | // --rdma_lane_rebuild_wait_ms. | ||
| 748 | const RawVerbsNodeMeta meta = lane.verbs->RefreshPeerMeta( | ||
| 749 | ✗ | server_node, static_cast<int>(remaining_ms)); | |
| 750 | ✗ | if (meta.lane_generation > server_baseline) { | |
| 751 | ✗ | LOG(WARNING) << "component=rdma_rc_client event=lane_rebuilt" | |
| 752 | ✗ | << " shard=" << config_.shard_id << " qp=" << qp_index | |
| 753 | ✗ | << " generation=" << lane.generation; | |
| 754 | ✗ | return true; | |
| 755 | } | ||
| 756 | ✗ | } catch (const std::exception& error) { | |
| 757 | ✗ | VLOG(1) << "component=rdma_rc_client event=lane_rebuild_ack_pending" | |
| 758 | ✗ | << " shard=" << config_.shard_id << " qp=" << qp_index | |
| 759 | ✗ | << " error=\"" << error.what() << "\""; | |
| 760 | ✗ | } | |
| 761 | ✗ | if (std::chrono::steady_clock::now() >= deadline) { | |
| 762 | ✗ | break; | |
| 763 | } | ||
| 764 | ✗ | std::this_thread::sleep_for(std::chrono::milliseconds(2)); | |
| 765 | ✗ | } | |
| 766 | ✗ | LOG(WARNING) << "component=rdma_rc_client event=lane_rebuild_timeout" | |
| 767 | ✗ | << " shard=" << config_.shard_id << " qp=" << qp_index | |
| 768 | ✗ | << " generation=" << lane.generation << " wait_ms=" << wait_ms; | |
| 769 | ✗ | return false; | |
| 770 | } | ||
| 771 | |||
| 772 | ✗ | std::uint64_t RcShardClientTransport::LaneGeneration(int qp_index) const { | |
| 773 | ✗ | return LaneAt(qp_index).generation; | |
| 774 | } | ||
| 775 | |||
| 776 | ✗ | RcShardServerTransport::RcShardServerTransport(const RcTransportConfig& config) | |
| 777 | ✗ | : config_(config) { | |
| 778 | ✗ | ValidateTransportConfig(config_); | |
| 779 | ✗ | if (config_.node_id < 0 || config_.node_id >= config_.num_servers) { | |
| 780 | ✗ | throw std::runtime_error("server global_id out of range"); | |
| 781 | } | ||
| 782 | ✗ | if (config_.slots_per_qp <= 0) { | |
| 783 | ✗ | throw std::runtime_error("slots_per_qp must be positive"); | |
| 784 | } | ||
| 785 | ✗ | const int raw_lane_count = RawLanesPerOsClient(config_); | |
| 786 | // Lane owns a mutex and an atomic epoch, so it is neither copyable nor | ||
| 787 | // movable; build the vector directly instead of growing it in place. | ||
| 788 | ✗ | lanes_ = std::vector<Lane>(static_cast<std::size_t>(raw_lane_count)); | |
| 789 | ✗ | for (int raw_lane = 0; raw_lane < raw_lane_count; ++raw_lane) { | |
| 790 | ✗ | Lane& lane = lanes_[static_cast<std::size_t>(raw_lane)]; | |
| 791 | ✗ | const int response_slots = config_.num_clients * config_.slots_per_qp; | |
| 792 | const std::size_t local_bytes = | ||
| 793 | ✗ | ServerLaneBytes(config_) + | |
| 794 | ✗ | static_cast<std::size_t>(response_slots) * config_.response_slot_bytes; | |
| 795 | ✗ | lane.verbs = std::make_unique<RawVerbsTransport>( | |
| 796 | ✗ | MakeRawConfig(config_, raw_lane, local_bytes, false, -1)); | |
| 797 | ✗ | lane.request_slots = | |
| 798 | ✗ | lane.verbs->AllocateRegistered(ServerLaneBytes(config_)); | |
| 799 | ✗ | std::memset(lane.request_slots, 0, ServerLaneBytes(config_)); | |
| 800 | ✗ | lane.response_staging.reserve(static_cast<std::size_t>(response_slots)); | |
| 801 | ✗ | lane.response_completion_pending.assign( | |
| 802 | ✗ | static_cast<std::size_t>(response_slots), 0); | |
| 803 | ✗ | lane.response_completion_ready.assign( | |
| 804 | ✗ | static_cast<std::size_t>(response_slots), 0); | |
| 805 | ✗ | lane.observed_client_generation.assign( | |
| 806 | ✗ | static_cast<std::size_t>(config_.num_clients), 0); | |
| 807 | ✗ | for (int slot = 0; slot < response_slots; ++slot) { | |
| 808 | void* response_slot = | ||
| 809 | ✗ | lane.verbs->AllocateRegistered(config_.response_slot_bytes); | |
| 810 | ✗ | std::memset(response_slot, 0, config_.response_slot_bytes); | |
| 811 | ✗ | lane.response_staging.push_back(response_slot); | |
| 812 | } | ||
| 813 | ✗ | lane.verbs->PublishAndConnect(); | |
| 814 | } | ||
| 815 | ✗ | } | |
| 816 | |||
| 817 | ✗ | RcShardServerTransport::~RcShardServerTransport() { | |
| 818 | try { | ||
| 819 | ✗ | const int logical_clients_per_process = LogicalClientsPerProcess(config_); | |
| 820 | ✗ | for (std::size_t raw_lane_index = 0; raw_lane_index < lanes_.size(); | |
| 821 | ++raw_lane_index) { | ||
| 822 | ✗ | Lane& lane = lanes_[raw_lane_index]; | |
| 823 | ✗ | if (!lane.verbs) { | |
| 824 | ✗ | continue; | |
| 825 | } | ||
| 826 | ✗ | const int local_logical_client = static_cast<int>( | |
| 827 | ✗ | raw_lane_index / | |
| 828 | ✗ | static_cast<std::size_t>(config_.qps_per_client_per_shard)); | |
| 829 | ✗ | const int qp_index = static_cast<int>( | |
| 830 | ✗ | raw_lane_index % | |
| 831 | ✗ | static_cast<std::size_t>(config_.qps_per_client_per_shard)); | |
| 832 | ✗ | for (int os_client = 0; os_client < config_.num_os_clients; ++os_client) { | |
| 833 | ✗ | const int client = | |
| 834 | ✗ | os_client * logical_clients_per_process + local_logical_client; | |
| 835 | ✗ | for (int slot_in_qp = 0; slot_in_qp < config_.slots_per_qp; | |
| 836 | ++slot_in_qp) { | ||
| 837 | const int raw_lane = | ||
| 838 | ✗ | RawLaneForLogicalClient(config_, client, qp_index); | |
| 839 | const int client_node_id = | ||
| 840 | ✗ | config_.num_servers + | |
| 841 | ✗ | OsClientIndexForLogicalClient(config_, client); | |
| 842 | const int response_slot = | ||
| 843 | ✗ | ResponseSlotOrdinal(config_, client, slot_in_qp); | |
| 844 | ✗ | DrainTrackedPendingWrite( | |
| 845 | lane.verbs.get(), | ||
| 846 | &lane.response_completion_pending, | ||
| 847 | &lane.response_completion_ready, | ||
| 848 | TrackedWrKind::kResponseStatus, | ||
| 849 | response_slot, | ||
| 850 | ✗ | WriteContext( | |
| 851 | ✗ | config_, | |
| 852 | client, | ||
| 853 | qp_index, | ||
| 854 | slot_in_qp, | ||
| 855 | 0, | ||
| 856 | ✗ | ClientResponseOffsetForRawLane( | |
| 857 | ✗ | config_, raw_lane, slot_in_qp) + | |
| 858 | ✗ | ResponseStatusOffset(config_), | |
| 859 | client_node_id, | ||
| 860 | "shutdown_response_status")); | ||
| 861 | } | ||
| 862 | } | ||
| 863 | } | ||
| 864 | ✗ | } catch (...) { | |
| 865 | ✗ | } | |
| 866 | ✗ | } | |
| 867 | |||
| 868 | RcShardServerTransport::Lane& | ||
| 869 | ✗ | RcShardServerTransport::LaneAt(int client_id, int qp_index) { | |
| 870 | ✗ | ValidateClientId(config_, client_id); | |
| 871 | ✗ | if (qp_index < 0 || qp_index >= config_.qps_per_client_per_shard) { | |
| 872 | ✗ | throw std::runtime_error("qp_index out of range"); | |
| 873 | } | ||
| 874 | ✗ | return lanes_.at(static_cast<std::size_t>( | |
| 875 | ✗ | RawLaneForLogicalClient(config_, client_id, qp_index))); | |
| 876 | } | ||
| 877 | |||
| 878 | const RcShardServerTransport::Lane& | ||
| 879 | ✗ | RcShardServerTransport::LaneAt(int client_id, int qp_index) const { | |
| 880 | ✗ | ValidateClientId(config_, client_id); | |
| 881 | ✗ | if (qp_index < 0 || qp_index >= config_.qps_per_client_per_shard) { | |
| 882 | ✗ | throw std::runtime_error("qp_index out of range"); | |
| 883 | } | ||
| 884 | ✗ | return lanes_.at(static_cast<std::size_t>( | |
| 885 | ✗ | RawLaneForLogicalClient(config_, client_id, qp_index))); | |
| 886 | } | ||
| 887 | |||
| 888 | ✗ | int RcShardServerTransport::TotalSlots() const { | |
| 889 | ✗ | return static_cast<int>(static_cast<std::size_t>(config_.num_clients) * | |
| 890 | ✗ | TotalClientSlotsPerShard(config_)); | |
| 891 | } | ||
| 892 | |||
| 893 | ✗ | void RcShardServerTransport::RegisterLocalMemoryRegion( | |
| 894 | void* base, std::size_t bytes) { | ||
| 895 | ✗ | for (auto& lane : lanes_) { | |
| 896 | ✗ | if (lane.verbs) { | |
| 897 | ✗ | lane.verbs->RegisterMemoryRegion(base, bytes); | |
| 898 | } | ||
| 899 | } | ||
| 900 | ✗ | } | |
| 901 | |||
| 902 | ✗ | int RcShardServerTransport::SlotIndex( | |
| 903 | int client_id, int qp_index, int slot_in_qp) const { | ||
| 904 | ✗ | ValidateClientId(config_, client_id); | |
| 905 | ✗ | ValidateSlotInQp(config_, slot_in_qp); | |
| 906 | ✗ | if (qp_index < 0 || qp_index >= config_.qps_per_client_per_shard) { | |
| 907 | ✗ | throw std::runtime_error("qp_index out of range"); | |
| 908 | } | ||
| 909 | ✗ | return GlobalSlotIndex(config_, client_id, qp_index, slot_in_qp); | |
| 910 | } | ||
| 911 | |||
| 912 | ✗ | void RcShardServerTransport::DecodeSlotIndex( | |
| 913 | int slot_index, int* client_id, int* qp_index, int* slot_in_qp) const { | ||
| 914 | ✗ | if (slot_index < 0 || slot_index >= TotalSlots()) { | |
| 915 | ✗ | throw std::runtime_error("slot_index out of range"); | |
| 916 | } | ||
| 917 | ✗ | DecodeGlobalSlotIndex(config_, slot_index, client_id, qp_index, slot_in_qp); | |
| 918 | ✗ | } | |
| 919 | |||
| 920 | ✗ | void* RcShardServerTransport::RequestSlot(int slot_index) const { | |
| 921 | ✗ | int client_id = -1; | |
| 922 | ✗ | int qp_index = -1; | |
| 923 | ✗ | int slot_in_qp = -1; | |
| 924 | ✗ | DecodeSlotIndex(slot_index, &client_id, &qp_index, &slot_in_qp); | |
| 925 | ✗ | const Lane& lane = LaneAt(client_id, qp_index); | |
| 926 | ✗ | return static_cast<char*>(lane.request_slots) + | |
| 927 | ✗ | ServerRequestOffset(config_, client_id, slot_in_qp); | |
| 928 | } | ||
| 929 | |||
| 930 | RequestDescriptor* | ||
| 931 | ✗ | RcShardServerTransport::RequestDescriptorAt(int slot_index) const { | |
| 932 | ✗ | return reinterpret_cast<RequestDescriptor*>(RequestSlot(slot_index)); | |
| 933 | } | ||
| 934 | |||
| 935 | ✗ | char* RcShardServerTransport::RequestPayloadAt(int slot_index) const { | |
| 936 | ✗ | return static_cast<char*>(RequestSlot(slot_index)) + | |
| 937 | ✗ | Align64(sizeof(RequestDescriptor)); | |
| 938 | } | ||
| 939 | |||
| 940 | ✗ | CommitWord* RcShardServerTransport::RequestCommitAt(int slot_index) const { | |
| 941 | return reinterpret_cast<CommitWord*>( | ||
| 942 | ✗ | static_cast<char*>(RequestSlot(slot_index)) + | |
| 943 | ✗ | RequestCommitOffset(config_)); | |
| 944 | } | ||
| 945 | |||
| 946 | ✗ | RcShardServerTransport::ResponseView RcShardServerTransport::OpenClientResponse( | |
| 947 | int client_id, int qp_index, int slot_in_qp) { | ||
| 948 | ✗ | ValidateClientId(config_, client_id); | |
| 949 | ✗ | ValidateSlotInQp(config_, slot_in_qp); | |
| 950 | ✗ | Lane& lane = LaneAt(client_id, qp_index); | |
| 951 | auto* slot = | ||
| 952 | ✗ | static_cast<char*>(lane.response_staging.at(static_cast<std::size_t>( | |
| 953 | ✗ | ResponseSlotOrdinal(config_, client_id, slot_in_qp)))); | |
| 954 | ✗ | auto* payload = static_cast<char*>(slot); | |
| 955 | auto* status = | ||
| 956 | ✗ | reinterpret_cast<StatusWord*>(payload + ResponseStatusOffset(config_)); | |
| 957 | ✗ | return ResponseView{slot, payload, status}; | |
| 958 | } | ||
| 959 | |||
| 960 | ✗ | void RcShardServerTransport::CompleteResponse( | |
| 961 | int client_id, | ||
| 962 | int qp_index, | ||
| 963 | int slot_in_qp, | ||
| 964 | const ResponseView& response, | ||
| 965 | std::uint64_t seq) { | ||
| 966 | ✗ | const bool profile_enabled = FLAGS_rdma_rc_profile_interval_ms > 0; | |
| 967 | ✗ | const std::uint64_t start_ns = profile_enabled ? NowNs() : 0; | |
| 968 | ✗ | ValidateClientId(config_, client_id); | |
| 969 | ✗ | ValidateSlotInQp(config_, slot_in_qp); | |
| 970 | ✗ | Lane& lane = LaneAt(client_id, qp_index); | |
| 971 | ✗ | const int response_slot = ResponseSlotOrdinal(config_, client_id, slot_in_qp); | |
| 972 | ✗ | const int raw_lane = RawLaneForLogicalClient(config_, client_id, qp_index); | |
| 973 | const int client_node_id = | ||
| 974 | ✗ | config_.num_servers + OsClientIndexForLogicalClient(config_, client_id); | |
| 975 | ✗ | auto& counters = TransportProfile(); | |
| 976 | ✗ | DrainTrackedPendingWrite( | |
| 977 | lane.verbs.get(), | ||
| 978 | &lane.response_completion_pending, | ||
| 979 | &lane.response_completion_ready, | ||
| 980 | TrackedWrKind::kResponseStatus, | ||
| 981 | response_slot, | ||
| 982 | ✗ | WriteContext( | |
| 983 | ✗ | config_, | |
| 984 | client_id, | ||
| 985 | qp_index, | ||
| 986 | slot_in_qp, | ||
| 987 | seq - 1, | ||
| 988 | ✗ | ClientResponseOffsetForRawLane(config_, raw_lane, slot_in_qp) + | |
| 989 | ✗ | ResponseStatusOffset(config_), | |
| 990 | client_node_id, | ||
| 991 | "previous_response_status"), | ||
| 992 | profile_enabled, | ||
| 993 | &counters.drain_pending_response_count, | ||
| 994 | &counters.drain_pending_response_ns); | ||
| 995 | ✗ | response.status->seq.store(seq, std::memory_order_release); | |
| 996 | ✗ | response.status->state.store(kRcSlotDone, std::memory_order_release); | |
| 997 | |||
| 998 | ✗ | if (response.status->response_bytes > 0) { | |
| 999 | const std::uint64_t response_payload_offset = | ||
| 1000 | ✗ | ClientResponseOffsetForRawLane(config_, raw_lane, slot_in_qp); | |
| 1001 | ✗ | lane.verbs->Write( | |
| 1002 | ✗ | response.payload, | |
| 1003 | GlobalAddress{ | ||
| 1004 | static_cast<std::uint16_t>(client_node_id), | ||
| 1005 | ✗ | response_payload_offset, | |
| 1006 | }, | ||
| 1007 | ✗ | response.status->response_bytes, | |
| 1008 | /*wr_id=*/0, | ||
| 1009 | false); | ||
| 1010 | ✗ | if (profile_enabled) { | |
| 1011 | ✗ | auto& profile = TransportProfile(); | |
| 1012 | ✗ | profile.response_payload_write_count.fetch_add( | |
| 1013 | 1, std::memory_order_relaxed); | ||
| 1014 | ✗ | profile.response_payload_bytes.fetch_add( | |
| 1015 | ✗ | response.status->response_bytes, std::memory_order_relaxed); | |
| 1016 | } | ||
| 1017 | } | ||
| 1018 | |||
| 1019 | const std::uint64_t response_status_offset = | ||
| 1020 | ✗ | ClientResponseOffsetForRawLane(config_, raw_lane, slot_in_qp) + | |
| 1021 | ✗ | ResponseStatusOffset(config_); | |
| 1022 | ✗ | lane.verbs->Write( | |
| 1023 | ✗ | response.status, | |
| 1024 | GlobalAddress{ | ||
| 1025 | static_cast<std::uint16_t>(client_node_id), | ||
| 1026 | ✗ | response_status_offset, | |
| 1027 | }, | ||
| 1028 | sizeof(StatusWord), | ||
| 1029 | MakeTrackedWrId(TrackedWrKind::kResponseStatus, response_slot), | ||
| 1030 | true); | ||
| 1031 | ✗ | lane.response_completion_pending[static_cast<std::size_t>(response_slot)] = 1; | |
| 1032 | ✗ | if (profile_enabled) { | |
| 1033 | ✗ | auto& profile = TransportProfile(); | |
| 1034 | ✗ | profile.complete_response_count.fetch_add(1, std::memory_order_relaxed); | |
| 1035 | ✗ | profile.response_status_write_count.fetch_add(1, std::memory_order_relaxed); | |
| 1036 | ✗ | profile.complete_response_ns.fetch_add( | |
| 1037 | ✗ | NowNs() - start_ns, std::memory_order_relaxed); | |
| 1038 | ✗ | MaybeReportTransportProfile(config_, "server"); | |
| 1039 | } | ||
| 1040 | ✗ | } | |
| 1041 | |||
| 1042 | std::uint64_t | ||
| 1043 | ✗ | RcShardServerTransport::ResponseEpoch(int client_id, int qp_index) const { | |
| 1044 | ✗ | return LaneAt(client_id, qp_index) | |
| 1045 | ✗ | .response_epoch.load(std::memory_order_acquire); | |
| 1046 | } | ||
| 1047 | |||
| 1048 | ✗ | void RcShardServerTransport::WriteResponsePayloadSg( | |
| 1049 | int client_id, | ||
| 1050 | int qp_index, | ||
| 1051 | int slot_in_qp, | ||
| 1052 | std::uint64_t expected_epoch, | ||
| 1053 | base::ConstArray<RawVerbsSge> sges, | ||
| 1054 | std::uint64_t response_offset, | ||
| 1055 | std::uint64_t bytes) { | ||
| 1056 | ✗ | ValidateClientId(config_, client_id); | |
| 1057 | ✗ | ValidateSlotInQp(config_, slot_in_qp); | |
| 1058 | ✗ | if (sges.Size() == 0 || bytes == 0) { | |
| 1059 | ✗ | return; | |
| 1060 | } | ||
| 1061 | ✗ | Lane& lane = LaneAt(client_id, qp_index); | |
| 1062 | ✗ | const int raw_lane = RawLaneForLogicalClient(config_, client_id, qp_index); | |
| 1063 | const int client_node_id = | ||
| 1064 | ✗ | config_.num_servers + OsClientIndexForLogicalClient(config_, client_id); | |
| 1065 | { | ||
| 1066 | ✗ | const std::lock_guard<std::mutex> guard(lane.rebuild_mu); | |
| 1067 | ✗ | if (lane.response_epoch.load(std::memory_order_acquire) != expected_epoch) { | |
| 1068 | ✗ | response_rejected_epoch_count_.fetch_add(1, std::memory_order_relaxed); | |
| 1069 | ✗ | throw RcLaneRebuiltDuringResponse( | |
| 1070 | ✗ | "lane rebuilt while the response payload was in flight: shard=" + | |
| 1071 | ✗ | std::to_string(config_.shard_id) + " client_id=" + | |
| 1072 | ✗ | std::to_string(client_id) + " qp=" + std::to_string(qp_index)); | |
| 1073 | } | ||
| 1074 | ✗ | lane.verbs->WriteSg( | |
| 1075 | sges, | ||
| 1076 | GlobalAddress{ | ||
| 1077 | static_cast<std::uint16_t>(client_node_id), | ||
| 1078 | ✗ | ClientResponseOffsetForRawLane(config_, raw_lane, slot_in_qp) + | |
| 1079 | response_offset, | ||
| 1080 | }, | ||
| 1081 | /*wr_id=*/0, | ||
| 1082 | false); | ||
| 1083 | ✗ | } | |
| 1084 | ✗ | if (FLAGS_rdma_rc_profile_interval_ms > 0) { | |
| 1085 | ✗ | auto& profile = TransportProfile(); | |
| 1086 | ✗ | profile.response_payload_sg_write_count.fetch_add( | |
| 1087 | 1, std::memory_order_relaxed); | ||
| 1088 | ✗ | profile.response_payload_sg_wr_count.fetch_add( | |
| 1089 | 1, std::memory_order_relaxed); | ||
| 1090 | ✗ | profile.response_payload_bytes.fetch_add(bytes, std::memory_order_relaxed); | |
| 1091 | } | ||
| 1092 | } | ||
| 1093 | |||
| 1094 | ✗ | void RcShardServerTransport::CompleteResponseStatusOnly( | |
| 1095 | int client_id, | ||
| 1096 | int qp_index, | ||
| 1097 | int slot_in_qp, | ||
| 1098 | const ResponseView& response, | ||
| 1099 | std::uint64_t seq) { | ||
| 1100 | ✗ | const bool profile_enabled = FLAGS_rdma_rc_profile_interval_ms > 0; | |
| 1101 | ✗ | const std::uint64_t start_ns = profile_enabled ? NowNs() : 0; | |
| 1102 | ✗ | ValidateClientId(config_, client_id); | |
| 1103 | ✗ | ValidateSlotInQp(config_, slot_in_qp); | |
| 1104 | ✗ | Lane& lane = LaneAt(client_id, qp_index); | |
| 1105 | ✗ | const int response_slot = ResponseSlotOrdinal(config_, client_id, slot_in_qp); | |
| 1106 | ✗ | const int raw_lane = RawLaneForLogicalClient(config_, client_id, qp_index); | |
| 1107 | const int client_node_id = | ||
| 1108 | ✗ | config_.num_servers + OsClientIndexForLogicalClient(config_, client_id); | |
| 1109 | ✗ | auto& counters = TransportProfile(); | |
| 1110 | ✗ | DrainTrackedPendingWrite( | |
| 1111 | lane.verbs.get(), | ||
| 1112 | &lane.response_completion_pending, | ||
| 1113 | &lane.response_completion_ready, | ||
| 1114 | TrackedWrKind::kResponseStatus, | ||
| 1115 | response_slot, | ||
| 1116 | ✗ | WriteContext( | |
| 1117 | ✗ | config_, | |
| 1118 | client_id, | ||
| 1119 | qp_index, | ||
| 1120 | slot_in_qp, | ||
| 1121 | seq - 1, | ||
| 1122 | ✗ | ClientResponseOffsetForRawLane(config_, raw_lane, slot_in_qp) + | |
| 1123 | ✗ | ResponseStatusOffset(config_), | |
| 1124 | client_node_id, | ||
| 1125 | "previous_response_status"), | ||
| 1126 | profile_enabled, | ||
| 1127 | &counters.drain_pending_response_count, | ||
| 1128 | &counters.drain_pending_response_ns); | ||
| 1129 | ✗ | response.status->seq.store(seq, std::memory_order_release); | |
| 1130 | ✗ | response.status->state.store(kRcSlotDone, std::memory_order_release); | |
| 1131 | |||
| 1132 | const std::uint64_t response_status_offset = | ||
| 1133 | ✗ | ClientResponseOffsetForRawLane(config_, raw_lane, slot_in_qp) + | |
| 1134 | ✗ | ResponseStatusOffset(config_); | |
| 1135 | ✗ | lane.verbs->Write( | |
| 1136 | ✗ | response.status, | |
| 1137 | GlobalAddress{ | ||
| 1138 | static_cast<std::uint16_t>(client_node_id), | ||
| 1139 | ✗ | response_status_offset, | |
| 1140 | }, | ||
| 1141 | sizeof(StatusWord), | ||
| 1142 | MakeTrackedWrId(TrackedWrKind::kResponseStatus, response_slot), | ||
| 1143 | true); | ||
| 1144 | ✗ | lane.response_completion_pending[static_cast<std::size_t>(response_slot)] = 1; | |
| 1145 | ✗ | if (profile_enabled) { | |
| 1146 | ✗ | auto& profile = TransportProfile(); | |
| 1147 | ✗ | profile.complete_response_count.fetch_add(1, std::memory_order_relaxed); | |
| 1148 | ✗ | profile.response_status_write_count.fetch_add(1, std::memory_order_relaxed); | |
| 1149 | ✗ | profile.complete_response_ns.fetch_add( | |
| 1150 | ✗ | NowNs() - start_ns, std::memory_order_relaxed); | |
| 1151 | ✗ | MaybeReportTransportProfile(config_, "server"); | |
| 1152 | } | ||
| 1153 | ✗ | } | |
| 1154 | |||
| 1155 | ✗ | bool RcShardServerTransport::RebuildClientLaneIfStale( | |
| 1156 | int client_id, int qp_index) { | ||
| 1157 | ✗ | if (config_.num_os_clients != 1) { | |
| 1158 | // One lane CQ multiplexes every OS client QP, so a reset here cannot be | ||
| 1159 | // fenced by draining the CQ (it would discard the other clients' | ||
| 1160 | // completions). Keep the lane as-is; the client refuses the same case. | ||
| 1161 | ✗ | return false; | |
| 1162 | } | ||
| 1163 | ✗ | Lane& lane = LaneAt(client_id, qp_index); | |
| 1164 | const int client_node_id = | ||
| 1165 | ✗ | config_.num_servers + OsClientIndexForLogicalClient(config_, client_id); | |
| 1166 | ✗ | RawVerbsNodeMeta meta; | |
| 1167 | try { | ||
| 1168 | // Don't inherit the 30s control-plane deadline: this runs on a data-plane | ||
| 1169 | // polling thread, and a sick control plane must not stall request service. | ||
| 1170 | ✗ | meta = lane.verbs->RefreshPeerMeta( | |
| 1171 | static_cast<std::uint16_t>(client_node_id), | ||
| 1172 | ✗ | std::max(1, FLAGS_rdma_lane_rebuild_poll_timeout_ms)); | |
| 1173 | ✗ | } catch (const std::exception&) { | |
| 1174 | ✗ | return false; | |
| 1175 | ✗ | } | |
| 1176 | std::uint64_t& observed = | ||
| 1177 | ✗ | lane.observed_client_generation[static_cast<std::size_t>(client_id)]; | |
| 1178 | ✗ | if (meta.lane_generation == observed) { | |
| 1179 | ✗ | return false; | |
| 1180 | } | ||
| 1181 | // The CQ is exclusive to this QP (checked above), so the reset's flush | ||
| 1182 | // completions can be drained. The response flags are cleared before any new | ||
| 1183 | // response is posted, so a stale completion cannot be matched to a fresh one. | ||
| 1184 | // | ||
| 1185 | // Payload offload posts from the get-payload worker threads are fenced by | ||
| 1186 | // rebuild_mu: a post either finishes before the reset (and is flushed with | ||
| 1187 | // the rest of the old generation) or observes the bumped response_epoch and | ||
| 1188 | // refuses. Without the fence a worker can land payload rows in the client's | ||
| 1189 | // new-generation slot while the status word already reports kOk. | ||
| 1190 | { | ||
| 1191 | ✗ | const std::lock_guard<std::mutex> guard(lane.rebuild_mu); | |
| 1192 | ✗ | lane.response_epoch.fetch_add(1, std::memory_order_acq_rel); | |
| 1193 | ✗ | lane.verbs->RebuildQp(static_cast<std::uint16_t>(client_node_id)); | |
| 1194 | ✗ | for (int slot_in_qp = 0; slot_in_qp < config_.slots_per_qp; ++slot_in_qp) { | |
| 1195 | const std::size_t ordinal = static_cast<std::size_t>( | ||
| 1196 | ✗ | ResponseSlotOrdinal(config_, client_id, slot_in_qp)); | |
| 1197 | ✗ | lane.response_completion_pending[ordinal] = 0; | |
| 1198 | ✗ | lane.response_completion_ready[ordinal] = 0; | |
| 1199 | } | ||
| 1200 | ✗ | } | |
| 1201 | ✗ | observed = meta.lane_generation; | |
| 1202 | ✗ | const std::uint16_t client_node = static_cast<std::uint16_t>(client_node_id); | |
| 1203 | const std::uint64_t next_generation = | ||
| 1204 | ✗ | lane.verbs->AdvanceLaneGeneration(client_node); | |
| 1205 | ✗ | LOG(WARNING) << "component=rdma_rc_server event=lane_rebuilt" | |
| 1206 | ✗ | << " shard=" << config_.shard_id << " client_id=" << client_id | |
| 1207 | ✗ | << " qp=" << qp_index << " generation=" << next_generation; | |
| 1208 | ✗ | return true; | |
| 1209 | ✗ | } | |
| 1210 | |||
| 1211 | ✗ | int RcShardServerTransport::PollClientRebuildRequests(int qp_index) { | |
| 1212 | ✗ | if (qp_index < 0 || qp_index >= config_.qps_per_client_per_shard) { | |
| 1213 | ✗ | throw std::runtime_error("qp_index out of range"); | |
| 1214 | } | ||
| 1215 | ✗ | const int clients_per_process = LogicalClientsPerProcess(config_); | |
| 1216 | ✗ | int rebuilt = 0; | |
| 1217 | ✗ | for (int os_client = 0; os_client < config_.num_os_clients; ++os_client) { | |
| 1218 | ✗ | for (int local = 0; local < clients_per_process; ++local) { | |
| 1219 | ✗ | const int client_id = os_client * clients_per_process + local; | |
| 1220 | ✗ | if (RebuildClientLaneIfStale(client_id, qp_index)) { | |
| 1221 | ✗ | ++rebuilt; | |
| 1222 | } | ||
| 1223 | } | ||
| 1224 | } | ||
| 1225 | ✗ | return rebuilt; | |
| 1226 | } | ||
| 1227 | |||
| 1228 | } // namespace petps | ||
| 1229 |