diff --git a/.github/scripts/run-performance-matrix.sh b/.github/scripts/run-performance-matrix.sh new file mode 100755 index 0000000..5902204 --- /dev/null +++ b/.github/scripts/run-performance-matrix.sh @@ -0,0 +1,274 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)" +cd "$ROOT_DIR" + +MATRIX_DIR="rpc-consumer/target/performance-matrix" +SNAPSHOT_DIR="rpc-consumer/target/rpc-performance" +LOG_DIR="$MATRIX_DIR/logs" +RPC_PORT_START=19090 +RPC_PORT_END=19120 +BASE_PAYLOAD_BYTES="1024" +NACOS_ADDRESS="127.0.0.1:8848" + +rm -rf "$MATRIX_DIR" +mkdir -p "$MATRIX_DIR" "$LOG_DIR" + +if ! command -v tc >/dev/null 2>&1; then + echo "Linux tc is required for network emulation." >&2 + exit 1 +fi + +cleanup() { + sudo tc qdisc del dev lo root 2>/dev/null || true + docker rm -f rpc-perf-nacos >/dev/null 2>&1 || true +} +trap cleanup EXIT + +# Resolve and compile everything before tc is applied so dependency downloads never pollute network scenarios. +# JaCoCo is intentionally disabled in this performance job; coverage remains in the unit-test job. +mvn -B -ntp test-compile \ + -pl rpc-consumer -am \ + -DskipTests \ + -Djacoco.skip=true \ + -Drpc.registry=local + +scenario_index=0 +scenario_failures=0 + +run_scenario() { + local category="$1" + local scenario="$2" + local registry="$3" + local protocol="$4" + local serializer="$5" + local network="$6" + local payload_bytes="$7" + local warmup="$8" + local sequential="$9" + local concurrent="${10}" + local concurrency="${11}" + local request_timeout_ms="${12}" + local call_timeout_ms="${13}" + local require_all_success="${14}" + local registry_address="${15:-$NACOS_ADDRESS}" + + scenario_index=$((scenario_index + 1)) + local rpc_port=$((RPC_PORT_START + scenario_index)) + if (( rpc_port > RPC_PORT_END )); then + echo "Scenario count exceeded reserved tc RPC port range ${RPC_PORT_START}-${RPC_PORT_END}." >&2 + exit 1 + fi + + rm -rf "$SNAPSHOT_DIR" + + local output log status + output=$(printf "%s/%02d-%s.md" "$MATRIX_DIR" "$scenario_index" "$scenario") + log="$LOG_DIR/$scenario.log" + + echo "::group::Performance scenario: $scenario (RPC port $rpc_port)" + set +e + mvn -B -ntp test \ + -pl rpc-consumer -am \ + -Dtest=RpcPerformanceSnapshotTest \ + -Dsurefire.failIfNoSpecifiedTests=false \ + -Djacoco.skip=true \ + -Dlogback.configurationFile="$ROOT_DIR/rpc-benchmark/src/main/resources/logback.xml" \ + -Drpc.registry="$registry" \ + -Drpc.registry-address="$registry_address" \ + -Drpc.protocol="$protocol" \ + -Drpc.serializer="$serializer" \ + -Drpc.request-timeout-ms="$request_timeout_ms" \ + -Drpc.perf.call-timeout-ms="$call_timeout_ms" \ + -Drpc.perf.server-port="$rpc_port" \ + -Drpc.perf.payload-bytes="$payload_bytes" \ + -Drpc.perf.require-all-success="$require_all_success" \ + -Drpc.perf.warmup="$warmup" \ + -Drpc.perf.sequential-requests="$sequential" \ + -Drpc.perf.concurrent-requests="$concurrent" \ + -Drpc.perf.concurrency="$concurrency" \ + >"$log" 2>&1 + status=$? + set -e + echo "::endgroup::" + + local summary="$SNAPSHOT_DIR/summary.md" + if [[ "$status" -eq 0 && ! -s "$summary" ]]; then + echo "Scenario $scenario completed without producing $summary" >&2 + status=2 + fi + + { + echo "### \`$scenario\`" + echo + echo "- Status: $([[ "$status" -eq 0 ]] && echo 'PASS' || echo 'FAIL')" + echo "- Matrix dimension: \`$category\`" + echo "- Registry / Protocol / Serializer: \`$registry / $protocol / $serializer\`" + echo "- Request payload: \`${payload_bytes} bytes\`" + echo "- RPC port: \`$rpc_port\`" + echo "- Network profile: \`$network\`" + echo "- Strict success requirement: \`$require_all_success\`" + echo + + if [[ "$status" -eq 0 ]]; then + sed \ + -e '1{/^# RPC CI Performance Snapshot$/d;}' \ + -e '/^> Observational snapshot only\./d' \ + -e '/^Raw attempt samples are available/d' \ + -e 's/^## Framework metrics$/#### Framework metrics/' \ + "$summary" + else + echo '> Scenario failed. The tail of its Maven log is included below so unsupported or broken modes remain visible.' + echo + echo '```text' + tail -n 60 "$log" + echo '```' + fi + } > "$output" + + if [[ "$status" -ne 0 ]]; then + scenario_failures=$((scenario_failures + 1)) + echo "Scenario $scenario FAILED; continuing so the rest of the matrix still runs." >&2 + else + echo "Scenario $scenario passed." + fi +} + +start_nacos() { + docker rm -f rpc-perf-nacos >/dev/null 2>&1 || true + docker run -d --name rpc-perf-nacos \ + -e MODE=standalone \ + -e NACOS_AUTH_ENABLE=false \ + -e JVM_XMS=256m \ + -e JVM_XMX=256m \ + -e JVM_XMN=128m \ + -p 8848:8848 \ + -p 9848:9848 \ + -p 9849:9849 \ + nacos/nacos-server:v2.5.4 >/dev/null + + for _ in $(seq 1 60); do + if curl -fsS "http://127.0.0.1:8848/nacos/v1/console/health/readiness" >/dev/null; then + echo "Nacos is ready." + return + fi + sleep 2 + done + + docker logs rpc-perf-nacos || true + echo "Nacos did not become ready in time." >&2 + exit 1 +} + +reset_network() { + sudo tc qdisc del dev lo root 2>/dev/null || true +} + +shape_rpc_ports() { + reset_network + + # Every scenario gets a unique RPC endpoint in a reserved local port range. This prevents a previous Nacos + # ephemeral instance shutdown from racing a new JVM that re-registers the exact same ip:port identity. + # Only this reserved RPC data-plane range is shaped; Nacos 8848/9848/9849 stays untouched. + sudo tc qdisc add dev lo root handle 1: prio bands 3 + sudo tc qdisc add dev lo parent 1:1 handle 10: netem "$@" + + local port priority=1 + for port in $(seq $((RPC_PORT_START + 1)) "$RPC_PORT_END"); do + sudo tc filter add dev lo protocol ip parent 1:0 prio "$priority" u32 \ + match ip sport "$port" 0xffff flowid 1:1 + priority=$((priority + 1)) + sudo tc filter add dev lo protocol ip parent 1:0 prio "$priority" u32 \ + match ip dport "$port" 0xffff flowid 1:1 + priority=$((priority + 1)) + done +} + +apply_lan_profile() { + shape_rpc_ports delay 1ms 200us distribution normal rate 1gbit +} + +start_nacos + +# Production-like baseline: gRPC + Nacos + Protobuf + synthetic LAN conditions. +# On loopback, request and response packets are both shaped, so the configured one-way delay roughly doubles +# into added request/response RTT before framework processing time is included. +apply_lan_profile +run_scenario "baseline" "baseline-grpc-protobuf-nacos-lan" \ + "nacos" "grpc" "protobuf" "tc LAN: 1ms +/-0.2ms one-way, ~2ms added RTT, 1gbit" \ + "$BASE_PAYLOAD_BYTES" 20 40 120 8 8000 10000 true + +# Payload matrix: keep the production baseline stack and LAN network profile fixed. +for payload in 64 16384 262144; do + run_scenario "payload" "payload-${payload}b" \ + "nacos" "grpc" "protobuf" "tc LAN: 1ms +/-0.2ms one-way, ~2ms added RTT, 1gbit" \ + "$payload" 10 20 60 4 10000 12000 true +done + +# Serializer matrix: gRPC + Nacos + LAN remain fixed; Protobuf is represented by the baseline. +for serializer in kryo java json; do + run_scenario "serializer" "serializer-${serializer}" \ + "nacos" "grpc" "$serializer" "tc LAN: 1ms +/-0.2ms one-way, ~2ms added RTT, 1gbit" \ + "$BASE_PAYLOAD_BYTES" 10 20 60 4 8000 10000 true +done + +# Registry matrix: Nacos is represented by the baseline; Local is retained only as a comparison control. +run_scenario "registry" "registry-local-control" \ + "local" "grpc" "protobuf" "tc LAN: 1ms +/-0.2ms one-way, ~2ms added RTT, 1gbit" \ + "$BASE_PAYLOAD_BYTES" 10 20 60 4 8000 10000 true + +# Network matrix: gRPC + Nacos + Protobuf + 1 KiB payload remain fixed. +shape_rpc_ports delay 3ms 1ms distribution normal rate 500mbit +run_scenario "network" "network-cross-az" \ + "nacos" "grpc" "protobuf" "tc cross-AZ/private: 3ms +/-1ms one-way, ~6ms added RTT, 500mbit" \ + "$BASE_PAYLOAD_BYTES" 10 20 60 4 10000 12000 true + +shape_rpc_ports delay 25ms 5ms distribution normal loss 0.05% rate 100mbit +run_scenario "network" "network-public-internet" \ + "nacos" "grpc" "protobuf" "tc public internet: 25ms +/-5ms one-way, ~50ms added RTT, 0.05% loss, 100mbit" \ + "$BASE_PAYLOAD_BYTES" 10 20 60 4 12000 14000 false + +shape_rpc_ports delay 60ms 10ms distribution normal loss 0.1% rate 50mbit +run_scenario "network" "network-cross-region" \ + "nacos" "grpc" "protobuf" "tc cross-region: 60ms +/-10ms one-way, ~120ms added RTT, 0.1% loss, 50mbit" \ + "$BASE_PAYLOAD_BYTES" 8 16 48 4 15000 17000 false + +shape_rpc_ports delay 120ms 40ms distribution normal loss 1% rate 5mbit +run_scenario "network" "network-weak-mobile" \ + "nacos" "grpc" "protobuf" "tc weak/mobile: 120ms +/-40ms one-way, ~240ms added RTT, 1% loss, 5mbit" \ + "$BASE_PAYLOAD_BYTES" 6 12 36 3 20000 22000 false + +# Protocol matrix runs last so a currently broken protocol cannot hide the baseline/network/serializer data. +apply_lan_profile +for protocol in netty http http2; do + run_scenario "protocol" "protocol-${protocol}" \ + "nacos" "$protocol" "protobuf" "tc LAN: 1ms +/-0.2ms one-way, ~2ms added RTT, 1gbit" \ + "$BASE_PAYLOAD_BYTES" 10 20 60 4 10000 12000 true +done + +reset_network + +MATRIX_FILE="$MATRIX_DIR/matrix.md" +{ + echo "## CI Matrix — $(date -u '+%Y-%m-%d %H:%M:%S UTC') — \`${GITHUB_SHA:-local}\`" + echo + echo "- Runner: \`${RUNNER_OS:-local} ${RUNNER_ARCH:-unknown}\`" + echo "- Java: \`$(java -version 2>&1 | head -n 1 | tr -d '"')\`" + echo "- CPU visible: \`$(nproc)\`" + echo "- Baseline: \`gRPC + Nacos + Protobuf + 1 KiB + tc LAN\`" + echo "- Strategy: orthogonal matrix around the production-like baseline; Local Registry is a control only" + echo "- Network simulation: Linux \`tc netem\` on \`lo\`, filtered to reserved RPC ports \`$((RPC_PORT_START + 1))-$RPC_PORT_END\` only" + echo "- Nacos isolation: every scenario gets a unique RPC ip:port instance identity" + echo "- Coverage instrumentation: disabled for performance scenarios" + echo "- Scenario failures: \`$scenario_failures\`" + echo + cat "$MATRIX_DIR"/[0-9][0-9]-*.md +} > "$MATRIX_FILE" + +cat "$MATRIX_FILE" + +if [[ "$scenario_failures" -ne 0 ]]; then + echo "$scenario_failures performance scenario(s) failed; matrix data was still generated for diagnosis." >&2 + exit 1 +fi diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7cb1e5b..aa27e0c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -20,18 +20,15 @@ jobs: name: Build runs-on: ubuntu-24.04 timeout-minutes: 10 - steps: - name: Checkout code uses: actions/checkout@v7 - - name: Set up JDK 17 uses: actions/setup-java@v6 with: java-version: '17' distribution: 'temurin' cache: 'maven' - - name: Build all modules run: mvn -B -ntp clean package -DskipTests -Drpc.registry=local @@ -39,24 +36,20 @@ jobs: name: Unit Tests & Coverage runs-on: ubuntu-24.04 timeout-minutes: 10 - steps: - name: Checkout code uses: actions/checkout@v7 - - name: Set up JDK 17 uses: actions/setup-java@v6 with: java-version: '17' distribution: 'temurin' cache: 'maven' - - name: Run core, transport and starter unit tests run: >- mvn -B -ntp test -pl rpc-core,rpc-transport-netty,rpc-spring-boot-starter -am -Drpc.registry=local - - name: Upload unit test and coverage reports if: always() uses: actions/upload-artifact@v7 @@ -72,20 +65,15 @@ jobs: name: RPC Integration runs-on: ubuntu-24.04 timeout-minutes: 10 - outputs: - performance_markdown: ${{ steps.performance-record.outputs.markdown }} - steps: - name: Checkout code uses: actions/checkout@v7 - - name: Set up JDK 17 uses: actions/setup-java@v6 with: java-version: '17' distribution: 'temurin' cache: 'maven' - - name: Run local-registry Netty end-to-end test run: >- mvn -B -ntp test @@ -94,41 +82,6 @@ jobs: -Dsurefire.failIfNoSpecifiedTests=false -Drpc.registry=local -Drpc.protocol=netty - - # Performance data is observational only. Hosted Runner variance makes absolute - # latency/QPS unsuitable as a merge threshold. PR CI validates the snapshot; - # successful main CI runs persist the result into docs/PERFORMANCE.md. - - name: Run RPC performance snapshot - run: >- - mvn -B -ntp test - -pl rpc-consumer -am - -Dtest=RpcPerformanceSnapshotTest - -Dsurefire.failIfNoSpecifiedTests=false - -Drpc.registry=local - -Drpc.protocol=netty - -Drpc.serializer=kryo - -Drpc.perf.warmup=100 - -Drpc.perf.sequential-requests=200 - -Drpc.perf.concurrent-requests=1000 - -Drpc.perf.concurrency=16 - - - name: Prepare performance record - id: performance-record - shell: bash - run: | - summary='rpc-consumer/target/rpc-performance/summary.md' - test -f "$summary" - { - echo 'markdown< Observational snapshot only\./d' \ - -e '/^Raw per-request samples are available/d' \ - -e 's/^## Framework metrics$/### Framework metrics/' \ - "$summary" - echo 'RPC_PERFORMANCE_EOF' - } >> "$GITHUB_OUTPUT" - - name: Upload integration test reports if: always() uses: actions/upload-artifact@v7 @@ -138,66 +91,104 @@ jobs: if-no-files-found: warn path: '**/target/surefire-reports/' + performance-matrix: + name: RPC Performance Matrix + runs-on: ubuntu-24.04 + timeout-minutes: 35 + outputs: + performance_markdown: ${{ steps.performance-record.outputs.markdown }} + steps: + - name: Checkout code + uses: actions/checkout@v7 + - name: Set up JDK 17 + uses: actions/setup-java@v6 + with: + java-version: '17' + distribution: 'temurin' + cache: 'maven' + - name: Verify tc and Docker + shell: bash + run: | + command -v tc + sudo tc qdisc show dev lo + docker --version + - name: Run production-baseline performance matrix + shell: bash + run: bash .github/scripts/run-performance-matrix.sh + - name: Prepare performance record + id: performance-record + if: always() + shell: bash + run: | + matrix='rpc-consumer/target/performance-matrix/matrix.md' + if [[ ! -s "$matrix" ]]; then + echo 'Performance matrix file was not produced.' >&2 + exit 1 + fi + delimiter="RPC_PERFORMANCE_${GITHUB_RUN_ID}_${GITHUB_RUN_ATTEMPT}" + { + echo "markdown<<${delimiter}" + cat "$matrix" + echo "${delimiter}" + } >> "$GITHUB_OUTPUT" + + ci-gate: + name: CI Gate + if: always() + needs: [ build, unit-tests, rpc-integration, performance-matrix ] + runs-on: ubuntu-24.04 + timeout-minutes: 2 + steps: + - name: Verify required checks + shell: bash + run: | + test "${{ needs.build.result }}" = "success" + test "${{ needs.unit-tests.result }}" = "success" + test "${{ needs.rpc-integration.result }}" = "success" + test "${{ needs.performance-matrix.result }}" = "success" + record-performance: name: Record Performance if: >- github.event_name == 'push' && github.ref == 'refs/heads/main' && - needs.rpc-integration.result == 'success' - needs: [ rpc-integration ] + needs.ci-gate.result == 'success' && + needs.performance-matrix.result == 'success' + needs: [ ci-gate, performance-matrix ] runs-on: ubuntu-24.04 timeout-minutes: 5 permissions: contents: write - steps: - name: Checkout main uses: actions/checkout@v7 with: ref: main fetch-depth: 0 - - - name: Append performance snapshot + - name: Append performance matrix shell: bash env: - PERFORMANCE_MARKDOWN: ${{ needs.rpc-integration.outputs.performance_markdown }} + PERFORMANCE_MARKDOWN: ${{ needs.performance-matrix.outputs.performance_markdown }} run: | test -n "$PERFORMANCE_MARKDOWN" + git pull --ff-only origin main mkdir -p docs test -f docs/PERFORMANCE.md - - short_sha="${GITHUB_SHA:0:7}" - recorded_at="$(date -u '+%Y-%m-%d %H:%M:%S UTC')" - { echo echo '---' echo - echo "## CI Snapshot — ${recorded_at} — \`${short_sha}\`" - echo printf '%s\n' "$PERFORMANCE_MARKDOWN" } >> docs/PERFORMANCE.md - - name: Commit performance history shell: bash run: | git config user.name 'github-actions[bot]' git config user.email '41898282+github-actions[bot]@users.noreply.github.com' git add docs/PERFORMANCE.md - git commit -m 'docs: record RPC performance snapshot [skip ci]' + if git diff --cached --quiet; then + echo "No performance history change." + exit 0 + fi + git commit -m 'docs: record RPC performance matrix [skip ci]' git push origin HEAD:main - - ci-gate: - name: CI Gate - if: always() - needs: [ build, unit-tests, rpc-integration ] - runs-on: ubuntu-24.04 - timeout-minutes: 2 - - steps: - - name: Verify required checks - shell: bash - run: | - test "${{ needs.build.result }}" = "success" - test "${{ needs.unit-tests.result }}" = "success" - test "${{ needs.rpc-integration.result }}" = "success" diff --git a/docs/PERFORMANCE.md b/docs/PERFORMANCE.md index 765253c..32421e4 100644 --- a/docs/PERFORMANCE.md +++ b/docs/PERFORMANCE.md @@ -1,33 +1,109 @@ # RPC Performance History -该文档由 CI 持续维护，用于记录 Java-RPC 真实端到端调用链路的性能快照。 +该文档由 CI 持续维护，用于记录 Java-RPC 真实端到端调用链路在不同协议、序列化器、注册中心、payload 和网络条件下的性能数据。 -> 这些数据来自 GitHub Hosted Runner，适合观察趋势和发现明显回归，不作为绝对性能承诺或硬性合并门槛。测试固定使用 Local Registry + Netty + Kryo，并保持固定 warmup、请求量和并发度以提高不同运行之间的可比性。 +> 数据来自 GitHub Hosted Runner，适合比较实现差异、观察趋势和发现明显回归，不作为绝对性能承诺或硬性合并门槛。CI 使用以生产常见组合为中心的正交矩阵：一次只改变一个主要变量，避免全笛卡尔积既昂贵又难解释。 -## 测试配置 +## CI 性能矩阵 -- Warmup：100 次 RPC -- 顺序阶段：200 次 RPC，并发度 1 -- 并发阶段：1000 次 RPC，并发度 16 -- Registry：`local` -- Protocol：`netty` -- Serializer：`kryo` -- 记录指标：吞吐、Avg、P50、P95、P99、Max，以及 Client / Server 内置 metrics +### 生产基线 + +默认基线不再使用 Local Registry 或裸 loopback，而是： + +- Protocol: `grpc` +- Registry: `nacos`（CI 启动真实 `nacos/nacos-server:v2.5.4`） +- Serializer: `protobuf` +- Request payload: `1 KiB` +- Network: Linux `tc netem` 模拟内网：`delay 1ms ± 0.2ms` + `rate 1gbit` + +`tc` 只过滤固定 RPC 数据端口 `19090`，不会把 Nacos 的 `8848/9848/9849` 控制面流量一起塑形，因此 RPC 数据面的网络差异不会被注册中心心跳/发现请求混入。 + +由于请求和响应都会经过 loopback 的 egress qdisc，文档中的 `tc delay` 可理解为单向注入值，RPC 请求/响应增加的 RTT 通常约为其两倍，再叠加框架处理时间。例如 `delay 25ms` 的场景大致增加约 `50ms` RTT。这里的网络档位是稳定、可复现的合成条件，不声称精确代表某个云厂商或运营商 SLA。 + +### 协议 + +固定 `Nacos + Protobuf + 1 KiB + tc LAN`： + +- `grpc`（baseline） +- `netty` +- `http` +- `http2` + +### 序列化 + +固定 `gRPC + Nacos + 1 KiB + tc LAN`： + +- `protobuf`（baseline） +- `kryo` +- `java` +- `json` + +### 注册中心 + +固定 `gRPC + Protobuf + 1 KiB + tc LAN`： + +- `nacos`（baseline，真实 Docker） +- `local`（仅作为 control，不再作为默认基线） + +### Payload + +固定 `gRPC + Nacos + Protobuf + tc LAN`： + +- `64 B` +- `1 KiB`（baseline） +- `16 KiB` +- `256 KiB` + +这样可以区分协议/序列化固定开销与数据量扩大后的拷贝、编码和带宽成本。 + +### 网络 + +固定 `gRPC + Nacos + Protobuf + 1 KiB`，只塑形 RPC 端口： + +- 内网 LAN（baseline）：`delay 1ms ± 0.2ms`，约 `2ms` added RTT，`1gbit` +- 跨 AZ / 私网：`delay 3ms ± 1ms`，约 `6ms` added RTT，`500mbit` +- 普通公网：`delay 25ms ± 5ms`，约 `50ms` added RTT，`loss 0.05%`，`100mbit` +- 跨地域公网：`delay 60ms ± 10ms`，约 `120ms` added RTT，`loss 0.1%`，`50mbit` +- 弱网 / 移动网络：`delay 120ms ± 40ms`，约 `240ms` added RTT，`loss 1%`，`5mbit` + +公网、跨地域和弱网场景允许出现真实 RPC 失败，不再以“必须 100% 成功”作为采样前提；报告会继续跑完整个阶段并记录成功率、failed、timeout、成功吞吐与成功请求的 P95/P99。至少需要保留成功样本，否则该场景仍判定失败。 + +## 测量约束 + +性能 job 与覆盖率 job 分离：性能矩阵显式关闭 JaCoCo，避免 coverage instrumentation 污染延迟与吞吐。性能 JVM 使用低日志配置，避免逐请求 INFO 输出成为热路径瓶颈；普通单元测试和覆盖率仍按原 CI 配置执行。 + +矩阵采用“围绕生产基线一次只改变一个主要变量”的策略。某个协议如果失败，脚本不会立即停止整个矩阵，而是把该场景标记为 FAIL 后继续运行剩余场景，最后再统一让 job 失败。这样一个坏模式不会遮住其他协议、序列化器、Nacos 或网络档位的数据。 + +## 记录指标 + +每个场景都会记录： + +- 尝试请求数、成功数、失败数、成功率 +- Attempt throughput 与 successful throughput +- 成功 RPC 的 Avg / P50 / P95 / P99 / Max latency +- Client / Server 的 total、success、failed、timeout、active +- Client / Server 内置 metric 的平均与最大耗时 +- Registry / Protocol / Serializer / Payload / Network profile +- Java / CPU / Runner 信息 + +PR CI 会执行完整矩阵作为验证；只有 `main` CI 全绿后才自动把矩阵结果追加到本文档。 ## 历史记录 -### PR #11 CI 基线 +### PR #11 CI 单一基线（矩阵启用前） - Commit: `281d526ebe6ccd0e970453fc9094b85bf08cfe96` - Java: `17.0.20.1` - CPU visible to JVM: `4` +- Registry / Protocol / Serializer: `local / netty / kryo` +- Network: loopback | Phase | Requests | Concurrency | Throughput req/s | Avg ms | P50 ms | P95 ms | P99 ms | Max ms | | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | | sequential | 200 | 1 | 482.033 | 2.066 | 1.535 | 4.578 | 6.195 | 9.340 | | concurrent | 1000 | 16 | 1537.520 | 10.250 | 9.376 | 17.876 | 22.806 | 27.916 | -#### Framework metrics +
Framework metrics | Phase | Side | Total | Success | Failed | Timeout | Active | Metric avg ms | Metric max ms | | --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | @@ -36,12 +112,12 @@ | concurrent | CLIENT | 1000 | 1000 | 0 | 0 | 0 | 9.692 | 25.216 | | concurrent | SERVER | 1000 | 1000 | 0 | 0 | 0 | 0.661 | 12.092 | +
+ --- ## CI Snapshot — 2026-09-14 09:14:08 UTC — `c525edb` - - - Commit: `c525edb58daffb37267b544e5928f0fb4db5ab49` - Protocol: `netty` - Serializer: `kryo` @@ -62,4 +138,3 @@ | sequential | SERVER | 200 | 200 | 0 | 0 | 0 | 0.190 | 2.493 | | concurrent | CLIENT | 1000 | 1000 | 0 | 0 | 0 | 8.364 | 20.758 | | concurrent | SERVER | 1000 | 1000 | 0 | 0 | 0 | 0.791 | 5.687 | - diff --git a/rpc-benchmark/src/main/resources/logback.xml b/rpc-benchmark/src/main/resources/logback.xml index 2a299a7..07546d6 100644 --- a/rpc-benchmark/src/main/resources/logback.xml +++ b/rpc-benchmark/src/main/resources/logback.xml @@ -1,14 +1,12 @@ - + + + + %d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n + + - - - %d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n - - + diff --git a/rpc-consumer/src/test/java/com/xiaoyu/rpc/consumer/RpcPerformanceSnapshotTest.java b/rpc-consumer/src/test/java/com/xiaoyu/rpc/consumer/RpcPerformanceSnapshotTest.java index c838576..9514ac1 100644 --- a/rpc-consumer/src/test/java/com/xiaoyu/rpc/consumer/RpcPerformanceSnapshotTest.java +++ b/rpc-consumer/src/test/java/com/xiaoyu/rpc/consumer/RpcPerformanceSnapshotTest.java @@ -2,6 +2,7 @@ import com.google.protobuf.ByteString; import com.xiaoyu.rpc.api.HelloService; +import com.xiaoyu.rpc.common.extension.ExtensionLoader; import com.xiaoyu.rpc.common.serialization.Serializer; import com.xiaoyu.rpc.common.serialization.SerializerCode; import com.xiaoyu.rpc.common.vo.RpcRequest; @@ -9,22 +10,23 @@ import com.xiaoyu.rpc.core.config.RpcConfig; import com.xiaoyu.rpc.core.observability.RpcMetricSide; import com.xiaoyu.rpc.core.observability.RpcMetrics; +import com.xiaoyu.rpc.core.registry.ServiceDiscovery; import com.xiaoyu.rpc.core.server.RpcServer; import org.junit.jupiter.api.Test; import java.io.IOException; +import java.net.InetSocketAddress; import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.Path; import java.time.Instant; -import java.util.ArrayList; import java.util.Arrays; -import java.util.List; import java.util.Locale; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -35,7 +37,7 @@ * CI 中的轻量端到端性能快照。 * *

