From 4292c49f4e141e98f2b92483f67fb8768f6c921c Mon Sep 17 00:00:00 2001 From: yuwnloy Date: Mon, 20 Jul 2026 17:44:13 +0800 Subject: [PATCH] perf: add reproducible messaging benchmarks --- .github/workflows/ci.yml | 1 + .gitignore | 1 + BENCHMARKS.md | 146 +++++++++ README.md | 2 + README_zh.md | 2 + cmd/jimbench/main.go | 81 +++++ docs/benchmarks/results/README.md | 17 + .../results/local-smoke-20260720-group.json | 1 + .../results/local-smoke-20260720-private.json | 1 + scripts/run-benchmark.sh | 108 ++++++ simulator/benchmark/api.go | 122 +++++++ simulator/benchmark/benchmark_test.go | 75 +++++ simulator/benchmark/config.go | 110 +++++++ simulator/benchmark/report.go | 168 ++++++++++ simulator/benchmark/runner.go | 309 ++++++++++++++++++ simulator/benchmark/stats.go | 121 +++++++ simulator/wsclients/wsimclient.go | 6 +- 17 files changed, 1270 insertions(+), 1 deletion(-) create mode 100644 BENCHMARKS.md create mode 100644 cmd/jimbench/main.go create mode 100644 docs/benchmarks/results/README.md create mode 100644 docs/benchmarks/results/local-smoke-20260720-group.json create mode 100644 docs/benchmarks/results/local-smoke-20260720-private.json create mode 100755 scripts/run-benchmark.sh create mode 100644 simulator/benchmark/api.go create mode 100644 simulator/benchmark/benchmark_test.go create mode 100644 simulator/benchmark/config.go create mode 100644 simulator/benchmark/report.go create mode 100644 simulator/benchmark/runner.go create mode 100644 simulator/benchmark/stats.go diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 68f3ba2f..636bf929 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -58,6 +58,7 @@ jobs: - name: Run self-contained unit tests run: | + go test ./simulator/benchmark go test ./services/commonservices/... go test ./commons/metrics/... go test ./services/pushmanager/services/httputil diff --git a/.gitignore b/.gitignore index 067b16d6..4b8ed691 100644 --- a/.gitignore +++ b/.gitignore @@ -11,3 +11,4 @@ launcher/uploads/ CLAUDE.md .claude/ .promotion-metrics/ +/benchmark-results/ diff --git a/BENCHMARKS.md b/BENCHMARKS.md new file mode 100644 index 00000000..19a16935 --- /dev/null +++ b/BENCHMARKS.md @@ -0,0 +1,146 @@ +# Reproducible benchmarks + +JuggleIM's benchmark harness measures connection establishment separately from steady-state message delivery. It favors reproducibility and explicit limitations over headline numbers. + +The harness runs against an isolated JuggleIM deployment, creates synthetic users through the signed server API, connects real WebSocket clients using the production protocol, and records a machine-readable JSON report. It must never target a public or production deployment. + +## Workloads + +### Private chat + +Every connected client is a sender. Clients form a ring: client `N` sends to client `N+1`, and the final client sends to the first. The configured rate is the aggregate publish rate across the ring. + +### Group chat + +All clients join one synthetic group. A configurable subset of clients publishes to the group at the aggregate target rate. The report distinguishes accepted publishes from observed recipient deliveries, making fan-out visible without treating it as sender throughput. + +Both workloads use stored, counted text messages by default. Each JSON payload contains a run ID, phase, sequence number, nanosecond send timestamp, and padding to reach the configured minimum size. + +## Metrics + +The JSON report contains: + +- connection attempts, successes, failures, error codes, and connection latency; +- publish-to-ACK throughput and P50/P95/P99 latency; +- publish-to-recipient-callback throughput and P50/P95/P99 latency; +- workload settings, including clients, senders, message rate, payload size, warm-up, duration, and delivery grace; +- server commit, dirty-tree state, OS, architecture, Go version, CPU model, logical CPU count, memory, database label, and deployment label; +- explicit limitations that must accompany any published result. + +Warm-up traffic is excluded from all message measurements. Connection establishment is completed before warm-up and is never mixed into steady-state latency. + +A small **[verified local smoke run](./docs/benchmarks/results/README.md)** and its machine-readable private-chat and group-chat reports are checked into the repository. They validate the full harness; they are explicitly not production capacity claims. + +## One-command local run + +Requirements: + +- Docker with Compose; +- Go version from [`go.mod`](./go.mod); +- `curl`, `jq`, `openssl`, and `shasum`. + +Run both baseline workloads: + +```bash +scripts/run-benchmark.sh all +``` + +The script builds and starts the local Compose stack, creates an isolated application, runs private and group workloads, and writes JSON files under `benchmark-results/`. It leaves the stack running so logs, profiles, and database state can be inspected. Stop it without deleting the database volume: + +```bash +docker compose down +``` + +The default baseline is intentionally modest: + +| Setting | Default | +| --- | ---: | +| Connected clients | 50 | +| Aggregate publish rate | 200 messages/second | +| Group senders | 2 | +| Warm-up | 10 seconds | +| Measurement | 30 seconds | +| Delivery grace | 3 seconds | +| Minimum payload | 256 bytes | + +Override settings with environment variables: + +```bash +BENCH_CLIENTS=500 \ +BENCH_RATE=1000 \ +BENCH_WARMUP=30s \ +BENCH_DURATION=2m \ +BENCH_PAYLOAD_BYTES=1024 \ +scripts/run-benchmark.sh all +``` + +Useful variables are `BENCH_CLIENTS`, `BENCH_RATE`, `BENCH_GROUP_SENDERS`, `BENCH_WARMUP`, `BENCH_DURATION`, `BENCH_DELIVERY_GRACE`, `BENCH_PAYLOAD_BYTES`, and `BENCH_OUTPUT_DIR`. + +To avoid conflicting with local development services, the runner uses host ports `19001` (API), `19002` (navigator), `19003` (WebSocket), `18090` (admin), `16060` (pprof), and `13306` (MySQL). Override them with the matching `*_HOST_PORT` variables. Containers still use JuggleIM's standard ports over the internal Compose network. + +## Running the CLI against an isolated environment + +The CLI reads the application secret only from `JIM_BENCH_APP_SECRET`, keeping it out of command arguments and result files: + +```bash +export JIM_BENCH_APP_KEY='benchmark-app' +export JIM_BENCH_APP_SECRET='replace-with-isolated-app-secret' + +go run ./cmd/jimbench \ + --scenario private \ + --clients 100 \ + --rate 500 \ + --warmup 30s \ + --duration 2m \ + --payload-bytes 256 +``` + +Loopback targets are enforced by default. `--allow-non-loopback` exists only for a dedicated, isolated benchmark environment. Never use it with shared staging, public, or production infrastructure. + +## Reproduction checklist + +Before publishing a result: + +1. Use a clean, named server commit and include the generated JSON files. +2. Run the load generator on a separate host for serious capacity measurements. +3. Record server and database hardware separately in `--environment` and `--database` labels. +4. State container CPU/memory limits, MySQL configuration, network topology, TLS status, and storage type. +5. Run at least three trials after warm-up and publish every trial, not only the best one. +6. Report errors and saturation behavior alongside throughput. +7. Keep private-chat and group fan-out results separate. +8. Use server-side CPU, memory, network, disk, database, and pprof data to explain bottlenecks. + +Example for a dedicated environment: + +```bash +go run ./cmd/jimbench \ + --scenario group \ + --ws-url wss://benchmark-im.example.internal \ + --api-url https://benchmark-api.example.internal/apigateway \ + --allow-non-loopback \ + --clients 1000 \ + --group-senders 10 \ + --rate 1000 \ + --warmup 1m \ + --duration 5m \ + --environment 'server: 8 vCPU/16 GiB, load generator: 8 vCPU/16 GiB, 10 GbE, TLS enabled' \ + --database 'MySQL 8.0.42, 8 vCPU/32 GiB, local NVMe, configuration linked in report notes' +``` + +## Interpreting results + +- ACK throughput is the rate at which JuggleIM accepts publishes; it is not the same as recipient deliveries. +- Delivery throughput counts callbacks observed by connected clients. Group fan-out can therefore be much larger than publish throughput. +- ACK latency includes client serialization, WebSocket transport, server processing, and the ACK return path. +- Delivery latency includes the full sender-to-recipient path but assumes synchronized timing because the load generator hosts all benchmark clients. +- A local Compose result is a development baseline, not a production capacity claim. + +Do not compare JuggleIM with another project unless both systems use equivalent delivery semantics, persistence, acknowledgements, payloads, fan-out, hardware, databases, and test duration. + +## Development checks + +```bash +go test ./simulator/benchmark ./cmd/jimbench +go test -race ./simulator/benchmark +go vet ./simulator/benchmark ./cmd/jimbench +``` diff --git a/README.md b/README.md index 40981a0c..9ce2e612 100644 --- a/README.md +++ b/README.md @@ -100,6 +100,8 @@ JuggleIM runs as a modular Go service: HTTP and WebSocket gateways route request Read the **[architecture guide](./docs/architecture.md)** for component boundaries, data ownership, private and group message flows, security boundaries, and deployment constraints. [简体中文版](./docs/architecture_zh.md) +Evaluate performance with the **[reproducible benchmark harness](./BENCHMARKS.md)**, which reports connection setup separately from private-chat and group-chat ACK and delivery throughput, including P50/P95/P99 latency and machine-readable results. + ## 🚀 Quick Start with Docker Run a complete local JuggleIM stack with MySQL and the admin console: diff --git a/README_zh.md b/README_zh.md index 1064bfaa..6a9e823f 100644 --- a/README_zh.md +++ b/README_zh.md @@ -102,6 +102,8 @@ JuggleIM 以模块化 Go 服务运行：HTTP 和 WebSocket 网关通过内部 Ac 完整的组件边界、数据职责、私聊与群聊链路、安全边界和部署约束，请阅读 **[架构文档](./docs/architecture_zh.md)**。[English version](./docs/architecture.md) +可通过 **[可复现性能测试工具](./BENCHMARKS.md)** 分别评估连接建立、私聊和群聊的 ACK 与端到端投递吞吐，并生成包含 P50/P95/P99 延迟及运行环境的机器可读结果。 + ## 🚀 使用 Docker 快速开始 通过 Docker Compose 一次启动 MySQL、JuggleIM 服务和管理后台： diff --git a/cmd/jimbench/main.go b/cmd/jimbench/main.go new file mode 100644 index 00000000..8085ede4 --- /dev/null +++ b/cmd/jimbench/main.go @@ -0,0 +1,81 @@ +package main + +import ( + "context" + "flag" + "fmt" + "os" + "os/signal" + "syscall" + + "im-server/simulator/benchmark" +) + +func main() { + config := benchmark.DefaultConfig() + var scenario string + flag.StringVar(&scenario, "scenario", string(config.Scenario), "workload scenario: private or group") + flag.StringVar(&config.WSURL, "ws-url", config.WSURL, "IM WebSocket base URL") + flag.StringVar(&config.APIURL, "api-url", config.APIURL, "signed server API base URL") + flag.StringVar(&config.AppKey, "app-key", os.Getenv("JIM_BENCH_APP_KEY"), "benchmark application key") + flag.IntVar(&config.Clients, "clients", config.Clients, "number of connected clients") + flag.IntVar(&config.GroupSenders, "group-senders", config.GroupSenders, "number of group-message senders") + flag.IntVar(&config.Rate, "rate", config.Rate, "aggregate target messages per second") + flag.DurationVar(&config.Warmup, "warmup", config.Warmup, "warm-up duration excluded from measurements") + flag.DurationVar(&config.Duration, "duration", config.Duration, "measurement duration") + flag.DurationVar(&config.DeliveryGrace, "delivery-grace", config.DeliveryGrace, "time to collect late deliveries after sending stops") + flag.IntVar(&config.PayloadBytes, "payload-bytes", config.PayloadBytes, "minimum JSON message payload size") + flag.IntVar(&config.SetupConcurrency, "setup-concurrency", config.SetupConcurrency, "concurrent user registration requests") + flag.IntVar(&config.ConnectConcurrency, "connect-concurrency", config.ConnectConcurrency, "concurrent WebSocket connection attempts") + flag.BoolVar(&config.StoreMessages, "store-messages", config.StoreMessages, "set the stored-message flag") + flag.BoolVar(&config.CountMessages, "count-messages", config.CountMessages, "set the counted-message flag") + flag.BoolVar(&config.AllowNonLoopback, "allow-non-loopback", false, "allow an isolated non-loopback target (never use against production)") + flag.StringVar(&config.OutputPath, "output", "", "JSON result path (default: benchmark-results/-.json)") + flag.StringVar(&config.EnvironmentLabel, "environment", envOr("JIM_BENCH_ENVIRONMENT", config.EnvironmentLabel), "human-readable deployment/configuration label") + flag.StringVar(&config.DatabaseLabel, "database", envOr("JIM_BENCH_DATABASE", config.DatabaseLabel), "database engine and configuration label") + flag.Parse() + + config.Scenario = benchmark.Scenario(scenario) + config.AppSecret = os.Getenv("JIM_BENCH_APP_SECRET") + config.ServerCommit = os.Getenv("JIM_BENCH_SERVER_COMMIT") + runner, err := benchmark.NewRunner(config) + if err != nil { + exitf("invalid benchmark configuration: %v", err) + } + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + fmt.Printf("Preparing %s benchmark: clients=%d rate=%d/s warmup=%s duration=%s payload>=%d bytes\n", + config.Scenario, config.Clients, config.Rate, config.Warmup, config.Duration, config.PayloadBytes) + report, err := runner.Run(ctx) + if err != nil { + exitf("benchmark failed: %v", err) + } + path, err := benchmark.WriteReport(config.OutputPath, report) + if err != nil { + exitf("save benchmark report: %v", err) + } + + fmt.Printf("Benchmark complete.\n") + fmt.Printf(" Connections: %d/%d, P95 %.2f ms\n", report.Connections.Succeeded, report.Connections.Attempted, report.Connections.Latency.P95MS) + fmt.Printf(" ACKs: %d succeeded, %d failed, %.2f/s, P95 %.2f ms, P99 %.2f ms\n", + report.MessageAcknowledgements.Succeeded, report.MessageAcknowledgements.Failed, + report.MessageAcknowledgements.ThroughputPerSec, report.MessageAcknowledgements.Latency.P95MS, + report.MessageAcknowledgements.Latency.P99MS) + fmt.Printf(" Deliveries: %d observed, %.2f/s, P95 %.2f ms, P99 %.2f ms\n", + report.Deliveries.Succeeded, report.Deliveries.ThroughputPerSec, + report.Deliveries.Latency.P95MS, report.Deliveries.Latency.P99MS) + fmt.Printf(" Result: %s\n", path) +} + +func envOr(name, fallback string) string { + if value := os.Getenv(name); value != "" { + return value + } + return fallback +} + +func exitf(format string, args ...any) { + fmt.Fprintf(os.Stderr, format+"\n", args...) + os.Exit(1) +} diff --git a/docs/benchmarks/results/README.md b/docs/benchmarks/results/README.md new file mode 100644 index 00000000..60c96dbc --- /dev/null +++ b/docs/benchmarks/results/README.md @@ -0,0 +1,17 @@ +# Verified local smoke run + +These files prove that the complete benchmark path works against the repository's Docker Compose stack. They are development smoke results, not production capacity claims: the server and load generator shared one Apple M2 host, the measurement lasted only 10 seconds, and the working tree contained the new benchmark harness. + +Run date: 2026-07-20 UTC. Server base commit: `b1f4ff52c3e5e6da2184f6b8ccb441fe04f13e21`. Database: the repository's default MySQL 8.0 Compose service. Workload: 20 WebSocket clients, stored and counted 256-byte messages, 3-second warm-up, 10-second measurement, 2-second delivery grace, target 50 publishes/second. + +| Scenario | Connections | Successful ACKs | ACK rate | ACK P95 / P99 | Observed deliveries | Delivery rate | Delivery P95 / P99 | +| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | +| Private ring | 20/20 | 499/499 | 49.9/s | 2.37 / 17.09 ms | 499 | 49.9/s | 8.47 / 36.75 ms | +| 20-member group, 2 senders | 20/20 | 499/499 | 49.9/s | 4.54 / 30.19 ms | 9,492 | 949.2/s | 35.29 / 115.88 ms | + +Machine-readable reports: + +- [`local-smoke-20260720-private.json`](./local-smoke-20260720-private.json) +- [`local-smoke-20260720-group.json`](./local-smoke-20260720-group.json) + +For publishable capacity results, follow the clean-commit, separate-load-generator, multi-trial, resource-monitoring, and full-environment checklist in [`BENCHMARKS.md`](../../../BENCHMARKS.md). diff --git a/docs/benchmarks/results/local-smoke-20260720-group.json b/docs/benchmarks/results/local-smoke-20260720-group.json new file mode 100644 index 00000000..9eed3cde --- /dev/null +++ b/docs/benchmarks/results/local-smoke-20260720-group.json @@ -0,0 +1 @@ +{"schema_version":1,"run_id":"1784536641-70553608","generated_at":"2026-07-20T08:37:36.993036Z","started_at":"2026-07-20T08:37:21.734704Z","measurement_started_at":"2026-07-20T08:37:24.860001Z","measurement_ended_at":"2026-07-20T08:37:34.860001Z","workload":{"scenario":"group","clients":20,"group_senders":2,"target_messages_per_second":50,"warmup":"3s","duration":"10s","delivery_grace":"2s","payload_bytes":256,"store_messages":true,"count_messages":true,"websocket_url":"ws://127.0.0.1:19003","api_url":"http://127.0.0.1:19001/apigateway","environment_label":"Local Docker Compose; server and load generator share the host; no explicit container CPU or memory limits; compose_sha256=b85237996448ead20f1c45e20211f254918268a152247520ba7b73881ae13cda","database_label":"MySQL 8.0 Docker image; utf8mb4_0900_ai_ci; default Compose configuration"},"environment":{"server_commit":"b1f4ff52c3e5e6da2184f6b8ccb441fe04f13e21","working_tree_dirty":true,"os":"darwin","architecture":"amd64","go_version":"go1.25.12","cpu_model":"Apple M2","logical_cpus":8,"total_memory_bytes":17179869184},"connections":{"attempted":20,"succeeded":20,"failed":0,"throughput_per_second":0,"latency":{"count":20,"min_ms":17.579166,"p50_ms":19.765167,"p95_ms":26.728375,"p99_ms":27.584125,"max_ms":27.584125,"mean_ms":21.951995}},"message_acknowledgements":{"attempted":499,"succeeded":499,"failed":0,"throughput_per_second":49.9,"latency":{"count":499,"min_ms":0.361709,"p50_ms":0.892416,"p95_ms":4.544583,"p99_ms":30.185166,"max_ms":70.9605,"mean_ms":2.129866}},"deliveries":{"attempted":9492,"succeeded":9492,"failed":0,"throughput_per_second":949.2,"latency":{"count":9492,"min_ms":3.394,"p50_ms":7.319,"p95_ms":35.286,"p99_ms":115.884,"max_ms":158.39,"mean_ms":12.0033}},"limitations":["ACK latency uses the load generator's monotonic clock; delivery latency uses wall-clock timestamps within the same load-generator process.","Acknowledgement latency measures client publish to server ACK; delivery latency measures client publish to recipient callback.","The load generator and local Docker stack may share hardware unless the environment label says otherwise.","Results from different hardware or workload settings are not directly comparable."]} diff --git a/docs/benchmarks/results/local-smoke-20260720-private.json b/docs/benchmarks/results/local-smoke-20260720-private.json new file mode 100644 index 00000000..62e35765 --- /dev/null +++ b/docs/benchmarks/results/local-smoke-20260720-private.json @@ -0,0 +1 @@ +{"schema_version":1,"run_id":"1784536624-998a7415","generated_at":"2026-07-20T08:37:19.728472Z","started_at":"2026-07-20T08:37:04.251776Z","measurement_started_at":"2026-07-20T08:37:07.392392Z","measurement_ended_at":"2026-07-20T08:37:17.392392Z","workload":{"scenario":"private","clients":20,"target_messages_per_second":50,"warmup":"3s","duration":"10s","delivery_grace":"2s","payload_bytes":256,"store_messages":true,"count_messages":true,"websocket_url":"ws://127.0.0.1:19003","api_url":"http://127.0.0.1:19001/apigateway","environment_label":"Local Docker Compose; server and load generator share the host; no explicit container CPU or memory limits; compose_sha256=b85237996448ead20f1c45e20211f254918268a152247520ba7b73881ae13cda","database_label":"MySQL 8.0 Docker image; utf8mb4_0900_ai_ci; default Compose configuration"},"environment":{"server_commit":"b1f4ff52c3e5e6da2184f6b8ccb441fe04f13e21","working_tree_dirty":true,"os":"darwin","architecture":"amd64","go_version":"go1.25.12","cpu_model":"Apple M2","logical_cpus":8,"total_memory_bytes":17179869184},"connections":{"attempted":20,"succeeded":20,"failed":0,"throughput_per_second":0,"latency":{"count":20,"min_ms":24.325417,"p50_ms":30.048125,"p95_ms":52.907875,"p99_ms":52.968208,"max_ms":52.968208,"mean_ms":32.666724}},"message_acknowledgements":{"attempted":499,"succeeded":499,"failed":0,"throughput_per_second":49.9,"latency":{"count":499,"min_ms":0.591375,"p50_ms":1.1375,"p95_ms":2.36675,"p99_ms":17.09125,"max_ms":81.083375,"mean_ms":1.742864}},"deliveries":{"attempted":499,"succeeded":499,"failed":0,"throughput_per_second":49.9,"latency":{"count":499,"min_ms":2.13,"p50_ms":5.008,"p95_ms":8.47,"p99_ms":36.751,"max_ms":128.973,"mean_ms":6.127274}},"limitations":["ACK latency uses the load generator's monotonic clock; delivery latency uses wall-clock timestamps within the same load-generator process.","Acknowledgement latency measures client publish to server ACK; delivery latency measures client publish to recipient callback.","The load generator and local Docker stack may share hardware unless the environment label says otherwise.","Results from different hardware or workload settings are not directly comparable."]} diff --git a/scripts/run-benchmark.sh b/scripts/run-benchmark.sh new file mode 100755 index 00000000..0a78475b --- /dev/null +++ b/scripts/run-benchmark.sh @@ -0,0 +1,108 @@ +#!/usr/bin/env bash +set -euo pipefail + +for command_name in curl docker go jq openssl shasum; do + if ! command -v "${command_name}" >/dev/null 2>&1; then + echo "Required command not found: ${command_name}" >&2 + exit 1 + fi +done + +scenario="${1:-all}" +if [[ "${scenario}" != "all" && "${scenario}" != "private" && "${scenario}" != "group" ]]; then + echo "Usage: $0 [all|private|group]" >&2 + exit 1 +fi + +clients="${BENCH_CLIENTS:-50}" +rate="${BENCH_RATE:-200}" +warmup="${BENCH_WARMUP:-10s}" +duration="${BENCH_DURATION:-30s}" +delivery_grace="${BENCH_DELIVERY_GRACE:-3s}" +payload_bytes="${BENCH_PAYLOAD_BYTES:-256}" +group_senders="${BENCH_GROUP_SENDERS:-2}" +output_dir="${BENCH_OUTPUT_DIR:-benchmark-results}" +export MYSQL_HOST_PORT="${MYSQL_HOST_PORT:-13306}" +export API_HOST_PORT="${API_HOST_PORT:-19001}" +export NAV_HOST_PORT="${NAV_HOST_PORT:-19002}" +export WS_HOST_PORT="${WS_HOST_PORT:-19003}" +export ADMIN_HOST_PORT="${ADMIN_HOST_PORT:-18090}" +export PPROF_HOST_PORT="${PPROF_HOST_PORT:-16060}" +admin_base_url="${ADMIN_BASE_URL:-http://127.0.0.1:${ADMIN_HOST_PORT}/admingateway}" +api_base_url="${API_BASE_URL:-http://127.0.0.1:${API_HOST_PORT}/apigateway}" +ws_url="${WS_URL:-ws://127.0.0.1:${WS_HOST_PORT}}" +admin_account="${ADMIN_ACCOUNT:-admin}" +admin_password="${ADMIN_PASSWORD:-123456}" +run_timestamp="$(date -u +%Y%m%dT%H%M%SZ)" +app_key="bench$(date +%s)" + +echo "Building and starting the local Docker Compose stack..." +docker compose up --detach --build + +login_response="" +for _ in $(seq 1 60); do + if login_response="$(curl --fail --silent --show-error \ + --request POST "${admin_base_url}/login" \ + --header 'Content-Type: application/json' \ + --data "$(jq -nc --arg account "${admin_account}" --arg password "${admin_password}" \ + '{account: $account, password: $password}')" 2>/dev/null)"; then + if [[ "$(jq -r '.code // -1' <<<"${login_response}")" == "0" ]]; then + break + fi + fi + sleep 2 +done + +if [[ -z "${login_response}" || "$(jq -r '.code // -1' <<<"${login_response}")" != "0" ]]; then + echo "Local stack did not become ready: ${login_response:-no response}" >&2 + exit 1 +fi + +admin_token="$(jq -er '.data.authorization' <<<"${login_response}")" +app_response="$(curl --fail --silent --show-error \ + --request POST "${admin_base_url}/apps/create" \ + --header 'Content-Type: application/json' \ + --header "Authorization: ${admin_token}" \ + --data "$(jq -nc --arg app_key "${app_key}" \ + '{app_key: $app_key, app_name: "Reproducible Benchmark"}')")" +if [[ "$(jq -r '.code // -1' <<<"${app_response}")" != "0" ]]; then + echo "Create benchmark application failed: ${app_response}" >&2 + exit 1 +fi + +export JIM_BENCH_APP_KEY="${app_key}" +export JIM_BENCH_APP_SECRET +JIM_BENCH_APP_SECRET="$(jq -er '.data.app_secret' <<<"${app_response}")" +sleep 2 +export JIM_BENCH_SERVER_COMMIT="$(git rev-parse HEAD)" +compose_sha="$(shasum -a 256 docker-compose.yml | awk '{print $1}')" +export JIM_BENCH_ENVIRONMENT="Local Docker Compose; server and load generator share the host; no explicit container CPU or memory limits; compose_sha256=${compose_sha}" +export JIM_BENCH_DATABASE="MySQL 8.0 Docker image; utf8mb4_0900_ai_ci; default Compose configuration" +mkdir -p "${output_dir}" + +run_scenario() { + local selected_scenario="$1" + echo "Running ${selected_scenario} workload..." + go run ./cmd/jimbench \ + --scenario "${selected_scenario}" \ + --ws-url "${ws_url}" \ + --api-url "${api_base_url}" \ + --clients "${clients}" \ + --group-senders "${group_senders}" \ + --rate "${rate}" \ + --warmup "${warmup}" \ + --duration "${duration}" \ + --delivery-grace "${delivery_grace}" \ + --payload-bytes "${payload_bytes}" \ + --output "${output_dir}/${run_timestamp}-${selected_scenario}.json" +} + +if [[ "${scenario}" == "all" || "${scenario}" == "private" ]]; then + run_scenario private +fi +if [[ "${scenario}" == "all" || "${scenario}" == "group" ]]; then + run_scenario group +fi + +echo "Benchmark results are in ${output_dir}/." +echo "The local stack remains running for inspection. Stop it with: docker compose down" diff --git a/simulator/benchmark/api.go b/simulator/benchmark/api.go new file mode 100644 index 00000000..109eea2e --- /dev/null +++ b/simulator/benchmark/api.go @@ -0,0 +1,122 @@ +package benchmark + +import ( + "bytes" + "context" + "crypto/rand" + "crypto/sha1" + "encoding/hex" + "encoding/json" + "fmt" + "net/http" + "strconv" + "strings" + "time" +) + +type apiClient struct { + baseURL string + appKey string + appSecret string + http *http.Client +} + +type apiResponse struct { + Code int `json:"code"` + Msg string `json:"msg"` + Data json.RawMessage `json:"data"` +} + +type registeredUser struct { + ID string + Token string +} + +func newAPIClient(config Config) *apiClient { + return &apiClient{ + baseURL: strings.TrimRight(config.APIURL, "/"), + appKey: config.AppKey, + appSecret: config.AppSecret, + http: &http.Client{Timeout: 15 * time.Second}, + } +} + +func (c *apiClient) registerUser(ctx context.Context, userID string) (registeredUser, error) { + request := struct { + UserID string `json:"user_id"` + Nickname string `json:"nickname"` + }{UserID: userID, Nickname: userID} + var response struct { + UserID string `json:"user_id"` + Token string `json:"token"` + } + if err := c.post(ctx, "/users/register", request, &response); err != nil { + return registeredUser{}, err + } + if response.Token == "" { + return registeredUser{}, fmt.Errorf("register user %s returned an empty token", userID) + } + return registeredUser{ID: response.UserID, Token: response.Token}, nil +} + +func (c *apiClient) createGroup(ctx context.Context, groupID string, memberIDs []string) error { + request := struct { + GroupID string `json:"group_id"` + GroupName string `json:"group_name"` + MemberIDs []string `json:"member_ids"` + }{GroupID: groupID, GroupName: "JuggleIM benchmark group", MemberIDs: memberIDs} + return c.post(ctx, "/groups/add", request, nil) +} + +func (c *apiClient) post(ctx context.Context, endpoint string, payload, output any) error { + body, err := json.Marshal(payload) + if err != nil { + return fmt.Errorf("encode %s request: %w", endpoint, err) + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+endpoint, bytes.NewReader(body)) + if err != nil { + return fmt.Errorf("create %s request: %w", endpoint, err) + } + nonce, err := randomNonce() + if err != nil { + return fmt.Errorf("create request nonce: %w", err) + } + timestamp := strconv.FormatInt(time.Now().UnixMilli(), 10) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("appkey", c.appKey) + req.Header.Set("nonce", nonce) + req.Header.Set("timestamp", timestamp) + req.Header.Set("signature", signature(c.appSecret, nonce, timestamp)) + + resp, err := c.http.Do(req) + if err != nil { + return fmt.Errorf("call %s: %w", endpoint, err) + } + defer resp.Body.Close() + var envelope apiResponse + if err := json.NewDecoder(resp.Body).Decode(&envelope); err != nil { + return fmt.Errorf("decode %s response (HTTP %d): %w", endpoint, resp.StatusCode, err) + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 || envelope.Code != 0 { + return fmt.Errorf("%s failed: HTTP %d, code %d, message %q", endpoint, resp.StatusCode, envelope.Code, envelope.Msg) + } + if output != nil && len(envelope.Data) > 0 && string(envelope.Data) != "null" { + if err := json.Unmarshal(envelope.Data, output); err != nil { + return fmt.Errorf("decode %s data: %w", endpoint, err) + } + } + return nil +} + +func signature(secret, nonce, timestamp string) string { + sum := sha1.Sum([]byte(secret + nonce + timestamp)) + return hex.EncodeToString(sum[:]) +} + +func randomNonce() (string, error) { + buffer := make([]byte, 12) + if _, err := rand.Read(buffer); err != nil { + return "", err + } + return hex.EncodeToString(buffer), nil +} diff --git a/simulator/benchmark/benchmark_test.go b/simulator/benchmark/benchmark_test.go new file mode 100644 index 00000000..30b9fd2c --- /dev/null +++ b/simulator/benchmark/benchmark_test.go @@ -0,0 +1,75 @@ +package benchmark + +import ( + "encoding/json" + "strings" + "testing" + "time" +) + +func TestSummarizeLatenciesUsesNearestRank(t *testing.T) { + values := make([]time.Duration, 100) + for index := range values { + values[index] = time.Duration(index+1) * time.Millisecond + } + summary := summarizeLatencies(values) + if summary.Count != 100 || summary.P50MS != 50 || summary.P95MS != 95 || summary.P99MS != 99 { + t.Fatalf("unexpected latency summary: %+v", summary) + } + if summary.MinMS != 1 || summary.MaxMS != 100 || summary.MeanMS != 50.5 { + t.Fatalf("unexpected latency bounds or mean: %+v", summary) + } +} + +func TestMetricRecorderSeparatesFailures(t *testing.T) { + recorder := newMetricRecorder() + recorder.recordSuccess(10 * time.Millisecond) + recorder.recordFailure("timeout", 20*time.Millisecond) + snapshot := recorder.snapshot(time.Second) + if snapshot.Attempted != 2 || snapshot.Succeeded != 1 || snapshot.Failed != 1 { + t.Fatalf("unexpected counters: %+v", snapshot) + } + if snapshot.Errors["timeout"] != 1 || snapshot.ThroughputPerSec != 1 { + t.Fatalf("unexpected errors or throughput: %+v", snapshot) + } +} + +func TestConfigRejectsRemoteTargetsByDefault(t *testing.T) { + config := DefaultConfig() + config.AppKey = "test" + config.AppSecret = "secret" + config.WSURL = "wss://example.com" + if err := config.Validate(); err == nil || !strings.Contains(err.Error(), "not loopback") { + t.Fatalf("expected non-loopback validation error, got %v", err) + } + config.AllowNonLoopback = true + if err := config.Validate(); err != nil { + t.Fatalf("allow non-loopback should permit isolated remote targets: %v", err) + } +} + +func TestMakePayloadHasRequestedMinimumSizeAndRoundTrips(t *testing.T) { + sentAt := time.Unix(123, 456) + data, err := makePayload("run", "measure", 42, sentAt, 512) + if err != nil { + t.Fatal(err) + } + if len(data) < 512 { + t.Fatalf("payload length %d is less than requested 512", len(data)) + } + var payload benchmarkPayload + if err := json.Unmarshal(data, &payload); err != nil { + t.Fatal(err) + } + if payload.BenchmarkID != "run" || payload.Phase != "measure" || payload.Sequence != 42 || payload.SentAtNS != sentAt.UnixNano() { + t.Fatalf("unexpected payload: %+v", payload) + } +} + +func TestSignatureMatchesServerContract(t *testing.T) { + got := signature("appsecret", "nonce", "1672568121910") + const want = "2e639ae3600a48b6c595495dcf61fb88b76f485b" + if got != want { + t.Fatalf("signature mismatch: got %s, want %s", got, want) + } +} diff --git a/simulator/benchmark/config.go b/simulator/benchmark/config.go new file mode 100644 index 00000000..50f5b92e --- /dev/null +++ b/simulator/benchmark/config.go @@ -0,0 +1,110 @@ +package benchmark + +import ( + "fmt" + "net" + "net/url" + "strings" + "time" +) + +type Scenario string + +const ( + ScenarioPrivate Scenario = "private" + ScenarioGroup Scenario = "group" +) + +type Config struct { + Scenario Scenario + WSURL string + APIURL string + AppKey string + AppSecret string + Clients int + GroupSenders int + Rate int + Warmup time.Duration + Duration time.Duration + DeliveryGrace time.Duration + PayloadBytes int + SetupConcurrency int + ConnectConcurrency int + StoreMessages bool + CountMessages bool + AllowNonLoopback bool + OutputPath string + EnvironmentLabel string + DatabaseLabel string + ServerCommit string +} + +func DefaultConfig() Config { + return Config{ + Scenario: ScenarioPrivate, + WSURL: "ws://127.0.0.1:9003", + APIURL: "http://127.0.0.1:9001/apigateway", + Clients: 50, + GroupSenders: 2, + Rate: 200, + Warmup: 10 * time.Second, + Duration: 30 * time.Second, + DeliveryGrace: 3 * time.Second, + PayloadBytes: 256, + SetupConcurrency: 10, + ConnectConcurrency: 20, + StoreMessages: true, + CountMessages: true, + EnvironmentLabel: "unspecified", + DatabaseLabel: "unspecified", + } +} + +func (c Config) Validate() error { + if c.Scenario != ScenarioPrivate && c.Scenario != ScenarioGroup { + return fmt.Errorf("scenario must be %q or %q", ScenarioPrivate, ScenarioGroup) + } + if c.AppKey == "" { + return fmt.Errorf("app key is required") + } + if c.AppSecret == "" { + return fmt.Errorf("app secret is required; set JIM_BENCH_APP_SECRET") + } + if c.Clients < 2 { + return fmt.Errorf("clients must be at least 2") + } + if c.GroupSenders < 1 || c.GroupSenders > c.Clients { + return fmt.Errorf("group senders must be between 1 and clients") + } + if c.Rate < 1 { + return fmt.Errorf("rate must be at least 1 message per second") + } + if c.Warmup < 0 || c.Duration <= 0 || c.DeliveryGrace < 0 { + return fmt.Errorf("warmup and delivery grace cannot be negative, and duration must be positive") + } + if c.PayloadBytes < 128 { + return fmt.Errorf("payload bytes must be at least 128") + } + if c.SetupConcurrency < 1 || c.ConnectConcurrency < 1 { + return fmt.Errorf("setup and connect concurrency must be positive") + } + for name, rawURL := range map[string]string{"WebSocket": c.WSURL, "API": c.APIURL} { + u, err := url.Parse(rawURL) + if err != nil || u.Hostname() == "" { + return fmt.Errorf("invalid %s URL %q", name, rawURL) + } + if !c.AllowNonLoopback && !isLoopbackHost(u.Hostname()) { + return fmt.Errorf("%s URL %q is not loopback; pass --allow-non-loopback only for an isolated benchmark environment", name, rawURL) + } + } + return nil +} + +func isLoopbackHost(host string) bool { + host = strings.TrimSpace(strings.ToLower(host)) + if host == "localhost" { + return true + } + ip := net.ParseIP(host) + return ip != nil && ip.IsLoopback() +} diff --git a/simulator/benchmark/report.go b/simulator/benchmark/report.go new file mode 100644 index 00000000..f181aaa3 --- /dev/null +++ b/simulator/benchmark/report.go @@ -0,0 +1,168 @@ +package benchmark + +import ( + "bufio" + "encoding/json" + "fmt" + "os" + "os/exec" + "path/filepath" + "runtime" + "strconv" + "strings" + "time" +) + +type WorkloadMetadata struct { + Scenario Scenario `json:"scenario"` + Clients int `json:"clients"` + GroupSenders int `json:"group_senders,omitempty"` + TargetRate int `json:"target_messages_per_second"` + Warmup string `json:"warmup"` + Duration string `json:"duration"` + DeliveryGrace string `json:"delivery_grace"` + PayloadBytes int `json:"payload_bytes"` + StoreMessages bool `json:"store_messages"` + CountMessages bool `json:"count_messages"` + WebSocketURL string `json:"websocket_url"` + APIURL string `json:"api_url"` + EnvironmentLabel string `json:"environment_label"` + DatabaseLabel string `json:"database_label"` +} + +type EnvironmentMetadata struct { + ServerCommit string `json:"server_commit"` + WorkingTreeDirty bool `json:"working_tree_dirty"` + OS string `json:"os"` + Architecture string `json:"architecture"` + GoVersion string `json:"go_version"` + CPUModel string `json:"cpu_model"` + LogicalCPUs int `json:"logical_cpus"` + TotalMemoryBytes uint64 `json:"total_memory_bytes"` +} + +type Report struct { + SchemaVersion int `json:"schema_version"` + RunID string `json:"run_id"` + GeneratedAt time.Time `json:"generated_at"` + StartedAt time.Time `json:"started_at"` + MeasurementStartedAt time.Time `json:"measurement_started_at"` + MeasurementEndedAt time.Time `json:"measurement_ended_at"` + Workload WorkloadMetadata `json:"workload"` + Environment EnvironmentMetadata `json:"environment"` + Connections MetricSnapshot `json:"connections"` + MessageAcknowledgements MetricSnapshot `json:"message_acknowledgements"` + Deliveries MetricSnapshot `json:"deliveries"` + Limitations []string `json:"limitations"` +} + +func newReport(config Config, runID string, startedAt, measurementStarted, measurementEnded time.Time) Report { + groupSenders := 0 + if config.Scenario == ScenarioGroup { + groupSenders = config.GroupSenders + } + commit, dirty := gitState() + if config.ServerCommit != "" { + commit = config.ServerCommit + } + return Report{ + SchemaVersion: 1, + RunID: runID, + GeneratedAt: time.Now().UTC(), + StartedAt: startedAt, + MeasurementStartedAt: measurementStarted, + MeasurementEndedAt: measurementEnded, + Workload: WorkloadMetadata{ + Scenario: config.Scenario, Clients: config.Clients, GroupSenders: groupSenders, + TargetRate: config.Rate, Warmup: config.Warmup.String(), Duration: config.Duration.String(), + DeliveryGrace: config.DeliveryGrace.String(), PayloadBytes: config.PayloadBytes, + StoreMessages: config.StoreMessages, CountMessages: config.CountMessages, + WebSocketURL: config.WSURL, APIURL: config.APIURL, + EnvironmentLabel: config.EnvironmentLabel, DatabaseLabel: config.DatabaseLabel, + }, + Environment: EnvironmentMetadata{ + ServerCommit: commit, WorkingTreeDirty: dirty, OS: runtime.GOOS, + Architecture: runtime.GOARCH, GoVersion: runtime.Version(), CPUModel: cpuModel(), + LogicalCPUs: runtime.NumCPU(), TotalMemoryBytes: totalMemoryBytes(), + }, + Limitations: []string{ + "ACK latency uses the load generator's monotonic clock; delivery latency uses wall-clock timestamps within the same load-generator process.", + "Acknowledgement latency measures client publish to server ACK; delivery latency measures client publish to recipient callback.", + "The load generator and local Docker stack may share hardware unless the environment label says otherwise.", + "Results from different hardware or workload settings are not directly comparable.", + }, + } +} + +func WriteReport(path string, report Report) (string, error) { + if path == "" { + path = filepath.Join("benchmark-results", report.RunID+"-"+string(report.Workload.Scenario)+".json") + } + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return "", fmt.Errorf("create report directory: %w", err) + } + data, err := json.MarshalIndent(report, "", " ") + if err != nil { + return "", fmt.Errorf("encode benchmark report: %w", err) + } + data = append(data, '\n') + if err := os.WriteFile(path, data, 0o644); err != nil { + return "", fmt.Errorf("write benchmark report: %w", err) + } + return path, nil +} + +func gitState() (string, bool) { + commitOutput, err := exec.Command("git", "rev-parse", "HEAD").Output() + if err != nil { + return "unknown", false + } + dirtyOutput, _ := exec.Command("git", "status", "--porcelain").Output() + return strings.TrimSpace(string(commitOutput)), len(strings.TrimSpace(string(dirtyOutput))) > 0 +} + +func cpuModel() string { + if runtime.GOOS == "darwin" { + if output, err := exec.Command("sysctl", "-n", "machdep.cpu.brand_string").Output(); err == nil { + return strings.TrimSpace(string(output)) + } + } + file, err := os.Open("/proc/cpuinfo") + if err != nil { + return "unknown" + } + defer file.Close() + scanner := bufio.NewScanner(file) + for scanner.Scan() { + line := scanner.Text() + if strings.HasPrefix(line, "model name") || strings.HasPrefix(line, "Hardware") { + if _, value, ok := strings.Cut(line, ":"); ok { + return strings.TrimSpace(value) + } + } + } + return "unknown" +} + +func totalMemoryBytes() uint64 { + if runtime.GOOS == "darwin" { + if output, err := exec.Command("sysctl", "-n", "hw.memsize").Output(); err == nil { + value, _ := strconv.ParseUint(strings.TrimSpace(string(output)), 10, 64) + return value + } + } + file, err := os.Open("/proc/meminfo") + if err != nil { + return 0 + } + defer file.Close() + scanner := bufio.NewScanner(file) + for scanner.Scan() { + fields := strings.Fields(scanner.Text()) + if len(fields) >= 2 && fields[0] == "MemTotal:" { + value, _ := strconv.ParseUint(fields[1], 10, 64) + return value * 1024 + } + } + return 0 +} diff --git a/simulator/benchmark/runner.go b/simulator/benchmark/runner.go new file mode 100644 index 00000000..540ef033 --- /dev/null +++ b/simulator/benchmark/runner.go @@ -0,0 +1,309 @@ +package benchmark + +import ( + "context" + "encoding/json" + "fmt" + "strconv" + "strings" + "sync" + "sync/atomic" + "time" + + "im-server/commons/pbdefines/pbobjs" + "im-server/services/commonservices/msgdefines" + "im-server/simulator/utils" + "im-server/simulator/wsclients" +) + +type benchmarkPayload struct { + BenchmarkID string `json:"benchmark_id"` + Phase string `json:"phase"` + Sequence uint64 `json:"sequence"` + SentAtNS int64 `json:"sent_at_unix_nano"` + Padding string `json:"padding"` +} + +type connectedClient struct { + user registeredUser + client *wsclients.WsImClient +} + +type Runner struct { + config Config + runID string + groupID string + api *apiClient + connections *metricRecorder + acknowledged *metricRecorder + deliveries *metricRecorder + sequence atomic.Uint64 +} + +func NewRunner(config Config) (*Runner, error) { + if err := config.Validate(); err != nil { + return nil, err + } + nonce, err := randomNonce() + if err != nil { + return nil, fmt.Errorf("create benchmark run ID: %w", err) + } + runID := fmt.Sprintf("%d-%s", time.Now().UTC().Unix(), nonce[:8]) + return &Runner{ + config: config, + runID: runID, + groupID: "bench-group-" + runID, + api: newAPIClient(config), + connections: newMetricRecorder(), + acknowledged: newMetricRecorder(), + deliveries: newMetricRecorder(), + }, nil +} + +func (r *Runner) Run(ctx context.Context) (Report, error) { + startedAt := time.Now().UTC() + users, err := r.prepareUsers(ctx) + if err != nil { + return Report{}, err + } + if r.config.Scenario == ScenarioGroup { + memberIDs := make([]string, len(users)) + for index, user := range users { + memberIDs[index] = user.ID + } + if err := r.api.createGroup(ctx, r.groupID, memberIDs); err != nil { + return Report{}, fmt.Errorf("prepare benchmark group: %w", err) + } + } + + clients := r.connect(ctx, users) + if len(clients) != len(users) { + r.disconnect(clients) + return Report{}, fmt.Errorf("connected %d of %d clients; refusing to publish a partial baseline", len(clients), len(users)) + } + defer r.disconnect(clients) + + if r.config.Warmup > 0 { + if err := r.runTraffic(ctx, clients, r.config.Warmup, false); err != nil { + return Report{}, fmt.Errorf("warm-up: %w", err) + } + } + measurementStarted := time.Now().UTC() + if err := r.runTraffic(ctx, clients, r.config.Duration, true); err != nil { + return Report{}, fmt.Errorf("measurement: %w", err) + } + measurementEnded := measurementStarted.Add(r.config.Duration) + if r.config.DeliveryGrace > 0 { + timer := time.NewTimer(r.config.DeliveryGrace) + select { + case <-ctx.Done(): + timer.Stop() + return Report{}, ctx.Err() + case <-timer.C: + } + } + + report := newReport(r.config, r.runID, startedAt, measurementStarted, measurementEnded) + report.Connections = r.connections.snapshot(0) + report.MessageAcknowledgements = r.acknowledged.snapshot(r.config.Duration) + report.Deliveries = r.deliveries.snapshot(r.config.Duration) + return report, nil +} + +func (r *Runner) prepareUsers(ctx context.Context) ([]registeredUser, error) { + users := make([]registeredUser, r.config.Clients) + jobs := make(chan int) + errCh := make(chan error, r.config.Clients) + var workers sync.WaitGroup + workerCount := min(r.config.SetupConcurrency, r.config.Clients) + for worker := 0; worker < workerCount; worker++ { + workers.Add(1) + go func() { + defer workers.Done() + for index := range jobs { + userID := fmt.Sprintf("bench-%s-%06d", r.runID, index+1) + user, err := r.api.registerUser(ctx, userID) + if err != nil { + errCh <- fmt.Errorf("register %s: %w", userID, err) + continue + } + users[index] = user + } + }() + } + for index := range users { + select { + case <-ctx.Done(): + close(jobs) + workers.Wait() + return nil, ctx.Err() + case jobs <- index: + } + } + close(jobs) + workers.Wait() + close(errCh) + for err := range errCh { + return nil, err + } + return users, nil +} + +func (r *Runner) connect(ctx context.Context, users []registeredUser) []connectedClient { + clients := make([]connectedClient, len(users)) + semaphore := make(chan struct{}, r.config.ConnectConcurrency) + var wait sync.WaitGroup + for index, user := range users { + select { + case <-ctx.Done(): + return compactClients(clients) + case semaphore <- struct{}{}: + } + wait.Add(1) + go func(index int, user registeredUser) { + defer wait.Done() + defer func() { <-semaphore }() + client := wsclients.NewWsImClient(r.config.WSURL, r.config.AppKey, user.Token, r.onMessage, nil, nil) + client.Verbose = false + started := time.Now() + code, _ := client.Connect("benchmark", "local") + latency := time.Since(started) + if code != utils.ClientErrorCode_Success { + r.connections.recordFailure(strconv.Itoa(int(code)), latency) + return + } + r.connections.recordSuccess(latency) + clients[index] = connectedClient{user: user, client: client} + }(index, user) + } + wait.Wait() + return compactClients(clients) +} + +func compactClients(clients []connectedClient) []connectedClient { + result := make([]connectedClient, 0, len(clients)) + for _, client := range clients { + if client.client != nil { + result = append(result, client) + } + } + return result +} + +func (r *Runner) disconnect(clients []connectedClient) { + for _, client := range clients { + client.client.Disconnect() + } +} + +func (r *Runner) runTraffic(parent context.Context, clients []connectedClient, duration time.Duration, measured bool) error { + ctx, cancel := context.WithTimeout(parent, duration) + defer cancel() + senderCount := len(clients) + if r.config.Scenario == ScenarioGroup { + senderCount = min(r.config.GroupSenders, len(clients)) + } + tokens := make(chan struct{}) + var workers sync.WaitGroup + for senderIndex := 0; senderIndex < senderCount; senderIndex++ { + workers.Add(1) + go func(senderIndex int) { + defer workers.Done() + for range tokens { + r.sendOne(clients, senderIndex, measured) + } + }(senderIndex) + } + + interval := time.Second / time.Duration(r.config.Rate) + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + close(tokens) + workers.Wait() + if parent.Err() != nil { + return parent.Err() + } + return nil + case <-ticker.C: + select { + case tokens <- struct{}{}: + case <-ctx.Done(): + } + } + } +} + +func (r *Runner) sendOne(clients []connectedClient, senderIndex int, measured bool) { + sequence := r.sequence.Add(1) + phase := "warmup" + if measured { + phase = "measure" + } + payload, err := makePayload(r.runID, phase, sequence, time.Now(), r.config.PayloadBytes) + if err != nil { + if measured { + r.acknowledged.recordFailure("payload", 0) + } + return + } + flags := int32(0) + if r.config.StoreMessages { + flags = msgdefines.SetStoreMsg(flags) + } + if r.config.CountMessages { + flags = msgdefines.SetCountMsg(flags) + } + message := &pbobjs.UpMsg{MsgType: msgdefines.InnerMsgType_Text, MsgContent: payload, Flags: flags} + started := time.Now() + var code utils.ClientErrorCode + if r.config.Scenario == ScenarioPrivate { + target := clients[(senderIndex+1)%len(clients)].user.ID + code, _ = clients[senderIndex].client.SendPrivateMsg(target, message) + } else { + code, _ = clients[senderIndex].client.SendGroupMsg(r.groupID, message) + } + if !measured { + return + } + latency := time.Since(started) + if code == utils.ClientErrorCode_Success { + r.acknowledged.recordSuccess(latency) + } else { + r.acknowledged.recordFailure(strconv.Itoa(int(code)), latency) + } +} + +func (r *Runner) onMessage(message *pbobjs.DownMsg) { + var payload benchmarkPayload + if err := json.Unmarshal(message.MsgContent, &payload); err != nil { + return + } + if payload.BenchmarkID != r.runID || payload.Phase != "measure" || payload.SentAtNS <= 0 { + return + } + latency := time.Since(time.Unix(0, payload.SentAtNS)) + if latency < 0 { + return + } + r.deliveries.recordObservation(latency) +} + +func makePayload(runID, phase string, sequence uint64, sentAt time.Time, size int) ([]byte, error) { + payload := benchmarkPayload{ + BenchmarkID: runID, + Phase: phase, + Sequence: sequence, + SentAtNS: sentAt.UnixNano(), + } + base, err := json.Marshal(payload) + if err != nil { + return nil, err + } + if missing := size - len(base); missing > 0 { + payload.Padding = strings.Repeat("x", missing) + } + return json.Marshal(payload) +} diff --git a/simulator/benchmark/stats.go b/simulator/benchmark/stats.go new file mode 100644 index 00000000..0b7fbfb1 --- /dev/null +++ b/simulator/benchmark/stats.go @@ -0,0 +1,121 @@ +package benchmark + +import ( + "math" + "sort" + "sync" + "time" +) + +type LatencySummary struct { + Count int `json:"count"` + MinMS float64 `json:"min_ms"` + P50MS float64 `json:"p50_ms"` + P95MS float64 `json:"p95_ms"` + P99MS float64 `json:"p99_ms"` + MaxMS float64 `json:"max_ms"` + MeanMS float64 `json:"mean_ms"` +} + +type MetricSnapshot struct { + Attempted int64 `json:"attempted"` + Succeeded int64 `json:"succeeded"` + Failed int64 `json:"failed"` + ThroughputPerSec float64 `json:"throughput_per_second"` + Latency LatencySummary `json:"latency"` + Errors map[string]int64 `json:"errors,omitempty"` +} + +type metricRecorder struct { + mu sync.Mutex + attempted int64 + succeeded int64 + errors map[string]int64 + latencies []time.Duration +} + +func newMetricRecorder() *metricRecorder { + return &metricRecorder{errors: make(map[string]int64)} +} + +func (r *metricRecorder) recordSuccess(latency time.Duration) { + r.mu.Lock() + defer r.mu.Unlock() + r.attempted++ + r.succeeded++ + r.latencies = append(r.latencies, latency) +} + +func (r *metricRecorder) recordFailure(code string, latency time.Duration) { + r.mu.Lock() + defer r.mu.Unlock() + r.attempted++ + r.errors[code]++ +} + +func (r *metricRecorder) recordObservation(latency time.Duration) { + r.mu.Lock() + defer r.mu.Unlock() + r.attempted++ + r.succeeded++ + r.latencies = append(r.latencies, latency) +} + +func (r *metricRecorder) snapshot(window time.Duration) MetricSnapshot { + r.mu.Lock() + defer r.mu.Unlock() + errors := make(map[string]int64, len(r.errors)) + for key, value := range r.errors { + errors[key] = value + } + throughput := 0.0 + if window > 0 { + throughput = float64(r.succeeded) / window.Seconds() + } + return MetricSnapshot{ + Attempted: r.attempted, + Succeeded: r.succeeded, + Failed: r.attempted - r.succeeded, + ThroughputPerSec: throughput, + Latency: summarizeLatencies(r.latencies), + Errors: errors, + } +} + +func summarizeLatencies(values []time.Duration) LatencySummary { + if len(values) == 0 { + return LatencySummary{} + } + ordered := append([]time.Duration(nil), values...) + sort.Slice(ordered, func(i, j int) bool { return ordered[i] < ordered[j] }) + var total time.Duration + for _, value := range ordered { + total += value + } + toMS := func(value time.Duration) float64 { + return float64(value) / float64(time.Millisecond) + } + return LatencySummary{ + Count: len(ordered), + MinMS: toMS(ordered[0]), + P50MS: toMS(nearestRank(ordered, 0.50)), + P95MS: toMS(nearestRank(ordered, 0.95)), + P99MS: toMS(nearestRank(ordered, 0.99)), + MaxMS: toMS(ordered[len(ordered)-1]), + MeanMS: toMS(total / time.Duration(len(ordered))), + } +} + +func nearestRank(values []time.Duration, percentile float64) time.Duration { + if len(values) == 0 { + return 0 + } + index := int(math.Ceil(percentile*float64(len(values)))) - 1 + if index < 0 { + index = 0 + } + if index >= len(values) { + index = len(values) - 1 + } + return values[index] +} diff --git a/simulator/wsclients/wsimclient.go b/simulator/wsclients/wsimclient.go index b3a01ae3..557ff807 100644 --- a/simulator/wsclients/wsimclient.go +++ b/simulator/wsclients/wsimclient.go @@ -28,6 +28,7 @@ type WsImClient struct { DeviceOsVersion string PushToken string VoipToken string + Verbose bool DisconnectCallback func(code utils.ClientErrorCode, disMsg *codec.DisconnectMsgBody) OnMessageCallBack func(msg *pbobjs.DownMsg) @@ -63,6 +64,7 @@ func NewWsImClient(address, appkey, token string, onMessage func(msg *pbobjs.Dow DisconnectCallback: onDisconnect, Platform: "Web", // "Android", DeviceId: "testDevice", + Verbose: true, obfCode: [8]byte{1, 2, 3, 4, 5, 6, 7, 8}, isEncrypt: true, lock: &sync.RWMutex{}, @@ -183,7 +185,9 @@ func (client *WsImClient) startListener() { client.state = utils.State_Disconnect } } - fmt.Println("Stop client listener.") + if client.Verbose { + fmt.Println("Stop client listener.") + } } func (client *WsImClient) startPing() {