IB CI fixes: correct CQ-full warning bound, drain in-flight posts on stopProxy, expand multi-QP perf tests

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.");
   }
 }
 

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_;

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) {