Companion files for QuestDB blog: Finding Java False Sharing with perf c2c

README.md

perf c2c false-sharing tutorial: companion files

Companion files for the QuestDB blog post Finding Java False Sharing with perf c2c.

file	purpose
setup.sql	creates markout_trades / markout_quotes, 2M rows each
query.sql	the HORIZON JOIN markout query under test
server.conf.fragment	worker pinning and page-frame cap for the benchmark server
query_client.py	HTTP workload client: timings plus result-invariance check
group_hitm.py	groups perf script HITM samples into 64-byte cache lines
run-c2c.sh	full baseline-vs-fixed perf c2c comparison
run-e2e.sh	counterbalanced end-to-end latency comparison

The runner scripts locate QuestDB's bundled jemalloc via the enclosing git checkout; when running them from this gist, point JEMALLOC at the library explicitly, e.g. JEMALLOC=/path/to/questdb/core/src/main/bin/linux-x86-64/libjemalloc.so.

group_hitm.py

#!/usr/bin/env python3

import argparse
import csv
import re
import sys
from collections import defaultdict


SAMPLE_PREFIX = re.compile(
    r"^\s*\S+\s+(?P<tid>\d+)\s+\[(?P<cpu>\d+)\]\s+"
    r"(?P<address>[0-9a-fA-F]+)\s+"
)


def parse_args():
    parser = argparse.ArgumentParser(
        description="Group perf-script HITM samples into 64-byte cache lines"
    )
    parser.add_argument("samples", help="hitm.txt produced by perf script")
    parser.add_argument(
        "--symbol",
        help="keep only samples whose resolved symbol contains this text",
    )
    return parser.parse_args()


def main():
    args = parse_args()
    groups = defaultdict(lambda: {"count": 0, "tids": set(), "cpus": set()})

    with open(args.samples, encoding="utf-8") as samples:
        for line_number, line in enumerate(samples, 1):
            if args.symbol and args.symbol not in line:
                continue
            match = SAMPLE_PREFIX.match(line)
            if not match:
                raise RuntimeError(
                    f"cannot parse {args.samples}:{line_number}: {line.rstrip()}"
                )
            address = int(match.group("address"), 16)
            cache_line = address & ~63
            group = groups[cache_line]
            group["count"] += 1
            group["tids"].add(int(match.group("tid")))
            group["cpus"].add(int(match.group("cpu")))
            group.setdefault("address_lo", address)
            group["address_lo"] = min(group["address_lo"], address)
            group.setdefault("address_hi", address)
            group["address_hi"] = max(group["address_hi"], address)

    writer = csv.writer(sys.stdout, lineterminator="\n")
    writer.writerow(
        ["cache_line", "hitm_samples", "tids", "cpus", "address_range"]
    )
    for cache_line, group in sorted(
        groups.items(), key=lambda item: (-item[1]["count"], item[0])
    ):
        writer.writerow(
            [
                hex(cache_line),
                group["count"],
                "|".join(map(str, sorted(group["tids"]))),
                "|".join(map(str, sorted(group["cpus"]))),
                f'{hex(group["address_lo"])}-{hex(group["address_hi"])}',
            ]
        )


if __name__ == "__main__":
    main()

query.sql

SELECT
    h.offset / 1000 AS horizon_ms,
    avg(q.mid - t.price) AS avg_markout
FROM markout_trades t
HORIZON JOIN markout_quotes q
    LIST (0, 1000U, 2000U, 3000U) AS h
ORDER BY h.offset;

query_client.py

#!/usr/bin/env python3

import argparse
import http.client
import json
import signal
import time
import urllib.parse


def parse_args():
    parser = argparse.ArgumentParser()
    query = parser.add_mutually_exclusive_group(required=True)
    query.add_argument("--query")
    query.add_argument("--query-file")
    parser.add_argument("--output", required=True)
    parser.add_argument("--warmups", type=int, default=20)
    parser.add_argument("--max-requests", type=int, default=0)
    return parser.parse_args()


