feat: add incrWorker and run to server
This commit is contained in:
dualfroz committed
1 parent
aa1e1992e4
commit
638c66a0ab
1 file changed
+158
+158
@@ -6753,3 +6753,161 @@ test "coll_op survives many real concurrent connections hammering the same count
|
||||
ok: bool = false,
|
||||
};
|
||||
|
||||
const incrWorker = struct {
|
||||
fn run(ctx: *IncrCtx) void {
|
||||
var client = client_mod.Client.connect(ctx.allocator, ctx.addr) catch return;
|
||||
defer client.close();
|
||||
var i: usize = 0;
|
||||
while (i < perThread) : (i += 1) {
|
||||
var reply = client.collOp(ctx.allocator, .counter_incrby, "shared-counter", &[_][]const u8{"1"}) catch return;
|
||||
reply.deinit(ctx.allocator);
|
||||
}
|
||||
ctx.ok = true;
|
||||
}
|
||||
}.run;
|
||||
|
||||
var incrContexts: [threadCount]IncrCtx = undefined;
|
||||
var incrThreads: [threadCount]std.Thread = undefined;
|
||||
for (0..threadCount) |i| {
|
||||
incrContexts[i] = .{ .allocator = allocator, .addr = server.boundAddress() };
|
||||
incrThreads[i] = try std.Thread.spawn(.{}, incrWorker, .{&incrContexts[i]});
|
||||
}
|
||||
for (0..threadCount) |i| incrThreads[i].join();
|
||||
for (0..threadCount) |i| try std.testing.expect(incrContexts[i].ok);
|
||||
|
||||
var client = try client_mod.Client.connect(allocator, server.boundAddress());
|
||||
defer client.close();
|
||||
var total = try client.collOp(allocator, .counter_get, "shared-counter", &.{});
|
||||
defer total.deinit(allocator);
|
||||
// Every increment is a separate request from a separate connection, so
|
||||
// a missing lock around `runColl`'s read-modify-write would show up
|
||||
// here as a total strictly less than threadCount*perThread (lost
|
||||
// updates), not as a crash - the mutex in `Server.dispatch` must make
|
||||
// this land on the exact total every single run.
|
||||
try std.testing.expectEqual(@as(i64, threadCount * perThread), total.int);
|
||||
}
|
||||
|
||||
// Same shape as the counter test above but for a structural collection
|
||||
// (`Set`) instead of a single scalar: many connections each add their OWN
|
||||
// disjoint members to one shared set, so a race would show up as a
|
||||
// missing member (an add silently lost to a concurrent writer) rather
|
||||
// than a wrong total, which is a different failure mode than incrBy's
|
||||
// lost-update and worth its own proof.
|
||||
test "coll_op survives many real concurrent connections adding disjoint members to the same set" {
|
||||
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");
|
||||
const threadCount = 8;
|
||||
const perThread = 50;
|
||||
|
||||
const AddCtx = struct {
|
||||
allocator: Allocator,
|
||||
addr: std.net.Address,
|
||||
thread: usize,
|
||||
ok: bool = false,
|
||||
};
|
||||
|
||||
const addWorker = struct {
|
||||
fn run(ctx: *AddCtx) void {
|
||||
var client = client_mod.Client.connect(ctx.allocator, ctx.addr) catch return;
|
||||
defer client.close();
|
||||
var i: usize = 0;
|
||||
while (i < perThread) : (i += 1) {
|
||||
var buf: [32]u8 = undefined;
|
||||
const member = std.fmt.bufPrint(&buf, "t{d}-m{d}", .{ ctx.thread, i }) catch return;
|
||||
var reply = client.collOp(ctx.allocator, .set_add, "shared-set", &[_][]const u8{member}) catch return;
|
||||
reply.deinit(ctx.allocator);
|
||||
}
|
||||
ctx.ok = true;
|
||||
}
|
||||
}.run;
|
||||
|
||||
var addContexts: [threadCount]AddCtx = undefined;
|
||||
var addThreads: [threadCount]std.Thread = undefined;
|
||||
for (0..threadCount) |i| {
|
||||
addContexts[i] = .{ .allocator = allocator, .addr = server.boundAddress(), .thread = i };
|
||||
addThreads[i] = try std.Thread.spawn(.{}, addWorker, .{&addContexts[i]});
|
||||
}
|
||||
for (0..threadCount) |i| addThreads[i].join();
|
||||
for (0..threadCount) |i| try std.testing.expect(addContexts[i].ok);
|
||||
|
||||
var client = try client_mod.Client.connect(allocator, server.boundAddress());
|
||||
defer client.close();
|
||||
var card = try client.collOp(allocator, .set_card, "shared-set", &.{});
|
||||
defer card.deinit(allocator);
|
||||
try std.testing.expectEqual(@as(i64, threadCount * perThread), card.int);
|
||||
|
||||
// Spot-check a member from every thread, not just the aggregate count -
|
||||
// a bug that dropped exactly one member per thread but "gained" one
|
||||
// duplicate elsewhere could still pass the cardinality check above.
|
||||
for (0..threadCount) |t| {
|
||||
var buf: [32]u8 = undefined;
|
||||
const member = try std.fmt.bufPrint(&buf, "t{d}-m0", .{t});
|
||||
var isMem = try client.collOp(allocator, .set_ismember, "shared-set", &[_][]const u8{member});
|
||||
defer isMem.deinit(allocator);
|
||||
try std.testing.expectEqual(@as(i64, 1), isMem.int);
|
||||
}
|
||||
}
|
||||
|
||||
test "coll_op applies the same lazy-expiry check as coll_ttl_op before every dispatch" {
|
||||
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 collmod.Hash.set(&db, allocator, "session", "user", "ada");
|
||||
try client.collTtlExpire(.hash, "session", 0);
|
||||
std.time.sleep(5 * std.time.ns_per_ms);
|
||||
|
||||
// The lazy check in `handleCollOp` reaps "session" before hash_get
|
||||
// ever runs, so the field reads back as gone, not as its old value.
|
||||
var got = try client.collOp(allocator, .hash_get, "session", &[_][]const u8{"user"});
|
||||
defer got.deinit(allocator);
|
||||
try std.testing.expect(got == .nil);
|
||||
try std.testing.expect(!try collmod.collectionExists(&db, allocator, .hash, "session"));
|
||||
}
|
||||
|
||||
test "coll_op set_inter/union/diff lazily expire every extra set name, not just the first" {
|
||||
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();
|
||||
|
||||
Reference in new issue
Block a user