Skip to content

[release/1.5.2] libfabric: serialize all endpoint/CQ/AV calls with the rail EP lock #2542

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
nv-nmailhot merged 1 commit into from
Oct 28, 2026
Merged

[release/1.5.2] libfabric: serialize all endpoint/CQ/AV calls with the rail EP lock #2542

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
84 changes: 56 additions & 28 deletions src/utils/libfabric/libfabric_rail.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -786,6 +786,8 @@ nixlLibfabricRail::progressCompletionQueue() {
struct fi_cq_data_entry completions[NIXL_LIBFABRIC_CQ_BATCH_SIZE];

int ret;
struct fi_cq_err_entry err_entry;
int err_ret = 0;

// Only protect libfabric CQ hardware operations
{
Expand All @@ -795,36 +797,37 @@ nixlLibfabricRail::progressCompletionQueue() {
ret = fi_cq_read(cq, completions, NIXL_LIBFABRIC_CQ_BATCH_SIZE);

if (ret < 0 && ret != -FI_EAGAIN) {
NIXL_ERROR << "fi_cq_read returned error " << ret << " on rail " << rail_id << ": "
<< fi_strerror(-ret);

// Handle error - but be careful about fi_cq_readerr
struct fi_cq_err_entry err_entry;
memset(&err_entry, 0, sizeof(err_entry));
err_ret = fi_cq_readerr(cq, &err_entry, 0);
}
}
// CQ lock released here - completion is now local data

int err_ret = fi_cq_readerr(cq, &err_entry, 0);
if (err_ret > 0) {
NIXL_ERROR << "CQ read failed on rail " << rail_id
<< " with error: " << fi_strerror(err_entry.err)
<< " prov_errno: " << err_entry.prov_errno << " len: " << err_entry.len;

// Notify the owning handle of the error and release the request
if (err_entry.op_context) {
nixlLibfabricReq *req = findRequestFromContext(err_entry.op_context);
if (req && req->in_use) {
if (req->completion_callback) {
req->completion_callback(NIXL_ERR_BACKEND);
}
releaseRequest(req);
if (ret < 0 && ret != -FI_EAGAIN) {
NIXL_ERROR << "fi_cq_read returned error " << ret << " on rail " << rail_id << ": "
<< fi_strerror(-ret);
if (err_ret > 0) {
NIXL_ERROR << "CQ read failed on rail " << rail_id
<< " with error: " << fi_strerror(err_entry.err)
<< " prov_errno: " << err_entry.prov_errno << " len: " << err_entry.len;

// Notify the owning handle of the error and release the request. Done without the
// EP lock: the callback and the request pool take their own locks.
if (err_entry.op_context) {
nixlLibfabricReq *req = findRequestFromContext(err_entry.op_context);
if (req && req->in_use) {
if (req->completion_callback) {
req->completion_callback(NIXL_ERR_BACKEND);
}
releaseRequest(req);
}
} else {
NIXL_ERROR << "fi_cq_readerr failed with " << err_ret;
}
return NIXL_ERR_BACKEND;
} else {
NIXL_ERROR << "fi_cq_readerr failed with " << err_ret;
}
return NIXL_ERR_BACKEND;
}
// CQ lock released here - completion is now local data

if (ret == -FI_EAGAIN) {
// No completions - fall through to drainPostQueue
Expand Down Expand Up @@ -859,14 +862,20 @@ nixlLibfabricRail::progressCompletionQueue() {
void
nixlLibfabricRail::pollForCompletions() {
struct fi_cq_data_entry cq_buf[16];
int cq_ret = fi_cq_read(cq, cq_buf, 16);
struct fi_cq_err_entry err_entry = {};
int cq_ret;
{
const std::lock_guard<std::mutex> ep_lock(ep_mutex_);
cq_ret = fi_cq_read(cq, cq_buf, 16);
if (cq_ret < 0 && cq_ret != -FI_EAGAIN) {
fi_cq_readerr(cq, &err_entry, 0);
}
}
if (cq_ret > 0) {
for (int c = 0; c < cq_ret; c++) {
processCompletionQueueEntry(&cq_buf[c]);
}
} else if (cq_ret < 0 && cq_ret != -FI_EAGAIN) {
struct fi_cq_err_entry err_entry = {};
fi_cq_readerr(cq, &err_entry, 0);
NIXL_ERROR << "CQ error in drain interleave on rail " << rail_id << ": "
<< fi_strerror(err_entry.err);
}
Expand Down Expand Up @@ -1305,6 +1314,7 @@ nixlLibfabricRail::drainPostQueue() {
msg.context = &pr.req->ctx;
msg.data = pr.immediate_data;

const std::lock_guard<std::mutex> ep_lock(ep_mutex_);
ret = fi_writemsg(endpoint, &msg, pr.fi_flags | FI_REMOTE_CQ_DATA);
} else {
struct iovec iov{};
Expand All @@ -1326,6 +1336,7 @@ nixlLibfabricRail::drainPostQueue() {
msg.rma_iov_count = 1;
msg.context = &pr.req->ctx;

const std::lock_guard<std::mutex> ep_lock(ep_mutex_);
ret = fi_readmsg(endpoint, &msg, pr.fi_flags);
}

Expand Down Expand Up @@ -1629,6 +1640,8 @@ nixlLibfabricRail::registerMemory(void *buffer,
mr_attr.mr_iov = &iov;
mr_attr.iov_count = 1;

// fi_mr_regattr is a domain-level call, not tied to the endpoint or CQ, so it runs
// without the EP lock.
int ret = fi_mr_regattr(domain, &mr_attr, 0, &mr);
if (ret) {
NIXL_ERROR << "fi_mr_reg failed on rail " << rail_id << ": " << fi_strerror(-ret)
Expand All @@ -1637,7 +1650,9 @@ nixlLibfabricRail::registerMemory(void *buffer,
return NIXL_ERR_BACKEND;
}

// Only an MR bound to the endpoint (FI_MR_ENDPOINT) needs the lock.
if (info->domain_attr->mr_mode & FI_MR_ENDPOINT) {
const std::lock_guard<std::mutex> ep_lock(ep_mutex_);
ret = fi_mr_bind(mr, &endpoint->fid, 0);
if (ret) {
NIXL_ERROR << "fi_mr_bind failed on rail " << rail_id << ": " << fi_strerror(-ret);
Expand All @@ -1656,7 +1671,7 @@ nixlLibfabricRail::registerMemory(void *buffer,
uint64_t key = fi_mr_key(mr);
if (key == FI_KEY_NOTAVAIL) {
NIXL_ERROR << "fi_mr_key returned FI_KEY_NOTAVAIL on rail " << rail_id;
fi_close(&mr->fid);
deregisterMemory(mr);
return NIXL_ERR_BACKEND;
} else {
NIXL_TRACE << "MR key obtained: " << key << " (" << std::hex << key << std::dec << ")";
Expand All @@ -1679,6 +1694,11 @@ nixlLibfabricRail::deregisterMemory(struct fid_mr *mr) const {
return NIXL_ERR_INVALID_PARAM;
}

// Same rule as registerMemory: only an MR bound to the endpoint needs the EP lock.
std::unique_lock<std::mutex> ep_lock(ep_mutex_, std::defer_lock);
if (info->domain_attr->mr_mode & FI_MR_ENDPOINT) {
ep_lock.lock();
}
int ret = fi_close(&mr->fid);
if (ret) {
NIXL_ERROR << "fi_close failed on rail " << rail_id << ": " << fi_strerror(-ret);
Expand All @@ -1701,7 +1721,11 @@ nixlLibfabricRail::insertAddress(const void *addr, fi_addr_t *fi_addr_out) const
return NIXL_ERR_BACKEND;
}

int ret = fi_av_insert(av, addr, 1, fi_addr_out, 0, NULL);
int ret;
{
const std::lock_guard<std::mutex> ep_lock(ep_mutex_);
ret = fi_av_insert(av, addr, 1, fi_addr_out, 0, NULL);
}
if (ret != 1) {
NIXL_ERROR << "fi_av_insert failed on rail " << rail_id << ": " << fi_strerror(-ret);
return NIXL_ERR_BACKEND;
Expand All @@ -1721,7 +1745,11 @@ nixlLibfabricRail::removeAddress(fi_addr_t fi_addr) const {
return NIXL_ERR_BACKEND;
}

int ret = fi_av_remove(av, &fi_addr, 1, 0);
int ret;
{
const std::lock_guard<std::mutex> ep_lock(ep_mutex_);
ret = fi_av_remove(av, &fi_addr, 1, 0);
}
if (ret != 0) {
NIXL_ERROR << "fi_av_remove failed on rail " << rail_id << ": " << fi_strerror(-ret);
return NIXL_ERR_BACKEND;
Expand Down
10 changes: 9 additions & 1 deletion src/utils/libfabric/libfabric_rail.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -576,7 +576,15 @@ class nixlLibfabricRail {
struct fid_cq *cq; // from rail_cqs[rail_id]
struct fid_av *av; // from rail_avs[rail_id]

// EP mutex to protect endpoint and CQ operations
// Serializes every libfabric call on this rail's endpoint, CQ and AV. The rail opens its
// domain with FI_THREAD_COMPLETION, under which the provider takes no locks for these
// objects and the application must serialize them. With the progress thread enabled the
// PT posts data transfers and reads the CQ while the caller's thread still posts control
// messages (postSend), so both paths must hold this lock.
// - Hold it only around the fi_* call itself, never around an FI_EAGAIN retry loop:
// the thread that reaps completions (and so clears EAGAIN) needs the lock too.
// - Keep it innermost: never run completion handling or callbacks while holding it,
// they take other locks and may re-enter (e.g. postRecv from a recv completion).
mutable std::mutex ep_mutex_;

// Lock-free MPSC ring: main thread pushes, PT pops and posts
Expand Down
Loading