atproto pds in zig
pds.zat.dev
pds
atproto
Instrument write path benchmarks
+222
-24
+1
-1
README.md
+1
-1
README.md
+37
-20
bench/README.md
+37
-20
bench/README.md
···
13
13
just bench repo
14
14
just bench blob
15
15
just bench metastore 10 1000
16
+
just bench write-profile 10 500
16
17
just bench run --scenario write --records 10000
17
18
```
18
19
···
27
28
- `blob`: blob write/read against disk blobstore plus SQLite metadata.
28
29
- `metastore`: Tranquil-shaped apply/get/list benchmark with caller counts and
29
30
latency percentiles.
31
+
- `write-profile`: `applyWrites` stage timing under write concurrency.
30
32
31
33
Each run uses temporary SQLite and blobstore state.
32
34
···
36
38
37
39
| scenario | zds |
38
40
|---|---:|
39
-
| write records | 294 ops/s |
41
+
| write records | 144 ops/s |
40
42
| list records | 30k ops/s |
41
-
| write repo car | 226 ops/s, 449 MB/s |
42
-
| blob put+get | 5.6k ops/s, 697 MB/s |
43
+
| write repo car | 252 ops/s, 440 MB/s |
44
+
| blob put+get | 368 ops/s, 46 MB/s |
43
45
44
46
## Tranquil comparison
45
47
···
67
69
68
70
| operation | callers | ops | zds | tranquil |
69
71
|---|---:|---:|---:|---:|
70
-
| apply/write | 1 | 1000 | 259 ops/s, p95 8.7 ms | not measured at this count |
71
-
| apply/write | 1 | 5000 | did not finish in 150 s | 268 ops/s, p95 4.1 ms |
72
-
| apply/write | 10 | 10000 | 172 ops/s, p95 48.3 ms | 738 ops/s, p95 16.0 ms |
73
-
| apply/write | 100 | 20000 | 1531 ops/s, p95 32.1 ms | 888 ops/s, p95 123.9 ms |
74
-
| apply/write | 1000 | 50000 | did not finish in 150 s | 1169 ops/s, then read backpressure |
75
-
| get record | 1 | 1000 | 94k ops/s, p95 15 us | not measured at this count |
76
-
| get record | 10 | 10000 | 88k ops/s, p95 14 us | 339k ops/s, p95 49 us |
77
-
| get record | 100 | 20000 | 82k ops/s, p95 849 us | 425k ops/s, p95 349 us |
78
-
| list records | 1 | 1000 | 58k ops/s, p95 18 us | not measured at this count |
79
-
| list records | 10 | 10000 | 53k ops/s, p95 22 us | 33k ops/s, p95 411 us |
80
-
| list records | 100 | 20000 | 44k ops/s, p95 941 us | 33k ops/s, p95 3.6 ms |
72
+
| apply/write | 1 | 1000 | 206 ops/s, p95 11.9 ms | not measured at this count |
73
+
| apply/write | 1 | 5000 | not rerun | 292 ops/s, p95 5.8 ms |
74
+
| apply/write | 10 | 10000 | 106 ops/s, p95 119 ms | 436 ops/s, p95 36.0 ms |
75
+
| apply/write | 100 | 20000 | 125 ops/s, p95 110 ms | 944 ops/s, p95 119 ms |
76
+
| apply/write | 1000 | 50000 | not rerun | 1174 ops/s, then read backpressure |
77
+
| get record | 1 | 1000 | 55k ops/s, p95 25 us | not measured at this count |
78
+
| get record | 10 | 10000 | 56k ops/s, p95 64 us | 571k ops/s, p95 31 us |
79
+
| get record | 100 | 20000 | 63k ops/s, p95 4.1 ms | 571k ops/s, p95 244 us |
80
+
| list records | 1 | 1000 | 51k ops/s, p95 22 us | not measured at this count |
81
+
| list records | 10 | 10000 | 51k ops/s, p95 21 us | 38k ops/s, p95 309 us |
82
+
| list records | 100 | 20000 | 51k ops/s, p95 21 us | 37k ops/s, p95 3.0 ms |
83
+
84
+
The 1000-caller Tranquil run reached 1174 apply commits/sec, then the next read
85
+
phase failed with `metastore handler backpressure` on this machine.
86
+
87
+
## write profile
88
+
89
+
`write-profile` shows that the current ZDS write path is dominated by serialized
90
+
time around the store mutex, not SQLite alone:
91
+
92
+
| callers | ops | throughput | lock wait | load repo | build commit | sqlite/event |
93
+
|---:|---:|---:|---:|---:|---:|---:|
94
+
| 10 | 500 | 2386 ops/s | 83% | 3% | 8% | 6% |
95
+
| 10 | 5000 | 353 ops/s | 82% | 13% | 2% | 3% |
96
+
| 100 | 2000 | 1310 ops/s | 98% | 1% | 1% | 1% |
81
97
82
-
The ZDS 1-caller/5000-write tier and 1000-caller/50000-write tier were stopped
83
-
after roughly 150 seconds. The 1000-caller Tranquil run reached 1169 apply
84
-
commits/sec, then the next read phase failed with `metastore handler
85
-
backpressure` on this machine.
98
+
The next write-path work should shrink or restructure the serialized section:
99
+
keep validation and request parsing outside the store lock, move toward a
100
+
Tranquil-style write worker/queue for repo mutation, and avoid rebuilding more
101
+
repo state than a single write needs.
86
102
87
103
## comparison work
88
104
···
101
117
- full repo export cost
102
118
- blob/block store throughput
103
119
104
-
Tranquil already measures latency percentiles. ZDS should add p50/p95/p99
105
-
latency reporting before we treat the comparison as mature.
120
+
ZDS and Tranquil both report latency percentiles for the metastore-shaped
121
+
benchmark. The remaining comparison gap is broader coverage of blockstore,
122
+
eventlog, and full repo export behavior.
+4
bench/justfile
+4
bench/justfile
···
28
28
# benchmark Tranquil metastore-shaped store operations
29
29
metastore callers="1" ops_per_caller="100":
30
30
{{zig}} build bench -Doptimize=ReleaseFast -- --scenario metastore --callers {{callers}} --ops-per-caller {{ops_per_caller}}
31
+
32
+
# profile applyWrites stage timing under write concurrency
33
+
write-profile callers="10" ops_per_caller="500":
34
+
{{zig}} build bench -Doptimize=ReleaseFast -- --scenario write-profile --callers {{callers}} --ops-per-caller {{ops_per_caller}}
+102
-1
bench/main.zig
+102
-1
bench/main.zig
···
8
8
blob,
9
9
repo,
10
10
metastore,
11
+
write_profile,
11
12
};
12
13
13
14
const bench_collection = "dev.zds.bench.record";
···
93
94
},
94
95
.blob => (try benchBlob(allocator, state.account, options.blobs, options.blob_size)).print(),
95
96
.metastore => try benchMetastore(allocator, options),
97
+
.write_profile => try benchWriteProfile(allocator, options),
96
98
}
97
99
}
98
100
···
296
298
return concurrentResult("applyWrites", level, total_ops, nowNs() - start, latencies);
297
299
}
298
300
301
+
fn benchWriteProfile(allocator: std.mem.Allocator, options: Options) !void {
302
+
const level: ConcurrencyLevel = .{
303
+
.callers = options.callers orelse 10,
304
+
.ops_per_caller = options.ops_per_caller orelse 100,
305
+
};
306
+
var state = try initBench(allocator);
307
+
defer state.deinit();
308
+
309
+
var arena = std.heap.ArenaAllocator.init(allocator);
310
+
defer arena.deinit();
311
+
const a = arena.allocator();
312
+
313
+
const accounts = try a.alloc(zds.auth.tokens.Account, level.callers);
314
+
for (accounts, 0..) |*account, i| {
315
+
account.* = try zds.storage.store.createAccount(
316
+
a,
317
+
try std.fmt.allocPrint(a, "profile-{d}.test", .{i}),
318
+
try std.fmt.allocPrint(a, "profile-{d}@test.com", .{i}),
319
+
"password",
320
+
try std.fmt.allocPrint(a, "did:plc:zdsprofile{d}", .{i}),
321
+
true,
322
+
);
323
+
}
324
+
325
+
const total_ops = level.callers * level.ops_per_caller;
326
+
const latencies = try allocator.alloc(u64, total_ops);
327
+
defer allocator.free(latencies);
328
+
const profiles = try allocator.alloc(zds.storage.store.WriteProfile, total_ops);
329
+
defer allocator.free(profiles);
330
+
const threads = try allocator.alloc(std.Thread, level.callers);
331
+
defer allocator.free(threads);
332
+
333
+
const start = nowNs();
334
+
for (threads, 0..) |*thread, i| {
335
+
thread.* = try std.Thread.spawn(.{}, profileWorker, .{
336
+
accounts[i],
337
+
i * level.ops_per_caller,
338
+
latencies[i * level.ops_per_caller ..][0..level.ops_per_caller],
339
+
profiles[i * level.ops_per_caller ..][0..level.ops_per_caller],
340
+
});
341
+
}
342
+
for (threads) |thread| thread.join();
343
+
344
+
const result = concurrentResult("applyWrites profile", level, total_ops, nowNs() - start, latencies);
345
+
result.print();
346
+
printWriteProfile(profiles);
347
+
}
348
+
299
349
fn benchConcurrentGets(allocator: std.mem.Allocator, level: ConcurrencyLevel) !ConcurrentResult {
300
350
var state = try initBench(allocator);
301
351
defer state.deinit();
···
360
410
}
361
411
}
362
412
413
+
fn profileWorker(
414
+
account: zds.auth.tokens.Account,
415
+
offset: usize,
416
+
latencies: []u64,
417
+
profiles: []zds.storage.store.WriteProfile,
418
+
) !void {
419
+
var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
420
+
defer arena.deinit();
421
+
for (latencies, profiles, 0..) |*latency, *profile, i| {
422
+
_ = arena.reset(.retain_capacity);
423
+
const value = try benchRecordValue(arena.allocator(), offset + i);
424
+
const start = nowNs();
425
+
const result = try zds.storage.store.applyWritesProfiled(arena.allocator(), account, &.{.{ .create = .{
426
+
.collection = bench_collection,
427
+
.rkey = try std.fmt.allocPrint(arena.allocator(), "r{d:0>10}", .{offset + i}),
428
+
.value = value,
429
+
} }}, profile);
430
+
if (result.records.len != 1) return error.UnexpectedWriteResult;
431
+
latency.* = nowNs() - start;
432
+
}
433
+
}
434
+
435
+
fn printWriteProfile(profiles: []const zds.storage.store.WriteProfile) void {
436
+
var total: zds.storage.store.WriteProfile = .{};
437
+
for (profiles) |profile| {
438
+
total.validation_ns += profile.validation_ns;
439
+
total.lock_wait_ns += profile.lock_wait_ns;
440
+
total.load_repo_ns += profile.load_repo_ns;
441
+
total.stage_records_ns += profile.stage_records_ns;
442
+
total.build_commit_ns += profile.build_commit_ns;
443
+
total.sql_ns += profile.sql_ns;
444
+
total.event_publish_ns += profile.event_publish_ns;
445
+
total.total_ns += profile.total_ns;
446
+
}
447
+
std.debug.print("\nwrite stage averages across {d} ops:\n", .{profiles.len});
448
+
printProfileStage("validation", total.validation_ns, profiles.len, total.total_ns);
449
+
printProfileStage("lock wait", total.lock_wait_ns, profiles.len, total.total_ns);
450
+
printProfileStage("load repo", total.load_repo_ns, profiles.len, total.total_ns);
451
+
printProfileStage("stage records", total.stage_records_ns, profiles.len, total.total_ns);
452
+
printProfileStage("build commit", total.build_commit_ns, profiles.len, total.total_ns);
453
+
printProfileStage("sqlite/event", total.sql_ns, profiles.len, total.total_ns);
454
+
printProfileStage("publish", total.event_publish_ns, profiles.len, total.total_ns);
455
+
}
456
+
457
+
fn printProfileStage(name: []const u8, ns: u64, ops: usize, total_ns: u64) void {
458
+
const avg_ms = @as(f64, @floatFromInt(ns)) / @as(f64, @floatFromInt(ops)) / std.time.ns_per_ms;
459
+
const pct = if (total_ns == 0) 0 else @as(f64, @floatFromInt(ns)) * 100.0 / @as(f64, @floatFromInt(total_ns));
460
+
std.debug.print(" {s: <14} {d: >9.3} ms/op {d: >6.1}%\n", .{ name, avg_ms, pct });
461
+
}
462
+
363
463
fn getWorker(did: []const u8, task_id: usize, latencies: []u64) !void {
364
464
for (latencies, 0..) |*latency, i| {
365
465
const rec_idx = (task_id * 7 + i * 13) % 1000;
···
503
603
if (std.mem.eql(u8, value, "blob")) return .blob;
504
604
if (std.mem.eql(u8, value, "repo")) return .repo;
505
605
if (std.mem.eql(u8, value, "metastore")) return .metastore;
606
+
if (std.mem.eql(u8, value, "write-profile")) return .write_profile;
506
607
return error.UnknownScenario;
507
608
}
508
609
509
610
fn usage() void {
510
611
std.debug.print(
511
-
\\usage: zds-bench [--scenario all|write|read|blob|repo|metastore] [--records N] [--blobs N] [--blob-size BYTES] [--callers N --ops-per-caller N]
612
+
\\usage: zds-bench [--scenario all|write|read|blob|repo|metastore|write-profile] [--records N] [--blobs N] [--blob-size BYTES] [--callers N --ops-per-caller N]
512
613
\\
513
614
, .{});
514
615
}
+2
-2
src/http/landing/mod.zig
+2
-2
src/http/landing/mod.zig
···
93
93
\\ <meta name="viewport" content="width=device-width, initial-scale=1">
94
94
\\ <meta name="color-scheme" content="dark light">
95
95
\\ <title>zds</title>
96
-
\\ <meta name="description" content="a zig implementation of an at protocol personal data server">
96
+
\\ <meta name="description" content="an at protocol personal data server">
97
97
\\ <link rel="icon" href="/favicon.svg" type="image/svg+xml">
98
98
\\ <style>{s}</style>
99
99
\\</head>
···
105
105
\\ </header>
106
106
\\ <section class="hero" aria-labelledby="title">
107
107
\\ <h1 id="title">{s}</h1>
108
-
\\ <p class="lede">an <a href="https://atproto.com">atproto</a> personal data server, written in zig.</p>
108
+
\\ <p class="lede">an <a href="https://atproto.com">atproto</a> personal data server.</p>
109
109
\\ </section>
110
110
\\ {s}
111
111
\\ <div class="grid">
+76
src/storage/store.zig
+76
src/storage/store.zig
···
133
133
records: []Record,
134
134
};
135
135
136
+
pub const WriteProfile = struct {
137
+
validation_ns: u64 = 0,
138
+
lock_wait_ns: u64 = 0,
139
+
load_repo_ns: u64 = 0,
140
+
stage_records_ns: u64 = 0,
141
+
build_commit_ns: u64 = 0,
142
+
sql_ns: u64 = 0,
143
+
event_publish_ns: u64 = 0,
144
+
total_ns: u64 = 0,
145
+
};
146
+
136
147
pub const WriteOp = union(enum) {
137
148
create: struct {
138
149
collection: []const u8,
···
680
691
}
681
692
682
693
pub fn applyWrites(allocator: std.mem.Allocator, account: auth.Account, ops: []const WriteOp) !WriteResult {
694
+
return applyWritesMeasured(allocator, account, ops, null);
695
+
}
696
+
697
+
pub fn applyWritesProfiled(allocator: std.mem.Allocator, account: auth.Account, ops: []const WriteOp, profile: *WriteProfile) !WriteResult {
698
+
profile.* = .{};
699
+
return applyWritesMeasured(allocator, account, ops, profile);
700
+
}
701
+
702
+
fn applyWritesMeasured(allocator: std.mem.Allocator, account: auth.Account, ops: []const WriteOp, profile: ?*WriteProfile) !WriteResult {
703
+
const total_start = monotonicNs();
683
704
if (ops.len == 0) return Error.MissingRecord;
684
705
706
+
const validation_start = monotonicNs();
685
707
for (ops) |op| switch (op) {
686
708
.create => |create_op| {
687
709
if (zat.Nsid.parse(create_op.collection) == null) return Error.InvalidCollection;
···
698
720
if (zat.Rkey.parse(delete_op.rkey) == null) return Error.InvalidRecordKey;
699
721
},
700
722
};
723
+
addProfile(profile, .validation_ns, elapsedNs(validation_start));
701
724
725
+
const lock_start = monotonicNs();
702
726
mutex.lockUncancelable(store_io);
727
+
addProfile(profile, .lock_wait_ns, elapsedNs(lock_start));
703
728
defer mutex.unlock(store_io);
704
729
try requireInitialized();
705
730
731
+
const load_start = monotonicNs();
706
732
const seq = try nextSeqLocked();
707
733
const current = try latestCommitRawLocked(allocator, account.did);
708
734
const rev = try revForSeq(allocator, seq, if (current) |root| root.rev else null);
···
711
737
try zat.mst.Mst.loadFromBlocks(allocator, repo_car, root.data_cid_raw)
712
738
else
713
739
zat.mst.Mst.init(allocator);
740
+
addProfile(profile, .load_repo_ns, elapsedNs(load_start));
714
741
715
742
var records: std.ArrayList(Record) = .empty;
716
743
var record_blocks: std.ArrayList(ImportedBlock) = .empty;
717
744
var mst_blocks: std.ArrayList(zat.car.Block) = .empty;
718
745
var blob_refs: std.ArrayList(BlobRef) = .empty;
719
746
747
+
const stage_start = monotonicNs();
720
748
for (ops) |op| switch (op) {
721
749
.create => |create_op| {
722
750
const rkey = create_op.rkey orelse try nextRkeyLocked(allocator);
···
732
760
_ = try tree.deleteReturn(path);
733
761
},
734
762
};
763
+
addProfile(profile, .stage_records_ns, elapsedNs(stage_start));
735
764
765
+
const build_start = monotonicNs();
736
766
const data_cid = try tree.rootCid();
737
767
try writeMstBlocks(allocator, &tree, &mst_blocks);
738
768
const commit = try signedCommit(allocator, account.did, rev, data_cid, if (current) |root| root.commit_cid_raw else null);
769
+
addProfile(profile, .build_commit_ns, elapsedNs(build_start));
739
770
771
+
const sql_start = monotonicNs();
740
772
try conn.exclusiveTransaction();
741
773
errdefer conn.rollback();
742
774
for (record_blocks.items) |block| {
···
814
846
\\ evt = excluded.evt
815
847
, .{ @as(i64, @intCast(seq)), account.did, commit.cid, zqlite.blob(event_frame) });
816
848
try conn.commit();
849
+
addProfile(profile, .sql_ns, elapsedNs(sql_start));
850
+
851
+
const publish_start = monotonicNs();
817
852
eventlog.publish(seq);
853
+
addProfile(profile, .event_publish_ns, elapsedNs(publish_start));
854
+
if (profile) |p| p.total_ns = elapsedNs(total_start);
818
855
819
856
return .{
820
857
.commit = .{ .cid = commit.cid, .rev = rev },
821
858
.records = try records.toOwnedSlice(allocator),
822
859
};
860
+
}
861
+
862
+
const ProfileField = enum {
863
+
validation_ns,
864
+
lock_wait_ns,
865
+
load_repo_ns,
866
+
stage_records_ns,
867
+
build_commit_ns,
868
+
sql_ns,
869
+
event_publish_ns,
870
+
};
871
+
872
+
fn addProfile(profile: ?*WriteProfile, field: ProfileField, ns: u64) void {
873
+
const p = profile orelse return;
874
+
switch (field) {
875
+
.validation_ns => p.validation_ns += ns,
876
+
.lock_wait_ns => p.lock_wait_ns += ns,
877
+
.load_repo_ns => p.load_repo_ns += ns,
878
+
.stage_records_ns => p.stage_records_ns += ns,
879
+
.build_commit_ns => p.build_commit_ns += ns,
880
+
.sql_ns => p.sql_ns += ns,
881
+
.event_publish_ns => p.event_publish_ns += ns,
882
+
}
883
+
}
884
+
885
+
fn monotonicNs() u64 {
886
+
var ts: std.posix.timespec = undefined;
887
+
const timestamp = switch (std.posix.errno(std.posix.system.clock_gettime(.MONOTONIC, &ts))) {
888
+
.SUCCESS => ts,
889
+
else => std.posix.timespec{ .sec = 0, .nsec = 0 },
890
+
};
891
+
const seconds: u64 = if (timestamp.sec < 0) 0 else @intCast(timestamp.sec);
892
+
const nanos: u64 = if (timestamp.nsec < 0) 0 else @intCast(timestamp.nsec);
893
+
return seconds * std.time.ns_per_s + nanos;
894
+
}
895
+
896
+
fn elapsedNs(start: u64) u64 {
897
+
const now = monotonicNs();
898
+
return if (now >= start) now - start else 0;
823
899
}
824
900
825
901
pub fn revForSeq(allocator: std.mem.Allocator, seq: u64, current_rev: ?[]const u8) ![]const u8 {