Skip to content

fix(rdma): v2.6.1 hardening — revert unilateral, fix MR leak, drop per-request logs #267

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ketor merged 1 commit into from
Aug 6, 2026
Merged

fix(rdma): v2.6.1 hardening — revert unilateral, fix MR leak, drop per-request logs #267

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 3 additions & 69 deletions src/cache/rdma_server.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -302,7 +302,7 @@ int ServerIdleMs() {
// the client re-dials a stale pooled connection via RdmaTransport's 2-attempt
// retry. Default 10 min keeps active/recently-used pooled conns alive; set
// DFKV_RDMA_IDLE_MS=0 disables the reaper and waits indefinitely.
int out = 30000; // 30 seconds (was 10 min; Mooncake has no idle timeout but uses SIEVE eviction)
int out = 600000; // 10 min; K8s launcher exports DFKV_RDMA_IDLE_MS=30000 explicitly
const char* e = std::getenv("DFKV_RDMA_IDLE_MS");
if (e && *e) {
long v = std::strtol(e, nullptr, 10);
Expand Down Expand Up @@ -543,7 +543,6 @@ void RdmaServer::Serve(int boot_fd) {
request->recv_slot = recv_slot;
request->data_slot = recv_slot;
request->recv_bytes = completion.byte_len;
DFKV_LOG_INFO("rdma: recv slot=" + std::to_string(recv_slot) + " byte_len=" + std::to_string(completion.byte_len) + " opcode=" + std::to_string(completion.opcode) + " has_imm=" + std::to_string(has_immediate));
if (write_imm_recv) {
const size_t data_slot = static_cast<size_t>(ntohl(completion.imm_data));
if (data_slot >= K || completion.byte_len < kReqPrefix) return false;
Expand Down Expand Up @@ -578,73 +577,8 @@ void RdmaServer::Serve(int boot_fd) {
}
return request->get.targets.size() <= rdma::kV2MaxGetTargets;
}
// Unilateral PUT path: the notification is a 50B request prefix sent via
// SEND. The offset field carries the data_slot index where [header|payload]
// was written via plain RDMA WRITE to the shared receive segment.
if (request->fields.op == static_cast<uint8_t>(WireOp::kCache)) {
if (completion.byte_len < kReqPrefix) return false;
const size_t data_slot = static_cast<size_t>(request->fields.offset);
if (data_slot >= K) return false;
const char* data_frame = recv_lease.data() + data_slot * slot_size +
rdma::kV2PutPrefixOffset;
DFKV_LOG_INFO("rdma: unilateral PUT data_slot=" + std::to_string(data_slot) +
" slot_size=" + std::to_string(slot_size) +
" data_frame_ver=" + std::to_string(static_cast<int>(static_cast<unsigned char>(data_frame[0]))) +
" data_frame_op=" + std::to_string(static_cast<int>(static_cast<unsigned char>(data_frame[1]))) +
" payload_len=" + std::to_string(*reinterpret_cast<const uint64_t*>(data_frame + 42)));
if (!DecodeReqVersion(
data_frame, kNativeProtoRdmaV2, &request->fields,
static_cast<uint64_t>(logical_data_cap)) ||
request->fields.op != static_cast<uint8_t>(WireOp::kCache) ||
request->fields.payload_len > static_cast<uint64_t>(logical_data_cap)) {
return false;
}
request->data_slot = data_slot;
request->contiguous_payload = data_frame + kReqPrefix;
v2_put_writes_.fetch_add(1, std::memory_order_relaxed);
return true;
}
// Unilateral PUT path: the notification is a 50B request prefix sent via
// SEND on the control QP. The offset field carries the data_slot index
// where [header|payload] was written via plain RDMA WRITE to the shared
// receive segment. nbuf_ is separate from sbuf_ so no race.
if (request->fields.op == static_cast<uint8_t>(WireOp::kCache)) {
if (completion.byte_len < kReqPrefix) return false;
const size_t data_slot = static_cast<size_t>(request->fields.offset);
if (data_slot >= K) return false;
const char* data_frame = recv_lease.data() + data_slot * slot_size +
rdma::kV2PutPrefixOffset;
if (!DecodeReqVersion(
data_frame, kNativeProtoRdmaV2, &request->fields,
static_cast<uint64_t>(logical_data_cap)) ||
request->fields.op != static_cast<uint8_t>(WireOp::kCache) ||
request->fields.payload_len > static_cast<uint64_t>(logical_data_cap)) {
return false;
}
request->data_slot = data_slot;
request->contiguous_payload = data_frame + kReqPrefix;
v2_put_writes_.fetch_add(1, std::memory_order_relaxed);
return true;
}
// Unilateral PUT: notification SEND carries offset=data_slot.
// Data was written via unsignaled WRITE to recv segment slot.
if (request->fields.op == static_cast<uint8_t>(WireOp::kCache)) {
const size_t data_slot = static_cast<size_t>(request->fields.offset);
if (data_slot >= K) return false;
const char* data_frame = recv_lease.data() + data_slot * slot_size +
rdma::kV2PutPrefixOffset;
if (!DecodeReqVersion(
data_frame, kNativeProtoRdmaV2, &request->fields,
static_cast<uint64_t>(logical_data_cap)) ||
request->fields.op != static_cast<uint8_t>(WireOp::kCache) ||
request->fields.payload_len > static_cast<uint64_t>(logical_data_cap)) {
return false;
}
request->data_slot = data_slot;
request->contiguous_payload = data_frame + kReqPrefix;
v2_put_writes_.fetch_add(1, std::memory_order_relaxed);
return true;
}


// Other control ops (Exist/Remove/Members/Lookup) with inline payload
if (completion.byte_len <
kReqPrefix + request->fields.payload_len) {
Expand Down
8 changes: 3 additions & 5 deletions src/transport/rdma_transport.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -907,12 +907,10 @@ Status RdmaTransport::RoundTrip(const std::string& node, WireOp op,
return Status::kIOError;
}
conn->Encode(ep.sbuf(0), op, key, offset, length, payload_len);
std::memcpy(ep.nbuf(0), ep.sbuf(0), kReqPrefix);
ok = ep.PostWriteScatter(
ok = ep.PostRecv(0) &&
ep.PostWriteImmScatter(
0, kReqPrefix, payload, static_cast<size_t>(payload_len),
payload_mr, conn->put_addr(0), conn->recv_segment.rkey);
if (ok)
ok = ep.PostRecv(0) && ep.PostSendNotify(0, kReqPrefix);
payload_mr, conn->put_addr(0), conn->recv_segment.rkey, 0);
if (ok)
v2_put_writes_.fetch_add(1, std::memory_order_relaxed);
} else if (op == WireOp::kRange) {
Expand Down
1 change: 0 additions & 1 deletion src/transport/rdma_transport.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -209,7 +209,6 @@ class RdmaTransport : public Transport {
std::atomic<uint64_t> numa_caller_unknown_fallbacks_{0};
std::atomic<uint64_t> numa_no_local_fallbacks_{0};
// #1: async CQ reaper for high-concurrency batch operations
bool reaper_enabled_ = false;
};

} // namespace dfkv
Expand Down
56 changes: 1 addition & 55 deletions src/transport/rdma_verbs.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -273,10 +273,8 @@ void RcEndpoint::Close() {
// only after the last endpoint that can still return it has gone idle/closed.
for (const auto& pool : pool_mr_) SharedReleasePoolMr(ctx_, pool.mr);
pool_mr_.clear();
smr_.clear(); rmr_.clear(); dmr_.clear(); nmr_.clear(); nbuf_.clear();
smr_.clear(); rmr_.clear(); dmr_.clear();
for (auto* b : sbuf_) delete[] b;
for (auto* m : nmr_) if (m) ibv_dereg_mr(m);
for (auto* b : nbuf_) delete[] b;
for (auto* b : rbuf_) delete[] b;
for (auto* b : dbuf_) std::free(b);
sbuf_.clear(); rbuf_.clear(); dbuf_.clear();
Expand Down Expand Up @@ -414,7 +412,6 @@ bool RcEndpoint::Open(const char* dev_name, size_t cap, size_t depth,
// sizes (>=128 KiB) new[] is mmap-backed = page-aligned, which mbind needs.
numa_node_ = numa::DeviceNode(dev_name);
sbuf_.resize(depth_, nullptr); rbuf_.resize(depth_, nullptr);
nbuf_.resize(depth_, nullptr); nmr_.resize(depth_, nullptr);
smr_.resize(depth_, nullptr); rmr_.resize(depth_, nullptr);
if (direct_io_buffers) {
const size_t dio_cap = direct_io_cap ? direct_io_cap : cap_;
Expand All @@ -439,9 +436,6 @@ bool RcEndpoint::Open(const char* dev_name, size_t cap, size_t depth,
}
}
for (size_t i = 0; i < depth_; ++i) {
nbuf_[i] = new char[64];
nmr_[i] = ibv_reg_mr(pd_, nbuf_[i], 64, IBV_ACCESS_LOCAL_WRITE);
if (!nmr_[i]) { Close(); return false; }
}

// local addressing info
Expand Down Expand Up @@ -897,21 +891,6 @@ bool RcEndpoint::PostWriteScatter(
return ibv_post_send(qp_, &wr, &bad) == 0;
}

bool RcEndpoint::PostSendNotify(size_t slot, size_t len) {
if (slot >= depth_ || len > 64 || !nbuf_[slot] || !nmr_[slot]) return false;
ibv_sge sge{};
sge.addr = reinterpret_cast<uintptr_t>(nbuf_[slot]);
sge.length = static_cast<uint32_t>(len);
sge.lkey = nmr_[slot]->lkey;
ibv_send_wr wr{}, *bad = nullptr;
wr.wr_id = slot;
wr.sg_list = &sge;
wr.num_sge = 1;
wr.opcode = IBV_WR_SEND;
wr.send_flags = IBV_SEND_SIGNALED;
return ibv_post_send(qp_, &wr, &bad) == 0;
}

bool RcEndpoint::PostWriteImmScatterMulti(
size_t slot, size_t header_len,
const std::vector<std::pair<const void*, uint32_t>>& segs,
Expand Down Expand Up @@ -1005,38 +984,5 @@ void RcEndpoint::Wake() {
}


void RcEndpoint::StartReaper(std::vector<std::atomic<uint32_t>*>* slots) {
reap_slots_ = *slots;
reap_stop_.store(false, std::memory_order_relaxed);
reap_thread_ = new std::thread([this] {
std::vector<ibv_wc> wcs(64);
while (!reap_stop_.load(std::memory_order_relaxed)) {
int got = ibv_poll_cq(cq_, static_cast<int>(wcs.size()), wcs.data());
if (got > 0) {
for (int i = 0; i < got; ++i) {
size_t slot = static_cast<size_t>(wcs[i].wr_id);
if (slot < reap_slots_.size() && reap_slots_[slot]) {
// Store byte_len for RECV, mark done for SEND
uint32_t val = (wcs[i].opcode == IBV_WC_RECV) ? wcs[i].byte_len : 0xFFFFFFFF;
reap_slots_[slot]->store(val, std::memory_order_release);
}
}
} else if (got == 0) {
// No completion: brief yield to avoid 100% CPU
std::this_thread::yield();
}
}
});
}

void RcEndpoint::StopReaper() {
if (!reap_thread_) return;
reap_stop_.store(true, std::memory_order_relaxed);
reap_thread_->join();
delete reap_thread_;
reap_thread_ = nullptr;
reap_slots_.clear();
}

} // namespace rdma
} // namespace dfkv
12 changes: 0 additions & 12 deletions src/transport/rdma_verbs.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -129,7 +129,6 @@ class RcEndpoint {
int numa_node() const { return numa_node_; } // device's NUMA node, or -1
char* sbuf(size_t slot) { return sbuf_[slot]; }
char* rbuf(size_t slot) { return rbuf_[slot]; }
char* nbuf(size_t slot) { return nbuf_[slot]; }
char* dbuf(size_t slot) { return dbuf_.empty() ? nullptr : dbuf_[slot]; }
size_t dbuf_cap() const { return dbuf_cap_; }
ibv_mr* dmr(size_t slot) { return dmr_.empty() ? nullptr : dmr_[slot]; }
Expand Down Expand Up @@ -248,8 +247,6 @@ class RcEndpoint {
size_t slot, size_t header_len, const void* payload,
size_t payload_len, ibv_mr* payload_mr, uint64_t remote_addr,
uint32_t remote_rkey);
// Signaled SEND from nbuf_[slot] (notification buffer).
bool PostSendNotify(size_t slot, size_t len);

// QP scatter-gather capability negotiated in Open() = min(kMaxSge, device cap).
// The SG datapath caps raw-payload segments at max_sge()-1 (SGE0 is the wire prefix).
Expand All @@ -271,8 +268,6 @@ class RcEndpoint {
// #1: Start a background CQ reaper that polls ibv_poll_cq in a tight
// loop and fills per-slot atomic flags. Eliminates CQ channel
// syscall and CQ contention at high thread counts.
void StartReaper(std::vector<std::atomic<uint32_t>*>* slots);
void StopReaper();

private:
void Close();
Expand All @@ -290,19 +285,12 @@ class RcEndpoint {
unsigned cq_armed_unacked_ = 0;
bool busy_poll_ = false;
size_t num_qp_ = 1;
// #1: async CQ reaper support. Per-slot atomic completion flags
// filled by a background reaper thread, checked by ReapPosted.
std::vector<std::atomic<uint32_t>*> reap_slots_;
std::atomic<bool> reap_stop_{false};
std::thread* reap_thread_ = nullptr;

size_t cap_ = 0, depth_ = 0, dbuf_cap_ = 0;
uint16_t remote_depth_ = 0; // Set from the required v2 QP advertisement.
size_t max_sge_ = 2; // QP max_send_sge/max_recv_sge = min(kMaxSge, device cap)
std::vector<char*> sbuf_, rbuf_, dbuf_;
std::vector<ibv_mr*> smr_, rmr_, dmr_;
std::vector<char*> nbuf_;
std::vector<ibv_mr*> nmr_;
// Big pre-registered caller regions (the host KV pool). Each cache entry owns
// one shared-registry reference. Successful growth replaces the idle
// endpoint's older same-base generation; endpoints with an in-flight user
Expand Down
Loading