feat: implement ReaderCtx and readerWorker to server
This commit is contained in:
dualfroz committed
1 parent
1f8f067bfc
commit
4e3fa110d0
1 file changed
+159
+159
@@ -5329,3 +5329,162 @@ test "table_join survives concurrent flush and compact reshaping the underlying
|
|||||||
const path = try tmp.dir.realpathAlloc(allocator, ".");
|
const path = try tmp.dir.realpathAlloc(allocator, ".");
|
||||||
defer allocator.free(path);
|
defer allocator.free(path);
|
||||||
|
|
||||||
|
// Small target size so a handful of `flush`/`compact` calls do real
|
||||||
|
// structural work instead of needing megabytes of data first - same
|
||||||
|
// trick as the `table_select` concurrency test above.
|
||||||
|
var db = try Db.open(allocator, path, .{ .sstableTargetBytes = 512 });
|
||||||
|
defer db.close();
|
||||||
|
try seedJoinConcurrencyTables(allocator, &db);
|
||||||
|
try db.flush();
|
||||||
|
|
||||||
|
const address = try std.net.Address.parseIp("127.0.0.1", 0);
|
||||||
|
var server = try Server.init(allocator, &db, address);
|
||||||
|
try server.start();
|
||||||
|
defer server.stop();
|
||||||
|
|
||||||
|
const client_mod = @import("client.zig");
|
||||||
|
const readerThreads = 4;
|
||||||
|
const readerIters = 20;
|
||||||
|
const mutatorThreads = 2;
|
||||||
|
const mutatorIters = 15;
|
||||||
|
const matchedCount = 8;
|
||||||
|
const customerCount = 10;
|
||||||
|
|
||||||
|
const ReaderCtx = struct {
|
||||||
|
allocator: Allocator,
|
||||||
|
addr: std.net.Address,
|
||||||
|
ok: bool = false,
|
||||||
|
};
|
||||||
|
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 inner = client.tableJoin(ctx.allocator, .inner, "customers", querymod.pkField, "orders", "customer_id", 0, 0, "", "") catch return;
|
||||||
|
defer protocol.freeJoinPairs(ctx.allocator, inner);
|
||||||
|
if (inner.len != matchedCount) return;
|
||||||
|
for (inner) |pair| {
|
||||||
|
const left = pair.left orelse return;
|
||||||
|
const right = pair.right orelse return;
|
||||||
|
const pk = left.get(querymod.pkField) orelse return;
|
||||||
|
const total = right.get("total") orelse return;
|
||||||
|
if (total.int != pk.int * 10) return;
|
||||||
|
}
|
||||||
|
|
||||||
|
const left = client.tableJoin(ctx.allocator, .left, "customers", querymod.pkField, "orders", "customer_id", 0, 0, "", "") catch return;
|
||||||
|
defer protocol.freeJoinPairs(ctx.allocator, left);
|
||||||
|
if (left.len != customerCount) return;
|
||||||
|
|
||||||
|
const right = client.tableJoin(ctx.allocator, .right, "customers", querymod.pkField, "orders", "customer_id", 0, 0, "", "") catch return;
|
||||||
|
defer protocol.freeJoinPairs(ctx.allocator, right);
|
||||||
|
if (right.len != matchedCount + 2) return;
|
||||||
|
|
||||||
|
const full = client.tableJoin(ctx.allocator, .full, "customers", querymod.pkField, "orders", "customer_id", 0, 0, "", "") catch return;
|
||||||
|
defer protocol.freeJoinPairs(ctx.allocator, full);
|
||||||
|
if (full.len != customerCount + 2) return;
|
||||||
|
}
|
||||||
|
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, "join-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: all four join shapes are
|
||||||
|
// still exactly what they were before the server ever started - the
|
||||||
|
// concurrent churn wrote only unrelated `join-churn-*` keys and
|
||||||
|
// reshaped physical sstable layout, never touched `customers`/`orders`
|
||||||
|
// values.
|
||||||
|
var client = try client_mod.Client.connect(allocator, server.boundAddress());
|
||||||
|
defer client.close();
|
||||||
|
const finalInner = try client.tableJoin(allocator, .inner, "customers", querymod.pkField, "orders", "customer_id", 0, 0, "", "");
|
||||||
|
defer protocol.freeJoinPairs(allocator, finalInner);
|
||||||
|
try std.testing.expectEqual(@as(usize, matchedCount), finalInner.len);
|
||||||
|
const finalFull = try client.tableJoin(allocator, .full, "customers", querymod.pkField, "orders", "customer_id", 0, 0, "", "");
|
||||||
|
defer protocol.freeJoinPairs(allocator, finalFull);
|
||||||
|
try std.testing.expectEqual(@as(usize, customerCount + 2), finalFull.len);
|
||||||
|
}
|
||||||
|
|
||||||
|
test "server without enableAuth accepts any auth attempt and every op" {
|
||||||
|
const allocator = std.testing.allocator;
|
||||||
|
var tmp = std.testing.tmpDir(.{});
|
||||||
|
defer tmp.cleanup();
|
||||||
|
const path = try tmp.dir.realpathAlloc(allocator, ".");
|
||||||
|
defer allocator.free(path);
|
||||||
|
|
||||||
|
var db = try testDb(allocator, path);
|
||||||
|
defer db.close();
|
||||||
|
|
||||||
|
const address = try std.net.Address.parseIp("127.0.0.1", 0);
|
||||||
|
var server = try Server.init(allocator, &db, address);
|
||||||
|
try server.start();
|
||||||
|
defer server.stop();
|
||||||
|
|
||||||
|
const client_mod = @import("client.zig");
|
||||||
|
var client = try client_mod.Client.connect(allocator, server.boundAddress());
|
||||||
|
defer client.close();
|
||||||
|
|
||||||
|
try client.authenticate("nobody", "wrong");
|
||||||
|
try client.set("k", "v");
|
||||||
|
}
|
||||||
|
|
||||||
|
test "server with auth enabled rejects unauthenticated requests, ping excepted" {
|
||||||
|
const allocator = std.testing.allocator;
|
||||||
|
var tmp = std.testing.tmpDir(.{});
|
||||||
|
defer tmp.cleanup();
|
||||||
|
const path = try tmp.dir.realpathAlloc(allocator, ".");
|
||||||
|
defer allocator.free(path);
|
||||||
|
|
||||||
|
var db = try testDb(allocator, path);
|
||||||
|
defer db.close();
|
||||||
|
const store = authmod.Store.init(&db);
|
||||||
|
try store.createUser(allocator, "alice", "hunter2", .readwrite, 16);
|
||||||
|
|
||||||
|
const address = try std.net.Address.parseIp("127.0.0.1", 0);
|
||||||
|
var server = try Server.init(allocator, &db, address);
|
||||||
|
server.enableAuth(store);
|
||||||
|
try server.start();
|
||||||
|
defer server.stop();
|
||||||
|
|
||||||
|
const client_mod = @import("client.zig");
|
||||||
|
var client = try client_mod.Client.connect(allocator, server.boundAddress());
|
||||||
|
defer client.close();
|
||||||
|
|
||||||
Reference in new issue
Block a user