Skip to content

Commit af58d7d

Browse files
bluestreak01claude
bluestreak01
and
claude
committed
test(ilp): SegmentLog.append latency benchmark
Plain main()-style benchmark (same idiom as StacBenchmarkClient — no JMH dependency required). Measures the per-frame latency of the SF persist path: CRC32C over the payload, frame-envelope construction, two pwrite syscalls (header + payload), bookkeeping, and an optional fsync when --fsync=each. Reports min / p50 / p90 / p99 / p99.9 / max in nanoseconds plus throughput in frames/sec and MB/sec. Smoke run on darwin-aarch64 (APFS): --payload-bytes=512 --measure=20000 --fsync=off p50 ≈ 4 µs, p99 ≈ 14 µs, ~150K frames/sec, ~74 MB/sec --payload-bytes=512 --measure=5000 --fsync=each p50 ≈ 28 µs, p99 ≈ 900 µs, ~16K frames/sec, ~8 MB/sec Run via Maven exec or directly from the IDE; the class lives under core/src/test so it has free access to the SF code path without adding to the production classpath. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1 parent 07c20e0 commit af58d7d

1 file changed

Lines changed: 329 additions & 0 deletions

File tree

‎core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/SegmentLogLatencyBenchmark.java‎

Lines changed: 329 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,329 @@
1+
/*+*****************************************************************************
2+
* ___ _ ____ ____
3+
* / _ \ _ _ ___ ___| |_| _ \| __ )
4+
* | | | | | | |/ _ \/ __| __| | | | _ \
5+
* | |_| | |_| | __/\__ \ |_| |_| | |_) |
6+
* \__\_\\__,_|\___||___/\__|____/|____/
7+
*
8+
* Copyright (c) 2014-2019 Appsicle
9+
* Copyright (c) 2019-2026 QuestDB
10+
*
11+
* Licensed under the Apache License, Version 2.0 (the "License");
12+
* you may not use this file except in compliance with the License.
13+
* You may obtain a copy of the License at
14+
*
15+
* http://www.apache.org/licenses/LICENSE-2.0
16+
*
17+
* Unless required by applicable law or agreed to in writing, software
18+
* distributed under the License is distributed on an "AS IS" BASIS,
19+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
20+
* See the License for the specific language governing permissions and
21+
* limitations under the License.
22+
*
23+
******************************************************************************/
24+
25+
package io.questdb.client.test.cutlass.qwp.client.sf;
26+
27+
import io.questdb.client.cutlass.qwp.client.sf.SegmentLog;
28+
import io.questdb.client.std.Files;
29+
import io.questdb.client.std.MemoryTag;
30+
import io.questdb.client.std.Unsafe;
31+
32+
import java.nio.file.Paths;
33+
import java.util.Arrays;
34+
35+
/**
36+
* Latency benchmark for {@link SegmentLog#append}, the per-frame entry point
37+
* the QWiP store-and-forward layer uses to persist outgoing batches before
38+
* they leave the wire.
39+
* <p>
40+
* Measures the wall-clock latency of a single {@code append} call from the
41+
* caller's perspective: CRC32C over the payload, frame-envelope construction,
42+
* two pwrite syscalls (header + payload), bookkeeping, and an optional
43+
* {@code fsync} when {@code --fsync=each}. Reports min / p50 / p90 / p99 /
44+
* p99.9 / max in nanoseconds, plus throughput in frames/sec and MB/sec.
45+
* <p>
46+
* Run via Maven exec:
47+
* <pre>
48+
* mvn -pl core test-compile
49+
* mvn -pl core exec:java \
50+
* -Dexec.classpathScope=test \
51+
* -Dexec.mainClass=io.questdb.client.test.cutlass.qwp.client.sf.SegmentLogLatencyBenchmark \
52+
* -Dexec.args="--payload-bytes=512 --measure=100000 --fsync=off"
53+
* </pre>
54+
* Or directly via your IDE — it's a plain {@code main} method, no JMH.
55+
* <p>
56+
* Defaults are tuned for a quick local sanity check (~1 second runtime). For
57+
* publication-quality numbers run with {@code --warmup=200000 --measure=1000000}
58+
* on an idle machine; the SF code path is short enough that JIT effects fade
59+
* within a few thousand iterations.
60+
*/
61+
public final class SegmentLogLatencyBenchmark {
62+
63+
private static final long DEFAULT_MAX_BYTES_PER_SEGMENT = 64L * 1024 * 1024; // 64 MiB
64+
private static final long DEFAULT_MAX_TOTAL_BYTES = Long.MAX_VALUE;
65+
private static final int DEFAULT_MEASURE = 100_000;
66+
private static final int DEFAULT_PAYLOAD_BYTES = 512;
67+
private static final int DEFAULT_WARMUP = 10_000;
68+
69+
public static void main(String[] args) throws Exception {
70+
int payloadBytes = DEFAULT_PAYLOAD_BYTES;
71+
int warmup = DEFAULT_WARMUP;
72+
int measure = DEFAULT_MEASURE;
73+
long maxBytesPerSegment = DEFAULT_MAX_BYTES_PER_SEGMENT;
74+
long maxTotalBytes = DEFAULT_MAX_TOTAL_BYTES;
75+
FsyncMode fsyncMode = FsyncMode.OFF;
76+
String dirOverride = null;
77+
78+
for (String arg : args) {
79+
if (arg.equals("--help") || arg.equals("-h")) {
80+
printUsage();
81+
System.exit(0);
82+
} else if (arg.startsWith("--payload-bytes=")) {
83+
payloadBytes = Integer.parseInt(arg.substring("--payload-bytes=".length()));
84+
} else if (arg.startsWith("--warmup=")) {
85+
warmup = Integer.parseInt(arg.substring("--warmup=".length()));
86+
} else if (arg.startsWith("--measure=")) {
87+
measure = Integer.parseInt(arg.substring("--measure=".length()));
88+
} else if (arg.startsWith("--max-bytes-per-segment=")) {
89+
maxBytesPerSegment = parseSize(arg.substring("--max-bytes-per-segment=".length()));
90+
} else if (arg.startsWith("--max-total-bytes=")) {
91+
maxTotalBytes = parseSize(arg.substring("--max-total-bytes=".length()));
92+
} else if (arg.startsWith("--fsync=")) {
93+
fsyncMode = FsyncMode.parse(arg.substring("--fsync=".length()));
94+
} else if (arg.startsWith("--dir=")) {
95+
dirOverride = arg.substring("--dir=".length());
96+
} else {
97+
System.err.println("Unknown option: " + arg);
98+
printUsage();
99+
System.exit(1);
100+
}
101+
}
102+
103+
if (payloadBytes <= 0) {
104+
System.err.println("--payload-bytes must be > 0");
105+
System.exit(1);
106+
}
107+
if (measure <= 0) {
108+
System.err.println("--measure must be > 0");
109+
System.exit(1);
110+
}
111+
if (warmup < 0) {
112+
System.err.println("--warmup must be >= 0");
113+
System.exit(1);
114+
}
115+
long oneFrameTotal = 8L /* FRAME_HEADER_SIZE */ + payloadBytes;
116+
if (24L /* HEADER_SIZE */ + oneFrameTotal > maxBytesPerSegment) {
117+
System.err.println("--max-bytes-per-segment too small for a single frame "
118+
+ "(need >= " + (24 + oneFrameTotal) + " bytes for the configured payload)");
119+
System.exit(1);
120+
}
121+
122+
String dir = dirOverride != null
123+
? dirOverride
124+
: Paths.get(System.getProperty("java.io.tmpdir"),
125+
"qdb-sf-bench-" + System.nanoTime()).toString();
126+
boolean ownDir = dirOverride == null;
127+
if (ownDir) {
128+
int rc = Files.mkdir(dir, 0755);
129+
if (rc != 0) {
130+
System.err.println("Failed to create benchmark dir: " + dir + " (rc=" + rc + ")");
131+
System.exit(1);
132+
}
133+
}
134+
135+
System.out.println("SegmentLog.append latency benchmark");
136+
System.out.println("====================================");
137+
System.out.println("Payload bytes: " + format(payloadBytes));
138+
System.out.println("Warmup iterations: " + format(warmup));
139+
System.out.println("Measure iterations: " + format(measure));
140+
System.out.println("Max bytes per segment: " + format(maxBytesPerSegment));
141+
System.out.println("Max total bytes: "
142+
+ (maxTotalBytes == Long.MAX_VALUE ? "unlimited" : format(maxTotalBytes)));
143+
System.out.println("Fsync mode: " + fsyncMode);
144+
System.out.println("SF directory: " + dir);
145+
System.out.println();
146+
147+
long buf = Unsafe.malloc(payloadBytes, MemoryTag.NATIVE_DEFAULT);
148+
try {
149+
// Deterministic-but-non-zero payload so the CRC isn't trivially short-circuited
150+
// by an all-zero stream and so any branch on payload content is exercised.
151+
for (int i = 0; i < payloadBytes; i++) {
152+
Unsafe.getUnsafe().putByte(buf + i, (byte) (i * 31 + 17));
153+
}
154+
155+
try (SegmentLog log = SegmentLog.open(dir, maxBytesPerSegment, maxTotalBytes,
156+
fsyncMode == FsyncMode.EACH)) {
157+
158+
// Warmup — discard timing, let the JIT settle and the first segment fill.
159+
for (int i = 0; i < warmup; i++) {
160+
log.append(buf, payloadBytes);
161+
}
162+
163+
long[] samples = new long[measure];
164+
long startNs = System.nanoTime();
165+
for (int i = 0; i < measure; i++) {
166+
long t0 = System.nanoTime();
167+
log.append(buf, payloadBytes);
168+
samples[i] = System.nanoTime() - t0;
169+
}
170+
long elapsedNs = System.nanoTime() - startNs;
171+
172+
// Optional final fsync when the per-call mode was OFF, so disk
173+
// committed bytes are stable before we report.
174+
if (fsyncMode == FsyncMode.FINAL_ONLY) {
175+
log.fsync();
176+
}
177+
178+
report(samples, elapsedNs, payloadBytes, log);
179+
}
180+
} finally {
181+
Unsafe.free(buf, payloadBytes, MemoryTag.NATIVE_DEFAULT);
182+
if (ownDir) {
183+
rmTree(dir);
184+
}
185+
}
186+
}
187+
188+
private static String format(long n) {
189+
return String.format("%,d", n);
190+
}
191+
192+
private static String formatDouble(double d) {
193+
if (d >= 1000) {
194+
return String.format("%,.0f", d);
195+
}
196+
if (d >= 10) {
197+
return String.format("%,.1f", d);
198+
}
199+
return String.format("%,.2f", d);
200+
}
201+
202+
private static long parseSize(String s) {
203+
s = s.trim().toUpperCase();
204+
long mult = 1;
205+
if (s.endsWith("K") || s.endsWith("KB")) {
206+
mult = 1024L;
207+
s = s.substring(0, s.length() - (s.endsWith("KB") ? 2 : 1));
208+
} else if (s.endsWith("M") || s.endsWith("MB")) {
209+
mult = 1024L * 1024;
210+
s = s.substring(0, s.length() - (s.endsWith("MB") ? 2 : 1));
211+
} else if (s.endsWith("G") || s.endsWith("GB")) {
212+
mult = 1024L * 1024 * 1024;
213+
s = s.substring(0, s.length() - (s.endsWith("GB") ? 2 : 1));
214+
}
215+
return Long.parseLong(s.trim()) * mult;
216+
}
217+
218+
private static void printUsage() {
219+
System.out.println("Usage: SegmentLogLatencyBenchmark [options]");
220+
System.out.println();
221+
System.out.println("Options:");
222+
System.out.println(" --payload-bytes=<n> Frame payload size in bytes (default: 512)");
223+
System.out.println(" --warmup=<n> Warmup append count (default: 10,000)");
224+
System.out.println(" --measure=<n> Measured append count (default: 100,000)");
225+
System.out.println(" --max-bytes-per-segment=<sz> Segment rotation threshold (default: 64M)");
226+
System.out.println(" Suffixes: K, M, G");
227+
System.out.println(" --max-total-bytes=<sz> Total disk cap (default: unlimited)");
228+
System.out.println(" --fsync=off|each|final Per-append fsync mode (default: off)");
229+
System.out.println(" off: no fsync, fastest");
230+
System.out.println(" each: fsync after every append (durability max)");
231+
System.out.println(" final: fsync once after the run (closer to flush())");
232+
System.out.println(" --dir=<path> Use this dir instead of an autogenerated tmp dir");
233+
System.out.println(" -h, --help Show this help");
234+
}
235+
236+
private static void report(long[] samples, long elapsedNs, int payloadBytes, SegmentLog log) {
237+
Arrays.sort(samples);
238+
int n = samples.length;
239+
long min = samples[0];
240+
long p50 = samples[(int) (n * 0.50)];
241+
long p90 = samples[(int) (n * 0.90)];
242+
long p99 = samples[(int) (n * 0.99)];
243+
long p999 = samples[Math.min(n - 1, (int) (n * 0.999))];
244+
long max = samples[n - 1];
245+
246+
long sum = 0;
247+
for (long s : samples) {
248+
sum += s;
249+
}
250+
double meanNs = (double) sum / n;
251+
252+
double seconds = elapsedNs / 1e9;
253+
double framesPerSec = n / seconds;
254+
// payload + 8-byte SF envelope; the segment header is amortised across
255+
// every frame in a segment and small enough to ignore here.
256+
double mbPerSec = framesPerSec * (payloadBytes + 8) / (1024.0 * 1024.0);
257+
258+
System.out.println("Latency (ns):");
259+
System.out.println(" min: " + format(min));
260+
System.out.println(" p50: " + format(p50));
261+
System.out.println(" p90: " + format(p90));
262+
System.out.println(" p99: " + format(p99));
263+
System.out.println(" p99.9: " + format(p999));
264+
System.out.println(" max: " + format(max));
265+
System.out.println(" mean: " + format((long) meanNs));
266+
System.out.println();
267+
System.out.println("Throughput:");
268+
System.out.println(" frames/sec: " + formatDouble(framesPerSec));
269+
System.out.println(" MB/sec (payload+env): " + formatDouble(mbPerSec));
270+
System.out.println();
271+
System.out.println("Final SegmentLog state:");
272+
System.out.println(" segments: " + log.segmentCount());
273+
System.out.println(" bytesOnDisk: " + format(log.bytesOnDisk()));
274+
System.out.println(" nextSeq: " + format(log.nextSeq()));
275+
}
276+
277+
private static void rmTree(String dir) {
278+
if (dir == null || !Files.exists(dir)) {
279+
return;
280+
}
281+
long find = Files.findFirst(dir);
282+
if (find != 0) {
283+
try {
284+
int rc = 1;
285+
while (rc > 0) {
286+
String name = Files.utf8ToString(Files.findName(find));
287+
if (name != null && !".".equals(name) && !"..".equals(name)) {
288+
Files.remove(dir + "/" + name);
289+
}
290+
rc = Files.findNext(find);
291+
}
292+
} finally {
293+
Files.findClose(find);
294+
}
295+
}
296+
Files.remove(dir);
297+
}
298+
299+
private enum FsyncMode {
300+
OFF, EACH, FINAL_ONLY;
301+
302+
static FsyncMode parse(String s) {
303+
switch (s.toLowerCase()) {
304+
case "off":
305+
return OFF;
306+
case "each":
307+
return EACH;
308+
case "final":
309+
return FINAL_ONLY;
310+
default:
311+
throw new IllegalArgumentException("--fsync must be off|each|final, got: " + s);
312+
}
313+
}
314+
315+
@Override
316+
public String toString() {
317+
switch (this) {
318+
case OFF:
319+
return "off";
320+
case EACH:
321+
return "each";
322+
case FINAL_ONLY:
323+
return "final";
324+
default:
325+
return name();
326+
}
327+
}
328+
}
329+
}

0 commit comments

Comments
 (0)