From 1f8f067bfca6d430d53de57d78407574dac7edec Mon Sep 17 00:00:00 2001 From: DualFroz Date: Sat, 12 Sep 2026 12:14:59 +0200 Subject: [PATCH] feat: add readerWorker and run to server --- src/server.zig | 160 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 160 insertions(+) diff --git a/src/server.zig b/src/server.zig index f21641b..14edeeb 100644 --- a/src/server.zig +++ b/src/server.zig @@ -5169,3 +5169,163 @@ test "table_select survives concurrent flush and compact reshaping the underlyin .{ .field = "age", .descending = true }, }; + const readerWorker = struct { + fn run(ctx: *ReaderCtx) void { + var client = client_mod.Client.connect(ctx.allocator, ctx.addr) catch return; + defer client.close(); + var i: usize = 0; + while (i < readerIters) : (i += 1) { + const warsawRows = client.tableSelect(ctx.allocator, "people", "city eq s:warsaw", 0, 0, &.{}) catch return; + defer protocol.freeRows(ctx.allocator, warsawRows); + if (warsawRows.len != rowCount / 3) return; + + const allRows = client.tableSelect(ctx.allocator, "people", "", 0, 0, &.{}) catch return; + defer protocol.freeRows(ctx.allocator, allRows); + if (allRows.len != rowCount) return; + for (allRows) |row| { + const pk = row.get(querymod.pkField) orelse return; + const age = row.get("age") orelse return; + if (age.int != pk.int * 2) return; + } + + const ordered = client.tableSelect(ctx.allocator, "people", "", 0, 0, &cityOrderBy) catch return; + defer protocol.freeRows(ctx.allocator, ordered); + if (ordered.len != rowCount) return; + var prevCity: ?[]const u8 = null; + var prevAge: i64 = std.math.maxInt(i64); + for (ordered) |row| { + const city = (row.get("city") orelse return).text; + const age = (row.get("age") orelse return).int; + if (prevCity) |pc| { + const cmp = std.mem.order(u8, pc, city); + if (cmp == .gt) return; + if (cmp == .eq and age > prevAge) return; + if (cmp != .eq) prevAge = std.math.maxInt(i64); + } + prevCity = city; + prevAge = age; + } + } + ctx.ok = true; + } + }.run; + + const MutatorCtx = struct { + allocator: Allocator, + addr: std.net.Address, + thread: usize, + ok: bool = false, + }; + const mutatorWorker = struct { + fn run(ctx: *MutatorCtx) void { + var client = client_mod.Client.connect(ctx.allocator, ctx.addr) catch return; + defer client.close(); + var i: usize = 0; + while (i < mutatorIters) : (i += 1) { + var buf: [32]u8 = undefined; + const key = std.fmt.bufPrint(&buf, "churn-{d}-{d}", .{ ctx.thread, i }) catch return; + client.set(key, "x") catch return; + client.flush() catch return; + client.compact() catch return; + } + ctx.ok = true; + } + }.run; + + var readerContexts: [readerThreads]ReaderCtx = undefined; + var readerHandles: [readerThreads]std.Thread = undefined; + for (0..readerThreads) |i| { + readerContexts[i] = .{ .allocator = allocator, .addr = server.boundAddress() }; + readerHandles[i] = try std.Thread.spawn(.{}, readerWorker, .{&readerContexts[i]}); + } + var mutatorContexts: [mutatorThreads]MutatorCtx = undefined; + var mutatorHandles: [mutatorThreads]std.Thread = undefined; + for (0..mutatorThreads) |i| { + mutatorContexts[i] = .{ .allocator = allocator, .addr = server.boundAddress(), .thread = i }; + mutatorHandles[i] = try std.Thread.spawn(.{}, mutatorWorker, .{&mutatorContexts[i]}); + } + + for (0..readerThreads) |i| readerHandles[i].join(); + for (0..mutatorThreads) |i| mutatorHandles[i].join(); + for (0..readerThreads) |i| try std.testing.expect(readerContexts[i].ok); + for (0..mutatorThreads) |i| try std.testing.expect(mutatorContexts[i].ok); + + // Final sanity check on a fresh connection: after all the concurrent + // churn, the table data itself must be byte-for-byte what was written + // before the server ever started - `flush`/`compact` change physical + // layout, never live values. + var client = try client_mod.Client.connect(allocator, server.boundAddress()); + defer client.close(); + const finalAll = try client.tableSelect(allocator, "people", "", 0, 0, &.{}); + defer protocol.freeRows(allocator, finalAll); + try std.testing.expectEqual(@as(usize, rowCount), finalAll.len); + const finalWarsaw = try client.tableSelect(allocator, "people", "city eq s:warsaw", 0, 0, &.{}); + defer protocol.freeRows(allocator, finalWarsaw); + try std.testing.expectEqual(@as(usize, rowCount / 3), finalWarsaw.len); + + // Precise check of the multi-column order-by: city groups sorted + // alphabetically (gdansk < krakow < warsaw), each group internally by + // age (== pk*2) descending. pk%3 maps 2->gdansk, 1->krakow, 0->warsaw, + // so each group's highest pk (== highest age) leads that group. + const finalOrdered = try client.tableSelect(allocator, "people", "", 0, 0, &cityOrderBy); + defer protocol.freeRows(allocator, finalOrdered); + try std.testing.expectEqual(@as(usize, rowCount), finalOrdered.len); + try std.testing.expectEqualStrings("gdansk", finalOrdered[0].get("city").?.text); + try std.testing.expectEqual(@as(i64, 29), finalOrdered[0].get(querymod.pkField).?.int); + try std.testing.expectEqualStrings("gdansk", finalOrdered[9].get("city").?.text); + try std.testing.expectEqual(@as(i64, 2), finalOrdered[9].get(querymod.pkField).?.int); + try std.testing.expectEqualStrings("krakow", finalOrdered[10].get("city").?.text); + try std.testing.expectEqual(@as(i64, 28), finalOrdered[10].get(querymod.pkField).?.int); + try std.testing.expectEqualStrings("krakow", finalOrdered[19].get("city").?.text); + try std.testing.expectEqual(@as(i64, 1), finalOrdered[19].get(querymod.pkField).?.int); + try std.testing.expectEqualStrings("warsaw", finalOrdered[20].get("city").?.text); + try std.testing.expectEqual(@as(i64, 30), finalOrdered[20].get(querymod.pkField).?.int); + try std.testing.expectEqualStrings("warsaw", finalOrdered[29].get("city").?.text); + try std.testing.expectEqual(@as(i64, 3), finalOrdered[29].get(querymod.pkField).?.int); +} + +// `table_join` shares the exact same underlying-sstable-reshaping hazard as +// `table_select` above (see that test's header comment) - `flush`/`compact` +// swap `Db.sstables` out from under any in-flight reader while +// `Server.dispatch`'s `self.mutex` is what keeps a join's two-table scan +// from tearing. Ten customers / ten orders, deliberately NOT 1:1: two +// customers (9, 10) never appear in `orders`, and two orders (9, 10) carry a +// `customer_id` nobody has, so inner/left/right/full each land on a +// different, independently-checkable pair count instead of degenerating to +// the same number. +fn seedJoinConcurrencyTables(allocator: Allocator, db: *Db) !void { + var customers = try querymod.Table.open(db, allocator, "customers"); + defer customers.deinit(); + var pk: i64 = 1; + while (pk <= 10) : (pk += 1) { + var buf: [16]u8 = undefined; + const name = try std.fmt.bufPrint(&buf, "cust{d}", .{pk}); + try customers.put(.{ .int = pk }, &.{.{ .name = "name", .value = .{ .text = name } }}); + } + + var orders = try querymod.Table.open(db, allocator, "orders"); + defer orders.deinit(); + pk = 1; + while (pk <= 8) : (pk += 1) { + try orders.put(.{ .int = pk }, &.{ + .{ .name = "customer_id", .value = .{ .int = pk } }, + .{ .name = "total", .value = .{ .int = pk * 10 } }, + }); + } + try orders.put(.{ .int = 9 }, &.{ + .{ .name = "customer_id", .value = .{ .int = 999 } }, + .{ .name = "total", .value = .{ .int = 900 } }, + }); + try orders.put(.{ .int = 10 }, &.{ + .{ .name = "customer_id", .value = .{ .int = 998 } }, + .{ .name = "total", .value = .{ .int = 800 } }, + }); +} + +test "table_join survives concurrent flush and compact reshaping the underlying sstables" { + const allocator = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + const path = try tmp.dir.realpathAlloc(allocator, "."); + defer allocator.free(path); +