mooncake-transfer-engine/include/transport/efa_transport/efa_endpoint.h

@@ -124,6 +124,13 @@ class EfaEndPoint {
     RWSpinlock lock_;  // protects peer_nic_path_ and status_
     std::string peer_nic_path_;
     fi_addr_t peer_fi_addr_;  // slot in context_.av()
+    // Last peer EFA address successfully inserted into the AV.  Used to
+    // make passive handshakes idempotent: when both peers initiate a
+    // handshake concurrently (the common case under bilateral sglang PD
+    // traffic), the second passive handshake carries the same EFA
+    // address as the cached value, and we can skip the fi_av_remove/
+    // fi_av_insert churn entirely.
+    std::string cached_peer_addr_;
 };
 
 }  // namespace mooncake

mooncake-transfer-engine/src/transport/efa_transport/efa_context.cpp

@@ -458,16 +458,31 @@ void* EfaContext::mrDesc(void* addr) {
 
 std::shared_ptr<EfaEndPoint> EfaContext::endpoint(
     const std::string& peer_nic_path) {
-    // Use normalized key (strip port) so the same physical peer reuses its
-    // handle across reconnections.  Each P2PHANDSHAKE run picks a random
-    // port, producing a different peer_nic_path for the same peer host+NIC.
-    std::string key = normalizeNicPath(peer_nic_path);
+    // Key the peer map by the full "host:port@nic" path, not the
+    // port-stripped form.  Under sglang DP>1 every DP worker on a peer
+    // host is a distinct process with its own Mooncake TransferEngine
+    // and its own P2PHANDSHAKE RPC port.  They share the same host+NIC
+    // but map to different EFA QPNs / memory regions.  If we normalize
+    // the port away, every DP worker on that host collapses onto the
+    // same EfaEndPoint slot: each arriving handshake looks like a
+    // "peer reconnected with new address" to the previous holder,
+    // triggering fi_av_remove + fi_av_insert (and an AH warm-up) on
+    // every KV transfer.  At high DP this devolves into permanent
+    // thrashing and eventually "Remote MR invalid" when an in-flight
+    // fi_write runs against a stale AV slot.
+    //
+    // The port is stable for the lifetime of a sglang worker process
+    // (Mooncake only calls initialize() once), so keying by full path
+    // costs nothing in steady state.  A genuine peer restart (process
+    // re-launch → new RPC port) creates a new map entry and leaks the
+    // old EfaEndPoint — that's at most a few bytes per ex-worker and
+    // far cheaper than the churn the old normalization caused.
+    const std::string& key = peer_nic_path;
 
     {
         RWSpinlock::ReadGuard guard(peer_map_lock_);
         auto it = peer_map_.find(key);
         if (it != peer_map_.end()) {
-            it->second->setPeerNicPath(peer_nic_path);
             return it->second;
         }
     }
@@ -478,7 +493,6 @@ std::shared_ptr<EfaEndPoint> EfaContext::endpoint(
     RWSpinlock::WriteGuard guard(peer_map_lock_);
     auto it = peer_map_.find(key);
     if (it != peer_map_.end()) {
-        it->second->setPeerNicPath(peer_nic_path);
         return it->second;
     }
     peer_map_[key] = new_ep;
@@ -488,7 +502,7 @@ std::shared_ptr<EfaEndPoint> EfaContext::endpoint(
 std::shared_ptr<EfaEndPoint> EfaContext::peekEndpoint(
     const std::string& peer_nic_path) {
     RWSpinlock::ReadGuard guard(peer_map_lock_);
-    auto it = peer_map_.find(normalizeNicPath(peer_nic_path));
+    auto it = peer_map_.find(peer_nic_path);
     if (it == peer_map_.end()) return nullptr;
     return it->second;
 }
@@ -497,7 +511,7 @@ int EfaContext::deleteEndpoint(const std::string& peer_nic_path) {
     std::shared_ptr<EfaEndPoint> ep;
     {
         RWSpinlock::WriteGuard guard(peer_map_lock_);
-        auto it = peer_map_.find(normalizeNicPath(peer_nic_path));
+        auto it = peer_map_.find(peer_nic_path);
         if (it == peer_map_.end()) return 0;
         ep = it->second;
         peer_map_.erase(it);
@@ -656,7 +670,7 @@ int EfaContext::submitPostSend(
         // freeing its AV entry for reuse.  Under the shared-endpoint model
         // this is cheap (no fid_ep to destroy).
         if (rc != 0 && !ep->connected()) {
-            deleteEndpoint(normalizeNicPath(peer_nic_path));
+            deleteEndpoint(peer_nic_path);
         }
     }
 

mooncake-transfer-engine/src/transport/efa_transport/efa_endpoint.cpp

@@ -80,6 +80,7 @@ int EfaEndPoint::setupConnectionsByActive() {
 
     rc = context_.insertPeerAddr(peer_desc.efa_addr, peer_fi_addr_);
     if (rc != 0) return rc;
+    cached_peer_addr_ = peer_desc.efa_addr;
 
     status_.store(CONNECTED, std::memory_order_release);
     VLOG(1) << "EFA connection established: " << toString()
@@ -90,10 +91,6 @@ int EfaEndPoint::setupConnectionsByActive() {
 int EfaEndPoint::setupConnectionsByPassive(const HandShakeDesc& peer_desc,
                                            HandShakeDesc& local_desc) {
     RWSpinlock::WriteGuard guard(lock_);
-    if (status_.load(std::memory_order_relaxed) == CONNECTED) {
-        LOG(WARNING) << "Re-establish EFA connection: " << toString();
-        disconnectUnlocked();
-    }
 
     if (peer_desc.peer_nic_path != context_.nicPath() ||
         peer_desc.local_nic_path != peer_nic_path_) {
@@ -112,11 +109,37 @@ int EfaEndPoint::setupConnectionsByPassive(const HandShakeDesc& peer_desc,
         return ERR_REJECT_HANDSHAKE;
     }
 
+    // Classify this passive handshake so we can emit the right log level
+    // without changing functional behavior: the AV reinsert below must
+    // still happen on every handshake because libfabric's EFA provider
+    // tracks per-peer transport state that depends on a fresh
+    // fi_av_insert (e.g. provider-internal AH activation and RNR state).
+    // Skipping reinsert caused request-level stalls under bilateral
+    // sglang P/D load even when the peer EFA address was unchanged.
+    //
+    // Log semantics:
+    //   * Same cached peer address → benign symmetric handshake under
+    //     bilateral traffic.  Demote to INFO to keep decode logs readable
+    //     without hiding real reconnects.
+    //   * Different cached peer address → genuine reconnect (peer
+    //     restart, port reshuffle that reached disconnect first, etc.).
+    //     Keep at WARNING so it stays visible.
+    if (status_.load(std::memory_order_relaxed) == CONNECTED) {
+        if (!cached_peer_addr_.empty() &&
+            peer_desc.efa_addr == cached_peer_addr_) {
+            VLOG(1) << "EFA passive handshake (same peer addr): " << toString();
+        } else {
+            LOG(WARNING) << "Re-establish EFA connection: " << toString();
+        }
+        disconnectUnlocked();
+    }
+
     int ret = context_.insertPeerAddr(peer_desc.efa_addr, peer_fi_addr_);
     if (ret != 0) {
         local_desc.reply_msg = "Failed to insert peer address";
         return ret;
     }
+    cached_peer_addr_ = peer_desc.efa_addr;
 
     local_desc.local_nic_path = context_.nicPath();
     local_desc.peer_nic_path = peer_nic_path_;
@@ -138,12 +161,14 @@ void EfaEndPoint::disconnectUnlocked() {
         context_.removePeerAddr(peer_fi_addr_);
         peer_fi_addr_ = FI_ADDR_UNSPEC;
     }
+    cached_peer_addr_.clear();
     status_.store(UNCONNECTED, std::memory_order_release);
 }
 
 void EfaEndPoint::markDetachedForTeardown() {
     RWSpinlock::WriteGuard guard(lock_);
     peer_fi_addr_ = FI_ADDR_UNSPEC;
+    cached_peer_addr_.clear();
     status_.store(UNCONNECTED, std::memory_order_release);
 }
 
@@ -165,11 +190,26 @@ int EfaEndPoint::submitPostSend(
         }
     }
 
-    fi_addr_t peer;
-    {
-        RWSpinlock::ReadGuard guard(lock_);
-        peer = peer_fi_addr_;
+    // Hold a read lock for the entire submit window so setPeerNicPath() /
+    // disconnect() (which take the write lock) cannot fi_av_remove() our
+    // slot while an fi_write() inside submitSlicesOnPeer is still in
+    // flight.  The previous code latched peer_fi_addr_ under the lock and
+    // released it before calling into the context, leaving a race: a
+    // concurrent peer reconnect could remove the AV entry after the latch,
+    // and libfabric would segfault on the stale fi_addr_t.  Read locks
+    // stack, so multiple senders to the same peer still submit in parallel.
+    RWSpinlock::ReadGuard guard(lock_);
+
+    // Re-check status under the lock — a racing disconnect() could have
+    // flipped us to UNCONNECTED between the outer check and acquiring the
+    // lock.  Fail the batch so the caller retries with a fresh setup.
+    if (status_.load(std::memory_order_acquire) != CONNECTED) {
+        for (auto* slice : slice_list) failed_slice_list.push_back(slice);
+        slice_list.clear();
+        return ERR_ENDPOINT;
     }
+
+    fi_addr_t peer = peer_fi_addr_;
     if (peer == FI_ADDR_UNSPEC) {
         for (auto* slice : slice_list) failed_slice_list.push_back(slice);
         slice_list.clear();

mooncake-transfer-engine/src/transport/efa_transport/efa_transport.cpp

@@ -745,7 +745,7 @@ int EfaTransport::warmupSegment(const std::string& segment_name) {
                         // slot is freed and the next warmup retry starts
                         // clean.  Cheap under the shared-endpoint model —
                         // no fi_endpoint teardown required.
-                        ctx->deleteEndpoint(normalizeNicPath(path));
+                        ctx->deleteEndpoint(path);
                     }
                     return rc;
                 }));