diff --git a/.github/workflows/urma-stability.yml b/.github/workflows/urma-stability.yml new file mode 100644 index 0000000000..68feb79bc6 --- /dev/null +++ b/.github/workflows/urma-stability.yml @@ -0,0 +1,52 @@ +# 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. + +name: URMA stability +on: + pull_request: + paths: + - 'src/brpc/urma/**' + - 'src/brpc/urma_transport.*' + - 'test/brpc_urma*' + - 'test/CMakeLists.txt' + - 'tools/urma_stability.py' + - '.github/workflows/urma-stability.yml' + workflow_dispatch: + +permissions: + contents: read + +jobs: + mock: + runs-on: ubuntu-24.04 + timeout-minutes: 20 + steps: + - uses: actions/checkout@v4 + - name: Install dependencies + run: | + sudo apt-get update + sudo apt-get install -y cmake ninja-build libssl-dev libgflags-dev \ + libprotobuf-dev protobuf-compiler libleveldb-dev libsnappy-dev \ + zlib1g-dev libgtest-dev + - name: Build + run: | + cmake -S . -B build -G Ninja -DWITH_URMA=ON -DURMA_USE_MOCK=ON \ + -DBUILD_UNIT_TESTS=ON -DDOWNLOAD_GTEST=OFF \ + -DBRPC_SYSTEM_GTEST_SOURCE_DIR=/usr/src/googletest -DBUILD_BRPC_TOOLS=OFF + cmake --build build --target brpc_urma_unittest brpc_urma_stability_unittest -j2 + - name: Run fault and data-path regressions + run: ctest --test-dir build -R '^(brpc_urma_unittest|brpc_urma_stability_unittest|urma_stability_.*)$' --output-on-failure diff --git a/docs/cn/urma.md b/docs/cn/urma.md index 62e10389a8..7de680a3e0 100644 --- a/docs/cn/urma.md +++ b/docs/cn/urma.md @@ -193,3 +193,16 @@ UrmaTransport 推荐用于**大包和跨节点**高吞吐路径，而 UBRing `ContextInitOrDie` 阶段拒绝。 - 硬件数据通路需要受支持的 UMDK provider 和 `liburma`。 - 当前实现面向 Linux。 + +## 稳定性测试 + +URMA mock 构建可运行故障注入和端到端测试： + +```sh +ctest --test-dir build -R '^(brpc_urma_stability_unittest|urma_stability_.*)$' --output-on-failure +python3 tools/urma_stability.py build/test/brpc_urma_stability_unittest --connections 1024 --rounds 100 +``` + +覆盖 credit/SQ 延迟、发送与接收重投失败、异常完成、断连、资源重建及注册池耗尽。 +规模测试维持独立 URMA 连接，逐请求校验混合大小消息，依次运行事件、轮询和轮询内联配置。 +使用真实 `liburma` 构建的测试程序可在硬件上运行同一规模测试，通过 `--device` 选择设备。 diff --git a/docs/en/urma.md b/docs/en/urma.md index a5b80f45b9..d11d2d2012 100644 --- a/docs/en/urma.md +++ b/docs/en/urma.md @@ -207,3 +207,18 @@ is tracked as a future enhancement. rejected at `ContextInitOrDie`. - A hardware data path requires a supported UMDK provider and `liburma`. - The current implementation targets Linux. + +## Stability tests + +Run fault injection and end-to-end tests with a URMA mock build: + +```sh +ctest --test-dir build -R '^(brpc_urma_stability_unittest|urma_stability_.*)$' --output-on-failure +python3 tools/urma_stability.py build/test/brpc_urma_stability_unittest --connections 1024 --rounds 100 +``` + +Coverage includes credit/SQ delays, failed sends and receive reposts, completion errors, +disconnects, resource recreation and registered-pool exhaustion. The scale test maintains +independent URMA connections and checks each mixed-size response in event, polling and +inline polling modes. A test binary linked with real `liburma` runs the same scale test +on hardware; use `--device` to select a device. diff --git a/src/brpc/urma/mock_urma.cpp b/src/brpc/urma/mock_urma.cpp index b0572c2e86..5e63003ea8 100644 --- a/src/brpc/urma/mock_urma.cpp +++ b/src/brpc/urma/mock_urma.cpp @@ -34,7 +34,10 @@ #if BRPC_WITH_URMA -#include "urma_api.h" +#include "brpc/urma/mock_urma.h" + +#include +#include #include #include @@ -44,10 +47,10 @@ #include #include #include -#include -#include #include +#include "urma_api.h" + namespace { struct JfcState { @@ -79,6 +82,28 @@ std::map jetty_id_map; std::map target_jetty_map; std::atomic next_jetty_id{1}; +struct DeferredCompletion { + urma_jfc_t *jfc; + urma_cr_t completion; +}; +struct FaultState { + brpc::urma::MockFault fault; + int remaining; + std::vector deferred; +}; +std::map faults; + +bool TakeFault(urma_jetty_t *jetty, brpc::urma::MockFault fault) { + auto it = faults.find(jetty); + if (it == faults.end() || it->second.fault != fault || it->second.remaining == 0) { + return false; + } + if (it->second.remaining > 0) { + --it->second.remaining; + } + return true; +} + void PushCompletion(urma_jfc_t* jfc, JfcState* state, const urma_cr_t& completion) { bool signal = false; @@ -118,6 +143,43 @@ urma_eid_info_t mock_eid_info = { extern "C" { +void brpc_urma_mock_fault(urma_jetty_t *jetty, brpc::urma::MockFault fault, int count) { + std::unique_lock lock(g_rw_mutex); + if (jetty_map.count(jetty)) { + auto &state = faults[jetty]; + state.fault = fault; + state.remaining = count; + } +} + +size_t brpc_urma_mock_release(urma_jetty_t *jetty) { + std::unique_lock lock(g_rw_mutex); + auto it = faults.find(jetty); + if (it == faults.end()) { + return 0; + } + size_t released = 0; + for (const auto &pending : it->second.deferred) { + auto jfc = jfc_state_map.find(pending.jfc); + if (jfc != jfc_state_map.end()) { + PushCompletion(jfc->first, jfc->second, pending.completion); + ++released; + } + } + faults.erase(it); + return released; +} + +brpc::urma::MockResources brpc_urma_mock_resources() { + std::shared_lock lock(g_rw_mutex); + size_t deferred = 0; + for (const auto &entry : faults) { + deferred += entry.second.deferred.size(); + } + return {jetty_map.size(), jfc_state_map.size(), jfr_map.size(), jfce_map.size(), + target_jetty_map.size(), seg_map.size(), deferred}; +} + urma_status_t urma_init(urma_init_attr_t *init_attr) { std::unique_lock lock(g_rw_mutex); if (initialized) { @@ -148,6 +210,7 @@ urma_status_t urma_uninit(void) { jetty_id_map.clear(); target_jetty_map.clear(); next_jetty_id.store(1); + faults.clear(); return URMA_SUCCESS; } @@ -351,6 +414,12 @@ urma_status_t urma_delete_jfc(urma_jfc_t *jfc) { if (!jfc || jfc_state_map.find(jfc) == jfc_state_map.end()) { return URMA_EINVAL; } + for (auto &entry : faults) { + auto &pending = entry.second.deferred; + pending.erase(std::remove_if(pending.begin(), pending.end(), + [jfc](const DeferredCompletion &c) { return c.jfc == jfc; }), + pending.end()); + } delete jfc_state_map[jfc]; jfc_state_map.erase(jfc); delete jfc; @@ -472,6 +541,7 @@ urma_status_t urma_delete_jetty(urma_jetty_t *jetty) { if (!jetty || jetty_map.find(jetty) == jetty_map.end()) { return URMA_EINVAL; } + faults.erase(jetty); jetty_id_map.erase(jetty->jetty_id.id); jetty_map.erase(jetty); delete jetty; @@ -533,94 +603,76 @@ 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); - auto local_it = jetty_map.find(jetty); - auto local_jfc_it = - jetty ? jfc_state_map.find(jetty->jetty_cfg.jfs_cfg.jfc) - : jfc_state_map.end(); - if (!jetty || !wr || local_it == jetty_map.end() || - local_jfc_it == jfc_state_map.end()) { - if (bad_wr) { - *bad_wr = wr; - } + // Keep resources alive until source SGEs have been consumed and completion + // records published. A completion permits the caller to recycle its SGEs. + std::unique_lock lock(g_rw_mutex); + auto local = jetty_map.find(jetty); + if (local == jetty_map.end() || !wr) { + if (bad_wr) *bad_wr = wr; return URMA_EINVAL; } - JfcState* local_state = local_jfc_it->second; - read_lock.unlock(); - - for (urma_jfs_wr_t* current = wr; current; current = current->next) { - if (!current->tjetty) { - if (bad_wr) { - *bad_wr = current; - } + auto local_jfc = jfc_state_map.find(jetty->jetty_cfg.jfs_cfg.jfc); + if (local_jfc == jfc_state_map.end()) { + if (bad_wr) *bad_wr = wr; + return URMA_EINVAL; + } + for (auto *current = wr; current; current = current->next) { + if (!current->tjetty || !target_jetty_map.count(current->tjetty)) { + if (bad_wr) *bad_wr = current; return URMA_EINVAL; } - - urma_cr_t send_cr{}; - send_cr.status = URMA_CR_SUCCESS; - send_cr.user_ctx = current->user_ctx; - send_cr.flag.bs.s_r = 0; - if (current->flag.bs.complete_enable) { - PushCompletion(jetty->jetty_cfg.jfs_cfg.jfc, - local_state, send_cr); + const bool credit = current->opcode == URMA_OPC_SEND_IMM; + if (TakeFault(jetty, brpc::urma::MOCK_FAIL_SEND) || + (credit && TakeFault(jetty, brpc::urma::MOCK_FAIL_CREDIT))) { + if (bad_wr) *bad_wr = current; + errno = EIO; + return static_cast(EIO); } - - PendingRecv recv{}; - urma_jfc_t* remote_jfc = nullptr; - { - 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; - } - urma_jfr_t* remote_jfr = - remote_it->second->jetty_cfg.shared.jfr; - auto recv_it = jfr_recv_map.find(remote_jfr); - auto jfc_it = jfr_jfc_map.find(remote_jfr); - if (recv_it == jfr_recv_map.end() || recv_it->second.empty() || - jfc_it == jfr_jfc_map.end()) { - continue; + const bool error = TakeFault(jetty, brpc::urma::MOCK_ERROR_COMPLETION); + auto remote = jetty_id_map.find(current->tjetty->id.id); + if (!error && remote != jetty_id_map.end()) { + auto *jfr = remote->second->jetty_cfg.shared.jfr; + auto recv_it = jfr_recv_map.find(jfr); + auto jfc_it = jfr_jfc_map.find(jfr); + if (recv_it != jfr_recv_map.end() && !recv_it->second.empty() && + jfc_it != jfr_jfc_map.end()) { + const auto recv = recv_it->second.front(); + recv_it->second.pop_front(); + uint32_t copied = 0; + for (uint32_t i = 0; i < current->send.src.num_sge && copied < recv.len; ++i) { + const auto &sge = current->send.src.sge[i]; + const uint32_t n = std::min(sge.len, recv.len - copied); + std::memcpy(reinterpret_cast(recv.addr + copied), + reinterpret_cast(sge.addr), n); + copied += n; + } + urma_cr_t cr{}; + cr.status = URMA_CR_SUCCESS; + cr.user_ctx = recv.user_ctx; + cr.flag.bs.s_r = 1; + cr.completion_len = copied; + cr.opcode = credit ? URMA_CR_OPC_SEND_WITH_IMM : URMA_CR_OPC_SEND; + cr.imm_data = current->send.imm_data; + if (credit && TakeFault(jetty, brpc::urma::MOCK_HOLD_CREDIT)) { + faults[jetty].deferred.push_back({jfc_it->second, cr}); + } else { + PushCompletion(jfc_it->second, jfc_state_map.at(jfc_it->second), cr); + } } - recv = recv_it->second.front(); - recv_it->second.pop_front(); - remote_jfc = jfc_it->second; } - - uint32_t copied = 0; - for (uint32_t i = 0; - i < current->send.src.num_sge && copied < recv.len; ++i) { - const urma_sge_t& sge = current->send.src.sge[i]; - const uint32_t n = std::min(sge.len, recv.len - copied); - std::memcpy(reinterpret_cast(recv.addr + copied), - reinterpret_cast(sge.addr), n); - copied += n; - } - - JfcState* remote_state = nullptr; - { - 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; + if (current->flag.bs.complete_enable) { + urma_cr_t cr{}; + cr.status = error ? URMA_CR_LOC_ACCESS_ERR : URMA_CR_SUCCESS; + cr.user_ctx = current->user_ctx; + cr.flag.bs.s_r = 0; + if (TakeFault(jetty, brpc::urma::MOCK_HOLD_SEND_COMPLETION)) { + faults[jetty].deferred.push_back({local_jfc->first, cr}); + } else { + PushCompletion(local_jfc->first, local_jfc->second, cr); } } - if (remote_state) { - urma_cr_t recv_cr{}; - recv_cr.status = URMA_CR_SUCCESS; - recv_cr.user_ctx = recv.user_ctx; - recv_cr.flag.bs.s_r = 1; - recv_cr.completion_len = copied; - recv_cr.opcode = - current->opcode == URMA_OPC_SEND_IMM - ? URMA_CR_OPC_SEND_WITH_IMM - : URMA_CR_OPC_SEND; - recv_cr.imm_data = current->send.imm_data; - PushCompletion(remote_jfc, remote_state, recv_cr); - } - } - if (bad_wr) { - *bad_wr = nullptr; } + if (bad_wr) *bad_wr = nullptr; return URMA_SUCCESS; } @@ -634,6 +686,14 @@ urma_status_t urma_post_jfr_wr(urma_jfr_t *jfr, urma_jfr_wr_t *wr, } return URMA_EINVAL; } + for (const auto &entry : jetty_map) { + if (entry.first->jetty_cfg.shared.jfr == jfr && + TakeFault(entry.first, brpc::urma::MOCK_FAIL_RECV)) { + if (bad_wr) *bad_wr = wr; + errno = ENOMEM; + return URMA_ENOMEM; + } + } for (urma_jfr_wr_t* current = wr; current; current = current->next) { if (current->src.num_sge == 0 || !current->src.sge) { if (bad_wr) { @@ -670,16 +730,13 @@ 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); - auto it = jfc_state_map.find(jfc); - if (it == jfc_state_map.end()) { - return -1; - } - state = it->second; + std::shared_lock lock(g_rw_mutex); + auto it = jfc_state_map.find(jfc); + if (it == jfc_state_map.end()) { + return -1; } - std::lock_guard lock(state->mutex); + auto *state = it->second; + std::lock_guard state_lock(state->mutex); int count = 0; while (count < num_entries && !state->completions.empty()) { cr_list[count++] = state->completions.front(); diff --git a/src/brpc/urma/mock_urma.h b/src/brpc/urma/mock_urma.h new file mode 100644 index 0000000000..4dcb3eca4f --- /dev/null +++ b/src/brpc/urma/mock_urma.h @@ -0,0 +1,56 @@ +// 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_MOCK_URMA_H +#define BRPC_URMA_MOCK_URMA_H + +#include + +#include "urma_api.h" + +namespace brpc { +namespace urma { +enum MockFault { + MOCK_FAIL_SEND, + MOCK_FAIL_CREDIT, + MOCK_HOLD_CREDIT, + MOCK_HOLD_SEND_COMPLETION, + MOCK_FAIL_RECV, + MOCK_ERROR_COMPLETION +}; + +struct MockResources { + size_t jettys; + size_t jfcs; + size_t jfrs; + size_t jfces; + size_t targets; + size_t segments; + size_t deferred; +}; +} // namespace urma +} // namespace brpc + +// These entry points exist only in the link-time mock. Faults are scoped to +// a jetty; count=-1 applies until release, otherwise count matching WRs. +extern "C" { +void brpc_urma_mock_fault(urma_jetty_t*, brpc::urma::MockFault, int count); +size_t brpc_urma_mock_release(urma_jetty_t*); +brpc::urma::MockResources brpc_urma_mock_resources(); +} + +#endif // BRPC_URMA_MOCK_URMA_H diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index a0fc2d0536..9391810b1b 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -321,3 +321,22 @@ if(BUILD_FUZZ_TESTS) INSTALL_RPATH "$ORIGIN/lib") endforeach() endif() + +if(WITH_URMA) + string(TOUPPER "${URMA_USE_MOCK}" URMA_STABILITY_MOCK_MODE) + if(URMA_STABILITY_MOCK_MODE STREQUAL "ON" OR + (URMA_STABILITY_MOCK_MODE STREQUAL "AUTO" AND NOT URMA_LIB)) + foreach(mode IN ITEMS false true) + add_test(NAME urma_stability_polling_${mode} + COMMAND brpc_urma_stability_unittest --urma_stability_datapath + --urma_use_polling=${mode} + --gtest_filter=UrmaStabilityDatapathTest.*:UrmaHandshakeFaultTest.*) + set_tests_properties(urma_stability_polling_${mode} PROPERTIES TIMEOUT 60) + endforeach() + add_test(NAME urma_stability_polling_inline + COMMAND brpc_urma_stability_unittest --urma_stability_datapath + --urma_use_polling=true --urma_disable_bthread=true + --gtest_filter=UrmaStabilityDatapathTest.*:UrmaHandshakeFaultTest.*) + set_tests_properties(urma_stability_polling_inline PROPERTIES TIMEOUT 60) + endif() +endif() diff --git a/test/brpc_urma_stability_unittest.cpp b/test/brpc_urma_stability_unittest.cpp new file mode 100644 index 0000000000..b90e33ab94 --- /dev/null +++ b/test/brpc_urma_stability_unittest.cpp @@ -0,0 +1,600 @@ +// 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 +#include + +#if BRPC_WITH_URMA +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +#include "brpc/channel.h" +#include "brpc/closure_guard.h" +#include "brpc/controller.h" +#include "brpc/server.h" +#include "brpc/urma/mock_urma.h" +#include "brpc/urma/urma_endpoint.h" +#include "brpc/urma/urma_helper.h" +#include "brpc/urma_transport.h" +#include "bthread/bthread.h" +#include "butil/time.h" +#include "echo.pb.h" + +DEFINE_bool(urma_stability_datapath, false, "Run end-to-end URMA tests"); +DEFINE_int32(urma_stability_connections, 8, "Independent live URMA connections"); +DEFINE_int32(urma_stability_rounds, 2, "Mixed-size rounds per connection"); + +namespace brpc { +namespace urma { +extern bool g_skip_urma_init; +} // namespace urma +} // namespace brpc +namespace butil { +namespace iobuf { +extern void* (*blockmem_allocate)(size_t); +extern void (*blockmem_deallocate)(void*); +} // namespace iobuf +} // namespace butil + +namespace { +using namespace brpc; +using namespace brpc::urma; + +// Resolve mock-only entry points so this binary also links with real liburma. +auto mock_fault = + reinterpret_cast(dlsym(RTLD_DEFAULT, "brpc_urma_mock_fault")); +auto mock_release = reinterpret_cast( + dlsym(RTLD_DEFAULT, "brpc_urma_mock_release")); +auto mock_resources = reinterpret_cast( + dlsym(RTLD_DEFAULT, "brpc_urma_mock_resources")); + +class UrmaFaultTest : public testing::Test { + protected: + void SetUp() override { + if (!mock_fault || !mock_release || !mock_resources) { + GTEST_SKIP() << "Requires the URMA link-time mock"; + } + before = mock_resources(); + for (size_t i = 0; i < 2; ++i) { + SocketId id; + ASSERT_EQ(0, Socket::Create(SocketOptions(), &id)); + ASSERT_EQ(0, Socket::Address(id, &sockets[i])); + endpoints[i].reset(new UrmaEndpoint(sockets[i].get())); + InitEndpoint(endpoints[i].get()); + ASSERT_FALSE(HasFatalFailure()); + } + for (size_t i = 0; i < 2; ++i) { + ParsedHello hello; + endpoints[1 - i]->MakeLocalParsedHello(&hello); + endpoints[i]->ApplyRemoteHello(hello); + ASSERT_EQ(0, endpoints[i]->ImportPeer(hello)); + } + } + + // Construct resources without registering a background CQ drainer. Tests + // advance actual endpoint completions explicitly to select each interleaving. + void InitEndpoint(UrmaEndpoint* ep) { + auto* ctx = GetUrmaContext(); + ep->_resource = new UrmaResource; + auto* r = ep->_resource; + r->jfce = urma_create_jfce(ctx); + ASSERT_NE(nullptr, r->jfce); + urma_jfc_cfg_t cq{}; + cq.depth = ep->_sq_size + ep->_rq_size; + cq.jfce = r->jfce; + r->jfc = urma_create_jfc(ctx, &cq); + ASSERT_NE(nullptr, r->jfc); + urma_jfr_cfg_t rq{}; + rq.depth = ep->_rq_size; + rq.trans_mode = URMA_TM_RM; + rq.max_sge = 1; + rq.jfc = r->jfc; + r->jfr = urma_create_jfr(ctx, &rq); + ASSERT_NE(nullptr, r->jfr); + urma_jetty_cfg_t cfg{}; + cfg.flag.bs.share_jfr = 1; + cfg.jfs_cfg.depth = ep->_sq_size; + cfg.jfs_cfg.trans_mode = URMA_TM_RM; + cfg.jfs_cfg.max_sge = GetUrmaMaxSge(); + cfg.jfs_cfg.jfc = r->jfc; + cfg.shared.jfr = r->jfr; + cfg.shared.jfc = r->jfc; + r->jetty = urma_create_jetty(ctx, &cfg); + ASSERT_NE(nullptr, r->jetty); + ep->_sbuf.resize(ep->_sq_size - ep->RESERVED_WR_NUM); + ep->_rbuf.resize(ep->_rq_size); + ep->_rbuf_data.resize(ep->_rq_size, nullptr); + ASSERT_EQ(0, ep->PostRecv(ep->_rq_size, true)); + } + + void TearDown() override { + for (size_t i = 0; i < 2; ++i) { + endpoints[i].reset(); + if (sockets[i]) sockets[i]->SetFailed(); + } + if (mock_resources) { + const auto after = mock_resources(); + EXPECT_EQ(before.jettys, after.jettys); + EXPECT_EQ(before.jfcs, after.jfcs); + EXPECT_EQ(before.jfrs, after.jfrs); + EXPECT_EQ(before.jfces, after.jfces); + EXPECT_EQ(before.targets, after.targets); + EXPECT_EQ(before.segments, after.segments); + EXPECT_EQ(before.deferred, after.deferred); + } + } + + ssize_t Send(size_t i, const std::string& payload) { + butil::IOBuf buf; + buf.append(payload); + butil::IOBuf* data[] = {&buf}; + return endpoints[i]->CutFromIOBufList(data, 1); + } + + void Pump() { + for (size_t turn = 0; turn < 256; ++turn) { + int count = 0; + for (const auto& ep : endpoints) { + urma_cr_t cr[32]{}; + const int n = urma_poll_jfc(ep->_resource->jfc, 32, cr); + ASSERT_GE(n, 0); + count += n; + for (int i = 0; i < n; ++i) { + ASSERT_GE(ep->HandleCompletion(cr[i]), 0); + } + } + if (count == 0) return; + } + FAIL() << "Completion processing did not quiesce"; + } + + MockResources before{}; + SocketUniquePtr sockets[2]; + std::unique_ptr endpoints[2]; +}; + +TEST_F(UrmaFaultTest, delayed_credit_stalls_then_resumes_without_consuming_input) { + auto* sender = endpoints[0].get(); + auto* receiver = endpoints[1].get(); + mock_fault(receiver->_resource->jetty, MOCK_HOLD_CREDIT, -1); + const auto capacity = sender->_local_window_capacity; + for (size_t i = 0; i < capacity; ++i) ASSERT_EQ(1, Send(0, "x")); + Pump(); + ASSERT_FALSE(HasFatalFailure()); + EXPECT_EQ(0, sender->_remote_rq_window_size.load()); + EXPECT_EQ(capacity, sender->_sq_window_size.load()); + butil::IOBuf blocked; + blocked.append("resume"); + butil::IOBuf* data[] = {&blocked}; + EXPECT_EQ(-1, sender->CutFromIOBufList(data, 1)); + EXPECT_EQ(EAGAIN, errno); + EXPECT_EQ("resume", blocked.to_string()); + ASSERT_GT(mock_release(receiver->_resource->jetty), 0u); + Pump(); + ASSERT_EQ(6, sender->CutFromIOBufList(data, 1)); + Pump(); + EXPECT_EQ(std::string(capacity, 'x') + "resume", + receiver->_input_processor.read_buf().to_string()); +} + +TEST_F(UrmaFaultTest, delayed_send_completions_hold_sq_until_buffers_are_reclaimed) { + auto* sender = endpoints[0].get(); + mock_fault(sender->_resource->jetty, MOCK_HOLD_SEND_COMPLETION, -1); + const auto capacity = sender->_local_window_capacity; + for (size_t i = 0; i < capacity; ++i) ASSERT_EQ(1, Send(0, "s")); + Pump(); + EXPECT_EQ(0, sender->_sq_window_size.load()); + EXPECT_GT(sender->_remote_rq_window_size.load(), 0); + EXPECT_EQ(-1, Send(0, "blocked")); + EXPECT_EQ(EAGAIN, errno); + ASSERT_EQ(capacity, mock_release(sender->_resource->jetty)); + Pump(); + EXPECT_EQ(capacity, sender->_sq_window_size.load()); + for (const auto& buf : sender->_sbuf) EXPECT_TRUE(buf.empty()); + ASSERT_EQ(2, Send(0, "ok")); + Pump(); +} + +TEST_F(UrmaFaultTest, rejected_data_post_restores_windows) { + auto* sender = endpoints[0].get(); + const auto sq = sender->_sq_window_size.load(); + const auto remote = sender->_remote_rq_window_size.load(); + mock_fault(sender->_resource->jetty, MOCK_FAIL_SEND, 1); + EXPECT_EQ(-1, Send(0, "rejected")); + EXPECT_EQ(EIO, errno); + EXPECT_EQ(sq, sender->_sq_window_size.load()); + EXPECT_EQ(remote, sender->_remote_rq_window_size.load()); + urma_cr_t cr{}; + EXPECT_EQ(0, urma_poll_jfc(sender->_resource->jfc, 1, &cr)); + sender->Reset(); + EXPECT_EQ(nullptr, sender->_resource); + EXPECT_TRUE(sender->_sbuf.empty()); +} + +TEST_F(UrmaFaultTest, rejected_credit_post_retains_credit_for_retry) { + auto* receiver = endpoints[1].get(); + const auto budget = receiver->_sq_imm_window_size; + mock_fault(receiver->_resource->jetty, MOCK_FAIL_CREDIT, 1); + const size_t count = receiver->_remote_window_capacity / 2 + 1; + for (size_t i = 0; i < count; ++i) ASSERT_EQ(1, Send(0, "c")); + Pump(); + EXPECT_EQ(count, receiver->_new_rq_wrs.load()); + EXPECT_EQ(budget, receiver->_sq_imm_window_size); + ASSERT_EQ(0, receiver->SendAck(0)); + Pump(); + EXPECT_EQ(0, receiver->_new_rq_wrs.load()); + EXPECT_EQ(budget, receiver->_sq_imm_window_size); + EXPECT_EQ(endpoints[0]->_local_window_capacity, endpoints[0]->_remote_rq_window_size.load()); +} + +TEST_F(UrmaFaultTest, completion_error_does_not_restore_a_successful_send_slot) { + auto* sender = endpoints[0].get(); + mock_fault(sender->_resource->jetty, MOCK_ERROR_COMPLETION, 1); + ASSERT_EQ(3, Send(0, "bad")); + urma_cr_t cr{}; + ASSERT_EQ(1, urma_poll_jfc(sender->_resource->jfc, 1, &cr)); + EXPECT_EQ(-1, sender->HandleCompletion(cr)); + EXPECT_EQ(EIO, errno); + EXPECT_EQ(sender->_local_window_capacity - 1, sender->_sq_window_size.load()); +} + +TEST_F(UrmaFaultTest, receive_repost_failure_is_reported) { + auto* receiver = endpoints[1].get(); + mock_fault(receiver->_resource->jetty, MOCK_FAIL_RECV, 1); + ASSERT_EQ(7, Send(0, "receive")); + urma_cr_t cr{}; + ASSERT_EQ(1, urma_poll_jfc(receiver->_resource->jfc, 1, &cr)); + EXPECT_EQ(-1, receiver->HandleCompletion(cr)); + EXPECT_EQ(ENOMEM, errno); +} + +TEST_F(UrmaFaultTest, invalid_and_duplicate_credit_cannot_overflow_window) { + auto* sender = endpoints[0].get(); + urma_cr_t cr{}; + cr.status = URMA_CR_SUCCESS; + cr.flag.bs.s_r = 1; + cr.opcode = URMA_CR_OPC_SEND_WITH_IMM; + cr.imm_data = sender->_local_window_capacity + 1; + EXPECT_EQ(-1, sender->HandleCompletion(cr)); + EXPECT_EQ(EPROTO, errno); + cr.imm_data = 1; + EXPECT_EQ(-1, sender->HandleCompletion(cr)); + EXPECT_EQ(EPROTO, errno); + EXPECT_EQ(sender->_local_window_capacity, sender->_remote_rq_window_size.load()); +} + +TEST_F(UrmaFaultTest, reset_discards_old_completions_before_recreating_resources) { + auto* sender = endpoints[0].get(); + const auto old_id = sender->_resource->jetty->jetty_id.id; + mock_fault(sender->_resource->jetty, MOCK_HOLD_SEND_COMPLETION, -1); + ASSERT_EQ(3, Send(0, "old")); + ASSERT_GT(mock_resources().deferred, before.deferred); + sender->Reset(); + EXPECT_EQ(before.deferred, mock_resources().deferred); + InitEndpoint(sender); + ASSERT_FALSE(HasFatalFailure()); + EXPECT_NE(old_id, sender->_resource->jetty->jetty_id.id); + urma_cr_t cr{}; + EXPECT_EQ(0, urma_poll_jfc(sender->_resource->jfc, 1, &cr)); + EXPECT_EQ(0, sender->_sq_sent); + EXPECT_EQ(0, sender->_sq_current); +} + +TEST_F(UrmaFaultTest, pool_exhaustion_rejects_unregistered_send_and_recovers) { + std::string value; + ASSERT_TRUE(gflags::GetCommandLineOption("urma_buffer_count", &value)); + const size_t capacity = std::stoul(value); + ASSERT_TRUE(gflags::GetCommandLineOption("urma_buffer_size", &value)); + const size_t size = std::stoul(value); + using Block = std::unique_ptr; + std::vector held; + void* fallback = nullptr; + for (size_t i = 0; i <= capacity; ++i) { + void* p = butil::iobuf::blockmem_allocate(size); + ASSERT_NE(nullptr, p); + if (!GetPoolSegFor(p)) { + fallback = p; + break; + } + held.emplace_back(p, butil::iobuf::blockmem_deallocate); + } + ASSERT_NE(nullptr, fallback); + butil::IOBuf buf; + ASSERT_EQ(0, buf.append_user_data(fallback, 16, butil::iobuf::blockmem_deallocate)); + butil::IOBuf* data[] = {&buf}; + EXPECT_EQ(-1, endpoints[0]->CutFromIOBufList(data, 1)); + EXPECT_EQ(ERDMAMEM, errno); + buf.clear(); + held.clear(); + ASSERT_EQ(9, Send(0, "recovered")); + Pump(); + EXPECT_EQ("recovered", endpoints[1]->_input_processor.read_buf().to_string()); +} + +TEST(UrmaPoolTest, exhaustion_falls_back_and_released_blocks_are_reusable) { + if (!mock_resources) GTEST_SKIP() << "Requires the URMA mock"; + std::string value; + ASSERT_TRUE(gflags::GetCommandLineOption("urma_buffer_count", &value)); + const size_t capacity = std::stoul(value); + ASSERT_TRUE(gflags::GetCommandLineOption("urma_buffer_size", &value)); + const size_t size = std::stoul(value); + std::vector blocks; + size_t registered = 0; + for (size_t i = 0; i <= capacity; ++i) { + void* p = butil::iobuf::blockmem_allocate(size); + ASSERT_NE(nullptr, p); + blocks.push_back(p); + registered += GetPoolSegFor(p) != nullptr; + } + EXPECT_GT(registered, 0u); + EXPECT_LT(registered, blocks.size()); + for (auto* p : blocks) butil::iobuf::blockmem_deallocate(p); + void* reused = butil::iobuf::blockmem_allocate(size); + ASSERT_NE(nullptr, reused); + EXPECT_NE(nullptr, GetPoolSegFor(reused)); + butil::iobuf::blockmem_deallocate(reused); +} + +class StabilityEchoService : public test::EchoService { + public: + void Echo(google::protobuf::RpcController* controller, const test::EchoRequest* request, + test::EchoResponse* response, google::protobuf::Closure* done) override { + ClosureGuard guard(done); + auto* cntl = static_cast(controller); + response->set_message(request->message()); + response->set_receiving_socket_id(cntl->_current_call.sending_sock->id()); + cntl->response_attachment().append(cntl->request_attachment()); + if (request->close_fd()) { + entered.store(true); + while (!release.load()) bthread_usleep(1000); + } + } + std::atomic entered{false}; + std::atomic release{false}; +}; + +class UrmaStabilityDatapathTest : public testing::Test { + protected: + void SetUp() override { + if (!FLAGS_urma_stability_datapath) GTEST_SKIP() << "Pass --urma_stability_datapath"; + ASSERT_TRUE(IsUrmaAvailable()); + ASSERT_EQ(0, server.AddService(&service, SERVER_DOESNT_OWN_SERVICE)); + ServerOptions options; + options.socket_mode = SOCKET_MODE_URMA; + ASSERT_EQ(0, server.Start("127.0.0.1:0", &options)); + } + void TearDown() override { + service.release.store(true); + server.Stop(0); + server.Join(); + } + void InitChannel(Channel* channel, const std::string& group) { + ChannelOptions options; + options.socket_mode = SOCKET_MODE_URMA; + options.timeout_ms = 3000; + options.connect_timeout_ms = 3000; + options.max_retry = 0; + options.connection_group = group; + ASSERT_EQ(0, channel->Init(server.listen_address(), &options)); + } + void CheckEcho(Channel* channel, size_t size, size_t sequence, SocketId* peer = nullptr) { + test::EchoService_Stub stub(channel); + test::EchoRequest request; + test::EchoResponse response; + Controller cntl; + std::string payload(size, '\0'); + for (size_t i = 0; i < size; ++i) payload[i] = (i * 31 + sequence * 17) % 251; + request.set_message("request-" + std::to_string(sequence)); + cntl.request_attachment().append(payload); + stub.Echo(&cntl, &request, &response, nullptr); + ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText(); + ASSERT_EQ(request.message(), response.message()); + ASSERT_TRUE(payload == cntl.response_attachment().to_string()) + << "size=" << size << " sequence=" << sequence; + SocketUniquePtr socket; + ASSERT_EQ(0, Socket::Address(response.receiving_socket_id(), &socket)); + ASSERT_EQ(SOCKET_MODE_URMA, socket->socket_mode()); + auto* transport = static_cast(socket->_transport.get()); + ASSERT_EQ(UrmaTransport::URMA_ON, transport->_urma_state.load()); + if (peer) *peer = socket->id(); + } + StabilityEchoService service; + Server server; +}; + +TEST_F(UrmaStabilityDatapathTest, mixed_sizes_on_independent_live_connections) { + ASSERT_GT(FLAGS_urma_stability_connections, 0); + ASSERT_GT(FLAGS_urma_stability_rounds, 0); + std::vector> channels; + std::vector channel_peers; + std::set peers; + for (int i = 0; i < FLAGS_urma_stability_connections; ++i) { + channels.emplace_back(new Channel); + InitChannel(channels.back().get(), "stability-" + std::to_string(i)); + ASSERT_FALSE(HasFatalFailure()); + SocketId peer; + CheckEcho(channels.back().get(), 1, i, &peer); + ASSERT_FALSE(HasFatalFailure()); + ASSERT_TRUE(peers.insert(peer).second); + channel_peers.push_back(peer); + } + ASSERT_EQ(static_cast(FLAGS_urma_stability_connections), peers.size()); + const size_t block = GetUrmaRecvBlockSize(); + const size_t sizes[] = {0, 1, 511, 512, 513, block - 1, + block, block + 1, 65535, 65536, 65537, 128 * 1024}; + for (int round = 0; round < FLAGS_urma_stability_rounds; ++round) { + for (size_t first = 0; first < channels.size(); first += 8) { + std::vector workers; + for (size_t i = first; i < channels.size() && i < first + 8; ++i) { + workers.emplace_back([&, i] { + SocketId peer = INVALID_SOCKET_ID; + CheckEcho(channels[i].get(), sizes[(i + round) % 12], + channels.size() * round + i, &peer); + EXPECT_EQ(channel_peers[i], peer); + }); + } + for (auto& worker : workers) worker.join(); + ASSERT_FALSE(HasFatalFailure()); + } + // Exercise every boundary even with a small connection count. + for (size_t i = 0; i < 12; ++i) { + SocketId peer = INVALID_SOCKET_ID; + CheckEcho(channels[i % channels.size()].get(), sizes[i], round * 12 + i, &peer); + EXPECT_EQ(channel_peers[i % channels.size()], peer); + ASSERT_FALSE(HasFatalFailure()); + } + } +} + +TEST_F(UrmaStabilityDatapathTest, peer_close_fails_inflight_rpc_and_new_connection_recovers) { + Channel channel; + InitChannel(&channel, "peer-close"); + SocketId peer; + CheckEcho(&channel, 512, 1, &peer); + ASSERT_FALSE(HasFatalFailure()); + test::EchoService_Stub stub(&channel); + test::EchoRequest request; + test::EchoResponse response; + Controller cntl; + request.set_message("inflight"); + request.set_close_fd(true); + const auto call = cntl.call_id(); + const int64_t start = butil::gettimeofday_us(); + stub.Echo(&cntl, &request, &response, DoNothing()); + const int64_t deadline = start + 2000000; + while (!service.entered.load() && butil::gettimeofday_us() < deadline) bthread_usleep(1000); + EXPECT_TRUE(service.entered.load()); + SocketUniquePtr socket; + if (Socket::Address(peer, &socket) == 0) { + shutdown(socket->fd(), SHUT_RDWR); + socket->SetFailed(ECONNRESET, "Injected peer close"); + } + // Let the CQ drainer run even when user code is processed inline. + service.release.store(true); + Join(call); + EXPECT_TRUE(cntl.Failed()); + EXPECT_LT(butil::gettimeofday_us() - start, 4000000); + Channel recovered; + InitChannel(&recovered, "peer-recovered"); + SocketId new_peer; + CheckEcho(&recovered, 65537, 2, &new_peer); + EXPECT_NE(peer, new_peer); +} + +TEST_F(UrmaStabilityDatapathTest, peer_close_while_waiting_for_credit_finishes_writer) { + if (!mock_fault) GTEST_SKIP() << "Requires credit injection in the URMA mock"; + Channel channel; + InitChannel(&channel, "credit-close"); + SocketId peer; + CheckEcho(&channel, 1, 1, &peer); + ASSERT_FALSE(HasFatalFailure()); + SocketUniquePtr socket; + ASSERT_EQ(0, Socket::Address(peer, &socket)); + auto* ep = static_cast(socket->_transport.get())->GetUrmaEp(); + const auto deferred = mock_resources().deferred; + mock_fault(ep->_resource->jetty, MOCK_HOLD_CREDIT, -1); + test::EchoService_Stub stub(&channel); + test::EchoRequest request; + test::EchoResponse response; + Controller cntl; + request.set_message("waiting-for-credit"); + cntl.request_attachment().append(std::string(256 * 1024, 'w')); + const auto call = cntl.call_id(); + const int64_t start = butil::gettimeofday_us(); + stub.Echo(&cntl, &request, &response, DoNothing()); + while (mock_resources().deferred == deferred && butil::gettimeofday_us() < start + 2000000) + bthread_usleep(1000); + EXPECT_GT(mock_resources().deferred, deferred); + shutdown(socket->fd(), SHUT_RDWR); + socket->SetFailed(ECONNRESET, "Injected close while writer waits for credit"); + Join(call); + EXPECT_TRUE(cntl.Failed()); + EXPECT_LT(butil::gettimeofday_us() - start, 4000000); + Channel recovered; + InitChannel(&recovered, "credit-recovered"); + CheckEcho(&recovered, 65537, 3); +} + +TEST(UrmaHandshakeFaultTest, peer_exit_during_handshake_finishes_rpc) { + if (!FLAGS_urma_stability_datapath) GTEST_SKIP() << "Pass --urma_stability_datapath"; + const int listener = socket(AF_INET, SOCK_STREAM, 0); + ASSERT_GE(listener, 0); + sockaddr_in addr{}; + addr.sin_family = AF_INET; + addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + ASSERT_EQ(0, bind(listener, reinterpret_cast(&addr), sizeof(addr))); + ASSERT_EQ(0, listen(listener, 1)); + socklen_t len = sizeof(addr); + ASSERT_EQ(0, getsockname(listener, reinterpret_cast(&addr), &len)); + std::thread peer([listener] { + const int fd = accept(listener, nullptr, nullptr); + if (fd >= 0) { + char magic[4]; + (void)recv(fd, magic, sizeof(magic), MSG_WAITALL); + close(fd); + } + }); + Channel channel; + ChannelOptions options; + options.socket_mode = SOCKET_MODE_URMA; + options.timeout_ms = 1000; + options.connect_timeout_ms = 1000; + options.max_retry = 0; + const int rc = channel.Init("127.0.0.1", ntohs(addr.sin_port), &options); + EXPECT_EQ(0, rc); + if (rc == 0) { + test::EchoService_Stub stub(&channel); + Controller cntl; + test::EchoRequest request; + test::EchoResponse response; + request.set_message("handshake"); + const int64_t start = butil::gettimeofday_us(); + stub.Echo(&cntl, &request, &response, nullptr); + EXPECT_TRUE(cntl.Failed()); + EXPECT_LT(butil::gettimeofday_us() - start, 2000000); + } + shutdown(listener, SHUT_RDWR); + peer.join(); + close(listener); +} +} // namespace +#endif // BRPC_WITH_URMA + +int main(int argc, char** argv) { +#if BRPC_WITH_URMA + gflags::SetCommandLineOption("urma_buffer_count", "4096"); + gflags::SetCommandLineOption("urma_sq_size", "16"); + gflags::SetCommandLineOption("urma_rq_size", "16"); + gflags::SetCommandLineOption("urma_prepared_jetty_cnt", "0"); +#endif + testing::InitGoogleTest(&argc, argv); + gflags::ParseCommandLineFlags(&argc, &argv, true); +#if BRPC_WITH_URMA + brpc::urma::g_skip_urma_init = !(mock_resources || FLAGS_urma_stability_datapath); + if (!brpc::urma::g_skip_urma_init) brpc::urma::GlobalUrmaInitializeOrDie(); +#endif + return RUN_ALL_TESTS(); +} diff --git a/tools/urma_stability.py b/tools/urma_stability.py new file mode 100644 index 0000000000..732de512b1 --- /dev/null +++ b/tools/urma_stability.py @@ -0,0 +1,87 @@ +#!/usr/bin/env python3 +# 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. + +"""Run mixed-size URMA regression tests with independent live connections.""" + +import argparse +import os +from pathlib import Path +import resource +import signal +import subprocess +import tempfile +import xml.etree.ElementTree as ET + + +def positive(value): + result = int(value) + if result <= 0: + raise argparse.ArgumentTypeError("must be positive") + return result + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("binary", type=Path) + parser.add_argument("--connections", type=positive, default=1024) + parser.add_argument("--rounds", type=positive, default=100) + parser.add_argument("--timeout", type=positive, default=1800) + parser.add_argument("--device", default="") + args = parser.parse_args() + # Each connection has client/server TCP fds and two CQ event fds. + needed = args.connections * 8 + 1024 + soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE) + if soft < needed: + if hard != resource.RLIM_INFINITY and hard < needed: + parser.error(f"RLIMIT_NOFILE must be at least {needed}") + resource.setrlimit(resource.RLIMIT_NOFILE, (needed, hard)) + binary = str(args.binary.resolve(strict=True)) + with tempfile.TemporaryDirectory(prefix="urma-stability-") as directory: + for polling, inline in ((False, False), (True, False), (True, True)): + report = Path(directory) / "result.xml" + command = [ + binary, "--urma_stability_datapath", + "--gtest_filter=UrmaStabilityDatapathTest.mixed_sizes_on_independent_live_connections", + f"--gtest_output=xml:{report}", + f"--urma_stability_connections={args.connections}", + f"--urma_stability_rounds={args.rounds}", + f"--urma_buffer_count={max(4096, args.connections * 40 + 4096)}", + f"--urma_use_polling={str(polling).lower()}", + f"--urma_disable_bthread={str(inline).lower()}", + ] + if args.device: + command.append(f"--urma_device={args.device}") + print(f"connections={args.connections} rounds={args.rounds} " + f"polling={polling} inline={inline}", flush=True) + with subprocess.Popen(command, start_new_session=True) as process: + try: + code = process.wait(timeout=args.timeout) + except (subprocess.TimeoutExpired, KeyboardInterrupt): + os.killpg(process.pid, signal.SIGKILL) + process.wait() + raise + if code: + raise RuntimeError(f"URMA test process exited with status {code}") + result = ET.parse(report).getroot() + if int(result.get("tests", "0")) != 1 or result.findall(".//skipped"): + raise RuntimeError("URMA mixed-size test did not run") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())