[CRL] Avoid sticky device error state on host buffer registration

flagcx/adaptor/device/cann_adaptor.cc

@@ -368,6 +368,7 @@ struct flagcxDeviceAdaptor cannAdaptor {
       cannAdaptorSymFlatUnmap, cannAdaptorSymMulticastSupported,
       cannAdaptorSymMulticastCreate, cannAdaptorSymMulticastBind,
       cannAdaptorSymMulticastTeardown, cannAdaptorSymMulticastFree,
+      NULL, // flagcxResult_t (*getLastError)();
 };
 
 #endif // USE_ASCEND_ADAPTOR

flagcx/adaptor/device/cuda_adaptor.cc

@@ -930,6 +930,11 @@ flagcxResult_t cudaAdaptorSymMulticastFree(void *) { return flagcxSuccess; }
 
 #endif // CUDART_VERSION >= 12010
 
+flagcxResult_t cudaAdaptorGetLastError() {
+  cudaError_t err = cudaGetLastError();
+  return err == cudaSuccess ? flagcxSuccess : flagcxSystemError;
+}
+
 struct flagcxDeviceAdaptor cudaAdaptor {
   "CUDA",
       // Basic functions
@@ -996,6 +1001,7 @@ struct flagcxDeviceAdaptor cudaAdaptor {
       cudaAdaptorSymFlatUnmap, cudaAdaptorSymMulticastSupported,
       cudaAdaptorSymMulticastCreate, cudaAdaptorSymMulticastBind,
       cudaAdaptorSymMulticastTeardown, cudaAdaptorSymMulticastFree,
+      cudaAdaptorGetLastError,
 };
 
 #endif // USE_NVIDIA_ADAPTOR

flagcx/adaptor/device/ducuda_adaptor.cc

@@ -608,6 +608,7 @@ struct flagcxDeviceAdaptor ducudaAdaptor {
       ducudaAdaptorSymMulticastSupported, ducudaAdaptorSymMulticastCreate,
       ducudaAdaptorSymMulticastBind, ducudaAdaptorSymMulticastTeardown,
       ducudaAdaptorSymMulticastFree,
+      NULL, // flagcxResult_t (*getLastError)();
 };
 
 #endif // USE_DU_ADAPTOR

flagcx/adaptor/device/hip_adaptor.cc

@@ -432,6 +432,7 @@ struct flagcxDeviceAdaptor hipAdaptor {
       hipAdaptorSymFlatUnmap, hipAdaptorSymMulticastSupported,
       hipAdaptorSymMulticastCreate, hipAdaptorSymMulticastBind,
       hipAdaptorSymMulticastTeardown, hipAdaptorSymMulticastFree,
+      NULL, // flagcxResult_t (*getLastError)();
 };
 
 #endif // USE_AMD_ADAPTOR

flagcx/adaptor/device/ixcuda_adaptor.cc

@@ -443,5 +443,6 @@ struct flagcxDeviceAdaptor ixcudaAdaptor {
       ixcudaAdaptorSymMulticastSupported, ixcudaAdaptorSymMulticastCreate,
       ixcudaAdaptorSymMulticastBind, ixcudaAdaptorSymMulticastTeardown,
       ixcudaAdaptorSymMulticastFree,
+      NULL, // flagcxResult_t (*getLastError)();
 };
 #endif // USE_ILUVATAR_COREX_ADAPTOR

flagcx/adaptor/device/kunlunxin_adaptor.cc

@@ -522,5 +522,6 @@ struct flagcxDeviceAdaptor kunlunAdaptor {
       kunlunxinAdaptorSymMulticastSupported, kunlunxinAdaptorSymMulticastCreate,
       kunlunxinAdaptorSymMulticastBind, kunlunxinAdaptorSymMulticastTeardown,
       kunlunxinAdaptorSymMulticastFree,
+      NULL, // flagcxResult_t (*getLastError)();
 };
 #endif // USE_KUNLUNXIN_ADAPTOR

flagcx/adaptor/device/maca_adaptor.cc

@@ -450,6 +450,7 @@ struct flagcxDeviceAdaptor macaAdaptor {
       macaAdaptorSymFlatUnmap, macaAdaptorSymMulticastSupported,
       macaAdaptorSymMulticastCreate, macaAdaptorSymMulticastBind,
       macaAdaptorSymMulticastTeardown, macaAdaptorSymMulticastFree,
+      NULL, // flagcxResult_t (*getLastError)();
 };
 
 #endif // USE_METAX_ADAPTOR

flagcx/adaptor/device/mlu_adaptor.cc

@@ -396,6 +396,7 @@ struct flagcxDeviceAdaptor mluAdaptor {
       mluAdaptorSymFlatUnmap, mluAdaptorSymMulticastSupported,
       mluAdaptorSymMulticastCreate, mluAdaptorSymMulticastBind,
       mluAdaptorSymMulticastTeardown, mluAdaptorSymMulticastFree,
+      NULL, // flagcxResult_t (*getLastError)();
 };
 
 #endif // USE_CAMBRICON_ADAPTOR

flagcx/adaptor/device/musa_adaptor.cc

@@ -425,6 +425,7 @@ struct flagcxDeviceAdaptor musaAdaptor {
       musaAdaptorSymFlatUnmap, musaAdaptorSymMulticastSupported,
       musaAdaptorSymMulticastCreate, musaAdaptorSymMulticastBind,
       musaAdaptorSymMulticastTeardown, musaAdaptorSymMulticastFree,
+      NULL, // flagcxResult_t (*getLastError)();
 };
 
 #endif // USE_MUSA_ADAPTOR

flagcx/adaptor/device/ppu_cuda_adaptor.cc

@@ -686,6 +686,7 @@ struct flagcxDeviceAdaptor ppucudaAdaptor {
       ppucudaAdaptorSymMulticastSupported, ppucudaAdaptorSymMulticastCreate,
       ppucudaAdaptorSymMulticastBind, ppucudaAdaptorSymMulticastTeardown,
       ppucudaAdaptorSymMulticastFree,
+      NULL, // flagcxResult_t (*getLastError)();
 };
 
 #endif // USE_PPU_ADAPTOR

flagcx/adaptor/device/ptpu_adaptor.cc

@@ -423,5 +423,6 @@ struct flagcxDeviceAdaptor ptpuAdaptor {
       // DMA buffer
       ptpuAdaptorDmaSupport, ptpuAdaptorGetHandleForAddressRange,
       ptpuAdaptorHostRegister, ptpuAdaptorHostUnregister,
+      NULL, // flagcxResult_t (*getLastError)();
 };
 #endif // USE_SUNRISE_ADAPTOR

flagcx/adaptor/device/tops_adaptor.cc

@@ -489,6 +489,7 @@ struct flagcxDeviceAdaptor topsAdaptor {
       topsAdaptorSymFlatUnmap, topsAdaptorSymMulticastSupported,
       topsAdaptorSymMulticastCreate, topsAdaptorSymMulticastBind,
       topsAdaptorSymMulticastTeardown, topsAdaptorSymMulticastFree,
+      NULL, // flagcxResult_t (*getLastError)();
 };
 
 #endif // USE_ENFLAME_ADAPTOR

flagcx/adaptor/device/tsmicro_adaptor.cc

@@ -471,6 +471,7 @@ struct flagcxDeviceAdaptor tsmicroAdaptor {
       tsmicroAdaptorSymMulticastSupported, tsmicroAdaptorSymMulticastCreate,
       tsmicroAdaptorSymMulticastBind, tsmicroAdaptorSymMulticastTeardown,
       tsmicroAdaptorSymMulticastFree,
+      NULL, // flagcxResult_t (*getLastError)();
 };
 
 #endif // USE_TSM_ADAPTOR

flagcx/adaptor/include/flagcx_device_adaptor.h

@@ -243,6 +243,8 @@ struct flagcxDeviceAdaptor_latest {
   // Release the multicast object handle returned by symMulticastCreate.
   // Must be called after all ranks have torn down their mappings.
   flagcxResult_t (*symMulticastFree)(void *mcHandle);
+
+  flagcxResult_t (*getLastError)();
 };
 
 #define flagcxDeviceAdaptor flagcxDeviceAdaptor_latest

flagcx/core/flagcx_p2p.cc

@@ -92,8 +92,10 @@ void loadGlobalConfig(FlagcxP2pGlobalConfig &c) {
                       4, "P2P_QPS_PER_CONN");
   c.workersPerPool = clampParam<int>(flagcxParamP2pWorkersPerPool(), 1, 8, 4,
                                      "P2P_WORKERS_PER_POOL");
+  c.workersPerPool = std::min(c.workersPerPool, c.qpsPerConn);
   c.shardCount =
       clampParam<int>(flagcxParamP2pShardCount(), 1, 64, 8, "P2P_SHARD_COUNT");
+  c.shardCount = std::max(c.shardCount, c.workersPerPool);
   c.sharedCqDepth = clampParam<size_t>(flagcxParamP2pCqDepth(), 1, 1u << 20,
                                        4096, "P2P_CQ_DEPTH");
   c.maxWrPerPost = clampParam<size_t>(flagcxParamP2pMaxWrPerPost(), 1, 1024,
@@ -1477,7 +1479,8 @@ static int detectPtrTypeAndMaybeCacheIpc(void *ptr, char *ipcHandleBuf,
     deviceAdaptor->ipcMemHandleFree(handle);
     return FLAGCX_PTR_CUDA;
   }
-
+  if (deviceAdaptor->getLastError)
+    deviceAdaptor->getLastError();
   deviceAdaptor->ipcMemHandleFree(handle);
   return FLAGCX_PTR_HOST;
 }
@@ -2632,11 +2635,22 @@ bool flagcxP2pEngineConnIsLocal(FlagcxP2pConn *conn) {
   return conn != NULL && conn->isLocal;
 }
 
-int flagcxP2pEngineReg(FlagcxP2pEngine *engine, uintptr_t data, size_t size,
-                       FlagcxP2pMr &mrId) {
+int flagcxP2pEngineRegEx(FlagcxP2pEngine *engine, uintptr_t data, size_t size,
+                         int hintType, FlagcxP2pMr &mrId) {
   if (engine == NULL || data == 0)
     return -1;
 
+  auto resolvePtrType = [&](char *ipcHandleBuf,
+                            uint32_t *ipcHandleSize) -> int {
+    if (hintType == FLAGCX_PTR_HOST || hintType == FLAGCX_PTR_CUDA) {
+      if (ipcHandleSize)
+        *ipcHandleSize = 0;
+      return hintType;
+    }
+    return detectPtrTypeAndMaybeCacheIpc(reinterpret_cast<void *>(data),
+                                         ipcHandleBuf, ipcHandleSize);
+  };
+
   if (!flagcxParamMrSortedLookup()) {
     /* Legacy: mutex + hash maps */
     std::lock_guard<std::mutex> lock(gMemMutex);
@@ -2666,8 +2680,7 @@ int flagcxP2pEngineReg(FlagcxP2pEngine *engine, uintptr_t data, size_t size,
     entry.ibDevN = ibDevN;
 
     setEngineDevice(engine);
-    entry.ptrType = detectPtrTypeAndMaybeCacheIpc(
-        reinterpret_cast<void *>(data), entry.ipcHandle, &entry.ipcHandleSize);
+    entry.ptrType = resolvePtrType(entry.ipcHandle, &entry.ipcHandleSize);
     entry.hasIpc = entry.ptrType == FLAGCX_PTR_CUDA && entry.ipcHandleSize > 0;
 
     if (engine->adaptor->regMr(&devCtx, reinterpret_cast<void *>(data), size,
@@ -2721,8 +2734,7 @@ int flagcxP2pEngineReg(FlagcxP2pEngine *engine, uintptr_t data, size_t size,
   memset(ipcHandle, 0, sizeof(ipcHandle));
 
   setEngineDevice(engine);
-  int ptrType = detectPtrTypeAndMaybeCacheIpc(reinterpret_cast<void *>(data),
-                                              ipcHandle, &ipcHandleSize);
+  int ptrType = resolvePtrType(ipcHandle, &ipcHandleSize);
   bool hasIpc = ptrType == FLAGCX_PTR_CUDA && ipcHandleSize > 0;
 
   /* Register with adaptor */
@@ -2766,6 +2778,11 @@ int flagcxP2pEngineReg(FlagcxP2pEngine *engine, uintptr_t data, size_t size,
   return 0;
 }
 
+int flagcxP2pEngineReg(FlagcxP2pEngine *engine, uintptr_t data, size_t size,
+                       FlagcxP2pMr &mrId) {
+  return flagcxP2pEngineRegEx(engine, data, size, 0, mrId);
+}
+
 void flagcxP2pEngineMrDestroy(FlagcxP2pEngine *engine, FlagcxP2pMr mr) {
   if (engine == NULL)
     return;
@@ -3480,6 +3497,20 @@ int flagcxP2pRpcRegister(void *engine, uint64_t addr, uint64_t size,
   return 0;
 }
 
+int flagcxP2pRpcRegisterHost(void *engine, uint64_t addr, uint64_t size,
+                             uint64_t *mrIdOut) {
+  if (mrIdOut == NULL)
+    return -1;
+  FlagcxP2pMr mrId = 0;
+  const int rc = flagcxP2pEngineRegEx(
+      reinterpret_cast<FlagcxP2pEngine *>(engine), static_cast<uintptr_t>(addr),
+      static_cast<size_t>(size), FLAGCX_PTR_HOST, mrId);
+  if (rc != 0)
+    return rc;
+  *mrIdOut = mrId;
+  return 0;
+}
+
 void *flagcxP2pRpcGetConn(void *engine, const char *session) {
   return reinterpret_cast<void *>(flagcxP2pEngineGetConn(
       reinterpret_cast<FlagcxP2pEngine *>(engine), session));

flagcx/include/flagcx_p2p.h

@@ -618,6 +618,12 @@ int flagcxP2pRpcStartServer(void *engine);
 int flagcxP2pRpcRegister(void *engine, uint64_t addr, uint64_t size,
                          uint64_t *mrIdOut);
 
+/* Same interface as flagcxP2pRpcRegister, but on chips/adaptors whose runtime
+   cannot reset the current error state, this host entry avoids producing the
+   error at all. */
+int flagcxP2pRpcRegisterHost(void *engine, uint64_t addr, uint64_t size,
+                             uint64_t *mrIdOut);
+
 /* Get (or lazily establish) a cached connection to "host:port". */
 void *flagcxP2pRpcGetConn(void *engine, const char *session);
 

plugin/interservice/flagcx_wrapper.py

@@ -419,6 +419,10 @@ class FLAGCXLibrary:
             flagcxP2pEngine_t, ctypes.c_uint64, ctypes.c_uint64,
             ctypes.POINTER(ctypes.c_uint64)
         ]),
+        Function("flagcxP2pRpcRegisterHost", ctypes.c_int, [
+            flagcxP2pEngine_t, ctypes.c_uint64, ctypes.c_uint64,
+            ctypes.POINTER(ctypes.c_uint64)
+        ]),
         Function("flagcxP2pRpcGetConn", flagcxP2pConn_t, [
             flagcxP2pEngine_t, ctypes.c_char_p
         ]),
@@ -447,9 +451,12 @@ def _find_default_library() -> str:
             so_path = os.path.join(flagcx_path, "lib", "libflagcx.so")
             if os.path.isfile(so_path):
                 return so_path
+            build_so_path = os.path.join(flagcx_path, "build", "lib", "libflagcx.so")
+            if os.path.isfile(build_so_path):
+                return build_so_path
             raise FileNotFoundError(
-                f"FLAGCX_PATH is set to '{flagcx_path}' but "
-                f"'{so_path}' does not exist. "
+                f"FLAGCX_PATH is set to '{flagcx_path}' but neither "
+                f"'{so_path}' nor '{build_so_path}' exists. "
                 f"Please build FlagCX or check FLAGCX_PATH."
             )
         # 2. Fall back to <repo_root>/build/lib/libflagcx.so
@@ -818,6 +825,17 @@ def flagcxP2pRegister(self, engine: flagcxP2pEngine_t,
                 f"flagcxP2pRegister failed (addr={hex(addr)}, size={size})")
         return mr_id.value
 
+    def flagcxP2pRegisterHost(self, engine: flagcxP2pEngine_t,
+                              addr: int, size: int) -> int:
+        mr_id = ctypes.c_uint64(0)
+        rc = self._funcs["flagcxP2pRpcRegisterHost"](
+            engine, ctypes.c_uint64(addr), ctypes.c_uint64(size),
+            ctypes.byref(mr_id))
+        if rc != 0:
+            raise RuntimeError(
+                f"flagcxP2pRegisterHost failed (addr={hex(addr)}, size={size})")
+        return mr_id.value
+
     def flagcxP2pGetConn(self, engine: flagcxP2pEngine_t,
                          session: str) -> flagcxP2pConn_t:
         conn = self._funcs["flagcxP2pRpcGetConn"](