def main():
    args = parse_args()
    if args.query_file:
        with open(args.query_file, encoding="utf-8") as query_file:
            sql = query_file.read().strip().rstrip(";")
    else:
        sql = args.query

    stopping = False

    def stop(_signum, _frame):
        nonlocal stopping
        stopping = True

    signal.signal(signal.SIGINT, stop)
    signal.signal(signal.SIGTERM, stop)

    path = "/exec?" + urllib.parse.urlencode({"query": sql})
    connection = http.client.HTTPConnection("127.0.0.1", 9000, timeout=600)
    expected_dataset = None
    requests = 0
    result_changes = 0

    def execute():
        nonlocal expected_dataset, result_changes
        connection.request("GET", path)
        response = connection.getresponse()
        body = response.read()
        if response.status != 200:
            raise RuntimeError(f"HTTP {response.status}: {body!r}")
        payload = json.loads(body)
        if "error" in payload:
            raise RuntimeError(payload["error"])
        dataset = payload.get("dataset")
        if expected_dataset is None:
            expected_dataset = dataset
        elif dataset != expected_dataset:
            result_changes += 1
            raise RuntimeError("query result changed during recording")

    for _ in range(args.warmups):
        execute()

    start = time.monotonic()
    while not stopping and (args.max_requests == 0 or requests < args.max_requests):
        execute()
        requests += 1
    elapsed = time.monotonic() - start
    connection.close()

    with open(args.output, "w", encoding="utf-8") as output:
        json.dump(
            {
                "query": sql,
                "warmups": args.warmups,
                "recorded_requests": requests,
                "recorded_seconds": elapsed,
                "result": expected_dataset,
                "result_changes": result_changes,
            },
            output,
            indent=2,
        )
        output.write("\n")


if __name__ == "__main__":
    main()

run-c2c.sh

#!/usr/bin/env bash

set -euo pipefail