它的目标是生成可比较的观测数据，而不是用 GitHub Hosted Runner 的绝对性能值作为合并门槛。 - * JMH 仍然负责更严格的微基准；这里关注真实 RPC 主链路的延迟、吞吐和框架内置 metrics 是否一致。

+ * JMH 仍然负责更严格的微基准；这里关注真实 RPC 主链路的延迟、吞吐、失败率和框架内置 metrics 是否一致。

*/ public class RpcPerformanceSnapshotTest { @@ -43,6 +45,11 @@ public class RpcPerformanceSnapshotTest { private static final int SEQUENTIAL_REQUESTS = Integer.getInteger("rpc.perf.sequential-requests", 200); private static final int CONCURRENT_REQUESTS = Integer.getInteger("rpc.perf.concurrent-requests", 1000); private static final int CONCURRENCY = Integer.getInteger("rpc.perf.concurrency", 16); + private static final int PAYLOAD_BYTES = Math.max(1, Integer.getInteger("rpc.perf.payload-bytes", 1024)); + private static final int FIXED_SERVER_PORT = Integer.getInteger("rpc.perf.server-port", 0); + private static final int CALL_TIMEOUT_MS = Math.max(1, Integer.getInteger("rpc.perf.call-timeout-ms", 6000)); + private static final boolean REQUIRE_ALL_SUCCESS = Boolean.parseBoolean( + System.getProperty("rpc.perf.require-all-success", "true")); private static final Path OUTPUT_DIR = Path.of("target", "rpc-performance"); public static class HelloServiceImpl implements HelloService { @@ -54,10 +61,10 @@ public String sayHello(String name) { @Test void generatePerformanceSnapshot() throws Exception { - String registry = System.getProperty("rpc.registry", "local"); - String protocol = System.getProperty("rpc.protocol", "netty"); - String serializerName = System.getProperty("rpc.serializer", "kryo"); - int port = findFreePort(); + String registry = System.getProperty("rpc.registry", "nacos"); + String protocol = System.getProperty("rpc.protocol", "grpc"); + String serializerName = System.getProperty("rpc.serializer", "protobuf"); + int port = FIXED_SERVER_PORT > 0 ? FIXED_SERVER_PORT : findFreePort(); System.setProperty("rpc.registry", registry); System.setProperty("rpc.server-host", "127.0.0.1"); @@ -71,15 +78,16 @@ void generatePerformanceSnapshot() throws Exception { server = new RpcServer(); server.register(HelloService.class, new HelloServiceImpl()); server.start(); + // Use the same SPI discovery instance as RpcClient, not a direct-address shortcut. + // Registration acknowledgement does not imply subscriber cache visibility. + awaitDiscovery(registry, port); Serializer serializer = SerializerCode.getSerializerByCode(RpcConfig.getInstance().getSerializerCode()); try (RpcClient client = new RpcClient()) { warmUp(client, serializer); - PhaseResult sequential = runSequential(client, serializer); PhaseResult concurrent = runConcurrent(client, serializer); - - writeArtifacts(registry, protocol, serializerName, sequential, concurrent); + writeArtifacts(registry, protocol, serializerName, port, sequential, concurrent); } } finally { if (server != null) { @@ -95,23 +103,59 @@ void generatePerformanceSnapshot() throws Exception { } } + private static void awaitDiscovery(String registry, int port) throws Exception { + ServiceDiscovery discovery = ExtensionLoader.getExtensionLoader(ServiceDiscovery.class).getExtension(registry); + long started = System.nanoTime(); + long deadline = started + TimeUnit.SECONDS.toNanos(30); + Exception lastError = null; + InetSocketAddress expected = new InetSocketAddress("127.0.0.1", port); + while (System.nanoTime() < deadline) { + try { + InetSocketAddress actual = discovery.lookupService(HelloService.class.getName()); + if (expected.equals(actual)) { + System.out.printf(Locale.ROOT, "Discovery ready for %s in %.3f ms (excluded from RPC timing)%n", + expected, nanosToMillis(System.nanoTime() - started)); + return; + } + lastError = new IllegalStateException("Expected current instance " + expected + ", discovered " + actual); + } catch (Exception error) { + lastError = error; + } + Thread.sleep(200); + } + throw new IllegalStateException("Consumer did not discover current provider within 30 seconds: " + expected, lastError); + } + private static void warmUp(RpcClient client, Serializer serializer) throws Exception { for (int i = 0; i < WARMUP_REQUESTS; i++) { - invokeOnce(client, serializer, "Warmup-" + i); + try { + invokeOnce(client, serializer, requestName("Warmup", i)); + } catch (Exception e) { + if (REQUIRE_ALL_SUCCESS) { + throw e; + } + } } + waitForMetricsToSettle(); } private static PhaseResult runSequential(RpcClient client, Serializer serializer) throws Exception { RpcMetrics.getInstance().reset(); long[] samples = new long[SEQUENTIAL_REQUESTS]; + int successfulRequests = 0; long phaseStart = System.nanoTime(); - for (int i = 0; i < SEQUENTIAL_REQUESTS; i++) { - samples[i] = invokeOnce(client, serializer, "Sequential-" + i); + try { + samples[i] = invokeOnce(client, serializer, requestName("Sequential", i)); + successfulRequests++; + } catch (Exception e) { + if (REQUIRE_ALL_SUCCESS) { + throw e; + } + } } - long elapsedNanos = System.nanoTime() - phaseStart; - return createPhaseResult("sequential", SEQUENTIAL_REQUESTS, 1, elapsedNanos, samples); + return createPhaseResult("sequential", SEQUENTIAL_REQUESTS, successfulRequests, 1, elapsedNanos, samples); } private static PhaseResult runConcurrent(RpcClient client, Serializer serializer) throws Exception { @@ -120,35 +164,39 @@ private static PhaseResult runConcurrent(RpcClient client, Serializer serializer ExecutorService executor = Executors.newFixedThreadPool(CONCURRENCY); CountDownLatch startGate = new CountDownLatch(1); CountDownLatch doneGate = new CountDownLatch(CONCURRENT_REQUESTS); - AtomicReference failure = new AtomicReference<>(); - + AtomicInteger successfulRequests = new AtomicInteger(); + AtomicReference firstFailure = new AtomicReference<>(); + AtomicReference invalidResponse = new AtomicReference<>(); try { for (int i = 0; i < CONCURRENT_REQUESTS; i++) { final int requestIndex = i; executor.execute(() -> { try { startGate.await(); - samples[requestIndex] = invokeOnce(client, serializer, "Concurrent-" + requestIndex); - } catch (Throwable t) { - failure.compareAndSet(null, t); + samples[requestIndex] = invokeOnce(client, serializer, requestName("Concurrent", requestIndex)); + successfulRequests.incrementAndGet(); + } catch (AssertionError error) { + invalidResponse.compareAndSet(null, error); + } catch (Exception error) { + firstFailure.compareAndSet(null, error); } finally { doneGate.countDown(); } }); } - long phaseStart = System.nanoTime(); startGate.countDown(); - boolean completed = doneGate.await(45, TimeUnit.SECONDS); + boolean completed = doneGate.await(90, TimeUnit.SECONDS); long elapsedNanos = System.nanoTime() - phaseStart; - - assertTrue(completed, "Concurrent performance phase should finish within 45 seconds"); - Throwable error = failure.get(); - if (error != null) { - fail("Concurrent RPC performance phase failed", error); + assertTrue(completed, "Concurrent performance phase should finish within 90 seconds"); + if (invalidResponse.get() != null) { + fail("Data corruption is never an allowed weak-network failure", invalidResponse.get()); } - - return createPhaseResult("concurrent", CONCURRENT_REQUESTS, CONCURRENCY, elapsedNanos, samples); + if (REQUIRE_ALL_SUCCESS && firstFailure.get() != null) { + fail("Concurrent RPC performance phase failed", firstFailure.get()); + } + return createPhaseResult("concurrent", CONCURRENT_REQUESTS, successfulRequests.get(), + CONCURRENCY, elapsedNanos, samples); } finally { executor.shutdownNow(); executor.awaitTermination(5, TimeUnit.SECONDS); @@ -157,102 +205,120 @@ private static PhaseResult runConcurrent(RpcClient client, Serializer serializer private static long invokeOnce(RpcClient client, Serializer serializer, String name) throws Exception { long startNanos = System.nanoTime(); - String result = (String) client - .sendRequest(buildRequest(name, serializer), String.class) - .get(5, TimeUnit.SECONDS); + String result = (String) client.sendRequest(buildRequest(name, serializer), String.class) + .get(CALL_TIMEOUT_MS, TimeUnit.MILLISECONDS); long elapsedNanos = System.nanoTime() - startNanos; - - if (result == null || !result.contains(name)) { - throw new AssertionError("Unexpected RPC result for " + name + ": " + result); + if (!("Hello, " + name + "! (from Netty Server)").equals(result)) { + throw new AssertionError("Unexpected RPC result for payload marker: " + name.substring(0, Math.min(32, name.length()))); } return elapsedNanos; } - private static PhaseResult createPhaseResult(String name, int requests, int concurrency, - long elapsedNanos, long[] samples) { + private static PhaseResult createPhaseResult(String name, int requests, int successfulRequests, int concurrency, + long elapsedNanos, long[] samples) throws InterruptedException { + waitForMetricsToSettle(); RpcMetrics.Snapshot clientMetrics = RpcMetrics.getInstance().snapshot(RpcMetricSide.CLIENT); RpcMetrics.Snapshot serverMetrics = RpcMetrics.getInstance().snapshot(RpcMetricSide.SERVER); - - assertMetrics("client", requests, clientMetrics); - assertMetrics("server", requests, serverMetrics); - - long[] sorted = samples.clone(); - Arrays.sort(sorted); + assertClientMetrics(requests, successfulRequests, clientMetrics); + assertServerMetrics(requests, successfulRequests, serverMetrics); + long[] successfulSamples = Arrays.stream(samples).filter(sample -> sample > 0L).toArray(); + Arrays.sort(successfulSamples); long sum = 0L; - for (long sample : samples) { + for (long sample : successfulSamples) { sum += sample; } + double seconds = elapsedNanos / 1_000_000_000D; + double attemptedThroughput = seconds <= 0D ? 0D : requests / seconds; + double successfulThroughput = seconds <= 0D ? 0D : successfulRequests / seconds; + double successRatePct = requests == 0 ? 100D : successfulRequests * 100D / requests; + double averageMillis = successfulRequests == 0 ? 0D : nanosToMillis(sum / (double) successfulRequests); + double minMillis = successfulRequests == 0 ? 0D : nanosToMillis(successfulSamples[0]); + double p50Millis = successfulRequests == 0 ? 0D : nanosToMillis(percentile(successfulSamples, 0.50D)); + double p95Millis = successfulRequests == 0 ? 0D : nanosToMillis(percentile(successfulSamples, 0.95D)); + double p99Millis = successfulRequests == 0 ? 0D : nanosToMillis(percentile(successfulSamples, 0.99D)); + double maxMillis = successfulRequests == 0 ? 0D : nanosToMillis(successfulSamples[successfulSamples.length - 1]); + if (REQUIRE_ALL_SUCCESS) { + assertEquals(requests, successfulRequests, name + " successful request count mismatch"); + } else { + assertTrue(successfulRequests > 0, name + " should retain at least one successful RPC sample"); + } + return new PhaseResult(name, requests, successfulRequests, requests - successfulRequests, + concurrency, elapsedNanos, attemptedThroughput, successfulThroughput, successRatePct, + minMillis, averageMillis, p50Millis, p95Millis, p99Millis, maxMillis, + clientMetrics, serverMetrics, samples.clone()); + } - double elapsedSeconds = elapsedNanos / 1_000_000_000D; - double throughput = elapsedSeconds <= 0D ? 0D : requests / elapsedSeconds; - double averageMillis = requests == 0 ? 0D : nanosToMillis(sum / (double) requests); - - return new PhaseResult( - name, - requests, - concurrency, - elapsedNanos, - throughput, - nanosToMillis(sorted[0]), - averageMillis, - nanosToMillis(percentile(sorted, 0.50D)), - nanosToMillis(percentile(sorted, 0.95D)), - nanosToMillis(percentile(sorted, 0.99D)), - nanosToMillis(sorted[sorted.length - 1]), - clientMetrics, - serverMetrics, - samples.clone()); + private static void assertClientMetrics(int expectedRequests, int successfulRequests, RpcMetrics.Snapshot snapshot) { + assertEquals(expectedRequests, snapshot.totalRequests(), "client total requests mismatch"); + assertEquals(0, snapshot.activeRequests(), "client active requests should return to zero"); + assertEquals(successfulRequests, snapshot.successRequests(), "client success requests mismatch"); + assertEquals(expectedRequests - successfulRequests, snapshot.failedRequests(), "client failed requests mismatch"); + if (REQUIRE_ALL_SUCCESS) { + assertEquals(0, snapshot.timeoutRequests(), "client timeout requests should be zero"); + } } - private static void assertMetrics(String side, int expectedRequests, RpcMetrics.Snapshot snapshot) { - assertEquals(expectedRequests, snapshot.totalRequests(), side + " total requests mismatch"); - assertEquals(expectedRequests, snapshot.successRequests(), side + " success requests mismatch"); - assertEquals(0, snapshot.failedRequests(), side + " failed requests should be zero"); - assertEquals(0, snapshot.timeoutRequests(), side + " timeout requests should be zero"); - assertEquals(0, snapshot.activeRequests(), side + " active requests should return to zero"); + private static void assertServerMetrics(int expectedRequests, int successfulRequests, RpcMetrics.Snapshot snapshot) { + assertEquals(0, snapshot.activeRequests(), "server active requests should return to zero"); + if (REQUIRE_ALL_SUCCESS) { + assertEquals(expectedRequests, snapshot.totalRequests(), "server total requests mismatch"); + assertEquals(successfulRequests, snapshot.successRequests(), "server success requests mismatch"); + assertEquals(0, snapshot.failedRequests(), "server failed requests should be zero"); + assertEquals(0, snapshot.timeoutRequests(), "server timeout requests should be zero"); + } + // In lossy profiles a request from a previous phase can arrive after client timeout. + // Server metrics describe the observation interval, not client completion counts. + } + + private static void waitForMetricsToSettle() throws InterruptedException { + for (int i = 0; i < 100; i++) { + RpcMetrics.Snapshot client = RpcMetrics.getInstance().snapshot(RpcMetricSide.CLIENT); + RpcMetrics.Snapshot server = RpcMetrics.getInstance().snapshot(RpcMetricSide.SERVER); + if (client.activeRequests() == 0 && server.activeRequests() == 0) { + return; + } + Thread.sleep(20L); + } } private static long percentile(long[] sorted, double percentile) { int rank = (int) Math.ceil(percentile * sorted.length); - int index = Math.min(sorted.length - 1, Math.max(0, rank - 1)); - return sorted[index]; + return sorted[Math.min(sorted.length - 1, Math.max(0, rank - 1))]; } private static double nanosToMillis(double nanos) { return nanos / 1_000_000D; } + private static String requestName(String phase, int index) { + String prefix = phase + '-' + index + '|'; + return prefix + "x".repeat(Math.max(0, PAYLOAD_BYTES - prefix.length())); + } + private static RpcRequest buildRequest(String name, Serializer serializer) { - return RpcRequest.newBuilder() - .setInterfaceName(HelloService.class.getName()) - .setMethodName("sayHello") - .addParamTypes(String.class.getName()) - .addParameters(ByteString.copyFrom(serializer.serialize(name))) - .build(); + return RpcRequest.newBuilder().setInterfaceName(HelloService.class.getName()).setMethodName("sayHello") + .addParamTypes(String.class.getName()).addParameters(ByteString.copyFrom(serializer.serialize(name))).build(); } private static int findFreePort() throws IOException { - try (java.net.ServerSocket socket = new java.net.ServerSocket(0, 1, - java.net.InetAddress.getByName("127.0.0.1"))) { + try (java.net.ServerSocket socket = new java.net.ServerSocket(0, 1, java.net.InetAddress.getByName("127.0.0.1"))) { return socket.getLocalPort(); } } - private static void writeArtifacts(String registry, String protocol, String serializer, + private static void writeArtifacts(String registry, String protocol, String serializer, int serverPort, PhaseResult sequential, PhaseResult concurrent) throws IOException { Files.createDirectories(OUTPUT_DIR); Files.writeString(OUTPUT_DIR.resolve("performance.json"), - toJson(registry, protocol, serializer, sequential, concurrent), StandardCharsets.UTF_8); + toJson(registry, protocol, serializer, serverPort, sequential, concurrent), StandardCharsets.UTF_8); Files.writeString(OUTPUT_DIR.resolve("summary.md"), - toMarkdown(registry, protocol, serializer, sequential, concurrent), StandardCharsets.UTF_8); - Files.writeString(OUTPUT_DIR.resolve("latency-samples.csv"), - toCsv(sequential, concurrent), StandardCharsets.UTF_8); + toMarkdown(registry, protocol, serializer, serverPort, sequential, concurrent), StandardCharsets.UTF_8); + Files.writeString(OUTPUT_DIR.resolve("latency-samples.csv"), toCsv(sequential, concurrent), StandardCharsets.UTF_8); } - private static String toJson(String registry, String protocol, String serializer, + private static String toJson(String registry, String protocol, String serializer, int serverPort, PhaseResult sequential, PhaseResult concurrent) { - StringBuilder out = new StringBuilder(); - out.append("{\n"); + StringBuilder out = new StringBuilder("{\n"); field(out, "generatedAt", Instant.now().toString(), true, 1); field(out, "gitSha", envOrDefault("GITHUB_SHA", "local"), true, 1); out.append(" \"environment\": {\n"); @@ -260,21 +326,22 @@ private static String toJson(String registry, String protocol, String serializer field(out, "os", System.getProperty("os.name") + " " + System.getProperty("os.arch"), true, 2); numberField(out, "availableProcessors", Runtime.getRuntime().availableProcessors(), true, 2); numberField(out, "maxHeapMb", Runtime.getRuntime().maxMemory() / (1024D * 1024D), false, 2); - out.append(" },\n"); - out.append(" \"config\": {\n"); + out.append(" },\n \"config\": {\n"); field(out, "registry", registry, true, 2); field(out, "protocol", protocol, true, 2); field(out, "serializer", serializer, true, 2); + numberField(out, "serverPort", serverPort, true, 2); + numberField(out, "payloadBytes", PAYLOAD_BYTES, true, 2); + numberField(out, "callTimeoutMs", CALL_TIMEOUT_MS, true, 2); + booleanField(out, "requireAllSuccess", REQUIRE_ALL_SUCCESS, true, 2); numberField(out, "warmupRequests", WARMUP_REQUESTS, true, 2); numberField(out, "sequentialRequests", SEQUENTIAL_REQUESTS, true, 2); numberField(out, "concurrentRequests", CONCURRENT_REQUESTS, true, 2); numberField(out, "concurrency", CONCURRENCY, false, 2); - out.append(" },\n"); - out.append(" \"phases\": [\n"); + out.append(" },\n \"phases\": [\n"); appendPhaseJson(out, sequential, true); appendPhaseJson(out, concurrent, false); - out.append(" ]\n"); - out.append("}\n"); + out.append(" ]\n}\n"); return out.toString(); } @@ -282,24 +349,25 @@ private static void appendPhaseJson(StringBuilder out, PhaseResult phase, boolea out.append(" {\n"); field(out, "name", phase.name(), true, 3); numberField(out, "requests", phase.requests(), true, 3); + numberField(out, "successfulRequests", phase.successfulRequests(), true, 3); + numberField(out, "failedRequests", phase.failedRequests(), true, 3); + numberField(out, "successRatePct", phase.successRatePct(), true, 3); numberField(out, "concurrency", phase.concurrency(), true, 3); numberField(out, "durationMs", nanosToMillis(phase.elapsedNanos()), true, 3); - numberField(out, "throughputRps", phase.throughputRps(), true, 3); - out.append(" \"latencyMs\": {\n"); + numberField(out, "attemptedThroughputRps", phase.attemptedThroughputRps(), true, 3); + numberField(out, "successfulThroughputRps", phase.successfulThroughputRps(), true, 3); + out.append(" \"successfulLatencyMs\": {\n"); numberField(out, "min", phase.minMillis(), true, 4); numberField(out, "average", phase.averageMillis(), true, 4); numberField(out, "p50", phase.p50Millis(), true, 4); numberField(out, "p95", phase.p95Millis(), true, 4); numberField(out, "p99", phase.p99Millis(), true, 4); numberField(out, "max", phase.maxMillis(), false, 4); - out.append(" },\n"); - out.append(" \"clientMetrics\": "); + out.append(" },\n \"clientMetrics\": "); appendMetricsJson(out, phase.clientMetrics()); - out.append(",\n"); - out.append(" \"serverMetrics\": "); + out.append(",\n \"serverMetrics\": "); appendMetricsJson(out, phase.serverMetrics()); - out.append("\n }"); - out.append(comma ? ",\n" : "\n"); + out.append("\n }").append(comma ? ",\n" : "\n"); } private static void appendMetricsJson(StringBuilder out, RpcMetrics.Snapshot snapshot) { @@ -309,23 +377,24 @@ private static void appendMetricsJson(StringBuilder out, RpcMetrics.Snapshot sna .append(",\"timeout\":").append(snapshot.timeoutRequests()) .append(",\"active\":").append(snapshot.activeRequests()) .append(",\"averageLatencyMs\":").append(format(snapshot.averageLatencyMillis())) - .append(",\"maxLatencyMs\":").append(format(snapshot.maxLatencyMillis())) - .append('}'); + .append(",\"maxLatencyMs\":").append(format(snapshot.maxLatencyMillis())).append('}'); } - private static String toMarkdown(String registry, String protocol, String serializer, + private static String toMarkdown(String registry, String protocol, String serializer, int serverPort, PhaseResult sequential, PhaseResult concurrent) { - StringBuilder out = new StringBuilder(); - out.append("# RPC CI Performance Snapshot\n\n"); + StringBuilder out = new StringBuilder("# RPC CI Performance Snapshot\n\n"); out.append("> Observational snapshot only. GitHub Hosted Runner performance varies; these values are not merge thresholds.\n\n"); out.append("- Commit: `").append(envOrDefault("GITHUB_SHA", "local")).append("`\n"); out.append("- Protocol: `").append(protocol).append("`\n"); out.append("- Serializer: `").append(serializer).append("`\n"); out.append("- Registry: `").append(registry).append("`\n"); + out.append("- RPC port: `").append(serverPort).append("`\n"); + out.append("- Request payload: `").append(PAYLOAD_BYTES).append(" bytes`\n"); + out.append("- Require all requests to succeed: `").append(REQUIRE_ALL_SUCCESS).append("`\n"); out.append("- Java: `").append(System.getProperty("java.version")).append("`\n"); out.append("- CPU visible to JVM: `").append(Runtime.getRuntime().availableProcessors()).append("`\n\n"); - out.append("| Phase | Requests | Concurrency | Throughput req/s | Avg ms | P50 ms | P95 ms | P99 ms | Max ms |\n"); - out.append("| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |\n"); + out.append("| Phase | Attempts | Success | Failed | Success % | Concurrency | Attempt req/s | Success req/s | Avg ms | P50 ms | P95 ms | P99 ms | Max ms |\n"); + out.append("| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |\n"); appendPhaseMarkdown(out, sequential); appendPhaseMarkdown(out, concurrent); out.append("\n## Framework metrics\n\n"); @@ -335,63 +404,58 @@ private static String toMarkdown(String registry, String protocol, String serial appendMetricsMarkdown(out, sequential, "SERVER", sequential.serverMetrics()); appendMetricsMarkdown(out, concurrent, "CLIENT", concurrent.clientMetrics()); appendMetricsMarkdown(out, concurrent, "SERVER", concurrent.serverMetrics()); - out.append("\nRaw per-request samples are available in `latency-samples.csv`; machine-readable totals are in `performance.json`.\n"); + out.append("\nLatency percentiles include successful RPCs only. Low sample counts do not establish a reliable tail-latency SLA.\n"); + out.append("Nacos visibility polling and warmup are excluded from timing. In lossy profiles, server observation intervals may include late requests.\n"); return out.toString(); } private static void appendPhaseMarkdown(StringBuilder out, PhaseResult phase) { - out.append("| ").append(phase.name()) - .append(" | ").append(phase.requests()) - .append(" | ").append(phase.concurrency()) - .append(" | ").append(format(phase.throughputRps())) - .append(" | ").append(format(phase.averageMillis())) - .append(" | ").append(format(phase.p50Millis())) - .append(" | ").append(format(phase.p95Millis())) - .append(" | ").append(format(phase.p99Millis())) - .append(" | ").append(format(phase.maxMillis())) - .append(" |\n"); + out.append("| ").append(phase.name()).append(" | ").append(phase.requests()) + .append(" | ").append(phase.successfulRequests()).append(" | ").append(phase.failedRequests()) + .append(" | ").append(format(phase.successRatePct())).append(" | ").append(phase.concurrency()) + .append(" | ").append(format(phase.attemptedThroughputRps())).append(" | ").append(format(phase.successfulThroughputRps())) + .append(" | ").append(format(phase.averageMillis())).append(" | ").append(format(phase.p50Millis())) + .append(" | ").append(format(phase.p95Millis())).append(" | ").append(format(phase.p99Millis())) + .append(" | ").append(format(phase.maxMillis())).append(" |\n"); } - private static void appendMetricsMarkdown(StringBuilder out, PhaseResult phase, String side, - RpcMetrics.Snapshot snapshot) { - out.append("| ").append(phase.name()) - .append(" | ").append(side) - .append(" | ").append(snapshot.totalRequests()) - .append(" | ").append(snapshot.successRequests()) - .append(" | ").append(snapshot.failedRequests()) - .append(" | ").append(snapshot.timeoutRequests()) - .append(" | ").append(snapshot.activeRequests()) - .append(" | ").append(format(snapshot.averageLatencyMillis())) - .append(" | ").append(format(snapshot.maxLatencyMillis())) - .append(" |\n"); + private static void appendMetricsMarkdown(StringBuilder out, PhaseResult phase, String side, RpcMetrics.Snapshot snapshot) { + out.append("| ").append(phase.name()).append(" | ").append(side) + .append(" | ").append(snapshot.totalRequests()).append(" | ").append(snapshot.successRequests()) + .append(" | ").append(snapshot.failedRequests()).append(" | ").append(snapshot.timeoutRequests()) + .append(" | ").append(snapshot.activeRequests()).append(" | ").append(format(snapshot.averageLatencyMillis())) + .append(" | ").append(format(snapshot.maxLatencyMillis())).append(" |\n"); } private static String toCsv(PhaseResult... phases) { - StringBuilder out = new StringBuilder("phase,index,latency_ms\n"); + StringBuilder out = new StringBuilder("phase,index,status,latency_ms\n"); for (PhaseResult phase : phases) { long[] samples = phase.samplesNanos(); for (int i = 0; i < samples.length; i++) { - out.append(phase.name()).append(',') - .append(i).append(',') - .append(format(nanosToMillis(samples[i]))) - .append('\n'); + boolean success = samples[i] > 0L; + out.append(phase.name()).append(',').append(i).append(',').append(success ? "success" : "failure").append(','); + if (success) { + out.append(format(nanosToMillis(samples[i]))); + } + out.append('\n'); } } return out.toString(); } private static void field(StringBuilder out, String name, String value, boolean comma, int indent) { - out.append(" ".repeat(indent)) - .append('\"').append(name).append("\": \"") - .append(escapeJson(value)).append('\"') - .append(comma ? ",\n" : "\n"); + out.append(" ".repeat(indent)).append('\"').append(name).append("\": \"") + .append(escapeJson(value)).append('\"').append(comma ? ",\n" : "\n"); } private static void numberField(StringBuilder out, String name, double value, boolean comma, int indent) { - out.append(" ".repeat(indent)) - .append('\"').append(name).append("\": ") - .append(format(value)) - .append(comma ? ",\n" : "\n"); + out.append(" ".repeat(indent)).append('\"').append(name).append("\": ") + .append(format(value)).append(comma ? ",\n" : "\n"); + } + + private static void booleanField(StringBuilder out, String name, boolean value, boolean comma, int indent) { + out.append(" ".repeat(indent)).append('\"').append(name).append("\": ") + .append(value).append(comma ? ",\n" : "\n"); } private static String escapeJson(String value) { @@ -407,20 +471,10 @@ private static String envOrDefault(String name, String fallback) { return value == null || value.isBlank() ? fallback : value; } - private record PhaseResult( - String name, - int requests, - int concurrency, - long elapsedNanos, - double throughputRps, - double minMillis, - double averageMillis, - double p50Millis, - double p95Millis, - double p99Millis, - double maxMillis, - RpcMetrics.Snapshot clientMetrics, - RpcMetrics.Snapshot serverMetrics, - long[] samplesNanos) { + private record PhaseResult(String name, int requests, int successfulRequests, int failedRequests, + int concurrency, long elapsedNanos, double attemptedThroughputRps, double successfulThroughputRps, + double successRatePct, double minMillis, double averageMillis, double p50Millis, double p95Millis, + double p99Millis, double maxMillis, RpcMetrics.Snapshot clientMetrics, + RpcMetrics.Snapshot serverMetrics, long[] samplesNanos) { } } diff --git a/rpc-core/src/main/java/com/xiaoyu/rpc/core/registry/LocalRegistry.java b/rpc-core/src/main/java/com/xiaoyu/rpc/core/registry/LocalRegistry.java index 3185946..58110b1 100644 --- a/rpc-core/src/main/java/com/xiaoyu/rpc/core/registry/LocalRegistry.java +++ b/rpc-core/src/main/java/com/xiaoyu/rpc/core/registry/LocalRegistry.java @@ -1,28 +1,32 @@ package com.xiaoyu.rpc.core.registry; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import java.net.InetSocketAddress; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; public class LocalRegistry implements ServiceRegistry, ServiceDiscovery { + private static final Logger log = LoggerFactory.getLogger(LocalRegistry.class); private static final Map SERVICES = new ConcurrentHashMap<>(); @Override public void registerService(String serviceName, InetSocketAddress inetSocketAddress) { SERVICES.put(serviceName, inetSocketAddress); - System.out.println("LocalRegistry: Registered " + serviceName + " at " + inetSocketAddress); + log.info("LocalRegistry registered {} at {}", serviceName, inetSocketAddress); } @Override public InetSocketAddress lookupService(String serviceName) { - System.out.println("LocalRegistry: Looking up " + serviceName); + log.debug("LocalRegistry lookup: {}", serviceName); return SERVICES.get(serviceName); } @Override public void clearRegistry() { SERVICES.clear(); - System.out.println("LocalRegistry: Cleared all services."); + log.info("LocalRegistry cleared all services"); } } diff --git a/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/client/RpcStreamResponseHandler.java b/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/client/RpcStreamResponseHandler.java new file mode 100644 index 0000000..5229308 --- /dev/null +++ b/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/client/RpcStreamResponseHandler.java @@ -0,0 +1,58 @@ +package com.xiaoyu.rpc.core.client; + +import com.xiaoyu.rpc.common.vo.RpcResponse; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.SimpleChannelInboundHandler; +import io.netty.handler.codec.http2.Http2ResetFrame; +import io.netty.util.ReferenceCountUtil; + +import java.nio.channels.ClosedChannelException; + +/** + * One handler per HTTP/2 stream. Never install the connection-scoped pending-request + * handler on a child channel: closing that stream must not fail sibling requests. + */ +public final class RpcStreamResponseHandler extends SimpleChannelInboundHandler { + private final NettyRpcClientHandler connectionHandler; + private final String requestId; + + public RpcStreamResponseHandler(NettyRpcClientHandler connectionHandler, String requestId) { + this.connectionHandler = connectionHandler; + this.requestId = requestId; + } + + @Override + protected void channelRead0(ChannelHandlerContext ctx, RpcResponse response) throws Exception { + if (!requestId.equals(response.getRequestId())) { + exceptionCaught(ctx, new IllegalStateException("Response requestId does not match its HTTP/2 stream")); + return; + } + // Invoke only the response-routing operation, not connection lifecycle callbacks. + connectionHandler.channelRead(ctx, response); + } + + @Override + public void channelInactive(ChannelHandlerContext ctx) throws Exception { + connectionHandler.failRequest(requestId, new ClosedChannelException()); + super.channelInactive(ctx); + } + + @Override + public void userEventTriggered(ChannelHandlerContext ctx, Object event) throws Exception { + if (event instanceof Http2ResetFrame reset) { + try { + exceptionCaught(ctx, new IllegalStateException("HTTP/2 stream reset: " + reset.errorCode())); + } finally { + ReferenceCountUtil.release(event); + } + } else { + super.userEventTriggered(ctx, event); + } + } + + @Override + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { + connectionHandler.failRequest(requestId, cause); + ctx.close(); + } +} diff --git a/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/protocol/grpc/GrpcClientResponseHandler.java b/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/protocol/grpc/GrpcClientResponseHandler.java index 6fd54e0..cab3170 100644 --- a/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/protocol/grpc/GrpcClientResponseHandler.java +++ b/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/protocol/grpc/GrpcClientResponseHandler.java @@ -2,75 +2,80 @@ import com.xiaoyu.rpc.common.vo.RpcResponse; import com.xiaoyu.rpc.core.client.NettyRpcClientHandler; -import io.netty.buffer.ByteBuf; +import com.xiaoyu.rpc.core.config.RpcConfig; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; +import io.netty.handler.codec.CorruptedFrameException; import io.netty.handler.codec.http2.Http2DataFrame; import io.netty.handler.codec.http2.Http2Frame; import io.netty.handler.codec.http2.Http2Headers; import io.netty.handler.codec.http2.Http2HeadersFrame; -/** - * 将 gRPC/HTTP2 帧转换为内部 RpcResponse，并交给通用客户端处理器完成 requestId 关联。 - */ +/** Decode a unary response and wait for successful gRPC trailers before completing the RPC. */ class GrpcClientResponseHandler extends SimpleChannelInboundHandler { - private final NettyRpcClientHandler clientHandler; private final String requestId; + private final GrpcMessageAccumulator accumulator; + private RpcResponse response; + private boolean finished; GrpcClientResponseHandler(NettyRpcClientHandler clientHandler, String requestId) { this.clientHandler = clientHandler; this.requestId = requestId; + this.accumulator = new GrpcMessageAccumulator(RpcConfig.getInstance().getMaxMessageSize()); } @Override protected void channelRead0(ChannelHandlerContext ctx, Http2Frame frame) throws Exception { - if (frame instanceof Http2DataFrame) { - Http2DataFrame dataFrame = (Http2DataFrame) frame; - ByteBuf content = dataFrame.content(); - if (content.readableBytes() < 5) { - clientHandler.failRequest(requestId, new IllegalStateException("Invalid gRPC frame: missing 5-byte prefix")); - return; + if (finished) { + return; + } + if (frame instanceof Http2DataFrame data) { + byte[] payload = accumulator.append(ctx.alloc(), data.content()); + if (payload != null) { + response = RpcResponse.parseFrom(payload); } - - byte compressedFlag = content.readByte(); - if (compressedFlag != 0) { - clientHandler.failRequest(requestId, new UnsupportedOperationException("Compressed gRPC payload is not supported")); - return; + if (data.isEndStream()) { + throw new CorruptedFrameException("gRPC response ended without status trailers"); } - - int length = content.readInt(); - if (content.readableBytes() < length) { - clientHandler.failRequest(requestId, new IllegalStateException("Invalid gRPC frame: payload length mismatch")); - return; + } else if (frame instanceof Http2HeadersFrame headersFrame) { + Http2Headers headers = headersFrame.headers(); + if (headers.status() != null && !"200".contentEquals(headers.status())) { + throw new CorruptedFrameException("Unexpected gRPC HTTP status: " + headers.status()); } - - ByteBuf slice = content.readSlice(length); - RpcResponse response; - if (slice.nioBufferCount() > 0) { - response = RpcResponse.parseFrom(slice.nioBuffer()); - } else { - byte[] bytes = new byte[length]; - slice.readBytes(bytes); - response = RpcResponse.parseFrom(bytes); + if (headersFrame.isEndStream()) { + CharSequence status = headers.get("grpc-status"); + if (status == null || !"0".contentEquals(status)) { + throw new CorruptedFrameException("gRPC status=" + status + ", message=" + headers.get("grpc-message")); + } + accumulator.requireComplete(); + if (response == null || !requestId.equals(response.getRequestId())) { + throw new CorruptedFrameException("Missing response or mismatched gRPC requestId"); + } + finished = true; + ctx.fireChannelRead(response); } - ctx.fireChannelRead(response); - return; } + } - if (frame instanceof Http2HeadersFrame) { - Http2Headers headers = ((Http2HeadersFrame) frame).headers(); - CharSequence grpcStatus = headers.get("grpc-status"); - if (grpcStatus != null && !"0".contentEquals(grpcStatus)) { - CharSequence grpcMessage = headers.get("grpc-message"); - String message = grpcMessage == null ? "unknown grpc error" : grpcMessage.toString(); - clientHandler.failRequest(requestId, new RuntimeException("gRPC error status=" + grpcStatus + ", message=" + message)); - } + @Override + public void channelInactive(ChannelHandlerContext ctx) throws Exception { + accumulator.close(); + if (!finished) { + clientHandler.failRequest(requestId, new CorruptedFrameException("gRPC stream closed before successful trailers")); } + super.channelInactive(ctx); + } + + @Override + public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { + accumulator.close(); + super.handlerRemoved(ctx); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { + accumulator.close(); clientHandler.failRequest(requestId, cause); ctx.close(); } diff --git a/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/protocol/grpc/GrpcMessageAccumulator.java b/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/protocol/grpc/GrpcMessageAccumulator.java new file mode 100644 index 0000000..35fb690 --- /dev/null +++ b/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/protocol/grpc/GrpcMessageAccumulator.java @@ -0,0 +1,72 @@ +package com.xiaoyu.rpc.core.protocol.grpc; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.ByteBufAllocator; +import io.netty.handler.codec.CorruptedFrameException; +import io.netty.handler.codec.TooLongFrameException; + +/** Bounded, unary gRPC deframer. HTTP/2 DATA boundaries are not message boundaries. */ +final class GrpcMessageAccumulator implements AutoCloseable { + private final int maxMessageSize; + private ByteBuf buffer; + private boolean complete; + + GrpcMessageAccumulator(int maxMessageSize) { + if (maxMessageSize < 1 || maxMessageSize > Integer.MAX_VALUE - 5) { + throw new IllegalArgumentException("Invalid gRPC maximum message size"); + } + this.maxMessageSize = maxMessageSize; + } + + byte[] append(ByteBufAllocator allocator, ByteBuf fragment) { + int size = fragment.readableBytes(); + if (size == 0) { + return null; + } + if (complete) { + throw new CorruptedFrameException("Multiple messages on a unary gRPC stream"); + } + if (buffer == null) { + buffer = allocator.buffer(Math.min(256, maxMessageSize + 5), maxMessageSize + 5); + } + if (size > maxMessageSize + 5 - buffer.readableBytes()) { + throw new TooLongFrameException("gRPC message exceeds configured maximum"); + } + buffer.writeBytes(fragment, fragment.readerIndex(), size); + if (buffer.readableBytes() < 5) { + return null; + } + if (buffer.getUnsignedByte(0) != 0) { + throw new CorruptedFrameException("Compressed gRPC messages are not supported"); + } + int length = buffer.getInt(1); + if (length < 0 || length > maxMessageSize) { + throw new TooLongFrameException("Invalid gRPC message length: " + length); + } + if (buffer.readableBytes() > length + 5) { + throw new CorruptedFrameException("Extra bytes after unary gRPC message"); + } + if (buffer.readableBytes() < length + 5) { + return null; + } + byte[] payload = new byte[length]; + buffer.getBytes(5, payload); + complete = true; + close(); + return payload; + } + + void requireComplete() { + if (!complete) { + throw new CorruptedFrameException("gRPC stream ended with a missing or truncated message"); + } + } + + @Override + public void close() { + if (buffer != null) { + buffer.release(); + buffer = null; + } + } +} diff --git a/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/protocol/grpc/GrpcProtocol.java b/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/protocol/grpc/GrpcProtocol.java index 3271df3..135d827 100644 --- a/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/protocol/grpc/GrpcProtocol.java +++ b/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/protocol/grpc/GrpcProtocol.java @@ -2,7 +2,9 @@ import com.xiaoyu.rpc.common.vo.RpcRequest; import com.xiaoyu.rpc.core.client.NettyRpcClientHandler; +import com.xiaoyu.rpc.core.client.RpcStreamResponseHandler; import com.xiaoyu.rpc.core.protocol.Protocol; +import com.xiaoyu.rpc.core.protocol.http2.Http2ClientConnectionReadyHandler; import io.netty.channel.Channel; import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; @@ -33,17 +35,12 @@ public String getName() { @Override public void config(ChannelPipeline pipeline, boolean isServer, ChannelHandler serverHandler) { if (isServer) { - // 先接入 HTTP/2 帧编解码，处理握手并产出 Frame pipeline.addLast(Http2FrameCodecBuilder.forServer().build()); - - // 再通过 MultiplexHandler 为每个 Stream 创建子 Channel pipeline.addLast(new Http2MultiplexHandler(new ChannelInitializer() { @Override - protected void initChannel(Channel ch) throws Exception { + protected void initChannel(Channel ch) { ChannelPipeline p = ch.pipeline(); - // 在子 Channel 中添加 gRPC 适配器 p.addLast(new GrpcServerHandler(serverHandler)); - // 添加业务处理器 (复用现有的 NettyRpcHandler) p.addLast(serverHandler); } })); @@ -53,10 +50,10 @@ protected void initChannel(Channel ch) throws Exception { .autoAckPingFrame(true) .initialSettings(Http2Settings.defaultSettings().maxHeaderListSize(8192)) .build()); + Http2ClientConnectionReadyHandler.install(pipeline); pipeline.addLast(new Http2MultiplexHandler(new ChannelInboundHandlerAdapter() { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { - // 连接级残留帧统一释放，避免引用计数对象泄漏 ReferenceCountUtil.release(msg); } })); @@ -64,7 +61,17 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) { } @Override - public void sendRequest(Channel channel, RpcRequest request, NettyRpcClientHandler clientHandler) throws Exception { + public void sendRequest(Channel channel, RpcRequest request, NettyRpcClientHandler clientHandler) { + Http2ClientConnectionReadyHandler.readinessFuture(channel).whenComplete((ignored, readinessError) -> { + if (readinessError != null) { + clientHandler.failRequest(request.getRequestId(), readinessError); + return; + } + openStreamAndSend(channel, request, clientHandler); + }); + } + + private void openStreamAndSend(Channel channel, RpcRequest request, NettyRpcClientHandler clientHandler) { Http2StreamChannelBootstrap streamBootstrap = new Http2StreamChannelBootstrap(channel); streamBootstrap.open().addListener(openFuture -> { if (!openFuture.isSuccess()) { @@ -74,11 +81,11 @@ public void sendRequest(Channel channel, RpcRequest request, NettyRpcClientHandl Http2StreamChannel streamChannel = (Http2StreamChannel) openFuture.getNow(); streamChannel.pipeline().addLast(new GrpcClientResponseHandler(clientHandler, request.getRequestId())); - streamChannel.pipeline().addLast(clientHandler); + streamChannel.pipeline().addLast(new RpcStreamResponseHandler(clientHandler, request.getRequestId())); byte[] payload = request.toByteArray(); io.netty.buffer.ByteBuf body = streamChannel.alloc().buffer(payload.length + 5); - body.writeByte(0); // compressed-flag + body.writeByte(0); body.writeInt(payload.length); body.writeBytes(payload); @@ -88,9 +95,7 @@ public void sendRequest(Channel channel, RpcRequest request, NettyRpcClientHandl .scheme("http") .set(HttpHeaderNames.CONTENT_TYPE, "application/grpc") .set(HttpHeaderNames.TE, "trailers"); - - if (channel.remoteAddress() instanceof InetSocketAddress) { - InetSocketAddress remote = (InetSocketAddress) channel.remoteAddress(); + if (channel.remoteAddress() instanceof InetSocketAddress remote) { headers.authority(remote.getHostString() + ":" + remote.getPort()); } @@ -98,6 +103,7 @@ public void sendRequest(Channel channel, RpcRequest request, NettyRpcClientHandl streamChannel.writeAndFlush(new DefaultHttp2DataFrame(body, true)).addListener(writeFuture -> { if (!writeFuture.isSuccess()) { clientHandler.failRequest(request.getRequestId(), writeFuture.cause()); + streamChannel.close(); } }); }); diff --git a/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/protocol/grpc/GrpcServerHandler.java b/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/protocol/grpc/GrpcServerHandler.java index 2468985..8f4d638 100644 --- a/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/protocol/grpc/GrpcServerHandler.java +++ b/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/protocol/grpc/GrpcServerHandler.java @@ -2,111 +2,101 @@ import com.xiaoyu.rpc.common.vo.RpcRequest; import com.xiaoyu.rpc.common.vo.RpcResponse; +import com.xiaoyu.rpc.core.config.RpcConfig; import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelDuplexHandler; +import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelPromise; +import io.netty.handler.codec.CorruptedFrameException; import io.netty.handler.codec.http.HttpHeaderNames; -import io.netty.handler.codec.http2.*; +import io.netty.handler.codec.http2.DefaultHttp2DataFrame; +import io.netty.handler.codec.http2.DefaultHttp2Headers; +import io.netty.handler.codec.http2.DefaultHttp2HeadersFrame; +import io.netty.handler.codec.http2.Http2DataFrame; +import io.netty.handler.codec.http2.Http2Frame; +import io.netty.handler.codec.http2.Http2Headers; +import io.netty.handler.codec.http2.Http2HeadersFrame; import io.netty.util.ReferenceCountUtil; -import lombok.extern.slf4j.Slf4j; +import io.netty.util.concurrent.PromiseCombiner; -@Slf4j +/** One instance per unary gRPC stream, including its bounded fragmented-message state. */ public class GrpcServerHandler extends ChannelDuplexHandler { + private final GrpcMessageAccumulator accumulator; + private RpcRequest request; + private boolean dispatched; - // 透传的业务处理器（例如 NettyRpcHandler） - private final io.netty.channel.ChannelHandler busineesHandler; - - public GrpcServerHandler(io.netty.channel.ChannelHandler busineesHandler) { - this.busineesHandler = busineesHandler; + public GrpcServerHandler(ChannelHandler businessHandler) { + // The business handler follows this adapter in the child pipeline. + accumulator = new GrpcMessageAccumulator(RpcConfig.getInstance().getMaxMessageSize()); } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { - if (msg instanceof Http2Frame) { - try { - // 只在这里处理 gRPC 对应的 HTTP/2 Frame，转换成内部 RpcRequest - processFrame(ctx, (Http2Frame) msg); - } finally { - ReferenceCountUtil.release(msg); - } - } else { + if (!(msg instanceof Http2Frame)) { ctx.fireChannelRead(msg); + return; } - } - - private void processFrame(ChannelHandlerContext ctx, Http2Frame frame) throws Exception { - if (frame instanceof Http2HeadersFrame) { - Http2HeadersFrame headersFrame = (Http2HeadersFrame) frame; - Http2Headers headers = headersFrame.headers(); - CharSequence contentType = headers.get(HttpHeaderNames.CONTENT_TYPE); - if (contentType != null && contentType.toString().startsWith("application/grpc")) { - // Initial gRPC header received + try { + if (msg instanceof Http2DataFrame data) { + byte[] payload = accumulator.append(ctx.alloc(), data.content()); + if (payload != null) { + request = RpcRequest.parseFrom(payload); + } + if (data.isEndStream()) { + dispatch(ctx); + } + } else if (msg instanceof Http2HeadersFrame headers && headers.isEndStream()) { + dispatch(ctx); } + } finally { + ReferenceCountUtil.release(msg); } + } - if (frame instanceof Http2DataFrame) { - Http2DataFrame dataFrame = (Http2DataFrame) frame; - ByteBuf content = dataFrame.content(); - - // gRPC 数据帧固定前缀：1 字节压缩标记 + 4 字节消息长度 - if (content.readableBytes() < 5) - return; - - content.readByte(); // Compressed-Flag - int length = content.readInt(); - - if (content.readableBytes() < length) { - // 当前帧数据不足，回退读指针等待后续数据（简化处理，生产环境建议引入缓冲聚合） - content.resetReaderIndex(); - return; - } - - // 尽量避免中间大数组拷贝：先切片，再按底层存储类型选择解析路径 - ByteBuf slice = content.readSlice(length); - - RpcRequest rpcRequest; - if (slice.nioBufferCount() > 0) { - // 直接走 NIO Buffer 解析，少一次复制 - rpcRequest = RpcRequest.parseFrom(slice.nioBuffer()); - } else { - // 兜底路径：内存布局不支持 NIO Buffer 时退回字节数组解析 - byte[] bytes = new byte[length]; - slice.readBytes(bytes); - rpcRequest = RpcRequest.parseFrom(bytes); - } - - ctx.fireChannelRead(rpcRequest); + private void dispatch(ChannelHandlerContext ctx) { + accumulator.requireComplete(); + if (dispatched || request == null) { + throw new CorruptedFrameException("Invalid unary gRPC request"); } + dispatched = true; + ctx.fireChannelRead(request); } @Override - public void write(ChannelHandlerContext ctx, Object msg, io.netty.channel.ChannelPromise promise) throws Exception { - if (msg instanceof RpcResponse) { - RpcResponse response = (RpcResponse) msg; - try { - byte[] bytes = response.toByteArray(); - // gRPC 响应体同样要补上 5 字节前缀（压缩位 + 长度） - ByteBuf out = ctx.alloc().buffer(); - out.writeByte(0); - out.writeInt(bytes.length); - out.writeBytes(bytes); + public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { + if (!(msg instanceof RpcResponse response)) { + super.write(ctx, msg, promise); + return; + } + byte[] bytes = response.toByteArray(); + ByteBuf body = ctx.alloc().buffer(bytes.length + 5); + body.writeByte(0).writeInt(bytes.length).writeBytes(bytes); + Http2Headers headers = new DefaultHttp2Headers().status("200") + .set(HttpHeaderNames.CONTENT_TYPE, "application/grpc"); + Http2Headers trailers = new DefaultHttp2Headers().set("grpc-status", "0"); + PromiseCombiner writes = new PromiseCombiner(ctx.executor()); + writes.add(ctx.write(new DefaultHttp2HeadersFrame(headers, false))); + writes.add(ctx.write(new DefaultHttp2DataFrame(body, false))); + writes.add(ctx.write(new DefaultHttp2HeadersFrame(trailers, true))); + writes.finish(promise); + } - Http2Headers headers = new DefaultHttp2Headers().status("200") - .set(HttpHeaderNames.CONTENT_TYPE, "application/grpc"); - ctx.write(new DefaultHttp2HeadersFrame(headers)); + @Override + public void channelInactive(ChannelHandlerContext ctx) throws Exception { + accumulator.close(); + super.channelInactive(ctx); + } - ctx.write(new DefaultHttp2DataFrame(out, false)); + @Override + public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { + accumulator.close(); + super.handlerRemoved(ctx); + } - Http2Headers trailers = new DefaultHttp2Headers() - .set("grpc-status", "0") - .set("grpc-message", ""); - ctx.writeAndFlush(new DefaultHttp2HeadersFrame(trailers, true), promise); - } catch (Exception e) { - log.error("Failed to write gRPC response", e); - promise.setFailure(e); - } - return; - } - super.write(ctx, msg, promise); + @Override + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { + accumulator.close(); + ctx.close(); } } diff --git a/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/protocol/http2/Http2ClientConnectionReadyHandler.java b/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/protocol/http2/Http2ClientConnectionReadyHandler.java new file mode 100644 index 0000000..69b79c9 --- /dev/null +++ b/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/protocol/http2/Http2ClientConnectionReadyHandler.java @@ -0,0 +1,72 @@ +package com.xiaoyu.rpc.core.protocol.http2; + +import io.netty.channel.Channel; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInboundHandlerAdapter; +import io.netty.channel.ChannelPipeline; +import io.netty.handler.codec.http2.Http2ConnectionPrefaceAndSettingsFrameWrittenEvent; +import io.netty.util.AttributeKey; + +import java.nio.channels.ClosedChannelException; +import java.util.concurrent.CompletableFuture; + +/** + * Gates client-created HTTP/2 streams until Netty has written the connection preface and initial SETTINGS frame. + * + *

Netty explicitly requires clients not to write HTTP/2 application data before + * {@link Http2ConnectionPrefaceAndSettingsFrameWrittenEvent} has been processed. Without this gate a request can race + * channel activation and put a HEADERS frame on the wire before the HTTP/2 client preface.

+ */ +public final class Http2ClientConnectionReadyHandler extends ChannelInboundHandlerAdapter { + + private static final AttributeKey> READY_FUTURE = + AttributeKey.valueOf(Http2ClientConnectionReadyHandler.class, "readyFuture"); + + private final CompletableFuture readyFuture; + + private Http2ClientConnectionReadyHandler(CompletableFuture readyFuture) { + this.readyFuture = readyFuture; + } + + public static void install(ChannelPipeline pipeline) { + CompletableFuture readyFuture = new CompletableFuture<>(); + pipeline.channel().attr(READY_FUTURE).set(readyFuture); + pipeline.addLast(new Http2ClientConnectionReadyHandler(readyFuture)); + } + + public static CompletableFuture readinessFuture(Channel channel) { + CompletableFuture readyFuture = channel.attr(READY_FUTURE).get(); + if (readyFuture != null) { + return readyFuture; + } + + CompletableFuture failed = new CompletableFuture<>(); + failed.completeExceptionally(new IllegalStateException( + "HTTP/2 client readiness handler is not installed on channel " + channel)); + return failed; + } + + @Override + public void userEventTriggered(ChannelHandlerContext ctx, Object evt) { + if (evt instanceof Http2ConnectionPrefaceAndSettingsFrameWrittenEvent) { + readyFuture.complete(null); + } + ctx.fireUserEventTriggered(evt); + } + + @Override + public void channelInactive(ChannelHandlerContext ctx) { + if (!readyFuture.isDone()) { + readyFuture.completeExceptionally(new ClosedChannelException()); + } + ctx.fireChannelInactive(); + } + + @Override + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { + if (!readyFuture.isDone()) { + readyFuture.completeExceptionally(cause); + } + ctx.fireExceptionCaught(cause); + } +} diff --git a/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/protocol/http2/Http2Protocol.java b/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/protocol/http2/Http2Protocol.java index 80f509a..a49ba32 100644 --- a/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/protocol/http2/Http2Protocol.java +++ b/rpc-transport-netty/src/main/java/com/xiaoyu/rpc/core/protocol/http2/Http2Protocol.java @@ -4,16 +4,25 @@ import com.xiaoyu.rpc.common.serialization.SerializerCode; import com.xiaoyu.rpc.common.vo.RpcRequest; import com.xiaoyu.rpc.common.vo.RpcResponse; +import com.xiaoyu.rpc.core.client.RpcStreamResponseHandler; import com.xiaoyu.rpc.core.config.RpcConfig; +import com.xiaoyu.rpc.core.protocol.Protocol; +import com.xiaoyu.rpc.core.protocol.http.HttpRpcDecoder; +import com.xiaoyu.rpc.core.protocol.http.HttpRpcEncoder; +import io.netty.channel.Channel; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelPipeline; -import io.netty.handler.codec.http2.*; +import io.netty.handler.codec.http.HttpObjectAggregator; +import io.netty.handler.codec.http2.Http2FrameCodec; +import io.netty.handler.codec.http2.Http2FrameCodecBuilder; +import io.netty.handler.codec.http2.Http2MultiplexHandler; +import io.netty.handler.codec.http2.Http2Settings; +import io.netty.handler.codec.http2.Http2StreamChannel; +import io.netty.handler.codec.http2.Http2StreamChannelBootstrap; +import io.netty.handler.codec.http2.Http2StreamFrameToHttpObjectCodec; import io.netty.util.ReferenceCountUtil; -import com.xiaoyu.rpc.core.protocol.http.HttpRpcDecoder; -import com.xiaoyu.rpc.core.protocol.http.HttpRpcEncoder; -import com.xiaoyu.rpc.core.protocol.Protocol; public class Http2Protocol implements Protocol { @@ -29,84 +38,71 @@ public void config(ChannelPipeline pipeline, boolean isServer, io.netty.channel. if (isServer) { Http2FrameCodec frameCodec = Http2FrameCodecBuilder.forServer() - .autoAckSettingsFrame(true) // 自动确认设置帧 - .autoAckPingFrame(true) // 自动确认ping帧 + .autoAckSettingsFrame(true) + .autoAckPingFrame(true) .build(); - - // 使用新的构造方式，指定子通道的处理器 Http2MultiplexHandler multiplexHandler = new Http2MultiplexHandler( new ChannelInitializer() { @Override protected void initChannel(Http2StreamChannel ch) { - // 在子通道（Stream）中注入转换层 ch.pipeline().addLast(new Http2StreamFrameToHttpObjectCodec(true)); - ch.pipeline().addLast(new io.netty.handler.codec.http.HttpObjectAggregator(512 * 1024)); + ch.pipeline().addLast(new HttpObjectAggregator(512 * 1024)); ch.pipeline().addLast(new HttpRpcDecoder(serializer, RpcRequest.class)); ch.pipeline().addLast(new HttpRpcEncoder(serializer)); - if (serverHandler != null) { ch.pipeline().addLast(serverHandler); } } }); pipeline.addLast(frameCodec, multiplexHandler); - } else { - // 客户端核心配置 Http2FrameCodec frameCodec = Http2FrameCodecBuilder.forClient() - // 强制自动处理一些基础帧，防止它们掉到 TailContext .autoAckSettingsFrame(true) .autoAckPingFrame(true) .initialSettings(Http2Settings.defaultSettings().maxHeaderListSize(8192)) .build(); - pipeline.addLast(frameCodec); - // MultiplexHandler 必须紧跟其后 + Http2ClientConnectionReadyHandler.install(pipeline); pipeline.addLast(new Http2MultiplexHandler(new ChannelInboundHandlerAdapter() { - @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { - // 如果还有残留的设置帧传到这里，说明 FrameCodec 没拦截住 - // 这里直接释放，避免引用计数对象泄漏 ReferenceCountUtil.release(msg); } - - @Override - public void userEventTriggered(ChannelHandlerContext ctx, Object evt) { - // 处理用户事件，如Http2SettingsAckFrame等 - ctx.fireUserEventTriggered(evt); - } })); } } @Override - public void sendRequest(io.netty.channel.Channel channel, RpcRequest request, - com.xiaoyu.rpc.core.client.NettyRpcClientHandler clientHandler) throws Exception { - RpcConfig rpcConfig = RpcConfig.getInstance(); - Serializer serializer = SerializerCode.getSerializerByCode(rpcConfig.getSerializerCode()); - - // HTTP/2 每个请求走独立 Stream，底层 TCP 连接仍然复用同一个 Channel - io.netty.handler.codec.http2.Http2StreamChannelBootstrap streamBootstrap = new io.netty.handler.codec.http2.Http2StreamChannelBootstrap( - channel); + public void sendRequest(Channel channel, RpcRequest request, + com.xiaoyu.rpc.core.client.NettyRpcClientHandler clientHandler) { + Serializer serializer = SerializerCode.getSerializerByCode(RpcConfig.getInstance().getSerializerCode()); + Http2ClientConnectionReadyHandler.readinessFuture(channel).whenComplete((ignored, readinessError) -> { + if (readinessError != null) { + clientHandler.failRequest(request.getRequestId(), readinessError); + return; + } + openStreamAndSend(channel, request, serializer, clientHandler); + }); + } + private void openStreamAndSend(Channel channel, RpcRequest request, Serializer serializer, + com.xiaoyu.rpc.core.client.NettyRpcClientHandler clientHandler) { + Http2StreamChannelBootstrap streamBootstrap = new Http2StreamChannelBootstrap(channel); streamBootstrap.open().addListener(f -> { if (!f.isSuccess()) { clientHandler.failRequest(request.getRequestId(), f.cause()); return; } - Http2StreamChannel streamChannel = (Http2StreamChannel) f.getNow(); - // 每个 Stream 都有独立 pipeline，避免多请求之间相互干扰 streamChannel.pipeline().addLast(new Http2StreamFrameToHttpObjectCodec(false)); - streamChannel.pipeline().addLast(new io.netty.handler.codec.http.HttpObjectAggregator(512 * 1024)); + streamChannel.pipeline().addLast(new HttpObjectAggregator(512 * 1024)); streamChannel.pipeline().addLast(new HttpRpcEncoder(serializer)); streamChannel.pipeline().addLast(new HttpRpcDecoder(serializer, RpcResponse.class)); - streamChannel.pipeline().addLast(clientHandler); - + streamChannel.pipeline().addLast(new RpcStreamResponseHandler(clientHandler, request.getRequestId())); streamChannel.writeAndFlush(request).addListener(writeFuture -> { if (!writeFuture.isSuccess()) { clientHandler.failRequest(request.getRequestId(), writeFuture.cause()); + streamChannel.close(); } }); }); diff --git a/rpc-transport-netty/src/test/java/com/xiaoyu/rpc/core/client/RpcStreamResponseHandlerTest.java b/rpc-transport-netty/src/test/java/com/xiaoyu/rpc/core/client/RpcStreamResponseHandlerTest.java new file mode 100644 index 0000000..d1956f0 --- /dev/null +++ b/rpc-transport-netty/src/test/java/com/xiaoyu/rpc/core/client/RpcStreamResponseHandlerTest.java @@ -0,0 +1,78 @@ +package com.xiaoyu.rpc.core.client; + +import com.xiaoyu.rpc.common.vo.RpcResponse; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.http2.DefaultHttp2ResetFrame; +import io.netty.handler.codec.http2.Http2Error; +import org.junit.jupiter.api.Test; +import java.util.concurrent.CompletableFuture; +import static org.junit.jupiter.api.Assertions.*; + +class RpcStreamResponseHandlerTest { + @Test + void completedStreamDoesNotFailSiblingOrNextRequest() { + NettyRpcClientHandler shared = new NettyRpcClientHandler(); + EmbeddedChannel parent = new EmbeddedChannel(shared); + EmbeddedChannel first = new EmbeddedChannel(new RpcStreamResponseHandler(shared, "a")); + EmbeddedChannel second = new EmbeddedChannel(new RpcStreamResponseHandler(shared, "b")); + try { + CompletableFuture a = new CompletableFuture<>(); + CompletableFuture b = new CompletableFuture<>(); + shared.addFuture("a", a); + shared.addFuture("b", b); + first.writeInbound(RpcResponse.newBuilder().setRequestId("a").build()); + first.close(); + assertFalse(a.isCompletedExceptionally()); + assertTrue(a.isDone()); + assertFalse(b.isDone()); + second.writeInbound(RpcResponse.newBuilder().setRequestId("b").build()); + assertTrue(b.isDone()); + assertFalse(b.isCompletedExceptionally()); + assertEquals(0, shared.pendingRequestCount()); + } finally { + first.finishAndReleaseAll(); + second.finishAndReleaseAll(); + parent.finishAndReleaseAll(); + } + } + + @Test + void resetFailsOnlyItsStreamAndParentCloseStillFailsAll() { + NettyRpcClientHandler shared = new NettyRpcClientHandler(); + EmbeddedChannel parent = new EmbeddedChannel(shared); + EmbeddedChannel child = new EmbeddedChannel(new RpcStreamResponseHandler(shared, "a")); + try { + CompletableFuture a = new CompletableFuture<>(); + CompletableFuture b = new CompletableFuture<>(); + shared.addFuture("a", a); + shared.addFuture("b", b); + child.pipeline().fireUserEventTriggered(new DefaultHttp2ResetFrame(Http2Error.CANCEL)); + assertTrue(a.isCompletedExceptionally()); + assertFalse(b.isDone()); + parent.close(); + assertTrue(b.isCompletedExceptionally()); + assertEquals(0, shared.pendingRequestCount()); + } finally { + child.finishAndReleaseAll(); + parent.finishAndReleaseAll(); + } + } + + @Test + void mismatchedResponseCannotCompleteAnotherStream() { + NettyRpcClientHandler shared = new NettyRpcClientHandler(); + EmbeddedChannel channel = new EmbeddedChannel(new RpcStreamResponseHandler(shared, "a")); + try { + CompletableFuture a = new CompletableFuture<>(); + CompletableFuture b = new CompletableFuture<>(); + shared.addFuture("a", a); + shared.addFuture("b", b); + channel.writeInbound(RpcResponse.newBuilder().setRequestId("b").build()); + assertTrue(a.isCompletedExceptionally()); + assertFalse(b.isDone()); + } finally { + channel.finishAndReleaseAll(); + shared.failAll(new IllegalStateException("test cleanup")); + } + } +} diff --git a/rpc-transport-netty/src/test/java/com/xiaoyu/rpc/core/protocol/grpc/GrpcClientResponseHandlerTest.java b/rpc-transport-netty/src/test/java/com/xiaoyu/rpc/core/protocol/grpc/GrpcClientResponseHandlerTest.java index ca0808f..2d53df5 100644 --- a/rpc-transport-netty/src/test/java/com/xiaoyu/rpc/core/protocol/grpc/GrpcClientResponseHandlerTest.java +++ b/rpc-transport-netty/src/test/java/com/xiaoyu/rpc/core/protocol/grpc/GrpcClientResponseHandlerTest.java @@ -3,17 +3,20 @@ import com.google.protobuf.ByteString; import com.xiaoyu.rpc.common.vo.RpcResponse; import com.xiaoyu.rpc.core.client.NettyRpcClientHandler; +import com.xiaoyu.rpc.core.client.RpcStreamResponseHandler; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.CorruptedFrameException; import io.netty.handler.codec.http2.DefaultHttp2DataFrame; +import io.netty.handler.codec.http2.DefaultHttp2Headers; import io.netty.handler.codec.http2.DefaultHttp2HeadersFrame; -import io.netty.handler.codec.http2.Http2Headers; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; import static org.junit.jupiter.api.Assertions.*; @@ -21,57 +24,74 @@ class GrpcClientResponseHandlerTest { @Test - @DisplayName("应把 DataFrame 解码为 RpcResponse 并完成 future") + @DisplayName("完整消息加成功状态尾帧才完成 future") void testDecodeDataFrameToRpcResponse() throws Exception { NettyRpcClientHandler clientHandler = new NettyRpcClientHandler(); - EmbeddedChannel channel = new EmbeddedChannel( - new GrpcClientResponseHandler(clientHandler, "req-1"), - clientHandler); - + EmbeddedChannel channel = channel(clientHandler, "req-1"); CompletableFuture future = new CompletableFuture<>(); clientHandler.addFuture("req-1", future); - - RpcResponse response = RpcResponse.newBuilder() - .setRequestId("req-1") - .setMessage("Success") - .setData(ByteString.copyFromUtf8("ok")) - .build(); - - byte[] payload = response.toByteArray(); - ByteBuf buf = Unpooled.buffer(); - buf.writeByte(0); - buf.writeInt(payload.length); - buf.writeBytes(payload); - - channel.writeInbound(new DefaultHttp2DataFrame(buf, true)); - - assertTrue(future.isDone(), "Future should be completed"); - Object result = future.get(); - assertInstanceOf(RpcResponse.class, result); - assertEquals("req-1", ((RpcResponse) result).getRequestId()); - assertEquals("Success", ((RpcResponse) result).getMessage()); + RpcResponse response = response("req-1"); + try { + channel.writeInbound(new DefaultHttp2HeadersFrame(new DefaultHttp2Headers() + .status("200").set("content-type", "application/grpc"), false)); + channel.writeInbound(new DefaultHttp2DataFrame(framed(response), false)); + assertFalse(future.isDone(), "A DATA frame does not establish gRPC success"); + channel.writeInbound(new DefaultHttp2HeadersFrame(new DefaultHttp2Headers() + .set("grpc-status", "0"), true)); + assertEquals(response, future.get(1, TimeUnit.SECONDS)); + } finally { + channel.finishAndReleaseAll(); + } } @Test @DisplayName("收到 grpc-status 非 0 时应异常完成 future") void testFailFutureOnGrpcErrorStatus() { NettyRpcClientHandler clientHandler = new NettyRpcClientHandler(); - EmbeddedChannel channel = new EmbeddedChannel( - new GrpcClientResponseHandler(clientHandler, "req-2"), - clientHandler); - + EmbeddedChannel channel = channel(clientHandler, "req-2"); CompletableFuture future = new CompletableFuture<>(); clientHandler.addFuture("req-2", future); + try { + channel.writeInbound(new DefaultHttp2HeadersFrame(new DefaultHttp2Headers() + .status("200").set("grpc-status", "13").set("grpc-message", "internal"), true)); + assertTrue(future.isCompletedExceptionally()); + ExecutionException error = assertThrows(ExecutionException.class, + () -> future.get(1, TimeUnit.SECONDS)); + assertTrue(error.getCause().getMessage().contains("gRPC status=13")); + } finally { + channel.finishAndReleaseAll(); + } + } - Http2Headers trailers = new io.netty.handler.codec.http2.DefaultHttp2Headers() - .set("grpc-status", "13") - .set("grpc-message", "internal"); - channel.writeInbound(new DefaultHttp2HeadersFrame(trailers, true)); + @Test + @DisplayName("DATA 直接结束且缺少状态尾帧时必须失败") + void testMissingTrailersCannotBeSuccess() { + NettyRpcClientHandler clientHandler = new NettyRpcClientHandler(); + EmbeddedChannel channel = channel(clientHandler, "req-3"); + CompletableFuture future = new CompletableFuture<>(); + clientHandler.addFuture("req-3", future); + try { + channel.writeInbound(new DefaultHttp2DataFrame(framed(response("req-3")), true)); + ExecutionException error = assertThrows(ExecutionException.class, + () -> future.get(1, TimeUnit.SECONDS)); + assertInstanceOf(CorruptedFrameException.class, error.getCause()); + } finally { + channel.finishAndReleaseAll(); + } + } + + private static EmbeddedChannel channel(NettyRpcClientHandler handler, String requestId) { + return new EmbeddedChannel(new GrpcClientResponseHandler(handler, requestId), + new RpcStreamResponseHandler(handler, requestId)); + } + + private static RpcResponse response(String requestId) { + return RpcResponse.newBuilder().setRequestId(requestId).setMessage("Success") + .setData(ByteString.copyFromUtf8("ok")).build(); + } - assertTrue(future.isCompletedExceptionally(), "Future should be completed exceptionally"); - ExecutionException ex = assertThrows(ExecutionException.class, future::get); - String message = ex.getCause().getMessage(); - assertNotNull(message, "Error message should not be null"); - assertTrue(message.toLowerCase().contains("grpc"), "Error message should include grpc details"); + private static ByteBuf framed(RpcResponse response) { + byte[] payload = response.toByteArray(); + return Unpooled.buffer(payload.length + 5).writeByte(0).writeInt(payload.length).writeBytes(payload); } } diff --git a/rpc-transport-netty/src/test/java/com/xiaoyu/rpc/core/protocol/grpc/GrpcFramingTest.java b/rpc-transport-netty/src/test/java/com/xiaoyu/rpc/core/protocol/grpc/GrpcFramingTest.java new file mode 100644 index 0000000..e71a87b --- /dev/null +++ b/rpc-transport-netty/src/test/java/com/xiaoyu/rpc/core/protocol/grpc/GrpcFramingTest.java @@ -0,0 +1,113 @@ +package com.xiaoyu.rpc.core.protocol.grpc; + +import com.google.protobuf.ByteString; +import com.xiaoyu.rpc.common.vo.RpcRequest; +import com.xiaoyu.rpc.common.vo.RpcResponse; +import com.xiaoyu.rpc.core.client.NettyRpcClientHandler; +import com.xiaoyu.rpc.core.client.RpcStreamResponseHandler; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.buffer.UnpooledByteBufAllocator; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.CorruptedFrameException; +import io.netty.handler.codec.TooLongFrameException; +import io.netty.handler.codec.http2.DefaultHttp2DataFrame; +import io.netty.handler.codec.http2.DefaultHttp2Headers; +import io.netty.handler.codec.http2.DefaultHttp2HeadersFrame; +import org.junit.jupiter.api.Test; +import java.util.concurrent.CompletableFuture; +import static org.junit.jupiter.api.Assertions.*; + +class GrpcFramingTest { + @Test + void serverAccumulatesSplitPrefixAndLargeMessage() { + RpcRequest request = RpcRequest.newBuilder().setRequestId("large") + .addParameters(ByteString.copyFrom(new byte[256 * 1024])).build(); + ByteBuf wire = framed(request.toByteArray()); + EmbeddedChannel server = new EmbeddedChannel(new GrpcServerHandler(null)); + try { + server.writeInbound(new DefaultHttp2DataFrame(wire.readRetainedSlice(2), false)); + assertNull(server.readInbound()); + while (wire.readableBytes() > 8192) { + server.writeInbound(new DefaultHttp2DataFrame(wire.readRetainedSlice(8192), false)); + assertNull(server.readInbound()); + } + server.writeInbound(new DefaultHttp2DataFrame(wire.readRetainedSlice(wire.readableBytes()), true)); + assertEquals(request, server.readInbound()); + assertNull(server.readInbound()); + } finally { + wire.release(); + server.finishAndReleaseAll(); + } + } + + @Test + void clientAccumulatesLargeResponseAndWaitsForTrailers() { + NettyRpcClientHandler shared = new NettyRpcClientHandler(); + CompletableFuture future = new CompletableFuture<>(); + shared.addFuture("large", future); + RpcResponse response = RpcResponse.newBuilder().setRequestId("large") + .setData(ByteString.copyFrom(new byte[256 * 1024])).build(); + EmbeddedChannel channel = new EmbeddedChannel(new GrpcClientResponseHandler(shared, "large"), + new RpcStreamResponseHandler(shared, "large")); + ByteBuf wire = framed(response.toByteArray()); + try { + channel.writeInbound(new DefaultHttp2HeadersFrame(new DefaultHttp2Headers().status("200"), false)); + channel.writeInbound(new DefaultHttp2DataFrame(wire.readRetainedSlice(3), false)); + while (wire.isReadable()) { + channel.writeInbound(new DefaultHttp2DataFrame( + wire.readRetainedSlice(Math.min(8192, wire.readableBytes())), false)); + } + assertFalse(future.isDone(), "A body alone is not gRPC success"); + channel.writeInbound(new DefaultHttp2HeadersFrame(new DefaultHttp2Headers().set("grpc-status", "0"), true)); + assertEquals(response, future.join()); + channel.close(); + assertFalse(future.isCompletedExceptionally()); + } finally { + wire.release(); + channel.finishAndReleaseAll(); + } + } + + @Test + void errorTrailersDoNotBecomeSuccessfulResponse() { + NettyRpcClientHandler shared = new NettyRpcClientHandler(); + CompletableFuture future = new CompletableFuture<>(); + shared.addFuture("error", future); + EmbeddedChannel channel = new EmbeddedChannel(new GrpcClientResponseHandler(shared, "error"), + new RpcStreamResponseHandler(shared, "error")); + try { + channel.writeInbound(new DefaultHttp2DataFrame(framed(RpcResponse.newBuilder().setRequestId("error").build().toByteArray()), false)); + channel.writeInbound(new DefaultHttp2HeadersFrame(new DefaultHttp2Headers().set("grpc-status", "13"), true)); + assertTrue(future.isCompletedExceptionally()); + } finally { + channel.finishAndReleaseAll(); + } + } + + @Test + void boundsAndTruncatedMessagesAreRejected() { + try (GrpcMessageAccumulator accumulator = new GrpcMessageAccumulator(32)) { + ByteBuf header = Unpooled.buffer().writeByte(0).writeInt(33); + try { + assertThrows(TooLongFrameException.class, + () -> accumulator.append(UnpooledByteBufAllocator.DEFAULT, header)); + } finally { + header.release(); + } + } + try (GrpcMessageAccumulator accumulator = new GrpcMessageAccumulator(32)) { + ByteBuf truncated = Unpooled.buffer().writeByte(0).writeInt(8).writeByte(1); + try { + assertNull(accumulator.append(UnpooledByteBufAllocator.DEFAULT, truncated)); + assertThrows(CorruptedFrameException.class, accumulator::requireComplete); + } finally { + truncated.release(); + } + } + } + + private static ByteBuf framed(byte[] payload) { + return Unpooled.buffer(payload.length + 5).writeByte(0).writeInt(payload.length).writeBytes(payload); + } +}