From 2ad1004538bc66dd34aba51b4af7361572bf9d3a Mon Sep 17 00:00:00 2001 From: Spicy-cream <2667314914@qq.com> Date: Mon, 2 Nov 2026 00:39:05 -0800 Subject: [PATCH 1/2] Add URMA transport observability and diagnostic CLI --- docs/cn/urma.md | 59 +++ docs/en/urma.md | 96 +++++ src/brpc/builtin/sockets_service.cpp | 52 ++- src/brpc/controller.cpp | 3 + src/brpc/policy/baidu_rpc_protocol.cpp | 3 + src/brpc/socket.cpp | 53 ++- src/brpc/socket.h | 9 + src/brpc/urma/mock_urma.cpp | 16 +- src/brpc/urma/urma_endpoint.cpp | 155 ++++++-- src/brpc/urma/urma_endpoint.h | 17 +- src/brpc/urma/urma_helper.cpp | 7 + src/brpc/urma/urma_observability.cpp | 342 +++++++++++++++++ src/brpc/urma/urma_observability.h | 179 +++++++++ src/brpc/urma_transport.cpp | 97 ++++- src/brpc/urma_transport.h | 11 + test/CMakeLists.txt | 18 + test/brpc_urma_observability_unittest.cpp | 432 ++++++++++++++++++++++ test/urma_diag_test.py | 106 ++++++ tools/urma_diag.py | 138 +++++++ 19 files changed, 1755 insertions(+), 38 deletions(-) create mode 100644 src/brpc/urma/urma_observability.cpp create mode 100644 src/brpc/urma/urma_observability.h create mode 100644 test/brpc_urma_observability_unittest.cpp create mode 100644 test/urma_diag_test.py create mode 100644 tools/urma_diag.py diff --git a/docs/cn/urma.md b/docs/cn/urma.md index 62e10389a8..4670bff114 100644 --- a/docs/cn/urma.md +++ b/docs/cn/urma.md @@ -193,3 +193,62 @@ UrmaTransport 推荐用于**大包和跨节点**高吞吐路径，而 UBRing `ContextInitOrDie` 阶段拒绝。 - 硬件数据通路需要受支持的 UMDK provider 和 `liburma`。 - 当前实现面向 Linux。 + + +## 可观测性与诊断 + +URMA 通过已有 `/vars` 和 Prometheus 导出进程级 `urma_*` 指标，通过已有 +内置服务入口提供连接快照。设置 `internal_port` 时使用内网端口，沿用现有 +内置服务访问控制。 + +```sh +curl http://127.0.0.1:8003/sockets/urma +curl 'http://127.0.0.1:8003/sockets/123?format=json' +python3 tools/urma_diag.py --server 127.0.0.1:8003 connections +python3 tools/urma_diag.py --server 127.0.0.1:8003 pool +python3 tools/urma_diag.py --server 127.0.0.1:8003 socket 123 +python3 tools/urma_diag.py --server 127.0.0.1:8003 --json watch --interval 1 --count 10 +python3 tools/urma_diag.py --server 127.0.0.1:8003 trace 123 --trace-id 456 +``` + +CLI 仅依赖 Python 3 标准库。列表最多返回 1024 条连接并输出 `truncated`； +按 SocketId 查询不受列表上限影响。ID 用十进制 JSON 字符串保存，避免 64 位 +精度丢失。列表 `schema_version` 为 1。 + +连接快照包括建链/回退状态和原因、实际协商窗口容量、剩余 SQ/远端 credit、 +数据 WR 提交与完成、在途字节、错误及两种资源等待。进程指标带 `urma_` 前缀， +连接快照的 `counters` 使用不带前缀的名称。 + +* `writes`、`bulk_writes`、`bulk_write_bytes`：进入已建立 URMA 数据路径的完整 + 序列化写入。`urma_bulk_min_bytes` 默认 65536 字节，非正值关闭分类。大小包括 + 协议头和附件；写入、WR 和业务 RPC 的数量不能混用。 +* `send_bytes` / `send_wrs` 与 `send_completed_bytes` / `send_completed_wrs` + 分别表示数据提交和本地完成。提交失败会回滚；本地完成不等于远端业务已处理。 +* `sq_blocked` / `remote_credit_blocked` 统计尝试发送时窗口为零的观察次数； + `sq_wait_us` / `remote_credit_wait_us` 按 butex 等待入口观察到的原因归因。 + 两者可能重叠，不能作为互斥阶段相加。 +* `urma_pool_capacity`、`urma_pool_buffer_size`、`urma_pool_in_use` 和 + `urma_pool_peak_in_use` 统计注册 BufferPool。IOBuf 引用和线程缓存可能在 RPC + 结束后继续持有 buffer。`urma_pool_exhausted` 和 `urma_malloc_fallback` 统计池耗尽 + 及分配回退尝试，后者还包括超大分配；分配回退不等于连接回退 TCP。 + +开启 rpcz 并设置 `urma_trace_sample_every=N`，按 span ID 每 N 条采样一条； +默认 0 关闭，1 跟踪所有可用 rpcz span，仍受 rpcz 自身采样影响。客户端请求和 +baidu_std 服务端响应显式携带 Span 进入写队列，不依赖 KeepWrite 的当前 TLS。 +支持一条 RPC 拆成多个 WR、一个 WR 合并多个请求数据，纯 credit WR 不计入。 +旁路关联表仅用于采样写入，保持 `Socket::WriteRequest` 为 64 字节。 + +排队、首次提交和完成/失败事件写入 rpcz 注释。完成可能晚于 rpcz 序列化， +不会为观测延迟 RPC 完成或 rpcz 收集，因此每连接另保留最近 64 条已结束采样， +包含 trace/span/log ID，供 CLI 查询。`path=no_urma_wr` 表示没有提交 URMA 数据 WR， +例如走了 TCP 回退。历史在连接重置或销毁时清除。这些时间描述本地传输进度， +不能直接当作远端处理耗时或纯链路延迟。 + +快照是各原子字段的并发观察，不保证所有字段来自同一瞬间。Endpoint 重用时 +连接计数清零且 `generation` 递增，进程指标累计保留；CLI 不跨 generation 计算速率。 +完整指标口径与验证说明见 [英文文档](../en/urma.md#observability-and-diagnostics)。 + +使用 URMA mock 构建后运行 `brpc_urma_unittest`、 +`brpc_urma_observability_unittest` 和 `python3 test/urma_diag_test.py`；同时检查 +`WITH_URMA=OFF` 构建。上线前需在真实硬件上比较关闭跟踪及代表性采样率下的 +吞吐、CPU 和尾延迟；mock 测试不能代替硬件性能验证。 diff --git a/docs/en/urma.md b/docs/en/urma.md index a5b80f45b9..c021065319 100644 --- a/docs/en/urma.md +++ b/docs/en/urma.md @@ -207,3 +207,99 @@ 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. + + +## Observability and diagnostics + +URMA exports process-wide `urma_*` bvars through `/vars` (and the existing +Prometheus exporter), with per-connection snapshots through the existing +builtin-services listener: + +```sh +curl http://127.0.0.1:8003/sockets/urma +curl 'http://127.0.0.1:8003/sockets/123?format=json' +python3 tools/urma_diag.py --server 127.0.0.1:8003 connections +python3 tools/urma_diag.py --server 127.0.0.1:8003 pool +python3 tools/urma_diag.py --server 127.0.0.1:8003 socket 123 +python3 tools/urma_diag.py --server 127.0.0.1:8003 --json watch --interval 1 --count 10 +python3 tools/urma_diag.py --server 127.0.0.1:8003 trace 123 --trace-id 456 +``` + +Use the internal port when `ServerOptions.internal_port` is configured. These +queries use the existing builtin-service access controls. Python 3 and its +standard library are sufficient for the CLI; there are no third-party packages. +The inventory returns at most 1024 connections and explicitly sets `truncated`. +An individual socket query bypasses that inventory limit. IDs are decimal JSON +strings to preserve all 64 bits. `schema_version` is currently 1 for the inventory. + +### Counter definitions + +All names below have the `urma_` prefix in `/vars`. Connection snapshots contain +these names without the prefix under `counters`. + +| Counters | Meaning | +| --- | --- | +| `writes`, `bulk_writes`, `bulk_write_bytes` | Serialized writes entering the established URMA data path. Bulk means at least `--urma_bulk_min_bytes` bytes (default 65536); a nonpositive value disables classification. Includes protocol headers and attachments. A write is not a WR or necessarily an application RPC. | +| `send_bytes`, `send_wrs` | Payload bytes and data WRs accepted for submission. Failed single-WR posts are rolled back. | +| `send_completed_bytes`, `send_completed_wrs` | Locally completed data sends; this does not mean the peer application has processed them. | +| `recv_bytes`, `recv_wrs` | Received nonempty payloads accepted by the URMA receive path. Credit-only traffic is excluded. | +| `send_errors`, `recv_post_errors`, `completion_errors` | Data-send post failures, receive-post failures, and provider completion errors. | +| `sq_blocked`, `remote_credit_blocked` | Observations of a zero local SQ window or zero remote receive credit while attempting a send. Counts attempts, not distinct congestion episodes. | +| `sq_wait_us`, `remote_credit_wait_us` | Time spent in a writer's butex wait, attributed to windows observed empty at wait entry. Both can increase for the same wait; do not sum them as exclusive phases. | +| `handshake_success`, `handshake_failed`, `tcp_fallback` | Established URMA handshakes, terminal handshake failures, and negotiated/unavailable-resource TCP fallbacks. | +| `pool_exhausted`, `malloc_fallback` | Registered-pool allocation misses and fallback allocation attempts (also includes allocations larger than one pool buffer). These are process-wide. | + +The pool gauges are `urma_pool_capacity`, `urma_pool_buffer_size`, +`urma_pool_in_use`, and `urma_pool_peak_in_use`. They count registered buffers, +not live RPCs; IOBuf references and thread-local caches may retain buffers after +an RPC returns. The implementation calls this resource `BufferPool`, not a +separate bulk arena. Pool exhaustion and allocation fallback do not by themselves +imply that the entire connection switched to TCP. + +Snapshots include the connection state, negotiated data-window capacity, +remaining SQ/remote credit, in-flight data bytes/WRs, and the last diagnostic +reason/provider error. `FALLBACK_TCP` and `FAILED` remain queryable while the +Socket is addressable. Reads are race-free observations of individual counters, +not a transaction across the data path. Connection counters reset on endpoint +reuse and `generation` increases; process bvars remain cumulative. The CLI +suppresses rates across a changed generation. + +### Sampled RPC transport traces + +Enable rpcz and set `--urma_trace_sample_every=N` to trace one in N span IDs. +The default is 0 (disabled). `N=1` traces every available rpcz span; rpcz's own +sampling still applies. The client request and baidu_std server response carry +their explicit Span through Socket's write queue, so background KeepWrite and +batched writes do not borrow an unrelated bthread's trace context. + +A sampled write records queue-to-first-post time, bytes, data WR count, +local completions, per-cause wait time and failure. One RPC can span several WRs, +and a single WR can contain contributions from several RPCs. Credit-only WRs +never count as RPC data. Tracing uses a side table only for sampled writes and +keeps `Socket::WriteRequest` at 64 bytes. It adds no global data-path lock when +transport tracing is disabled. + +Annotations are added to the existing rpcz span at queue, first-post and local +completion/failure boundaries. Completion may occur after rpcz has serialized +the span; it does not delay RPC completion or rpcz collection. For that reason, +each connection also retains its latest 64 finished sampled writes in the JSON +snapshot, including trace/span/log IDs. `path=no_urma_wr` identifies writes that +never submitted a URMA data WR (for example, TCP fallback). The bounded history +lasts until connection reset or destruction. These measurements describe local +transport progress, not remote execution or wire-only latency. + +### Validation + +Build with `WITH_URMA=ON`, `URMA_USE_MOCK=ON`, and `BUILD_UNIT_TESTS=ON`, then run +`brpc_urma_unittest`, `brpc_urma_observability_unittest`, and +`python3 test/urma_diag_test.py`. Also build with `WITH_URMA=OFF` to check the +optional-transport boundary. Hardware throughput, CPU and tail latency should be +compared with tracing disabled and representative sampling rates before rollout; +mock results are functional validation, not a hardware performance measurement. + +CMake automatically registers `urma_datapath_polling_false` and +`urma_datapath_polling_true` in mock builds. These send concurrent 128 KiB RPCs, +check split/merged WR trace completion, inject empty SQ/remote-credit windows, +and exhaust the actual registered buffer pool. With rpcz enabled, URMA schedules +application callbacks outside the CQ drainer so a large traced response cannot +block the credit completions needed to finish its own write. diff --git a/src/brpc/builtin/sockets_service.cpp b/src/brpc/builtin/sockets_service.cpp index c17e86cf6e..acad228492 100644 --- a/src/brpc/builtin/sockets_service.cpp +++ b/src/brpc/builtin/sockets_service.cpp @@ -17,12 +17,15 @@ #include +#include #include "brpc/closure_guard.h" // ClosureGuard #include "brpc/controller.h" // Controller #include "brpc/builtin/common.h" #include "brpc/builtin/sockets_service.h" #include "brpc/socket.h" - +#if BRPC_WITH_URMA +#include "brpc/urma/urma_observability.h" +#endif namespace brpc { @@ -35,15 +38,56 @@ void SocketsService::default_method(::google::protobuf::RpcController* cntl_base cntl->http_response().set_content_type("text/plain"); butil::IOBufBuilder os; const std::string& constraint = cntl->http_request().unresolved_path(); - - if (constraint.empty()) { + +#if BRPC_WITH_URMA + if (constraint == "urma") { + cntl->http_response().set_content_type("application/json"); + os << "{\"schema_version\":1,\"pool\":"; + urma::DescribePoolJson(os); + os << ",\"connections\":["; + bool first = true; + // Fetch one extra ID to report truncation, rather than silently hide it. + const auto ids = urma::ListDiagnosticSockets(1025); + for (size_t i = 0; i < ids.size() && i < 1024; ++i) { + SocketUniquePtr socket; + if (Socket::AddressFailedAsWell(ids[i], &socket) < 0) { + continue; + } + std::ostringstream snapshot; + if (!socket->DescribeUrmaJson(snapshot)) { + continue; + } + if (!first) { + os << ','; + } + first = false; + os << snapshot.str(); + } + os << "],\"truncated\":" << (ids.size() > 1024 ? "true" : "false") << '}'; + } else +#endif + if (constraint.empty()) { os << "# Use /sockets/\n" << butil::describe_resources() << '\n'; } else { char* endptr = nullptr; SocketId sid = strtoull(constraint.c_str(), &endptr, 10); if (*endptr == '\0' || *endptr == '/') { - Socket::DebugSocket(os, sid); +#if BRPC_WITH_URMA + const std::string* format = cntl->http_request().uri().GetQuery("format"); + if (format && *format == "json") { + SocketUniquePtr socket; + if (Socket::AddressFailedAsWell(sid, &socket) < 0 || + !socket->DescribeUrmaJson(os)) { + cntl->SetFailed(ENOMETHOD, "Socket has no URMA diagnostics"); + } else { + cntl->http_response().set_content_type("application/json"); + } + } else +#endif + { + Socket::DebugSocket(os, sid); + } } else { cntl->SetFailed(ENOMETHOD, "path=%s is not a SocketId", constraint.c_str()); diff --git a/src/brpc/controller.cpp b/src/brpc/controller.cpp index 4fff9fd2f4..cde95527f5 100644 --- a/src/brpc/controller.cpp +++ b/src/brpc/controller.cpp @@ -1509,6 +1509,9 @@ void Controller::IssueRPC(int64_t start_realtime_us) { } Socket::WriteOptions wopt; wopt.id_wait = cid; +#if BRPC_WITH_URMA + wopt.urma_trace_span = _span; +#endif wopt.abstime = pabstime; wopt.pipelined_count = _pipelined_count; wopt.auth_flags = _auth_flags; diff --git a/src/brpc/policy/baidu_rpc_protocol.cpp b/src/brpc/policy/baidu_rpc_protocol.cpp index 54e611861c..bda65dd635 100644 --- a/src/brpc/policy/baidu_rpc_protocol.cpp +++ b/src/brpc/policy/baidu_rpc_protocol.cpp @@ -467,6 +467,9 @@ void SendRpcResponse(int64_t correlation_id, Controller* cntl, // Have the risk of unlimited pending responses, in which case, tell // users to set max_concurrency. Socket::WriteOptions wopt; +#if BRPC_WITH_URMA + wopt.urma_trace_span = span; +#endif wopt.ignore_eovercrowded = true; if (INVALID_BTHREAD_ID != response_id) { wopt.id_wait = response_id; diff --git a/src/brpc/socket.cpp b/src/brpc/socket.cpp index 57cfd2296a..ecf1704bd2 100644 --- a/src/brpc/socket.cpp +++ b/src/brpc/socket.cpp @@ -41,6 +41,9 @@ #include "brpc/errno.pb.h" #include "brpc/event_dispatcher.h" // RemoveConsumer #include "brpc/socket.h" +#if BRPC_WITH_URMA +#include "brpc/urma_transport.h" +#endif #include "brpc/describable.h" // Describable #include "brpc/circuit_breaker.h" // CircuitBreaker #include "brpc/input_messenger.h" @@ -399,6 +402,11 @@ void Socket::WriteRequest::Setup(Socket* s) { return; } } +#if BRPC_WITH_URMA + if (s->socket_mode() == SOCKET_MODE_URMA) { + static_cast(s->_transport.get())->ObserveWrite(data.size()); + } +#endif const int64_t before_write = s->_unwritten_bytes.fetch_add(data.size(), butil::memory_order_relaxed); if (before_write + (int64_t)data.size() >= FLAGS_socket_max_unwritten_bytes) { @@ -508,6 +516,11 @@ Socket::~Socket() { void Socket::ReturnSuccessfulWriteRequest(Socket::WriteRequest* p) { DCHECK(p->data.empty()); +#if BRPC_WITH_URMA + if (socket_mode() == SOCKET_MODE_URMA) { + static_cast(_transport.get())->FinishWrite(&p->data, 0); + } +#endif AddOutputMessages(1); const bthread_id_t id_wait = p->id_wait; bool is_notify_on_success = p->is_notify_on_success(); @@ -528,6 +541,11 @@ void Socket::ReturnFailedWriteRequest(Socket::WriteRequest* p, int error_code, CancelUnwrittenBytes(p->data.size()); } p->data.clear(); // data is probably not written. +#if BRPC_WITH_URMA + if (socket_mode() == SOCKET_MODE_URMA) { + static_cast(_transport.get())->FinishWrite(&p->data, error_code); + } +#endif const bthread_id_t id_wait = p->id_wait; butil::return_object(p); if (id_wait != INVALID_BTHREAD_ID) { @@ -1677,6 +1695,11 @@ int Socket::Write(butil::IOBuf* data, const WriteOptions* options_in) { // wait until it points to a valid WriteRequest or nullptr. req->next = WriteRequest::UNCONNECTED; req->id_wait = opt.id_wait; +#if BRPC_WITH_URMA + if (socket_mode() == SOCKET_MODE_URMA) { + static_cast(_transport.get())->TrackWrite(&req->data, opt.urma_trace_span); + } +#endif req->clear_and_set_control_bits(opt.notify_on_success, opt.shutdown_write); req->set_pipelined_count_and_user_message( opt.pipelined_count, DUMMY_USER_MESSAGE, opt.auth_flags); @@ -1714,6 +1737,11 @@ int Socket::Write(SocketMessagePtr<>& msg, const WriteOptions* options_in) { // wait until it points to a valid WriteRequest or nullptr. req->next = WriteRequest::UNCONNECTED; req->id_wait = opt.id_wait; +#if BRPC_WITH_URMA + if (socket_mode() == SOCKET_MODE_URMA) { + static_cast(_transport.get())->TrackWrite(&req->data, opt.urma_trace_span); + } +#endif req->clear_and_set_control_bits(opt.notify_on_success, opt.shutdown_write); req->set_pipelined_count_and_user_message( opt.pipelined_count, msg.release(), opt.auth_flags); @@ -1780,7 +1808,15 @@ int Socket::StartWrite(WriteRequest* req, const WriteOptions& opt) { butil::IOBuf* data_arr[1] = { &req->data }; nw = _conn->CutMessageIntoFileDescriptor(fd(), data_arr, 1); } else { - nw = _transport->CutFromIOBuf(&req->data); +#if BRPC_WITH_URMA + if (socket_mode() == SOCKET_MODE_URMA) { + butil::IOBuf* data_arr[] = {&req->data}; + nw = static_cast(_transport.get())->CutWithTrace(data_arr, 1); + } else +#endif + { + nw = _transport->CutFromIOBuf(&req->data); + } } if (nw < 0) { // RTMP may return EOVERCROWDED @@ -1925,6 +1961,12 @@ ssize_t Socket::DoWrite(WriteRequest* req) { if (_conn) { return _conn->CutMessageIntoFileDescriptor(fd(), data_list, ndata); } else { +#if BRPC_WITH_URMA + if (socket_mode() == SOCKET_MODE_URMA) { + return static_cast(_transport.get()) + ->CutWithTrace(data_list, ndata); + } +#endif return _transport->CutFromIOBufList(data_list, ndata); } } @@ -2327,6 +2369,15 @@ std::ostream& operator<<(std::ostream& os, const ObjectPtr& obj) { return os << obj._obj; } +#if BRPC_WITH_URMA +bool Socket::DescribeUrmaJson(std::ostream& os) const { + if (socket_mode() != SOCKET_MODE_URMA || !_transport) { + return false; + } + return static_cast(_transport.get())->DescribeJson(os); +} +#endif + void Socket::DebugSocket(std::ostream& os, SocketId id) { SocketUniquePtr ptr; int ret = AddressFailedAsWell(id, &ptr); diff --git a/src/brpc/socket.h b/src/brpc/socket.h index da701e9ad7..bd2912657f 100644 --- a/src/brpc/socket.h +++ b/src/brpc/socket.h @@ -46,6 +46,7 @@ #include "brpc/input_messenger_processor.h" // InputMessengerProcessor namespace brpc { +class Span; namespace policy { class ConsistentHashingLoadBalancer; class RtmpContext; @@ -393,6 +394,10 @@ friend class TransportFactory; // - Write once when uncontended(most cases). // - Wait-free when contended. struct WriteOptions { +#if BRPC_WITH_URMA + // Explicit RPC span; never infer this from the KeepWrite bthread's TLS. + std::shared_ptr urma_trace_span; +#endif // `id_wait' is signalled when this Socket is SetFailed or data is written // successfully with `notify_on_success=true'. To disable the signal, set // this field to INVALID_BTHREAD_ID. `on_reset' of `id_wait' is NOT called @@ -632,6 +637,10 @@ friend class TransportFactory; // The socket mode decided at creation. SocketMode socket_mode() const { return _socket_mode; } +#if BRPC_WITH_URMA + // Called with an addressed Socket, keeping its transport alive. + bool DescribeUrmaJson(std::ostream& os) const; +#endif // Get an UNUSED socket connecting to the same place as this socket // from the SocketPool of this socket. diff --git a/src/brpc/urma/mock_urma.cpp b/src/brpc/urma/mock_urma.cpp index b0572c2e86..fa5ce3f15a 100644 --- a/src/brpc/urma/mock_urma.cpp +++ b/src/brpc/urma/mock_urma.cpp @@ -560,10 +560,11 @@ urma_status_t urma_post_jetty_send_wr(urma_jetty_t *jetty, urma_jfs_wr_t *wr, 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); - } + auto complete_send = [&] { + if (current->flag.bs.complete_enable) { + PushCompletion(jetty->jetty_cfg.jfs_cfg.jfc, local_state, send_cr); + } + }; PendingRecv recv{}; urma_jfc_t* remote_jfc = nullptr; @@ -571,6 +572,7 @@ urma_status_t urma_post_jetty_send_wr(urma_jetty_t *jetty, urma_jfs_wr_t *wr, 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()) { + complete_send(); continue; } urma_jfr_t* remote_jfr = @@ -579,6 +581,7 @@ urma_status_t urma_post_jetty_send_wr(urma_jetty_t *jetty, urma_jfs_wr_t *wr, 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()) { + complete_send(); continue; } recv = recv_it->second.front(); @@ -596,6 +599,11 @@ urma_status_t urma_post_jetty_send_wr(urma_jetty_t *jetty, urma_jfs_wr_t *wr, copied += n; } + // A local completion releases the sender's IOBuf references. Publish + // it only after the mock has consumed every source SGE, matching the + // lifetime guarantee made by real providers. + complete_send(); + JfcState* remote_state = nullptr; { std::shared_lock lock(g_rw_mutex); diff --git a/src/brpc/urma/urma_endpoint.cpp b/src/brpc/urma/urma_endpoint.cpp index 26c7a1b9dc..80a5bdc0ab 100644 --- a/src/brpc/urma/urma_endpoint.cpp +++ b/src/brpc/urma/urma_endpoint.cpp @@ -155,9 +155,11 @@ UrmaEndpoint::UrmaEndpoint(Socket* s) _read_butex = bthread::butex_create_checked>(); _read_butex->store(0, butil::memory_order_relaxed); _input_processor.Init(s, InputMessengerProcessor::STREAM_URMA_JETTY); + RegisterDiagnosticSocket(s->id()); } UrmaEndpoint::~UrmaEndpoint() { + UnregisterDiagnosticSocket(_socket->id()); DeallocateResources(); if (_read_butex) { bthread::butex_destroy(_read_butex); @@ -167,6 +169,8 @@ UrmaEndpoint::~UrmaEndpoint() { void UrmaEndpoint::Reset() { DeallocateResources(); + _stats.Reset(); + _waiting_traces.clear(); _state = UNINIT; _handshake_version = 0; _remote_recv_block_size = 0; @@ -467,6 +471,7 @@ int UrmaEndpoint::AllocateResources() { } _sbuf.resize(_sq_size - RESERVED_WR_NUM); + _send_observations.resize(_sbuf.size()); _rbuf.resize(_rq_size); _rbuf_data.resize(_rq_size, nullptr); @@ -530,6 +535,12 @@ void UrmaEndpoint::DeallocateResources() { // Reusing a Jetty requires a driver-supported RESET plus a complete JFC // drain. Until that lifecycle is implemented, prepared resources are // one-shot: they accelerate connection setup but are destroyed on close. + for (auto& observation : _send_observations) { + for (auto& contribution : observation.traces) { + contribution.trace->PostingDone(ECONNRESET); + } + observation.traces.clear(); + } delete _resource; _resource = nullptr; } @@ -672,7 +683,9 @@ class UrmaIOBuf : private butil::IOBuf { } }; -ssize_t UrmaEndpoint::CutFromIOBufList(butil::IOBuf** from, size_t ndata) { +ssize_t UrmaEndpoint::CutFromIOBufList(butil::IOBuf** from, size_t ndata, + const std::shared_ptr* traces) { + _waiting_traces.clear(); if (!_resource || !_resource->jetty || !_resource->remote_jetty) { errno = ENOTCONN; return -1; @@ -693,8 +706,21 @@ ssize_t UrmaEndpoint::CutFromIOBufList(butil::IOBuf** from, size_t ndata) { ssize_t total_len = 0; while (current < ndata) { uint16_t remote_wnd = _remote_rq_window_size.load(butil::memory_order_relaxed); - uint16_t sq_wnd = _sq_window_size.load(butil::memory_order_relaxed); + uint16_t sq_wnd = _sq_window_size.load(butil::memory_order_acquire); if (remote_wnd == 0 || sq_wnd == 0) { + if (sq_wnd == 0) { + _stats.Add(sq_blocked); + } + if (remote_wnd == 0) { + _stats.Add(remote_credit_blocked); + } + if (traces) { + for (size_t i = current; i < ndata; ++i) { + if (traces[i]) { + _waiting_traces.push_back(traces[i]); + } + } + } if (total_len > 0) { break; } @@ -702,6 +728,9 @@ ssize_t UrmaEndpoint::CutFromIOBufList(butil::IOBuf** from, size_t ndata) { return -1; } butil::IOBuf* to = &_sbuf[_sq_current]; + auto& observation = _send_observations[_sq_current]; + observation.bytes = 0; + observation.traces.clear(); size_t sge_index = 0; size_t this_len = 0; size_t max_len = _remote_recv_block_size > 0 @@ -719,6 +748,14 @@ ssize_t UrmaEndpoint::CutFromIOBufList(butil::IOBuf** from, size_t ndata) { if (n < 0) { return -1; } + if (n > 0 && traces && traces[current]) { + if (!observation.traces.empty() && + observation.traces.back().trace == traces[current]) { + observation.traces.back().bytes += n; + } else { + observation.traces.push_back({traces[current], static_cast(n)}); + } + } this_len += n; } if (sge_index == 0) { @@ -748,11 +785,27 @@ ssize_t UrmaEndpoint::CutFromIOBufList(butil::IOBuf** from, size_t ndata) { // Decrementing after post therefore creates a transient capacity + 1 // window and makes the strict credit check tear down a healthy // connection. + observation.bytes = this_len; + _stats.Add(send_bytes, this_len); + _stats.Add(send_wrs); + for (auto& contribution : observation.traces) { + contribution.trace->Posted(contribution.bytes); + } _remote_rq_window_size.fetch_sub(1, butil::memory_order_relaxed); - _sq_window_size.fetch_sub(1, butil::memory_order_relaxed); + // Publish slot metadata before a polling thread can observe its completion. + _sq_window_size.fetch_sub(1, butil::memory_order_release); int rc = urma_post_jetty_send_wr(_resource->jetty, &wr, &bad_wr); if (rc != URMA_SUCCESS) { const int provider_errno = errno; + _stats.Add(send_bytes, -static_cast(this_len)); + _stats.Add(send_wrs, -1); + _stats.Add(send_errors); + _stats.SetReason("post send", rc); + for (auto& contribution : observation.traces) { + contribution.trace->Rejected(contribution.bytes); + contribution.trace->PostingDone(rc); + } + observation.traces.clear(); _remote_rq_window_size.fetch_add(1, butil::memory_order_relaxed); _sq_window_size.fetch_add(1, butil::memory_order_relaxed); LOG(WARNING) << "urma_post_jetty_send_wr failed: " << rc @@ -805,6 +858,8 @@ int UrmaEndpoint::DoPostRecv(void* block, size_t block_size) { const urma_status_t status = urma_post_jfr_wr(_resource->jfr, &wr, &bad); if (status != URMA_SUCCESS) { + _stats.Add(recv_post_errors); + _stats.SetReason("post receive", status); LOG(WARNING) << "Failed to post URMA receive WR: status=" << status << " bonding=" << IsUrmaBondingDevice() << " bad_wr=" << static_cast(bad) @@ -927,6 +982,8 @@ int UrmaEndpoint::SendAck(int num) { ssize_t UrmaEndpoint::HandleCompletion(const urma_cr_t& cr) { bool zerocopy = FLAGS_urma_recv_zerocopy; if (cr.status != URMA_CR_SUCCESS) { + _stats.Add(completion_errors); + _stats.SetReason("completion", cr.status); LOG(WARNING) << "URMA completion failed, status=" << cr.status; errno = EIO; return -1; @@ -950,6 +1007,20 @@ ssize_t UrmaEndpoint::HandleCompletion(const urma_cr_t& cr) { return 0; } uint16_t wnd = 1; // We signal every WR (complete_enable=1). + if (_sq_window_size.load(butil::memory_order_acquire) >= _local_window_capacity) { + _stats.Add(completion_errors); + errno = EPROTO; + return -1; + } + auto& observation = _send_observations[_sq_sent]; + _stats.Add(send_completed_bytes, observation.bytes); + _stats.Add(send_completed_wrs); + for (auto& contribution : observation.traces) { + contribution.trace->Completed(); + } + observation.traces.clear(); + _sbuf[_sq_sent].clear(); + _sq_sent = (_sq_sent + 1) % (_sq_size - RESERVED_WR_NUM); uint16_t old = _sq_window_size.load(butil::memory_order_relaxed); while (true) { @@ -963,16 +1034,12 @@ ssize_t UrmaEndpoint::HandleCompletion(const urma_cr_t& cr) { errno = EPROTO; return -1; } - if (_sq_window_size.compare_exchange_weak( - old, static_cast(old + wnd), - butil::memory_order_relaxed)) { + if (_sq_window_size.compare_exchange_weak(old, static_cast(old + wnd), + butil::memory_order_release, + butil::memory_order_relaxed)) { break; } } - for (uint16_t i = 0; i < wnd; ++i) { - _sbuf[_sq_sent].clear(); - _sq_sent = (_sq_sent + 1) % (_sq_size - RESERVED_WR_NUM); - } butil::subtle::MemoryBarrier(); if (_remote_rq_window_size.load(butil::memory_order_relaxed) >= _local_window_capacity / 8) { @@ -1036,6 +1103,8 @@ ssize_t UrmaEndpoint::HandleCompletion(const urma_cr_t& cr) { return -1; } if (cr.completion_len > 0) { + _stats.Add(recv_bytes, cr.completion_len); + _stats.Add(recv_wrs); SendAck(1); } return static_cast(cr.completion_len); @@ -1194,6 +1263,7 @@ void UrmaEndpoint::ApplyRemoteHello(const ParsedHello& remote) { _remote_window_capacity = _rq_size > RESERVED_WR_NUM ? _rq_size - RESERVED_WR_NUM : 0; _sq_imm_window_size = RESERVED_WR_NUM; + _stats.window_capacity.store(_local_window_capacity, butil::memory_order_relaxed); _remote_rq_window_size.store(_local_window_capacity, butil::memory_order_relaxed); _sq_window_size.store(_local_window_capacity, butil::memory_order_relaxed); @@ -1227,6 +1297,8 @@ void UrmaEndpoint::OnNewDataFromTcp(Socket* m) { // Server side: kick off the handshake bthread. if (!IsUrmaAvailable()) { ep->_state = FALLBACK_TCP; + ep->_stats.Add(tcp_fallback); + ep->_stats.SetReason("URMA unavailable"); tp->_urma_state = UrmaTransport::URMA_OFF; InputMessenger::OnNewMessages(m); return; @@ -1298,7 +1370,10 @@ static void TryReadOnTcpDuringUrmaEst(Socket* socket) { } } -void UrmaEndpoint::FallbackToTcp(UrmaTransport* transport, bool process_tcp) { +void UrmaEndpoint::FallbackToTcp(UrmaTransport* transport, bool process_tcp, const char* reason, + int error) { + _stats.Add(tcp_fallback); + _stats.SetReason(reason, error); transport->_urma_state = UrmaTransport::URMA_OFF; _state.store(FALLBACK_TCP, butil::memory_order_release); DeallocateResources(); @@ -1309,6 +1384,8 @@ void UrmaEndpoint::FallbackToTcp(UrmaTransport* transport, bool process_tcp) { void UrmaEndpoint::FailHandshake(UrmaTransport* transport, int error, const char* reason) { + _stats.Add(handshake_failed); + _stats.SetReason(reason, error); LOG(ERROR) << "URMA handshake failed in state=" << GetStateStr() << " on " << _socket->description() << ": " << reason << ", error=" << error @@ -1334,23 +1411,24 @@ void* UrmaEndpoint::ProcessHandshakeAtClient(void* arg) { auto* tp = static_cast(s->_transport.get()); UrmaConnect::RunGuard guard(static_cast(s->_app_connect.get())); if (!IsUrmaAvailable()) { - ep->FallbackToTcp(tp, true); + ep->FallbackToTcp(tp, true, "URMA unavailable"); return nullptr; } ep->_state = C_ALLOC_RES; if (ep->AllocateResources() < 0) { - ep->FallbackToTcp(tp, true); + ep->FallbackToTcp(tp, true, "resource allocation", errno); return nullptr; } // Prepost the shared JFR before sending the client hello so the peer sees // a ready receive queue as soon as its import completes. if (ep->PostRecv(ep->_rq_size, FLAGS_urma_recv_zerocopy) < 0) { - ep->FallbackToTcp(tp, true); + ep->FallbackToTcp(tp, true, "initial receive posting", errno); return nullptr; } ep->_state = C_HELLO_SEND; std::unique_ptr hs(CreateClientHandshake(ep)); ep->_handshake_version = hs->ProtocolVersion(); + ep->_stats.handshake_version.store(ep->_handshake_version, butil::memory_order_relaxed); if (hs->SendLocalHello() < 0) { const int saved_errno = errno ? errno : EIO; ep->FailHandshake(tp, saved_errno, "send client hello"); @@ -1385,6 +1463,7 @@ void* UrmaEndpoint::ProcessHandshakeAtClient(void* arg) { } tp->_urma_state = UrmaTransport::URMA_ON; ep->_state = ESTABLISHED; + ep->_stats.Add(handshake_success); ep->DispatchReceivedBytes(s, 0); return nullptr; } @@ -1409,6 +1488,7 @@ void* UrmaEndpoint::ProcessHandshakeAtServer(void* arg) { return nullptr; } ep->_handshake_version = hs->ProtocolVersion(); + ep->_stats.handshake_version.store(ep->_handshake_version, butil::memory_order_relaxed); ParsedHello remote; bool negotiated = false; if (hs->ReceiveAndParseRemoteHello(&remote, &negotiated) < 0) { @@ -1470,6 +1550,7 @@ void* UrmaEndpoint::ProcessHandshakeAtServer(void* arg) { } tp->_urma_state = UrmaTransport::URMA_ON; ep->_state = ESTABLISHED; + ep->_stats.Add(handshake_success); ep->DispatchReceivedBytes(s, 0); } else { ep->FallbackToTcp(tp, true); @@ -1499,6 +1580,8 @@ void UrmaConnect::StartConnect(const Socket* socket, // Fall back to TCP immediately. if (tp->_urma_ep) { tp->_urma_ep->_state = UrmaEndpoint::FALLBACK_TCP; + tp->_urma_ep->_stats.Add(tcp_fallback); + tp->_urma_ep->_stats.SetReason("URMA unavailable"); } tp->_urma_state = UrmaTransport::URMA_OFF; Run(); @@ -1511,6 +1594,8 @@ void UrmaConnect::StartConnect(const Socket* socket, UrmaEndpoint::ProcessHandshakeAtClient, tp->_urma_ep) != 0) { tp->_urma_ep->_state = UrmaEndpoint::FALLBACK_TCP; + tp->_urma_ep->_stats.Add(tcp_fallback); + tp->_urma_ep->_stats.SetReason("start handshake bthread"); tp->_urma_state = UrmaTransport::URMA_OFF; Run(); } else { @@ -1554,14 +1639,42 @@ std::string UrmaEndpoint::GetStateStr() const { return "UNKNOWN"; } +ConnectionSnapshot UrmaEndpoint::GetSnapshot() const { + ConnectionSnapshot snapshot; + snapshot.socket_id = _socket->id(); + snapshot.socket_failed = _socket->Failed(); + snapshot.state = GetStateStr(); + snapshot.sq_size = _sq_size; + snapshot.rq_size = _rq_size; + snapshot.sq_remaining = _sq_window_size.load(butil::memory_order_relaxed); + snapshot.remote_credit_remaining = _remote_rq_window_size.load(butil::memory_order_relaxed); + _stats.Snapshot(&snapshot); + return snapshot; +} + +void UrmaEndpoint::RecordWait(bool sq, bool remote, uint64_t elapsed_us) { + if (sq) { + _stats.Add(sq_wait_us, elapsed_us); + } + if (remote) { + _stats.Add(remote_credit_wait_us, elapsed_us); + } + for (auto& trace : _waiting_traces) { + trace->Waited(sq, remote, elapsed_us); + } + _waiting_traces.clear(); +} + void UrmaEndpoint::DebugInfo(std::ostream& os, butil::StringPiece) const { - os << "state=" << GetStateStr() - << " sq_size=" << _sq_size << " rq_size=" << _rq_size - << " remote_recv_block_size=" << _remote_recv_block_size - << " sq_window=" << _sq_window_size.load(butil::memory_order_relaxed) - << " remote_rq_window=" << _remote_rq_window_size.load(butil::memory_order_relaxed) - << " handshake_version=" << _handshake_version - << " read_buf=" << _input_processor.read_buf().size(); + const auto snapshot = GetSnapshot(); + os << "state=" << snapshot.state << " sq_size=" << snapshot.sq_size + << " rq_size=" << snapshot.rq_size << " sq_window=" << snapshot.sq_remaining + << " remote_rq_window=" << snapshot.remote_credit_remaining + << " handshake_version=" << snapshot.handshake_version << " reason=" << snapshot.reason + << " error=" << snapshot.error; + for (size_t i = 0; i < COUNTER_COUNT; ++i) { + os << '\n' << CounterName(static_cast(i)) << '=' << snapshot.counters[i]; + } } int UrmaEndpoint::WaitCqEvent(SocketUniquePtr& s, diff --git a/src/brpc/urma/urma_endpoint.h b/src/brpc/urma/urma_endpoint.h index e5aed5ac66..1d3c0a634e 100644 --- a/src/brpc/urma/urma_endpoint.h +++ b/src/brpc/urma/urma_endpoint.h @@ -37,6 +37,7 @@ #include "urma_api.h" #include "urma_types.h" #include "brpc/urma/urma_handshake.h" +#include "brpc/urma/urma_observability.h" #include "brpc/urma/urma_handshake.pb.h" namespace brpc { @@ -126,7 +127,11 @@ class BAIDU_CACHELINE_ALIGNMENT UrmaEndpoint : public SocketUser { // ---- Data path (called by UrmaTransport) ---- // Cut data from the IOBuf list and post it via URMA SEND. // Returns bytes sent, or -1 (errno set; EAGAIN when windows are full). - ssize_t CutFromIOBufList(butil::IOBuf** data, size_t ndata); + ssize_t CutFromIOBufList(butil::IOBuf** data, size_t ndata, + const std::shared_ptr* traces = nullptr); + ConnectionSnapshot GetSnapshot() const; + void RecordWait(bool sq, bool remote, uint64_t elapsed_us); + ConnectionStats& stats() { return _stats; } // Whether the endpoint can post more sends (both windows non-empty). bool IsWritable() const; @@ -260,7 +265,8 @@ class BAIDU_CACHELINE_ALIGNMENT UrmaEndpoint : public SocketUser { void PollerRemoveCqSid(); inline void TryReadOnTcp(); - void FallbackToTcp(UrmaTransport* transport, bool process_tcp); + void FallbackToTcp(UrmaTransport* transport, bool process_tcp, + const char* reason = "peer declined URMA", int error = 0); void FailHandshake(UrmaTransport* transport, int error, const char* reason); @@ -294,6 +300,13 @@ class BAIDU_CACHELINE_ALIGNMENT UrmaEndpoint : public SocketUser { // Per-WR send buffers (own the IOBuf until the SEND completes). std::vector _sbuf; + struct SendObservation { + size_t bytes = 0; + std::vector traces; + }; + std::vector _send_observations; + std::vector> _waiting_traces; + ConnectionStats _stats; // Per-WR recv buffers (zero-copy targets). std::vector _rbuf; std::vector _rbuf_data; diff --git a/src/brpc/urma/urma_helper.cpp b/src/brpc/urma/urma_helper.cpp index ded662fb22..3456864edc 100644 --- a/src/brpc/urma/urma_helper.cpp +++ b/src/brpc/urma/urma_helper.cpp @@ -46,6 +46,7 @@ #include "urma_types.h" #include "brpc/urma/urma_bonding.h" #include "brpc/urma/urma_endpoint.h" +#include "brpc/urma/urma_observability.h" DECLARE_int32(task_group_ntags); @@ -278,6 +279,7 @@ void* PoolAllocate(size_t size) { // Only serve the configured buffer size; callers always ask for that. if (size > g_pool_buffer_size) { // Larger than a single buffer -- fall back to the system allocator. + AddGlobalCounter(malloc_fallback); return g_mem_alloc_orig ? g_mem_alloc_orig(size) : malloc(size); } auto start = PreferredShard(); @@ -296,8 +298,11 @@ void* PoolAllocate(size_t size) { g_pool->in_use[idx] = 1; } g_pool->outstanding.fetch_add(1, butil::memory_order_relaxed); + PoolAllocated(); return buf; } + AddGlobalCounter(pool_exhausted); + AddGlobalCounter(malloc_fallback); LOG_EVERY_SECOND(WARNING) << "URMA buffer pool exhausted; falling back to malloc"; return g_mem_alloc_orig ? g_mem_alloc_orig(size) : malloc(size); @@ -337,6 +342,7 @@ void PoolDeallocate(void* buf) { g_pool->free_lists[shard].push_back(buf); if (g_pool->outstanding.load(butil::memory_order_relaxed) > 0) { g_pool->outstanding.fetch_sub(1, butil::memory_order_relaxed); + PoolDeallocated(); } } @@ -389,6 +395,7 @@ bool InitPool() { g_pool->free_lists[shard].push_back( static_cast(g_pool_base) + i * g_pool_buffer_size); } + SetPoolCapacity(count, g_pool_buffer_size); return true; } diff --git a/src/brpc/urma/urma_observability.cpp b/src/brpc/urma/urma_observability.cpp new file mode 100644 index 0000000000..c63d4c5e2b --- /dev/null +++ b/src/brpc/urma/urma_observability.cpp @@ -0,0 +1,342 @@ +// 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_observability.h" +#if BRPC_WITH_URMA +#include + +#include +#include +#include + +#include "brpc/span.h" +#include "butil/third_party/rapidjson/stringbuffer.h" +#include "butil/third_party/rapidjson/writer.h" +#include "butil/time.h" +#include "bvar/passive_status.h" +#include "bvar/reducer.h" + +namespace brpc { +namespace urma { + +DEFINE_int64(urma_bulk_min_bytes, 65536, + "Minimum serialized Socket::Write size classified as a bulk write"); + +DEFINE_int32(urma_trace_sample_every, 0, + "Trace one in N rpcz spans through URMA; 0 disables transport tracing"); + +namespace { +const char* const COUNTER_NAMES[] = { +#define BRPC_URMA_NAME(name) #name, + BRPC_URMA_COUNTERS(BRPC_URMA_NAME) +#undef BRPC_URMA_NAME +}; + +struct Diagnostics { + std::array, COUNTER_COUNT> counters; + butil::atomic capacity{0}, buffer_size{0}, in_use{0}, peak{0}; + std::mutex mutex; + std::set sockets; + bvar::PassiveStatus capacity_var, size_var, used_var, peak_var; + Diagnostics() + : capacity_var("urma_pool_capacity", Read, &capacity), + size_var("urma_pool_buffer_size", Read, &buffer_size), + used_var("urma_pool_in_use", Read, &in_use), + peak_var("urma_pool_peak_in_use", Read, &peak) { + for (size_t i = 0; i < COUNTER_COUNT; ++i) { + counters[i].expose(std::string("urma_") + COUNTER_NAMES[i]); + } + } + static uint64_t Read(void* arg) { + return static_cast*>(arg)->load(butil::memory_order_relaxed); + } +}; + +Diagnostics& Global() { + // bvar callbacks remain valid during process shutdown. + static Diagnostics* const diagnostics = new Diagnostics; + return *diagnostics; +} + +using Writer = butil::rapidjson::Writer; +void Number(Writer& writer, const char* key, uint64_t value) { + writer.Key(key); + writer.AddUint64(value); +} +void Id(Writer& writer, const char* key, uint64_t value) { + writer.Key(key); + writer.String(std::to_string(value).c_str()); +} +} // namespace + +const char* CounterName(Counter counter) { return COUNTER_NAMES[counter]; } +void AddGlobalCounter(Counter counter, int64_t value) { Global().counters[counter] << value; } + +void SetPoolCapacity(uint64_t count, uint64_t size) { + Global().capacity.store(count, butil::memory_order_relaxed); + Global().buffer_size.store(size, butil::memory_order_relaxed); +} +void PoolAllocated() { + auto& diagnostics = Global(); + const uint64_t used = diagnostics.in_use.fetch_add(1, butil::memory_order_relaxed) + 1; + uint64_t peak = diagnostics.peak.load(butil::memory_order_relaxed); + while (used > peak && + !diagnostics.peak.compare_exchange_weak(peak, used, butil::memory_order_relaxed)) { + } +} +void PoolDeallocated() { Global().in_use.fetch_sub(1, butil::memory_order_relaxed); } +PoolSnapshot GetPoolSnapshot() { + auto& diagnostics = Global(); + PoolSnapshot snapshot; + snapshot.capacity = diagnostics.capacity.load(butil::memory_order_relaxed); + snapshot.buffer_size = diagnostics.buffer_size.load(butil::memory_order_relaxed); + snapshot.in_use = diagnostics.in_use.load(butil::memory_order_relaxed); + snapshot.peak_in_use = diagnostics.peak.load(butil::memory_order_relaxed); + return snapshot; +} +void DescribePoolJson(std::ostream& os) { + const auto snapshot = GetPoolSnapshot(); + butil::rapidjson::StringBuffer buffer; + Writer writer(buffer); + writer.StartObject(); + Number(writer, "capacity", snapshot.capacity); + Number(writer, "buffer_size", snapshot.buffer_size); + Number(writer, "in_use", snapshot.in_use); + Number(writer, "peak_in_use", snapshot.peak_in_use); + Number(writer, "free", + snapshot.capacity > snapshot.in_use ? snapshot.capacity - snapshot.in_use : 0); + Number(writer, "exhausted", Global().counters[pool_exhausted].get_value()); + Number(writer, "malloc_fallback", Global().counters[malloc_fallback].get_value()); + writer.EndObject(); + os << buffer.GetString(); +} + +void TraceHistory::Push(const TraceRecord& record) { + std::lock_guard lock(_mutex); + if (_records.size() == 64) { + _records.pop_front(); + } + _records.push_back(record); +} +std::vector TraceHistory::Snapshot() const { + std::lock_guard lock(_mutex); + return {_records.begin(), _records.end()}; +} +void TraceHistory::Clear() { + std::lock_guard lock(_mutex); + _records.clear(); +} + +WriteTrace::WriteTrace(std::shared_ptr span, SocketId sid, + std::shared_ptr history) + : _span(std::move(span)), + _sid(sid), + _history(std::move(history)), + _start_us(butil::cpuwide_time_us()) { + _span->Annotate("URMA queued socket=%llu", (unsigned long long)_sid); +} +void WriteTrace::Posted(size_t bytes) { + _bytes.fetch_add(bytes, butil::memory_order_relaxed); + if (_wrs.fetch_add(1, butil::memory_order_release) == 0) { + const uint64_t elapsed = butil::cpuwide_time_us() - _start_us; + _queue_us.store(elapsed, butil::memory_order_relaxed); + _span->Annotate("URMA first post socket=%llu queue_us=%llu", (unsigned long long)_sid, + (unsigned long long)elapsed); + } +} +void WriteTrace::Rejected(size_t bytes) { + _bytes.fetch_sub(bytes, butil::memory_order_relaxed); + _wrs.fetch_sub(1, butil::memory_order_relaxed); +} +void WriteTrace::Completed() { + _completed.fetch_add(1, butil::memory_order_acq_rel); + Finish(); +} +void WriteTrace::Waited(bool sq, bool remote, uint64_t elapsed_us) { + if (sq) { + _sq_wait_us.fetch_add(elapsed_us, butil::memory_order_relaxed); + } + if (remote) { + _remote_wait_us.fetch_add(elapsed_us, butil::memory_order_relaxed); + } +} +void WriteTrace::PostingDone(int error) { + if (error) { + _error.store(error, butil::memory_order_relaxed); + } + _posting_done.store(true, butil::memory_order_release); + Finish(); +} +void WriteTrace::Finish() { + if (!_posting_done.load(butil::memory_order_acquire)) { + return; + } + const int error = _error.load(butil::memory_order_relaxed); + if (!error && + _completed.load(butil::memory_order_acquire) != _wrs.load(butil::memory_order_acquire)) { + return; + } + if (_finished.exchange(true, butil::memory_order_acq_rel)) { + return; + } + TraceRecord record; + record.trace_id = _span->trace_id(); + record.span_id = _span->span_id(); + record.log_id = _span->log_id(); + record.bytes = _bytes.load(butil::memory_order_relaxed); + record.wrs = _wrs.load(butil::memory_order_relaxed); + record.completed_wrs = _completed.load(butil::memory_order_relaxed); + record.queue_us = _queue_us.load(butil::memory_order_relaxed); + record.elapsed_us = butil::cpuwide_time_us() - _start_us; + record.sq_wait_us = _sq_wait_us.load(butil::memory_order_relaxed); + record.remote_credit_wait_us = _remote_wait_us.load(butil::memory_order_relaxed); + record.error = error; + _history->Push(record); + _span->Annotate( + "URMA local send finished socket=%llu bytes=%llu wrs=%llu " + "completed_wrs=%llu sq_wait_us=%llu remote_credit_wait_us=%llu error=%d", + (unsigned long long)_sid, (unsigned long long)record.bytes, (unsigned long long)record.wrs, + (unsigned long long)record.completed_wrs, (unsigned long long)record.sq_wait_us, + (unsigned long long)record.remote_credit_wait_us, error); +} + +ConnectionStats::ConnectionStats() : _history(new TraceHistory) { + (void)Global(); + Reset(); +} +void ConnectionStats::Reset() { + for (auto& counter : _counters) { + counter.store(0, butil::memory_order_relaxed); + } + window_capacity.store(0, butil::memory_order_relaxed); + handshake_version.store(0, butil::memory_order_relaxed); + SetReason(""); + _history->Clear(); + _generation.fetch_add(1, butil::memory_order_relaxed); +} +void ConnectionStats::Add(Counter counter, int64_t value) { + _counters[counter].fetch_add(value, butil::memory_order_relaxed); + AddGlobalCounter(counter, value); +} +void ConnectionStats::SetReason(const char* reason, int error) { + _error.store(error, butil::memory_order_relaxed); + _reason.store(reason, butil::memory_order_release); +} +void ConnectionStats::Snapshot(ConnectionSnapshot* snapshot) const { + snapshot->generation = _generation.load(butil::memory_order_relaxed); + snapshot->reason = _reason.load(butil::memory_order_acquire); + snapshot->error = _error.load(butil::memory_order_relaxed); + snapshot->window_capacity = window_capacity.load(butil::memory_order_relaxed); + snapshot->handshake_version = handshake_version.load(butil::memory_order_relaxed); + for (size_t i = 0; i < COUNTER_COUNT; ++i) { + snapshot->counters[i] = _counters[i].load(butil::memory_order_relaxed); + } + snapshot->traces = _history->Snapshot(); +} +std::shared_ptr ConnectionStats::NewTrace(std::shared_ptr span, SocketId sid) { + const int every = FLAGS_urma_trace_sample_every; + if (!span || every <= 0 || span->span_id() % every != 0) { + return nullptr; + } + return std::make_shared(std::move(span), sid, _history); +} + +void DescribeConnectionJson(std::ostream& os, const ConnectionSnapshot& snapshot) { + butil::rapidjson::StringBuffer buffer; + Writer writer(buffer); + writer.StartObject(); + Id(writer, "socket_id", snapshot.socket_id); + writer.Key("socket_failed"); + writer.Bool(snapshot.socket_failed); + writer.Key("state"); + writer.String(snapshot.state.c_str()); + writer.Key("reason"); + writer.String(snapshot.reason.c_str()); + writer.Key("error"); + writer.AddInt(snapshot.error); + Number(writer, "generation", snapshot.generation); + Number(writer, "sq_size", snapshot.sq_size); + Number(writer, "rq_size", snapshot.rq_size); + Number(writer, "window_capacity", snapshot.window_capacity); + Number(writer, "sq_remaining", snapshot.sq_remaining); + Number(writer, "remote_credit_remaining", snapshot.remote_credit_remaining); + Number(writer, "handshake_version", snapshot.handshake_version); + const auto submitted_bytes = snapshot.counters[send_bytes]; + const auto completed_bytes = snapshot.counters[send_completed_bytes]; + const auto submitted_wrs = snapshot.counters[send_wrs]; + const auto completed_wrs = snapshot.counters[send_completed_wrs]; + Number(writer, "inflight_bytes", + submitted_bytes > completed_bytes ? submitted_bytes - completed_bytes : 0); + Number(writer, "inflight_wrs", + submitted_wrs > completed_wrs ? submitted_wrs - completed_wrs : 0); + writer.Key("counters"); + writer.StartObject(); + for (size_t i = 0; i < COUNTER_COUNT; ++i) { + Number(writer, COUNTER_NAMES[i], snapshot.counters[i]); + } + writer.EndObject(); + writer.Key("traces"); + writer.StartArray(); + for (const auto& trace : snapshot.traces) { + writer.StartObject(); + writer.Key("path"); + writer.String(trace.wrs ? "urma" : "no_urma_wr"); + Id(writer, "trace_id", trace.trace_id); + Id(writer, "span_id", trace.span_id); + Id(writer, "log_id", trace.log_id); + Number(writer, "bytes", trace.bytes); + Number(writer, "wrs", trace.wrs); + Number(writer, "completed_wrs", trace.completed_wrs); + Number(writer, "queue_us", trace.queue_us); + Number(writer, "elapsed_us", trace.elapsed_us); + Number(writer, "sq_wait_us", trace.sq_wait_us); + Number(writer, "remote_credit_wait_us", trace.remote_credit_wait_us); + writer.Key("error"); + writer.AddInt(trace.error); + writer.EndObject(); + } + writer.EndArray(); + writer.EndObject(); + os << buffer.GetString(); +} + +void RegisterDiagnosticSocket(SocketId sid) { + auto& diagnostics = Global(); + std::lock_guard lock(diagnostics.mutex); + diagnostics.sockets.insert(sid); +} +void UnregisterDiagnosticSocket(SocketId sid) { + auto& diagnostics = Global(); + std::lock_guard lock(diagnostics.mutex); + diagnostics.sockets.erase(sid); +} +std::vector ListDiagnosticSockets(size_t limit) { + auto& diagnostics = Global(); + std::lock_guard lock(diagnostics.mutex); + std::vector result; + for (auto sid : diagnostics.sockets) { + if (result.size() == limit) { + break; + } + result.push_back(sid); + } + return result; +} + +} // namespace urma +} // namespace brpc +#endif // BRPC_WITH_URMA diff --git a/src/brpc/urma/urma_observability.h b/src/brpc/urma/urma_observability.h new file mode 100644 index 0000000000..1d6798e613 --- /dev/null +++ b/src/brpc/urma/urma_observability.h @@ -0,0 +1,179 @@ +// 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_OBSERVABILITY_H +#define BRPC_URMA_OBSERVABILITY_H + +#if BRPC_WITH_URMA +#include +#include +#include +#include +#include +#include +#include + +#include "brpc/socket_id.h" +#include "butil/atomicops.h" + +namespace brpc { +class Span; +namespace urma { + +// Counters count data WRs, not RPCs. Credit-only WRs are excluded. +// clang-format off +#define BRPC_URMA_COUNTERS(V) \ + V(writes) \ + V(bulk_writes) \ + V(bulk_write_bytes) \ + V(send_bytes) \ + V(send_wrs) \ + V(send_completed_bytes) \ + V(send_completed_wrs) \ + V(recv_bytes) \ + V(recv_wrs) \ + V(send_errors) \ + V(recv_post_errors) \ + V(completion_errors) \ + V(sq_blocked) \ + V(remote_credit_blocked) \ + V(sq_wait_us) \ + V(remote_credit_wait_us) \ + V(handshake_success) \ + V(handshake_failed) \ + V(tcp_fallback) \ + V(pool_exhausted) \ + V(malloc_fallback) + +enum Counter { +#define BRPC_URMA_ENUM(name) name, + BRPC_URMA_COUNTERS(BRPC_URMA_ENUM) +#undef BRPC_URMA_ENUM + COUNTER_COUNT +}; +// clang-format on +const char* CounterName(Counter counter); +void AddGlobalCounter(Counter counter, int64_t value = 1); + +struct PoolSnapshot { + uint64_t capacity = 0; + uint64_t buffer_size = 0; + uint64_t in_use = 0; + uint64_t peak_in_use = 0; +}; +void SetPoolCapacity(uint64_t count, uint64_t buffer_size); +void PoolAllocated(); +void PoolDeallocated(); +PoolSnapshot GetPoolSnapshot(); +void DescribePoolJson(std::ostream& os); + +struct TraceRecord { + uint64_t trace_id = 0; + uint64_t span_id = 0; + uint64_t log_id = 0; + uint64_t bytes = 0; + uint64_t wrs = 0; + uint64_t completed_wrs = 0; + uint64_t queue_us = 0; + uint64_t elapsed_us = 0; + uint64_t sq_wait_us = 0; + uint64_t remote_credit_wait_us = 0; + int error = 0; +}; + +// Storage is bounded per connection, independent of rpcz's disk retention. +class TraceHistory { + public: + void Push(const TraceRecord& record); + std::vector Snapshot() const; + void Clear(); + + private: + mutable std::mutex _mutex; + std::deque _records; +}; + +// Explicitly carried from Socket::Write through KeepWrite and each data WR. +// A WR may reference multiple traces; a trace may span multiple WRs. +class WriteTrace { + public: + WriteTrace(std::shared_ptr span, SocketId sid, std::shared_ptr history); + void Posted(size_t bytes); + void Completed(); + void Rejected(size_t bytes); + void Waited(bool sq, bool remote, uint64_t elapsed_us); + void PostingDone(int error = 0); + + private: + void Finish(); + std::shared_ptr _span; + const SocketId _sid; + std::shared_ptr _history; + const int64_t _start_us; + butil::atomic _bytes{0}, _wrs{0}, _completed{0}; + butil::atomic _queue_us{0}, _sq_wait_us{0}, _remote_wait_us{0}; + butil::atomic _posting_done{false}, _finished{false}; + butil::atomic _error{0}; +}; + +struct TraceContribution { + std::shared_ptr trace; + size_t bytes; +}; + +struct ConnectionSnapshot { + SocketId socket_id = INVALID_SOCKET_ID; + bool socket_failed = false; + std::string state; + std::string reason; + int error = 0; + uint64_t generation = 0; + uint64_t sq_size = 0, rq_size = 0, window_capacity = 0; + uint64_t sq_remaining = 0, remote_credit_remaining = 0; + uint64_t handshake_version = 0; + std::array counters{}; + std::vector traces; +}; +void DescribeConnectionJson(std::ostream& os, const ConnectionSnapshot& snapshot); + +class ConnectionStats { + public: + ConnectionStats(); + void Reset(); + void Add(Counter counter, int64_t value = 1); + void SetReason(const char* static_reason, int error = 0); + void Snapshot(ConnectionSnapshot* snapshot) const; + std::shared_ptr NewTrace(std::shared_ptr span, SocketId sid); + butil::atomic window_capacity{0}; + butil::atomic handshake_version{0}; + + private: + std::array, COUNTER_COUNT> _counters; + butil::atomic _generation{0}; + butil::atomic _reason{""}; + butil::atomic _error{0}; + std::shared_ptr _history; +}; + +void RegisterDiagnosticSocket(SocketId sid); +void UnregisterDiagnosticSocket(SocketId sid); +std::vector ListDiagnosticSockets(size_t limit); + +} // namespace urma +} // namespace brpc +#endif // BRPC_WITH_URMA +#endif // BRPC_URMA_OBSERVABILITY_H diff --git a/src/brpc/urma_transport.cpp b/src/brpc/urma_transport.cpp index c5cc163ffa..d91e026fbb 100644 --- a/src/brpc/urma_transport.cpp +++ b/src/brpc/urma_transport.cpp @@ -23,12 +23,14 @@ #include "butil/iobuf.h" #include "butil/logging.h" +#include "butil/time.h" #include "bthread/bthread.h" #include "bthread/types.h" #include "brpc/event_dispatcher.h" #include "brpc/input_messenger.h" #include "brpc/socket.h" +#include "brpc/span.h" #include "brpc/tcp_transport.h" #include "brpc/urma/urma_endpoint.h" #include "brpc/urma/urma_helper.h" @@ -38,6 +40,9 @@ namespace brpc { // Defined in urma_helper.cpp. DECLARE_bool(urma_use_polling); DECLARE_bool(urma_disable_bthread); +namespace urma { +DECLARE_int64(urma_bulk_min_bytes); +} void UrmaTransport::Init(Socket* socket, const SocketOptions& options) { CHECK(_urma_ep == nullptr); @@ -117,8 +122,15 @@ int UrmaTransport::WaitEpollOut(butil::atomic* epollout_butex, if (!_urma_ep->IsWritable()) { // Same caveat as RDMA: URMA cannot detect failure by writing, so // after a failed butex wait we must re-check _socket->Failed(). + const bool sq = _urma_ep->_sq_window_size.load(butil::memory_order_relaxed) == 0; + const bool remote = + _urma_ep->_remote_rq_window_size.load(butil::memory_order_relaxed) == 0; + const int64_t start_us = butil::cpuwide_time_us(); int rc = bthread::butex_wait(epollout_butex, expected_val, &duetime); + const int saved_wait_errno = errno; + _urma_ep->RecordWait(sq, remote, butil::cpuwide_time_us() - start_us); + errno = saved_wait_errno; if (rc < 0 && errno != EWOULDBLOCK && errno != ETIMEDOUT) { const int saved_errno = errno; PLOG(ERROR) << "Fail to wait epollout butex"; @@ -157,14 +169,17 @@ void UrmaTransport::ProcessEvent(bthread_attr_t attr) { void UrmaTransport::QueueMessage(InputMessageClosure& input_msg, int* num_bthread_created, bool last_msg) { - if (last_msg && !urma::FLAGS_urma_use_polling) { + // rpcz's server response path joins the queued write. A large response may + // need receive-credit completions from this same CQ before that write can + // finish. Keep the CQ drainer runnable while the application waits. + if (last_msg && !urma::FLAGS_urma_use_polling && !FLAGS_enable_rpcz) { return; } InputMessageBase* to_run_msg = input_msg.release(); if (!to_run_msg) { return; } - if (urma::FLAGS_urma_disable_bthread) { + if (urma::FLAGS_urma_disable_bthread && !FLAGS_enable_rpcz) { Transport::ProcessInputMessage(to_run_msg); return; } @@ -175,21 +190,91 @@ void UrmaTransport::QueueMessage(InputMessageClosure& input_msg, tmp.keytable_pool = _socket->keytable_pool(); tmp.tag = bthread_self_tag(); bthread_attr_set_name(&tmp, "ProcessInputMessage"); - if (!FLAGS_usercode_in_coroutine && bthread_start_background( - &th, &tmp, Transport::ProcessInputMessage, to_run_msg) == 0) { + if ((!FLAGS_usercode_in_coroutine || FLAGS_enable_rpcz) && + bthread_start_background(&th, &tmp, Transport::ProcessInputMessage, to_run_msg) == 0) { ++*num_bthread_created; } else { Transport::ProcessInputMessage(to_run_msg); } } +void UrmaTransport::ObserveWrite(size_t bytes) { + if (!_urma_ep || _urma_state.load(butil::memory_order_acquire) != URMA_ON) { + return; + } + _urma_ep->stats().Add(urma::writes); + const int64_t threshold = urma::FLAGS_urma_bulk_min_bytes; + if (threshold > 0 && bytes >= static_cast(threshold)) { + _urma_ep->stats().Add(urma::bulk_writes); + _urma_ep->stats().Add(urma::bulk_write_bytes, bytes); + } +} + +void UrmaTransport::TrackWrite(const butil::IOBuf* data, std::shared_ptr span) { + if (!_urma_ep) { + return; + } + auto trace = _urma_ep->stats().NewTrace(std::move(span), _socket->id()); + if (!trace) { + return; + } + std::lock_guard lock(_trace_mutex); + _write_traces[data] = std::move(trace); + _has_traces.store(true, butil::memory_order_release); +} + +void UrmaTransport::FinishWrite(const butil::IOBuf* data, int error) { + if (!_has_traces.load(butil::memory_order_acquire)) { + return; + } + std::shared_ptr trace; + { + std::lock_guard lock(_trace_mutex); + auto it = _write_traces.find(data); + if (it == _write_traces.end()) { + return; + } + trace = std::move(it->second); + _write_traces.erase(it); + _has_traces.store(!_write_traces.empty(), butil::memory_order_release); + } + trace->PostingDone(error); +} + +ssize_t UrmaTransport::CutWithTrace(butil::IOBuf** data, size_t ndata) { + if (!_urma_ep || _urma_state.load(butil::memory_order_acquire) != URMA_ON) { + return _tcp_transport->CutFromIOBufList(data, ndata); + } + if (!_has_traces.load(butil::memory_order_acquire)) { + return _urma_ep->CutFromIOBufList(data, ndata); + } + std::vector> traces(ndata); + { + std::lock_guard lock(_trace_mutex); + for (size_t i = 0; i < ndata; ++i) { + auto it = _write_traces.find(data[i]); + if (it != _write_traces.end()) { + traces[i] = it->second; + } + } + } + return _urma_ep->CutFromIOBufList(data, ndata, traces.data()); +} + void UrmaTransport::Debug(std::ostream& os) { - if (_urma_state.load(butil::memory_order_acquire) == URMA_ON && - _urma_ep) { + if (_urma_ep) { _urma_ep->DebugInfo(os); } } +bool UrmaTransport::DescribeJson(std::ostream& os) const { + if (!_urma_ep) { + return false; + } + urma::DescribeConnectionJson(os, _urma_ep->GetSnapshot()); + return true; +} + int UrmaTransport::ContextInitOrDie(bool server_or_not, const void* options) { if (server_or_not) { if (!OptionsAvailableOverUrma( diff --git a/src/brpc/urma_transport.h b/src/brpc/urma_transport.h index 9a496d56be..ef47376390 100644 --- a/src/brpc/urma_transport.h +++ b/src/brpc/urma_transport.h @@ -20,6 +20,8 @@ #if BRPC_WITH_URMA +#include +#include #include "brpc/channel.h" #include "brpc/socket.h" #include "brpc/transport.h" @@ -54,6 +56,11 @@ class UrmaTransport : public Transport { void QueueMessage(InputMessageClosure& input_msg, int* num_bthread_created, bool last_msg) override; void Debug(std::ostream& os) override; + bool DescribeJson(std::ostream& os) const; + ssize_t CutWithTrace(butil::IOBuf** data, size_t ndata); + void TrackWrite(const butil::IOBuf* data, std::shared_ptr span); + void ObserveWrite(size_t bytes); + void FinishWrite(const butil::IOBuf* data, int error); urma::UrmaEndpoint* GetUrmaEp() { CHECK(_urma_ep != nullptr); @@ -73,6 +80,10 @@ class UrmaTransport : public Transport { URMA_UNKNOWN }; + // Only sampled writes enter this side table; Socket::WriteRequest stays 64 bytes. + std::mutex _trace_mutex; + std::unordered_map> _write_traces; + butil::atomic _has_traces{false}; urma::UrmaEndpoint* _urma_ep = nullptr; butil::atomic _urma_state{URMA_UNKNOWN}; std::shared_ptr _tcp_transport; diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index a0fc2d0536..127d85c9ce 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -321,3 +321,21 @@ if(BUILD_FUZZ_TESTS) INSTALL_RPATH "$ORIGIN/lib") endforeach() endif() + +# The URMA diagnostic CLI only depends on the Python standard library. +find_package(Python3 COMPONENTS Interpreter QUIET) +if(Python3_Interpreter_FOUND) + add_test(NAME urma_diag_test + COMMAND ${Python3_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/urma_diag_test.py) +endif() + +# These functional data-path tests use the in-process mock, never a real device. +if(WITH_URMA AND (URMA_USE_MOCK STREQUAL "ON" OR + (URMA_USE_MOCK STREQUAL "AUTO" AND NOT URMA_LIB))) + foreach(mode IN ITEMS false true) + add_test(NAME urma_datapath_polling_${mode} + COMMAND brpc_urma_observability_unittest --urma_test_datapath + --urma_use_polling=${mode} --gtest_filter=UrmaDatapathTest.*) + set_tests_properties(urma_datapath_polling_${mode} PROPERTIES TIMEOUT 30) + endforeach() +endif() diff --git a/test/brpc_urma_observability_unittest.cpp b/test/brpc_urma_observability_unittest.cpp new file mode 100644 index 0000000000..3a026478a4 --- /dev/null +++ b/test/brpc_urma_observability_unittest.cpp @@ -0,0 +1,432 @@ +// 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 + +#include +#include +#include + +#if BRPC_WITH_URMA +#include "brpc/builtin/sockets_service.h" +#include "brpc/channel.h" +#include "brpc/closure_guard.h" +#include "brpc/controller.h" +#include "brpc/server.h" +#include "brpc/span.h" +#include "brpc/urma/urma_observability.h" +#include "brpc/urma/urma_helper.h" +#include "brpc/urma_transport.h" +#include "bthread/bthread.h" +#include "butil/third_party/rapidjson/document.h" +#include "butil/time.h" +#include "bvar/variable.h" +#include "echo.pb.h" + +namespace butil { +namespace iobuf { +extern void* (*blockmem_allocate)(size_t); +extern void (*blockmem_deallocate)(void*); +} // namespace iobuf +} // namespace butil + +DEFINE_bool(urma_test_datapath, false, + "Exercise the complete URMA data path (use a mock build or URMA hardware)"); + +namespace brpc { +namespace urma { +DECLARE_int32(urma_trace_sample_every); +extern bool g_skip_urma_init; +} // namespace urma +} // namespace brpc + +using namespace brpc; +using namespace brpc::urma; + +TEST(UrmaObservabilityTest, counters_export_and_reset) { + ConnectionStats stats; + const auto before = bvar::Variable::describe_exposed("urma_send_bytes"); + stats.Add(send_bytes, 123); + ConnectionSnapshot snapshot; + stats.Snapshot(&snapshot); + EXPECT_EQ(123u, snapshot.counters[send_bytes]); + const uint64_t generation = snapshot.generation; + EXPECT_EQ(std::stoll(before) + 123, + std::stoll(bvar::Variable::describe_exposed("urma_send_bytes"))); + stats.Reset(); + stats.Snapshot(&snapshot); + EXPECT_EQ(0u, snapshot.counters[send_bytes]); + EXPECT_EQ(generation + 1, snapshot.generation); +} + +TEST(UrmaObservabilityTest, concurrent_updates_and_snapshots) { + ConnectionStats stats; + std::thread writer([&] { + for (size_t i = 0; i < 10000; ++i) { + stats.Add(recv_bytes); + } + }); + ConnectionSnapshot snapshot; + for (size_t i = 0; i < 100; ++i) { + stats.Snapshot(&snapshot); + EXPECT_LE(snapshot.counters[recv_bytes], 10000u); + } + writer.join(); + stats.Snapshot(&snapshot); + EXPECT_EQ(10000u, snapshot.counters[recv_bytes]); +} + +TEST(UrmaObservabilityTest, json_preserves_identifiers_and_escapes_reason) { + ConnectionSnapshot snapshot; + snapshot.socket_id = UINT64_MAX - 1; + snapshot.reason = "quoted \"reason\"\n"; + std::ostringstream os; + DescribeConnectionJson(os, snapshot); + butil::rapidjson::Document document; + document.Parse(os.str().c_str()); + ASSERT_FALSE(document.HasParseError()); + EXPECT_EQ(std::to_string(snapshot.socket_id), document["socket_id"].GetString()); + EXPECT_EQ(snapshot.reason, document["reason"].GetString()); +} + +TEST(UrmaObservabilityTest, pool_watermarks_survive_free) { + const auto before = GetPoolSnapshot(); + PoolAllocated(); + PoolAllocated(); + EXPECT_EQ(before.in_use + 2, GetPoolSnapshot().in_use); + PoolDeallocated(); + PoolDeallocated(); + EXPECT_EQ(before.in_use, GetPoolSnapshot().in_use); + EXPECT_GE(GetPoolSnapshot().peak_in_use, before.in_use + 2); +} + +TEST(UrmaObservabilityTest, history_is_bounded) { + TraceHistory history; + for (size_t i = 0; i < 100; ++i) { + TraceRecord record; + record.span_id = i; + history.Push(record); + } + const auto records = history.Snapshot(); + ASSERT_EQ(64u, records.size()); + EXPECT_EQ(36u, records.front().span_id); + EXPECT_EQ(99u, records.back().span_id); +} + +TEST(UrmaObservabilityTest, trace_split_and_merged_wrs_keep_rpc_identity) { + auto history = std::make_shared(); + auto span1 = Span::CreateClientSpan("first", 0); + auto span2 = Span::CreateClientSpan("second", 0); + auto first = std::make_shared(span1, 7, history); + auto second = std::make_shared(span2, 7, history); + // First RPC spans two WRs; the second WR also carries a second RPC. + first->Posted(100); + first->Completed(); // May complete before posting has finished. + EXPECT_TRUE(history->Snapshot().empty()); + first->Posted(20); + second->Posted(40); + first->Waited(true, false, 23); + first->PostingDone(); + second->PostingDone(); + std::thread completion([&] { + first->Completed(); + second->Completed(); + }); + completion.join(); + const auto records = history->Snapshot(); + ASSERT_EQ(2u, records.size()); + EXPECT_EQ(span1->span_id(), records[0].span_id); + EXPECT_EQ(120u, records[0].bytes); + EXPECT_EQ(2u, records[0].completed_wrs); + EXPECT_EQ(23u, records[0].sq_wait_us); + EXPECT_EQ(span2->span_id(), records[1].span_id); + EXPECT_EQ(40u, records[1].bytes); + EXPECT_EQ(1u, records[1].completed_wrs); +} + +TEST(UrmaObservabilityTest, failed_post_is_not_counted_as_submitted) { + auto history = std::make_shared(); + auto span = Span::CreateClientSpan("failed", 0); + WriteTrace trace(span, 9, history); + trace.Posted(512); + trace.Rejected(512); + trace.PostingDone(EIO); + trace.PostingDone(EIO); // Failure cleanup must not duplicate the record. + auto records = history->Snapshot(); + ASSERT_EQ(1u, records.size()); + EXPECT_EQ(0u, records[0].wrs); + EXPECT_EQ(0u, records[0].bytes); + EXPECT_EQ(EIO, records[0].error); +} + +TEST(UrmaObservabilityTest, transport_trace_is_opt_in) { + ConnectionStats stats; + const int previous = FLAGS_urma_trace_sample_every; + auto span = Span::CreateClientSpan("sample", 0); + FLAGS_urma_trace_sample_every = 0; + EXPECT_FALSE(stats.NewTrace(span, 1)); + FLAGS_urma_trace_sample_every = 1; + EXPECT_TRUE(stats.NewTrace(span, 1)); + EXPECT_FALSE(stats.NewTrace(nullptr, 1)); + FLAGS_urma_trace_sample_every = previous; +} + +class UrmaDiagnosticsTest : public testing::Test { + protected: + void SetUp() override { + SocketOptions options; + options.socket_mode = SOCKET_MODE_URMA; + ASSERT_EQ(0, Socket::Create(options, &_sid)); + ASSERT_EQ(0, Socket::Address(_sid, &_socket)); + _transport = static_cast(_socket->_transport.get()); + _endpoint = _transport->GetUrmaEp(); + } + void TearDown() override { + if (_socket) { _socket->SetFailed(); _socket.reset(); } + } + SocketId _sid; + SocketUniquePtr _socket; + UrmaTransport* _transport; + UrmaEndpoint* _endpoint; +}; + +TEST_F(UrmaDiagnosticsTest, negotiated_capacity_uses_peer_limits) { + ParsedHello hello; + hello.buffer_size = 8192; + hello.recv_buffer_cnt = 15; + _endpoint->ApplyRemoteHello(hello); + const auto snapshot = _endpoint->GetSnapshot(); + EXPECT_EQ(13u, snapshot.window_capacity); // min(local SQ, peer RQ) - 3. + EXPECT_EQ(13u, snapshot.sq_remaining); + EXPECT_EQ(13u, snapshot.remote_credit_remaining); +} + +TEST_F(UrmaDiagnosticsTest, separate_wait_causes_and_reset) { + _endpoint->RecordWait(true, false, 40); + _endpoint->RecordWait(false, true, 80); + auto snapshot = _endpoint->GetSnapshot(); + EXPECT_EQ(40u, snapshot.counters[sq_wait_us]); + EXPECT_EQ(80u, snapshot.counters[remote_credit_wait_us]); + _endpoint->Reset(); + snapshot = _endpoint->GetSnapshot(); + EXPECT_EQ(0u, snapshot.counters[sq_wait_us]); + EXPECT_EQ(0u, snapshot.counters[remote_credit_wait_us]); + EXPECT_EQ("UNINIT", snapshot.state); +} + +TEST_F(UrmaDiagnosticsTest, fallback_reason_is_available_after_resources_release) { + _endpoint->FallbackToTcp(_transport, false, "injected allocation failure", ENOMEM); + std::ostringstream os; + EXPECT_TRUE(_socket->DescribeUrmaJson(os)); + butil::rapidjson::Document document; + document.Parse(os.str().c_str()); + ASSERT_FALSE(document.HasParseError()); + EXPECT_STREQ("FALLBACK_TCP", document["state"].GetString()); + EXPECT_STREQ("injected allocation failure", document["reason"].GetString()); + EXPECT_EQ(ENOMEM, document["error"].GetInt()); + EXPECT_EQ(1u, document["counters"]["tcp_fallback"].GetUint64()); +} + +TEST_F(UrmaDiagnosticsTest, failed_handshake_retains_reason) { + _endpoint->FailHandshake(_transport, EPROTO, "invalid peer hello"); + const auto snapshot = _endpoint->GetSnapshot(); + EXPECT_EQ("FAILED", snapshot.state); + EXPECT_TRUE(snapshot.socket_failed); + EXPECT_EQ("invalid peer hello", snapshot.reason); + EXPECT_EQ(1u, snapshot.counters[handshake_failed]); +} + +TEST_F(UrmaDiagnosticsTest, builtin_inventory_returns_structured_snapshot) { + Controller controller; + controller.http_request()._unresolved_path = "urma"; + SocketsService service; + service.default_method(&controller, nullptr, nullptr, nullptr); + ASSERT_FALSE(controller.Failed()); + butil::rapidjson::Document document; + document.Parse(controller.response_attachment().to_string().c_str()); + ASSERT_FALSE(document.HasParseError()); + EXPECT_EQ(1, document["schema_version"].GetInt()); + bool found = false; + for (auto it = document["connections"].Begin(); it != document["connections"].End(); ++it) { + if ((*it)["socket_id"].GetString() == std::to_string(_sid)) { + found = true; + } + } + EXPECT_TRUE(found); +} + +class DiagnosticEchoService : public test::EchoService { + void Echo(google::protobuf::RpcController*, const test::EchoRequest* request, + test::EchoResponse* response, google::protobuf::Closure* done) override { + ClosureGuard guard(done); + response->set_message(request->message()); + } +}; + +TEST(UrmaDatapathTest, large_concurrent_rpcs_have_counters_and_trace_completion) { + if (!FLAGS_urma_test_datapath) { + GTEST_SKIP() << "Pass --urma_test_datapath in a mock build or on URMA hardware"; + } + gflags::SetCommandLineOption("urma_buffer_count", "256"); + gflags::SetCommandLineOption("urma_sq_size", "16"); + gflags::SetCommandLineOption("urma_rq_size", "16"); + gflags::SetCommandLineOption("urma_prepared_jetty_cnt", "0"); + FLAGS_enable_rpcz = true; + FLAGS_urma_trace_sample_every = 1; + DiagnosticEchoService service; + Server server; + ASSERT_EQ(0, server.AddService(&service, SERVER_DOESNT_OWN_SERVICE)); + ServerOptions server_options; + server_options.socket_mode = SOCKET_MODE_URMA; + ASSERT_EQ(0, server.Start("127.0.0.1:0", &server_options)); + Channel channel; + ChannelOptions options; + options.socket_mode = SOCKET_MODE_URMA; + options.timeout_ms = 5000; + options.max_retry = 0; + ASSERT_EQ(0, channel.Init(server.listen_address(), &options)); + test::EchoService_Stub stub(&channel); + test::EchoRequest requests[4]; + test::EchoResponse responses[4]; + Controller controllers[4]; + CallId calls[4]; + for (size_t i = 0; i < 4; ++i) { + requests[i].set_message(std::string(128 * 1024, 'a' + i)); + calls[i] = controllers[i].call_id(); + stub.Echo(&controllers[i], &requests[i], &responses[i], DoNothing()); + } + for (size_t i = 0; i < 4; ++i) { + Join(calls[i]); + if (controllers[i].Failed()) { + for (auto sid : ListDiagnosticSockets(1024)) { + SocketUniquePtr socket; + if (Socket::AddressFailedAsWell(sid, &socket) >= 0) { + std::ostringstream debug; + socket->DescribeUrmaJson(debug); + ADD_FAILURE() << controllers[i].ErrorText() << " " << debug.str(); + socket->SetFailed(); + } + } + server.Stop(0); + server.Join(); + return; + } + EXPECT_TRUE(requests[i].message() == responses[i].message()) + << "payload mismatch in RPC " << i; + } + uint64_t bulk = 0, submitted = 0, completed = 0, traced = 0; + // The peer can respond before our final local send completion is drained. + for (size_t attempt = 0; attempt < 1000; ++attempt) { + bulk = submitted = completed = traced = 0; + for (auto sid : ListDiagnosticSockets(1024)) { + SocketUniquePtr socket; + if (Socket::Address(sid, &socket) != 0 || socket->socket_mode() != SOCKET_MODE_URMA) { + continue; + } + auto* transport = static_cast(socket->_transport.get()); + auto snapshot = transport->GetUrmaEp()->GetSnapshot(); + bulk += snapshot.counters[bulk_writes]; + submitted += snapshot.counters[send_wrs]; + completed += snapshot.counters[send_completed_wrs]; + for (const auto& trace : snapshot.traces) { + if (trace.wrs > 1 && trace.error == 0) { + EXPECT_EQ(trace.wrs, trace.completed_wrs); + ++traced; + } + } + } + if (bulk >= 8 && submitted == completed && traced >= 4) { + break; + } + bthread_usleep(1000); + } + EXPECT_GE(bulk, 8u); + EXPECT_GT(submitted, 8u); + EXPECT_EQ(submitted, completed); + EXPECT_GE(traced, 4u); + EXPECT_GT(GetPoolSnapshot().peak_in_use, 0u); + for (auto sid : ListDiagnosticSockets(1024)) { + SocketUniquePtr socket; + if (Socket::Address(sid, &socket) != 0 || socket->socket_mode() != SOCKET_MODE_URMA) { + continue; + } + auto* transport = static_cast(socket->_transport.get()); + auto* endpoint = transport->GetUrmaEp(); + const auto before = endpoint->GetSnapshot(); + if (before.state != "ESTABLISHED") { + continue; + } + butil::IOBuf probe; + probe.append("probe"); + butil::IOBuf* data[] = {&probe}; + const auto sq = endpoint->_sq_window_size.exchange(0, butil::memory_order_acq_rel); + EXPECT_EQ(-1, endpoint->CutFromIOBufList(data, 1)); + EXPECT_EQ(EAGAIN, errno); + const timespec deadline = butil::milliseconds_from_now(2); + transport->WaitEpollOut(socket->_epollout_butex, false, deadline); + endpoint->_sq_window_size.store(sq, butil::memory_order_release); + const auto remote = + endpoint->_remote_rq_window_size.exchange(0, butil::memory_order_acq_rel); + EXPECT_EQ(-1, endpoint->CutFromIOBufList(data, 1)); + EXPECT_EQ(EAGAIN, errno); + endpoint->_remote_rq_window_size.store(remote, butil::memory_order_release); + const auto after = endpoint->GetSnapshot(); + EXPECT_GT(after.counters[sq_blocked], before.counters[sq_blocked]); + EXPECT_GT(after.counters[remote_credit_blocked], before.counters[remote_credit_blocked]); + EXPECT_GT(after.counters[sq_wait_us], before.counters[sq_wait_us]); + EXPECT_EQ("probe", probe.to_string()); + break; + } + server.Stop(0); + server.Join(); +} + +TEST(UrmaDatapathTest, pool_exhaustion_exports_allocation_fallback) { + if (!FLAGS_urma_test_datapath) { + GTEST_SKIP() << "Requires mock or URMA hardware"; + } + gflags::SetCommandLineOption("urma_buffer_count", "256"); + gflags::SetCommandLineOption("urma_prepared_jetty_cnt", "0"); + GlobalUrmaInitializeOrDie(); + const auto before = GetPoolSnapshot(); + ASSERT_GT(before.capacity, 0u); + const auto misses = std::stoll(bvar::Variable::describe_exposed("urma_pool_exhausted")); + std::vector allocations; + for (size_t i = 0; i <= before.capacity; ++i) { + allocations.push_back(butil::iobuf::blockmem_allocate(before.buffer_size)); + } + EXPECT_EQ(before.capacity, GetPoolSnapshot().in_use); + EXPECT_GT(std::stoll(bvar::Variable::describe_exposed("urma_pool_exhausted")), misses); + for (auto allocation : allocations) { + if (allocation) { + butil::iobuf::blockmem_deallocate(allocation); + } + } + EXPECT_EQ(before.in_use, GetPoolSnapshot().in_use); +} + +#endif // BRPC_WITH_URMA + +int main(int argc, char** argv) { + testing::InitGoogleTest(&argc, argv); + gflags::ParseCommandLineFlags(&argc, &argv, true); +#if BRPC_WITH_URMA + brpc::urma::g_skip_urma_init = !FLAGS_urma_test_datapath; +#endif + return RUN_ALL_TESTS(); +} diff --git a/test/urma_diag_test.py b/test/urma_diag_test.py new file mode 100644 index 0000000000..c0fc4fa57a --- /dev/null +++ b/test/urma_diag_test.py @@ -0,0 +1,106 @@ +# 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. + +import contextlib +import copy +import http.server +import importlib.util +import io +import json +import pathlib +import threading +import unittest + +spec = importlib.util.spec_from_file_location( + "urma_diag", pathlib.Path(__file__).resolve().parents[1] / "tools/urma_diag.py") +diag = importlib.util.module_from_spec(spec) +spec.loader.exec_module(diag) + +SNAPSHOT = {"schema_version": 1, "pool": {"capacity": 8, "in_use": 2}, + "truncated": False, "connections": [ + {"socket_id": "18446744073709551614", "state": "ESTABLISHED", + "generation": 1, "counters": {"send_bytes": 100}, + "traces": [{"trace_id": "42", "span_id": "17"}]}]} + + +class Handler(http.server.BaseHTTPRequestHandler): + def do_GET(self): + if self.path == "/sockets/urma": + body = SNAPSHOT + elif self.path == "/sockets/" + SNAPSHOT["connections"][0]["socket_id"] + "?format=json": + body = SNAPSHOT["connections"][0] + else: + self.send_error(404, "socket not present") + return + self.send_response(200) + self.end_headers() + self.wfile.write(json.dumps(body).encode()) + + def log_message(self, *_): + pass + + +class UrmaDiagTest(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) + cls.thread = threading.Thread(target=cls.server.serve_forever) + cls.thread.start() + cls.address = "127.0.0.1:" + str(cls.server.server_port) + + @classmethod + def tearDownClass(cls): + cls.server.shutdown() + cls.server.server_close() + cls.thread.join() + + def run_cli(self, *args): + output = io.StringIO() + with contextlib.redirect_stdout(output): + code = diag.main(["--server", self.address, "--json", *args]) + self.assertEqual(0, code) + return [json.loads(line) for line in output.getvalue().splitlines()] + + def test_inventory_pool_socket_and_trace(self): + self.assertEqual(SNAPSHOT["connections"], self.run_cli("connections")[0]["connections"]) + self.assertEqual(SNAPSHOT["pool"], self.run_cli("pool")[0]) + sid = SNAPSHOT["connections"][0]["socket_id"] + self.assertEqual(sid, self.run_cli("socket", sid)[0]["socket_id"]) + self.assertEqual("42", self.run_cli("trace", sid, "--trace-id", "42")[0][0]["trace_id"]) + + def test_delta_and_connection_reuse(self): + current = copy.deepcopy(SNAPSHOT) + current["connections"][0]["counters"]["send_bytes"] += 80 + self.assertEqual(40, diag.delta(SNAPSHOT, current, 2)[0]["rates"]["send_bytes_per_second"]) + current["connections"][0]["generation"] += 1 + self.assertTrue(diag.delta(SNAPSHOT, current, 2)[0]["reset"]) + self.assertEqual({}, diag.delta(SNAPSHOT, current, 2)[0]["rates"]) + + def test_watch_is_finite(self): + self.assertEqual(2, len(self.run_cli("watch", "--interval", "0.001", "--count", "2"))) + + def test_missing_socket_returns_error(self): + error = io.StringIO() + with contextlib.redirect_stderr(error): + self.assertEqual(1, diag.main(["--server", self.address, "socket", "123"])) + self.assertIn("404", error.getvalue()) + + def test_invalid_scheme(self): + with self.assertRaisesRegex(ValueError, "HTTP"): + diag.fetch("file:///etc/passwd", 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/urma_diag.py b/tools/urma_diag.py new file mode 100644 index 0000000000..66c43f39d9 --- /dev/null +++ b/tools/urma_diag.py @@ -0,0 +1,138 @@ +#!/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. + +"""Read URMA snapshots from brpc's existing builtin-services listener.""" + +import argparse +import json +import sys +import time +import urllib.error +import urllib.parse +import urllib.request + + +def fetch(server, timeout, socket_id=None): + if "://" not in server: + server = "http://" + server + url = urllib.parse.urlsplit(server) + if url.scheme not in ("http", "https") or not url.netloc: + raise ValueError("server must be an HTTP(S) address") + if socket_id is not None and not socket_id.isdecimal(): + raise ValueError("socket ID must be a decimal integer") + path = "/sockets/" + socket_id if socket_id is not None else "/sockets/urma" + request_url = urllib.parse.urlunsplit( + (url.scheme, url.netloc, url.path.rstrip("/") + path, + "format=json" if socket_id is not None else "", "") + ) + with urllib.request.urlopen(request_url, timeout=timeout) as response: + snapshot = json.load(response) + if not isinstance(snapshot, dict): + raise ValueError("expected a URMA JSON object") + if socket_id is not None: + return {"schema_version": 1, "pool": {}, "connections": [snapshot], "truncated": False} + if snapshot.get("schema_version") != 1: + raise ValueError("unsupported URMA snapshot schema") + return snapshot + + +def delta(previous, current, elapsed): + """Compute rates only within the same SocketId and connection generation.""" + old = {item["socket_id"]: item for item in previous["connections"]} + result = [] + for item in current["connections"]: + before = old.get(item["socket_id"]) + row = {"socket_id": item["socket_id"], "state": item["state"]} + reset = before is None or before["generation"] != item["generation"] + row["reset"] = reset + row["rates"] = {} if reset else { + key + "_per_second": max(0, value - before["counters"].get(key, 0)) / elapsed + for key, value in item["counters"].items() + } + result.append(row) + return result + + +def select(snapshot, args): + if args.command == "pool": + return snapshot["pool"] + connections = snapshot["connections"] + if args.command in ("socket", "trace"): + connections = [item for item in connections if item["socket_id"] == args.socket_id] + if not connections: + suffix = " (inventory truncated)" if snapshot.get("truncated") else "" + raise ValueError("socket not present in snapshot" + suffix) + if args.command == "trace": + traces = connections[0]["traces"] + if args.trace_id: + traces = [item for item in traces if item["trace_id"] == args.trace_id] + return traces + return connections[0] if args.command == "socket" else { + "connections": connections, "truncated": snapshot.get("truncated", False) + } + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--server", required=True, help="builtin-services host:port or URL") + parser.add_argument("--timeout", type=float, default=5) + parser.add_argument("--json", action="store_true", help="compact JSON, one object per sample") + commands = parser.add_subparsers(dest="command", required=True) + commands.add_parser("connections") + commands.add_parser("pool") + socket = commands.add_parser("socket") + socket.add_argument("socket_id") + trace = commands.add_parser("trace") + trace.add_argument("socket_id") + trace.add_argument("--trace-id") + watch = commands.add_parser("watch") + watch.add_argument("--interval", type=float, default=1) + watch.add_argument("--count", type=int, default=0, help="0 runs until interrupted") + args = parser.parse_args(argv) + if args.timeout <= 0: + parser.error("timeout must be positive") + if args.command == "watch" and (args.interval <= 0 or args.count < 0): + parser.error("interval must be positive and count must be nonnegative") + + def emit(value): + print(json.dumps(value, indent=None if args.json else 2), flush=True) + + try: + previous = fetch(args.server, args.timeout, getattr(args, "socket_id", None)) + if args.command != "watch": + emit(select(previous, args)) + return 0 + previous_time = time.monotonic() + sample = 0 + while args.count == 0 or sample < args.count: + time.sleep(args.interval) + current = fetch(args.server, args.timeout) + current_time = time.monotonic() + emit({"elapsed_seconds": current_time - previous_time, + "connections": delta(previous, current, max(current_time - previous_time, 1e-6)), + "pool": current["pool"], "truncated": current.get("truncated", False)}) + previous, previous_time = current, current_time + sample += 1 + return 0 + except (ValueError, OSError, urllib.error.URLError) as error: + print("urma_diag: " + str(error), file=sys.stderr) + return 1 + except KeyboardInterrupt: + return 0 + + +if __name__ == "__main__": + sys.exit(main()) From 4bc6256a2e1b7b3698aa2eb744b45c71b881d9b3 Mon Sep 17 00:00:00 2001 From: Spicy-cream <2667314914@qq.com> Date: Mon, 2 Nov 2026 00:45:36 -0800 Subject: [PATCH 2/2] Simplify URMA observability documentation --- docs/cn/urma.md | 66 +++++++++-------------------- docs/en/urma.md | 107 +++++++++++------------------------------------- 2 files changed, 44 insertions(+), 129 deletions(-) diff --git a/docs/cn/urma.md b/docs/cn/urma.md index 4670bff114..da916bcfad 100644 --- a/docs/cn/urma.md +++ b/docs/cn/urma.md @@ -194,61 +194,35 @@ UrmaTransport 推荐用于**大包和跨节点**高吞吐路径，而 UBRing - 硬件数据通路需要受支持的 UMDK provider 和 `liburma`。 - 当前实现面向 Linux。 - ## 可观测性与诊断 -URMA 通过已有 `/vars` 和 Prometheus 导出进程级 `urma_*` 指标，通过已有 -内置服务入口提供连接快照。设置 `internal_port` 时使用内网端口，沿用现有 -内置服务访问控制。 +通过 `/vars/urma_*` 和 Prometheus 查看进程指标，通过 `/sockets/urma` 查看连接与 +注册 BufferPool 快照。配置 `internal_port` 时使用该端口。 ```sh curl http://127.0.0.1:8003/sockets/urma curl 'http://127.0.0.1:8003/sockets/123?format=json' -python3 tools/urma_diag.py --server 127.0.0.1:8003 connections python3 tools/urma_diag.py --server 127.0.0.1:8003 pool -python3 tools/urma_diag.py --server 127.0.0.1:8003 socket 123 -python3 tools/urma_diag.py --server 127.0.0.1:8003 --json watch --interval 1 --count 10 +python3 tools/urma_diag.py --server 127.0.0.1:8003 --json watch --interval 1 python3 tools/urma_diag.py --server 127.0.0.1:8003 trace 123 --trace-id 456 ``` -CLI 仅依赖 Python 3 标准库。列表最多返回 1024 条连接并输出 `truncated`； -按 SocketId 查询不受列表上限影响。ID 用十进制 JSON 字符串保存，避免 64 位 -精度丢失。列表 `schema_version` 为 1。 - -连接快照包括建链/回退状态和原因、实际协商窗口容量、剩余 SQ/远端 credit、 -数据 WR 提交与完成、在途字节、错误及两种资源等待。进程指标带 `urma_` 前缀， -连接快照的 `counters` 使用不带前缀的名称。 - -* `writes`、`bulk_writes`、`bulk_write_bytes`：进入已建立 URMA 数据路径的完整 - 序列化写入。`urma_bulk_min_bytes` 默认 65536 字节，非正值关闭分类。大小包括 - 协议头和附件；写入、WR 和业务 RPC 的数量不能混用。 -* `send_bytes` / `send_wrs` 与 `send_completed_bytes` / `send_completed_wrs` - 分别表示数据提交和本地完成。提交失败会回滚；本地完成不等于远端业务已处理。 -* `sq_blocked` / `remote_credit_blocked` 统计尝试发送时窗口为零的观察次数； - `sq_wait_us` / `remote_credit_wait_us` 按 butex 等待入口观察到的原因归因。 - 两者可能重叠，不能作为互斥阶段相加。 -* `urma_pool_capacity`、`urma_pool_buffer_size`、`urma_pool_in_use` 和 - `urma_pool_peak_in_use` 统计注册 BufferPool。IOBuf 引用和线程缓存可能在 RPC - 结束后继续持有 buffer。`urma_pool_exhausted` 和 `urma_malloc_fallback` 统计池耗尽 - 及分配回退尝试，后者还包括超大分配；分配回退不等于连接回退 TCP。 +CLI 使用 Python 3 标准库，支持 `connections`、`pool`、`socket`、`trace` 和 `watch`。 + +| 指标 | 内容 | +| --- | --- | +| 写入与传输 | bulk 写入、数据字节、WR 提交与本地完成、错误 | +| 流控 | SQ/远端 credit 剩余量、阻塞次数与等待时间 | +| 建链 | 握手成功、失败、TCP 回退及原因 | +| 注册池 | 容量、占用、峰值、耗尽与回退分配次数 | + +`urma_bulk_min_bytes` 默认 65536，按包含协议头和附件的序列化写入大小分类。 +等待时间按进入 butex 等待时的窗口状态归因，SQ 与远端 credit 等待可重叠。 开启 rpcz 并设置 `urma_trace_sample_every=N`，按 span ID 每 N 条采样一条； -默认 0 关闭，1 跟踪所有可用 rpcz span，仍受 rpcz 自身采样影响。客户端请求和 -baidu_std 服务端响应显式携带 Span 进入写队列，不依赖 KeepWrite 的当前 TLS。 -支持一条 RPC 拆成多个 WR、一个 WR 合并多个请求数据，纯 credit WR 不计入。 -旁路关联表仅用于采样写入，保持 `Socket::WriteRequest` 为 64 字节。 - -排队、首次提交和完成/失败事件写入 rpcz 注释。完成可能晚于 rpcz 序列化， -不会为观测延迟 RPC 完成或 rpcz 收集，因此每连接另保留最近 64 条已结束采样， -包含 trace/span/log ID，供 CLI 查询。`path=no_urma_wr` 表示没有提交 URMA 数据 WR， -例如走了 TCP 回退。历史在连接重置或销毁时清除。这些时间描述本地传输进度， -不能直接当作远端处理耗时或纯链路延迟。 - -快照是各原子字段的并发观察，不保证所有字段来自同一瞬间。Endpoint 重用时 -连接计数清零且 `generation` 递增，进程指标累计保留；CLI 不跨 generation 计算速率。 -完整指标口径与验证说明见 [英文文档](../en/urma.md#observability-and-diagnostics)。 - -使用 URMA mock 构建后运行 `brpc_urma_unittest`、 -`brpc_urma_observability_unittest` 和 `python3 test/urma_diag_test.py`；同时检查 -`WITH_URMA=OFF` 构建。上线前需在真实硬件上比较关闭跟踪及代表性采样率下的 -吞吐、CPU 和尾延迟；mock 测试不能代替硬件性能验证。 +默认 0 关闭，1 跟踪全部可用 span。采样记录排队、提交、本地完成及资源等待， +写入 rpcz 注释，并在每连接保留最近 64 条已结束记录。 + +快照中的 ID 为十进制字符串。列表最多返回 1024 条连接，超出时标记 `truncated`； +按 SocketId 可直接查询。连接重用时 `generation` 递增、连接计数清零，CLI 按 +同一 generation 计算速率。 diff --git a/docs/en/urma.md b/docs/en/urma.md index c021065319..758699a503 100644 --- a/docs/en/urma.md +++ b/docs/en/urma.md @@ -208,98 +208,39 @@ is tracked as a future enhancement. - A hardware data path requires a supported UMDK provider and `liburma`. - The current implementation targets Linux. - ## Observability and diagnostics -URMA exports process-wide `urma_*` bvars through `/vars` (and the existing -Prometheus exporter), with per-connection snapshots through the existing -builtin-services listener: +Use `/vars/urma_*` and Prometheus for process metrics, and `/sockets/urma` for +connection and registered BufferPool snapshots. Use `internal_port` when configured. ```sh curl http://127.0.0.1:8003/sockets/urma curl 'http://127.0.0.1:8003/sockets/123?format=json' -python3 tools/urma_diag.py --server 127.0.0.1:8003 connections python3 tools/urma_diag.py --server 127.0.0.1:8003 pool -python3 tools/urma_diag.py --server 127.0.0.1:8003 socket 123 -python3 tools/urma_diag.py --server 127.0.0.1:8003 --json watch --interval 1 --count 10 +python3 tools/urma_diag.py --server 127.0.0.1:8003 --json watch --interval 1 python3 tools/urma_diag.py --server 127.0.0.1:8003 trace 123 --trace-id 456 ``` -Use the internal port when `ServerOptions.internal_port` is configured. These -queries use the existing builtin-service access controls. Python 3 and its -standard library are sufficient for the CLI; there are no third-party packages. -The inventory returns at most 1024 connections and explicitly sets `truncated`. -An individual socket query bypasses that inventory limit. IDs are decimal JSON -strings to preserve all 64 bits. `schema_version` is currently 1 for the inventory. - -### Counter definitions - -All names below have the `urma_` prefix in `/vars`. Connection snapshots contain -these names without the prefix under `counters`. +The CLI uses the Python 3 standard library and supports `connections`, `pool`, +`socket`, `trace`, and `watch`. -| Counters | Meaning | +| Metrics | Contents | | --- | --- | -| `writes`, `bulk_writes`, `bulk_write_bytes` | Serialized writes entering the established URMA data path. Bulk means at least `--urma_bulk_min_bytes` bytes (default 65536); a nonpositive value disables classification. Includes protocol headers and attachments. A write is not a WR or necessarily an application RPC. | -| `send_bytes`, `send_wrs` | Payload bytes and data WRs accepted for submission. Failed single-WR posts are rolled back. | -| `send_completed_bytes`, `send_completed_wrs` | Locally completed data sends; this does not mean the peer application has processed them. | -| `recv_bytes`, `recv_wrs` | Received nonempty payloads accepted by the URMA receive path. Credit-only traffic is excluded. | -| `send_errors`, `recv_post_errors`, `completion_errors` | Data-send post failures, receive-post failures, and provider completion errors. | -| `sq_blocked`, `remote_credit_blocked` | Observations of a zero local SQ window or zero remote receive credit while attempting a send. Counts attempts, not distinct congestion episodes. | -| `sq_wait_us`, `remote_credit_wait_us` | Time spent in a writer's butex wait, attributed to windows observed empty at wait entry. Both can increase for the same wait; do not sum them as exclusive phases. | -| `handshake_success`, `handshake_failed`, `tcp_fallback` | Established URMA handshakes, terminal handshake failures, and negotiated/unavailable-resource TCP fallbacks. | -| `pool_exhausted`, `malloc_fallback` | Registered-pool allocation misses and fallback allocation attempts (also includes allocations larger than one pool buffer). These are process-wide. | - -The pool gauges are `urma_pool_capacity`, `urma_pool_buffer_size`, -`urma_pool_in_use`, and `urma_pool_peak_in_use`. They count registered buffers, -not live RPCs; IOBuf references and thread-local caches may retain buffers after -an RPC returns. The implementation calls this resource `BufferPool`, not a -separate bulk arena. Pool exhaustion and allocation fallback do not by themselves -imply that the entire connection switched to TCP. - -Snapshots include the connection state, negotiated data-window capacity, -remaining SQ/remote credit, in-flight data bytes/WRs, and the last diagnostic -reason/provider error. `FALLBACK_TCP` and `FAILED` remain queryable while the -Socket is addressable. Reads are race-free observations of individual counters, -not a transaction across the data path. Connection counters reset on endpoint -reuse and `generation` increases; process bvars remain cumulative. The CLI -suppresses rates across a changed generation. - -### Sampled RPC transport traces - -Enable rpcz and set `--urma_trace_sample_every=N` to trace one in N span IDs. -The default is 0 (disabled). `N=1` traces every available rpcz span; rpcz's own -sampling still applies. The client request and baidu_std server response carry -their explicit Span through Socket's write queue, so background KeepWrite and -batched writes do not borrow an unrelated bthread's trace context. - -A sampled write records queue-to-first-post time, bytes, data WR count, -local completions, per-cause wait time and failure. One RPC can span several WRs, -and a single WR can contain contributions from several RPCs. Credit-only WRs -never count as RPC data. Tracing uses a side table only for sampled writes and -keeps `Socket::WriteRequest` at 64 bytes. It adds no global data-path lock when -transport tracing is disabled. - -Annotations are added to the existing rpcz span at queue, first-post and local -completion/failure boundaries. Completion may occur after rpcz has serialized -the span; it does not delay RPC completion or rpcz collection. For that reason, -each connection also retains its latest 64 finished sampled writes in the JSON -snapshot, including trace/span/log IDs. `path=no_urma_wr` identifies writes that -never submitted a URMA data WR (for example, TCP fallback). The bounded history -lasts until connection reset or destruction. These measurements describe local -transport progress, not remote execution or wire-only latency. - -### Validation - -Build with `WITH_URMA=ON`, `URMA_USE_MOCK=ON`, and `BUILD_UNIT_TESTS=ON`, then run -`brpc_urma_unittest`, `brpc_urma_observability_unittest`, and -`python3 test/urma_diag_test.py`. Also build with `WITH_URMA=OFF` to check the -optional-transport boundary. Hardware throughput, CPU and tail latency should be -compared with tracing disabled and representative sampling rates before rollout; -mock results are functional validation, not a hardware performance measurement. - -CMake automatically registers `urma_datapath_polling_false` and -`urma_datapath_polling_true` in mock builds. These send concurrent 128 KiB RPCs, -check split/merged WR trace completion, inject empty SQ/remote-credit windows, -and exhaust the actual registered buffer pool. With rpcz enabled, URMA schedules -application callbacks outside the CQ drainer so a large traced response cannot -block the credit completions needed to finish its own write. +| Writes and transfers | Bulk writes, data bytes, submitted and locally completed WRs, errors | +| Flow control | Remaining SQ/remote credit, blocked attempts, wait time | +| Connections | Handshake success, failure, TCP fallback and reasons | +| Registered pool | Capacity, occupancy, peak occupancy, exhaustion, allocation fallback | + +`urma_bulk_min_bytes` defaults to 65536 and classifies serialized writes including +protocol headers and attachments. Wait time is attributed to the window state +at butex-wait entry; SQ and remote-credit waits can overlap. + +Enable rpcz and set `urma_trace_sample_every=N` to sample one in N span IDs. +The default is 0 (disabled); 1 traces every available span. Samples record queueing, +submission, local completion and resource waits in rpcz annotations. Each connection +also retains its latest 64 finished samples. + +Snapshot IDs are decimal strings. The inventory returns up to 1024 connections +and sets `truncated` for larger lists; individual sockets can be queried directly. +Connection reuse increments `generation` and resets connection counters. The CLI +calculates rates within the same generation.