Urma event mode

cmake/config.cmake

@@ -78,6 +78,43 @@ if (YLT_ENABLE_IBV)
         target_link_libraries(${ylt_target_name} INTERFACE -libverbs -lmlx5)
     endif ()
 endif ()
+option(YLT_ENABLE_URMA "Enable URMA support" OFF)
+if (YLT_ENABLE_URMA)
+    message(STATUS "Enable URMA support")
+    find_path(URMA_INCLUDE_PATH NAMES urma_api.h
+              HINTS ${URMA_ROOT} ENV URMA_ROOT
+              PATHS /usr/include /usr/local/include
+              PATH_SUFFIXES ub/umdk/urma umdk/urma urma
+                            src/urma/lib/urma/core/include)
+    find_path(URMA_BOND_INCLUDE_PATH NAMES urma_ubagg.h
+              HINTS ${URMA_ROOT} ENV URMA_ROOT
+              PATHS /usr/include /usr/local/include
+              PATH_SUFFIXES ub/umdk/urma umdk/urma urma
+                            src/urma/lib/urma/bond/include)
+    if (NOT URMA_INCLUDE_PATH OR NOT URMA_BOND_INCLUDE_PATH)
+        message(FATAL_ERROR
+                "Fail to find URMA headers. Install UMDK headers or set URMA_ROOT.")
+    endif()
+    message(STATUS "Found URMA headers: ${URMA_INCLUDE_PATH}")
+    if (URMA_BOND_INCLUDE_PATH)
+        message(STATUS "Found URMA bonding headers: ${URMA_BOND_INCLUDE_PATH}")
+    endif()
+    if(CMAKE_PROJECT_NAME STREQUAL "yaLanTingLibs")
+        add_compile_definitions("YLT_ENABLE_URMA")
+        include_directories(${URMA_INCLUDE_PATH})
+        if (URMA_BOND_INCLUDE_PATH)
+            include_directories(${URMA_BOND_INCLUDE_PATH})
+        endif()
+        link_libraries(-lurma)
+    else ()
+        target_compile_definitions(${ylt_target_name} INTERFACE "YLT_ENABLE_URMA")
+        target_include_directories(${ylt_target_name} INTERFACE ${URMA_INCLUDE_PATH})
+        if (URMA_BOND_INCLUDE_PATH)
+            target_include_directories(${ylt_target_name} INTERFACE ${URMA_BOND_INCLUDE_PATH})
+        endif()
+        target_link_libraries(${ylt_target_name} INTERFACE -lurma)
+    endif ()
+endif ()
 if (YLT_ENABLE_CUDA)
     message(STATUS "Enable cuda support")
     find_package(CUDAToolkit REQUIRED)

docs/urma_rpc.md

