feat: add readerWorker and run to server
This commit is contained in:
dualfroz committed
1 parent
eee83a0ec7
commit
1f8f067bfc
1 file changed
+160
+160
@@ -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);
|
||||
|
||||
Reference in new issue
Block a user