diff --git a/.github/workflows/ci-linux.yml b/.github/workflows/ci-linux.yml index e15d81db4f..8befc0bf5f 100644 --- a/.github/workflows/ci-linux.yml +++ b/.github/workflows/ci-linux.yml @@ -2,7 +2,7 @@ name: Build and Test on Linux on: push: - branches: [ master ] + branches: [ master, urma_transport_dev ] paths-ignore: - '**.md' pull_request: @@ -85,6 +85,60 @@ jobs: -DWITH_ASAN=ON -DCMAKE_POLICY_VERSION_MINIMUM=3.5 .. make -j ${{env.proc_num}} && make clean + cmake-unittest-urma-mock: + # No URMA hardware is available in CI, so this builds against brpc's + # URMA link-time mock (WITH_URMA_MOCK=ON) instead of a real liburma, and + # actually runs UrmaTransport's unit tests against it. + runs-on: ubuntu-22.04 + steps: + - uses: actions/checkout@v2 + - uses: ./.github/actions/install-all-dependencies + - name: install gperftools + # test/CMakeLists.txt links ${GPERFTOOLS_LIBRARIES} into every + # brpc_*_unittest target unconditionally, and FindGperftools.cmake + # leaves it as "...-NOTFOUND" when the library is absent, which then + # reaches the linker verbatim. The existing unittest jobs build via + # test/Makefile and get gperftools from init-ut-make-config; this is + # the first CMake-based unittest job, so install it explicitly. + run: sudo apt-get install -y libgoogle-perftools-dev + - name: configure + run: | + mkdir build && cd build + cmake -DBUILD_UNIT_TESTS=ON -DDOWNLOAD_GTEST=ON \ + -DWITH_URMA=ON -DWITH_URMA_MOCK=ON \ + -DCMAKE_POLICY_VERSION_MINIMUM=3.5 .. + - name: build + run: | + cd build + make -j ${{env.proc_num}} brpc_urma_unittest + - name: run tests + run: | + cd build + ctest -R brpc_urma_unittest --output-on-failure + + make-compile-urma-mock: + # The Makefile build is the one that pins -std=c++14 (config_brpc.sh), and + # it is also the only build where --werror is on by default in CI. The + # CMake job above builds the same URMA sources with the compiler's default + # standard, so it cannot catch C++17-only constructs in the URMA transport + # or in its link-time mock. Keep both. + runs-on: ubuntu-22.04 + steps: + - uses: actions/checkout@v2 + - uses: ./.github/actions/install-all-dependencies + - name: fetch UMDK headers + # config_brpc.sh has no DOWNLOAD_URMA_HEADERS equivalent (CMake) and no + # @umdk repository rule (Bazel), so fetch the same pinned UMDK revision + # by hand and hand the include dirs to --headers. + run: git clone --depth 1 --branch v26.06.0_CAM https://atomgit.com/openeuler/umdk.git "$HOME/umdk" + - name: compile with make (URMA link-time mock) + run: | + sh config_brpc.sh --nodebugsymbols --werror --cc=gcc --cxx=g++ \ + --headers="/usr/include $HOME/umdk/src/urma/lib/urma/core/include $HOME/umdk/src/urma/lib/urma/bond/include" \ + --libs="/usr/lib /usr/lib64" \ + --with-urma --with-urma-mock + cat config.mk && make -j ${{env.proc_num}} + gcc-compile-with-make-protobuf: runs-on: ubuntu-22.04 steps: diff --git a/BUILD.bazel b/BUILD.bazel index 8df080c4bb..cb94044461 100644 --- a/BUILD.bazel +++ b/BUILD.bazel @@ -54,6 +54,9 @@ DEFINES = [ }) + select({ "//bazel/config:brpc_with_urma": ["BRPC_WITH_URMA=1"], "//conditions:default": [], + }) + select({ + "//bazel/config:brpc_with_urma_mock": ["BRPC_WITH_URMA_MOCK=1"], + "//conditions:default": [], }) + select({ "//bazel/config:brpc_with_debug_bthread_sche_safety": ["BRPC_DEBUG_BTHREAD_SCHE_SAFETY=1"], "//conditions:default": ["BRPC_DEBUG_BTHREAD_SCHE_SAFETY=0"], @@ -100,6 +103,16 @@ LINKOPTS = [ "-libverbs", ], "//conditions:default": [], +}) + select({ + # brpc_with_urma_mock is a specialization of brpc_with_urma (see + # bazel/config/BUILD.bazel), so it wins this select whenever both defines + # are set: mock builds compile against brpc's link-time mock instead and + # must not also link the real liburma, or urma_* symbols would clash. + "//bazel/config:brpc_with_urma_mock": [], + "//bazel/config:brpc_with_urma": [ + "-lurma", + ], + "//conditions:default": [], }) + select({ "//bazel/config:brpc_with_asan": ["-fsanitize=address"], "//conditions:default": [], diff --git a/CMakeLists.txt b/CMakeLists.txt index 10e9052dcb..1fa9f55b27 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -31,6 +31,9 @@ option(WITH_URMA "With URMA (openEuler Unified Remote Memory Access)" OFF) option(DOWNLOAD_URMA_HEADERS "Download UMDK headers when WITH_URMA is enabled and headers are absent" ON) +option(WITH_URMA_MOCK + "Explicitly allow linking brpc's URMA link-time mock when liburma is not found (WITH_URMA only). The mock cannot talk to real URMA hardware, so this must be opted into rather than silently substituted." + OFF) option(WITH_UBRING "With UB" OFF) option(WITH_DEBUG_BTHREAD_SCHE_SAFETY "With debugging bthread sche safety" OFF) option(WITH_DEBUG_LOCK "With debugging lock" OFF) @@ -368,14 +371,37 @@ if(WITH_URMA) find_library(URMA_LIB NAMES urma HINTS ENV URMA_ROOT PATH_SUFFIXES lib lib64) - if(URMA_LIB) + # WITH_URMA_MOCK is an explicit opt-in, so it wins over auto-detection. + # Deciding by "is liburma installed?" would make a mock build mean + # different things on different machines -- a CI image without liburma + # gets the mock, a developer box with liburma silently gets a hardware + # build even though it asked for the mock. Bazel already resolves it this + # way (brpc_with_urma_mock is a specialization of brpc_with_urma, so it + # wins whenever both defines are set); this keeps CMake consistent. + if(WITH_URMA_MOCK) + if(URMA_LIB) + message(WARNING + "WITH_URMA_MOCK=ON, so brpc's URMA link-time mock is used " + "instead of the liburma found at ${URMA_LIB}. The mock " + "cannot talk to real URMA hardware -- drop WITH_URMA_MOCK " + "for a hardware build.") + else() + message(STATUS + "liburma not found; WITH_URMA_MOCK=ON, building with the " + "URMA link-time mock") + endif() + set(URMA_USE_MOCK 1) + elseif(URMA_LIB) message(STATUS "Found URMA library: ${URMA_LIB}") set(URMA_USE_MOCK 0) else() - message(STATUS - "liburma not found; building with the URMA link-time mock") - set(URMA_USE_MOCK 1) + message(FATAL_ERROR + "Fail to find liburma. Install liburma, set URMA_ROOT, or " + "explicitly opt into brpc's link-time mock with " + "-DWITH_URMA_MOCK=ON (the mock cannot talk to real URMA " + "hardware; only enable it for CI/tests without URMA hardware).") endif() + list(APPEND BRPC_COMMON_DEFINITIONS BRPC_WITH_URMA_MOCK=${URMA_USE_MOCK}) endif() find_library(PROTOC_LIB NAMES protoc) diff --git a/MODULE.bazel b/MODULE.bazel index 391b8c2784..12e7f6fad0 100644 --- a/MODULE.bazel +++ b/MODULE.bazel @@ -53,13 +53,10 @@ git_override( commit = '1e08f8e0507b6b6b1f4416a9a22cf5c28beaba93', # Jun 28, 2024 ) -git_repository = use_repo_rule( - '@bazel_tools//tools/build_defs/repo:git.bzl', - 'git_repository', +umdk_repository = use_repo_rule( + '//bazel/third_party/umdk:repositories.bzl', + 'umdk_repository', ) -git_repository( +umdk_repository( name = 'umdk', - build_file = '//bazel/third_party/umdk:umdk.BUILD', - remote = 'https://atomgit.com/openeuler/umdk.git', - commit = '564ee727a55523d4351a8fb3c94292b388ebb924', # v26.06.0_CAM ) diff --git a/WORKSPACE b/WORKSPACE index 22fc411b32..37ff3875e1 100644 --- a/WORKSPACE +++ b/WORKSPACE @@ -19,6 +19,7 @@ workspace(name = "com_github_brpc_brpc") load("@bazel_tools//tools/build_defs/repo:http.bzl", "http_archive") load("@bazel_tools//tools/build_defs/repo:git.bzl", "git_repository") +load("//bazel/third_party/umdk:repositories.bzl", "umdk_repository") # # Constants @@ -279,11 +280,8 @@ http_archive( urls = ["https://archive.apache.org/dist/thrift/0.15.0/thrift-0.15.0.tar.gz"], ) -git_repository( +umdk_repository( name = "umdk", - build_file = "//bazel/third_party/umdk:umdk.BUILD", - remote = "https://atomgit.com/openeuler/umdk.git", - commit = "564ee727a55523d4351a8fb3c94292b388ebb924", # v26.06.0_CAM ) # Header-only JSON library used by iobuf_unittest's IOBuf<->std::iostream diff --git a/bazel/config/BUILD.bazel b/bazel/config/BUILD.bazel index 98e801de34..d5ae2c8552 100644 --- a/bazel/config/BUILD.bazel +++ b/bazel/config/BUILD.bazel @@ -116,6 +116,21 @@ config_setting( visibility = ["//visibility:public"], ) +# A specialization of :brpc_with_urma (its define_values is a superset), so +# Bazel resolves select()s using both to this setting whenever it matches. +# Passing --define BRPC_WITH_URMA=true alone builds against a real liburma; +# adding --define BRPC_WITH_URMA_MOCK=true switches to brpc's URMA link-time +# mock instead. The mock cannot talk to real URMA hardware, so it must be +# opted into rather than silently substituted. +config_setting( + name = "brpc_with_urma_mock", + define_values = { + "BRPC_WITH_URMA": "true", + "BRPC_WITH_URMA_MOCK": "true", + }, + visibility = ["//visibility:public"], +) + config_setting( name = "brpc_with_boringssl", define_values = {"BRPC_WITH_BORINGSSL": "true"}, diff --git a/bazel/third_party/umdk/umdk.BUILD b/bazel/third_party/umdk/BUILD.bazel similarity index 70% rename from bazel/third_party/umdk/umdk.BUILD rename to bazel/third_party/umdk/BUILD.bazel index 410e6a2736..e7578fc37e 100644 --- a/bazel/third_party/umdk/umdk.BUILD +++ b/bazel/third_party/umdk/BUILD.bazel @@ -12,17 +12,6 @@ # 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. - -package(default_visibility = ["//visibility:public"]) - -cc_library( - name = "urma_headers", - hdrs = glob([ - "src/urma/lib/urma/bond/include/*.h", - "src/urma/lib/urma/core/include/*.h", - ]), - includes = [ - "src/urma/lib/urma/bond/include", - "src/urma/lib/urma/core/include", - ], -) +# +# This empty BUILD.bazel file is required to make Bazel treat +# this directory as a package. diff --git a/bazel/third_party/umdk/repositories.bzl b/bazel/third_party/umdk/repositories.bzl new file mode 100644 index 0000000000..6cf2fad9ac --- /dev/null +++ b/bazel/third_party/umdk/repositories.bzl @@ -0,0 +1,141 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You 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. + +_UMDK_REMOTE = "https://atomgit.com/openeuler/umdk.git" +_UMDK_COMMIT = "564ee727a55523d4351a8fb3c94292b388ebb924" # v26.06.0_CAM + +_BUILD_FILE = """ +package(default_visibility = ["//visibility:public"]) + +cc_library( + name = "urma_headers", + hdrs = glob([ + "src/urma/lib/urma/bond/include/*.h", + "src/urma/lib/urma/core/include/*.h", + ]), + includes = [ + "src/urma/lib/urma/bond/include", + "src/urma/lib/urma/core/include", + ], +) +""" + +def _is_false(value): + return value.lower() in ["0", "false", "no", "off"] + +def _first_existing_file(ctx, root, relpaths): + for relpath in relpaths: + # get_child() joins one segment at a time; passing a relative path + # containing separators is not portable across Bazel versions. + path = ctx.path(root) + for segment in relpath.split("/"): + path = path.get_child(segment) + if path.exists: + return relpath + return None + +def _write_build_file(ctx): + ctx.file("BUILD.bazel", _BUILD_FILE) + +def _use_local_umdk(ctx, urma_root): + core_include = _first_existing_file(ctx, urma_root, [ + "urma_api.h", + "ub/umdk/urma/urma_api.h", + "umdk/urma/urma_api.h", + "urma/urma_api.h", + # CMake's find_path() implicitly also searches an include/ prefix + # under HINTS, so URMA_ROOT=/usr resolves there. Probe the same + # layouts here to keep URMA_ROOT meaning the same in both builds. + "include/ub/umdk/urma/urma_api.h", + "include/umdk/urma/urma_api.h", + "include/urma/urma_api.h", + "include/urma_api.h", + "src/urma/lib/urma/core/include/urma_api.h", + ]) + if not core_include: + fail("URMA_ROOT is set to '%s', but no urma_api.h was found under it" % + urma_root) + + core_include = core_include[:-len("/urma_api.h")] if "/" in core_include else "." + bond_include = _first_existing_file(ctx, urma_root, [ + "urma_ubagg.h", + "ub/umdk/urma/urma_ubagg.h", + "umdk/urma/urma_ubagg.h", + "urma/urma_ubagg.h", + "include/ub/umdk/urma/urma_ubagg.h", + "include/umdk/urma/urma_ubagg.h", + "include/urma/urma_ubagg.h", + "include/urma_ubagg.h", + "src/urma/lib/urma/bond/include/urma_ubagg.h", + ]) + if not bond_include: + fail(("URMA_ROOT is set to '%s' and urma_api.h was found under it, " + + "but no urma_ubagg.h was. brpc needs both the core and the bond " + + "UMDK headers; a partial tree would only fail later with a " + + "missing-include error far from its cause.") % urma_root) + bond_include = bond_include[:-len("/urma_ubagg.h")] if "/" in bond_include else "." + + ctx.symlink( + ctx.path(urma_root).get_child(core_include), + "src/urma/lib/urma/core/include", + ) + ctx.symlink( + ctx.path(urma_root).get_child(bond_include), + "src/urma/lib/urma/bond/include", + ) + _write_build_file(ctx) + +def _run(ctx, args): + result = ctx.execute(args, quiet = False) + if result.return_code != 0: + fail("Failed to run '%s'\nstdout:\n%s\nstderr:\n%s" % + (" ".join(args), result.stdout, result.stderr)) + +def _download_umdk(ctx): + checkout = "umdk_checkout" + _run(ctx, ["git", "init", checkout]) + _run(ctx, ["git", "-C", checkout, "remote", "add", "origin", _UMDK_REMOTE]) + _run(ctx, ["git", "-C", checkout, "fetch", "--depth", "1", "origin", _UMDK_COMMIT]) + _run(ctx, ["git", "-C", checkout, "checkout", "--detach", "FETCH_HEAD"]) + ctx.symlink( + ctx.path(checkout).get_child("src/urma/lib/urma/core/include"), + "src/urma/lib/urma/core/include", + ) + ctx.symlink( + ctx.path(checkout).get_child("src/urma/lib/urma/bond/include"), + "src/urma/lib/urma/bond/include", + ) + _write_build_file(ctx) + +def _umdk_repository_impl(ctx): + urma_root = ctx.os.environ.get("URMA_ROOT", "") + if urma_root: + _use_local_umdk(ctx, urma_root) + return + + download = ctx.os.environ.get("BRPC_DOWNLOAD_URMA_HEADERS", "true") + if _is_false(download): + fail("Failed to find urma_api.h. Set URMA_ROOT to an installed UMDK " + + "tree or allow downloading with --repo_env=BRPC_DOWNLOAD_URMA_HEADERS=true.") + + _download_umdk(ctx) + +umdk_repository = repository_rule( + implementation = _umdk_repository_impl, + environ = [ + "BRPC_DOWNLOAD_URMA_HEADERS", + "URMA_ROOT", + ], +) diff --git a/config_brpc.sh b/config_brpc.sh index 2c1394e840..d5ce17fc65 100755 --- a/config_brpc.sh +++ b/config_brpc.sh @@ -54,11 +54,12 @@ else LDD=ldd fi -TEMP=`getopt -o v: --long headers:,libs:,cc:,cxx:,with-glog,with-thrift,with-rdma,with-urma,with-mesalink,with-bthread-tracer,with-debug-bthread-sche-safety,with-debug-lock,with-asan,with-riscv-zvbc,with-riscv-zbc,with-cpu-frequency,nodebugsymbols,werror -n 'config_brpc' -- "$@"` +TEMP=`getopt -o v: --long headers:,libs:,cc:,cxx:,with-glog,with-thrift,with-rdma,with-urma,with-urma-mock,with-mesalink,with-bthread-tracer,with-debug-bthread-sche-safety,with-debug-lock,with-asan,with-riscv-zvbc,with-riscv-zbc,with-cpu-frequency,nodebugsymbols,werror -n 'config_brpc' -- "$@"` WITH_GLOG=0 WITH_THRIFT=0 WITH_RDMA=0 WITH_URMA=0 +WITH_URMA_MOCK=0 WITH_MESALINK=0 WITH_BTHREAD_TRACER=0 WITH_ASAN=0 @@ -92,6 +93,7 @@ while true; do --with-thrift) WITH_THRIFT=1; shift 1 ;; --with-rdma) WITH_RDMA=1; shift 1 ;; --with-urma) WITH_URMA=1; shift 1 ;; + --with-urma-mock) WITH_URMA_MOCK=1; shift 1 ;; --with-mesalink) WITH_MESALINK=1; shift 1 ;; --with-bthread-tracer) WITH_BTHREAD_TRACER=1; shift 1 ;; --with-debug-bthread-sche-safety ) BRPC_DEBUG_BTHREAD_SCHE_SAFETY=1; shift 1 ;; @@ -550,13 +552,26 @@ if [ $WITH_URMA != 0 ]; then if [ -n "$URMA_BOND_HDR" ]; then append_to_output_headers "$URMA_BOND_HDR" fi - if [ -n "$URMA_LIB" ]; then + # --with-urma-mock is an explicit opt-in and wins over auto-detection, so + # that a mock build means the same thing whether or not the machine + # happens to have liburma installed. Matches CMake's WITH_URMA_MOCK and + # Bazel's brpc_with_urma_mock. + if [ $WITH_URMA_MOCK != 0 ]; then + append_to_output "URMA_USE_MOCK=1" + CPPFLAGS="${CPPFLAGS} -DBRPC_WITH_URMA_MOCK=1" + if [ -n "$URMA_LIB" ]; then + print_info "--with-urma-mock given, using URMA link-time mock instead of the liburma found in $URMA_LIB (the mock cannot talk to real URMA hardware)" + else + print_info "liburma not found; --with-urma-mock given, using URMA link-time mock" + fi + elif [ -n "$URMA_LIB" ]; then append_to_output_libs "$URMA_LIB" append_to_output "DYNAMIC_LINKINGS+=-lurma" append_to_output "URMA_USE_MOCK=0" + CPPFLAGS="${CPPFLAGS} -DBRPC_WITH_URMA_MOCK=0" else - append_to_output "URMA_USE_MOCK=1" - print_info "liburma not found; using URMA link-time mock" + >&2 $ECHO "Fail to find liburma. Install liburma, or explicitly opt into brpc's link-time mock with --with-urma-mock (the mock cannot talk to real URMA hardware; only use it for CI/tests without URMA hardware)." + exit 1 fi fi diff --git a/docs/cn/urma.md b/docs/cn/urma.md index a9be634ec7..f4bc5ec26b 100644 --- a/docs/cn/urma.md +++ b/docs/cn/urma.md @@ -20,7 +20,7 @@ WR。完成事件既可由 JFC 忙轮询获取，也可通过 JFCE 事件 fd 获 ### CMake 编译 ```bash -# 带 URMA 支持编译 brpc +# 带 URMA 支持编译 brpc（需要 liburma；无硬件/CI 场景见下方 mock 说明） cmake -B build -DWITH_URMA=ON make -C build -j$(nproc) @@ -30,11 +30,70 @@ cmake -B build make -C build -j$(nproc) ``` +未安装 `liburma` 时（例如 CI 环境），需显式开启链接期 mock，而不是依赖 +隐式回退： + +```bash +cmake -B build -DWITH_URMA=ON -DWITH_URMA_MOCK=ON +make -C build -j$(nproc) +``` + `WITH_URMA=ON` 使用上游 UMDK 头文件进行编译。CMake 优先使用系统安装的 SDK；找不到头文件时，会参照 Mooncake 的 mock 构建方式下载固定版本的 -UMDK，可通过 `DOWNLOAD_URMA_HEADERS=OFF` 禁止下载。找到 `liburma` 时使用 -真实硬件数据通路，否则链接 brpc 的 mock，使 URMA 代码和测试仍可在无硬件 -环境编译。 +UMDK，可通过 `DOWNLOAD_URMA_HEADERS=OFF` 禁止下载。库的选择遵循两条规则。未开启 +`WITH_URMA_MOCK` 时，链接 `liburma` 走真实硬件数据通路；找不到 +`liburma` 则直接报错终止构建，而不是静默回退到 mock——那样会掩盖环境 +问题，产出一个看似支持 URMA、实际无法访问真实硬件的产物。传入 +`-DWITH_URMA_MOCK=ON`（Make 对应 `config_brpc.sh --with-urma-mock`， +Bazel 对应 `--define BRPC_WITH_URMA_MOCK=true`）时，**无论是否装有 +`liburma` 都使用 mock**，且 CMake 在覆盖真实 `liburma` 时会打印警告。 +这个显式开关刻意优先于自动探测：若按「是否装了 liburma」来决定，同一条 +mock 构建命令在不同机器上含义就不同了——CI 上得到 mock，而装了 SDK 的 +开发机上却悄悄变成硬件构建。 + +### Bazel 编译 + +```bash +# 带 URMA 支持编译 brpc（链接真实 liburma，需系统已安装） +bazel build --define BRPC_WITH_URMA=true //:brpc + +# 无硬件/CI 场景：显式选择链接期 mock（不链接 liburma） +bazel build --define BRPC_WITH_URMA=true --define BRPC_WITH_URMA_MOCK=true //:brpc +``` + +`bazel/config:brpc_with_urma_mock` 是 `bazel/config:brpc_with_urma` 的 +一个特化（`define_values` 是后者的超集），因此同时传入两个 `--define` +时 Bazel 会按更具体的设置解析：编译期定义 `BRPC_WITH_URMA_MOCK=1` +并跳过 `-lurma` 链接；只传 `BRPC_WITH_URMA=true` 时才会链接真实 +`liburma`。这与 CMake/Make 侧「找不到 liburma 就必须显式加 +`--with-urma-mock`，否则直接报错」的语义等价，只是 Bazel 没有 +`find_library` 式的自动探测，需要由调用方显式指定其中一个。 + +Bazel 使用的 `@umdk` 仓库与 CMake 的头文件查找顺序保持一致： + +1. 如果设置了 `URMA_ROOT`，Bazel 会优先使用这个本地 UMDK 安装/源码树中的 + 头文件。支持系统安装布局（例如 + `$URMA_ROOT/ub/umdk/urma/urma_api.h`）和 UMDK 源码布局（例如 + `$URMA_ROOT/src/urma/lib/urma/core/include/urma_api.h`）。 +2. 如果未设置 `URMA_ROOT`，Bazel 才会从 + `https://atomgit.com/openeuler/umdk.git` 下载固定版本 + `564ee727a55523d4351a8fb3c94292b388ebb924`（即 `v26.06.0_CAM`）。 +3. 离线或 hermetic CI 中若要禁止该下载兜底，传入 + `--repo_env=BRPC_DOWNLOAD_URMA_HEADERS=false`；此时 `URMA_ROOT` 不可用会 + 在仓库解析阶段直接报错。 + +如果要用一个准备好的本地仓库或内部镜像直接替换 `@umdk`，使用 Bazel 的 +仓库覆盖。被覆盖的仓库必须提供兼容的 `@umdk//:urma_headers` target；原始 +UMDK 安装目录或源码 checkout 更适合走 `URMA_ROOT`。 + +```bash +bazel build --define BRPC_WITH_URMA=true \ + --override_repository=umdk=/path/to/local/umdk \ + //:brpc +``` + +`--distdir` 只服务于 `http_archive` / `http_file` 这类走 Bazel downloader +的规则；它不会满足 `@umdk` 的 git 下载兜底。 ## 使用 diff --git a/docs/en/urma.md b/docs/en/urma.md index 964f648bd2..48a42630d2 100644 --- a/docs/en/urma.md +++ b/docs/en/urma.md @@ -20,7 +20,7 @@ from a JFC either by busy polling or through a JFCE event fd. ### Build with CMake ```bash -# Build brpc with URMA support +# Build brpc with URMA support (requires liburma; see below for CI/mock builds) cmake -B build -DWITH_URMA=ON make -C build -j$(nproc) @@ -30,13 +30,76 @@ cmake -B build make -C build -j$(nproc) ``` +Without `liburma` installed (e.g. in CI), explicitly opt into the link-time +mock instead of relying on an implicit fallback: + +```bash +cmake -B build -DWITH_URMA=ON -DWITH_URMA_MOCK=ON +make -C build -j$(nproc) +``` + `WITH_URMA=ON` compiles against upstream UMDK headers. CMake prefers an installed SDK and, following Mooncake's mock setup, downloads a pinned UMDK release when the headers are unavailable. Set `DOWNLOAD_URMA_HEADERS=OFF` to disable downloading. -When `liburma` is found it is linked for the hardware data path. Otherwise, -brpc uses its link-time mock so URMA code and tests can still be built without -hardware. +Library selection follows two rules. Without `WITH_URMA_MOCK`, `liburma` is +linked for the hardware data path, and a missing `liburma` fails the build +rather than silently falling back to the mock -- that fallback would mask a +broken environment and ship a binary that looks URMA-capable but cannot +reach real hardware. With `-DWITH_URMA_MOCK=ON` +(`config_brpc.sh --with-urma-mock`, Bazel `--define BRPC_WITH_URMA_MOCK=true`), +the mock is used *whether or not* `liburma` is installed, and CMake prints a +warning when it overrides a real `liburma`. The explicit switch deliberately +wins over auto-detection: deciding by "is liburma installed?" would make a +mock build mean different things on different machines -- the mock in CI, a +hardware build on a developer box that happens to have the SDK. + +### Build with Bazel + +```bash +# Build brpc with URMA support (links the real liburma; must be installed) +bazel build --define BRPC_WITH_URMA=true //:brpc + +# No hardware / CI: explicitly select the link-time mock (does not link liburma) +bazel build --define BRPC_WITH_URMA=true --define BRPC_WITH_URMA_MOCK=true //:brpc +``` + +`bazel/config:brpc_with_urma_mock` is a specialization of +`bazel/config:brpc_with_urma` (its `define_values` is a superset), so when +both `--define`s are passed Bazel resolves selects to the more specific +setting: it compiles with `BRPC_WITH_URMA_MOCK=1` and skips the `-lurma` +link. Passing only `BRPC_WITH_URMA=true` links the real `liburma`. This +mirrors the CMake/Make behavior of failing the build unless `liburma` is +found or the mock is explicitly requested — Bazel just has no +`find_library`-style auto-detection, so the caller must pick one of the two +`--define`s explicitly. + +The `@umdk` repository used by Bazel mirrors CMake's header lookup order: + +1. If `URMA_ROOT` is set, Bazel uses headers from that local UMDK install/tree. + Supported layouts include an installed SDK such as + `$URMA_ROOT/ub/umdk/urma/urma_api.h` and a UMDK source checkout such as + `$URMA_ROOT/src/urma/lib/urma/core/include/urma_api.h`. +2. Otherwise Bazel downloads the pinned UMDK revision + `564ee727a55523d4351a8fb3c94292b388ebb924` (`v26.06.0_CAM`) from + `https://atomgit.com/openeuler/umdk.git`. +3. To forbid that fallback in offline or hermetic CI, pass + `--repo_env=BRPC_DOWNLOAD_URMA_HEADERS=false`; repository resolution will + fail if `URMA_ROOT` is not usable. + +For a prepared local repository or internal mirror that should replace `@umdk` +directly, use Bazel's repository override instead. The override target must +expose a compatible `@umdk//:urma_headers` target; use `URMA_ROOT` for a raw +UMDK install or source checkout. + +```bash +bazel build --define BRPC_WITH_URMA=true \ + --override_repository=umdk=/path/to/local/umdk \ + //:brpc +``` + +`--distdir` only feeds Bazel's downloader-based rules such as `http_archive` +and `http_file`; it does not satisfy the git fallback used by `@umdk`. ## Usage diff --git a/example/cmake/BrpcExample.cmake b/example/cmake/BrpcExample.cmake index 6b2c7850ff..9f6baaffcf 100644 --- a/example/cmake/BrpcExample.cmake +++ b/example/cmake/BrpcExample.cmake @@ -86,6 +86,7 @@ macro(brpc_example_find_common_deps out_libs) ) endif() + # Search for libthrift* by best effort. If it is not found and brpc is # compiled with thrift protocol enabled, a link error would be reported. find_library(THRIFT_LIB NAMES thrift) @@ -123,6 +124,14 @@ macro(brpc_example_find_common_deps out_libs) find_package(OpenSSL REQUIRED) + # brpc built with -DWITH_URMA=ON carries undefined urma_* symbols, so every + # example has to link liburma. Search by best effort: when brpc was built + # without URMA the symbols are absent and the library is not needed. + find_library(URMA_LIB NAMES urma) + if(NOT URMA_LIB) + set(URMA_LIB "") + endif() + set(_common_libs Threads::Threads ${GFLAGS_LIBRARY} @@ -132,6 +141,7 @@ macro(brpc_example_find_common_deps out_libs) ${OPENSSL_CRYPTO_LIBRARY} ${OPENSSL_SSL_LIBRARY} ${THRIFT_LIB} + ${URMA_LIB} dl ) diff --git a/example/urma_performance/CMakeLists.txt b/example/urma_performance/CMakeLists.txt index 154970fbd3..eae472fe90 100644 --- a/example/urma_performance/CMakeLists.txt +++ b/example/urma_performance/CMakeLists.txt @@ -26,13 +26,7 @@ brpc_example_find_common_deps(DYNAMIC_LIB) protobuf_generate_cpp(PROTO_SRC PROTO_HEADER test.proto) set(BRPC_EXAMPLE_WITH_URMA ON) -find_library(URMA_LIB NAMES urma) -if(URMA_LIB) - list(APPEND DYNAMIC_LIB ${URMA_LIB}) -else() - message(STATUS - "liburma not found; using the URMA implementation linked into brpc") -endif() + add_executable(urma_performance_client client.cpp ${PROTO_SRC} ${PROTO_HEADER}) brpc_example_configure_target(urma_performance_client) diff --git a/src/brpc/policy/rdma_handshake_protocol.cpp b/src/brpc/policy/rdma_handshake_protocol.cpp index 580abdda5d..84039f27f8 100644 --- a/src/brpc/policy/rdma_handshake_protocol.cpp +++ b/src/brpc/policy/rdma_handshake_protocol.cpp @@ -16,7 +16,11 @@ // under the License. #include "brpc/policy/rdma_handshake_protocol.h" +#include +#if BRPC_WITH_URMA +#include "brpc/urma/urma_handshake_server.h" +#endif #include "butil/logging.h" #include "brpc/destroyable.h" #include "brpc/rdma/rdma_handshake_server.h" @@ -26,6 +30,18 @@ namespace policy { ParseResult ParseRdmaHandshake(butil::IOBuf* source, Socket* socket, bool /*read_eof*/, const void* /*arg*/) { +#if BRPC_WITH_URMA + if (source->size() >= 4) { + char magic[4]; + source->copy_to(magic, sizeof(magic)); + + if (std::memcmp(magic, "URMA", sizeof(magic)) == 0 || + std::memcmp(magic, "URM3", sizeof(magic)) == 0) { + return urma::ExecuteFallbackServerHandshake(source, socket); + } + } +#endif + return rdma::ExecuteServerHandshake(source, socket); } diff --git a/src/brpc/urma/mock_urma.cpp b/src/brpc/urma/mock_urma.cpp index 266c08b19a..4de9e5a168 100644 --- a/src/brpc/urma/mock_urma.cpp +++ b/src/brpc/urma/mock_urma.cpp @@ -32,7 +32,7 @@ // - Device-name contract: device->name == "mock_urma_device" so tests can // match it with --urma_device=mock_urma_device. -#if BRPC_WITH_URMA +#if BRPC_WITH_URMA && BRPC_WITH_URMA_MOCK #include "urma_api.h" @@ -62,7 +62,14 @@ struct PendingRecv { uint64_t user_ctx; }; -std::shared_mutex g_rw_mutex; +// `std::shared_mutex` is a C++17 type, but brpc's Makefile build defaults to +// -std=c++14 (config_brpc.sh), so the mock must not depend on it. +// `std::shared_timed_mutex` has the same reader/writer semantics and is +// available since C++14, which keeps the mock buildable under every +// toolchain brpc supports. +using MockSharedMutex = std::shared_timed_mutex; + +MockSharedMutex g_rw_mutex; bool initialized = false; std::vector device_list; std::map context_map; @@ -93,7 +100,12 @@ void PushCompletion(urma_jfc_t* jfc, JfcState* state, if (signal && jfc && jfc->jfc_cfg.jfce && jfc->jfc_cfg.jfce->fd >= 0) { uint64_t one = 1; - (void)write(jfc->jfc_cfg.jfce->fd, &one, sizeof(one)); + // The eventfd counter only signals readiness; a short/failed write + // just means the waiter polls again. Swallow the result explicitly so + // -Wunused-result (glibc marks write() warn_unused_result) stays quiet + // under --werror builds. + ssize_t rc = write(jfc->jfc_cfg.jfce->fd, &one, sizeof(one)); + (void)rc; } } @@ -119,7 +131,7 @@ urma_eid_info_t mock_eid_info = { extern "C" { urma_status_t urma_init(urma_init_attr_t *init_attr) { - std::unique_lock lock(g_rw_mutex); + std::unique_lock lock(g_rw_mutex); if (initialized) { return URMA_EEXIST; } @@ -128,7 +140,7 @@ urma_status_t urma_init(urma_init_attr_t *init_attr) { } urma_status_t urma_uninit(void) { - std::unique_lock lock(g_rw_mutex); + std::unique_lock lock(g_rw_mutex); initialized = false; for (auto device : device_list) { delete device; @@ -153,7 +165,7 @@ urma_status_t urma_uninit(void) { urma_device_t **urma_get_device_list(int *num_devices) { { - std::shared_lock lock(g_rw_mutex); + std::shared_lock lock(g_rw_mutex); if (!initialized) { *num_devices = 0; return nullptr; @@ -168,7 +180,7 @@ urma_device_t **urma_get_device_list(int *num_devices) { } } { - std::unique_lock write_lock(g_rw_mutex); + std::unique_lock write_lock(g_rw_mutex); if (!initialized) { *num_devices = 0; return nullptr; @@ -193,7 +205,7 @@ urma_device_t **urma_get_device_list(int *num_devices) { urma_device_t *urma_get_device_by_name(char *dev_name) { { - std::shared_lock lock(g_rw_mutex); + std::shared_lock lock(g_rw_mutex); if (!initialized) { return nullptr; } @@ -207,7 +219,7 @@ urma_device_t *urma_get_device_by_name(char *dev_name) { } } { - std::unique_lock write_lock(g_rw_mutex); + std::unique_lock write_lock(g_rw_mutex); if (!initialized) { return nullptr; } @@ -265,7 +277,7 @@ void urma_free_eid_list(urma_eid_info_t *eid_list) { } urma_context_t *urma_create_context(urma_device_t *device, uint32_t eid_index) { - std::unique_lock lock(g_rw_mutex); + std::unique_lock lock(g_rw_mutex); if (!device) { return nullptr; } @@ -276,8 +288,20 @@ urma_context_t *urma_create_context(urma_device_t *device, uint32_t eid_index) { return ctx; } +// The bonding provider extension. brpc only calls this with +// BONDP_USER_CTL_SET_BONDING_MODE, and only for a device whose name marks it +// as a bonding device -- which the mock never reports. Accept the call so the +// symbol resolves in mock builds where the UMDK tree does ship urma_ubagg.h. +urma_status_t urma_user_ctl(urma_context_t *ctx, urma_user_ctl_in_t *in, + urma_user_ctl_out_t *out) { + if (ctx == nullptr || in == nullptr || out == nullptr) { + return URMA_EINVAL; + } + return URMA_SUCCESS; +} + urma_status_t urma_delete_context(urma_context_t *ctx) { - std::unique_lock lock(g_rw_mutex); + std::unique_lock lock(g_rw_mutex); if (!ctx || context_map.find(ctx) == context_map.end()) { return URMA_EINVAL; } @@ -287,7 +311,7 @@ urma_status_t urma_delete_context(urma_context_t *ctx) { } urma_jfce_t *urma_create_jfce(urma_context_t *ctx) { - std::unique_lock lock(g_rw_mutex); + std::unique_lock lock(g_rw_mutex); if (!ctx || context_map.find(ctx) == context_map.end()) { return nullptr; } @@ -305,7 +329,7 @@ urma_jfce_t *urma_create_jfce(urma_context_t *ctx) { } urma_status_t urma_delete_jfce(urma_jfce_t *jfce) { - std::unique_lock lock(g_rw_mutex); + std::unique_lock lock(g_rw_mutex); if (!jfce || jfce_map.find(jfce) == jfce_map.end()) { return URMA_EINVAL; } @@ -316,7 +340,7 @@ urma_status_t urma_delete_jfce(urma_jfce_t *jfce) { } urma_jfc_t *urma_create_jfc(urma_context_t *ctx, urma_jfc_cfg_t *cfg) { - std::unique_lock lock(g_rw_mutex); + std::unique_lock lock(g_rw_mutex); if (!ctx || !cfg || context_map.find(ctx) == context_map.end()) { return nullptr; } @@ -334,7 +358,7 @@ urma_jfc_t *urma_create_jfc(urma_context_t *ctx, urma_jfc_cfg_t *cfg) { } urma_status_t urma_delete_jfc(urma_jfc_t *jfc) { - std::unique_lock lock(g_rw_mutex); + std::unique_lock lock(g_rw_mutex); if (!jfc || jfc_state_map.find(jfc) == jfc_state_map.end()) { return URMA_EINVAL; } @@ -345,7 +369,7 @@ urma_status_t urma_delete_jfc(urma_jfc_t *jfc) { } urma_jfr_t *urma_create_jfr(urma_context_t *ctx, urma_jfr_cfg_t *cfg) { - std::unique_lock lock(g_rw_mutex); + std::unique_lock lock(g_rw_mutex); if (!ctx || !cfg || context_map.find(ctx) == context_map.end()) { return nullptr; } @@ -361,7 +385,7 @@ urma_jfr_t *urma_create_jfr(urma_context_t *ctx, urma_jfr_cfg_t *cfg) { } urma_status_t urma_delete_jfr(urma_jfr_t *jfr) { - std::unique_lock lock(g_rw_mutex); + std::unique_lock lock(g_rw_mutex); if (!jfr || jfr_map.find(jfr) == jfr_map.end()) { return URMA_EINVAL; } @@ -373,7 +397,7 @@ urma_status_t urma_delete_jfr(urma_jfr_t *jfr) { } urma_target_seg_t *urma_register_seg(urma_context_t *ctx, urma_seg_cfg_t *cfg) { - std::unique_lock lock(g_rw_mutex); + std::unique_lock lock(g_rw_mutex); if (!ctx || !cfg || context_map.find(ctx) == context_map.end()) { return nullptr; } @@ -389,7 +413,7 @@ urma_target_seg_t *urma_register_seg(urma_context_t *ctx, urma_seg_cfg_t *cfg) { } urma_status_t urma_unregister_seg(urma_target_seg_t *seg) { - std::unique_lock lock(g_rw_mutex); + std::unique_lock lock(g_rw_mutex); if (!seg || seg_map.find(seg) == seg_map.end()) { return URMA_EINVAL; } @@ -401,7 +425,7 @@ urma_status_t urma_unregister_seg(urma_target_seg_t *seg) { urma_target_seg_t *urma_import_seg(urma_context_t *ctx, urma_seg_t *seg, urma_token_t *token_value, uint64_t addr, urma_import_seg_flag_t flag) { - std::unique_lock lock(g_rw_mutex); + std::unique_lock lock(g_rw_mutex); if (!ctx || !seg || !token_value || context_map.find(ctx) == context_map.end()) { return nullptr; @@ -414,7 +438,7 @@ urma_target_seg_t *urma_import_seg(urma_context_t *ctx, urma_seg_t *seg, } urma_status_t urma_unimport_seg(urma_target_seg_t *tseg) { - std::unique_lock lock(g_rw_mutex); + std::unique_lock lock(g_rw_mutex); if (!tseg || seg_map.find(tseg) == seg_map.end()) { return URMA_EINVAL; } @@ -428,7 +452,7 @@ urma_status_t urma_get_async_event(urma_context_t *ctx, if (!ctx || !event) { return URMA_EINVAL; } - std::shared_lock lock(g_rw_mutex); + std::shared_lock lock(g_rw_mutex); if (context_map.find(ctx) == context_map.end()) { return URMA_EINVAL; } @@ -438,7 +462,7 @@ urma_status_t urma_get_async_event(urma_context_t *ctx, void urma_ack_async_event(urma_async_event_t *event) {} urma_jetty_t *urma_create_jetty(urma_context_t *ctx, urma_jetty_cfg_t *cfg) { - std::unique_lock lock(g_rw_mutex); + std::unique_lock lock(g_rw_mutex); if (!ctx || !cfg || context_map.find(ctx) == context_map.end()) { return nullptr; } @@ -455,7 +479,7 @@ urma_jetty_t *urma_create_jetty(urma_context_t *ctx, urma_jetty_cfg_t *cfg) { } urma_status_t urma_delete_jetty(urma_jetty_t *jetty) { - std::unique_lock lock(g_rw_mutex); + std::unique_lock lock(g_rw_mutex); if (!jetty || jetty_map.find(jetty) == jetty_map.end()) { return URMA_EINVAL; } @@ -466,7 +490,7 @@ urma_status_t urma_delete_jetty(urma_jetty_t *jetty) { } urma_status_t urma_unbind_jetty(urma_jetty_t *jetty) { - std::unique_lock lock(g_rw_mutex); + std::unique_lock lock(g_rw_mutex); if (!jetty || jetty_map.find(jetty) == jetty_map.end()) { return URMA_EINVAL; } @@ -477,7 +501,7 @@ urma_status_t urma_unbind_jetty(urma_jetty_t *jetty) { urma_target_jetty_t *urma_import_jetty(urma_context_t *ctx, urma_rjetty_t *rjetty, urma_token_t *token_value) { - std::unique_lock lock(g_rw_mutex); + std::unique_lock lock(g_rw_mutex); if (!ctx || !rjetty || !token_value || context_map.find(ctx) == context_map.end()) { return nullptr; @@ -490,7 +514,7 @@ urma_target_jetty_t *urma_import_jetty(urma_context_t *ctx, } urma_status_t urma_unimport_jetty(urma_target_jetty_t *tjetty) { - std::unique_lock lock(g_rw_mutex); + std::unique_lock lock(g_rw_mutex); if (!tjetty || target_jetty_map.find(tjetty) == target_jetty_map.end()) { return URMA_EINVAL; } @@ -501,7 +525,7 @@ urma_status_t urma_unimport_jetty(urma_target_jetty_t *tjetty) { urma_status_t urma_bind_jetty(urma_jetty_t *jetty, urma_target_jetty_t *tjetty) { - std::unique_lock lock(g_rw_mutex); + std::unique_lock lock(g_rw_mutex); if (!jetty || !tjetty || jetty_map.find(jetty) == jetty_map.end() || target_jetty_map.find(tjetty) == target_jetty_map.end()) { return URMA_EINVAL; @@ -511,7 +535,7 @@ urma_status_t urma_bind_jetty(urma_jetty_t *jetty, } urma_status_t urma_modify_jetty(urma_jetty_t *jetty, urma_jetty_attr_t *attr) { - std::shared_lock lock(g_rw_mutex); + std::shared_lock lock(g_rw_mutex); if (!jetty || !attr || jetty_map.find(jetty) == jetty_map.end()) { return URMA_EINVAL; } @@ -520,7 +544,7 @@ urma_status_t urma_modify_jetty(urma_jetty_t *jetty, urma_jetty_attr_t *attr) { urma_status_t urma_post_jetty_send_wr(urma_jetty_t *jetty, urma_jfs_wr_t *wr, urma_jfs_wr_t **bad_wr) { - std::shared_lock read_lock(g_rw_mutex); + std::shared_lock read_lock(g_rw_mutex); auto local_it = jetty_map.find(jetty); auto local_jfc_it = jetty ? jfc_state_map.find(jetty->jetty_cfg.jfs_cfg.jfc) @@ -555,7 +579,7 @@ urma_status_t urma_post_jetty_send_wr(urma_jetty_t *jetty, urma_jfs_wr_t *wr, PendingRecv recv{}; urma_jfc_t* remote_jfc = nullptr; { - std::unique_lock lock(g_rw_mutex); + std::unique_lock lock(g_rw_mutex); auto remote_it = jetty_id_map.find(current->tjetty->id.id); if (remote_it == jetty_id_map.end()) { continue; @@ -585,7 +609,7 @@ urma_status_t urma_post_jetty_send_wr(urma_jetty_t *jetty, urma_jfs_wr_t *wr, JfcState* remote_state = nullptr; { - std::shared_lock lock(g_rw_mutex); + std::shared_lock lock(g_rw_mutex); auto state_it = jfc_state_map.find(remote_jfc); if (state_it != jfc_state_map.end()) { remote_state = state_it->second; @@ -613,7 +637,7 @@ urma_status_t urma_post_jetty_send_wr(urma_jetty_t *jetty, urma_jfs_wr_t *wr, urma_status_t urma_post_jfr_wr(urma_jfr_t *jfr, urma_jfr_wr_t *wr, urma_jfr_wr_t **bad_wr) { - std::unique_lock lock(g_rw_mutex); + std::unique_lock lock(g_rw_mutex); auto recv_it = jfr_recv_map.find(jfr); if (!jfr || !wr || recv_it == jfr_recv_map.end()) { if (bad_wr) { @@ -644,7 +668,7 @@ urma_status_t urma_post_jetty_recv_wr(urma_jetty_t *jetty, urma_jfr_wr_t **bad_wr) { urma_jfr_t* shared_jfr = nullptr; { - std::shared_lock lock(g_rw_mutex); + std::shared_lock lock(g_rw_mutex); if (!jetty || jetty_map.find(jetty) == jetty_map.end()) { if (bad_wr) { *bad_wr = wr; @@ -659,7 +683,7 @@ urma_status_t urma_post_jetty_recv_wr(urma_jetty_t *jetty, int urma_poll_jfc(urma_jfc_t *jfc, int num_entries, urma_cr_t *cr_list) { JfcState* state = nullptr; { - std::shared_lock lock(g_rw_mutex); + std::shared_lock lock(g_rw_mutex); auto it = jfc_state_map.find(jfc); if (it == jfc_state_map.end()) { return -1; @@ -689,8 +713,12 @@ int urma_wait_jfc(urma_jfce_t* jfce, uint32_t jfc_cnt, int, return -1; } uint64_t value = 0; - (void)read(jfce->fd, &value, sizeof(value)); - std::shared_lock lock(g_rw_mutex); + // Drain the eventfd counter; the value itself is irrelevant because the + // completions are re-scanned below. See the write() note above for why the + // result is captured instead of cast to void. + ssize_t rc = read(jfce->fd, &value, sizeof(value)); + (void)rc; + std::shared_lock lock(g_rw_mutex); uint32_t count = 0; for (const auto& item : jfc_state_map) { if (count >= jfc_cnt || item.first->jfc_cfg.jfce != jfce) { @@ -710,4 +738,4 @@ void urma_ack_jfc(urma_jfc_t*[], uint32_t[], uint32_t) { } // extern "C" -#endif // BRPC_WITH_URMA +#endif // BRPC_WITH_URMA && BRPC_WITH_URMA_MOCK diff --git a/src/brpc/urma/urma_endpoint.cpp b/src/brpc/urma/urma_endpoint.cpp index 78aa225989..a4bd83ef9e 100644 --- a/src/brpc/urma/urma_endpoint.cpp +++ b/src/brpc/urma/urma_endpoint.cpp @@ -148,10 +148,13 @@ UrmaEndpoint::UrmaEndpoint(Socket* s) _state(UNINIT), _handshake_version(0), _resource(nullptr) { - _sq_size = static_cast( - std::max(16, std::min(4096, static_cast(FLAGS_urma_sq_size)))); - _rq_size = static_cast( - std::max(16, std::min(4096, static_cast(FLAGS_urma_rq_size)))); + // FLAGS_urma_sq_size / FLAGS_urma_rq_size are range-checked to [16, 4096] + // once in GlobalUrmaInitializeImpl() (urma_helper.cpp), which every path + // that can construct a UrmaEndpoint (UrmaTransport::Init(), gated by + // ContextInitOrDie()) runs before any endpoint exists. No need to + // re-clamp here. + _sq_size = static_cast(FLAGS_urma_sq_size); + _rq_size = static_cast(FLAGS_urma_rq_size); _read_butex = bthread::butex_create_checked>(); _read_butex->store(0, butil::memory_order_relaxed); } @@ -670,6 +673,11 @@ class UrmaIOBuf : private butil::IOBuf { } }; +// urma_jfs_cfg_t::max_sge is uint8_t (see urma_helper.cpp), so +// GetUrmaMaxSge() never exceeds this. Bound the on-stack SGE array by it +// instead of alloca()-ing a runtime-controlled size. +static constexpr int kUrmaMaxSgePerWr = 255; + ssize_t UrmaEndpoint::CutFromIOBufList(butil::IOBuf** from, size_t ndata) { if (!_resource || !_resource->jetty || !_resource->remote_jetty) { errno = ENOTCONN; @@ -678,14 +686,14 @@ ssize_t UrmaEndpoint::CutFromIOBufList(butil::IOBuf** from, size_t ndata) { int max_sge = GetUrmaMaxSge(); if (max_sge < 1) { max_sge = 1; + } else if (max_sge > kUrmaMaxSgePerWr) { + max_sge = kUrmaMaxSgePerWr; } - urma_sge_t* sglist = static_cast( - alloca(sizeof(urma_sge_t) * max_sge)); - if (!sglist) { - errno = ENOMEM; - return -1; - } + // Zero-initialize: cut_into_sglist() fills addr/len/tseg but not + // user_tseg, and a stack-garbage handle there is undefined behaviour + // waiting to happen. + urma_sge_t sglist[kUrmaMaxSgePerWr] = {}; size_t current = 0; ssize_t total_len = 0; @@ -705,6 +713,19 @@ ssize_t UrmaEndpoint::CutFromIOBufList(butil::IOBuf** from, size_t ndata) { size_t max_len = _remote_recv_block_size > 0 ? _remote_recv_block_size : GetUrmaRecvBlockSize(); + // The peer's recv block is only half the story: a single SEND must + // also fit the transport's max message size (4096B on CTP, which is + // what brpc advertises). urma_post_jetty_send_wr accepts an over-size + // WR and reports a successful completion, but the payload never + // arrives intact, so the peer's parser stalls and the RPC fails only + // on timeout with no error logged anywhere. Split here instead; the + // receiver already appends each completion to _read_buf in order, so + // the protocol parser reassembles the message exactly as it does for + // messages larger than one recv block. + const size_t msg_cap = GetUrmaMaxMsgSize(); + if (msg_cap > 0 && max_len > msg_cap) { + max_len = msg_cap; + } while (sge_index < static_cast(max_sge) && this_len < max_len && current < ndata) { auto* data = reinterpret_cast(from[current]); @@ -922,10 +943,52 @@ int UrmaEndpoint::SendAck(int num) { return 0; } +// urma_cr_status_t has no vendor-provided to-string helper, and the bare +// number is not much use in a bug report. Keep the spelling identical to +// urma_opcode.h so it can be grepped against the SDK. +static const char* UrmaCrStatusStr(urma_cr_status_t status) { + switch (status) { + case URMA_CR_SUCCESS: return "SUCCESS"; + case URMA_CR_UNSUPPORTED_OPCODE_ERR: return "UNSUPPORTED_OPCODE_ERR"; + case URMA_CR_LOC_LEN_ERR: return "LOC_LEN_ERR"; + case URMA_CR_LOC_OPERATION_ERR: return "LOC_OPERATION_ERR"; + case URMA_CR_LOC_ACCESS_ERR: return "LOC_ACCESS_ERR"; + case URMA_CR_REM_RESP_LEN_ERR: return "REM_RESP_LEN_ERR"; + case URMA_CR_REM_UNSUPPORTED_REQ_ERR: return "REM_UNSUPPORTED_REQ_ERR"; + case URMA_CR_REM_OPERATION_ERR: return "REM_OPERATION_ERR"; + case URMA_CR_REM_ACCESS_ABORT_ERR: return "REM_ACCESS_ABORT_ERR"; + case URMA_CR_ACK_TIMEOUT_ERR: return "ACK_TIMEOUT_ERR"; + case URMA_CR_RNR_RETRY_CNT_EXC_ERR: return "RNR_RETRY_CNT_EXC_ERR"; + case URMA_CR_WR_FLUSH_ERR: return "WR_FLUSH_ERR"; + case URMA_CR_WR_SUSPEND_DONE: return "WR_SUSPEND_DONE"; + case URMA_CR_WR_FLUSH_ERR_DONE: return "WR_FLUSH_ERR_DONE"; + case URMA_CR_WR_UNHANDLED: return "WR_UNHANDLED"; + case URMA_CR_LOC_DATA_POISON: return "LOC_DATA_POISON"; + case URMA_CR_REM_DATA_POISON: return "REM_DATA_POISON"; + default: return "UNKNOWN"; + } +} + ssize_t UrmaEndpoint::HandleCompletion(const urma_cr_t& cr) { bool zerocopy = FLAGS_urma_recv_zerocopy; if (cr.status != URMA_CR_SUCCESS) { - LOG(WARNING) << "URMA completion failed, status=" << cr.status; + // A bare status number cannot tell a send failure from a recv one, + // which is the first thing anyone needs to know here. + LOG(WARNING) << "URMA completion failed: status=" << cr.status + << " (" << UrmaCrStatusStr(cr.status) << ')' + << ", dir=" << (cr.flag.bs.s_r == 0 ? "send" : "recv") + << ", opcode=" << static_cast(cr.opcode) + << ", user_ctx=" << cr.user_ctx + << ", completion_len=" << cr.completion_len + << ", local_id=" << cr.local_id + << ", remote_jetty_id=" << cr.remote_id.id + << ", tpn=" << cr.tpn + << ", sq_window=" << _sq_window_size.load( + butil::memory_order_relaxed) + << ", remote_rq_window=" << _remote_rq_window_size.load( + butil::memory_order_relaxed) + << ", state=" << GetStateStr() + << " on " << _socket->description(); errno = EIO; return -1; } @@ -1216,52 +1279,51 @@ void UrmaEndpoint::OnNewDataFromTcp(Socket* m) { InputMessenger::OnNewMessages(m); return; } - int progress = 0; - while (true) { - const State state = - ep->_state.load(butil::memory_order_acquire); - if (state == UNINIT) { - if (!m->CreatedByConnect()) { - // Server side: kick off the handshake bthread. - if (!IsUrmaAvailable()) { - ep->_state = FALLBACK_TCP; - tp->_urma_state = UrmaTransport::URMA_OFF; - InputMessenger::OnNewMessages(m); - return; - } - SocketUniquePtr s; - m->ReAddress(&s); - ep->_state = S_HELLO_WAIT; - bthread_t tid; - bthread_attr_t attr = BTHREAD_ATTR_NORMAL; - bthread_attr_set_name(&attr, "UrmaServerHandshake"); - if (bthread_start_background(&tid, &attr, - ProcessHandshakeAtServer, ep) != 0) { - ep->_state = UNINIT; - LOG(FATAL) << "Fail to start UrmaServerHandshake bthread"; - } else { - s.release(); - } + const State state = ep->_state.load(butil::memory_order_acquire); + if (state == UNINIT) { + if (!m->CreatedByConnect()) { + // Server side: kick off the handshake bthread. + if (!IsUrmaAvailable()) { + ep->_state = FALLBACK_TCP; + tp->_urma_state = UrmaTransport::URMA_OFF; + InputMessenger::OnNewMessages(m); return; } - // Client side: handled by ProcessHandshakeAtClient. - return; - } else if (state < ESTABLISHED) { - // During handshake: wake the handshake bthread parked in ReadFromFd. - ep->_read_butex->fetch_add(1, butil::memory_order_release); - bthread::butex_wake(ep->_read_butex); - return; - } else if (state == FALLBACK_TCP) { - InputMessenger::OnNewMessages(m); - return; - } else if (state == ESTABLISHED) { - TryReadOnTcpDuringUrmaEst(m); + SocketUniquePtr s; + m->ReAddress(&s); + ep->_state = S_HELLO_WAIT; + bthread_t tid; + bthread_attr_t attr = BTHREAD_ATTR_NORMAL; + bthread_attr_set_name(&attr, "UrmaServerHandshake"); + if (bthread_start_background(&tid, &attr, + ProcessHandshakeAtServer, ep) != 0) { + ep->_state = UNINIT; + LOG(FATAL) << "Fail to start UrmaServerHandshake bthread"; + } else { + s.release(); + } return; } - if (!m->MoreReadEvents(&progress)) { - break; - } + // Client side: handled by ProcessHandshakeAtClient. + return; + } + if (state < ESTABLISHED) { + // During handshake: wake the handshake bthread parked in ReadFromFd. + ep->_read_butex->fetch_add(1, butil::memory_order_release); + bthread::butex_wake(ep->_read_butex); + return; + } + if (state == FALLBACK_TCP) { + InputMessenger::OnNewMessages(m); + return; + } + if (state == ESTABLISHED) { + TryReadOnTcpDuringUrmaEst(m); + return; } + // state == FAILED: FailHandshake() already called _socket->SetFailed(), + // which tears the socket down through the normal Socket path. There is + // nothing left for the edge-trigger dispatcher to do. } inline void UrmaEndpoint::TryReadOnTcp() { @@ -1613,7 +1675,7 @@ int UrmaEndpoint::PollingModeInitialize( if (!FLAGS_urma_use_polling) { return 0; } - if (tag >= _poller_groups.size() || + if (tag < 0 || static_cast(tag) >= _poller_groups.size() || _poller_groups[tag].pollers.empty()) { errno = EINVAL; return -1; @@ -1702,7 +1764,8 @@ int UrmaEndpoint::PollingModeInitialize( } void UrmaEndpoint::PollingModeRelease(bthread_tag_t tag) { - if (!FLAGS_urma_use_polling || tag >= _poller_groups.size()) { + if (!FLAGS_urma_use_polling || tag < 0 || + static_cast(tag) >= _poller_groups.size()) { return; } auto& group = _poller_groups[tag]; @@ -1720,7 +1783,8 @@ void UrmaEndpoint::PollerAddCqSid() { return; } _poller_tag = bthread_self_tag(); - if (_poller_tag >= _poller_groups.size()) { + if (_poller_tag < 0 || + static_cast(_poller_tag) >= _poller_groups.size()) { return; } auto& pollers = _poller_groups[_poller_tag].pollers; @@ -1735,7 +1799,8 @@ void UrmaEndpoint::PollerAddCqSid() { void UrmaEndpoint::PollerRemoveCqSid() { if (_cq_sid == INVALID_SOCKET_ID || _poller_groups.empty() || - _poller_tag >= _poller_groups.size()) { + _poller_tag < 0 || + static_cast(_poller_tag) >= _poller_groups.size()) { return; } auto& pollers = _poller_groups[_poller_tag].pollers; diff --git a/src/brpc/urma/urma_handshake_server.cpp b/src/brpc/urma/urma_handshake_server.cpp new file mode 100644 index 0000000000..1f16462ade --- /dev/null +++ b/src/brpc/urma/urma_handshake_server.cpp @@ -0,0 +1,163 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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. + +#include "brpc/urma/urma_handshake_server.h" + +#if BRPC_WITH_URMA + +#include +#include + +#include "butil/iobuf.h" +#include "butil/sys_byteorder.h" +#include "brpc/socket.h" +#include "brpc/urma/urma_handshake.h" +#include "brpc/urma/urma_handshake.pb.h" + +namespace brpc { +namespace urma { + +namespace { + +constexpr size_t URMA_MAGIC_LEN = 4; +constexpr size_t URMA_V3_SIZE_LEN = 4; +constexpr uint32_t URMA_V3_MAX_PB_SIZE = 4096; + +ParseResult SendFallbackHelloV2(butil::IOBuf* source, Socket* socket) { + constexpr size_t HEADER_LEN = URMA_MAGIC_LEN + sizeof(uint16_t); + + if (source->size() < HEADER_LEN) { + return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA); + } + + uint8_t header[HEADER_LEN]; + source->copy_to(header, sizeof(header)); + + uint16_t msg_len_be = 0; + std::memcpy(&msg_len_be, header + URMA_MAGIC_LEN, + sizeof(msg_len_be)); + const uint16_t msg_len = butil::NetToHost16(msg_len_be); + + if (msg_len < v2_wire::HELLO_MSG_LEN_MIN || + msg_len > v2_wire::HELLO_MSG_LEN_MAX) { + return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG); + } + + if (source->size() < msg_len) { + return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA); + } + source->pop_front(msg_len); + + v2_wire::HelloMessage reply{}; + reply.msg_len = v2_wire::HELLO_PACKET_LEN; + reply.hello_ver = 0; + reply.impl_ver = 0; + + uint8_t packet[v2_wire::HELLO_PACKET_LEN]; + std::memcpy(packet, "URMA", URMA_MAGIC_LEN); + reply.Serialize(packet + URMA_MAGIC_LEN); + + butil::IOBuf output; + output.append(packet, sizeof(packet)); + if (socket->Write(&output) != 0) { + return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG); + } + + return MakeParseError(PARSE_ERROR_TRY_OTHERS); +} +ParseResult SendFallbackHelloV3(butil::IOBuf* source, Socket* socket) { + constexpr size_t HEADER_LEN = URMA_MAGIC_LEN + URMA_V3_SIZE_LEN; + + if (source->size() < HEADER_LEN) { + return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA); + } + + uint8_t header[HEADER_LEN]; + source->copy_to(header, sizeof(header)); + + uint32_t pb_size_be = 0; + std::memcpy(&pb_size_be, header + URMA_MAGIC_LEN, + sizeof(pb_size_be)); + const uint32_t pb_size = butil::NetToHost32(pb_size_be); + + if (pb_size == 0 || pb_size > URMA_V3_MAX_PB_SIZE) { + return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG); + } + + const size_t total_size = HEADER_LEN + pb_size; + if (source->size() < total_size) { + return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA); + } + source->pop_front(total_size); + + UrmaHello reply; + reply.set_buffer_size(0); + reply.set_recv_buffer_cnt(0); + reply.set_jetty_id(0); + reply.set_eid(std::string(16, '\0')); + reply.set_uasid(0); + reply.set_tp_type(0); + reply.set_seg_eid(std::string(16, '\0')); + reply.set_seg_uasid(0); + reply.set_seg_va(0); + reply.set_seg_len(0); + reply.set_seg_token_id(0); + + std::string body; + if (!reply.SerializeToString(&body)) { + return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG); + } + + butil::IOBuf output; + output.append("URM3", URMA_MAGIC_LEN); + + const uint32_t reply_size_be = + butil::HostToNet32(static_cast(body.size())); + output.append(&reply_size_be, sizeof(reply_size_be)); + output.append(body); + + if (socket->Write(&output) != 0) { + return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG); + } + + return MakeParseError(PARSE_ERROR_TRY_OTHERS); +} + +} // namespace +ParseResult ExecuteFallbackServerHandshake( + butil::IOBuf* source, Socket* socket) { + if (source->size() < URMA_MAGIC_LEN) { + return MakeParseError(PARSE_ERROR_NOT_ENOUGH_DATA); + } + + char magic[URMA_MAGIC_LEN]; + source->copy_to(magic, sizeof(magic)); + + if (std::memcmp(magic, "URMA", URMA_MAGIC_LEN) == 0) { + return SendFallbackHelloV2(source, socket); + } + + if (std::memcmp(magic, "URM3", URMA_MAGIC_LEN) == 0) { + return SendFallbackHelloV3(source, socket); + } + + return MakeParseError(PARSE_ERROR_TRY_OTHERS); +} +} // namespace urma +} // namespace brpc + +#endif // BRPC_WITH_URMA \ No newline at end of file diff --git a/src/brpc/urma/urma_handshake_server.h b/src/brpc/urma/urma_handshake_server.h new file mode 100644 index 0000000000..7ca2d53789 --- /dev/null +++ b/src/brpc/urma/urma_handshake_server.h @@ -0,0 +1,39 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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 BRPC_URMA_URMA_HANDSHAKE_SERVER_H +#define BRPC_URMA_URMA_HANDSHAKE_SERVER_H + +#include "brpc/parse_result.h" + +namespace butil { +class IOBuf; +} + +namespace brpc { + +class Socket; + +namespace urma { + +ParseResult ExecuteFallbackServerHandshake( + butil::IOBuf* source, Socket* socket); + +} // namespace urma +} // namespace brpc + +#endif // BRPC_URMA_URMA_HANDSHAKE_SERVER_H \ No newline at end of file diff --git a/src/brpc/urma/urma_helper.cpp b/src/brpc/urma/urma_helper.cpp index 648316f1ae..67943b9c82 100644 --- a/src/brpc/urma/urma_helper.cpp +++ b/src/brpc/urma/urma_helper.cpp @@ -79,6 +79,11 @@ DEFINE_int32(urma_zerocopy_min_size, 512, DEFINE_string(urma_device, "", "The name of the URMA device to use. Empty means the first one."); +DEFINE_int32(urma_max_msg_size, 0, + "Largest payload a single URMA SEND WR may carry, in bytes. " + "0 means the built-in default (the CTP message limit), and any " + "value is still clamped by the device's max_msg_size. Raise it " + "only if the transport in use really carries larger messages."); DEFINE_int32(urma_max_sge, 0, "Max SGEs per WR. 0 means the device maximum."); DEFINE_int32(urma_bonding_mode, 0, @@ -118,6 +123,15 @@ static bool g_has_local_eid = false; static urma_device_attr_t g_device_attr{}; static int g_max_sge = 1; static size_t g_recv_block_size = 8 * 1024; +// brpc always advertises URMA_CTP (see MakeLocalParsedHello) and selects the +// CTP jetty priority, and UMDK caps a single CTP message at 4096 bytes +// (UMQ_CTP_MAX_BUF_SIZE in src/urpc/umq/umq_ub/core/private/umq_ub.c). A +// larger SEND is not rejected by urma_post_jetty_send_wr -- it is silently +// dropped or truncated, so the peer's parser waits forever for the rest of a +// message that never arrives and the RPC only fails on timeout, with clean +// logs on both sides. The send path must therefore split at this bound. +static const size_t URMA_CTP_MAX_MSG_SIZE = 4096; +static size_t g_max_msg_size = URMA_CTP_MAX_MSG_SIZE; static bool g_is_bonding_device = false; // Prefer the device capability table and retain priority 6 as a compatibility // fallback for CTP providers that do not report a priority. @@ -582,6 +596,19 @@ static bool GlobalUrmaInitializeImpl() { static_cast(FLAGS_urma_buffer_size) - sizeof(butil::IOBuf::Block); + // Never post a SEND larger than the transport can actually carry. 0 means + // "use the built-in default", which is the CTP message limit clamped by + // whatever the device reports. + if (FLAGS_urma_max_msg_size > 0) { + g_max_msg_size = static_cast(FLAGS_urma_max_msg_size); + } else { + g_max_msg_size = URMA_CTP_MAX_MSG_SIZE; + } + const uint64_t dev_max_msg = g_device_attr.dev_cap.max_msg_size; + if (dev_max_msg > 0 && dev_max_msg < g_max_msg_size) { + g_max_msg_size = static_cast(dev_max_msg); + } + // User-segment table. g_user_segs_lock = new (std::nothrow) butil::Mutex; g_user_segs = new (std::nothrow) butil::FlatMap(); @@ -630,7 +657,9 @@ static bool GlobalUrmaInitializeImpl() { << " bonding=" << g_is_bonding_device << " max_sge=" << g_max_sge << " buffer_size=" << g_pool_buffer_size - << " buffer_count=" << g_pool->buffer_count(); + << " buffer_count=" << g_pool->buffer_count() + << " max_msg_size=" << g_max_msg_size + << " dev_max_msg_size=" << g_device_attr.dev_cap.max_msg_size; return true; } @@ -688,6 +717,7 @@ int FindUrmaPriorityForTpType(const urma_device_attr_t& attr, } uint8_t GetUrmaJettyPriority() { return g_jetty_priority; } int GetUrmaMaxSge() { return g_max_sge; } +size_t GetUrmaMaxMsgSize() { return g_max_msg_size; } size_t GetUrmaRecvBlockSize() { return g_recv_block_size; } // ============================================================================ diff --git a/src/brpc/urma/urma_helper.h b/src/brpc/urma/urma_helper.h index 871f1a6354..d31c77c833 100644 --- a/src/brpc/urma/urma_helper.h +++ b/src/brpc/urma/urma_helper.h @@ -96,6 +96,11 @@ bool SupportedByUrma(const std::string& protocol); // Return the configured recv buffer size (one URMA recv WR's payload size). size_t GetUrmaRecvBlockSize(); +// Largest payload a single URMA SEND WR may carry: the smaller of the +// device's max_msg_size and the CTP transport's 4096-byte message limit. +// Sends above this bound are silently dropped, not rejected. +size_t GetUrmaMaxMsgSize(); + // Return max_sge supported by the device. int GetUrmaMaxSge(); diff --git a/src/brpc/urma_transport.cpp b/src/brpc/urma_transport.cpp index 311786e8eb..5c76b64006 100644 --- a/src/brpc/urma_transport.cpp +++ b/src/brpc/urma_transport.cpp @@ -58,6 +58,12 @@ void UrmaTransport::Init(Socket* socket, const SocketOptions& options) { _default_connect = options.app_connect; _on_edge_trigger = options.on_edge_triggered_events; if (options.need_on_edge_trigger && _on_edge_trigger == nullptr) { + // Unlike RDMA (which only wires OnNewDataFromTcp for client sockets + // and relies on InputMessenger::OnNewMessages / ParseRdmaHandshake + // for the server side), URMA uses OnNewDataFromTcp for both roles: + // its UNINIT branch itself dispatches on m->CreatedByConnect(), + // starting the server handshake bthread directly instead of going + // through a protocol-level handshake message. _on_edge_trigger = urma::UrmaEndpoint::OnNewDataFromTcp; } _tcp_transport = std::make_shared(); @@ -87,8 +93,14 @@ std::shared_ptr UrmaTransport::Connect() { } int UrmaTransport::CutFromIOBuf(butil::IOBuf* buf) { + // Only send over the URMA channel once the handshake has NEGOTIATED it + // (URMA_ON). While the state is still URMA_UNKNOWN (handshake in + // progress, or a server connection that turned out to be plain TCP and + // never handshook) or URMA_OFF (fell back), _resource is not yet set up + // and everything must go over the TCP fd. Mirrors the URMA_ON check in + // WaitEpollOut() and RDMA's equivalent check. if (_urma_ep && - _urma_state.load(butil::memory_order_acquire) != URMA_OFF) { + _urma_state.load(butil::memory_order_acquire) == URMA_ON) { butil::IOBuf* data_arr[1] = {buf}; return _urma_ep->CutFromIOBufList(data_arr, 1); } else { @@ -98,7 +110,7 @@ int UrmaTransport::CutFromIOBuf(butil::IOBuf* buf) { ssize_t UrmaTransport::CutFromIOBufList(butil::IOBuf** buf, size_t ndata) { if (_urma_ep && - _urma_state.load(butil::memory_order_acquire) != URMA_OFF) { + _urma_state.load(butil::memory_order_acquire) == URMA_ON) { return _urma_ep->CutFromIOBufList(buf, ndata); } else { return _tcp_transport->CutFromIOBufList(buf, ndata); diff --git a/test/brpc_urma_unittest.cpp b/test/brpc_urma_unittest.cpp index d05784d009..c97e9a1fc7 100644 --- a/test/brpc_urma_unittest.cpp +++ b/test/brpc_urma_unittest.cpp @@ -22,6 +22,11 @@ #if BRPC_WITH_URMA #include "butil/atomicops.h" #include "butil/sys_byteorder.h" +#include "brpc/channel.h" +#include "brpc/closure_guard.h" +#include "brpc/controller.h" +#include "brpc/server.h" +#include "echo.pb.h" #include "urma_api.h" #include "brpc/urma/urma_handshake.h" #include "brpc/urma/urma_handshake.pb.h" @@ -39,7 +44,16 @@ extern butil::atomic g_urma_available; } // namespace urma } // namespace brpc - +class UrmaFallbackEchoService : public test::EchoService { +public: + void Echo(google::protobuf::RpcController*, + const test::EchoRequest* request, + test::EchoResponse* response, + google::protobuf::Closure* done) override { + brpc::ClosureGuard done_guard(done); + response->set_message(request->message()); + } +}; // --------------------------------------------------------------------------- // v2 binary HelloMessage: serialize + deserialize round-trips. // --------------------------------------------------------------------------- @@ -212,8 +226,10 @@ TEST(UrmaHandshakeTest, ack_bit_is_urma_ok) { // ParsedHello field layout: covers the flattened segment (seg_* fields). // --------------------------------------------------------------------------- TEST(UrmaHandshakeTest, parsed_hello_segment_fields) { + // ParsedHello zero-initializes every field with default member + // initializers, so memset() is redundant here -- and GCC rejects it under + // --werror (-Wclass-memaccess) because the type is not trivial. urma::ParsedHello p; - std::memset(&p, 0, sizeof(p)); p.buffer_size = 8192; p.recv_buffer_cnt = 127; p.jetty_id = 42; @@ -263,6 +279,7 @@ TEST(UrmaHelperTest, selects_priority_matching_transport_path_type) { // --------------------------------------------------------------------------- // SupportedByUrma: only baidu_std. // --------------------------------------------------------------------------- + TEST(UrmaHandshakeTest, supported_by_urma_protocol_allowlist) { EXPECT_TRUE(urma::SupportedByUrma("baidu_std")); EXPECT_FALSE(urma::SupportedByUrma("http")); @@ -589,7 +606,51 @@ TEST_F(UrmaMockTest, urma_delete_context(ctx); urma_free_device_list(devices); } +TEST(UrmaHandshakeTest, v2_client_falls_back_to_tcp_server) { + urma::g_skip_urma_init = false; + + UrmaFallbackEchoService service; + brpc::Server server; + ASSERT_EQ(0, server.AddService( + &service, brpc::SERVER_DOESNT_OWN_SERVICE)); + + brpc::ServerOptions server_options; + server_options.socket_mode = brpc::SOCKET_MODE_TCP; + server_options.internal_port = -1; + ASSERT_EQ(0, server.Start(0, &server_options)); + + const int saved_version = + urma::FLAGS_urma_client_handshake_version; + urma::FLAGS_urma_client_handshake_version = 2; + brpc::ChannelOptions channel_options; + channel_options.socket_mode = brpc::SOCKET_MODE_URMA; + channel_options.connect_timeout_ms = 1000; + channel_options.timeout_ms = 3000; + channel_options.max_retry = 0; + + brpc::Channel channel; + const int init_result = + channel.Init(server.listen_address(), &channel_options); + EXPECT_EQ(0, init_result); + + if (init_result == 0) { + test::EchoRequest request; + test::EchoResponse response; + brpc::Controller cntl; + request.set_message("urma-fallback"); + + test::EchoService::Stub stub(&channel); + stub.Echo(&cntl, &request, &response, nullptr); + + EXPECT_FALSE(cntl.Failed()) << cntl.ErrorText(); + EXPECT_EQ(request.message(), response.message()); + } + + urma::FLAGS_urma_client_handshake_version = saved_version; + server.Stop(0); + server.Join(); +} #else // BRPC_WITH_URMA // When URMA is not compiled in, the test file is a no-op so the build stays