diff --git a/src/core/ib.cc b/src/core/ib.cc index 5e3be071..bd34321c 100644 --- a/src/core/ib.cc +++ b/src/core/ib.cc @@ -188,6 +188,7 @@ IbQp::IbQp(ibv_context* ctx, ibv_pd* pd, int portNum, int gidIndex, int maxSendC numStagedRecv_(0), numPostedSignaledSend_(0), numStagedSignaledSend_(0), + maxSendCqSize_(maxSendCqSize), maxSendCqPollNum_(maxSendCqPollNum), maxSendWr_(maxSendWr), maxWrPerSend_(maxWrPerSend), @@ -402,9 +403,9 @@ void IbQp::postSend() { numStagedSend_ = 0; numPostedSignaledSend_ += numStagedSignaledSend_; numStagedSignaledSend_ = 0; - if (numPostedSignaledSend_ + 4 > sendCq_->cqe) { - WARN(NET, "IB: CQ is almost full ( ", numPostedSignaledSend_, " / ", sendCq_->cqe, - " ). The connection needs to be flushed to prevent timeout errors."); + if (numPostedSignaledSend_ + 4 > maxSendCqSize_) { + WARN(NET, "IB: CQ is almost full (", numPostedSignaledSend_, " / ", maxSendCqSize_, + "). The connection needs to be flushed to prevent timeout errors."); } } diff --git a/src/core/include/ib.hpp b/src/core/include/ib.hpp index 36c5a237..31a48a31 100644 --- a/src/core/include/ib.hpp +++ b/src/core/include/ib.hpp @@ -124,6 +124,7 @@ class IbQp { int numPostedSignaledSend_; int numStagedSignaledSend_; + const int maxSendCqSize_; const int maxSendCqPollNum_; const int maxSendWr_; const int maxWrPerSend_; diff --git a/src/core/port_channel.cc b/src/core/port_channel.cc index 210b2eeb..0601ef84 100644 --- a/src/core/port_channel.cc +++ b/src/core/port_channel.cc @@ -90,8 +90,8 @@ MSCCLPP_API_CPP void ProxyService::startProxy(bool blocking) { proxy_->start(blo MSCCLPP_API_CPP void ProxyService::stopProxy() { proxy_->stop(); - // Drain pending flushes. After a bounded loop, force-unblock any still-pending GPU - // waiters with a sentinel write (UINT64_MAX > any FIFO position). + // Drain pending TriggerSync flushes. After a bounded loop, force-unblock any still-pending + // GPU waiters with a sentinel write (UINT64_MAX > any FIFO position). for (int i = 0; i < 1000 && !pendingFlushPos_.empty(); ++i) { progressFlushes(); } @@ -102,6 +102,16 @@ MSCCLPP_API_CPP void ProxyService::stopProxy() { } pendingFlushPos_.clear(); } + // Drain any remaining in-flight signaled posts that the proxy issued but never sync-flushed. + for (auto& [conn, count] : inflightRequests_) { + if (count <= 0) continue; + try { + conn->flush(/*timeoutUsec=*/5'000'000); + } catch (const std::exception& e) { + WARN(CONN, "stopProxy: flush failed for a connection (continuing): ", e.what()); + } + } + inflightRequests_.clear(); } ProxyHandlerResult ProxyService::handleTrigger(ProxyTrigger trigger) {