@@ -0,0 +1,157 @@
+# URMA RPC
+
+本文介绍 `coro_rpc` 的 URMA 传输层、自动升级机制、配置方式和底层接口。URMA 是可选功能；未启用或初始化失败时，`coro_rpc` 继续使用 TCP。
+
+## 1. 架构与连接流程
+
+URMA RPC 由三层组成：
+
+1. `coro_rpc`：负责 RPC 编解码、handler、连接管理和 attachment。
+2. `coro_io::urma_socket_t`：负责 TCP 握手、URMA Jetty 建链、收发队列和完成事件。
+3. URMA 资源层：负责设备、context、JFC/JFCE、Jetty、Segment 和 work request。
+
+连接建立先使用 TCP 交换 EID、UASID、Jetty ID 和 buffer pool segment；随后双方导入对端资源，数据面切换到 URMA。TCP 只用于握手。
+
+```mermaid
+flowchart TD
+    RPC["coro_rpc client/server"] --> SOCKET["coro_io::urma_socket_t"]
+    SOCKET --> HANDSHAKE["TCP handshake"]
+    HANDSHAKE --> META["Peer metadata: EID, UASID, Jetty ID, Segment"]
+    SOCKET --> RESOURCES["URMA resources"]
+    RESOURCES --> CONTEXT["Device and context"]
+    RESOURCES --> QUEUES["JFC/JFCE and Jetty"]
+    RESOURCES --> SEGMENT["Registered buffer pool segment"]
+    META --> IMPORT["Import peer resources"]
+    QUEUES --> WR["Send/receive work requests"]
+    SEGMENT --> WR
+    IMPORT --> WR
+    WR --> COMPLETION["Completion polling or event loop"]
+```
+
+buffer pool 将一块连续内存注册为 Segment，再切分为固定大小的 buffer，减少重复注册开销。发送窗口受本地发送 buffer 数和对端接收 buffer 数共同限制。
+
+## 2. 构建与显式启用
+
+```bash
+cmake -S . -B build -DYLT_ENABLE_URMA=ON -DBUILD_EXAMPLES=ON
+cmake --build build --target coro_rpc_urma_example -j
+```
+
+CMake 从 `URMA_ROOT`、`/usr/include` 和 `/usr/local/include` 查找 UMDK
+头文件，支持 `urma/`、`umdk/urma/` 和 UMDK 源码目录布局。例如：
+
+```bash
+cmake -S . -B build -DYLT_ENABLE_URMA=ON \
+  -DURMA_ROOT=/opt/umdk
+```
+
+也可以通过环境变量指定：
+
+```bash
+export URMA_ROOT=/opt/umdk
+```
+
+配置必须同时包含 `urma_api.h` 和 bonding 扩展头 `urma_ubagg.h`，并在启用
+URMA 时链接系统 `liburma`。
+
+```cpp
+coro_rpc_server server(std::thread::hardware_concurrency(), 9000);
+server.init_urma();
+server.register_handler<echo>();
+server.start();
+```
+
+```cpp
+coro_rpc_client client;
+if (!client.init_urma()) {
+  co_return;
+}
+auto ec = co_await client.connect("127.0.0.1:9000");
+auto result = co_await client.call_for<echo>(30s, "hello urma");
+```
+
+## 3. 配置
+
+```cpp
+coro_io::urma_socket_t::config_t config{
+    .cq_size = 128,
+    .recv_buffer_cnt = 64,
+    .send_buffer_cnt = 64,
+    .buffer_size = 4096,
+    .max_memory_usage = 256ull * 1024 * 1024,
+    .device_name = "bonding_dev_0",
+    .eid_index = 0,
+    .tp_type = URMA_CTP,
+};
+
+server.init_urma(config);
+client.init_urma(config);
+```
+
+## 4. 环境变量自动升级
+
+默认 TCP 配置下，设置 `URMA_RPC_ENABLE=1` 可自动探测 URMA 设备并升级：
+
+```bash
+export URMA_RPC_ENABLE=1
+export URMA_RPC_DEVICE=bonding_dev_0
+export URMA_RPC_EID_INDEX=0
+```
+
+`URMA_RPC_ENABLE` 接受 `1`、`on`、`true`、`yes`。未设置、关闭或设备探测失败时回退 TCP。显式调用 `init_urma(config)` 的配置优先。
+
+| 变量 | 作用 |
+| --- | --- |
+| `URMA_RPC_ENABLE` | 启用自动升级 |
+| `URMA_RPC_DEVICE` | 设备名；为空时自动选择 |
+| `URMA_RPC_EID_INDEX` | EID 下标，默认 `0` |
+| `URMA_RPC_CQ_SIZE` | completion queue 大小 |
+| `URMA_RPC_RECV_BUFFER_CNT` / `URMA_RPC_SEND_BUFFER_CNT` | 收发 buffer 数 |
+| `URMA_RPC_BUFFER_SIZE` | 单个 buffer 大小 |
+| `URMA_RPC_MAX_MEMORY_USAGE` | buffer pool 上限，单位 byte |
+| `URMA_RPC_TP_TYPE` | `ctp`、`rtp` 或其他支持的类型 |
+| `URMA_RPC_EVENT_MODE` | JFCE 事件模式开关 |
+| `URMA_RPC_BUSY_POLL_BUDGET` | 事件唤醒后的忙轮询次数 |
+| `URMA_RPC_POLL_INTERVAL` | 轮询间隔，单位微秒 |
+
+实现位于 `include/ylt/coro_io/urma/urma_rpc_env.hpp`，行为测试位于 `src/coro_rpc/tests/test_urma_rpc_env.cpp`。
+
+## 5. URMA 接口
+
+| 层次 | 主要接口 | 作用 |
+| --- | --- | --- |
+| 生命周期 | `urma_init`、`urma_uninit` | 初始化和释放 runtime |
+| 设备 | `urma_get_device_list`、`urma_get_eid_list`、`urma_query_device` | 查询设备、EID 和能力 |
+| Context | `urma_create_context`、`urma_delete_context` | 创建设备上下文 |
+| 完成队列 | `urma_create_jfc`、`urma_create_jfce`、`urma_poll_jfc`、`urma_wait_jfc` | 等待并读取完成记录 |
+| 数据队列 | `urma_create_jfs`、`urma_create_jfr`、`urma_create_jetty` | 创建收发资源 |
+| 远端资源 | `urma_import_seg`、`urma_import_jetty` | 导入对端 Segment 和 Jetty |
+| 本地内存 | `urma_register_seg` | 注册 URMA 可访问内存 |
+| 数据收发 | `urma_post_jetty_send_wr`、`urma_post_jetty_recv_wr` | 提交 work request |
+
+典型顺序是：创建 context → 创建 JFC/JFCE 和 Jetty → 注册 Segment → 交换元数据 → 导入对端资源 → post WR → poll/wait completion → 释放资源。声明来自外部 UMDK 的 `urma_api.h` 和 `urma_types.h`。
+
+## 6. 大数据与压测
+
+大 payload 建议使用 attachment：
+
+```cpp
+client.set_req_attachment(payload);
+auto result = co_await client.call_for<upload>(30s, payload.size());
+```
+
+```bash
+cmake --build build --target coro_rpc_urma_benchmark -j
+```
+
+详细参数见 [`urma_benchmark/README.md`](../src/coro_rpc/examples/urma_benchmark/README.md)。`--transport raw` 用于排除 RPC 编解码开销，`--rpc attach_sink` 用于测试 attachment 快路径。
+
+## 7. 排障
+
+- 没有 URMA 目标：确认使用 `-DYLT_ENABLE_URMA=ON`，并安装匹配的库、头文件和驱动。
+- 自动升级未生效：确认 `URMA_RPC_ENABLE` 有效，且应用走默认 TCP 配置路径。
+- 初始化或连接失败：检查设备名、EID、TP 类型和 TCP 握手端口。
+- 高并发下出现 RNR 或 `WR_FLUSH_ERR`：降低连接数、pipeline depth、队列深度，或增加内存上限。
+- 定位延迟：启用 benchmark 的 `--profile`，观察握手、post send、completion wait、RPC dispatch 和 attachment 阶段。
+
+示例：`src/coro_rpc/examples/urma_example/urma_example.cpp`；实现：`include/ylt/coro_io/urma/`；公共 API 来自 UMDK 的 `urma_api.h`、`urma_types.h` 和 `urma_ubagg.h`。

