From 0be3cebf4d753582e987e2109a0b5fa3ce81bc94 Mon Sep 17 00:00:00 2001 From: Borja Castellano Date: Mon, 14 Sep 2026 14:54:21 +0000 Subject: [PATCH 1/8] feat(dash-spv-bench): opt-in in-process CPU and heap profiling, client always in docker `./run.sh --flame` builds the bench with the `cpu-profile` feature (pprof, 99 Hz SIGPROF sampler) and writes flamegraph.svg. `--memory-snapshot` builds with `heap-profile` (jemalloc as the global allocator, sampling heap profiling) and writes heap-peak.svg: the live heap at the highest RSS of the run. A thread re-dumps the profile each time VmRSS grows by 5 % and only the last dump is rendered, after the sync. Without the flags neither profiler is compiled in. perf is not usable on the benchmark host (perf_event_paranoid=4) nor reliably under Docker Desktop, and jeprof needed path and placeholder workarounds, so both profilers now run inside the process. peak_rss_mib is read before the profiles are rendered: rendering the CPU flamegraph of a 20-minute run allocates ~2 GB, which otherwise showed up as a 4703 MiB peak and filled half of the heap snapshot. The client now always runs in the client container, shaped or not, and the binary is always built inside the Rust image, so it links against the container's glibc (a host build fails to start on hosts with a newer glibc than bookworm's). This drops the host-run path, the perf/perl/FlameGraph/OS checks and the policer fallback for kernels without ifb: a run whose download rate cannot be enforced now fails instead of producing a result that cannot be compared. `--wallets` with a relative path now resolves against the caller's directory instead of dash-spv-bench/. run.sh goes from 664 to 366 lines. Mainnet restore, mainnet.100mbi.100ms, on dev, --flame --memory-snapshot: 20.3 min, peak_rss_mib 2280, heap snapshot at 2243 MiB RSS (1.88 GB live), 14114383 sat, 13389 addresses. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_017DruChNTWXwJoWPartZwCf --- dash-spv-bench/.gitignore | 4 - dash-spv-bench/Cargo.toml | 9 + dash-spv-bench/run.sh | 616 +++++++++------------------------- dash-spv-bench/src/main.rs | 15 +- dash-spv-bench/src/profile.rs | 91 +++++ 5 files changed, 271 insertions(+), 464 deletions(-) create mode 100644 dash-spv-bench/src/profile.rs diff --git a/dash-spv-bench/.gitignore b/dash-spv-bench/.gitignore index fb948c9f4..31fa05194 100644 --- a/dash-spv-bench/.gitignore +++ b/dash-spv-bench/.gitignore @@ -1,6 +1,5 @@ bench-results/ bench-storage/ -profiles/ *.lock @@ -17,8 +16,5 @@ chain-data/ results/ .bin/ -# FlameGraph tooling clone (run.sh --flame), fetched on demand. -.flamegraph/ - # Wallet mnemonics (one BIP39 phrase per line) — secrets, never committed. wallets.txt diff --git a/dash-spv-bench/Cargo.toml b/dash-spv-bench/Cargo.toml index 0cf11a8b9..58aa104b0 100644 --- a/dash-spv-bench/Cargo.toml +++ b/dash-spv-bench/Cargo.toml @@ -20,6 +20,15 @@ tracing = "0.1" tracing-subscriber = { version = "0.3.20", features = ["env-filter"] } indicatif = "0.18" +[target.'cfg(target_os = "linux")'.dependencies] +pprof = { version = "0.14", features = ["flamegraph"], optional = true } +tikv-jemallocator = { version = "0.7", features = ["profiling"], optional = true } +jemalloc_pprof = { version = "0.9", features = ["flamegraph", "symbolize"], optional = true } + +[features] +cpu-profile = ["dep:pprof"] +heap-profile = ["dep:tikv-jemallocator", "dep:jemalloc_pprof"] + [[bin]] name = "dash-spv-bench" path = "src/main.rs" diff --git a/dash-spv-bench/run.sh b/dash-spv-bench/run.sh index 09758e204..79e60be35 100755 --- a/dash-spv-bench/run.sh +++ b/dash-spv-bench/run.sh @@ -1,115 +1,68 @@ #!/bin/bash # -# dash-spv benchmark driver — runs ONE scenario end to end (build, bring up peers, sync, report). +# dash-spv benchmark driver: builds the client, brings up local peers when the +# scenario has them, syncs inside the client container and archives the results. # -# Usage: -# ./run.sh Run the scenario -# ./run.sh --flame Same, under the sampler (perf on Linux, sample on macOS) -# -> profiles/flamegraph.svg -# --wallets Wallets for this run: a file with one BIP39 mnemonic per -# line. No file => the run has no wallet. +# Usage: ./run.sh ... [--flame] [--memory-snapshot] [--wallets ] # -# ./run.sh 'scenarios/mainnet.*' Several scenarios: any argument that is not a file is -# ./run.sh scenarios/local.*.yml treated as a glob (quote it to let run.sh expand it, or -# ./run.sh scenarios/*.yml let your shell do it). Patterns resolve against the -# current directory and then against scenarios/, so -# 'mainnet.*' works from anywhere. +# --flame CPU flamegraph sampled by the client -> flamegraph.svg +# --memory-snapshot live-heap flamegraph at the RSS peak -> heap-peak.svg +# --wallets one BIP39 mnemonic per line (default: wallets.txt) # -# Every invocation, one scenario or twenty, writes results// containing per scenario -# .log stdout of the run (build, bring-up, summary) -# .run.log the sync trace, kept because bench-storage/ is wiped by the next run -# .summary.txt the metrics block -# plus results.tsv and report.md over all of them. A scenario that fails is recorded and the -# rest still run. -# -# RUST_LOG can be set to tweak logging, e.g. RUST_LOG=info ./run.sh scenarios/local.1ideal.yml -# It is the tracing filter for BOTH log sinks and overrides the defaults, which are -# terminal = "warn,dash_spv_bench=info" (kept light so the live bars stay readable) -# file = "warn,dash_spv=debug,dash_spv_bench=debug" (debug, for offline analysis) -# -# Outputs land in bench-storage/ (gitignored, wiped at the start of each run): -# run.log full trace of the sync (debug by default) for offline analysis -# summary.txt metrics + per-wallet tx/balance fingerprint (also printed to stdout) -# -# bench-storage/ is wiped at the start of every run, so both are archived into results/ too. -# plus the SPV storage the run produced (block_headers/, filters/, ...) +# An argument that is not a file is a glob, tried against the current directory +# and then scenarios/. Every invocation writes results// with, per +# scenario, .log, .run.log, .summary.txt and any flamegraph, plus +# results.tsv and report.md. RUST_LOG overrides the client's log filters. set -euo pipefail SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" REPO_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)" -COMPOSE_FILE="" # local mode generates one here; deleted on exit -DASH_VERSION="23.1.7" -IMAGE="dash-spv-bench/dashd:${DASH_VERSION}" +SELF="${SCRIPT_DIR}/$(basename "${BASH_SOURCE[0]}")" +INVOCATION_DIR="${PWD}" +IMAGE="dash-spv-bench/dashd:23.1.7" CLIENT_IMAGE="dash-spv-bench/client:1" -# Rust image for the Linux build, taken from the workspace's own pin so the two -# cannot drift: with a mismatched tag the image spends every run having rustup -# fetch the pinned toolchain before it can compile anything. RUST_CHANNEL="$(sed -n 's/^channel *= *"\(.*\)"/\1/p' "${REPO_ROOT}/rust-toolchain.toml" 2>/dev/null || true)" RUST_IMAGE="rust:${RUST_CHANNEL:-1.89}-bookworm" -# Separate from the host's `target/`: a different triple, and sharing one -# directory across both would make every switch a full rebuild. PROJECT="spv-bench" STATE="${SCRIPT_DIR}/.clonedir" -FLAME_SVG="${SCRIPT_DIR}/profiles/flamegraph.svg" -FLAMEGRAPH_DIR="${SCRIPT_DIR}/.flamegraph" # FlameGraph tooling clone (gitignored, inside the package) CHAIN_DIR="${SCRIPT_DIR}/chain-data" -# Absolute, because the script `cd`s to its own directory below: a relative `$0` -# stops resolving after that, which broke both `--help` and re-invoking self. -SELF="${SCRIPT_DIR}/$(basename "${BASH_SOURCE[0]}")" - -INVOCATION_DIR="${PWD}" abspath() { case "$1" in /*) printf '%s\n' "$1" ;; *) printf '%s\n' "${INVOCATION_DIR%/}/$1" ;; esac; } - cd "${SCRIPT_DIR}" -FLAME=0 +FEATURES="" +FLAGS=() SCN_ARGS=() WALLETS_ARG="" while [ $# -gt 0 ]; do case "$1" in - --flame) FLAME=1; shift ;; - --wallets) WALLETS_ARG="${2:?--wallets needs a file path}"; shift 2 ;; - -h | --help) sed -n '3,24p' "${SELF}"; exit 0 ;; + --flame) FEATURES="${FEATURES} cpu-profile"; FLAGS+=("$1"); shift ;; + --memory-snapshot) FEATURES="${FEATURES} heap-profile"; FLAGS+=("$1"); shift ;; + --wallets) WALLETS_ARG="$(abspath "${2:?--wallets needs a file path}")"; FLAGS+=("$1" "${WALLETS_ARG}"); shift 2 ;; + -h | --help) sed -n '3,/^set /p' "${SELF}" | sed '$d'; exit 0 ;; -*) echo "unknown flag: $1" >&2; exit 1 ;; *) SCN_ARGS+=("$1"); shift ;; esac done -[ "${#SCN_ARGS[@]}" -gt 0 ] || { echo "usage: $0 ... [--flame] [--wallets ]" >&2; exit 1; } +[ "${#SCN_ARGS[@]}" -gt 0 ] || { echo "usage: $0 ... [--flame] [--memory-snapshot] [--wallets ]" >&2; exit 1; } -# Resolve each argument to scenario files. -# -# A bare path stays a bare path, so the original single-file invocation is -# untouched. Anything that is not a file is treated as a glob — which covers -# both a quoted pattern (`'scenarios/mainnet.*'`, expanded here) and one the -# caller's shell already expanded into several arguments. Patterns are tried -# against the invocation directory first and then against `scenarios/`, so -# `mainnet.*` works from anywhere. SCN_FILES=() +add_matches() { + local m + while IFS= read -r m; do + case "${m}" in /*) ;; *) m="$1/${m}" ;; esac + if [ -f "${m}" ]; then SCN_FILES+=("${m}"); fi + done < <(cd "$1" && compgen -G "$2") +} for arg in "${SCN_ARGS[@]}"; do - if [ -f "${arg}" ] || [ -f "$(abspath "${arg}")" ]; then - SCN_FILES+=("$(abspath "${arg}")") - continue - fi - matched=0 - while IFS= read -r hit; do - [ -f "${hit}" ] || continue - SCN_FILES+=("${hit}") - matched=1 - done < <(cd "${INVOCATION_DIR}" 2>/dev/null && shopt -s nullglob && printf '%s\n' ${arg} | while IFS= read -r m; do abspath "${m}"; done - cd "${SCRIPT_DIR}/scenarios" 2>/dev/null && shopt -s nullglob && printf '%s\n' ${arg} | while IFS= read -r m; do printf '%s\n' "${SCRIPT_DIR}/scenarios/${m}"; done) - [ "${matched}" -eq 1 ] || { echo "Error: no scenario matched: ${arg}" >&2; exit 1; } + n="${#SCN_FILES[@]}" + add_matches "${INVOCATION_DIR}" "${arg}" + [ "${#SCN_FILES[@]}" -gt "${n}" ] || add_matches "${SCRIPT_DIR}/scenarios" "${arg}" + [ "${#SCN_FILES[@]}" -gt "${n}" ] || { echo "Error: no scenario matched: ${arg}" >&2; exit 1; } done -# Run each scenario through a fresh invocation of this script and collect the -# numbers. Re-invoking rather than looping in place keeps the per-scenario path -# — exports, compose, traps, teardown — byte for byte what a single-file run has -# always done, so the extension cannot change the thing it is measuring. -# -# Taken for one scenario as much as for twenty: naming a file and naming a glob -# that happens to match one file are the same request, and having them leave -# their results in different shapes is a trap for anyone scripting on top. The -# child is marked so it runs the scenario instead of wrapping it again. +# Each scenario runs in a fresh invocation of this script, so one that fails +# cannot take the others with it. if [ -z "${BENCH_BATCH:-}" ]; then RUN_TS="$(date +%Y%m%d-%H%M%S)" OUT_DIR="${SCRIPT_DIR}/results/${RUN_TS}" @@ -117,11 +70,6 @@ if [ -z "${BENCH_BATCH:-}" ]; then TSV="${OUT_DIR}/results.tsv" REPORT="${OUT_DIR}/report.md" printf 'scenario\tcompleted\ttotal_ms\tblock_headers_ms\tfilter_headers_ms\tfilters_ms\ttransactions\tconfirmed_sat\tpeak_rss_mib\n' >"${TSV}" - - child_flags=() - [ "${FLAME}" -eq 1 ] && child_flags+=(--flame) - [ -n "${WALLETS_ARG}" ] && child_flags+=(--wallets "${WALLETS_ARG}") - metric() { awk -F':[[:space:]]*' -v k="$2" '$1==k {gsub(/[[:space:]]+$/,"",$2); print $2; exit}' "$1"; } echo "==> ${#SCN_FILES[@]} scenario(s); results in ${OUT_DIR}" @@ -129,37 +77,25 @@ if [ -z "${BENCH_BATCH:-}" ]; then name="$(basename "${f}" .yml)" log="${OUT_DIR}/${name}.log" echo "===== ${name} =====" - # A scenario that fails must not take the batch with it: record it and move - # on, or one bad run costs every result after it. BENCH_BATCH=1 BENCH_ARCHIVE_DIR="${OUT_DIR}" BENCH_ARCHIVE_NAME="${name}" \ - "${SELF}" "${f}" "${child_flags[@]+"${child_flags[@]}"}" 2>&1 | tee "${log}" || true - # Read the metrics from the summary the child archived, not from its stdout: - # the summary is written by the binary and is plain text either way, while a - # pty-captured log carries the bars' escape codes and carriage returns. - src="${log}" - if [ -s "${OUT_DIR}/${name}.summary.txt" ]; then src="${OUT_DIR}/${name}.summary.txt"; fi - wallet_line="$(grep -m1 -o 'confirmed_sat=[0-9]*' "${src}" || true)" - printf '%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n' \ - "${name}" \ - "$(metric "${src}" completed)" \ - "$(metric "${src}" total_ms)" \ - "$(metric "${src}" block_headers_ms)" \ - "$(metric "${src}" filter_headers_ms)" \ - "$(metric "${src}" filters_ms)" \ - "$(metric "${src}" transactions)" \ - "$(sed -n 's/.*confirmed_sat=\([0-9]*\).*/\1/p' <<<"${wallet_line}")" \ - "$(metric "${src}" peak_rss_mib)" \ - >>"${TSV}" + "${SELF}" "${f}" ${FLAGS[@]+"${FLAGS[@]}"} 2>&1 | tee "${log}" || true + src="${OUT_DIR}/${name}.summary.txt" + [ -s "${src}" ] || src="${log}" + row="${name}" + for key in completed total_ms block_headers_ms filter_headers_ms filters_ms transactions; do + row+=$'\t'"$(metric "${src}" "${key}")" + done + row+=$'\t'"$(grep -m1 -o 'confirmed_sat=[0-9]*' "${src}" | cut -d= -f2 || true)" + row+=$'\t'"$(metric "${src}" peak_rss_mib)" + echo "${row}" >>"${TSV}" done { echo "# dash-spv bench — ${RUN_TS}" echo - echo "| scenario | completed | total_ms | headers_ms | filter_headers_ms | filters_ms | transactions | confirmed_sat | peak_rss_mib |" - echo "|---|---|---|---|---|---|---|---|---|" - tail -n +2 "${TSV}" | awk -F'\t' '{printf "| %s | %s | %s | %s | %s | %s | %s | %s | %s |\n", $1,$2,$3,$4,$5,$6,$7,$8,$9}' + awk -F'\t' '{ line = "|"; for (i = 1; i <= NF; i++) line = line " " $i " |"; print line + if (NR == 1) { line = "|"; for (i = 1; i <= NF; i++) line = line "---|"; print line } }' "${TSV}" } >"${REPORT}" - echo echo "==> report: ${REPORT}" cat "${REPORT}" @@ -167,33 +103,27 @@ if [ -z "${BENCH_BATCH:-}" ]; then fi SCN_FILE="${SCN_FILES[0]}" -[ -f "${SCN_FILE}" ] || { echo "Error: scenario file not found: ${SCN_FILE}" >&2; exit 1; } -YQ_VERSION="v4.44.6" -ensure_yq() { - if command -v yq >/dev/null 2>&1; then YQ=yq; return 0; fi - local bin="${SCRIPT_DIR}/.bin/yq" - if [ ! -x "${bin}" ]; then +if command -v yq >/dev/null 2>&1; then + YQ=yq +else + YQ="${SCRIPT_DIR}/.bin/yq" + if [ ! -x "${YQ}" ]; then + os="$(uname -s | tr '[:upper:]' '[:lower:]')" + case "$(uname -m)" in x86_64 | amd64) arch=amd64 ;; aarch64 | arm64) arch=arm64 ;; *) arch="$(uname -m)" ;; esac + echo "==> fetching yq into .bin/yq" mkdir -p "${SCRIPT_DIR}/.bin" - local os arch - os="$(uname -s | tr '[:upper:]' '[:lower:]')"; arch="$(uname -m)" - case "${arch}" in x86_64 | amd64) arch=amd64 ;; aarch64 | arm64) arch=arm64 ;; esac - echo "==> fetching yq ${YQ_VERSION} (${os}/${arch}) into .bin/yq" - curl -fsSL "https://github.com/mikefarah/yq/releases/download/${YQ_VERSION}/yq_${os}_${arch}" \ - -o "${bin}" || { echo "Error: could not download yq; install it manually." >&2; exit 1; } - chmod +x "${bin}" + curl -fsSL "https://github.com/mikefarah/yq/releases/download/v4.44.6/yq_${os}_${arch}" -o "${YQ}" \ + || { echo "Error: could not download yq; install it manually." >&2; exit 1; } + chmod +x "${YQ}" fi - YQ="${bin}" -} -ensure_yq -scn() { "${YQ}" "$1" "${SCN_FILE}"; } # evaluate a yq expression against the scenario file - -_peer_group() { - "${YQ}" ".peers[$1] | [.count, .latency_ms // 0, .jitter_ms // 0, .loss_pct // 0, .rate_kbit // 0, .corrupt_pct // 0, .reorder_pct // 0] | @tsv" "${SCN_FILE}" -} +fi +scn() { "${YQ}" "$1" "${SCN_FILE}"; } -build_netem() { - local lat="$1" jit="$2" loss="$3" rate="$4" corrupt="$5" reorder="$6" a="" +# `tc netem` arguments for the link described at yq path $1. +netem_args() { + local lat jit loss rate corrupt reorder a="" + read -r lat jit loss rate corrupt reorder <<<"$(scn "$1 | [.latency_ms // 0, .jitter_ms // 0, .loss_pct // 0, .rate_kbit // 0, .corrupt_pct // 0, .reorder_pct // 0] | @tsv")" [ "${lat}" != 0 ] && { a="delay ${lat}ms"; [ "${jit}" != 0 ] && a="${a} ${jit}ms"; } [ "${loss}" != 0 ] && a="${a} loss ${loss}%" [ "${rate}" != 0 ] && a="${a} rate ${rate}kbit" @@ -202,38 +132,8 @@ build_netem() { echo "${a# }" } -# The measured client's own link shaping, if the scenario asks for one -client_netem() { - local lat jit loss rate corrupt reorder - read -r lat jit loss rate corrupt reorder <<<"$("${YQ}" \ - '[.client.latency_ms // 0, .client.jitter_ms // 0, .client.loss_pct // 0, .client.rate_kbit // 0, .client.corrupt_pct // 0, .client.reorder_pct // 0] | @tsv' \ - "${SCN_FILE}")" - build_netem "${lat}" "${jit}" "${loss}" "${rate}" "${corrupt}" "${reorder}" -} - -# The client's bandwidth cap on its own, so the container can mirror it onto -# ingress -client_rate_kbit() { - "${YQ}" '.client.rate_kbit // 0' "${SCN_FILE}" -} - -peers_summary() { - local ng g count lat jit loss rate corrupt reorder tag out="" - ng="$(scn '.peers | length')" - case "${ng}" in ''|null|*[!0-9]*) ng=0 ;; esac - for ((g = 0; g < ng; g++)); do - read -r count lat jit loss rate corrupt reorder <<<"$(_peer_group "${g}")" - tag="${lat}ms" - [ "${jit}" != 0 ] && tag="${tag}±${jit}" - [ "${loss}" != 0 ] && tag="${tag}/${loss}%loss" - [ "${rate}" != 0 ] && tag="${tag}/${rate}kbit" - out="${out}, ${count}×${tag}" - done - echo "${out#, }" -} - emit_compose() { - local out="$1" peer_cpus="$2" + local out="$1" groups g count n peer=0 cat >"${out}" <
>"${out}" <>"${out}" </dev/null \ + if [ "\$\${INGRESS_RATE_KBIT:-0}" != 0 ]; then + ip link add ifb0 type ifb \ && ip link set ifb0 up \ && tc qdisc add dev eth0 handle ffff: ingress \ && tc filter add dev eth0 parent ffff: protocol all prio 1 u32 \ match u32 0 0 action mirred egress redirect dev ifb0 \ - && tc qdisc add dev ifb0 root netem rate \$\${INGRESS_RATE_KBIT}kbit; then - echo "client netem (ingress): rate \$\${INGRESS_RATE_KBIT}kbit via ifb0" - else - tc qdisc del dev eth0 ingress 2>/dev/null || true - if tc qdisc add dev eth0 handle ffff: ingress 2>/dev/null \ - && tc filter add dev eth0 parent ffff: protocol all prio 1 u32 \ - match u32 0 0 action police rate \$\${INGRESS_RATE_KBIT}kbit \ - burst \$\${INGRESS_BURST_K}k mtu 64k conform-exceed drop; then - echo "client police (ingress): rate \$\${INGRESS_RATE_KBIT}kbit burst \$\${INGRESS_BURST_K}k" - echo "NOTE: no ifb, so ingress is POLICED (drops) not shaped (queues) — throughput lands within ~3% but the loss pattern differs; do not compare against ifb runs" - else - echo "WARNING: ingress shaping failed (NET_ADMIN? no ifb and no act_police?)" - echo "WARNING: DOWNLOADS ARE UNSHAPED — rate_kbit is NOT enforced and this run is not comparable" - fi - fi + && tc qdisc add dev ifb0 root netem rate \$\${INGRESS_RATE_KBIT}kbit \ + || { echo "ERROR: cannot shape the client's downloads (no ifb in this kernel?)"; exit 1; } + echo "client netem (ingress): rate \$\${INGRESS_RATE_KBIT}kbit via ifb0" fi exec /usr/local/bin/dash-spv-bench SERVICE } - MODE="$(scn '.mode // "local"')" case "${MODE}" in local | testnet | mainnet) ;; *) echo "Error: mode must be 'local', 'testnet' or 'mainnet' (got '${MODE}')" >&2; exit 1 ;; esac -export BENCH_MODE="${MODE}" export CLONE_DIR="${CLONE_DIR:-/nonexistent}" -CLIENT_NETEM="$(client_netem)" -CLIENT_RATE_KBIT="$(client_rate_kbit)" +CLIENT_NETEM="$(netem_args .client)" +CLIENT_RATE_KBIT="$(scn '.client.rate_kbit // 0')" case "${CLIENT_RATE_KBIT}" in '' | *[!0-9]*) echo "Error: client.rate_kbit must be a whole number of kbit (got '${CLIENT_RATE_KBIT}')" >&2; exit 1 ;; esac -# Burst for the policer fallback: ~0.5s of the rate (kbit/16 => kbytes), floored -# so a very small rate still gets a workable bucket. Unused on the ifb path. -CLIENT_INGRESS_BURST_K=$(( CLIENT_RATE_KBIT / 16 )) -if [ "${CLIENT_INGRESS_BURST_K}" -lt 32 ]; then CLIENT_INGRESS_BURST_K=32; fi BENCH_CPUS="$(scn '.cpus // ""')" -BENCH_MAX_PEERS="$(scn '.max_peers // ""')" # only if set; else the ClientConfig default +BENCH_MAX_PEERS="$(scn '.max_peers // ""')" export BENCH_MAX_PEERS - -BENCH_WALLET_FILE="${SCRIPT_DIR}/wallets.txt" -[ -n "${WALLETS_ARG}" ] && BENCH_WALLET_FILE="$(abspath "${WALLETS_ARG}")" -export BENCH_WALLET_FILE +BENCH_WALLET_FILE="${WALLETS_ARG:-${SCRIPT_DIR}/wallets.txt}" export BENCH_STORAGE_DIR="${SCRIPT_DIR}/bench-storage" DESC="$(scn '.description // ""')" -[ -n "${DESC}" ] && echo "==> description: ${DESC}" +[ -z "${DESC}" ] || echo "==> description: ${DESC}" -BLOCKS="" if [ "${MODE}" = local ]; then - BLOCKS="$(scn '.blocks // 1000000')" - export BENCH_HEIGHT="${BLOCKS}" - unset BENCH_START_HEIGHT # local always syncs from genesis + BENCH_HEIGHT="$(scn '.blocks // 1000000')" + export BENCH_HEIGHT + unset BENCH_START_HEIGHT else BENCH_PEERS="$(scn '.peers // [] | join(",")')" export BENCH_PEERS - sh="$(scn '.start_height // ""')"; [ -n "${sh}" ] && export BENCH_START_HEIGHT="${sh}" + start_height="$(scn '.start_height // ""')" + [ -z "${start_height}" ] || export BENCH_START_HEIGHT="${start_height}" fi -expand_cpus() { - local part lo hi - local IFS=, - for part in $1; do - case "${part}" in - *-*) lo="${part%-*}"; hi="${part#*-}"; seq "${lo}" "${hi}" ;; - *) echo "${part}" ;; - esac - done -} - -CPU_PREFIX=() +# The client gets BENCH_CPUS, docker peers get every other core. +BENCH_PEER_CPUS="" if [ -n "${BENCH_CPUS}" ]; then ncpu="$(getconf _NPROCESSORS_ONLN 2>/dev/null || sysctl -n hw.ncpu 2>/dev/null || nproc 2>/dev/null || echo 0)" - bench_cores="$(expand_cpus "${BENCH_CPUS}" | sort -nu)" - if [ "${ncpu}" -gt 0 ]; then - peer_cores="" - for i in $(seq 0 $((ncpu - 1))); do - grep -qxF "${i}" <<<"${bench_cores}" || peer_cores="${peer_cores},${i}" - done - BENCH_PEER_CPUS="${peer_cores#,}" - fi - - if [ -n "${CLIENT_NETEM}" ]; then - # The client is a container: `cpuset` on the service does the pinning that - # `taskset` does for a host run, and it works on macOS where taskset does - # not exist at all. Emitted by `emit_client_service` from BENCH_CPUS. - echo "==> pinning the measured client container to CPUs ${BENCH_CPUS}${BENCH_PEER_CPUS:+; docker peers to ${BENCH_PEER_CPUS}}" - elif command -v taskset >/dev/null 2>&1; then - CPU_PREFIX=(taskset -c "${BENCH_CPUS}") - echo "==> pinning the measured run to CPUs ${BENCH_CPUS}${BENCH_PEER_CPUS:+; docker peers to ${BENCH_PEER_CPUS}}" - else - echo "==> note: 'taskset' not found (e.g. macOS); running the measured binary UNPINNED${BENCH_PEER_CPUS:+ (docker peers still pinned to ${BENCH_PEER_CPUS})}" >&2 - fi -fi - -# Where the measured binary will be, decided before the compose that mounts -# it is written. Everything lands in the workspace `target/`: a cross build -# gets cargo's own per-triple subdirectory, so it cannot collide with a host -# build, and neither duplicates the dependency graph. -if [ -n "${CLIENT_NETEM}" ] && [ "$(uname -s)" != Linux ]; then - CROSS_TRIPLE="$(docker run --rm "${RUST_IMAGE}" rustc -vV | sed -n 's/^host: //p')" - [ -n "${CROSS_TRIPLE}" ] || { echo "could not read the build image's target triple" >&2; exit 1; } - BIN="${REPO_ROOT}/target/${CROSS_TRIPLE}/release/dash-spv-bench" -else - CROSS_TRIPLE="" - BIN="${REPO_ROOT}/target/release/dash-spv-bench" -fi - -# A compose file is needed for the peers (local mode) and for the client -# container (any mode with a `client:` section) — testnet/mainnet shape the -# client against the real network, with no peer services at all. -if [ "${MODE}" = local ] || [ -n "${CLIENT_NETEM}" ]; then - COMPOSE_FILE="$(mktemp "${SCRIPT_DIR}/.scenario.XXXXXX")" - mv "${COMPOSE_FILE}" "${COMPOSE_FILE}.yml" - COMPOSE_FILE="${COMPOSE_FILE}.yml" - npeers="$(emit_compose "${COMPOSE_FILE}" "${BENCH_PEER_CPUS:-}")" - if [ "${MODE}" = local ]; then - echo "==> scenario '$(basename "${SCN_FILE}" .yml)': ${npeers} peers [$(peers_summary)], cpus=${BENCH_CPUS:-}, blocks=${BLOCKS}" - fi + bench_cores="$(IFS=,; for part in ${BENCH_CPUS}; do case "${part}" in *-*) seq "${part%-*}" "${part#*-}" ;; *) echo "${part}" ;; esac; done)" + for ((i = 0; i < ncpu; i++)); do + grep -qxF "${i}" <<<"${bench_cores}" || BENCH_PEER_CPUS="${BENCH_PEER_CPUS},${i}" + done + BENCH_PEER_CPUS="${BENCH_PEER_CPUS#,}" + echo "==> pinning the client container to CPUs ${BENCH_CPUS}${BENCH_PEER_CPUS:+; docker peers to ${BENCH_PEER_CPUS}}" fi -[ -n "${CLIENT_NETEM}" ] && echo "==> client link shaped: ${CLIENT_NETEM}" +# Built in the Rust image so the binary links against the client image's glibc. +BIN="${REPO_ROOT}/target/bench/release/dash-spv-bench" +echo "==> building bench binary in ${RUST_IMAGE}" +docker run --rm -v "${REPO_ROOT}:/src" -w /src --user "$(id -u):$(id -g)" \ + -e CARGO_HOME=/src/target/bench/cargo-home -e CARGO_TARGET_DIR=/src/target/bench \ + -e CARGO_NET_GIT_FETCH_WITH_CLI=true -e CARGO_PROFILE_RELEASE_DEBUG=line-tables-only \ + "${RUST_IMAGE}" cargo build --release -p dash-spv-bench ${FEATURES:+--features "${FEATURES# }"} +[ -x "${BIN}" ] || { echo "build produced no binary at ${BIN}" >&2; exit 1; } + +COMPOSE_FILE="$(mktemp "${SCRIPT_DIR}/.scenario.XXXXXX")" +mv "${COMPOSE_FILE}" "${COMPOSE_FILE}.yml" +COMPOSE_FILE="${COMPOSE_FILE}.yml" +emit_compose "${COMPOSE_FILE}" +[ -z "${CLIENT_NETEM}" ] || echo "==> client link shaped: ${CLIENT_NETEM}" compose() { docker compose -p "${PROJECT}" -f "${COMPOSE_FILE}" "$@"; } +teardown() { + compose down --remove-orphans >/dev/null 2>&1 || true + [ ! -f "${STATE}" ] || rm -rf "$(cat "${STATE}")" "${STATE}" 2>/dev/null || true + rm -rf "${SCRIPT_DIR}"/.bench-clones.* 2>/dev/null || true +} +trap 'teardown; rm -f "${COMPOSE_FILE}"' EXIT +trap 'exit 130' INT +trap 'exit 143' TERM +trap 'exit 129' HUP + wait_loaded() { - local c logs + local c logs all for _ in $(seq 1 400); do - local all=1 + all=1 for c in "$@"; do logs="$(docker logs "${c}" 2>&1)" || { all=0; break; } case "${logs}" in *"init message: Done loading"*) ;; *) all=0; break ;; esac done - [ "${all}" -eq 1 ] && return 0 + [ "${all}" -eq 0 ] || return 0 echo -n "."; sleep 3 done return 1 } -teardown() { - [ -n "${COMPOSE_FILE}" ] && compose down --remove-orphans >/dev/null 2>&1 || true - [ -f "${STATE}" ] && { rm -rf "$(cat "${STATE}")" 2>/dev/null || true; rm -f "${STATE}"; } - rm -rf "${SCRIPT_DIR}"/.bench-clones.* 2>/dev/null || true -} - -arm_teardown() { - # On exit: tear down, THEN delete the generated compose (down needs it to still exist). - trap 'teardown; [ -n "${COMPOSE_FILE}" ] && rm -f "${COMPOSE_FILE}"' EXIT - trap 'exit 130' INT - trap 'exit 143' TERM - trap 'exit 129' HUP -} - -bring_up() { +if [ "${MODE}" = local ]; then + bash "${SCRIPT_DIR}/snapshot-chain.sh" # builds ./chain-data to BENCH_HEIGHT via docker echo "==> ensuring a clean network" teardown docker image inspect "${IMAGE}" >/dev/null 2>&1 || { echo "==> building peer image"; compose build; } - - # The client is in the same compose file but is not a peer: it must not be - # counted, started here, or waited on for "Done loading". - local services; services="$(compose config --services | grep -v '^client$' || true)" - local n; n="$(echo ${services} | wc -w | tr -d ' ')" - # BENCH_PEERS is derived from the generated peers (host ports 19401..). - local peers_csv="" - for svc in ${services}; do peers_csv="${peers_csv},127.0.0.1:$((19400 + ${svc#dashd}))"; done - export BENCH_PEERS="${peers_csv#,}" - - echo "==> CoW-cloning ${CHAIN_DIR} for ${n} peers (instant)" - local clone_dir; clone_dir="$(mktemp -d "${SCRIPT_DIR}/.bench-clones.XXXXXX")" - echo "${clone_dir}" > "${STATE}" + services="$(compose config --services | grep -v '^client$' || true)" + npeers="$(echo ${services} | wc -w | tr -d ' ')" + peers_summary="$(scn '[.peers[] | (.count | tostring) + "×" + ([to_entries[] | select(.key != "count") | .key + "=" + (.value | tostring)] | join(" "))] | join(", ")')" + echo "==> scenario '$(basename "${SCN_FILE}" .yml)': ${npeers} peers [${peers_summary}], cpus=${BENCH_CPUS:-}, blocks=${BENCH_HEIGHT}" + echo "==> CoW-cloning ${CHAIN_DIR} for ${npeers} peers" + clone_dir="$(mktemp -d "${SCRIPT_DIR}/.bench-clones.XXXXXX")" + echo "${clone_dir}" >"${STATE}" for svc in ${services}; do - local dst="${clone_dir}/peer${svc#dashd}" + dst="${clone_dir}/peer${svc#dashd}" cp -c -R "${CHAIN_DIR}" "${dst}" 2>/dev/null \ || cp --reflink=auto -R "${CHAIN_DIR}" "${dst}" 2>/dev/null \ || cp -R "${CHAIN_DIR}" "${dst}" done - - local batch=4 - echo "==> starting ${n} peers in batches of ${batch}" - local started="" count=0 + echo "==> starting ${npeers} peers in batches of 4" + started="" + count=0 for svc in ${services}; do CLONE_DIR="${clone_dir}" compose up -d "${svc}" >/dev/null 2>&1 - started="${started} spv-bench-${svc}"; count=$((count + 1)) - if [ $((count % batch)) -eq 0 ]; then - echo -n " loaded ${count}/${n} " + started="${started} spv-bench-${svc}" + count=$((count + 1)) + if [ $((count % 4)) -eq 0 ] || [ "${count}" -eq "${npeers}" ]; then + echo -n " loaded ${count}/${npeers} " wait_loaded ${started} || { echo " timeout loading batch" >&2; exit 1; } echo " ok" fi done - if [ $((count % batch)) -ne 0 ]; then # wait on the trailing partial batch too - echo -n " loaded ${count}/${n} " - wait_loaded ${started} || { echo " timeout loading batch" >&2; exit 1; } - echo " ok" - fi - - echo "==> ${n} peers started" - - # A containerised client shares the compose network with the peers, so it must - # reach them at their container addresses; the published host ports only exist - # for a client running on the host. Read the addresses back from the running - # containers rather than pinning a subnet in the compose file: a pinned subnet - # is one more thing that can collide with whatever else the machine has up, - # VPNs included, and it buys nothing a lookup does not. - if [ -n "${CLIENT_NETEM}" ]; then - local ip csv="" - for svc in ${services}; do - ip="$(docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' "spv-bench-${svc}")" - [ -n "${ip}" ] || { echo "Error: could not resolve address of spv-bench-${svc}" >&2; exit 1; } - csv="${csv},${ip}:19400" - done - export BENCH_PEERS="${csv#,}" - echo "==> client will reach peers at ${BENCH_PEERS}" - fi -} - -build_bin() { - if [ -n "${CROSS_TRIPLE}" ]; then - # The client runs in a Linux container, so the binary has to be a Linux - # one. On a Linux host the ordinary build already is — same triple, and - # glibc is forward compatible — so only a non-Linux host comes here. - # Bind-mounts the workspace and builds into its `target/`, which cargo - # keeps under the triple, so this stays incremental — an image layer build - # would replay the whole workspace on every source change, which makes A/B - # runs unusable. - echo "==> cross-building bench binary (${RUST_IMAGE}, ${CROSS_TRIPLE})" - # As the invoking user, with CARGO_HOME inside the workspace: writing - # into the shared `target/` as root leaves artifacts the host build then - # cannot overwrite. - docker run --rm \ - -v "${REPO_ROOT}:/src" \ - -w /src \ - --user "$(id -u):$(id -g)" \ - -e CARGO_HOME=/src/target/.cross-cargo-home \ - -e CARGO_TARGET_DIR=/src/target \ - -e CARGO_NET_GIT_FETCH_WITH_CLI=true \ - -e CARGO_PROFILE_RELEASE_DEBUG=line-tables-only \ - "${RUST_IMAGE}" \ - cargo build --release --target "${CROSS_TRIPLE}" -p dash-spv-bench - else - echo "==> building bench binary (release + line-table symbols)" - ( cd "${REPO_ROOT}" && CARGO_PROFILE_RELEASE_DEBUG=line-tables-only \ - cargo build --release -p dash-spv-bench ) - fi - [ -x "${BIN}" ] || { echo "build produced no binary at ${BIN}" >&2; exit 1; } -} - -FLAME_TOOL="" -if [ "${FLAME}" -eq 1 ]; then # fail fast on missing profiler deps, before building - if command -v perf >/dev/null; then FLAME_TOOL=perf # Linux - elif command -v /usr/bin/sample >/dev/null; then FLAME_TOOL=sample # macOS - else echo "flame mode needs 'perf' (Linux) or '/usr/bin/sample' (macOS)" >&2; exit 1; fi - [ -f "${FLAMEGRAPH_DIR}/flamegraph.pl" ] || \ - git clone --depth 1 https://github.com/brendangregg/FlameGraph "${FLAMEGRAPH_DIR}" -fi - -build_bin - -if [ "${MODE}" = local ]; then - arm_teardown - bash "${SCRIPT_DIR}/snapshot-chain.sh" # builds ./chain-data to BENCH_HEIGHT via docker - bring_up + # Container addresses, read back rather than pinned to a subnet that could + # collide with whatever else the machine has up. + csv="" + for svc in ${services}; do + ip="$(docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' "spv-bench-${svc}")" + [ -n "${ip}" ] || { echo "Error: could not resolve address of spv-bench-${svc}" >&2; exit 1; } + csv="${csv},${ip}:19400" + done + export BENCH_PEERS="${csv#,}" + echo "==> ${npeers} peers started, reachable at ${BENCH_PEERS}" else echo "==> ${MODE} mode, peers: ${BENCH_PEERS:-}" fi -# Run the sync with its live output straight on the terminal instead of through -# the batch wrapper's pipe -run_live() { - # Not `[ -w /dev/tty ]`: the device node is writable even with no - # controlling terminal, and the redirect then fails with ENXIO. Only - # actually opening it answers the question. - if { : >/dev/tty; } 2>/dev/null; then - "$@" >/dev/tty 2>&1 - else - "$@" - fi -} - -if [ -n "${CLIENT_NETEM}" ]; then - # `run` rather than `up`: it streams the client's own stdout and gives back - # its exit code, which is what the report is read from. Profiling is not - # wired through the container, so `--flame` is refused rather than silently - # producing a graph of the host doing nothing. - [ "${FLAME}" -eq 0 ] || { echo "Error: --flame is not supported with a containerised client" >&2; exit 1; } - arm_teardown - echo "==> running sync in the client container" - run_live compose run --rm --build client -elif [ "${FLAME}" -eq 0 ]; then - echo "==> running sync" - run_live "${CPU_PREFIX[@]+"${CPU_PREFIX[@]}"}" "${BIN}" -elif [ "${FLAME_TOOL}" = perf ]; then - echo "==> running sync under perf" - mkdir -p "$(dirname "${FLAME_SVG}")" - perf record -F 499 -g -o /tmp/bench-perf.data -- "${CPU_PREFIX[@]+"${CPU_PREFIX[@]}"}" "${BIN}" - perf script -i /tmp/bench-perf.data \ - | "${FLAMEGRAPH_DIR}/stackcollapse-perf.pl" \ - | "${FLAMEGRAPH_DIR}/flamegraph.pl" --title "dash-spv sync" --colors hot > "${FLAME_SVG}" - echo "==> wrote ${FLAME_SVG}" +echo "==> running sync in the client container" +if { : >/dev/tty; } 2>/dev/null; then + compose run --rm --build client >/dev/tty 2>&1 else - echo "==> running sync under the sampler" - "${CPU_PREFIX[@]+"${CPU_PREFIX[@]}"}" "${BIN}" & bpid=$! - sleep 3 - /usr/bin/sample "${bpid}" 2000 1 -file /tmp/bench.sample.txt -mayDie >/dev/null 2>&1 || true - wait "${bpid}" 2>/dev/null || true - mkdir -p "$(dirname "${FLAME_SVG}")" - "${FLAMEGRAPH_DIR}/stackcollapse-sample.awk" /tmp/bench.sample.txt \ - | sed -E 's/^Thread_[^;]*;//' \ - | "${FLAMEGRAPH_DIR}/flamegraph.pl" --title "dash-spv sync" --colors hot > "${FLAME_SVG}" - echo "==> wrote ${FLAME_SVG}" + compose run --rm --build client fi -# Keep this run's outputs, which `bench-storage/` does not: it is wiped at the -# start of every run, so without this the trace of anything but the most recent -# scenario is gone — and in a batch that meant 15 of 16 syncs left nothing to -# look at afterwards. -# -# The batch passes its own directory in, so a child archives straight into it -# and there is one copy, named after the scenario. A lone run makes its own. -ARCHIVE_DIR="${BENCH_ARCHIVE_DIR:-${SCRIPT_DIR}/results/$(date +%Y%m%d-%H%M%S)-$(basename "${SCN_FILE}" .yml)}" -# Prefixed with the scenario only inside a batch, where one directory holds -# every scenario. A lone run's directory is already named after it. -ARCHIVE_NAME="${BENCH_ARCHIVE_NAME:-}" +ARCHIVE_DIR="${BENCH_ARCHIVE_DIR:?BENCH_ARCHIVE_DIR is set by the batch wrapper}" mkdir -p "${ARCHIVE_DIR}" -for out in run.log summary.txt; do +for out in run.log summary.txt flamegraph.svg heap-peak.svg; do [ -s "${BENCH_STORAGE_DIR}/${out}" ] || continue - cp "${BENCH_STORAGE_DIR}/${out}" "${ARCHIVE_DIR}/${ARCHIVE_NAME:+${ARCHIVE_NAME}.}${out}" + cp "${BENCH_STORAGE_DIR}/${out}" "${ARCHIVE_DIR}/${BENCH_ARCHIVE_NAME}.${out}" done echo "==> logs archived to ${ARCHIVE_DIR}" diff --git a/dash-spv-bench/src/main.rs b/dash-spv-bench/src/main.rs index 03b34da7c..35544319b 100644 --- a/dash-spv-bench/src/main.rs +++ b/dash-spv-bench/src/main.rs @@ -1,5 +1,7 @@ mod dashboard; mod metrics; +#[cfg(target_os = "linux")] +mod profile; use std::net::SocketAddr; use std::path::PathBuf; @@ -40,11 +42,11 @@ fn load_mnemonics() -> Vec { .unwrap_or_default() } -fn peak_rss_kb() -> Option { +fn proc_status_kb(field: &str) -> Option { std::fs::read_to_string("/proc/self/status") .ok()? .lines() - .find_map(|line| line.strip_prefix("VmHWM:"))? + .find_map(|line| line.strip_prefix(field))? .split_whitespace() .next()? .parse() @@ -89,6 +91,9 @@ async fn main() -> Result<()> { ) .init(); + #[cfg(target_os = "linux")] + let profilers = profile::start(); + let mode = env_or("BENCH_MODE", "local").trim().to_ascii_lowercase(); let (network, remote) = match mode.as_str() { "local" => (Network::Testnet, false), @@ -197,9 +202,13 @@ async fn main() -> Result<()> { run_handle.abort(); let _ = run_handle.await; + let peak_rss_kb = proc_status_kb("VmHWM:"); + #[cfg(target_os = "linux")] + profilers.finish(&output_dir); + use std::fmt::Write as _; let mut report = format!("{m}\n"); - if let Some(kb) = peak_rss_kb() { + if let Some(kb) = peak_rss_kb { let _ = writeln!(report, "peak_rss_mib: {}", kb / 1024); } diff --git a/dash-spv-bench/src/profile.rs b/dash-spv-bench/src/profile.rs new file mode 100644 index 000000000..05735115a --- /dev/null +++ b/dash-spv-bench/src/profile.rs @@ -0,0 +1,91 @@ +use std::path::Path; + +#[cfg(feature = "heap-profile")] +#[global_allocator] +static ALLOC: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc; + +#[cfg(feature = "heap-profile")] +#[export_name = "_rjem_malloc_conf"] +pub static MALLOC_CONF: &[u8] = b"prof:true,prof_active:true,lg_prof_sample:19\0"; + +#[cfg(feature = "heap-profile")] +type HeapPeak = std::sync::Arc>>; + +pub(crate) struct Profilers { + #[cfg(feature = "cpu-profile")] + cpu: Option>, + #[cfg(feature = "heap-profile")] + heap_peak: HeapPeak, +} + +pub(crate) fn start() -> Profilers { + let profilers = Profilers { + #[cfg(feature = "cpu-profile")] + cpu: pprof::ProfilerGuardBuilder::default() + .frequency(99) + .blocklist(&["libc", "libgcc", "pthread", "vdso"]) + .build() + .inspect_err(|e| tracing::warn!("cpu profiler not started: {e}")) + .ok(), + #[cfg(feature = "heap-profile")] + heap_peak: HeapPeak::default(), + }; + #[cfg(feature = "heap-profile")] + { + let peak = profilers.heap_peak.clone(); + std::thread::spawn(move || { + let Some(ctl) = jemalloc_pprof::PROF_CTL.as_ref() else { + tracing::warn!("jemalloc heap profiling is not enabled"); + return; + }; + let mut dumped_kb = 0; + while let Some(rss_kb) = crate::proc_status_kb("VmRSS:") { + if rss_kb > dumped_kb + dumped_kb / 20 { + match ctl.blocking_lock().dump_profile() { + Ok(profile) => { + dumped_kb = rss_kb; + if let Ok(mut slot) = peak.lock() { + *slot = Some((rss_kb, profile)); + } + } + Err(e) => { + tracing::warn!("heap snapshot failed: {e}"); + return; + } + } + } + std::thread::sleep(std::time::Duration::from_millis(500)); + } + }); + } + profilers +} + +impl Profilers { + #[allow(unused_variables)] + pub(crate) fn finish(self, dir: &Path) { + #[cfg(feature = "heap-profile")] + if let Some((rss_kb, profile)) = self.heap_peak.lock().ok().and_then(|mut slot| slot.take()) + { + let mut opts = jemalloc_pprof::FlamegraphOptions::default(); + opts.title = format!("dash-spv live heap at the RSS peak ({} MiB)", rss_kb / 1024); + opts.count_name = "bytes".to_string(); + let written = + profile.to_flamegraph(&mut opts).map_err(|e| e.to_string()).and_then(|svg| { + std::fs::write(dir.join("heap-peak.svg"), svg).map_err(|e| e.to_string()) + }); + if let Err(e) = written { + tracing::warn!("could not write heap-peak.svg: {e}"); + } + } + #[cfg(feature = "cpu-profile")] + if let Some(report) = self.cpu.as_ref().and_then(|guard| guard.report().build().ok()) { + let written = std::fs::File::create(dir.join("flamegraph.svg")) + .map_err(|e| e.to_string()) + .and_then(|file| report.flamegraph(file).map_err(|e| e.to_string())); + if let Err(e) = written { + tracing::warn!("could not write flamegraph.svg: {e}"); + } + } + } +} From 6e9bce955b68fb6c834524b24083cb738f8106c0 Mon Sep 17 00:00:00 2001 From: Borja Castellano Date: Mon, 14 Sep 2026 16:18:35 +0000 Subject: [PATCH 2/8] refactor(dash-spv): drop the committed-height segment release Reverts ae2a5d6f (#946). Segments were released from memory only when a sync manager remembered to call `set_committed_height` on the block and filter storages, and only on the next timed persist. The resident set should be bounded by the segment cache itself, not by callers reporting progress to storage. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_017DruChNTWXwJoWPartZwCf --- dash-spv/src/storage/blocks.rs | 58 ----- dash-spv/src/storage/filters.rs | 15 -- dash-spv/src/storage/mod.rs | 8 - dash-spv/src/storage/segments.rs | 337 --------------------------- dash-spv/src/sync/blocks/manager.rs | 18 -- dash-spv/src/sync/filters/manager.rs | 12 - 6 files changed, 448 deletions(-) diff --git a/dash-spv/src/storage/blocks.rs b/dash-spv/src/storage/blocks.rs index 34b90a76f..27324d5c9 100644 --- a/dash-spv/src/storage/blocks.rs +++ b/dash-spv/src/storage/blocks.rs @@ -33,17 +33,6 @@ pub trait BlockStorage: Send + Sync + 'static { /// A crash between `truncate_above` and `persist` may leave orphaned segment /// files on disk and cause the storage to reopen at the pre-truncation tip. async fn truncate_above(&mut self, target_height: CoreBlockHeight) -> StorageResult<()>; - - /// Declare the highest block height that has been applied to every - /// interested wallet, allowing the storage to stop holding those block - /// bodies in memory. - /// - /// Blocks at or below this height stay readable: `load_block` reloads them - /// from disk on demand. The watermark may move backwards when a rescan - /// re-processes lower heights. - /// - /// Defaults to a no-op for implementations that hold no in-memory cache. - async fn set_committed_height(&mut self, _height: CoreBlockHeight) {} } /// Persistent storage for full blocks using segmented files. @@ -92,10 +81,6 @@ impl BlockStorage for PersistentBlockStorage { async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()> { self.blocks.write().await.truncate_above(target_height).await } - - async fn set_committed_height(&mut self, height: u32) { - self.blocks.write().await.set_committed_height(height); - } } #[cfg(test)] @@ -158,49 +143,6 @@ mod tests { assert_eq!(storage.load_block(3).await.unwrap(), None); } - /// Block bodies are the dominant memory consumer during a long backfill. - /// Once applied, they must leave memory while staying loadable from disk — - /// `handle_sync_event` re-reads stored blocks by height on resume/rescan. - #[tokio::test] - async fn test_committed_blocks_are_released_but_still_loadable() { - let temp_dir = TempDir::new().unwrap(); - let mut storage = PersistentBlockStorage::open(temp_dir.path()).await.unwrap(); - - // One block in each of segments 0, 1 and 2 (50_000 heights per segment). - // Carry real transaction payloads so the reload proves the block bodies - // round-trip, not merely that a slot is occupied. - let txs = vec![dashcore::Transaction::dummy_empty()]; - let low = HashedBlock::dummy(10, txs.clone()); - let mid = HashedBlock::dummy(50_010, txs.clone()); - let tip = HashedBlock::dummy(100_010, txs); - - storage.store_block(10, low.clone()).await.unwrap(); - storage.store_block(50_010, mid.clone()).await.unwrap(); - storage.store_block(100_010, tip.clone()).await.unwrap(); - storage.persist(temp_dir.path()).await.unwrap(); - - assert_eq!(storage.blocks.read().await.resident_segment_ids(), vec![0, 1, 2]); - - // Everything below segment 2 has been applied to the wallets. - storage.set_committed_height(99_999).await; - storage.persist(temp_dir.path()).await.unwrap(); - - // Segments 0 and 1 are gone from memory; the frontier segment stays. - assert_eq!( - storage.blocks.read().await.resident_segment_ids(), - vec![2], - "applied block segments must be released" - ); - - // All three blocks still load, byte-identically, via the disk fallback. - assert_eq!(storage.load_block(10).await.unwrap(), Some(low)); - assert_eq!(storage.load_block(50_010).await.unwrap(), Some(mid)); - assert_eq!(storage.load_block(100_010).await.unwrap(), Some(tip)); - - // Gaps inside a released segment still report absent, not sentinel data. - assert_eq!(storage.load_block(11).await.unwrap(), None); - } - #[tokio::test] async fn test_returns_none_for_gaps() { let temp_dir = TempDir::new().unwrap(); diff --git a/dash-spv/src/storage/filters.rs b/dash-spv/src/storage/filters.rs index 249b966b9..3578a26ea 100644 --- a/dash-spv/src/storage/filters.rs +++ b/dash-spv/src/storage/filters.rs @@ -51,17 +51,6 @@ pub trait FilterStorage: Send + Sync + 'static { /// A crash between `truncate_above` and `persist` may leave orphaned segment /// files on disk and cause the storage to reopen at the pre-truncation tip. async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()>; - - /// Declare the highest filter height that has been scanned and committed - /// for every wallet, allowing the storage to stop holding those filters in - /// memory. - /// - /// Filters at or below this height stay readable: `load_filters` reloads - /// them from disk on demand. The watermark may move backwards when a - /// rescan rolls the scan position back. - /// - /// Defaults to a no-op for implementations that hold no in-memory cache. - async fn set_committed_height(&mut self, _height: u32) {} } pub struct PersistentFilterStorage { @@ -121,10 +110,6 @@ impl FilterStorage for PersistentFilterStorage { async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()> { self.filters.write().await.truncate_above(target_height).await } - - async fn set_committed_height(&mut self, height: u32) { - self.filters.write().await.set_committed_height(height); - } } #[cfg(test)] diff --git a/dash-spv/src/storage/mod.rs b/dash-spv/src/storage/mod.rs index 70a851acb..248ab10d0 100644 --- a/dash-spv/src/storage/mod.rs +++ b/dash-spv/src/storage/mod.rs @@ -386,10 +386,6 @@ impl filters::FilterStorage for DiskStorageManager { async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()> { self.filters.write().await.truncate_above(target_height).await } - - async fn set_committed_height(&mut self, height: u32) { - self.filters.write().await.set_committed_height(height).await; - } } #[async_trait] @@ -405,10 +401,6 @@ impl BlockStorage for DiskStorageManager { async fn truncate_above(&mut self, target_height: u32) -> StorageResult<()> { self.blocks.write().await.truncate_above(target_height).await } - - async fn set_committed_height(&mut self, height: u32) { - self.blocks.write().await.set_committed_height(height).await; - } } #[async_trait] diff --git a/dash-spv/src/storage/segments.rs b/dash-spv/src/storage/segments.rs index ec900899c..f76566aa0 100644 --- a/dash-spv/src/storage/segments.rs +++ b/dash-spv/src/storage/segments.rs @@ -83,14 +83,6 @@ pub struct SegmentCache { /// Segment ids whose backing files must be removed on the next `persist`. /// Populated by `truncate_above` for segments that are dropped entirely. to_delete: HashSet, - /// Highest height whose items are durably persisted *and* fully consumed - /// by every reader, as declared by the owning sync manager through - /// [`SegmentCache::set_committed_height`]. - /// - /// `None` — the default — disables committed-height release entirely, so a - /// cache whose owner never declares a watermark keeps exactly the residency - /// behavior it had before this field existed. - committed_height: Option, } impl SegmentCache { @@ -106,7 +98,6 @@ impl SegmentCache { start_height: None, segments_dir: segments_dir.clone(), to_delete: HashSet::new(), - committed_height: None, }; // Building the metadata @@ -454,15 +445,6 @@ impl SegmentCache { self.tip_height = Some(target_height); - // The watermark must never claim heights that no longer exist: a - // rescan re-reads this range, and a stale high watermark would keep - // releasing segments it is actively refilling. - if let Some(committed) = self.committed_height { - if committed > target_height { - self.committed_height = Some(target_height); - } - } - Ok(()) } @@ -511,8 +493,6 @@ impl SegmentCache { self.evicted.clear(); self.tip_height = None; self.start_height = None; - // Nothing is stored, so nothing is committed. - self.committed_height = None; Ok(()) } @@ -547,123 +527,6 @@ impl SegmentCache { tracing::error!("Failed to persist segment with id {id}: {e}"); } } - - // After the writes above, so segments cleaned by *this* pass are - // eligible immediately. A segment whose persist failed is still Dirty - // and is therefore skipped. - let released = self.release_committed_segments(); - if released > 0 { - tracing::debug!( - "SegmentCache: released {} committed segment(s) below height {:?}; {} resident", - released, - self.committed_height, - self.segments.len(), - ); - } - } - - /// Declare the highest height whose items are durably persisted and fully - /// consumed, so the cache may stop holding them in memory. - /// - /// Segments lying *entirely* at or below `height` are released from memory - /// by the next [`SegmentCache::persist`], once they are `Clean`. The - /// on-disk file remains the source of truth and a later read of a released - /// height transparently reloads the segment through the same lazy path used - /// for any never-resident segment — the data a reader observes is - /// unchanged, only the resident-set size is. - /// - /// The watermark may move **backwards**: a wallet rescan rolls it back - /// before re-reading lower heights. A regression only ever narrows the - /// release window, so it is always safe. Callers therefore assign rather - /// than accumulate, and no monotonicity is enforced here. - pub fn set_committed_height(&mut self, height: u32) { - self.committed_height = Some(height); - } - - /// The committed-height watermark, or `None` when the owner has never - /// declared one (in which case no committed-height release occurs). - /// - /// Test-only for now — nothing reads the watermark back in production. - /// Promote it to an unconditional accessor when a caller needs it. - #[cfg(test)] - #[inline] - pub fn committed_height(&self) -> Option { - self.committed_height - } - - /// Sorted ids of the segments currently held in memory. Test-only hook for - /// asserting residency from the storage wrappers. - #[cfg(test)] - pub(crate) fn resident_segment_ids(&self) -> Vec { - let mut ids: Vec = self.segments.keys().copied().collect(); - ids.sort_unstable(); - ids - } - - /// Release the in-memory items of every fully-committed clean segment, - /// keeping the on-disk file as the source of truth. - /// - /// This is what stops a long backfill from pinning every segment it ever - /// touched. `MAX_ACTIVE_SEGMENTS` alone cannot: a scan spanning fewer than - /// ten segments never trips the LRU, so every downloaded item stays - /// resident until the process dies. - /// - /// Releasing a `Clean` segment is invisible to readers. `Clean` is reachable - /// only via [`Segment::load`] from an existing file or a successful - /// [`Segment::persist`], so a clean segment's items are byte-identical to - /// its backing file, and `get_segment_mut` reloads exactly those bytes on - /// the next access. - /// - /// Three classes are deliberately kept: - /// - `Dirty` segments — their contents exist *only* in memory, so dropping - /// them would lose data outright. - /// - The segment holding the sync frontier (`tip_height`) — it is still - /// being written, and releasing it would force a reload on the very next - /// store. - /// - Any segment not entirely at or below the watermark, so a partially - /// committed segment is never released out from under an active reader. - /// - /// Segments queued in `to_delete` cannot appear here: both `truncate_above` - /// and `clear` remove a segment from `self.segments` in the same step that - /// queues it, so the two sets are disjoint by construction. - /// - /// Returns the number of segments released. - fn release_committed_segments(&mut self) -> usize { - let Some(committed) = self.committed_height else { - return 0; - }; - - // Still-written frontier; never release it. - let frontier_segment = self.tip_height.map(Self::height_to_segment_id); - - let items_per_segment = Segment::::ITEMS_PER_SEGMENT as u64; - let committed = committed as u64; - - // Widened to u64 so a segment id near u32::MAX cannot overflow the - // range arithmetic; such a segment simply fails the comparison. - let releasable: Vec = self - .segments - .iter() - .filter(|(id, segment)| { - if segment.state != SegmentState::Clean { - return false; - } - - if Some(**id) == frontier_segment { - return false; - } - - let last_height = (**id as u64) * items_per_segment + items_per_segment - 1; - last_height <= committed - }) - .map(|(id, _)| *id) - .collect(); - - for id in &releasable { - self.segments.remove(id); - } - - releasable.len() } #[inline] @@ -1444,206 +1307,6 @@ mod tests { ); } - /// The headline guarantee: a committed, clean segment is dropped from - /// memory on persist, and a later read transparently reloads it from disk - /// byte-identically. - /// - /// This is the leak that pinned a 350k-block backfill in RAM: the scan - /// spans fewer than `MAX_ACTIVE_SEGMENTS`, so the LRU never fires and every - /// item stays resident until the process dies. - #[tokio::test] - async fn test_release_committed_segments_reloads_from_disk() { - let tmp_dir = TempDir::new().unwrap(); - - const ITEMS_PER_SEGMENT: u32 = Segment::::ITEMS_PER_SEGMENT; - - // Dense across segments 0 and 1, plus a few items into segment 2 so the - // frontier lives above the range we expect to be released. - let items = FilterHeader::dummy_batch(0..ITEMS_PER_SEGMENT * 2 + 5); - - let mut cache = SegmentCache::::load_or_new(tmp_dir.path()).await.unwrap(); - cache.store_items_at_height(&items, 0).await.unwrap(); - cache.persist(tmp_dir.path()).await; - - // Baseline: without a watermark nothing is released, which is exactly - // the pre-fix behavior. - assert_eq!(cache.committed_height(), None); - assert_eq!(cache.segments.len(), 3, "no watermark declared, so nothing is released"); - - // Segments 0 and 1 are now fully committed (their last heights are - // ITEMS_PER_SEGMENT-1 and 2*ITEMS_PER_SEGMENT-1). - cache.set_committed_height(ITEMS_PER_SEGMENT * 2 - 1); - cache.persist(tmp_dir.path()).await; - - assert!(!cache.segments.contains_key(&0), "committed clean segment 0 must be released"); - assert!(!cache.segments.contains_key(&1), "committed clean segment 1 must be released"); - assert!(cache.segments.contains_key(&2), "frontier segment must stay resident"); - assert_eq!(cache.segments.len(), 1); - - // The cache-level watermarks are unaffected by residency. - assert_eq!(cache.start_height(), Some(0)); - assert_eq!(cache.tip_height(), Some(ITEMS_PER_SEGMENT * 2 + 4)); - - // Reading the released range reloads both segments from disk and - // returns exactly the bytes that were written. - let reread = cache.get_items(0..ITEMS_PER_SEGMENT * 2).await.unwrap(); - assert_eq!(reread, items[0..(ITEMS_PER_SEGMENT * 2) as usize]); - - // Single-item reads across the boundary agree too. - assert_eq!(cache.get_item(0).await.unwrap(), Some(items[0])); - assert_eq!( - cache.get_item(ITEMS_PER_SEGMENT).await.unwrap(), - Some(items[ITEMS_PER_SEGMENT as usize]) - ); - } - - /// Never release a segment that is dirty (its items exist only in memory), - /// the frontier segment (still being written), or a segment only partially - /// below the watermark. - #[tokio::test] - async fn test_release_skips_dirty_frontier_and_partial_segments() { - let tmp_dir = TempDir::new().unwrap(); - - const ITEMS_PER_SEGMENT: u32 = Segment::::ITEMS_PER_SEGMENT; - - let mut cache = SegmentCache::::load_or_new(tmp_dir.path()).await.unwrap(); - - // Sparse writes into segments 0, 1 and 2. - cache.store_items_at_height(&FilterHeader::dummy_batch(0..1), 10).await.unwrap(); - cache - .store_items_at_height(&FilterHeader::dummy_batch(1..2), ITEMS_PER_SEGMENT + 10) - .await - .unwrap(); - cache - .store_items_at_height(&FilterHeader::dummy_batch(2..3), ITEMS_PER_SEGMENT * 2 + 10) - .await - .unwrap(); - - // A watermark above everything, but nothing has been persisted yet, so - // all three segments are Dirty and none may be released. - cache.set_committed_height(u32::MAX); - let released = cache.release_committed_segments(); - assert_eq!(released, 0, "dirty segments must never be released"); - assert_eq!(cache.segments.len(), 3); - - cache.persist(tmp_dir.path()).await; - // persist() cleaned all three, then released every non-frontier one. - assert_eq!(cache.segments.keys().copied().collect::>(), vec![2]); - - // Now verify the partial-coverage rule: a watermark inside segment 0 - // does not release it, because its upper heights are not yet committed. - let mut cache = SegmentCache::::load_or_new(tmp_dir.path()).await.unwrap(); - let _ = cache.get_segment_mut(&0).await.unwrap(); - assert!(cache.segments.contains_key(&0)); - - cache.set_committed_height(ITEMS_PER_SEGMENT - 2); // one short of the segment's last height - assert_eq!(cache.release_committed_segments(), 0, "partially committed segment must stay"); - assert!(cache.segments.contains_key(&0)); - - cache.set_committed_height(ITEMS_PER_SEGMENT - 1); // exactly the last height - assert_eq!(cache.release_committed_segments(), 1, "fully committed segment is releasable"); - assert!(!cache.segments.contains_key(&0)); - } - - /// The rescan path re-reads heights far below the watermark. Those segments - /// were released, so this exercises the lazy reload under the access - /// pattern `reset_for_rescan` / `start_download` produce. - #[tokio::test] - async fn test_released_segments_serve_a_rescan_reread() { - let tmp_dir = TempDir::new().unwrap(); - - const ITEMS_PER_SEGMENT: u32 = Segment::::ITEMS_PER_SEGMENT; - - let items = FilterHeader::dummy_batch(0..ITEMS_PER_SEGMENT * 2 + 5); - - let mut cache = SegmentCache::::load_or_new(tmp_dir.path()).await.unwrap(); - cache.store_items_at_height(&items, 0).await.unwrap(); - cache.set_committed_height(ITEMS_PER_SEGMENT * 2 - 1); - cache.persist(tmp_dir.path()).await; - assert_eq!(cache.segments.len(), 1); - - // A wallet appears behind the scan: the manager rolls the watermark - // back and re-reads from a low height. - cache.set_committed_height(0); - let rescanned = cache.get_items(5..ITEMS_PER_SEGMENT + 5).await.unwrap(); - assert_eq!(rescanned, items[5..(ITEMS_PER_SEGMENT + 5) as usize]); - - // With the watermark rolled back, the re-read segments are retained - // rather than being dropped again underneath the rescan. - cache.persist(tmp_dir.path()).await; - assert!(cache.segments.contains_key(&0), "rolled-back watermark must stop re-release"); - - // And the data still round-trips after a full reopen. - let mut reloaded = SegmentCache::::load_or_new(tmp_dir.path()).await.unwrap(); - assert_eq!(reloaded.tip_height(), Some(ITEMS_PER_SEGMENT * 2 + 4)); - assert_eq!( - reloaded.get_items(0..ITEMS_PER_SEGMENT * 2 + 5).await.unwrap(), - items, - "released-then-reloaded data must survive a process restart byte-identically" - ); - } - - /// The watermark must never outlive the data it refers to: `truncate_above` - /// clamps it and `clear` drops it, so a later scan of the same range is not - /// released out from under itself. - #[tokio::test] - async fn test_committed_height_follows_truncate_and_clear() { - let tmp_dir = TempDir::new().unwrap(); - - let items = FilterHeader::dummy_batch(0..30); - - let mut cache = SegmentCache::::load_or_new(tmp_dir.path()).await.unwrap(); - cache.store_items_at_height(&items, 0).await.unwrap(); - - cache.set_committed_height(25); - assert_eq!(cache.committed_height(), Some(25)); - - // Truncating above a height below the watermark clamps it. - cache.truncate_above(10).await.unwrap(); - assert_eq!(cache.committed_height(), Some(10)); - - // Truncating above the watermark leaves it alone. - cache.set_committed_height(5); - cache.truncate_above(8).await.unwrap(); - assert_eq!(cache.committed_height(), Some(5)); - - cache.clear().unwrap(); - assert_eq!(cache.committed_height(), None, "an empty cache has nothing committed"); - } - - /// A released segment must still accept new writes into its unused slots, - /// reloading the existing contents first rather than silently starting from - /// a blank segment (which would drop the persisted items on the next write). - #[tokio::test] - async fn test_store_into_released_segment_preserves_existing_items() { - let tmp_dir = TempDir::new().unwrap(); - - const ITEMS_PER_SEGMENT: u32 = Segment::::ITEMS_PER_SEGMENT; - - let mut cache = SegmentCache::::load_or_new(tmp_dir.path()).await.unwrap(); - cache.store_items_at_height(&FilterHeader::dummy_batch(0..10), 0).await.unwrap(); - cache - .store_items_at_height(&FilterHeader::dummy_batch(50..55), ITEMS_PER_SEGMENT) - .await - .unwrap(); - - cache.set_committed_height(ITEMS_PER_SEGMENT - 1); - cache.persist(tmp_dir.path()).await; - assert!(!cache.segments.contains_key(&0), "segment 0 released"); - - // Write into a gap in the released segment. - cache.store_items_at_height(&FilterHeader::dummy_batch(90..95), 100).await.unwrap(); - - // Both the reloaded originals and the new items are present. - assert_eq!(cache.get_items(0..10).await.unwrap(), FilterHeader::dummy_batch(0..10)); - assert_eq!(cache.get_items(100..105).await.unwrap(), FilterHeader::dummy_batch(90..95)); - - cache.persist(tmp_dir.path()).await; - let mut reloaded = SegmentCache::::load_or_new(tmp_dir.path()).await.unwrap(); - assert_eq!(reloaded.get_items(0..10).await.unwrap(), FilterHeader::dummy_batch(0..10)); - assert_eq!(reloaded.get_items(100..105).await.unwrap(), FilterHeader::dummy_batch(90..95)); - } - #[test] fn test_segment_insert_get() { let segment_id = 10; diff --git a/dash-spv/src/sync/blocks/manager.rs b/dash-spv/src/sync/blocks/manager.rs index e4a1464d8..feb428da6 100644 --- a/dash-spv/src/sync/blocks/manager.rs +++ b/dash-spv/src/sync/blocks/manager.rs @@ -77,9 +77,6 @@ impl BlocksManager SyncResult> { let mut events = Vec::new(); - // Highest height applied in this drain, used below to advance the - // storage's committed watermark exactly once. - let mut last_applied: Option = None; // Process blocks in height order using pipeline's ordering logic while let Some((block, height, interested)) = self.pipeline.take_next_ordered_block() { @@ -127,7 +124,6 @@ impl BlocksManager BlocksManager= batch_end { @@ -759,13 +754,6 @@ impl Date: Mon, 14 Sep 2026 20:56:49 +0000 Subject: [PATCH 3/8] feat(dash-spv): trace segment cache hits and misses Every segment request now logs at trace level whether it was served from a resident segment or, on a miss, where the segment came from (`evicted`, `disk` or `new`). Enable it per module with RUST_LOG=dash_spv::storage::segments=trace. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_017DruChNTWXwJoWPartZwCf --- dash-spv/src/storage/segments.rs | 22 ++++++++++++++++++---- 1 file changed, 18 insertions(+), 4 deletions(-) diff --git a/dash-spv/src/storage/segments.rs b/dash-spv/src/storage/segments.rs index f76566aa0..57c7339e0 100644 --- a/dash-spv/src/storage/segments.rs +++ b/dash-spv/src/storage/segments.rs @@ -177,6 +177,10 @@ impl SegmentCache { let segments_len = self.segments.len(); if self.segments.contains_key(segment_id) { + tracing::trace!( + "SegmentCache<{}>: segment {segment_id} cache hit", + std::any::type_name::() + ); let segment = self.segments.get_mut(segment_id).expect("We already checked that it exists"); return Ok(segment); @@ -199,13 +203,23 @@ impl SegmentCache { // If the segment is queued for deletion, return a fresh empty segment. // The next `persist` will atomically overwrite the stale file. // Otherwise, load it from disk. - let segment = if let Some(segment) = self.evicted.remove(segment_id) { - segment + let (segment, source) = if let Some(segment) = self.evicted.remove(segment_id) { + (segment, "evicted") } else if self.to_delete.remove(segment_id) { - Segment::new(*segment_id, vec![], SegmentState::Dirty) + (Segment::new(*segment_id, vec![], SegmentState::Dirty), "new") } else { - Segment::load(&self.segments_dir, *segment_id).await? + let segment = Segment::load(&self.segments_dir, *segment_id).await?; + let source = if segment.state == SegmentState::Clean { + "disk" + } else { + "new" + }; + (segment, source) }; + tracing::trace!( + "SegmentCache<{}>: segment {segment_id} cache miss ({source})", + std::any::type_name::() + ); let segment = self.segments.entry(*segment_id).or_insert(segment); Ok(segment) From 99a4eeb8a9c1664ea529a35513218f05606127c5 Mon Sep 17 00:00:00 2001 From: Borja Castellano Date: Mon, 14 Sep 2026 21:00:47 +0000 Subject: [PATCH 4/8] perf(dash-spv): keep at most two segments resident per storage cache With the committed-height release gone, the segment cache itself has to bound the resident set. Ten 50 000-item segments per cache kept most of a restore in memory; two keep the peak below what the release achieved, without costing time. Mainnet restore, mainnet.100mbi.100ms, #1015/#1016/#1014 applied, jemalloc heap profiling, wallet identical in every run (14114383 sat, 7112 records, 13389 addresses): segments peak RSS time segment loads from disk (headers/filter headers/filters/blocks) 10 + #946 982 MiB 8.2 min - 10 1490 MiB 8.8 min 84 / 2 / 28 / 0 2 930 MiB 8.8 min 212 / 103 / 74 / 0 1 1000 MiB 7.7 min 427 / 742 / 156 / 33 Each cache served ~2.34M requests from memory in every run. One segment starts reloading block segments of up to 87 MB from disk, which raises the peak again. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_017DruChNTWXwJoWPartZwCf --- dash-spv/src/storage/segments.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dash-spv/src/storage/segments.rs b/dash-spv/src/storage/segments.rs index 57c7339e0..a61647caa 100644 --- a/dash-spv/src/storage/segments.rs +++ b/dash-spv/src/storage/segments.rs @@ -86,7 +86,7 @@ pub struct SegmentCache { } impl SegmentCache { - const MAX_ACTIVE_SEGMENTS: usize = 10; + const MAX_ACTIVE_SEGMENTS: usize = 2; pub async fn load_or_new(segments_dir: impl Into) -> StorageResult { let segments_dir = segments_dir.into(); From 78a3d8109f68c9aed684cfc59c84b9aa2f334e55 Mon Sep 17 00:00:00 2001 From: Borja Castellano Date: Mon, 14 Sep 2026 22:24:29 +0000 Subject: [PATCH 5/8] perf(dash-spv): persist a dirty segment when it is evicted A dirty segment leaving the resident set used to wait in `evicted`, in memory, until the next 5 s storage tick wrote it. The resident limit was therefore not a limit: every segment evicted between two ticks stayed in memory. Now eviction writes the segment first and only then drops it; if the write fails the segment stays resident and the error is returned. The `evicted` map is gone, and the tick only persists the resident segments. Mainnet restore, mainnet.100mbi.100ms, #1015/#1016/#1014 applied, two resident segments: 7.7 and 7.9 min, peak RSS 943 and 948 MiB, wallet identical (14114383 sat, 7112 records, 13389 addresses). No cache misses served from `evicted` any more; header, filter header and filter segments are reloaded from disk about as often as they used to come back from `evicted` (~270 / ~150 / ~90), block segments never. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_017DruChNTWXwJoWPartZwCf --- dash-spv/src/storage/segments.rs | 40 +++++++++++--------------------- 1 file changed, 13 insertions(+), 27 deletions(-) diff --git a/dash-spv/src/storage/segments.rs b/dash-spv/src/storage/segments.rs index a61647caa..0a9b29991 100644 --- a/dash-spv/src/storage/segments.rs +++ b/dash-spv/src/storage/segments.rs @@ -76,7 +76,6 @@ impl Persistable for HashedBlock { #[derive(Debug)] pub struct SegmentCache { segments: HashMap>, - evicted: HashMap>, tip_height: Option, start_height: Option, segments_dir: PathBuf, @@ -93,7 +92,6 @@ impl SegmentCache { let mut cache = Self { segments: HashMap::with_capacity(Self::MAX_ACTIVE_SEGMENTS), - evicted: HashMap::new(), tip_height: None, start_height: None, segments_dir: segments_dir.clone(), @@ -188,24 +186,20 @@ impl SegmentCache { if segments_len >= Self::MAX_ACTIVE_SEGMENTS { let key_to_evict = - self.segments.iter_mut().min_by_key(|(_, s)| s.last_accessed).map(|(k, v)| (*k, v)); + self.segments.iter().min_by_key(|(_, s)| s.last_accessed).map(|(k, _)| *k); - if let Some((key, _)) = key_to_evict { - if let Some(segment) = self.segments.remove(&key) { - if segment.state == SegmentState::Dirty { - self.evicted.insert(key, segment); - } + if let Some(key) = key_to_evict { + if let Some(segment) = self.segments.get_mut(&key) { + segment.persist(&self.segments_dir).await?; } + self.segments.remove(&key); } } - // If the segment is already in the to_persist map, load it from there. // If the segment is queued for deletion, return a fresh empty segment. // The next `persist` will atomically overwrite the stale file. // Otherwise, load it from disk. - let (segment, source) = if let Some(segment) = self.evicted.remove(segment_id) { - (segment, "evicted") - } else if self.to_delete.remove(segment_id) { + let (segment, source) = if self.to_delete.remove(segment_id) { (Segment::new(*segment_id, vec![], SegmentState::Dirty), "new") } else { let segment = Segment::load(&self.segments_dir, *segment_id).await?; @@ -453,7 +447,6 @@ impl SegmentCache { for segment_id in (boundary_segment_id + 1)..=max_segment_id { self.segments.remove(&segment_id); - self.evicted.remove(&segment_id); self.to_delete.insert(segment_id); } @@ -495,16 +488,14 @@ impl SegmentCache { Err(e) => return Err(StorageError::Io(e)), } - // Scan succeeded — now it is safe to mutate cache state. Resident and - // evicted segments may be dirty and not persisted yet, so the scan - // above cannot see them. Queue their ids too; `persist` ignores - // missing files. + // Scan succeeded — now it is safe to mutate cache state. Resident + // segments may be dirty and not persisted yet, so the scan above + // cannot see them. Queue their ids too; `persist` ignores missing + // files. to_delete.extend(self.segments.keys().copied()); - to_delete.extend(self.evicted.keys().copied()); self.to_delete.extend(to_delete); self.segments.clear(); - self.evicted.clear(); self.tip_height = None; self.start_height = None; @@ -528,14 +519,6 @@ impl SegmentCache { } self.to_delete.extend(failed); - for (id, segments) in self.evicted.iter_mut() { - if let Err(e) = segments.persist(&segments_dir).await { - tracing::error!("Failed to persist segment with id {id}: {e}"); - } - } - - self.evicted.clear(); - for (id, segments) in self.segments.iter_mut() { if let Err(e) = segments.persist(&segments_dir).await { tracing::error!("Failed to persist segment with id {id}: {e}"); @@ -781,11 +764,14 @@ mod tests { segment.insert(FilterHeader::dummy(i), 0); } + assert!(tmp_dir.path().join(FilterHeader::segment_file_name(0)).exists()); + for i in 0..=MAX_SEGMENTS { assert_eq!(cache.segments.len(), MAX_SEGMENTS as usize); let segment = cache.get_segment_mut(&i).await.expect("Failed to load segment"); + assert!(segment.state == SegmentState::Clean); assert_eq!(segment.get(0..1), [FilterHeader::dummy(i)]); } } From 6ec081bbb616f904e29f65ae64b54bd2ec022e02 Mon Sep 17 00:00:00 2001 From: Borja Castellano Date: Mon, 14 Sep 2026 23:02:32 +0000 Subject: [PATCH 6/8] perf(dash-spv): stream segments to disk instead of encoding them into one buffer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `Segment::persist` encoded the whole segment into a `Vec` before handing it to `atomic_write`. For a block segment near the tip (up to 87 MB on disk) that `Vec` grew to 128 MiB, and since dirty segments are now written on eviction this happened in the middle of the sync. `atomic_write_items` encodes one item at a time through a 1 MiB `BufWriter` into the temporary file, then syncs and renames it as before. `atomic_write` keeps its behaviour and shares the temporary file and rename logic. Mainnet restore, mainnet.100mbi.100ms, #1015/#1016/#1014 applied, two resident segments, wallet identical in every run (14114383 sat, 7112 records, 13389 addresses): whole-segment buffer peak RSS 943 / 948 MiB 7.7 / 7.9 min streamed peak RSS 845 / 855 / 856 MiB 9.4 / 7.7 / 6.5 min The persist buffer, 68–133 MiB in the earlier heap snapshots, no longer shows up. The 9.4 min run lost time to peers in the header and filter header phases. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_017DruChNTWXwJoWPartZwCf --- dash-spv/src/storage/io.rs | 56 +++++++++++++++++++++++++++----- dash-spv/src/storage/segments.rs | 12 ++----- 2 files changed, 49 insertions(+), 19 deletions(-) diff --git a/dash-spv/src/storage/io.rs b/dash-spv/src/storage/io.rs index 9854d5ab0..c965fbd30 100644 --- a/dash-spv/src/storage/io.rs +++ b/dash-spv/src/storage/io.rs @@ -3,7 +3,8 @@ use std::path::{Path, PathBuf}; use crate::error::{StorageError, StorageResult}; -use tokio::io::AsyncWriteExt; +use dashcore::consensus::Encodable; +use tokio::io::{AsyncWriteExt, BufWriter}; /// Get the temporary file path for atomic writes. /// Uses process ID and a counter to ensure uniqueness even with concurrent writes. @@ -22,6 +23,38 @@ fn get_temp_path(path: &Path) -> PathBuf { /// Atomically write data to a file. /// Uses temporary file + sync + rename pattern for crash resilience. pub(crate) async fn atomic_write(path: &Path, data: &[u8]) -> StorageResult<()> { + atomic_write_with(path, |mut file| async move { + file.write_all(data).await?; + file.sync_all().await + }) + .await +} + +/// Like [`atomic_write`], but encodes `items` into the file one at a time +/// instead of into a buffer holding all of them. +pub(crate) async fn atomic_write_items( + path: &Path, + items: &[I], +) -> StorageResult<()> { + atomic_write_with(path, |file| async move { + let mut writer = BufWriter::with_capacity(1 << 20, file); + let mut encoded = Vec::new(); + for item in items { + encoded.clear(); + item.consensus_encode(&mut encoded)?; + writer.write_all(&encoded).await?; + } + writer.flush().await?; + writer.into_inner().sync_all().await + }) + .await +} + +async fn atomic_write_with(path: &Path, write: F) -> StorageResult<()> +where + F: FnOnce(tokio::fs::File) -> Fut, + Fut: std::future::Future>, +{ // Ensure parent directory exists if let Some(parent) = path.parent() { tokio::fs::create_dir_all(parent) @@ -32,14 +65,7 @@ pub(crate) async fn atomic_write(path: &Path, data: &[u8]) -> StorageResult<()> let temp_path = get_temp_path(path); // Write to temporary file - let write_result = async { - let mut file = tokio::fs::File::create(&temp_path).await?; - file.write_all(data).await?; - file.sync_all().await?; - - Ok::<(), std::io::Error>(()) - } - .await; + let write_result = async { write(tokio::fs::File::create(&temp_path).await?).await }.await; // Clean up temp file on error if let Err(e) = write_result { @@ -62,6 +88,18 @@ mod tests { use std::fs; use tempfile::TempDir; + #[tokio::test] + async fn test_atomic_write_items_matches_encoding_all_at_once() { + let temp_dir = TempDir::new().unwrap(); + let path = temp_dir.path().join("items.dat"); + let items = vec![vec![1u8, 2, 3], vec![], vec![4u8; 300]]; + + atomic_write_items(&path, &items).await.unwrap(); + + let expected: Vec = items.iter().flat_map(dashcore::consensus::serialize).collect(); + assert_eq!(fs::read(&path).unwrap(), expected); + } + #[test] fn test_get_temp_path_uniqueness() { let path = Path::new("some").join("path").join("file.dat"); diff --git a/dash-spv/src/storage/segments.rs b/dash-spv/src/storage/segments.rs index 0a9b29991..2717857b1 100644 --- a/dash-spv/src/storage/segments.rs +++ b/dash-spv/src/storage/segments.rs @@ -19,7 +19,7 @@ use dashcore_hashes::Hash; use crate::{ error::StorageResult, - storage::io::atomic_write, + storage::io::atomic_write_items, types::{HashedBlock, HashedBlockHeader}, StorageError, }; @@ -648,15 +648,7 @@ impl Segment { return Err(StorageError::WriteFailed(format!("Failed to persist segment: {}", e))); } - let mut buffer = Vec::new(); - - for item in self.items.iter() { - item.consensus_encode(&mut buffer).map_err(|e| { - StorageError::WriteFailed(format!("Failed to encode segment item: {}", e)) - })?; - } - - atomic_write(&path, &buffer).await?; + atomic_write_items(&path, &self.items).await?; self.state = SegmentState::Clean; Ok(()) From b2bf00cdb59299ac899ca2a94688f8086b0f619f Mon Sep 17 00:00:00 2001 From: Borja Castellano Date: Tue, 15 Sep 2026 13:27:01 +0000 Subject: [PATCH 7/8] perf(dash-spv): size storage segments per item type Every segment cache held 50 000 items, whatever an item costs. Near the tip a block segment holds tens of MB of decoded blocks, while a header segment spanning the same heights is 5.6 MB and a filter header one 1.6 MB. `Persistable::ITEMS_PER_SEGMENT` lets each type choose: headers 10 000 (~1.1 MB per segment), filter headers 50 000 (~1.6 MB), filters 2 000 (~2 MB near the tip) and blocks 1 000. This changes the on-disk layout of the header, filter and block segments. Storage written with 50 000-item segments is misread by this layout and has to be deleted until a migration or a versioned folder lands. Mainnet restore, mainnet.100mbi.100ms, #1015/#1016/#1014 applied, two resident segments, jemalloc heap profiling, wallet identical in every run (14114383 sat, 7112 records, 13389 addresses): segment items time peak RSS segment caches 50 000 for every type 7.0 min 818 MiB 332 MiB 5 000 for every type 7.3-8.3 min 554-687 MiB 23-106 MiB 1 000 for every type 9.2-12.2 min 453-577 MiB 2-66 MiB per type (this commit) 7.5 min 538 MiB 38 MiB With 1 000 items everywhere, the header and filter header phases paid an fsync per evicted segment (105-141 s and 254-332 s instead of ~60 s and ~150 s). Blocks at 500 items saved ~18 MiB of cache but reloaded 50 % more block segments. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_017DruChNTWXwJoWPartZwCf --- dash-spv/src/storage/segments.rs | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/dash-spv/src/storage/segments.rs b/dash-spv/src/storage/segments.rs index 2717857b1..f0c98dcc7 100644 --- a/dash-spv/src/storage/segments.rs +++ b/dash-spv/src/storage/segments.rs @@ -27,6 +27,7 @@ use crate::{ pub trait Persistable: Sized + Encodable + Decodable + PartialEq + Clone { const SEGMENT_PREFIX: &'static str = "segment"; const DATA_FILE_EXTENSION: &'static str = "dat"; + const ITEMS_PER_SEGMENT: u32; fn segment_file_name(segment_id: u32) -> String { format!("{}_{:04}.{}", Self::SEGMENT_PREFIX, segment_id, Self::DATA_FILE_EXTENSION) @@ -36,12 +37,16 @@ pub trait Persistable: Sized + Encodable + Decodable + PartialEq + Clone { } impl Persistable for Vec { + const ITEMS_PER_SEGMENT: u32 = 2_000; + fn sentinel() -> Self { vec![] } } impl Persistable for HashedBlockHeader { + const ITEMS_PER_SEGMENT: u32 = 10_000; + fn sentinel() -> Self { let header = BlockHeader { version: Version::from_consensus(i32::MAX), // Invalid version @@ -57,12 +62,16 @@ impl Persistable for HashedBlockHeader { } impl Persistable for FilterHeader { + const ITEMS_PER_SEGMENT: u32 = 50_000; + fn sentinel() -> Self { FilterHeader::from_byte_array([0u8; 32]) } } impl Persistable for HashedBlock { + const ITEMS_PER_SEGMENT: u32 = 1_000; + fn sentinel() -> Self { let block = Block { header: *HashedBlockHeader::sentinel().header(), @@ -562,7 +571,7 @@ pub struct Segment { } impl Segment { - const ITEMS_PER_SEGMENT: u32 = 50_000; + const ITEMS_PER_SEGMENT: u32 = I::ITEMS_PER_SEGMENT; fn new(segment_id: u32, mut items: Vec, state: SegmentState) -> Self { debug_assert!(items.len() <= Self::ITEMS_PER_SEGMENT as usize); From 034a65f39890d7854f9b87b8c83d418e7a60d874 Mon Sep 17 00:00:00 2001 From: Borja Castellano Date: Tue, 15 Sep 2026 13:41:07 +0000 Subject: [PATCH 8/8] refactor(dash-spv): keep Persistable inside the storage module Nothing outside `storage` implements or names the trait, and the `segments` module it lives in is private to `storage` already. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_017DruChNTWXwJoWPartZwCf --- dash-spv/src/storage/segments.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dash-spv/src/storage/segments.rs b/dash-spv/src/storage/segments.rs index f0c98dcc7..4788bb0c7 100644 --- a/dash-spv/src/storage/segments.rs +++ b/dash-spv/src/storage/segments.rs @@ -24,7 +24,7 @@ use crate::{ StorageError, }; -pub trait Persistable: Sized + Encodable + Decodable + PartialEq + Clone { +pub(super) trait Persistable: Sized + Encodable + Decodable + PartialEq + Clone { const SEGMENT_PREFIX: &'static str = "segment"; const DATA_FILE_EXTENSION: &'static str = "dat"; const ITEMS_PER_SEGMENT: u32;