[Bugfix][TENT] Drain the async event queue and recover contexts whose PORT_ACTIVE was lost

mooncake-transfer-engine/tent/include/tent/transport/rdma/context.h

@@ -123,6 +123,15 @@ class RdmaContext {
     // open or the query fails.
     int refreshPortAttributes();
 
+    // Read the port's current state from the hardware. The monitor thread
+    // polls this for paused contexts so a link that came back without a
+    // usable IBV_EVENT_PORT_ACTIVE is still noticed. Deliberately leaves the
+    // cached speed/width alone: refreshLinkSpeed() diffs against them, so
+    // refreshing here would hide a renegotiated rate from the selector.
+    // Returns -1 and leaves *state untouched if no device is open or the
+    // query fails.
+    int queryPortState(ibv_port_state *state) const;
+
     int eventFd() const { return event_fd_; }
 
     RdmaCQ *cq(int index);

mooncake-transfer-engine/tent/include/tent/transport/rdma/workers.h

@@ -84,6 +84,9 @@ class Workers {
 
     // dev_id is the NicID (context_set_ index), which is also the
     // DeviceSelector id, so port events can flip that device's availability.
+    // Drains the whole async event queue: the async fd is edge-triggered and
+    // ibv_get_async_event() dequeues one record per call, so an event left
+    // behind waits for an unrelated later event to release it.
     int handleContextEvents(int dev_id, std::shared_ptr<RdmaContext>& context);
 
     // The decision half of handleContextEvents, kept apart from
@@ -94,11 +97,24 @@ class Workers {
     void applyContextEvent(int dev_id, RdmaContext& context,
                            const ibv_async_event& event);
 
+    // Everything a recovered port needs: resume the context, re-seed its
+    // bandwidth and make it selectable again. Shared by the
+    // IBV_EVENT_PORT_ACTIVE path and by resumePausedContexts().
+    void activateContext(int dev_id, RdmaContext& context);
+
     // Re-read the link speed after a port event and re-seed the selector if
     // it changed; a link that returns at the same speed keeps what it
     // learned.
     void refreshLinkSpeed(int dev_id, RdmaContext& context);
 
+    // 1 Hz heartbeat from monitorThread(): safety net for a lost
+    // IBV_EVENT_PORT_ACTIVE. Nothing else ever leaves DEVICE_PAUSED, so a
+    // context whose recovery event was dropped (edge-triggered fd, event
+    // queue overflow, ...) would fail every transfer on that NIC forever.
+    // Polls the port state of paused contexts and activates the ones the
+    // hardware reports as up.
+    void resumePausedContexts();
+
     Status generatePostPath(RdmaSlice* slice);
 
    private:

mooncake-transfer-engine/tent/src/transport/rdma/context.cpp

@@ -880,6 +880,21 @@ int RdmaContext::refreshPortAttributes() {
     return 0;
 }
 
+int RdmaContext::queryPortState(ibv_port_state* state) const {
+    if (!state || !native_context_ || !params_) return -1;
+    ibv_port_attr port_attr;
+    if (verbs_.ibv_query_port_default(native_context_, params_->device.port,
+                                      &port_attr)) {
+        PLOG(WARNING) << "Failed to query state of port "
+                      << static_cast<int>(params_->device.port) << " on "
+                      << device_name_;
+        return -1;
+    }
+    // No recordPortSpeed() here on purpose -- see the header.
+    *state = port_attr.state;
+    return 0;
+}
+
 double RdmaContext::linkSpeedGbps() const {
     return ibLinkSpeedGbps(active_speed_.load(std::memory_order_relaxed),
                            active_width_.load(std::memory_order_relaxed));

mooncake-transfer-engine/tent/src/transport/rdma/workers.cpp

@@ -20,6 +20,7 @@
 
 #include <algorithm>
 #include <cassert>
+#include <cerrno>
 #include <fstream>
 #include <sstream>
 
@@ -861,14 +862,35 @@ void Workers::workerThread(int thread_id) {
 
 int Workers::handleContextEvents(int dev_id,
                                  std::shared_ptr<RdmaContext>& context) {
-    ibv_async_event event;
-    if (ibv_get_async_event(context->nativeContext(), &event) < 0) return -1;
-    LOG(WARNING) << "Received context async event "
-                 << ibv_event_type_str(event.event_type) << " for context "
-                 << context->name();
-    applyContextEvent(dev_id, *context, event);
-    ibv_ack_async_event(&event);
-    return 0;
+    // The async fd is non-blocking and edge-triggered
+    // (joinNonblockingPollList), and ibv_get_async_event() dequeues one record
+    // per call, so every queued event has to be consumed here: epoll only
+    // reports readiness again once a *new* event arrives. Bursts are routine
+    // (IBV_EVENT_COMM_EST fires once per connection), and a PORT_ACTIVE
+    // stranded behind one keeps the context paused until some unrelated event
+    // happens to release it -- which may be never.
+    while (true) {
+        ibv_async_event event;
+        errno = 0;
+        if (ibv_get_async_event(context->nativeContext(), &event) < 0) {
+            if (errno == EAGAIN || errno == EWOULDBLOCK) return 0;  // drained
+            if (errno == EINTR) continue;
+            PLOG(ERROR) << "ibv_get_async_event for context "
+                        << context->name();
+            return -1;
+        }
+        if (event.event_type == IBV_EVENT_COMM_EST) {
+            VLOG(1) << "Received context async event "
+                    << ibv_event_type_str(event.event_type)
+                    << " for context " << context->name();
+        } else {
+            LOG(WARNING) << "Received context async event "
+                         << ibv_event_type_str(event.event_type)
+                         << " for context " << context->name();
+        }
+        applyContextEvent(dev_id, *context, event);
+        ibv_ack_async_event(&event);
+    }
 }
 
 void Workers::applyContextEvent(int dev_id, RdmaContext& context,
@@ -909,12 +931,7 @@ void Workers::applyContextEvent(int dev_id, RdmaContext& context,
                 device_selector_->setDeviceAvailable(dev_id, false);
                 LOG(WARNING) << "Action: " << context.name() << " down";
             } else {
-                context.resume();
-                // The link may have renegotiated while down: re-seed before
-                // the device becomes selectable so no worker scores it on
-                // the old rate.
-                refreshLinkSpeed(dev_id, context);
-                device_selector_->setDeviceAvailable(dev_id, true);
+                activateContext(dev_id, context);
                 LOG(WARNING) << "Action: " << context.name() << " up";
             }
             break;
@@ -933,6 +950,14 @@ void Workers::applyContextEvent(int dev_id, RdmaContext& context,
     }
 }
 
+void Workers::activateContext(int dev_id, RdmaContext& context) {
+    context.resume();
+    // The link may have renegotiated while down: re-seed before the device
+    // becomes selectable so no worker scores it on the old rate.
+    refreshLinkSpeed(dev_id, context);
+    if (device_selector_) device_selector_->setDeviceAvailable(dev_id, true);
+}
+
 void Workers::refreshLinkSpeed(int dev_id, RdmaContext& context) {
     if (!device_selector_) return;
     const double before = context.linkSpeedGbps();
@@ -956,6 +981,27 @@ void Workers::reclaimEndpoints() {
     }
 }
 
+void Workers::resumePausedContexts() {
+    for (size_t dev_id = 0; dev_id < transport_->context_set_.size();
+         ++dev_id) {
+        auto& context = transport_->context_set_[dev_id];
+        // Only a paused context is waiting for a recovery event; this also
+        // filters out inert slots, which never leave DEVICE_UNINIT.
+        if (!context || context->status() != RdmaContext::DEVICE_PAUSED)
+            continue;
+        ibv_port_state state;
+        if (context->queryPortState(&state) != 0) continue;  // already logged
+        // Only a fully active port carries traffic. Intermediate states
+        // (INIT/ARMED/ACTIVE_DEFER) mean the link is still settling, so leave
+        // the context paused and re-check on the next tick.
+        if (state != IBV_PORT_ACTIVE) continue;
+        LOG(WARNING) << "Action: " << context->name()
+                     << " up (port reports ACTIVE without an "
+                        "IBV_EVENT_PORT_ACTIVE event)";
+        activateContext(static_cast<int>(dev_id), *context);
+    }
+}
+
 void Workers::monitorThread() {
     // Track time for periodic endpoint reclaim (1 Hz heartbeat)
     auto last_reclaim_time = std::chrono::steady_clock::now();
@@ -971,6 +1017,8 @@ void Workers::monitorThread() {
 
         if (time_since_last_reclaim >= 1000) {  // 1 second = 1000 ms
             reclaimEndpoints();
+            // Safety net for a recovery event that never reached us.
+            resumePausedContexts();
             last_reclaim_time = current_time;
         }
 

mooncake-transfer-engine/tent/tests/CMakeLists.txt

@@ -283,6 +283,26 @@ target_include_directories(tent_rdma_transport_test
                            PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/../include)
 add_test(NAME tent_rdma_transport_test COMMAND tent_rdma_transport_test)
 
+# Regression test for the edge-triggered async event fd: one epoll wakeup must
+# drain the whole queue. Wraps ibv_get_async_event to script the event source,
+# so it needs no RDMA device. -Wl,--wrap is a GNU ld / lld feature that Apple's
+# linker lacks, and the test is meaningless without it, so skip the target
+# there rather than build one that calls the real symbols.
+if(UNIX AND NOT APPLE)
+  add_executable(tent_rdma_async_event_drain_test
+                 rdma_async_event_drain_test.cpp)
+  target_link_libraries(
+    tent_rdma_async_event_drain_test PRIVATE gtest gtest_main tent_link_group
+                                             ibverbs)
+  target_include_directories(tent_rdma_async_event_drain_test
+                             PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/../include)
+  target_link_options(
+    tent_rdma_async_event_drain_test PRIVATE "-Wl,--wrap=ibv_get_async_event"
+    "-Wl,--wrap=ibv_ack_async_event")
+  add_test(NAME tent_rdma_async_event_drain_test
+           COMMAND tent_rdma_async_event_drain_test)
+endif()
+
 if(USE_HIP)
   find_package(HIP REQUIRED)
   add_executable(tent_rocm_platform_test rocm_platform_test.cpp)