include/ylt/coro_io/client_pool.hpp

@@ -49,6 +49,7 @@
 #include "coro_io.hpp"
 #include "detail/client_queue.hpp"
 #include "io_context_pool.hpp"
+#include "ylt/coro_io/urma/urma_benchmark_profile.hpp"
 #include "ylt/easylog.hpp"
 #include "ylt/util/atomic_shared_ptr.hpp"
 #ifdef YLT_ENABLE_IBV
@@ -316,7 +317,13 @@ class client_pool : public std::enable_shared_from_this<
           ELOG_ERROR << "init client config failed.";
           co_return nullptr;
         }
+      auto connect_begin = coro_io::urma_benchmark_profile::enabled()
+                             ? coro_io::urma_benchmark_profile::now_ns()
+                             : 0;
       co_await reconnect(client, this->weak_from_this());
+      coro_io::urma_benchmark_profile::record_since(
+          coro_io::urma_benchmark_profile::stage::client_connect_total,
+          connect_begin);
     }
     else {
       ELOG_TRACE << "get free client{" << client.get() << "}. from queue";

include/ylt/coro_io/coro_io.hpp

@@ -612,6 +612,7 @@ inline async_simple::coro::Lazy<std::error_code> async_connect(
   co_return ec;
 }
 #endif
+
 class period_timer : public asio::steady_timer {
  public:
   using asio::steady_timer::steady_timer;

include/ylt/coro_io/data_view.hpp

@@ -16,6 +16,7 @@
 #pragma once
 #include <cstddef>
 #include <cstdint>
+#include <memory>
 #include <span>
 #include <string_view>
 
@@ -69,4 +70,20 @@ class data_view : public std::string_view {
  private:
   int gpu_id_;  // GPU ID (-1 for CPU memory, >=0 for GPU memory)
 };
-}  // namespace coro_io
+
+struct owned_data_view {
+  data_view view;
+  std::shared_ptr<void> owner;
+
+  owned_data_view() = default;
+  owned_data_view(data_view view, std::shared_ptr<void> owner)
+      : view(view), owner(std::move(owner)) {}
+
+  bool empty() const noexcept { return view.empty(); }
+  const char* data() const noexcept { return view.data(); }
+  std::size_t size() const noexcept { return view.size(); }
+  int gpu_id() const noexcept { return view.gpu_id(); }
+  operator data_view() const noexcept { return view; }
+  operator std::string_view() const noexcept { return view; }
+};
+}  // namespace coro_io