if [[ $# -ne 4 ]]; then
    echo "usage: $0 BASELINE_JAR FIXED_JAR DB_ROOT OUTPUT_DIR" >&2
    exit 2
fi

baseline_jar=$(realpath "$1")
fixed_jar=$(realpath "$2")
db_root=$(realpath "$3")
output_dir=$4
script_dir=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
repo_root=$(git -C "$script_dir" rev-parse --show-toplevel)
jemalloc=${JEMALLOC:-$repo_root/core/src/main/bin/linux-x86-64/libjemalloc.so}
server_cpus=${SERVER_CPUS:-0-14}
client_cpu=${CLIENT_CPU:-15}
warmups=${WARMUPS:-100}
requests=${REQUESTS:-500}

mkdir -p "$output_dir"
output_dir=$(realpath "$output_dir")
server_pid=
perf_pid=

if curl -fsS -G --data-urlencode "query=SELECT 1" \
    http://127.0.0.1:9000/exec >/dev/null 2>&1; then
    echo "port 9000 already serves a QuestDB instance" >&2
    exit 1
fi

stop_all() {
    if [[ -n "${perf_pid:-}" ]] && kill -0 "$perf_pid" 2>/dev/null; then
        kill -INT "$perf_pid"
        wait "$perf_pid" || true
    fi
    perf_pid=
    if [[ -n "${server_pid:-}" ]] && kill -0 "$server_pid" 2>/dev/null; then
        kill -TERM "$server_pid"
        wait "$server_pid" || true
    fi
    server_pid=
}

trap stop_all EXIT INT TERM

for variant in baseline fixed; do
    if [[ "$variant" == baseline ]]; then
        jar=$baseline_jar
    else
        jar=$fixed_jar
    fi
    variant_dir=$output_dir/$variant
    mkdir -p "$variant_dir"
    echo "START $variant"

    env LD_PRELOAD="$jemalloc" taskset -c "$server_cpus" java \
        --add-exports=java.base/jdk.internal.vm=ALL-UNNAMED \
        --add-opens=java.base/java.lang=ALL-UNNAMED \
        --sun-misc-unsafe-memory-access=allow \
        --enable-native-access=ALL-UNNAMED \
        -cp "$jar" io.questdb.ServerMain -d "$db_root" \
        >"$variant_dir/server.log" 2>&1 &
    server_pid=$!

    ready=false
    for _ in {1..120}; do
        if taskset -c "$client_cpu" curl -fsS -G \
            --data-urlencode "query=SELECT 1" \
            http://127.0.0.1:9000/exec >/dev/null 2>&1; then
            ready=true
            break
        fi
        if ! kill -0 "$server_pid" 2>/dev/null; then
            tail -100 "$variant_dir/server.log" >&2
            exit 1
        fi
        sleep 0.1
    done
    [[ "$ready" == true ]]
    grep -F "$jemalloc" "/proc/$server_pid/maps" >"$variant_dir/jemalloc-map.txt"

    taskset -c "$client_cpu" python3 "$script_dir/query_client.py" \
        --query-file "$script_dir/query.sql" \
        --output "$variant_dir/warmup.json" \
        --warmups 0 \
        --max-requests "$warmups"

    jcmd "$server_pid" Compiler.perfmap >"$variant_dir/jcmd-before.txt"
    perf c2c record -o "$variant_dir/perf.data" -p "$server_pid" \
        >"$variant_dir/record.stdout" 2>"$variant_dir/record.stderr" &
    perf_pid=$!
    sleep 0.5

    taskset -c "$client_cpu" python3 "$script_dir/query_client.py" \
        --query-file "$script_dir/query.sql" \
        --output "$variant_dir/client.json" \
        --warmups 0 \
        --max-requests "$requests"

    kill -INT "$perf_pid"
    wait "$perf_pid" || true
    perf_pid=
    jcmd "$server_pid" Compiler.perfmap >"$variant_dir/jcmd-after.txt"
    cp "/tmp/perf-$server_pid.map" "$variant_dir/perf.map"

    DEBUGINFOD_URLS= perf c2c report --stdio --stats \
        -i "$variant_dir/perf.data" \
        >"$variant_dir/c2c-stats.txt" \
        2>"$variant_dir/c2c-stats.stderr"
    DEBUGINFOD_URLS= perf c2c report --stdio --full-symbols --show-all \
        -i "$variant_dir/perf.data" \
        >"$variant_dir/c2c-report.txt" \
        2>"$variant_dir/c2c-report.stderr"
    DEBUGINFOD_URLS= perf script \
        -i "$variant_dir/perf.data" \
        -F comm,tid,cpu,addr,ip,sym,data_src \
        2>"$variant_dir/perf-script.stderr" \
        | awk '/SNP HitM/' >"$variant_dir/hitm.txt"
    python3 "$script_dir/group_hitm.py" "$variant_dir/hitm.txt" \
        --symbol AsyncHorizonJoinRecordCursorFactory.processHorizonTimestamps \
        >"$variant_dir/hot-cache-lines.csv"

    echo "DONE  $variant"
    stop_all
done

python3 "$script_dir/summarize.py" c2c "$output_dir"

run-e2e.sh

#!/usr/bin/env bash

set -euo pipefail

if [[ $# -ne 4 ]]; then
    echo "usage: $0 BASELINE_JAR FIXED_JAR DB_ROOT OUTPUT_DIR" >&2
    exit 2
fi

baseline_jar=$(realpath "$1")
fixed_jar=$(realpath "$2")
db_root=$(realpath "$3")
output_dir=$4
script_dir=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
repo_root=$(git -C "$script_dir" rev-parse --show-toplevel)
jemalloc=${JEMALLOC:-$repo_root/core/src/main/bin/linux-x86-64/libjemalloc.so}
server_cpus=${SERVER_CPUS:-0-14}
client_cpu=${CLIENT_CPU:-15}
warmups=${WARMUPS:-100}
requests=${REQUESTS:-500}
schedule=(baseline fixed fixed baseline baseline fixed fixed baseline baseline fixed fixed baseline)

mkdir -p "$output_dir"
output_dir=$(realpath "$output_dir")
server_pid=

if curl -fsS -G --data-urlencode "query=SELECT 1" \
    http://127.0.0.1:9000/exec >/dev/null 2>&1; then
    echo "port 9000 already serves a QuestDB instance" >&2
    exit 1
fi

stop_server() {
    if [[ -n "${server_pid:-}" ]] && kill -0 "$server_pid" 2>/dev/null; then
        kill -TERM "$server_pid"
        wait "$server_pid" || true
    fi
    server_pid=
}

trap stop_server EXIT INT TERM

start_server() {
    local jar=$1
    local log=$2
    env LD_PRELOAD="$jemalloc" taskset -c "$server_cpus" java \
        --add-exports=java.base/jdk.internal.vm=ALL-UNNAMED \
        --add-opens=java.base/java.lang=ALL-UNNAMED \
        --sun-misc-unsafe-memory-access=allow \
        --enable-native-access=ALL-UNNAMED \
        -cp "$jar" io.questdb.ServerMain -d "$db_root" >"$log" 2>&1 &
    server_pid=$!

    local ready=false
    for _ in {1..120}; do
        if taskset -c "$client_cpu" curl -fsS -G \
            --data-urlencode "query=SELECT 1" \
            http://127.0.0.1:9000/exec >/dev/null 2>&1; then
            ready=true
            break
        fi
        if ! kill -0 "$server_pid" 2>/dev/null; then
            tail -100 "$log" >&2
            exit 1
        fi
        sleep 0.1
    done
    if [[ "$ready" != true ]]; then
        echo "QuestDB did not become ready" >&2
        exit 1
    fi
    kill -0 "$server_pid"
}

for index in "${!schedule[@]}"; do
    run=$((index + 1))
    variant=${schedule[$index]}
    if [[ "$variant" == baseline ]]; then
        jar=$baseline_jar
    else
        jar=$fixed_jar
    fi
    prefix=$(printf "%02d-%s" "$run" "$variant")
    echo "START $prefix"
    start_server "$jar" "$output_dir/$prefix-server.log"
    grep -F "$jemalloc" "/proc/$server_pid/maps" >"$output_dir/$prefix-jemalloc-map.txt"

    if [[ ! -f "$output_dir/plan-$variant.json" ]]; then
        query=$(tr '\n' ' ' <"$script_dir/query.sql")
        taskset -c "$client_cpu" curl -fsS -G \
            --data-urlencode "query=EXPLAIN $query" \
            http://127.0.0.1:9000/exec >"$output_dir/plan-$variant.json"
        grep -q 'Async Horizon Join workers: 15 offsets: 4' "$output_dir/plan-$variant.json"
    fi

    taskset -c "$client_cpu" python3 "$script_dir/query_client.py" \
        --query-file "$script_dir/query.sql" \
        --output "$output_dir/$prefix.json" \
        --warmups "$warmups" \
        --max-requests "$requests"
    echo "DONE  $prefix"
    stop_server
done

python3 "$script_dir/summarize.py" e2e "$output_dir"

server.conf.fragment

shared.network.worker.count=1
shared.network.worker.affinity=0
shared.query.worker.count=15
shared.query.worker.affinity=0,1,2,3,4,5,6,7,8,9,10,11,12,13,14
shared.write.worker.count=1
shared.write.worker.affinity=0

cairo.partition.encoder.parquet.row.group.size=1000
cairo.sql.page.frame.max.rows=10000
cairo.sql.small.page.frame.max.rows=10000

line.tcp.enabled=false
pg.enabled=false
telemetry.enabled=false
query.tracing.enabled=false

setup.sql

DROP TABLE IF EXISTS markout_trades;
DROP TABLE IF EXISTS markout_quotes;

CREATE TABLE markout_trades AS (
    SELECT
        timestamp_sequence('2025-01-01T00:00:00.000000Z', 1000L) ts,
        x::double price
    FROM long_sequence(2000000)
) TIMESTAMP(ts) PARTITION BY DAY;

CREATE TABLE markout_quotes AS (
    SELECT
        timestamp_sequence('2025-01-01T00:00:00.000000Z', 1000L) ts,
        x::double mid
    FROM long_sequence(2000000)
) TIMESTAMP(ts) PARTITION BY DAY;

SELECT 'markout_trades' AS table_name, count() AS rows FROM markout_trades
UNION ALL
SELECT 'markout_quotes' AS table_name, count() AS rows FROM markout_quotes;