diff --git a/src/server.zig b/src/server.zig index d5ee9f9..f8f1ee7 100644 --- a/src/server.zig +++ b/src/server.zig @@ -7066,3 +7066,166 @@ test "ttl_op survives many real concurrent connections racing writes against swe ok: bool = false, }; + const writeWorker = struct { + fn run(ctx: *WriteCtx) void { + var client = client_mod.Client.connect(ctx.allocator, ctx.addr) catch return; + defer client.close(); + var i: usize = 0; + while (i < perWriter) : (i += 1) { + var buf: [32]u8 = undefined; + const key = std.fmt.bufPrint(&buf, "ttl-conc-{d}-{d}", .{ ctx.thread, i }) catch return; + client.set(key, "v") catch return; + client.ttlExpire(key, 0) catch return; + } + ctx.ok = true; + } + }.run; + + const SweepCtx = struct { + allocator: Allocator, + addr: std.net.Address, + reaped: std.atomic.Value(u64) = std.atomic.Value(u64).init(0), + ok: bool = false, + }; + + const sweepWorker = struct { + fn run(ctx: *SweepCtx) void { + var client = client_mod.Client.connect(ctx.allocator, ctx.addr) catch return; + defer client.close(); + var i: usize = 0; + while (i < sweepsPerSweeper) : (i += 1) { + const n = client.ttlSweep() catch return; + _ = ctx.reaped.fetchAdd(n, .monotonic); + } + ctx.ok = true; + } + }.run; + + var writeContexts: [writerCount]WriteCtx = undefined; + var writeThreads: [writerCount]std.Thread = undefined; + var sweepContexts: [sweeperCount]SweepCtx = undefined; + var sweepThreads: [sweeperCount]std.Thread = undefined; + + for (0..sweeperCount) |i| { + sweepContexts[i] = .{ .allocator = allocator, .addr = server.boundAddress() }; + sweepThreads[i] = try std.Thread.spawn(.{}, sweepWorker, .{&sweepContexts[i]}); + } + for (0..writerCount) |i| { + writeContexts[i] = .{ .allocator = allocator, .addr = server.boundAddress(), .thread = i }; + writeThreads[i] = try std.Thread.spawn(.{}, writeWorker, .{&writeContexts[i]}); + } + for (0..writerCount) |i| writeThreads[i].join(); + for (0..sweeperCount) |i| sweepThreads[i].join(); + for (0..writerCount) |i| try std.testing.expect(writeContexts[i].ok); + for (0..sweeperCount) |i| try std.testing.expect(sweepContexts[i].ok); + + var client = try client_mod.Client.connect(allocator, server.boundAddress()); + defer client.close(); + + // Any key that lost the race against every sweeper (written after the + // last sweep call landed) is mopped up by one final sweep here. + const finalReaped = try client.ttlSweep(); + + var total: u64 = finalReaped; + for (0..sweeperCount) |i| total += sweepContexts[i].reaped.load(.monotonic); + try std.testing.expectEqual(@as(u64, writerCount * perWriter), total); + + // Spot-check one key from every writer thread: all must be truly gone + // (not just marker-cleared), proving the sweeps were real deletes. + for (0..writerCount) |t| { + var buf: [32]u8 = undefined; + const key = try std.fmt.bufPrint(&buf, "ttl-conc-{d}-0", .{t}); + try std.testing.expectEqual(@as(?[]u8, null), try client.get(allocator, key)); + try std.testing.expectEqual(ttlmod.TtlResult.noKey, try client.ttlQuery(key)); + } +} + +// Mirrors the `ttl_op` concurrency test above but against `coll_ttl_op` +// and a real collection (`Counter`) instead of a flat key - the sibling +// gap the same end-of-#32 note called out explicitly ("ttl_op dziala na +// plaskim keyspace... analogiczny test... jest wykonalny, ale odlozony"). +// `collttl.sweepExpired` (collttl.zig) shares the exact same +// scan-then-delete-within-one-call shape as `ttl.sweepExpired`, so the +// same exactly-once accounting argument applies. +test "coll_ttl_op survives many real concurrent connections racing writes against sweeps" { + 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 writerCount = 6; + const perWriter = 40; + const sweeperCount = 3; + const sweepsPerSweeper = 200; + + const WriteCtx = struct { + allocator: Allocator, + addr: std.net.Address, + thread: usize, + ok: bool = false, + }; + + const writeWorker = struct { + fn run(ctx: *WriteCtx) void { + var client = client_mod.Client.connect(ctx.allocator, ctx.addr) catch return; + defer client.close(); + var i: usize = 0; + while (i < perWriter) : (i += 1) { + var buf: [32]u8 = undefined; + const name = std.fmt.bufPrint(&buf, "ttlcoll-conc-{d}-{d}", .{ ctx.thread, i }) catch return; + var reply = client.collOp(ctx.allocator, .counter_incrby, name, &[_][]const u8{"1"}) catch return; + reply.deinit(ctx.allocator); + client.collTtlExpire(.counter, name, 0) catch return; + } + ctx.ok = true; + } + }.run; + + const SweepCtx = struct { + allocator: Allocator, + addr: std.net.Address, + reaped: std.atomic.Value(u64) = std.atomic.Value(u64).init(0), + ok: bool = false, + }; + + const sweepWorker = struct { + fn run(ctx: *SweepCtx) void { + var client = client_mod.Client.connect(ctx.allocator, ctx.addr) catch return; + defer client.close(); + var i: usize = 0; + while (i < sweepsPerSweeper) : (i += 1) { + const n = client.collTtlSweep() catch return; + _ = ctx.reaped.fetchAdd(n, .monotonic); + } + ctx.ok = true; + } + }.run; + + var writeContexts: [writerCount]WriteCtx = undefined; + var writeThreads: [writerCount]std.Thread = undefined; + var sweepContexts: [sweeperCount]SweepCtx = undefined; + var sweepThreads: [sweeperCount]std.Thread = undefined; + + for (0..sweeperCount) |i| { + sweepContexts[i] = .{ .allocator = allocator, .addr = server.boundAddress() }; + sweepThreads[i] = try std.Thread.spawn(.{}, sweepWorker, .{&sweepContexts[i]}); + } + for (0..writerCount) |i| { + writeContexts[i] = .{ .allocator = allocator, .addr = server.boundAddress(), .thread = i }; + writeThreads[i] = try std.Thread.spawn(.{}, writeWorker, .{&writeContexts[i]}); + } + for (0..writerCount) |i| writeThreads[i].join(); + for (0..sweeperCount) |i| sweepThreads[i].join(); + for (0..writerCount) |i| try std.testing.expect(writeContexts[i].ok); + for (0..sweeperCount) |i| try std.testing.expect(sweepContexts[i].ok); +