include/ylt/coro_io/detail/circle_buffer.hpp

@@ -0,0 +1,68 @@
+/*
+ * Copyright (c) 2025, Alibaba Group Holding Limited;
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#ifndef YLT_CORO_IO_DETAIL_CIRCLE_BUFFER_HPP
+#define YLT_CORO_IO_DETAIL_CIRCLE_BUFFER_HPP
+
+#include <cassert>
+#include <cstdint>
+#include <vector>
+
+namespace coro_io {
+namespace detail {
+
+template <typename T>
+struct circle_buffer {
+  std::vector<T> queue;
+  uint32_t front_ = 0, end_ = 0;
+  bool may_empty = true;
+  circle_buffer() = default;
+  circle_buffer(uint32_t size) {
+    assert(size > 0);
+    queue.resize(size);
+  }
+  void push(T&& elem) {
+    assert(!full());
+    may_empty = false;
+    end_ = (end_ + 1) % queue.size();
+    queue[end_] = std::move(elem);
+  }
+  T pop() {
+    assert(!empty());
+    front_ = (front_ + 1) % queue.size();
+    may_empty = true;
+    return std::move(queue[front_]);
+  }
+  T& back() { return queue[end_]; }
+  T& front() { return queue[(front_ + 1) % queue.size()]; }
+  bool full() const noexcept { return end_ == front_ && !may_empty; }
+  bool empty() const noexcept { return end_ == front_ && may_empty; }
+  std::size_t size() const noexcept {
+    if (front_ > end_) {
+      return queue.size() + end_ - front_;
+    }
+    else if (front_ == end_) {
+      return empty() ? 0 : queue.size();
+    }
+    else {
+      return end_ - front_;
+    }
+  }
+};
+
+}  // namespace detail
+}  // namespace coro_io
+
+#endif  // YLT_CORO_IO_DETAIL_CIRCLE_BUFFER_HPP

include/ylt/coro_io/ibverbs/ib_socket.hpp

@@ -37,6 +37,7 @@
 #include "asio/ip/tcp.hpp"
 #include "asio/posix/stream_descriptor.hpp"
 #include "async_simple/Future.h"
+#include "async_simple/Promise.h"
 #include "async_simple/Signal.h"
 #include "async_simple/coro/FutureAwaiter.h"
 #include "async_simple/coro/Lazy.h"
@@ -45,6 +46,7 @@
 #include "ib_device.hpp"
 #include "ib_error.hpp"
 #include "ylt/coro_io/coro_io.hpp"
+#include "ylt/coro_io/detail/circle_buffer.hpp"
 #include "ylt/coro_io/ibverbs/ib_buffer.hpp"
 #include "ylt/coro_io/io_context_pool.hpp"
 #include "ylt/easylog.hpp"
@@ -54,43 +56,6 @@
 namespace coro_io {
 namespace detail {
 struct ib_socket_shared_state_t;
-template <typename T>
-struct circle_buffer {
-  std::vector<T> queue;
-  uint32_t front_ = 0, end_ = 0;
-  bool may_empty = true;
-  circle_buffer(uint32_t size) {
-    assert(size > 0);
-    queue.resize(size);
-  }
-  void push(T&& elem) {
-    assert(!full());
-    may_empty = false;
-    end_ = (end_ + 1) % queue.size();
-    queue[end_] = std::move(elem);
-  }
-  T pop() {
-    assert(!empty());
-    front_ = (front_ + 1) % queue.size();
-    may_empty = true;
-    return std::move(queue[front_]);
-  }
-  T& back() { return queue[end_]; }
-  T& front() { return queue[(front_ + 1) % queue.size()]; }
-  bool full() const noexcept { return end_ == front_ && !may_empty; }
-  bool empty() const noexcept { return end_ == front_ && may_empty; }
-  std::size_t size() const noexcept {
-    if (front_ > end_) {
-      return queue.size() + end_ - front_;
-    }
-    else if (front_ == end_) {
-      return empty() ? 0 : queue.size();
-    }
-    else {
-      return end_ - front_;
-    }
-  }
-};
 struct ib_buffer_queue : public circle_buffer<ib_buffer_t> {
   using circle_buffer::circle_buffer;
   std::error_code post_recv_real(ibv_sge buffer,