server: a query that finds every permit taken waits for one instead of failing, the budget is 64, permit and upstream turns wait on a futex with one thread each, the thread pool has no task limit, and transparent huge pages are off for the process
This commit is contained in:
@@ -91,7 +91,6 @@ pub const Snapshot = struct {
|
||||
idle_timeouts: u64,
|
||||
connection_errors: u64,
|
||||
bad_requests: u64,
|
||||
dropped_no_permit: u64,
|
||||
};
|
||||
|
||||
pub const DohServer = struct {
|
||||
@@ -117,6 +116,7 @@ pub const DohServer = struct {
|
||||
pub const Owner = DohServer;
|
||||
pub const ConnPayload = Payload;
|
||||
pub const serveConn = serveOne;
|
||||
pub const closing = wakeBudgetWaiters;
|
||||
pub const read_buffer_len = recv_buffer_len;
|
||||
pub const write_buffer_len = send_buffer_len;
|
||||
pub const log = std.log.scoped(.doh_server);
|
||||
@@ -165,7 +165,6 @@ pub const DohServer = struct {
|
||||
.idle_timeouts = core.idle_timeouts.load(.monotonic),
|
||||
.connection_errors = core.connection_errors.load(.monotonic),
|
||||
.bad_requests = self.stats.bad_requests.load(.monotonic),
|
||||
.dropped_no_permit = core.dropped_no_permit.load(.monotonic),
|
||||
};
|
||||
}
|
||||
|
||||
@@ -174,6 +173,10 @@ pub const DohServer = struct {
|
||||
self.core.serve(io);
|
||||
}
|
||||
|
||||
fn wakeBudgetWaiters(self: *DohServer, io: std.Io) void {
|
||||
self.budget.wakeWaiters(io);
|
||||
}
|
||||
|
||||
pub fn deinit(self: *DohServer, io: std.Io) void {
|
||||
self.core.deinit(io);
|
||||
self.* = undefined;
|
||||
@@ -389,10 +392,10 @@ pub const DohServer = struct {
|
||||
request: *http.Server.Request,
|
||||
query: []const u8,
|
||||
) error{ WriteFailed, HttpExpectationFailed }!Next {
|
||||
const work = self.budget.tryAcquire(io) orelse {
|
||||
listener.bump(&self.core.stats.dropped_no_permit);
|
||||
return .close;
|
||||
};
|
||||
// Bounded like a read: a query still waiting when the idle budget runs
|
||||
// out closes the connection, with the timeout counted by the budget.
|
||||
const permit_deadline: std.Io.Clock.Timestamp = .fromNow(io, self.options.idle_timeout);
|
||||
const work = self.budget.acquire(io, permit_deadline, &self.core.run_state) catch return .close;
|
||||
defer self.budget.release(io, work);
|
||||
|
||||
const outcome = self.handler.handle(
|
||||
|
||||
@@ -60,7 +60,6 @@ pub const StatsSnapshot = struct {
|
||||
tls_handshake_failures: u64,
|
||||
idle_timeouts: u64,
|
||||
connection_errors: u64,
|
||||
dropped_no_permit: u64,
|
||||
};
|
||||
|
||||
pub const DotServer = struct {
|
||||
@@ -83,6 +82,7 @@ pub const DotServer = struct {
|
||||
pub const Owner = DotServer;
|
||||
pub const ConnPayload = Payload;
|
||||
pub const serveConn = serveOne;
|
||||
pub const closing = wakeBudgetWaiters;
|
||||
pub const read_buffer_len = stream_buffer_len;
|
||||
pub const write_buffer_len = stream_buffer_len;
|
||||
pub const log = std.log.scoped(.dot_server);
|
||||
@@ -121,6 +121,10 @@ pub const DotServer = struct {
|
||||
self.core.serve(io);
|
||||
}
|
||||
|
||||
fn wakeBudgetWaiters(self: *DotServer, io: std.Io) void {
|
||||
self.budget.wakeWaiters(io);
|
||||
}
|
||||
|
||||
pub fn deinit(self: *DotServer, io: std.Io) void {
|
||||
self.core.deinit(io);
|
||||
self.* = undefined;
|
||||
@@ -138,7 +142,6 @@ pub const DotServer = struct {
|
||||
.tls_handshake_failures = self.stats.tls_handshake_failures.load(.monotonic),
|
||||
.idle_timeouts = core.idle_timeouts.load(.monotonic),
|
||||
.connection_errors = core.connection_errors.load(.monotonic),
|
||||
.dropped_no_permit = core.dropped_no_permit.load(.monotonic),
|
||||
};
|
||||
}
|
||||
|
||||
@@ -219,11 +222,11 @@ pub const DotServer = struct {
|
||||
},
|
||||
}
|
||||
|
||||
// There is no framing for "no answer", so the connection ends.
|
||||
const work = self.budget.tryAcquire(io) orelse {
|
||||
listener.bump(&stats.dropped_no_permit);
|
||||
return;
|
||||
};
|
||||
// Bounded like a read: a query still waiting when the idle budget
|
||||
// runs out ends the connection, with the timeout counted by the
|
||||
// budget. There is no framing for "no answer".
|
||||
const permit_deadline: std.Io.Clock.Timestamp = .fromNow(io, budget);
|
||||
const work = self.budget.acquire(io, permit_deadline, &self.core.run_state) catch return;
|
||||
defer self.budget.release(io, work);
|
||||
|
||||
const outcome = self.handler.handle(
|
||||
|
||||
@@ -91,10 +91,6 @@ pub const CoreStats = struct {
|
||||
accept_errors: std.atomic.Value(u64) = .init(0),
|
||||
connection_errors: std.atomic.Value(u64) = .init(0),
|
||||
idle_timeouts: std.atomic.Value(u64) = .init(0),
|
||||
/// A DNS listener read a whole query and found every permit of the shared
|
||||
/// query budget taken, so the connection ended unanswered. The web
|
||||
/// listener answers no DNS and keeps this at zero.
|
||||
dropped_no_permit: std.atomic.Value(u64) = .init(0),
|
||||
/// The most slots that have been taken at once (specs/memory-budget.md
|
||||
/// item 0). A gauge, so no listener's `Snapshot` carries it: those export
|
||||
/// every field as a counter. Written only under `Core.mutex`.
|
||||
@@ -268,6 +264,10 @@ pub fn decideClaim(conns: anytype, shutdown_begun: bool) Claim {
|
||||
/// - `initPayload` / `deinitPayload` — for a payload that owns memory (the web
|
||||
/// listener's per-connection arena). `initPayload` runs inside `listen`,
|
||||
/// `deinitPayload` inside `deinit` after every connection task has joined.
|
||||
/// - `closing: fn (*Owner, std.Io) void` — runs in `deinit` once `.closing` is
|
||||
/// published, for a connection blocked on something other than its socket,
|
||||
/// which the shutdown of each stream cannot reach. The DNS listeners wake
|
||||
/// their query-budget waiters here.
|
||||
pub fn Core(comptime Cfg: type) type {
|
||||
return struct {
|
||||
const Self = @This();
|
||||
@@ -336,6 +336,7 @@ pub fn Core(comptime Cfg: type) type {
|
||||
|
||||
pub fn deinit(self: *Self, io: std.Io) void {
|
||||
const was_serving = self.run_state.swap(.closing, .acq_rel) == .serving;
|
||||
if (@hasDecl(Cfg, "closing")) Cfg.closing(self.owner(), io);
|
||||
|
||||
// Shutting the listening socket down is the documented way to
|
||||
// unblock a pending `accept`: it fails with
|
||||
|
||||
+311
-18
@@ -7,21 +7,29 @@
|
||||
//! it when its reply is written, so a keep-alive connection between requests
|
||||
//! holds none.
|
||||
//!
|
||||
//! Nothing waits for a permit. A listener that finds the pool empty drops the
|
||||
//! query and counts it, because a wait would only turn a visible drop into
|
||||
//! invisible latency, and every permit is also the tasks its query may start,
|
||||
//! which `app.concurrent_limit` adds up from `permits`.
|
||||
//! A query that finds every permit taken waits for one, on its own thread,
|
||||
//! rather than being dropped or answered SERVFAIL: a stub resolver treats
|
||||
//! either as a failed lookup, while a short wait is invisible to it. Each held
|
||||
//! permit is bounded by the upstream deadline and the stream write budget, so a
|
||||
//! wait always ends. A UDP listener waits with its datagram staged and stops
|
||||
//! receiving meanwhile, so the backlog sits in the socket's kernel buffer,
|
||||
//! where `/proc/net/udp` counts what overflows. A stream query waits at most
|
||||
//! its connection's idle budget. Every wait is counted.
|
||||
|
||||
const std = @import("std");
|
||||
const Allocator = std.mem.Allocator;
|
||||
|
||||
const handler = @import("handler.zig");
|
||||
const listener = @import("listener.zig");
|
||||
const transport = @import("../upstream/transport.zig");
|
||||
const WakeWord = @import("../platform/wake_word.zig").WakeWord;
|
||||
|
||||
/// The most DNS queries answered at once, across every listener. The Pi's
|
||||
/// highest reading of `nxdns_dns_in_flight_peak` was 27 over 37 hours of
|
||||
/// household traffic, and a query beyond this is dropped and counted.
|
||||
pub const permits = 32;
|
||||
/// highest reading of `nxdns_dns_in_flight_peak` was 27 over a day of
|
||||
/// household traffic; a burst of 64 simultaneous uncached queries was answered
|
||||
/// whole on two upstreams with the slowest at 0.2 s. Each answered query can
|
||||
/// hold up to three threads while its exchange runs.
|
||||
pub const permits = 64;
|
||||
|
||||
/// The EDNS ceiling the UDP listener advertises. A larger datagram arrives
|
||||
/// truncated and is dropped rather than guessed at.
|
||||
@@ -48,6 +56,26 @@ pub const Work = struct {
|
||||
scratch: handler.Scratch,
|
||||
};
|
||||
|
||||
/// Why `acquire` returned without a permit.
|
||||
pub const AcquireError = error{
|
||||
/// The waiting task is being torn down.
|
||||
Canceled,
|
||||
/// The caller's deadline passed. Counted in `wait_timeouts`.
|
||||
Timeout,
|
||||
/// The caller's listener left `.serving`.
|
||||
Stopped,
|
||||
};
|
||||
|
||||
pub const Snapshot = struct {
|
||||
/// Queries that found every permit taken and had to wait.
|
||||
waits: u64,
|
||||
wait_seconds: f64,
|
||||
/// Waits that ran out of their deadline. Only stream queries have one.
|
||||
wait_timeouts: u64,
|
||||
/// Queries waiting right now.
|
||||
waiting: u64,
|
||||
};
|
||||
|
||||
pub const QueryBudget = struct {
|
||||
gpa: Allocator,
|
||||
work: []Work,
|
||||
@@ -56,6 +84,11 @@ pub const QueryBudget = struct {
|
||||
/// `free[0..free_len]`.
|
||||
free: []u16,
|
||||
free_len: usize,
|
||||
released: WakeWord,
|
||||
waits: std.atomic.Value(u64),
|
||||
wait_ns: std.atomic.Value(u64),
|
||||
wait_timeouts: std.atomic.Value(u64),
|
||||
waiting: std.atomic.Value(u64),
|
||||
|
||||
pub fn init(gpa: Allocator, count: u16) Allocator.Error!QueryBudget {
|
||||
std.debug.assert(count > 0);
|
||||
@@ -63,21 +96,52 @@ pub const QueryBudget = struct {
|
||||
errdefer gpa.free(work);
|
||||
const free = try gpa.alloc(u16, count);
|
||||
for (free, 0..) |*slot, index| slot.* = @intCast(index);
|
||||
return .{ .gpa = gpa, .work = work, .mutex = .init, .free = free, .free_len = count };
|
||||
return .{
|
||||
.gpa = gpa,
|
||||
.work = work,
|
||||
.mutex = .init,
|
||||
.free = free,
|
||||
.free_len = count,
|
||||
.released = .{},
|
||||
.waits = .init(0),
|
||||
.wait_ns = .init(0),
|
||||
.wait_timeouts = .init(0),
|
||||
.waiting = .init(0),
|
||||
};
|
||||
}
|
||||
|
||||
/// Every listener has stopped, so every permit is back.
|
||||
pub fn deinit(self: *QueryBudget) void {
|
||||
std.debug.assert(self.free_len == self.work.len);
|
||||
std.debug.assert(self.waiting.load(.monotonic) == 0);
|
||||
self.gpa.free(self.free);
|
||||
self.gpa.free(self.work);
|
||||
self.* = undefined;
|
||||
}
|
||||
|
||||
/// A permit, waiting for one while every permit is taken, until `deadline`
|
||||
/// (`null`: no deadline) or until `run_state` leaves `.serving`. A listener
|
||||
/// that publishes `.closing` calls `wakeWaiters` so a waiter sees it.
|
||||
pub fn acquire(
|
||||
self: *QueryBudget,
|
||||
io: std.Io,
|
||||
deadline: ?std.Io.Clock.Timestamp,
|
||||
run_state: *const std.atomic.Value(listener.State),
|
||||
) AcquireError!*Work {
|
||||
if (self.tryAcquire(io)) |work| return work;
|
||||
|
||||
_ = self.waits.fetchAdd(1, .monotonic);
|
||||
_ = self.waiting.fetchAdd(1, .monotonic);
|
||||
const started = std.Io.Clock.awake.now(io);
|
||||
const outcome = self.waitForPermit(io, deadline, run_state);
|
||||
const elapsed = started.durationTo(std.Io.Clock.awake.now(io)).toNanoseconds();
|
||||
_ = self.wait_ns.fetchAdd(@intCast(@max(elapsed, 0)), .monotonic);
|
||||
_ = self.waiting.fetchSub(1, .monotonic);
|
||||
return outcome;
|
||||
}
|
||||
|
||||
/// A permit, or null when every one is taken. Waits for the bookkeeping
|
||||
/// mutex, which no holder keeps across any Io, but never for a permit:
|
||||
/// treating a contended lock as an empty pool would drop queries the
|
||||
/// budget has room for.
|
||||
/// mutex, which no holder keeps across any Io, but never for a permit.
|
||||
pub fn tryAcquire(self: *QueryBudget, io: std.Io) ?*Work {
|
||||
self.mutex.lockUncancelable(io);
|
||||
defer self.mutex.unlock(io);
|
||||
@@ -90,11 +154,20 @@ pub const QueryBudget = struct {
|
||||
const index = (@intFromPtr(work) - @intFromPtr(self.work.ptr)) / @sizeOf(Work);
|
||||
std.debug.assert(&self.work[index] == work);
|
||||
|
||||
self.mutex.lockUncancelable(io);
|
||||
defer self.mutex.unlock(io);
|
||||
std.debug.assert(self.free_len < self.work.len);
|
||||
self.free[self.free_len] = @intCast(index);
|
||||
self.free_len += 1;
|
||||
{
|
||||
self.mutex.lockUncancelable(io);
|
||||
defer self.mutex.unlock(io);
|
||||
std.debug.assert(self.free_len < self.work.len);
|
||||
self.free[self.free_len] = @intCast(index);
|
||||
self.free_len += 1;
|
||||
}
|
||||
self.released.wakeOne(io);
|
||||
}
|
||||
|
||||
/// Wakes every waiter so each re-reads its listener's `run_state`. A
|
||||
/// listener calls this after publishing `.closing`.
|
||||
pub fn wakeWaiters(self: *QueryBudget, io: std.Io) void {
|
||||
self.released.wakeAll(io);
|
||||
}
|
||||
|
||||
/// Permits held right now.
|
||||
@@ -103,10 +176,61 @@ pub const QueryBudget = struct {
|
||||
defer self.mutex.unlock(io);
|
||||
return self.work.len - self.free_len;
|
||||
}
|
||||
|
||||
/// The counters, read one at a time; a scrape mid-wait can see a wait
|
||||
/// counted before its seconds are.
|
||||
pub fn snapshot(self: *const QueryBudget) Snapshot {
|
||||
const ns: f64 = @floatFromInt(self.wait_ns.load(.monotonic));
|
||||
return .{
|
||||
.waits = self.waits.load(.monotonic),
|
||||
.wait_seconds = ns / std.time.ns_per_s,
|
||||
.wait_timeouts = self.wait_timeouts.load(.monotonic),
|
||||
.waiting = self.waiting.load(.monotonic),
|
||||
};
|
||||
}
|
||||
|
||||
fn waitForPermit(
|
||||
self: *QueryBudget,
|
||||
io: std.Io,
|
||||
deadline: ?std.Io.Clock.Timestamp,
|
||||
run_state: *const std.atomic.Value(listener.State),
|
||||
) AcquireError!*Work {
|
||||
while (true) {
|
||||
// Read before the check, so a release or a `wakeWaiters` that
|
||||
// lands after the check moves the word and the wait returns at
|
||||
// once.
|
||||
const seen = self.released.observe();
|
||||
if (run_state.load(.acquire) != .serving) return self.leave(io, error.Stopped);
|
||||
// Before the take, so a permit freed just as the deadline passes
|
||||
// does not start a query its caller has already given up on.
|
||||
if (deadline) |at| {
|
||||
if (std.Io.Clock.Timestamp.now(io, at.clock).compare(.gte, at)) {
|
||||
_ = self.wait_timeouts.fetchAdd(1, .monotonic);
|
||||
return self.leave(io, error.Timeout);
|
||||
}
|
||||
}
|
||||
if (self.tryAcquire(io)) |work| return work;
|
||||
self.released.wait(io, seen, deadline) catch return self.leave(io, error.Canceled);
|
||||
}
|
||||
}
|
||||
|
||||
/// This waiter may have consumed the one wake a release sent, so a permit
|
||||
/// still free is handed on rather than left for nobody.
|
||||
fn leave(self: *QueryBudget, io: std.Io, err: AcquireError) AcquireError {
|
||||
const free_len = free: {
|
||||
self.mutex.lockUncancelable(io);
|
||||
defer self.mutex.unlock(io);
|
||||
break :free self.free_len;
|
||||
};
|
||||
if (free_len != 0) self.released.wakeOne(io);
|
||||
return err;
|
||||
}
|
||||
};
|
||||
|
||||
const testing = std.testing;
|
||||
|
||||
var test_serving: std.atomic.Value(listener.State) = .init(.serving);
|
||||
|
||||
test "a budget hands out every permit once, then refuses" {
|
||||
var budget: QueryBudget = try .init(testing.allocator, 3);
|
||||
defer budget.deinit();
|
||||
@@ -135,6 +259,175 @@ test "a released permit is handed out again" {
|
||||
budget.release(io, second);
|
||||
}
|
||||
|
||||
test "a free permit is taken without counting a wait" {
|
||||
var budget: QueryBudget = try .init(testing.allocator, 1);
|
||||
defer budget.deinit();
|
||||
const io = testing.io;
|
||||
|
||||
const work = try budget.acquire(io, null, &test_serving);
|
||||
budget.release(io, work);
|
||||
try testing.expectEqual(@as(u64, 0), budget.snapshot().waits);
|
||||
}
|
||||
|
||||
test "a wait past its deadline times out, holds nothing and is counted" {
|
||||
var budget: QueryBudget = try .init(testing.allocator, 1);
|
||||
defer budget.deinit();
|
||||
const io = testing.io;
|
||||
|
||||
const held = budget.tryAcquire(io).?;
|
||||
defer budget.release(io, held);
|
||||
const deadline: std.Io.Clock.Timestamp = .fromNow(io, .{ .raw = .fromMilliseconds(20), .clock = .awake });
|
||||
try testing.expectError(error.Timeout, budget.acquire(io, deadline, &test_serving));
|
||||
|
||||
const snap = budget.snapshot();
|
||||
try testing.expectEqual(@as(u64, 1), snap.waits);
|
||||
try testing.expectEqual(@as(u64, 1), snap.wait_timeouts);
|
||||
try testing.expectEqual(@as(u64, 0), snap.waiting);
|
||||
try testing.expect(snap.wait_seconds >= 0.02);
|
||||
try testing.expectEqual(@as(usize, 1), budget.taken(io));
|
||||
}
|
||||
|
||||
test "a listener that is not serving gets no permit, even a free one" {
|
||||
var budget: QueryBudget = try .init(testing.allocator, 1);
|
||||
defer budget.deinit();
|
||||
const io = testing.io;
|
||||
|
||||
const held = budget.tryAcquire(io).?;
|
||||
var closing: std.atomic.Value(listener.State) = .init(.closing);
|
||||
try testing.expectError(error.Stopped, budget.acquire(io, null, &closing));
|
||||
budget.release(io, held);
|
||||
try testing.expectEqual(@as(usize, 0), budget.taken(io));
|
||||
}
|
||||
|
||||
fn acquireAndHold(
|
||||
budget: *QueryBudget,
|
||||
io: std.Io,
|
||||
deadline: ?std.Io.Clock.Timestamp,
|
||||
run_state: *const std.atomic.Value(listener.State),
|
||||
) AcquireError!*Work {
|
||||
return budget.acquire(io, deadline, run_state);
|
||||
}
|
||||
|
||||
fn waitUntilWaiting(budget: *QueryBudget, io: std.Io, count: u64) !void {
|
||||
const tick: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(1), .clock = .awake };
|
||||
while (budget.waiting.load(.monotonic) != count) try tick.sleep(io);
|
||||
}
|
||||
|
||||
test "a release wakes a waiter and hands it the permit" {
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var budget: QueryBudget = try .init(testing.allocator, 1);
|
||||
defer budget.deinit();
|
||||
const held = budget.tryAcquire(io).?;
|
||||
|
||||
var waiter = io.concurrent(acquireAndHold, .{ &budget, io, null, &test_serving }) catch |err| switch (err) {
|
||||
error.ConcurrencyUnavailable => {
|
||||
budget.release(io, held);
|
||||
return error.SkipZigTest;
|
||||
},
|
||||
};
|
||||
try waitUntilWaiting(&budget, io, 1);
|
||||
budget.release(io, held);
|
||||
|
||||
const got = try waiter.await(io);
|
||||
try testing.expectEqual(held, got);
|
||||
budget.release(io, got);
|
||||
try testing.expectEqual(@as(u64, 1), budget.snapshot().waits);
|
||||
}
|
||||
|
||||
test "a canceled waiter hands its wake on to a peer" {
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var budget: QueryBudget = try .init(testing.allocator, 1);
|
||||
defer budget.deinit();
|
||||
const held = budget.tryAcquire(io).?;
|
||||
|
||||
var canceled = io.concurrent(acquireAndHold, .{ &budget, io, null, &test_serving }) catch |err| switch (err) {
|
||||
error.ConcurrencyUnavailable => {
|
||||
budget.release(io, held);
|
||||
return error.SkipZigTest;
|
||||
},
|
||||
};
|
||||
var peer = io.concurrent(acquireAndHold, .{ &budget, io, null, &test_serving }) catch |err| switch (err) {
|
||||
error.ConcurrencyUnavailable => {
|
||||
_ = canceled.cancel(io) catch {};
|
||||
budget.release(io, held);
|
||||
return error.SkipZigTest;
|
||||
},
|
||||
};
|
||||
try waitUntilWaiting(&budget, io, 2);
|
||||
|
||||
// One release, one wake. Whichever task the wake reaches, the peer must
|
||||
// end up with the permit: the canceled task either never saw the wake or
|
||||
// hands it on as it leaves. A lost wake hangs this test.
|
||||
budget.release(io, held);
|
||||
if (canceled.cancel(io)) |work| {
|
||||
// The cancel lost the race to the wake; give the permit to the peer.
|
||||
budget.release(io, work);
|
||||
} else |err| try testing.expectEqual(error.Canceled, err);
|
||||
|
||||
const got = try peer.await(io);
|
||||
budget.release(io, got);
|
||||
try testing.expectEqual(@as(usize, 0), budget.taken(io));
|
||||
}
|
||||
|
||||
test "wakeWaiters ends a wait whose listener is closing" {
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var budget: QueryBudget = try .init(testing.allocator, 1);
|
||||
defer budget.deinit();
|
||||
const held = budget.tryAcquire(io).?;
|
||||
defer budget.release(io, held);
|
||||
|
||||
var state: std.atomic.Value(listener.State) = .init(.serving);
|
||||
var waiter = io.concurrent(acquireAndHold, .{ &budget, io, null, &state }) catch |err| switch (err) {
|
||||
error.ConcurrencyUnavailable => return error.SkipZigTest,
|
||||
};
|
||||
try waitUntilWaiting(&budget, io, 1);
|
||||
state.store(.closing, .release);
|
||||
budget.wakeWaiters(io);
|
||||
try testing.expectError(error.Stopped, waiter.await(io));
|
||||
}
|
||||
|
||||
test "many waiters on few permits all get one, and every permit comes back" {
|
||||
var threaded: std.Io.Threaded = .init(testing.allocator, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var budget: QueryBudget = try .init(testing.allocator, 2);
|
||||
defer budget.deinit();
|
||||
|
||||
var group: std.Io.Group = .init;
|
||||
var taken_total: std.atomic.Value(u32) = .init(0);
|
||||
const rounds = 64;
|
||||
for (0..rounds) |_| {
|
||||
group.concurrent(io, takeBriefly, .{ &budget, io, &taken_total }) catch |err| switch (err) {
|
||||
error.ConcurrencyUnavailable => {
|
||||
try group.await(io);
|
||||
return error.SkipZigTest;
|
||||
},
|
||||
};
|
||||
}
|
||||
try group.await(io);
|
||||
try testing.expectEqual(@as(u32, rounds), taken_total.load(.monotonic));
|
||||
try testing.expectEqual(@as(usize, 0), budget.taken(io));
|
||||
try testing.expectEqual(@as(u64, 0), budget.snapshot().waiting);
|
||||
}
|
||||
|
||||
fn takeBriefly(budget: *QueryBudget, io: std.Io, taken_total: *std.atomic.Value(u32)) void {
|
||||
const work = budget.acquire(io, null, &test_serving) catch return;
|
||||
defer budget.release(io, work);
|
||||
_ = taken_total.fetchAdd(1, .monotonic);
|
||||
const hold: std.Io.Clock.Duration = .{ .raw = .fromMicroseconds(200), .clock = .awake };
|
||||
hold.sleep(io) catch {};
|
||||
}
|
||||
|
||||
test "init survives every allocation failure" {
|
||||
try testing.checkAllAllocationFailures(testing.allocator, struct {
|
||||
fn run(gpa: Allocator) !void {
|
||||
@@ -145,7 +438,7 @@ test "init survives every allocation failure" {
|
||||
}
|
||||
|
||||
test "the default budget stays small" {
|
||||
// 4096 + 65535 + scratch per permit: about 2.4 MiB for the whole process,
|
||||
// 4096 + 65535 + scratch per permit: under 5 MiB for the whole process,
|
||||
// where each UDP and TCP listener used to hold 4.8 and 8.7 MiB of its own.
|
||||
try testing.expect(@sizeOf(Work) * permits < 3 * 1024 * 1024);
|
||||
try testing.expect(@sizeOf(Work) * permits < 5 * 1024 * 1024);
|
||||
}
|
||||
|
||||
@@ -179,7 +179,7 @@ const EntryStorage = struct {
|
||||
.priority = priority,
|
||||
.enabled = true,
|
||||
.health = .init,
|
||||
.admission = .{ .permits = self.slots.len },
|
||||
.admission = .{ .permits = .init(self.slots.len) },
|
||||
.reuse_recoveries = &self.recoveries,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -50,7 +50,6 @@ pub const Snapshot = struct {
|
||||
accept_errors: u64,
|
||||
connection_errors: u64,
|
||||
idle_timeouts: u64,
|
||||
dropped_no_permit: u64,
|
||||
};
|
||||
|
||||
pub const TcpServer = struct {
|
||||
@@ -70,6 +69,7 @@ pub const TcpServer = struct {
|
||||
pub const Owner = TcpServer;
|
||||
pub const ConnPayload = Payload;
|
||||
pub const serveConn = serveOne;
|
||||
pub const closing = wakeBudgetWaiters;
|
||||
pub const read_buffer_len = stream_buffer_len;
|
||||
pub const write_buffer_len = stream_buffer_len;
|
||||
pub const log = std.log.scoped(.tcp_server);
|
||||
@@ -116,7 +116,6 @@ pub const TcpServer = struct {
|
||||
.accept_errors = stats.accept_errors.load(.monotonic),
|
||||
.connection_errors = stats.connection_errors.load(.monotonic),
|
||||
.idle_timeouts = stats.idle_timeouts.load(.monotonic),
|
||||
.dropped_no_permit = stats.dropped_no_permit.load(.monotonic),
|
||||
};
|
||||
}
|
||||
|
||||
@@ -125,6 +124,10 @@ pub const TcpServer = struct {
|
||||
self.core.serve(io);
|
||||
}
|
||||
|
||||
fn wakeBudgetWaiters(self: *TcpServer, io: std.Io) void {
|
||||
self.budget.wakeWaiters(io);
|
||||
}
|
||||
|
||||
pub fn deinit(self: *TcpServer, io: std.Io) void {
|
||||
self.core.deinit(io);
|
||||
self.* = undefined;
|
||||
@@ -182,11 +185,11 @@ pub const TcpServer = struct {
|
||||
},
|
||||
}
|
||||
|
||||
// There is no framing for "no answer", so the connection ends.
|
||||
const work = self.budget.tryAcquire(io) orelse {
|
||||
listener.bump(&stats.dropped_no_permit);
|
||||
return;
|
||||
};
|
||||
// Bounded like a read: a query still waiting when the idle budget
|
||||
// runs out ends the connection, with the timeout counted by the
|
||||
// budget. There is no framing for "no answer".
|
||||
const permit_deadline: std.Io.Clock.Timestamp = .fromNow(io, budget);
|
||||
const work = self.budget.acquire(io, permit_deadline, &self.core.run_state) catch return;
|
||||
defer self.budget.release(io, work);
|
||||
|
||||
const outcome = self.handler.handle(
|
||||
|
||||
@@ -527,9 +527,21 @@ fn askOn(reader: *std.Io.Reader, writer: *std.Io.Writer) ![]const u8 {
|
||||
return reader.take(len);
|
||||
}
|
||||
|
||||
fn waitForWaiters(permits: *query_budget.QueryBudget, io: std.Io, count: u64) !void {
|
||||
const tick: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(1), .clock = .awake };
|
||||
while (permits.snapshot().waiting != count) try tick.sleep(io);
|
||||
}
|
||||
|
||||
fn sendQuery(writer: *std.Io.Writer) !void {
|
||||
try writer.writeAll(&transport.framePrefix(@intCast(query_bytes.len)));
|
||||
try writer.writeAll(query_bytes);
|
||||
try writer.flush();
|
||||
}
|
||||
|
||||
/// One answered query, then the connection idles with every permit back in the
|
||||
/// budget; a second query that finds none taken ends the connection unanswered.
|
||||
fn keepAliveThenStarved(io: std.Io, address: net.IpAddress, permits: *query_budget.QueryBudget) anyerror!void {
|
||||
/// budget; a second query that finds every permit taken waits, and is answered
|
||||
/// once one is released.
|
||||
fn keepAliveThenWait(io: std.Io, address: net.IpAddress, permits: *query_budget.QueryBudget) anyerror!void {
|
||||
var stream = try address.connect(io, .{ .mode = .stream });
|
||||
defer stream.close(io);
|
||||
|
||||
@@ -546,17 +558,19 @@ fn keepAliveThenStarved(io: std.Io, address: net.IpAddress, permits: *query_budg
|
||||
while (permits.taken(io) != 0) try tick.sleep(io);
|
||||
|
||||
const held = permits.tryAcquire(io).?;
|
||||
defer permits.release(io, held);
|
||||
var released = false;
|
||||
defer if (!released) permits.release(io, held);
|
||||
|
||||
try writer.interface.writeAll(&transport.framePrefix(@intCast(query_bytes.len)));
|
||||
try writer.interface.writeAll(query_bytes);
|
||||
try writer.interface.flush();
|
||||
try sendQuery(&writer.interface);
|
||||
try waitForWaiters(permits, io, 1);
|
||||
permits.release(io, held);
|
||||
released = true;
|
||||
|
||||
var sink: [64]u8 = undefined;
|
||||
if (try reader.interface.readSliceShort(&sink) != 0) return error.TestUnexpectedBytes;
|
||||
const len = transport.parsePrefix((try reader.interface.takeArray(transport.prefix_len)).*);
|
||||
try expectAnswersQuery(try reader.interface.take(len));
|
||||
}
|
||||
|
||||
test "a keep-alive connection holds no permit between queries, and a query without one ends it" {
|
||||
test "a keep-alive connection holds no permit between queries, and a query without one waits for it" {
|
||||
if (!build_options.integration) return error.SkipZigTest;
|
||||
|
||||
const gpa = testing.allocator;
|
||||
@@ -583,11 +597,122 @@ test "a keep-alive connection holds no permit between queries, and a query witho
|
||||
};
|
||||
}
|
||||
|
||||
try bounded(io, keepAliveThenStarved, .{ io, server_address, &permits });
|
||||
try bounded(io, keepAliveThenWait, .{ io, server_address, &permits });
|
||||
|
||||
try testing.expectEqual(@as(u64, 1), server.core.stats.dropped_no_permit.load(.monotonic));
|
||||
try testing.expectEqual(@as(u64, 1), permits.snapshot().waits);
|
||||
try testing.expectEqual(@as(u64, 0), permits.snapshot().wait_timeouts);
|
||||
try testing.expectEqual(@as(u64, 0), server.core.stats.connection_errors.load(.monotonic));
|
||||
try testing.expectEqual(@as(u64, 1), h.stats.queries.load(.monotonic));
|
||||
try testing.expectEqual(@as(u64, 2), h.stats.queries.load(.monotonic));
|
||||
}
|
||||
|
||||
/// A query that waits out the whole idle budget for a permit: the server ends
|
||||
/// the connection without an answer.
|
||||
fn waitOutlastsIdle(io: std.Io, address: net.IpAddress) anyerror!void {
|
||||
var stream = try address.connect(io, .{ .mode = .stream });
|
||||
defer stream.close(io);
|
||||
|
||||
var read_buf: [1024]u8 = undefined;
|
||||
var write_buf: [1024]u8 = undefined;
|
||||
var reader = stream.reader(io, &read_buf);
|
||||
var writer = stream.writer(io, &write_buf);
|
||||
|
||||
try sendQuery(&writer.interface);
|
||||
var sink: [64]u8 = undefined;
|
||||
if (try reader.interface.readSliceShort(&sink) != 0) return error.TestUnexpectedBytes;
|
||||
}
|
||||
|
||||
test "a stream query whose permit wait outlasts the idle budget ends its connection, counted" {
|
||||
if (!build_options.integration) return error.SkipZigTest;
|
||||
|
||||
const gpa = testing.allocator;
|
||||
var threaded: std.Io.Threaded = .init(gpa, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var fake: FakeUpstream = .{ .reply = response_bytes };
|
||||
var h_owner: upstream_owner.Borrowed = .{};
|
||||
var h = bareHandler(h_owner.client(fake.client()));
|
||||
|
||||
const listen_address: net.IpAddress = try .parse("127.0.0.1", 0);
|
||||
var permits: query_budget.QueryBudget = try .init(gpa, 1);
|
||||
defer permits.deinit();
|
||||
var server = try tcp_server.TcpServer.listen(gpa, io, listen_address, &h, &permits, .{
|
||||
.max_connections = 2,
|
||||
.idle_timeout = short_idle,
|
||||
});
|
||||
const server_address = server.boundAddress();
|
||||
|
||||
var group: std.Io.Group = .init;
|
||||
try group.concurrent(io, tcp_server.TcpServer.serve, .{ &server, io });
|
||||
defer {
|
||||
server.deinit(io);
|
||||
group.await(io) catch |err| switch (err) {
|
||||
error.Canceled => unreachable,
|
||||
};
|
||||
}
|
||||
|
||||
const held = permits.tryAcquire(io).?;
|
||||
defer permits.release(io, held);
|
||||
|
||||
try bounded(io, waitOutlastsIdle, .{ io, server_address });
|
||||
|
||||
try testing.expectEqual(@as(u64, 1), permits.snapshot().wait_timeouts);
|
||||
try testing.expectEqual(@as(u64, 0), h.stats.queries.load(.monotonic));
|
||||
}
|
||||
|
||||
/// Leaves a query waiting for a permit and returns once the server sees it
|
||||
/// waiting.
|
||||
fn queryThenWait(io: std.Io, address: net.IpAddress, permits: *query_budget.QueryBudget, stream: *net.Stream) anyerror!void {
|
||||
stream.* = try address.connect(io, .{ .mode = .stream });
|
||||
var write_buf: [1024]u8 = undefined;
|
||||
var writer = stream.writer(io, &write_buf);
|
||||
try sendQuery(&writer.interface);
|
||||
try waitForWaiters(permits, io, 1);
|
||||
}
|
||||
|
||||
test "closing a tcp listener ends a connection that waits for a permit" {
|
||||
if (!build_options.integration) return error.SkipZigTest;
|
||||
|
||||
const gpa = testing.allocator;
|
||||
var threaded: std.Io.Threaded = .init(gpa, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var fake: FakeUpstream = .{ .reply = response_bytes };
|
||||
var h_owner: upstream_owner.Borrowed = .{};
|
||||
var h = bareHandler(h_owner.client(fake.client()));
|
||||
|
||||
const listen_address: net.IpAddress = try .parse("127.0.0.1", 0);
|
||||
var permits: query_budget.QueryBudget = try .init(gpa, 1);
|
||||
defer permits.deinit();
|
||||
// An idle budget far past `bounded`'s, so only the close can end the wait.
|
||||
const long_idle: std.Io.Clock.Duration = .{ .raw = .fromSeconds(60), .clock = .awake };
|
||||
var server = try tcp_server.TcpServer.listen(gpa, io, listen_address, &h, &permits, .{
|
||||
.max_connections = 2,
|
||||
.idle_timeout = long_idle,
|
||||
});
|
||||
const server_address = server.boundAddress();
|
||||
|
||||
var group: std.Io.Group = .init;
|
||||
try group.concurrent(io, tcp_server.TcpServer.serve, .{ &server, io });
|
||||
|
||||
const held = permits.tryAcquire(io).?;
|
||||
defer permits.release(io, held);
|
||||
|
||||
var stream: net.Stream = undefined;
|
||||
try bounded(io, queryThenWait, .{ io, server_address, &permits, &stream });
|
||||
defer stream.close(io);
|
||||
|
||||
try bounded(io, closeAndJoin, .{ &server, io, &group });
|
||||
try testing.expectEqual(@as(u64, 0), permits.snapshot().waiting);
|
||||
try testing.expectEqual(@as(u64, 0), h.stats.queries.load(.monotonic));
|
||||
}
|
||||
|
||||
fn closeAndJoin(server: *tcp_server.TcpServer, io: std.Io, group: *std.Io.Group) anyerror!void {
|
||||
server.deinit(io);
|
||||
group.await(io) catch |err| switch (err) {
|
||||
error.Canceled => unreachable,
|
||||
};
|
||||
}
|
||||
|
||||
/// Holds the only slot with an answered connection, then opens a second one,
|
||||
|
||||
+25
-19
@@ -1,9 +1,10 @@
|
||||
//! The UDP/53 listener.
|
||||
//!
|
||||
//! One receive loop hands each datagram to a task that holds a permit from the
|
||||
//! shared query budget. A datagram that finds no permit is dropped and counted,
|
||||
//! never queued: an unbounded queue only moves the failure from "visible drop"
|
||||
//! to "invisible latency", which is the failure mode PLAN §1 names.
|
||||
//! shared query budget. While every permit is taken the loop waits for one with
|
||||
//! the datagram it holds and receives nothing more, so the backlog stays in the
|
||||
//! socket's kernel buffer: bounded by `SO_RCVBUF`, and an overflow is counted
|
||||
//! by the kernel and exported as `nxdns_udp_server_kernel_drops_total`.
|
||||
//!
|
||||
//! The permit carries the query, the reply and the handler's scratch, so
|
||||
//! answering a datagram allocates nothing.
|
||||
@@ -12,6 +13,7 @@ const std = @import("std");
|
||||
const address = @import("../platform/address.zig");
|
||||
const concurrency = @import("../platform/concurrency.zig");
|
||||
const handler = @import("handler.zig");
|
||||
const listener = @import("listener.zig");
|
||||
const query_budget = @import("query_budget.zig");
|
||||
|
||||
const log = std.log.scoped(.udp_server);
|
||||
@@ -27,14 +29,14 @@ pub const Stats = struct {
|
||||
received: std.atomic.Value(u64) = .init(0),
|
||||
/// The datagram arrived with `flags.trunc`: the kernel discarded its tail.
|
||||
dropped_oversize: std.atomic.Value(u64) = .init(0),
|
||||
/// Every permit of the shared query budget was taken.
|
||||
dropped_no_slot: std.atomic.Value(u64) = .init(0),
|
||||
/// The handler answered `.drop` — an upstream exchange canceled by
|
||||
/// shutdown. The handler counts its own malformed drops; this counter is
|
||||
/// what keeps the listener's "every dropped datagram is counted" rule true
|
||||
/// for the drops it cannot see. A datagram that got a permit but no task
|
||||
/// is counted under `nxdns_concurrency_unavailable_total{site="udp_query"}`.
|
||||
dropped_handler: std.atomic.Value(u64) = .init(0),
|
||||
/// The listener closed while the datagram waited for a permit.
|
||||
dropped_at_shutdown: std.atomic.Value(u64) = .init(0),
|
||||
receive_errors: std.atomic.Value(u64) = .init(0),
|
||||
send_errors: std.atomic.Value(u64) = .init(0),
|
||||
};
|
||||
@@ -46,23 +48,20 @@ pub const Stats = struct {
|
||||
pub const Snapshot = struct {
|
||||
received: u64,
|
||||
dropped_oversize: u64,
|
||||
dropped_no_slot: u64,
|
||||
dropped_handler: u64,
|
||||
dropped_at_shutdown: u64,
|
||||
receive_errors: u64,
|
||||
send_errors: u64,
|
||||
};
|
||||
|
||||
/// Lifecycle of the receive loop. `serve` claims `.serving`, `deinit` publishes
|
||||
/// `.closing`, and the two meet at `stopped` so no task touches the socket
|
||||
/// after it is closed.
|
||||
const State = enum(u32) { idle, serving, closing };
|
||||
|
||||
pub const UdpServer = struct {
|
||||
socket: std.Io.net.Socket,
|
||||
handler: *handler.Handler,
|
||||
budget: *query_budget.QueryBudget,
|
||||
stats: Stats,
|
||||
state: std.atomic.Value(State),
|
||||
/// `serve` claims `.serving`, `deinit` publishes `.closing`, and the two
|
||||
/// meet at `stopped` so no task touches the socket after it is closed.
|
||||
state: std.atomic.Value(listener.State),
|
||||
stopped: std.Io.Event,
|
||||
|
||||
pub const BindError = std.Io.net.IpAddress.BindError;
|
||||
@@ -96,8 +95,8 @@ pub const UdpServer = struct {
|
||||
return .{
|
||||
.received = self.stats.received.load(.monotonic),
|
||||
.dropped_oversize = self.stats.dropped_oversize.load(.monotonic),
|
||||
.dropped_no_slot = self.stats.dropped_no_slot.load(.monotonic),
|
||||
.dropped_handler = self.stats.dropped_handler.load(.monotonic),
|
||||
.dropped_at_shutdown = self.stats.dropped_at_shutdown.load(.monotonic),
|
||||
.receive_errors = self.stats.receive_errors.load(.monotonic),
|
||||
.send_errors = self.stats.send_errors.load(.monotonic),
|
||||
};
|
||||
@@ -122,9 +121,13 @@ pub const UdpServer = struct {
|
||||
}
|
||||
|
||||
pub fn deinit(self: *UdpServer, io: std.Io) void {
|
||||
// `serve` reads this between receives, so it stops on its own and the
|
||||
// socket is closed only once nothing can be reading from it.
|
||||
if (self.state.swap(.closing, .acq_rel) == .serving) self.stopped.waitUncancelable(io);
|
||||
// `serve` reads this between receives and while it waits for a permit,
|
||||
// so it stops on its own and the socket is closed only once nothing
|
||||
// can be reading from it.
|
||||
if (self.state.swap(.closing, .acq_rel) == .serving) {
|
||||
self.budget.wakeWaiters(io);
|
||||
self.stopped.waitUncancelable(io);
|
||||
}
|
||||
self.socket.close(io);
|
||||
self.* = undefined;
|
||||
}
|
||||
@@ -161,9 +164,12 @@ pub const UdpServer = struct {
|
||||
continue;
|
||||
}
|
||||
|
||||
const work = self.budget.tryAcquire(io) orelse {
|
||||
bump(&self.stats.dropped_no_slot);
|
||||
continue;
|
||||
const work = self.budget.acquire(io, null, &self.state) catch |err| switch (err) {
|
||||
error.Canceled, error.Stopped => {
|
||||
bump(&self.stats.dropped_at_shutdown);
|
||||
return;
|
||||
},
|
||||
error.Timeout => unreachable,
|
||||
};
|
||||
@memcpy(work.datagram[0..msg.data.len], msg.data);
|
||||
|
||||
|
||||
@@ -260,7 +260,12 @@ test "deinit ends a serve loop that is blocked on receive" {
|
||||
};
|
||||
}
|
||||
|
||||
test "with every permit taken a udp query is dropped and counted" {
|
||||
fn waitForWaiters(permits: *query_budget.QueryBudget, io: std.Io, count: u64) !void {
|
||||
const tick: std.Io.Clock.Duration = .{ .raw = .fromMilliseconds(1), .clock = .awake };
|
||||
while (permits.snapshot().waiting != count) try tick.sleep(io);
|
||||
}
|
||||
|
||||
test "with every permit taken a udp query waits, and is answered once one frees" {
|
||||
if (!build_options.integration) return error.SkipZigTest;
|
||||
|
||||
const gpa = testing.allocator;
|
||||
@@ -288,6 +293,52 @@ test "with every permit taken a udp query is dropped and counted" {
|
||||
}
|
||||
|
||||
// Another listener's query, as far as this one can tell.
|
||||
var held: ?*query_budget.Work = permits.tryAcquire(io).?;
|
||||
defer if (held) |work| permits.release(io, work);
|
||||
|
||||
const client_address: net.IpAddress = try .parse("127.0.0.1", 0);
|
||||
var client = try client_address.bind(io, .{ .mode = .dgram });
|
||||
defer client.close(io);
|
||||
|
||||
try client.send(io, &server_address, query_bytes);
|
||||
try waitForWaiters(&permits, io, 1);
|
||||
|
||||
var buf: [udp_server.max_datagram]u8 = undefined;
|
||||
try testing.expectError(error.Timeout, client.receiveTimeout(io, &buf, silence));
|
||||
try testing.expectEqual(@as(u64, 0), h.stats.queries.load(.monotonic));
|
||||
|
||||
permits.release(io, held.?);
|
||||
held = null;
|
||||
|
||||
const msg = try client.receiveTimeout(io, &buf, budget);
|
||||
try expectAnswersQuery(msg.data);
|
||||
const waits = permits.snapshot();
|
||||
try testing.expectEqual(@as(u64, 1), waits.waits);
|
||||
try testing.expectEqual(@as(u64, 0), waits.wait_timeouts);
|
||||
try testing.expect(waits.wait_seconds > 0);
|
||||
}
|
||||
|
||||
test "a udp listener waiting for a permit stops when it is closed" {
|
||||
if (!build_options.integration) return error.SkipZigTest;
|
||||
|
||||
const gpa = testing.allocator;
|
||||
var threaded: std.Io.Threaded = .init(gpa, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var fake: FakeUpstream = .{ .reply = response_bytes };
|
||||
var h_owner: upstream_owner.Borrowed = .{};
|
||||
var h = bareHandler(h_owner.client(fake.client()));
|
||||
|
||||
const listen_address: net.IpAddress = try .parse("127.0.0.1", 0);
|
||||
var permits: query_budget.QueryBudget = try .init(gpa, 1);
|
||||
defer permits.deinit();
|
||||
var server = try udp_server.UdpServer.bind(io, listen_address, &h, &permits);
|
||||
const server_address = server.boundAddress();
|
||||
|
||||
var group: std.Io.Group = .init;
|
||||
try group.concurrent(io, udp_server.UdpServer.serve, .{ &server, io });
|
||||
|
||||
const held = permits.tryAcquire(io).?;
|
||||
defer permits.release(io, held);
|
||||
|
||||
@@ -296,13 +347,55 @@ test "with every permit taken a udp query is dropped and counted" {
|
||||
defer client.close(io);
|
||||
|
||||
try client.send(io, &server_address, query_bytes);
|
||||
try waitForWaiters(&permits, io, 1);
|
||||
|
||||
var buf: [udp_server.max_datagram]u8 = undefined;
|
||||
try testing.expectError(error.Timeout, client.receiveTimeout(io, &buf, silence));
|
||||
try testing.expectEqual(@as(u64, 1), server.stats.dropped_no_slot.load(.monotonic));
|
||||
// The permit is still held, so only the close can end this wait; a deinit
|
||||
// that missed the waiter would hang here.
|
||||
server.deinit(io);
|
||||
group.await(io) catch |err| switch (err) {
|
||||
error.Canceled => unreachable,
|
||||
};
|
||||
try testing.expectEqual(@as(u64, 0), permits.snapshot().waiting);
|
||||
try testing.expectEqual(@as(u64, 0), h.stats.queries.load(.monotonic));
|
||||
}
|
||||
|
||||
test "a udp listener canceled while it waits for a permit counts the datagram it held" {
|
||||
if (!build_options.integration) return error.SkipZigTest;
|
||||
|
||||
const gpa = testing.allocator;
|
||||
var threaded: std.Io.Threaded = .init(gpa, .{});
|
||||
defer threaded.deinit();
|
||||
const io = threaded.io();
|
||||
|
||||
var fake: FakeUpstream = .{ .reply = response_bytes };
|
||||
var h_owner: upstream_owner.Borrowed = .{};
|
||||
var h = bareHandler(h_owner.client(fake.client()));
|
||||
|
||||
const listen_address: net.IpAddress = try .parse("127.0.0.1", 0);
|
||||
var permits: query_budget.QueryBudget = try .init(gpa, 1);
|
||||
defer permits.deinit();
|
||||
var server = try udp_server.UdpServer.bind(io, listen_address, &h, &permits);
|
||||
defer server.deinit(io);
|
||||
const server_address = server.boundAddress();
|
||||
|
||||
var group: std.Io.Group = .init;
|
||||
try group.concurrent(io, udp_server.UdpServer.serve, .{ &server, io });
|
||||
|
||||
const held = permits.tryAcquire(io).?;
|
||||
defer permits.release(io, held);
|
||||
|
||||
const client_address: net.IpAddress = try .parse("127.0.0.1", 0);
|
||||
var client = try client_address.bind(io, .{ .mode = .dgram });
|
||||
defer client.close(io);
|
||||
|
||||
try client.send(io, &server_address, query_bytes);
|
||||
try waitForWaiters(&permits, io, 1);
|
||||
|
||||
group.cancel(io);
|
||||
try testing.expectEqual(@as(u64, 1), server.stats.dropped_at_shutdown.load(.monotonic));
|
||||
try testing.expectEqual(@as(u64, 0), permits.snapshot().waiting);
|
||||
}
|
||||
|
||||
test "a udp query with a permit but no task is dropped, counted, and returns its permit" {
|
||||
if (!build_options.integration) return error.SkipZigTest;
|
||||
|
||||
@@ -342,5 +435,4 @@ test "a udp query with a permit but no task is dropped, counted, and returns its
|
||||
try testing.expectError(error.Timeout, client.receiveTimeout(io, &buf, silence));
|
||||
try testing.expectEqual(@as(u64, 1), concurrency.snapshot().get(.udp_query) - before);
|
||||
try testing.expectEqual(@as(usize, 0), permits.taken(io));
|
||||
try testing.expectEqual(@as(u64, 0), server.stats.dropped_no_slot.load(.monotonic));
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user