diff --git a/CHANGELOG.md b/CHANGELOG.md index 21fde014..a9114c2f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 per-stream response field-count and name/value-byte limits across final headers and trailers, resetting an offending stream before the field is copied (#998). +- **Non-blocking RDMA endpoint teardown**: `rdma_ibverbs::endpoint` destruction + no longer sleeps on scheduler workers, and the new awaitable `shutdown()` joins + the CQ pump before releasing verbs resources. Provider teardown failures are + reported without discarding ownership, allowing a serialized retry; checked + QP destruction no longer loses provider failures through `rdma_destroy_qp()`. + Shutdown is terminal for data-path use and requires the pump scheduler to + remain live until completion. Fatal undrainable eager operations can call + `abandon_outstanding()` while their state and buffers remain alive; only a + `true` result confirms QP destruction before the remaining verbs resources + are intentionally relinquished. + Post-shutdown calls to `conn()`, `cas()`, `fetch_add()`, and `dispatcher_ref()` + now throw `std::logic_error` instead of retaining their former `noexcept` + specification (#975). - **Bounded raw WebSocket parsing by default**: `websocket::frame_parser` now applies the same 16 MiB aggregate message limit as the high-level client and server configurations. Direct parser users must explicitly set a limit of diff --git a/examples/rdma_gpu_bw.cpp b/examples/rdma_gpu_bw.cpp index d93b3879..c4f4a49f 100644 --- a/examples/rdma_gpu_bw.cpp +++ b/examples/rdma_gpu_bw.cpp @@ -192,38 +192,42 @@ coro::task async_main(int argc, char* argv[]) { ep.start_cq_pump(*sched); std::printf("Client connected.\n"); - // Allocate GPU buffer and register - elio::rdma_cuda::gpu_memory_region gpu_mr{ - ep.pd(), cfg.size, - IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE | - IBV_ACCESS_REMOTE_READ}; - - // Exchange remote info via SEND/RECV - auto remote = gpu_mr.remote(); - xchg_msg msg_out{remote.addr, remote.length, remote.rkey}; - - auto send_mr = ep.register_buffer(&msg_out, sizeof(msg_out), - IBV_ACCESS_LOCAL_WRITE); - auto send_wc = co_await ep.conn().send(send_mr.view()); - if (!send_wc.ok()) { - std::fprintf(stderr, "server send failed status=%d\n", - static_cast(send_wc.status)); - co_return 1; + int result = 0; + { + // This scope owns every MR; it ends before endpoint shutdown. + elio::rdma_cuda::gpu_memory_region gpu_mr{ + ep.pd(), cfg.size, + IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE | + IBV_ACCESS_REMOTE_READ}; + + auto remote = gpu_mr.remote(); + xchg_msg msg_out{remote.addr, remote.length, remote.rkey}; + + auto send_mr = ep.register_buffer(&msg_out, sizeof(msg_out), + IBV_ACCESS_LOCAL_WRITE); + auto send_wc = co_await ep.conn().send(send_mr.view()); + if (!send_wc.ok()) { + std::fprintf(stderr, "server send failed status=%d\n", + static_cast(send_wc.status)); + result = 1; + } else { + xchg_msg msg_in{}; + auto recv_mr = ep.register_buffer(&msg_in, sizeof(msg_in), + IBV_ACCESS_LOCAL_WRITE); + auto recv_wc = co_await ep.conn().recv(recv_mr.view()); + if (!recv_wc.ok()) { + std::fprintf(stderr, "server recv failed status=%d\n", + static_cast(recv_wc.status)); + result = 1; + } + } } - // Wait for client "done" signal - xchg_msg msg_in{}; - auto recv_mr = ep.register_buffer(&msg_in, sizeof(msg_in), - IBV_ACCESS_LOCAL_WRITE); - auto recv_wc = co_await ep.conn().recv(recv_mr.view()); - if (!recv_wc.ok()) { - std::fprintf(stderr, "server recv failed status=%d\n", - static_cast(recv_wc.status)); - co_return 1; + co_await ep.shutdown(); + if (result == 0) { + std::printf("Benchmark complete.\n"); } - - std::printf("Benchmark complete.\n"); - co_return 0; + co_return result; } // Client @@ -235,54 +239,60 @@ coro::task async_main(int argc, char* argv[]) { cm_ch, reinterpret_cast(&addr), sizeof(addr), ep_cfg); ep.start_cq_pump(*sched); - // Receive server's remote buffer info - xchg_msg srv_info{}; - auto recv_mr = ep.register_buffer(&srv_info, sizeof(srv_info), - IBV_ACCESS_LOCAL_WRITE); - auto recv_wc = co_await ep.conn().recv(recv_mr.view()); - if (!recv_wc.ok()) { - std::fprintf(stderr, "client recv failed status=%d\n", - static_cast(recv_wc.status)); - co_return 1; - } - - rdma::remote_buffer server_buf{srv_info.addr, srv_info.length, srv_info.rkey}; - std::printf("Connected. Server buffer: %zu bytes\n", cfg.size); - std::printf("Running RDMA WRITE bandwidth test (%d iters, %zu B):\n\n", + int result = 0; + { + // Receive server's remote buffer info. All MRs in this scope are + // released before endpoint shutdown below. + xchg_msg srv_info{}; + auto recv_mr = ep.register_buffer(&srv_info, sizeof(srv_info), + IBV_ACCESS_LOCAL_WRITE); + auto recv_wc = co_await ep.conn().recv(recv_mr.view()); + if (!recv_wc.ok()) { + std::fprintf(stderr, "client recv failed status=%d\n", + static_cast(recv_wc.status)); + result = 1; + } else { + rdma::remote_buffer server_buf{ + srv_info.addr, srv_info.length, srv_info.rkey}; + std::printf("Connected. Server buffer: %zu bytes\n", cfg.size); + std::printf( + "Running RDMA WRITE bandwidth test (%d iters, %zu B):\n\n", cfg.iters, cfg.size); - // GPU path - if (cfg.buf_mode == "gpu" || cfg.buf_mode == "both") { - elio::rdma_cuda::gpu_memory_region gpu_mr{ - ep.pd(), cfg.size, IBV_ACCESS_LOCAL_WRITE}; - auto r = co_await run_write_bw(ep, gpu_mr.view(), server_buf, - DEFAULT_WARMUP, cfg.iters); - print_result("GPU", r); - } + if (cfg.buf_mode == "gpu" || cfg.buf_mode == "both") { + elio::rdma_cuda::gpu_memory_region gpu_mr{ + ep.pd(), cfg.size, IBV_ACCESS_LOCAL_WRITE}; + auto r = co_await run_write_bw( + ep, gpu_mr.view(), server_buf, DEFAULT_WARMUP, cfg.iters); + print_result("GPU", r); + } - // CPU path - if (cfg.buf_mode == "cpu" || cfg.buf_mode == "both") { - std::vector cpu_buf(cfg.size, 'A'); - auto cpu_mr = ep.register_buffer(cpu_buf.data(), cfg.size, - IBV_ACCESS_LOCAL_WRITE); - auto r = co_await run_write_bw(ep, cpu_mr.view(), server_buf, - DEFAULT_WARMUP, cfg.iters); - print_result("CPU", r); - } + if (cfg.buf_mode == "cpu" || cfg.buf_mode == "both") { + std::vector cpu_buf(cfg.size, 'A'); + auto cpu_mr = ep.register_buffer( + cpu_buf.data(), cfg.size, IBV_ACCESS_LOCAL_WRITE); + auto r = co_await run_write_bw( + ep, cpu_mr.view(), server_buf, DEFAULT_WARMUP, cfg.iters); + print_result("CPU", r); + } - // Signal server we're done - xchg_msg done_msg{}; - auto done_mr = ep.register_buffer(&done_msg, sizeof(done_msg), - IBV_ACCESS_LOCAL_WRITE); - auto done_wc = co_await ep.conn().send(done_mr.view()); - if (!done_wc.ok()) { - std::fprintf(stderr, "client done send failed status=%d\n", - static_cast(done_wc.status)); - co_return 1; + xchg_msg done_msg{}; + auto done_mr = ep.register_buffer( + &done_msg, sizeof(done_msg), IBV_ACCESS_LOCAL_WRITE); + auto done_wc = co_await ep.conn().send(done_mr.view()); + if (!done_wc.ok()) { + std::fprintf(stderr, "client done send failed status=%d\n", + static_cast(done_wc.status)); + result = 1; + } + } } - std::printf("\nDone.\n"); - co_return 0; + co_await ep.shutdown(); + if (result == 0) { + std::printf("\nDone.\n"); + } + co_return result; } ELIO_ASYNC_MAIN(async_main) diff --git a/examples/rdma_perf.cpp b/examples/rdma_perf.cpp index f150381c..669e6da9 100644 --- a/examples/rdma_perf.cpp +++ b/examples/rdma_perf.cpp @@ -41,6 +41,7 @@ int main() { #include #include #include +#include #include #include #include @@ -58,6 +59,14 @@ namespace { // ------------------------------------------------------------------ enum class test_mode { send, write, read }; +enum class session_result { success, failed_safe, failed_outstanding }; + +void abandon_outstanding_or_terminate( + elio::rdma_ibverbs::endpoint& ep) noexcept { + if (!ep.abandon_outstanding()) { + std::terminate(); + } +} struct config { bool is_server = false; @@ -139,8 +148,8 @@ elio::rdma_ibverbs::endpoint_config make_ep_cfg(const config& cfg) { // SEND/RECV benchmark // ------------------------------------------------------------------ -task send_server(elio::rdma_ibverbs::endpoint& ep, - const config& cfg) { +task send_server(elio::rdma_ibverbs::endpoint& ep, + const config& cfg) { const auto depth = std::max(cfg.depth, 1); std::vector buf(cfg.msg_size * depth); auto mr = ep.register_buffer(buf.data(), buf.size(), @@ -163,16 +172,22 @@ task send_server(elio::rdma_ibverbs::endpoint& ep, if (!wc.ok()) { std::fprintf(stderr, "server recv error #%zu status=%d\n", completed, static_cast(wc.status)); - co_return; + if (!inflight.empty()) { + abandon_outstanding_or_terminate(ep); + co_return session_result::failed_outstanding; + } + co_return session_result::failed_safe; } ++completed; } // Send a single ACK so client knows we're done. - co_await ep.conn().send(mr.view(0, 1)); + auto ack = co_await ep.conn().send(mr.view(0, 1)); + co_return ack.ok() ? session_result::success + : session_result::failed_safe; } -task send_client(elio::rdma_ibverbs::endpoint& ep, - const config& cfg) { +task send_client(elio::rdma_ibverbs::endpoint& ep, + const config& cfg) { std::vector tx_buf(cfg.msg_size, 'A'); std::vector ack_buf(cfg.msg_size); auto tx_mr = ep.register_buffer(tx_buf.data(), tx_buf.size(), @@ -203,26 +218,30 @@ task send_client(elio::rdma_ibverbs::endpoint& ep, if (!wc.ok()) { std::fprintf(stderr, "client send error status=%d\n", static_cast(wc.status)); - co_return false; + abandon_outstanding_or_terminate(ep); + co_return session_result::failed_outstanding; } ++completed; } } // Wait for server ACK. - co_await std::move(ack_aw); + auto ack = co_await std::move(ack_aw); + if (!ack.ok()) { + co_return session_result::failed_safe; + } const auto end = std::chrono::steady_clock::now(); print_results(cfg, end - start, cfg.count * cfg.msg_size); - co_return true; + co_return session_result::success; } // ------------------------------------------------------------------ // RDMA WRITE benchmark // ------------------------------------------------------------------ -task write_server(elio::rdma_ibverbs::endpoint& ep, - const config& cfg) { +task write_server(elio::rdma_ibverbs::endpoint& ep, + const config& cfg) { // Expose a buffer for the client to RDMA WRITE into. std::vector buf(cfg.msg_size); auto mr = ep.register_buffer(buf.data(), buf.size(), @@ -238,7 +257,7 @@ task write_server(elio::rdma_ibverbs::endpoint& ep, if (!ready.ok()) { std::fprintf(stderr, "server: client-ready recv failed status=%d\n", static_cast(ready.status)); - co_return; + co_return session_result::failed_safe; } // Send our MR info to the client. @@ -251,18 +270,20 @@ task write_server(elio::rdma_ibverbs::endpoint& ep, if (!ctrl_wc.ok()) { std::fprintf(stderr, "server: failed to send MR info status=%d\n", static_cast(ctrl_wc.status)); - co_return; + co_return session_result::failed_safe; } // Wait for client "done" signal (a single SEND). std::vector done_buf(4); auto done_mr = ep.register_buffer(done_buf.data(), done_buf.size(), IBV_ACCESS_LOCAL_WRITE); - co_await ep.conn().recv(done_mr.view()); + auto done = co_await ep.conn().recv(done_mr.view()); + co_return done.ok() ? session_result::success + : session_result::failed_safe; } -task write_client(elio::rdma_ibverbs::endpoint& ep, - const config& cfg) { +task write_client(elio::rdma_ibverbs::endpoint& ep, + const config& cfg) { // Receive server's MR info. mr_info info{}; std::vector ctrl(sizeof(info)); @@ -276,13 +297,14 @@ task write_client(elio::rdma_ibverbs::endpoint& ep, auto ready_wc = co_await ep.conn().send(ready_mr.view(0, 1)); if (!ready_wc.ok()) { std::fprintf(stderr, "client: failed to send metadata-ready signal\n"); - co_return false; + abandon_outstanding_or_terminate(ep); + co_return session_result::failed_outstanding; } auto wc = co_await std::move(ctrl_recv); if (!wc.ok()) { std::fprintf(stderr, "client: failed to recv MR info\n"); - co_return false; + co_return session_result::failed_safe; } std::memcpy(&info, ctrl.data(), sizeof(info)); remote_buffer rb{info.addr, info.length, info.rkey}; @@ -309,7 +331,11 @@ task write_client(elio::rdma_ibverbs::endpoint& ep, if (!r.ok()) { std::fprintf(stderr, "write error status=%d\n", static_cast(r.status)); - co_return false; + if (!inflight.empty()) { + abandon_outstanding_or_terminate(ep); + co_return session_result::failed_outstanding; + } + co_return session_result::failed_safe; } ++completed; } @@ -321,18 +347,21 @@ task write_client(elio::rdma_ibverbs::endpoint& ep, std::vector done_buf(4, 'D'); auto done_mr = ep.register_buffer(done_buf.data(), done_buf.size(), IBV_ACCESS_LOCAL_WRITE); - co_await ep.conn().send(done_mr.view(0, 1)); + auto done = co_await ep.conn().send(done_mr.view(0, 1)); + if (!done.ok()) { + co_return session_result::failed_safe; + } print_results(cfg, end - start, cfg.count * cfg.msg_size); - co_return true; + co_return session_result::success; } // ------------------------------------------------------------------ // RDMA READ benchmark // ------------------------------------------------------------------ -task read_server(elio::rdma_ibverbs::endpoint& ep, - const config& cfg) { +task read_server(elio::rdma_ibverbs::endpoint& ep, + const config& cfg) { // Expose a source buffer for the client to RDMA READ from. std::vector buf(cfg.msg_size, 'R'); auto mr = ep.register_buffer(buf.data(), buf.size(), @@ -348,7 +377,7 @@ task read_server(elio::rdma_ibverbs::endpoint& ep, if (!ready.ok()) { std::fprintf(stderr, "server: client-ready recv failed status=%d\n", static_cast(ready.status)); - co_return; + co_return session_result::failed_safe; } mr_info info{remote.addr, remote.length, remote.rkey}; @@ -360,18 +389,20 @@ task read_server(elio::rdma_ibverbs::endpoint& ep, if (!ctrl_wc.ok()) { std::fprintf(stderr, "server: failed to send MR info status=%d\n", static_cast(ctrl_wc.status)); - co_return; + co_return session_result::failed_safe; } // Wait for client "done" signal. std::vector done_buf(4); auto done_mr = ep.register_buffer(done_buf.data(), done_buf.size(), IBV_ACCESS_LOCAL_WRITE); - co_await ep.conn().recv(done_mr.view()); + auto done = co_await ep.conn().recv(done_mr.view()); + co_return done.ok() ? session_result::success + : session_result::failed_safe; } -task read_client(elio::rdma_ibverbs::endpoint& ep, - const config& cfg) { +task read_client(elio::rdma_ibverbs::endpoint& ep, + const config& cfg) { // Receive server's MR info. mr_info info{}; std::vector ctrl(sizeof(info)); @@ -385,13 +416,14 @@ task read_client(elio::rdma_ibverbs::endpoint& ep, auto ready_wc = co_await ep.conn().send(ready_mr.view(0, 1)); if (!ready_wc.ok()) { std::fprintf(stderr, "client: failed to send metadata-ready signal\n"); - co_return false; + abandon_outstanding_or_terminate(ep); + co_return session_result::failed_outstanding; } auto wc = co_await std::move(ctrl_recv); if (!wc.ok()) { std::fprintf(stderr, "client: failed to recv MR info\n"); - co_return false; + co_return session_result::failed_safe; } std::memcpy(&info, ctrl.data(), sizeof(info)); remote_buffer rb{info.addr, info.length, info.rkey}; @@ -422,7 +454,11 @@ task read_client(elio::rdma_ibverbs::endpoint& ep, if (!r.ok()) { std::fprintf(stderr, "read error status=%d\n", static_cast(r.status)); - co_return false; + if (!inflight.empty()) { + abandon_outstanding_or_terminate(ep); + co_return session_result::failed_outstanding; + } + co_return session_result::failed_safe; } ++completed; } @@ -433,10 +469,13 @@ task read_client(elio::rdma_ibverbs::endpoint& ep, std::vector done_buf(4, 'D'); auto done_mr = ep.register_buffer(done_buf.data(), done_buf.size(), IBV_ACCESS_LOCAL_WRITE); - co_await ep.conn().send(done_mr.view(0, 1)); + auto done = co_await ep.conn().send(done_mr.view(0, 1)); + if (!done.ok()) { + co_return session_result::failed_safe; + } print_results(cfg, end - start, cfg.count * cfg.msg_size); - co_return true; + co_return session_result::success; } // ------------------------------------------------------------------ @@ -460,17 +499,27 @@ task run_server(elio::runtime::scheduler& sched, // Pre-post a recv for SEND-based modes (server needs recv WRs // ready before client sends). + auto result = session_result::failed_safe; switch (cfg.mode) { case test_mode::send: - co_await send_server(ep, cfg); + result = co_await send_server(ep, cfg); break; case test_mode::write: - co_await write_server(ep, cfg); + result = co_await write_server(ep, cfg); break; case test_mode::read: - co_await read_server(ep, cfg); + result = co_await read_server(ep, cfg); break; } + // Graceful paths have released every helper-owned MR/awaiter before + // shutdown. An outstanding failure already called abandon_outstanding() + // from inside that helper while those objects were still alive. + if (result != session_result::failed_outstanding) { + co_await ep.shutdown(); + } + if (result != session_result::success) { + std::fprintf(stderr, "session failed\n"); + } std::printf("session complete, waiting for next client...\n"); } } @@ -491,20 +540,27 @@ task run_client(elio::runtime::scheduler& sched, cfg.host.c_str(), cfg.port, mode_str, cfg.msg_size, cfg.count, cfg.depth); - bool ok = false; + auto result = session_result::failed_safe; switch (cfg.mode) { case test_mode::send: - ok = co_await send_client(ep, cfg); + result = co_await send_client(ep, cfg); break; case test_mode::write: - ok = co_await write_client(ep, cfg); + result = co_await write_client(ep, cfg); break; case test_mode::read: - ok = co_await read_client(ep, cfg); + result = co_await read_client(ep, cfg); break; } - co_return ok ? 0 : 1; + // A failed pipelined session can still own provider work through a + // pre-posted receive or remaining WRs. Its helper has already destroyed the + // QP fail-closed while the associated MRs and buffers were still alive; + // completed error paths are safe to shut down gracefully. + if (result != session_result::failed_outstanding) { + co_await ep.shutdown(); + } + co_return result == session_result::success ? 0 : 1; } void usage(const char* prog) { @@ -583,6 +639,7 @@ int main(int argc, char* argv[]) { sched.start(); std::atomic exit_code{0}; + std::atomic done{false}; if (cfg.is_server) { sched.go([&]() -> task { @@ -592,7 +649,6 @@ int main(int argc, char* argv[]) { std::printf("press Ctrl+C to stop\n"); pause(); } else { - std::atomic done{false}; sched.go([&]() -> task { exit_code.store(co_await run_client(sched, cfg)); done.store(true); diff --git a/examples/rdma_req_resp_ibverbs.cpp b/examples/rdma_req_resp_ibverbs.cpp index d98bba58..687f4ac5 100644 --- a/examples/rdma_req_resp_ibverbs.cpp +++ b/examples/rdma_req_resp_ibverbs.cpp @@ -31,6 +31,7 @@ int main() { return 0; } #include #include #include +#include #include #include #include @@ -49,6 +50,57 @@ struct request_header { constexpr std::uint16_t kPort = 18515; constexpr std::uint32_t kImmDoneTag = 0xABCDEFu; +void abandon_outstanding_or_terminate( + elio::rdma_ibverbs::endpoint& ep) noexcept { + if (!ep.abandon_outstanding()) { + std::terminate(); + } +} + +elio::coro::task serve_request( + elio::rdma_ibverbs::endpoint& ep) { + std::array req_buf{}; + std::array payload_buf{}; + std::array notify_buf{}; + + auto req_mr = ep.register_buffer( + req_buf.data(), req_buf.size(), IBV_ACCESS_LOCAL_WRITE); + auto payload_mr = ep.register_buffer( + payload_buf.data(), payload_buf.size(), IBV_ACCESS_LOCAL_WRITE); + auto notify_mr = ep.register_buffer( + notify_buf.data(), sizeof(notify_buf), IBV_ACCESS_LOCAL_WRITE); + + auto req_wc = co_await ep.conn().recv(req_mr.view()); + if (!req_wc.ok()) { + std::cerr << "server: recv failed status=" + << static_cast(req_wc.status) << "\n"; + co_return; + } + auto* hdr = reinterpret_cast(req_buf.data()); + std::cout << "server: got request from client " + << hdr->client_id << "\n"; + + std::string body = "response for client " + std::to_string(hdr->client_id); + std::memcpy(payload_buf.data(), body.data(), body.size()); + elio::rdma::remote_buffer dst{ + hdr->resp_addr, hdr->resp_length, hdr->resp_rkey}; + + auto wr = co_await ep.conn().rdma_write( + payload_mr.view(0, body.size()), dst); + if (!wr.ok()) { + std::cerr << "server: rdma_write failed status=" + << static_cast(wr.status) << "\n"; + co_return; + } + + auto notify = co_await ep.conn().send_with_imm( + notify_mr.view(0, 0), static_cast(body.size())); + if (!notify.ok()) { + std::cerr << "server: notify send failed status=" + << static_cast(notify.status) << "\n"; + } +} + elio::coro::task server(elio::runtime::scheduler& sched) { sockaddr_in bind_addr{}; bind_addr.sin_family = AF_INET; @@ -65,116 +117,79 @@ elio::coro::task server(elio::runtime::scheduler& sched) { .max_send_wr = 4, .max_recv_wr = 4}); ep.start_cq_pump(sched); - // Stash buffers for the lifetime of this connection. - std::array req_buf{}; - std::array payload_buf{}; - std::array notify_buf{}; - - auto req_mr = ep.register_buffer( - req_buf.data(), req_buf.size(), IBV_ACCESS_LOCAL_WRITE); - auto payload_mr = ep.register_buffer( - payload_buf.data(), payload_buf.size(), - IBV_ACCESS_LOCAL_WRITE); - auto notify_mr = ep.register_buffer( - notify_buf.data(), sizeof(notify_buf), IBV_ACCESS_LOCAL_WRITE); - - // Post recv for the request. - auto req_wc = co_await ep.conn().recv(req_mr.view()); - if (!req_wc.ok()) { - std::cerr << "server: recv failed status=" - << static_cast(req_wc.status) << "\n"; - co_return; - } - auto* hdr = reinterpret_cast(req_buf.data()); - std::cout << "server: got request from client " - << hdr->client_id << "\n"; - - // Build a response payload. - std::string body = "response for client " + std::to_string(hdr->client_id); - std::memcpy(payload_buf.data(), body.data(), body.size()); - - elio::rdma::remote_buffer dst{ - hdr->resp_addr, hdr->resp_length, hdr->resp_rkey}; - - // RDMA WRITE the response into the client's exposed buffer. - auto wr = co_await ep.conn().rdma_write( - payload_mr.view(0, body.size()), dst); - if (!wr.ok()) { - std::cerr << "server: rdma_write failed status=" - << static_cast(wr.status) << "\n"; - co_return; - } - - // SEND_WITH_IMM the OOB completion: imm carries the body length - // so the client knows exactly how much was written. - auto notify = co_await ep.conn().send_with_imm( - notify_mr.view(0, 0), // zero-length payload; carrier is imm - static_cast(body.size())); - if (!notify.ok()) { - std::cerr << "server: notify send failed status=" - << static_cast(notify.status) << "\n"; - } + // The helper frame owns every MR and completed operation. It is + // destroyed before terminal endpoint shutdown begins. + co_await serve_request(ep); + co_await ep.shutdown(); co_return; // single-shot demo } } -elio::coro::task client(elio::runtime::scheduler& sched) { - sockaddr_in dst_addr{}; - dst_addr.sin_family = AF_INET; - dst_addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); - dst_addr.sin_port = htons(kPort); - - elio::rdma_cm::event_channel cm_ch; - auto ep = co_await elio::rdma_ibverbs::connect( - cm_ch, reinterpret_cast(&dst_addr), sizeof(dst_addr), - elio::rdma_ibverbs::endpoint_config{ - .max_send_wr = 4, .max_recv_wr = 4}); - ep.start_cq_pump(sched); - - std::array req_buf{}; +elio::coro::task run_client_request( + elio::rdma_ibverbs::endpoint& ep) { + std::array req_buf{}; std::array resp_buf{}; std::array notify_buf{}; auto req_mr = ep.register_buffer( req_buf.data(), req_buf.size(), IBV_ACCESS_LOCAL_WRITE); - // resp_buf needs REMOTE_WRITE so the server can RDMA_WRITE into it. auto resp_mr = ep.register_buffer( resp_buf.data(), resp_buf.size(), IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE); auto notify_mr = ep.register_buffer( notify_buf.data(), sizeof(notify_buf), IBV_ACCESS_LOCAL_WRITE); - // CRITICAL: post the recv for the OOB notify BEFORE sending the - // request. Otherwise the server's send-with-imm will RNR-retry. auto notify_awaiter = ep.conn().recv(notify_mr.view()).start(); auto* hdr = reinterpret_cast(req_buf.data()); auto resp_remote = resp_mr.remote(); - hdr->resp_addr = resp_remote.addr; + hdr->resp_addr = resp_remote.addr; hdr->resp_length = resp_remote.length; - hdr->resp_rkey = resp_remote.rkey; - hdr->client_id = 7; + hdr->resp_rkey = resp_remote.rkey; + hdr->client_id = 7; auto send_wc = co_await ep.conn().send( req_mr.view(0, sizeof(request_header))); if (!send_wc.ok()) { std::cerr << "client: send failed status=" << static_cast(send_wc.status) << "\n"; - co_return; + // The pre-posted notify receive may still be provider-owned. Destroy + // the QP while this frame still owns its awaiter, MRs, and buffers; + // the terminal fail-closed path intentionally leaks the remaining + // endpoint resources. + abandon_outstanding_or_terminate(ep); + co_return false; } auto notify_wc = co_await std::move(notify_awaiter); if (!notify_wc.ok()) { std::cerr << "client: notify recv failed status=" << static_cast(notify_wc.status) << "\n"; - co_return; + co_return true; } - // imm_data carries the response length. const auto resp_len = notify_wc.imm_data; std::cout << "client: server wrote " << resp_len << " bytes: '" << std::string(resp_buf.data(), resp_len) << "'\n"; - co_return; + co_return true; +} + +elio::coro::task client(elio::runtime::scheduler& sched) { + sockaddr_in dst_addr{}; + dst_addr.sin_family = AF_INET; + dst_addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + dst_addr.sin_port = htons(kPort); + + elio::rdma_cm::event_channel cm_ch; + auto ep = co_await elio::rdma_ibverbs::connect( + cm_ch, reinterpret_cast(&dst_addr), sizeof(dst_addr), + elio::rdma_ibverbs::endpoint_config{ + .max_send_wr = 4, .max_recv_wr = 4}); + ep.start_cq_pump(sched); + + if (co_await run_client_request(ep)) { + co_await ep.shutdown(); + } } } // namespace diff --git a/include/elio/rdma_ibverbs/endpoint.hpp b/include/elio/rdma_ibverbs/endpoint.hpp index 721e6732..58cc61a3 100644 --- a/include/elio/rdma_ibverbs/endpoint.hpp +++ b/include/elio/rdma_ibverbs/endpoint.hpp @@ -20,8 +20,11 @@ /// elio::rdma_ibverbs::endpoint_config{.max_recv_wr = 8}); /// ep.start_cq_pump(sched); /// -/// auto mr = ep.register_buffer(buf, len, IBV_ACCESS_LOCAL_WRITE); -/// auto wc = co_await ep.conn().send(mr.view()); + /// { + /// auto mr = ep.register_buffer(buf, len, IBV_ACCESS_LOCAL_WRITE); + /// auto wc = co_await ep.conn().send(mr.view()); + /// } // The completed awaiter and MR are gone before shutdown. + /// co_await ep.shutdown(); /// @endcode /// /// `endpoint` is NOT for users who need fine-grained control over PD @@ -29,13 +32,14 @@ /// the low-level `elio::rdma::connection` surface. Both /// styles coexist; mixing within a process is fine. /// -/// **Shutdown contract**: the destructor sets the cq_pump cancel -/// token AND destroys the QP first (which generates flush CQEs that -/// wake the pump). It then waits up to ~1s for the pump to observe -/// the cancel and exit before tearing down the CQ / comp_channel / -/// PD. If the pump doesn't exit in time (e.g. a custom drain that -/// blocks indefinitely), the verbs resources are intentionally -/// leaked rather than risk a use-after-free on the polling thread. +/// **Shutdown contract**: after all posted operations have completed and all +/// endpoint memory regions have been destroyed, `shutdown()` cancels and joins +/// the cq_pump asynchronously before tearing down the CQ / comp_channel / PD. +/// The pump scheduler must remain operational until that await completes, and +/// starting shutdown permanently ends data-path use. The destructor does not +/// wait for the pump. If it observes an active pump, or if checked QP +/// destruction fails, the affected ownership graph is intentionally leaked +/// rather than risk a use-after-free or continue destroying dependencies. #if !defined(ELIO_HAS_RDMA_CM) || !ELIO_HAS_RDMA_CM // endpoint depends on elio_rdma_cm for the CM bootstrap step. Build @@ -46,6 +50,7 @@ namespace elio::rdma_ibverbs { /* endpoint API not available */ } #else #include +#include #include #include #include @@ -64,13 +69,13 @@ namespace elio::rdma_ibverbs { /* endpoint API not available */ } #include #include -#include #include #include +#include #include +#include #include #include -#include #include #if defined(ELIO_HAS_RDMA_CUDA) && ELIO_HAS_RDMA_CUDA @@ -98,6 +103,84 @@ inline void set_nonblocking(int fd) { } } +inline void require_endpoint_active(bool shutdown_started, + const char* operation) { + if (shutdown_started) { + throw std::logic_error( + std::string("endpoint: ") + operation + + " is not allowed after shutdown has started"); + } +} + +[[nodiscard]] inline bool pump_resources_are_idle(bool pump_started, + bool pump_exited, + bool pump_frame_destroyed) noexcept { + return !pump_started || pump_exited || pump_frame_destroyed; +} + +class pump_exit_state { +public: + class wait_awaitable { + public: + explicit wait_awaitable(pump_exit_state& state) noexcept + : state_(&state), waiter_(state.waiter_) {} + + [[nodiscard]] bool await_ready() const noexcept { + return state_->is_exited(); + } + + bool await_suspend(std::coroutine_handle<> handle) noexcept { + return state_->waiter_.register_waiter( + waiter_, handle, [this] { return state_->is_exited(); }); + } + + void await_resume() const noexcept {} + + private: + pump_exit_state* state_; + elio::coro::detail::completion_waiter waiter_; + }; + + [[nodiscard]] wait_awaitable wait() noexcept { + return wait_awaitable(*this); + } + + [[nodiscard]] bool is_exited() const noexcept { + return exited_.load(std::memory_order_acquire); + } + + void mark_exited() noexcept { + exited_.store(true, std::memory_order_release); + if (auto waiter = waiter_.take()) { + elio::runtime::schedule_handle(waiter); + } + } + +private: + std::atomic exited_{false}; + elio::coro::detail::completion_waiter_slot waiter_; +}; + +template +void destroy_resource_or_throw(Resource*& resource, + Destroy&& destroy, + const char* operation) { + if (!resource) return; + + errno = 0; + const int result = std::forward(destroy)(resource); + if (result != 0) { + const int error = result > 0 + ? result + : (errno != 0 ? errno : (result < -1 ? -result : EIO)); + errno = error; + throw std::runtime_error( + std::string("endpoint: ") + operation + " failed: " + + std::strerror(error)); + } + resource = nullptr; +} + } // namespace endpoint_detail /// Per-endpoint creation knobs. Most users only touch the cap fields. @@ -130,7 +213,7 @@ class endpoint { /// /// Throws `std::runtime_error` if any verbs setup step fails; /// partial resources are cleaned up before the exception - /// propagates. + /// propagates. The transferred cm_id must not already have an attached QP. endpoint(rdma_cm::cm_id id, endpoint_config cfg = {}) : id_(std::move(id)), config_(cfg), disp_(std::make_unique()) { @@ -142,24 +225,29 @@ class endpoint { "endpoint: cm_id has no verbs context " "(resolve_addr not yet completed)"); } + if (id_.qp()) { + throw std::runtime_error( + "endpoint: cm_id already has an attached QP"); + } try { build_resources_(); + conn_ = + std::make_unique>( + qp_, *disp_, elio::rdma::connection_config{ + .max_inline_data = cfg.max_inline_data}); } catch (...) { destroy_resources_(); throw; } - conn_ = std::make_unique>( - qp_, *disp_, - elio::rdma::connection_config{.max_inline_data = cfg.max_inline_data}); } endpoint(const endpoint&) = delete; endpoint& operator=(const endpoint&) = delete; - // Movable: every owned object that needs a stable address is - // behind a unique_ptr (dispatcher, connection, pump_state) or a - // shared_ptr (cancel_source state). Raw pointers are nulled in - // the source so its destructor doesn't double-free. + // Movable: every owned object that needs a stable address is behind a + // unique_ptr (dispatcher, connection) or shared ownership (pump/cancel + // state). Raw pointers are nulled in the source so its destructor doesn't + // double-free. endpoint(endpoint&& other) noexcept : id_(std::move(other.id_)), config_(other.config_), @@ -170,7 +258,9 @@ class endpoint { disp_(std::move(other.disp_)), conn_(std::move(other.conn_)), pump_cancel_(std::move(other.pump_cancel_)), - pump_state_(std::move(other.pump_state_)) {} + pump_state_(std::move(other.pump_state_)), + pump_handle_(std::exchange(other.pump_handle_, std::nullopt)), + shutdown_started_(std::exchange(other.shutdown_started_, true)) {} endpoint& operator=(endpoint&& other) noexcept { if (this != &other) { @@ -185,51 +275,37 @@ class endpoint { conn_ = std::move(other.conn_); pump_cancel_ = std::move(other.pump_cancel_); pump_state_ = std::move(other.pump_state_); + pump_handle_ = std::exchange(other.pump_handle_, std::nullopt); + shutdown_started_ = std::exchange(other.shutdown_started_, true); } return *this; } - ~endpoint() { - if (pump_state_) { - pump_cancel_.cancel(); - } - // Destroying QP first generates flush CQEs that wake the pump - // and unblock its async_poll_read. Use rdma_destroy_qp (not - // ibv_destroy_qp) so the cm_id's internal qp pointer is nulled; - // otherwise rdma_destroy_id would double-free it. - if (qp_) { - ::rdma_destroy_qp(id_.native()); - qp_ = nullptr; - } - // Wait briefly for the pump to observe the cancel and exit. - // If it doesn't (custom drain that blocks indefinitely, or no - // flush CQEs because the QP had no posted WRs), we leak the - // remaining verbs resources AND the dispatcher rather than - // risk a UAF when ibv_destroy_cq or ~dispatcher runs while the - // pump coroutine is still referencing them. - bool pump_clean = true; - if (pump_state_) { - auto deadline = std::chrono::steady_clock::now() - + std::chrono::milliseconds(1000); - while (!pump_state_->done.load(std::memory_order_acquire) - && std::chrono::steady_clock::now() < deadline) { - std::this_thread::sleep_for(std::chrono::milliseconds(5)); - } - pump_clean = pump_state_->done.load(std::memory_order_acquire); + ~endpoint() noexcept { + request_pump_stop_(); + if (!destroy_qp_noexcept_()) { + // rdma_cm's void-returning wrapper discards ibv_destroy_qp() + // failures and then lets cm_id destruction continue. Preserve the + // entire dependency graph, including the CM ID, instead. + relinquish_all_resources_(); + return; } - if (pump_clean) { - if (cq_) ::ibv_destroy_cq(cq_); - if (comp_ch_) ::ibv_destroy_comp_channel(comp_ch_); - if (pd_) ::ibv_dealloc_pd(pd_); + + if (pump_resources_are_idle_()) { + destroy_remaining_resources_(); } else { + // The pump still owns raw references to the CQ, completion channel, + // and dispatcher. A destructor cannot join it without blocking the + // current scheduler worker, so preserve the previous fail-closed + // policy and intentionally leak those resources. (void)disp_.release(); } - // cm_id RAII handles itself. } /// Returns the data-path connection. Lifetime is tied to the /// endpoint; do not move out. - [[nodiscard]] elio::rdma::connection& conn() noexcept { + [[nodiscard]] elio::rdma::connection& conn() { + require_active_("conn"); return *conn_; } @@ -237,6 +313,7 @@ class endpoint { /// move-only and auto-deregisters at scope exit. [[nodiscard]] elio::rdma::memory_region register_buffer(void* addr, std::size_t length, int access) { + require_active_("register_buffer"); return elio::rdma::memory_region{ pd_, addr, length, access}; } @@ -247,45 +324,119 @@ class endpoint { [[nodiscard]] auto cas(elio::rdma::buffer_view local, elio::rdma::remote_buffer remote, std::uint64_t compare, std::uint64_t swap, - elio::rdma::send_flags flags = {}) noexcept { + elio::rdma::send_flags flags = {}) { + require_active_("cas"); return conn_->cas(local, remote, compare, swap, flags); } [[nodiscard]] auto fetch_add(elio::rdma::buffer_view local, elio::rdma::remote_buffer remote, std::uint64_t add, - elio::rdma::send_flags flags = {}) noexcept { + elio::rdma::send_flags flags = {}) { + require_active_("fetch_add"); return conn_->fetch_add(local, remote, add, flags); } /// Spawn the cq_pump on the given scheduler. Idempotent: a /// second call is a no-op. void start_cq_pump(elio::runtime::scheduler& sched) { - if (pump_state_) return; - pump_state_ = std::make_shared(); + require_active_("start_cq_pump"); + if (pump_handle_) return; + pump_state_ = std::make_shared(); auto state = pump_state_; auto token = pump_cancel_.get_token(); auto cq = cq_; auto comp_ch = comp_ch_; auto* disp = disp_.get(); - sched.go([state, token, cq, comp_ch, disp]() -> elio::coro::task { - co_await elio::rdma::cq_pump(comp_ch->fd, *disp, - make_cq_drain(cq, comp_ch), - token); - state->done.store(true, std::memory_order_release); - }); + pump_handle_.emplace(sched.go_joinable(pump_runner{ + std::move(state), std::move(token), cq, comp_ch, disp})); } - /// Cancel the cq_pump's loop. Does not wait — the pump exits at - /// its next iteration. For a synchronous join, rely on the - /// destructor's bounded wait OR ensure outstanding WRs flush - /// before this call so a CQE wakes the pump. + /// Cancel the cq_pump's loop. Does not wait; use shutdown() to join the + /// pump and release its verbs resources without blocking a worker thread. void stop_cq_pump() noexcept { - if (pump_state_) pump_cancel_.cancel(); + request_pump_stop_(); + } + + /// Gracefully stop and join the CQ pump, then release the QP, CQ, + /// completion channel, and PD. The endpoint must remain alive until this + /// task completes. Every posted operation and its awaiter must have + /// completed, and all registered memory regions must be destroyed first. + /// The scheduler passed to start_cq_pump() must remain operational until + /// this task completes; force-stopping it can orphan in-flight I/O. Once + /// this task starts executing the endpoint is terminal: only a later + /// serialized shutdown() retry, raw null-state inspection, or destruction + /// is supported. Resource-destruction failures retain ownership for that + /// retry. Concurrent shutdown calls are not supported. + [[nodiscard]] elio::coro::task shutdown() { + shutdown_started_ = true; + request_pump_stop_(); + + std::exception_ptr pump_error; + if (pump_handle_) { + co_await pump_state_->wait(); + try { + if (pump_handle_->is_ready()) { + pump_handle_->await_resume(); + } + } catch (...) { + pump_error = std::current_exception(); + } + } + + // The callable object publishes exit from its destructor after its last + // raw-resource access. That covers normal completion, exceptions, + // admission rejection, and runner-frame destruction while the pump + // scheduler remains operational. Only consume the join result when the + // join state completed; destruction does not manufacture success. + destroy_qp_or_throw_(); + std::exception_ptr teardown_error; + if (pump_resources_are_idle_()) { + try { + destroy_remaining_resources_or_throw_(); + } catch (...) { + teardown_error = std::current_exception(); + } + } + + if (teardown_error) { + std::rethrow_exception(teardown_error); + } + + if (pump_error) { + std::rethrow_exception(pump_error); + } + } + + /// Terminal fail-closed escape hatch for fatal paths whose remaining + /// provider-posted operations were launched eagerly with .start() and have + /// never installed a coroutine waiter. Call this while those eager + /// awaitables, their memory regions, and payload buffers are still alive. + /// The QP is destroyed synchronously before returning, then the CQ, + /// completion channel, PD, and dispatcher are intentionally relinquished + /// so those eager objects can unwind without racing resource destruction. + /// This does not make a suspended operation frame safe to destroy: once + /// delivery snapshots its waiter, that coroutine must resume and consume + /// the result. This operation never waits, permanently ends endpoint use, + /// and intentionally leaks the relinquished resources. It returns true only + /// after QP destruction succeeds. A false result retains the QP and every + /// dependency: keep all operation/MR/buffer lifetimes intact, repair the + /// provider state through the raw accessors, and retry (or terminate). + /// Prefer shutdown() after normal completion. + [[nodiscard]] bool abandon_outstanding() noexcept { + shutdown_started_ = true; + request_pump_stop_(); + if (!destroy_qp_noexcept_()) { + return false; + } + + relinquish_remaining_resources_(); + return true; } #if defined(ELIO_HAS_RDMA_CUDA) && ELIO_HAS_RDMA_CUDA [[nodiscard]] elio::rdma_cuda::gpu_memory_region register_gpu_buffer(std::size_t size, int access) { + require_active_("register_gpu_buffer"); return elio::rdma_cuda::gpu_memory_region{pd_, size, access}; } #endif @@ -297,7 +448,8 @@ class endpoint { [[nodiscard]] ibv_qp* qp() const noexcept { return qp_; } [[nodiscard]] ibv_comp_channel* comp_ch() const noexcept { return comp_ch_; } [[nodiscard]] rdma_cm::cm_id& cm_id() noexcept { return id_; } - [[nodiscard]] elio::rdma::dispatcher& dispatcher_ref() noexcept { + [[nodiscard]] elio::rdma::dispatcher& dispatcher_ref() { + require_active_("dispatcher_ref"); return *disp_; } [[nodiscard]] const endpoint_config& config() const noexcept { @@ -305,10 +457,135 @@ class endpoint { } private: - struct pump_state { - std::atomic done{false}; + struct pump_runner { + std::shared_ptr state; + elio::coro::cancel_token token; + ibv_cq* cq; + ibv_comp_channel* comp_ch; + elio::rdma::dispatcher* disp; + + pump_runner(std::shared_ptr state_arg, + elio::coro::cancel_token token_arg, + ibv_cq* cq_arg, + ibv_comp_channel* comp_ch_arg, + elio::rdma::dispatcher* disp_arg) noexcept + : state(std::move(state_arg)), + token(std::move(token_arg)), + cq(cq_arg), + comp_ch(comp_ch_arg), + disp(disp_arg) {} + + pump_runner(const pump_runner&) = delete; + pump_runner& operator=(const pump_runner&) = delete; + + pump_runner(pump_runner&& other) noexcept + : state(std::move(other.state)), + token(std::move(other.token)), + cq(std::exchange(other.cq, nullptr)), + comp_ch(std::exchange(other.comp_ch, nullptr)), + disp(std::exchange(other.disp, nullptr)) {} + + ~pump_runner() { + if (state) { + state->mark_exited(); + } + } + + [[nodiscard]] elio::coro::task operator()() { + co_await elio::rdma::cq_pump(comp_ch->fd, *disp, + make_cq_drain(cq, comp_ch), token); + } }; + void request_pump_stop_() noexcept { + if (!pump_handle_) return; + try { + pump_cancel_.cancel(); + } catch (...) { + // The endpoint-owned token is only registered with library I/O + // waits, but teardown must remain noexcept even if that invariant + // changes in the future. + } + } + + void require_active_(const char* operation) const { + endpoint_detail::require_endpoint_active( + shutdown_started_, operation); + } + + [[nodiscard]] bool pump_resources_are_idle_() const noexcept { + const bool pump_exited = pump_state_ && pump_state_->is_exited(); + return endpoint_detail::pump_resources_are_idle( + pump_handle_.has_value(), pump_exited, + pump_handle_ && pump_handle_->is_destroyed()); + } + + void destroy_qp_or_throw_() { + if (!qp_) { + return; + } + auto* owned_qp = qp_; + if (!id_ || id_.native()->qp != owned_qp) { + throw std::runtime_error( + "endpoint: CM ID and endpoint QP ownership diverged"); + } + endpoint_detail::destroy_resource_or_throw( + qp_, [](ibv_qp* qp) { return ::ibv_destroy_qp(qp); }, + "ibv_destroy_qp"); + if (id_ && id_.native()->qp == owned_qp) { + // Mirror rdma_destroy_qp() only after this exact provider QP was + // destroyed, so cm_id cannot later attempt a double destroy. + id_.native()->qp = nullptr; + } + } + + [[nodiscard]] bool destroy_qp_noexcept_() noexcept { + try { + destroy_qp_or_throw_(); + return true; + } catch (...) { + return false; + } + } + + void relinquish_remaining_resources_() noexcept { + cq_ = nullptr; + comp_ch_ = nullptr; + pd_ = nullptr; + (void)disp_.release(); + } + + void relinquish_all_resources_() noexcept { + qp_ = nullptr; + relinquish_remaining_resources_(); + (void)id_.release(); + } + + void destroy_remaining_resources_() noexcept { + try { + destroy_remaining_resources_or_throw_(); + } catch (...) { + // Destructors and constructor rollback cannot report provider + // failures. Stop at the first dependency that remains live and + // intentionally retain the rest rather than invalidating it. + } + } + + void destroy_remaining_resources_or_throw_() { + endpoint_detail::destroy_resource_or_throw( + cq_, [](ibv_cq* cq) { return ::ibv_destroy_cq(cq); }, + "ibv_destroy_cq"); + endpoint_detail::destroy_resource_or_throw( + comp_ch_, + [](ibv_comp_channel* channel) { + return ::ibv_destroy_comp_channel(channel); + }, + "ibv_destroy_comp_channel"); + endpoint_detail::destroy_resource_or_throw( + pd_, [](ibv_pd* pd) { return ::ibv_dealloc_pd(pd); }, + "ibv_dealloc_pd"); + } + void build_resources_() { pd_ = ::ibv_alloc_pd(id_.verbs()); if (!pd_) throw_verbs_("ibv_alloc_pd"); @@ -349,13 +626,11 @@ class endpoint { } void destroy_resources_() noexcept { - if (qp_) { - ::rdma_destroy_qp(id_.native()); - qp_ = nullptr; + if (!destroy_qp_noexcept_()) { + relinquish_all_resources_(); + return; } - if (cq_) { ::ibv_destroy_cq(cq_); cq_ = nullptr; } - if (comp_ch_) { ::ibv_destroy_comp_channel(comp_ch_); comp_ch_ = nullptr; } - if (pd_) { ::ibv_dealloc_pd(pd_); pd_ = nullptr; } + destroy_remaining_resources_(); } [[noreturn]] void throw_verbs_(const char* op) { @@ -374,7 +649,9 @@ class endpoint { std::unique_ptr disp_; std::unique_ptr> conn_; elio::coro::cancel_source pump_cancel_; - std::shared_ptr pump_state_; + std::shared_ptr pump_state_; + std::optional> pump_handle_; + bool shutdown_started_ = false; }; // --------------------------------------------------------------------- diff --git a/tests/unit/test_rdma_ibverbs_endpoint.cpp b/tests/unit/test_rdma_ibverbs_endpoint.cpp index 34c55876..52e7a6bb 100644 --- a/tests/unit/test_rdma_ibverbs_endpoint.cpp +++ b/tests/unit/test_rdma_ibverbs_endpoint.cpp @@ -14,14 +14,31 @@ #include #include +#include #include #include #include +#include +#include +#include +#include + using elio::rdma_ibverbs::endpoint_config; +static_assert(std::is_nothrow_destructible_v); +static_assert(std::is_same_v< + decltype(std::declval().shutdown()), + elio::coro::task>); +static_assert(noexcept( + std::declval().abandon_outstanding())); +static_assert(std::is_same_v< + decltype(std::declval() + .abandon_outstanding()), + bool>); + TEST_CASE("endpoint comp channel fds are made non-blocking", "[rdma][ibverbs][endpoint]") { int pipe_fds[2]{-1, -1}; @@ -60,6 +77,116 @@ TEST_CASE("endpoint_config defaults match the documented values", REQUIRE(cfg.custom_qp_init_attr == nullptr); } +TEST_CASE("endpoint teardown only releases resources after the pump is idle", + "[rdma][ibverbs][endpoint][shutdown]") { + using elio::rdma_ibverbs::endpoint_detail::pump_resources_are_idle; + + REQUIRE(pump_resources_are_idle(false, false, false)); + REQUIRE_FALSE(pump_resources_are_idle(true, false, false)); + REQUIRE(pump_resources_are_idle(true, true, false)); + REQUIRE(pump_resources_are_idle(true, false, true)); +} + +TEST_CASE("endpoint pump exit signal wakes and unregisters coroutine waiters", + "[rdma][ibverbs][endpoint][shutdown]") { + using elio::rdma_ibverbs::endpoint_detail::pump_exit_state; + + SECTION("exit wakes a suspended shutdown waiter") { + pump_exit_state state; + std::atomic waiting{false}; + std::atomic resumed{false}; + elio::runtime::scheduler scheduler(1); + scheduler.start(); + + auto wait_for_exit = [&]() -> elio::coro::task { + waiting.store(true, std::memory_order_release); + waiting.notify_one(); + co_await state.wait(); + resumed.store(true, std::memory_order_release); + }; + auto waiter = scheduler.go_joinable(wait_for_exit); + + waiting.wait(false, std::memory_order_acquire); + auto publish_exit = [&]() -> elio::coro::task { + state.mark_exited(); + co_return; + }; + auto publisher = scheduler.go_joinable(publish_exit); + + publisher.wait_destroyed(); + waiter.wait_destroyed(); + scheduler.shutdown(); + + REQUIRE(resumed.load(std::memory_order_acquire)); + } + + SECTION("pre-published exit does not suspend") { + pump_exit_state state; + state.mark_exited(); + bool resumed = false; + auto wait_for_prepublished_exit = [&]() -> elio::coro::task { + co_await state.wait(); + resumed = true; + }; + auto waiting = wait_for_prepublished_exit(); + auto handle = elio::coro::detail::task_access::handle(waiting); + + handle.resume(); + REQUIRE(handle.done()); + REQUIRE(resumed); + } + + SECTION("destroyed shutdown waiter unregisters before pump exit") { + pump_exit_state state; + { + auto wait_for_abandoned_exit = [&]() -> elio::coro::task { + co_await state.wait(); + }; + auto abandoned = wait_for_abandoned_exit(); + auto handle = elio::coro::detail::task_access::handle(abandoned); + handle.resume(); + REQUIRE_FALSE(handle.done()); + } + + REQUIRE_NOTHROW(state.mark_exited()); + REQUIRE(state.is_exited()); + } +} + +TEST_CASE("endpoint resource teardown retains failures for retry", + "[rdma][ibverbs][endpoint][shutdown]") { + using elio::rdma_ibverbs::endpoint_detail::destroy_resource_or_throw; + + int storage = 0; + int* resource = &storage; + int attempts = 0; + auto destroy = [&](int*) { + ++attempts; + return attempts == 1 ? EBUSY : 0; + }; + + REQUIRE_THROWS_AS( + destroy_resource_or_throw(resource, destroy, "fake_destroy"), + std::runtime_error); + REQUIRE(resource == &storage); + REQUIRE(errno == EBUSY); + + REQUIRE_NOTHROW( + destroy_resource_or_throw(resource, destroy, "fake_destroy")); + REQUIRE(resource == nullptr); + REQUIRE(attempts == 2); +} + +TEST_CASE("endpoint data-path use is rejected after shutdown starts", + "[rdma][ibverbs][endpoint][shutdown]") { + using elio::rdma_ibverbs::endpoint_detail::require_endpoint_active; + + REQUIRE_NOTHROW(require_endpoint_active(false, "test_operation")); + REQUIRE_THROWS_AS( + require_endpoint_active(true, "test_operation"), + std::logic_error); +} + TEST_CASE("rdma_ibverbs module version matches S13", "[rdma][ibverbs][endpoint][version]") { REQUIRE(std::string(elio::rdma_ibverbs::module_version) == "0.0.14-S13"); diff --git a/wiki/API-Contracts.md b/wiki/API-Contracts.md index 52cc76ac..e38e2838 100644 --- a/wiki/API-Contracts.md +++ b/wiki/API-Contracts.md @@ -273,10 +273,10 @@ Names below use the convenience re-exports from `` and | `memory_region::view()`, `view(offset, sub_length)`, `remote()`, and `remote(offset, sub_length)` | Produce verbs-adjacent local/remote views. | Bounds and 32-bit sub-length preconditions are caller-owned where documented. | | `rdma::dispatcher` and `rdma::cq_pump` | Deliver completions to awaiting operations according to `wr_id`/`op_state` lifecycle rules. For a suspended operation, delivery snapshots its waiter before publishing `pending → completed`; forced frame destruction after that transition fails closed because safe revocation of the snapshot cannot be proven. | Drive the dispatcher/CQ pump until all outstanding operations complete or are intentionally drained. After delivery wins for a suspended operation, do not force-destroy its coroutine before the result is consumed. | | `rdma_cm::event_channel` | Wraps an RDMA-CM event channel/fd, exposes cancellable event waits, and requires acking consumed events. | Destroy derived `cm_id` objects before the channel. Ack every event exactly once according to the helper contract. | -| `rdma_cm::cm_id` | Owns one `rdma_cm_id*`, destroys attached QP defensively, and releases ownership only through `release()`. | Do not use moved-from IDs. If `release()` is called, the caller owns eventual `rdma_destroy_id` and any attached QP teardown. | +| `rdma_cm::cm_id` | Owns one `rdma_cm_id*`, destroys attached QP/CQ resources through the RDMA-CM teardown path, and releases ownership only through `release()`. | Do not use moved-from IDs. If `release()` is called, the caller owns eventual `rdma_destroy_id` and any attached QP teardown. | | `rdma_cm::resolve()` and `complete_connect()` | Drive client-side address/route resolve and final connect phases, returning `cm_id`/`cm_status` with normalized errors. | Configure addresses, routes, port space, QP attributes, connection parameters, and teardown policy valid for the provider. Create the QP before `complete_connect()`. | | `rdma_cm::cm_listener::accept()` and `rdma_cm::accept_connect()` | Consume connect requests for the listener and complete server-side accept after caller-created QP setup. | Keep listener/channel alive, create the QP on the returned ID, and choose connection parameters/security policy before accepting. | -| `rdma_ibverbs::endpoint` and `ibverbs_backend` | Provide a reference provider-backed endpoint/backend path when enabled. | Deploy with compatible libibverbs/RDMA-Core provider, permissions, device configuration, and CQ pump lifecycle. | +| `rdma_ibverbs::endpoint` and `ibverbs_backend` | Provide a reference provider-backed endpoint/backend path when enabled. `endpoint::shutdown()` cancels and asynchronously joins its CQ pump before releasing the QP, CQ, completion channel, and PD. Starting shutdown makes the endpoint terminal: data-path accessors and pump start reject further use, while a serialized shutdown retry is allowed. Checked provider failures while releasing QP/CQ/channel/PD are reported and the failed resource plus its dependencies remain owned for that retry. `abandon_outstanding()` is a non-waiting terminal escape hatch for eager `.start()` operations that never installed a waiter: `true` confirms checked QP destruction before the remaining verbs resources are intentionally relinquished; `false` retains the QP and dependencies. It does not make suspended operation frames revocable. The destructor does not wait for the CQ pump; an active pump or QP-destruction failure causes teardown to fail closed by relinquishing the affected ownership graph. | Deploy with compatible libibverbs/RDMA-Core provider, permissions, and device configuration. Ensure every posted operation and its awaiter has completed, destroy memory regions registered from the endpoint PD, serialize shutdown with endpoint use, keep the endpoint and the scheduler supplied to `start_cq_pump()` operational through `co_await endpoint::shutdown()`, and prefer that graceful path over destructor or forced-scheduler-shutdown fallback. If a fatal path cannot drain provider-owned eager `.start()` operations, call `abandon_outstanding()` while those never-awaited operation objects, MRs, and payload buffers are still alive; only a `true` result permits unwinding them. On `false`, keep all affected lifetimes intact, repair provider state through raw handles, and retry or terminate. Never use it to justify destroying a suspended operation frame; once delivery may have snapshotted a waiter, resume that coroutine and consume the result. | | `rdma_cuda::gpu_buffer` and `gpu_memory_region` | Expose CUDA-specific buffer and registration helpers when the feature is enabled. | Ensure CUDA context, device pointer validity, registration support, peer access, and driver/runtime compatibility. | ## Logging And Debugging diff --git a/wiki/Examples.md b/wiki/Examples.md index 8f536a03..e3f86489 100644 --- a/wiki/Examples.md +++ b/wiki/Examples.md @@ -934,15 +934,24 @@ elio::coro::task client(elio::runtime::scheduler& sched) { {.max_send_wr = 4, .max_recv_wr = 4}); ep.start_cq_pump(sched); - auto req_mr = ep.register_buffer(req_buf, sizeof(req_buf), IBV_ACCESS_LOCAL_WRITE); - auto resp_mr = ep.register_buffer(resp_buf, sizeof(resp_buf), - IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE); - - // Post recv for OOB notify BEFORE sending the request. - auto notify_aw = ep.conn().recv(notify_mr.view()).start(); - co_await ep.conn().send(req_mr.view(0, sizeof(request_header))); - auto wc = co_await std::move(notify_aw); - // wc.imm_data carries the response length + { + auto req_mr = ep.register_buffer( + req_buf, sizeof(req_buf), IBV_ACCESS_LOCAL_WRITE); + auto resp_mr = ep.register_buffer( + resp_buf, sizeof(resp_buf), + IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE); + std::uint32_t notify_buf{}; + auto notify_mr = ep.register_buffer( + ¬ify_buf, sizeof(notify_buf), IBV_ACCESS_LOCAL_WRITE); + + // Post recv for OOB notify BEFORE sending the request. + auto notify_aw = ep.conn().recv(notify_mr.view()).start(); + co_await ep.conn().send(req_mr.view(0, sizeof(request_header))); + auto wc = co_await std::move(notify_aw); + // wc.imm_data carries the response length + } // Every endpoint-backed MR and awaiter is gone before shutdown. + + co_await ep.shutdown(); } ``` diff --git a/wiki/RDMA-Guide.md b/wiki/RDMA-Guide.md index 76c944d5..953babf7 100644 --- a/wiki/RDMA-Guide.md +++ b/wiki/RDMA-Guide.md @@ -374,8 +374,11 @@ auto ep = co_await elio::rdma_ibverbs::connect( .max_send_wr = 8, .max_recv_wr = 8}); ep.start_cq_pump(sched); -auto mr = ep.register_buffer(buf, len, IBV_ACCESS_LOCAL_WRITE); -auto wc = co_await ep.conn().send(mr.view()); +{ + auto mr = ep.register_buffer(buf, len, IBV_ACCESS_LOCAL_WRITE); + auto wc = co_await ep.conn().send(mr.view()); +} // The completed awaiter and MR are gone before shutdown. +co_await ep.shutdown(); ``` `endpoint` bundles: @@ -387,6 +390,10 @@ auto wc = co_await ep.conn().send(mr.view()); * `register_buffer(...)` returning `memory_region` bound to the endpoint's PD. +The direct `endpoint(rdma_cm::cm_id, ...)` constructor takes ownership of a +resolved CM ID and requires that it does not already have an attached QP; the +wrapper creates and owns that QP itself. + Server side uses `acceptor`: ```cpp @@ -394,6 +401,7 @@ elio::rdma_ibverbs::acceptor ac{cm_ch, bind_addr, len}; auto ep = co_await ac.accept(); // accepts ONE connection ep.start_cq_pump(sched); // ... data path identical to client +co_await ep.shutdown(); ``` For full control over QP attributes, set @@ -401,12 +409,43 @@ For full control over QP attributes, set the wrapper hands it to `rdma_create_qp` unchanged (it will still fill `send_cq` / `recv_cq` if you left them null). -**Shutdown contract**: the destructor cancels the cq_pump, destroys -the QP first (its flush CQEs wake the pump), waits up to 1s for -the pump to observe the cancel, then tears down CQ / comp_channel / -PD. If the pump doesn't exit in time (custom drain that blocks -indefinitely) the verbs resources are intentionally leaked rather -than risk a use-after-free. +**Shutdown contract**: while the endpoint is still alive, explicitly await +`ep.shutdown()`. It requests CQ-pump cancellation, joins the pump asynchronously, +and then releases the QP, CQ, completion channel, and PD: + +```cpp +// Let every memory_region registered from ep leave scope first. +co_await ep.shutdown(); +``` + +The endpoint destructor does not wait for the CQ pump, so it does not block a +scheduler worker on pump completion. It requests cancellation and attempts +checked QP destruction. If QP destruction fails, it relinquishes the whole +dependency graph, including the CM ID; if destruction races an active pump +after the QP is gone, it intentionally leaks the remaining verbs resources and +dispatcher rather than wait or risk a use-after-free. +Before `shutdown()`, ensure every posted operation and its awaiter has completed +and destroy all memory regions registered from the endpoint PD. Serialize the +call with endpoint use, and keep both the endpoint and the scheduler supplied to +`start_cq_pump()` operational until it completes. `scheduler::shutdown_force()` +can orphan in-flight I/O and is not a substitute for endpoint shutdown. Once +`shutdown()` starts executing, the endpoint is terminal: do not restart its pump +or use its connection/data-path accessors. A provider failure while destroying +the QP, CQ, completion channel, or PD is reported; the failed resource and its +dependencies remain owned so a later serialized `shutdown()` can retry. + +If a fatal error leaves provider-owned operations that were launched eagerly +with `.start()` and never awaited, call `ep.abandon_outstanding()` while those +eager awaitables, memory regions, and payload buffers are still alive. A `true` +result confirms checked QP destruction, requests pump stop, and intentionally +relinquishes the CQ, completion channel, PD, and dispatcher; only then can the +eager objects unwind under the posted/orphaned lifecycle. A `false` result +retains the QP and dependencies: keep every affected lifetime intact, repair +provider state through the raw accessors, and retry or terminate. This +fail-closed path does not wait, leaks resources after success, and makes the +endpoint terminal. It does not make a suspended operation frame safe to +destroy: if delivery may have snapshotted its waiter, resume it and consume the +result. The method is not a substitute for normal `shutdown()`. Worked example: `examples/rdma_req_resp_ibverbs.cpp` runs a client / server in one process — client SEND request → server