diff --git a/README.md b/README.md index aa7c565..db9894f 100644 --- a/README.md +++ b/README.md @@ -24,6 +24,8 @@ Windows uses overlapped sockets and an I/O completion port (IOCP). Application callbacks can run on the I/O owner or on fixed startup workers. Inline execution is the default for trusted bounded, nonblocking handlers; blocking callbacks must explicitly select fixed startup workers. +Worker callbacks can [flush output without returning](docs/USING.md#flush-within-a-worker-callback). +The same reserved worker stack resumes after the owner completes transmission. The [Windows receipt](reports/2026-09-06-windows-iocp.md) records native x64 evidence and its limits. This is the M4 implementation informed by the adjacent [Zig LLM Wiki](https://technologylab-ai.github.io/zigllmwiki/?page=wiki/bounded-http-server-design.md). diff --git a/docs/OWNERSHIP.md b/docs/OWNERSHIP.md index 71a6e81..f2ef371 100644 --- a/docs/OWNERSHIP.md +++ b/docs/OWNERSHIP.md @@ -19,10 +19,19 @@ with release/acquire ordering. Wake signals are hints; the phase is authoritativ | I/O receive/parse | I/O owner appends initialized receive bytes and advances the parser. One transport operation may borrow the destination. | | Ready/running callback | Assigned worker borrows the immutable request and exclusively mutates its writer/state. The I/O owner can set the atomic cancellation flag and close networking, but cannot recycle the slot. | | Published callback result | I/O owner acquires the committed writer snapshot and action; the worker stops accessing request/writer/state until another dispatch. | +| Worker flush (`stream_ready`/`stream_wait`) | The callback stack stays live. The owner acquires only the frozen output. The worker waits for output release before continuing. | | Sending | I/O owner advances partial-send cursors. Committed payload storage stays immutable until every send using it completes. | | Flushed | Writer storage is released/reset and the same handler is dispatched with `.flushed`; the request and continuation state remain borrowed. | | Finished/closing | No continuation is promised. Slot storage is reused only after the callback has returned and all target/cancel completions have drained. | +`Context.flushAndWait()` retains the same worker callback across transmission. +Successful completion publishes `running` after restoring writable capacity. +Cancellation publishes `running` only after all kernel borrows end. +The worker then receives `Cancelled` and must return `.close`. +Only final callback return publishes `result` and permits eventual slot reuse. +Neither streaming phase ends the application borrow. +See [worker flush](USING.md#flush-within-a-worker-callback) for the public contract and worker occupancy limit. + The receive buffer stays at a stable address through parsing and all callbacks for that request. `Request.method`, `target`, `headers`, `header(name)` results, `body_wire` and each body-iterator span borrow it. Chunked decoding removes no diff --git a/docs/USING.md b/docs/USING.md index 3de6f93..19eaa1b 100644 --- a/docs/USING.md +++ b/docs/USING.md @@ -293,6 +293,35 @@ Return that result immediately. The owner completes response framing without another callback for that response. Finished responses can remain buffered within a batch before transmission. +### Flush within a worker callback + +`Context.supportsBlockingFlush()` reports whether the callback runs on a fixed application worker. +`Context.flushAndWait()` sends the current snapshot without returning from that callback. +The method freezes output and waits on the worker's existing notification mechanism. +The I/O owner continues processing network events and other connections. +Successful completion restores writable capacity and resumes the same worker stack. +Application code can then write more output and flush again. +Return `writer.finish()` after the final write. + +This operation requires `.workers` execution and a begun response without an outstanding reservation. +Inline callbacks receive `BlockingFlushUnavailable` before the writer changes. +Invalid writer state returns `InvalidState`. +Disconnect, deadline, shutdown, or response-limit failure can return `Cancelled`. +On cancellation, return `.close`; do not append output or reuse the frozen writer. +The original request deadline and cumulative `max_response_bytes` limit remain unchanged across waits. + +One active callback occupies one preallocated worker throughout its flush waits and application work. +Other slots assigned to that worker wait for it to return. +There is no new request thread, heap allocation, or general asynchronous task scheduler. +Arbitrary application blocking still requires the existing external watchdog. + +The context and writer stay on their original callback thread. +Cancellation waits for every kernel borrow before unwinding the waiting worker. +The slot remains retained until the callback also returns. +Successful flush proves local transmission completion, not peer application receipt. +HEAD suppresses payload transmission, and empty flushes do not end chunked responses. +The existing return-and-resume `Writer.flush()` API remains available. + ### Capacity and `WouldBlock` `Writer.capacity()` reports conservative body capacity after a flush. diff --git a/src/api.zig b/src/api.zig index 3f51fef..85f467e 100644 --- a/src/api.zig +++ b/src/api.zig @@ -5,6 +5,7 @@ const assert = std.debug.assert; pub const Action = enum { flush, finish, close }; pub const Event = enum { request, flushed }; pub const Handler = *const fn (*Context) Action; +pub const FlushError = error{ BlockingFlushUnavailable, InvalidState, Cancelled }; pub const Context = struct { request: *const http.Request, @@ -14,6 +15,31 @@ pub const Context = struct { state: *[8]usize, application: ?*anyopaque, cancelled: *const std.atomic.Value(bool), + /// The scheduler supplies this hook only for an existing application worker. + blocking_flush: ?BlockingFlush = null, + + pub const BlockingFlush = struct { + context: *anyopaque, + flush: *const fn (*anyopaque) FlushError!void, + }; + + pub fn supportsBlockingFlush(self: *const Context) bool { + return self.blocking_flush != null; + } + + /// Send this snapshot and resume the same callback with fresh output capacity. + /// The worker waits; the I/O owner continues processing other connections. + /// Completion ends the local transport borrow, not the peer's processing. + /// After publication, cancellation waits for transport borrows before returning an error. + /// The context and writer must remain on their original callback thread. + pub fn flushAndWait(self: *Context) FlushError!void { + const hook = self.blocking_flush orelse return error.BlockingFlushUnavailable; + if (self.cancelled.load(.acquire)) return error.Cancelled; + if (self.writer.frozen or self.writer.reserved != 0 or !self.writer.began) + return error.InvalidState; + _ = self.writer.flush(); + return hook.flush(hook.context); + } }; /// Largest response head this framework generates, plus the chunk-size field. @@ -444,6 +470,46 @@ fn testCache() HeaderCache { return cache; } +test "blocking flush rejects unavailable, cancelled and unpublished states without freezing" { + var arena: [1024]u8 = undefined; + const cache = testCache(); + var writer = Writer.init(&arena, &cache, 0); + writer.open(0, true, false); + var request: http.Request = undefined; + var state: [8]usize = @splat(0); + var cancelled: std.atomic.Value(bool) = .init(false); + var called = false; + var context: Context = .{ + .request = &request, + .writer = &writer, + .event = .request, + .state = &state, + .application = null, + .cancelled = &cancelled, + }; + try std.testing.expect(!context.supportsBlockingFlush()); + try std.testing.expectError(error.BlockingFlushUnavailable, context.flushAndWait()); + context.blocking_flush = .{ .context = &called, .flush = struct { + fn flush(pointer: *anyopaque) FlushError!void { + const value: *bool = @ptrCast(@alignCast(pointer)); + value.* = true; + } + }.flush }; + try std.testing.expect(context.supportsBlockingFlush()); + try std.testing.expectError(error.InvalidState, context.flushAndWait()); + try writer.begin(200, "text/plain", null); + _ = try writer.reserve(1); + try std.testing.expectError(error.InvalidState, context.flushAndWait()); + writer.commit(0); + cancelled.store(true, .release); + try std.testing.expectError(error.Cancelled, context.flushAndWait()); + try std.testing.expect(!writer.frozen and !called); + cancelled.store(false, .release); + try context.flushAndWait(); + try std.testing.expect(writer.frozen and called); + try std.testing.expectError(error.InvalidState, context.flushAndWait()); +} + test "begin writes the head into the arena and flush keeps the snapshot" { var arena: [1024]u8 = undefined; const cache = testCache(); diff --git a/src/server.zig b/src/server.zig index a496323..ed2ee27 100644 --- a/src/server.zig +++ b/src/server.zig @@ -220,7 +220,8 @@ pub const Stats = struct { } }; -const Phase = enum(u8) { io, ready, running, result }; +// Only result ends the callback borrow. Both streaming phases retain its stack. +const Phase = enum(u8) { io, ready, running, stream_ready, stream_wait, result }; const SendMode = enum { response, interim, reject }; const Kind = enum(u8) { accept = 1, recv, send, cancel_recv, cancel_send, cancel_accept }; const accept_token: u64 = @intFromEnum(Kind.accept); @@ -342,23 +343,7 @@ const Worker = struct { _ = self.server.ready_workers.fetchAdd(1, .release); defer _ = self.server.exited_workers.fetchAdd(1, .release); while (!self.server.stop_workers.load(.acquire)) { - // The phase owns work; finite waits make coalesced wakeups harmless. - if (builtin.os.tag == .windows) { - const result = win32.WaitForSingleObject(self.event.?, 10); - assert(result == 0 or result == 258); // signaled or timeout - } else { - var ready = [_]c.pollfd{.{ .fd = self.read_fd, .events = c.POLL.IN, .revents = 0 }}; - const polled = c.poll(&ready, 1, 10); - if (polled < 0) { - assert(c.errno(polled) == .INTR); - continue; - } - if (polled > 0) { - var bytes: [64]u8 = undefined; - const read = c.read(self.read_fd, &bytes, bytes.len); - assert(read > 0 or (read < 0 and c.errno(read) == .INTR)); - } - } + self.wait(); if (self.server.stop_workers.load(.acquire)) break; var index = self.index; while (index < self.server.slots.len) : (index += self.server.workers.len) { @@ -370,6 +355,26 @@ const Worker = struct { } } } + + fn wait(self: *Worker) void { + // The phase owns work; finite waits make coalesced wakeups harmless. + if (builtin.os.tag == .windows) { + const result = win32.WaitForSingleObject(self.event.?, 10); + assert(result == 0 or result == 258); // signaled or timeout + } else { + var ready = [_]c.pollfd{.{ .fd = self.read_fd, .events = c.POLL.IN, .revents = 0 }}; + const polled = c.poll(&ready, 1, 10); + if (polled < 0) { + assert(c.errno(polled) == .INTR); + return; + } + if (polled > 0) { + var bytes: [64]u8 = undefined; + const read = c.read(self.read_fd, &bytes, bytes.len); + assert(read > 0 or (read < 0 and c.errno(read) == .INTR)); + } + } + } }; /// One I/O owner: listener, transport, slots, arenas, operation cells, clock @@ -693,7 +698,9 @@ pub const Server = struct { fn anyResult(self: *Server) bool { for (self.slots) |*slot| { - if (slot.in_use and slot.phase.load(.acquire) == .result) return true; + if (!slot.in_use) continue; + const phase = slot.phase.load(.acquire); + if (phase == .result or phase == .stream_ready) return true; } return false; } @@ -721,7 +728,21 @@ pub const Server = struct { } self.invokeInline(slot); } - if (slot.phase.load(.acquire) != .result) break; + const phase = slot.phase.load(.acquire); + if (phase == .stream_ready) { + assert(self.config.execution == .workers and slot.action == .flush); + // The worker still owns its stack, request and application state. + // Only the frozen writer snapshot transfers to the I/O owner. + slot.phase.store(.stream_wait, .release); + if (slot.closing or self.stop_requested.load(.acquire) or self.now >= slot.deadline) { + if (!slot.closing and !self.stop_requested.load(.acquire)) self.stats.timeouts += 1; + try self.beginClose(slot); + } else { + try self.prepareResponse(index, false); + } + break; + } + if (phase != .result) break; if (self.config.callback_timing) { self.stats.max_handler_ns = @max(self.stats.max_handler_ns, slot.handler_ns); self.stats.max_queue_ns = @max(self.stats.max_queue_ns, slot.queue_ns); @@ -1168,6 +1189,7 @@ pub const Server = struct { if (slot.cancelled.load(.acquire)) { slot.action = .close; } else { + var streaming: WorkerFlush = .{ .server = self, .slot = slot }; var context: api.Context = .{ .request = &slot.request, .writer = &slot.writer, @@ -1175,14 +1197,42 @@ pub const Server = struct { .state = &slot.state, .application = self.application, .cancelled = &slot.cancelled, + .blocking_flush = if (self.config.execution == .workers) .{ + .context = &streaming, + .flush = WorkerFlush.flush, + } else null, }; slot.action = self.handler(&context); if (slot.action != .close) assert(slot.writer.frozen); } if (timing) slot.handler_ns = nowNs() - started; + assert(slot.phase.load(.acquire) == .running); slot.phase.store(.result, .release); } + /// This frame lives on the startup worker stack until the handler returns. + const WorkerFlush = struct { + server: *Server, + slot: *Slot, + + fn flush(pointer: *anyopaque) api.FlushError!void { + const self: *WorkerFlush = @ptrCast(@alignCast(pointer)); + const slot = self.slot; + assert(self.server.config.execution == .workers); + assert(slot.phase.load(.acquire) == .running and slot.writer.frozen); + const index = (@intFromPtr(slot) - @intFromPtr(self.server.slots.ptr)) / @sizeOf(Slot); + const worker = &self.server.workers[index % self.server.workers.len]; + slot.action = .flush; + slot.phase.store(.stream_ready, .release); + self.server.backend.wake(); + // Cancellation alone cannot return output or input ownership. + // Success releases output borrows. Cancellation releases all kernel borrows. + while (slot.phase.load(.acquire) != .running) worker.wait(); + if (slot.cancelled.load(.acquire)) return error.Cancelled; + assert(!slot.writer.frozen and slot.writer.headers_committed); + } + }; + fn reject(self: *Server, index: usize, status: u16) !void { const slot = &self.slots[index]; assert(slot.batch_count == 0 and slot.arena_used == 0); @@ -1426,7 +1476,13 @@ pub const Server = struct { slot.writer.resumeSnapshot(0); slot.event = .flushed; self.stats.resumed += 1; - self.dispatch(index); + if (slot.phase.load(.acquire) == .stream_wait) { + assert(self.config.execution == .workers); + slot.phase.store(.running, .release); + self.workers[index % self.workers.len].wake(); + } else { + self.dispatch(index); + } }, .close => try self.beginClose(slot), .parse => { @@ -1482,9 +1538,18 @@ pub const Server = struct { fn maybeFree(self: *Server, slot: *Slot) void { if (!slot.closing or slot.recv_pending or slot.send_pending or - slot.recv_cancel_pending or slot.send_cancel_pending or - slot.phase.load(.acquire) != .io) return; + slot.recv_cancel_pending or slot.send_cancel_pending) return; const index = (@intFromPtr(slot) - @intFromPtr(self.slots.ptr)) / @sizeOf(Slot); + const phase = slot.phase.load(.acquire); + if (phase == .stream_wait) { + assert(self.config.execution == .workers and slot.cancelled.load(.acquire)); + // Kernel borrows ended, but the same callback must still unwind. + // Keep the writer frozen because cancellation forbids more output. + slot.phase.store(.running, .release); + self.workers[index % self.workers.len].wake(); + return; + } + if (phase != .io) return; if (slot.fd >= 0) { self.backend.close(self.recvCell(index), slot.fd); slot.fd = -1; @@ -1505,6 +1570,137 @@ pub const Server = struct { self.free_count += 1; } + test "blocking flush resumes the same stack and preserves the request deadline" { + const server = try Server.init(std.testing.allocator, .{ + .execution = .workers, + .workers = 1, + .connections = 1, + .port = 0, + }, struct { + fn handler(_: *api.Context) api.Action { + @panic("blocking flush must not invoke the handler again"); + } + }.handler, null); + defer server.deinit(); + const slot = &server.slots[0]; + defer slot.phase.store(.io, .release); + server.header_cache.refresh("Sat, 05 Sep 2026 12:34:56 GMT"); + for ([_]bool{ false, true }) |head_only| { + slot.writer.open(0, true, head_only); + try slot.writer.begin(200, "text/plain", null); + _ = slot.writer.flush(); + slot.writer.headers_committed = true; + slot.request_active = true; + slot.request_started = 123; + slot.deadline = 123 + @as(u64, server.config.timeout_ms) * 1_000_000; + const deadline = slot.deadline; + server.now = 456; + slot.phase.store(.stream_wait, .release); + slot.batch_next = .resume_flush; + slot.cells[0] = .{}; + slot.batch_count = 1; + slot.part_count = 0; + slot.part = 0; + try server.completeBatch(0); + try std.testing.expectEqual(Phase.running, slot.phase.load(.acquire)); + try std.testing.expectEqual(deadline, slot.deadline); + try std.testing.expect(!slot.writer.frozen and slot.request_active); + try std.testing.expectEqual(@as(usize, 0), slot.writer.bodyBytes()); + try std.testing.expectEqual(@as(u64, 0), server.stats.worker_dispatches); + // A second empty flush completes without submitting a send. + _ = slot.writer.flush(); + slot.phase.store(.stream_wait, .release); + slot.batch_count = 1; + try server.completeBatch(0); + try std.testing.expectEqual(Phase.running, slot.phase.load(.acquire)); + try std.testing.expectEqual(deadline, slot.deadline); + } + try std.testing.expectEqual(@as(u64, 4), server.stats.resumed); + } + + test "cancelled flush retains kernel borrows and the callback until its final result" { + const server = try Server.init(std.testing.allocator, .{ + .execution = .workers, + .workers = 1, + .connections = 1, + .port = 0, + }, struct { + fn handler(_: *api.Context) api.Action { + return .close; + } + }.handler, null); + defer server.deinit(); + const slot = &server.slots[0]; + defer { + slot.phase.store(.io, .release); + server.stats.live_connections = 0; + } + slot.in_use = true; + slot.closing = true; + slot.cancelled.store(true, .release); + slot.phase.store(.stream_wait, .release); + server.stats.live_connections = 1; + server.free_count = 0; + inline for (.{ "recv_pending", "send_pending", "recv_cancel_pending", "send_cancel_pending" }) |field| { + @field(slot, field) = true; + server.maybeFree(slot); + try std.testing.expectEqual(Phase.stream_wait, slot.phase.load(.acquire)); + try std.testing.expect(slot.in_use and server.free_count == 0); + @field(slot, field) = false; + } + server.maybeFree(slot); + try std.testing.expectEqual(Phase.running, slot.phase.load(.acquire)); + try std.testing.expect(slot.in_use and server.free_count == 0); + // Cancellation releases kernel ownership, but the callback can still unwind. + server.maybeFree(slot); + try std.testing.expectEqual(@as(usize, 1), server.stats.live_connections); + slot.action = .close; + slot.phase.store(.result, .release); + try server.serviceSlot(0); + try std.testing.expect(!slot.in_use); + try std.testing.expectEqual(@as(usize, 1), server.free_count); + try std.testing.expectEqual(@as(usize, 0), server.stats.live_connections); + } + + test "blocking flush applies cumulative response bounds before transmission" { + const server = try Server.init(std.testing.allocator, .{ + .execution = .workers, + .workers = 1, + .connections = 1, + .port = 0, + .max_response_bytes = 4, + }, struct { + fn handler(_: *api.Context) api.Action { + return .close; + } + }.handler, null); + defer server.deinit(); + const slot = &server.slots[0]; + defer { + slot.phase.store(.io, .release); + server.stats.live_connections = 0; + } + server.header_cache.refresh("Sat, 05 Sep 2026 12:34:56 GMT"); + slot.in_use = true; + slot.request_active = true; + slot.deadline = 100; + slot.logical_written = 3; + server.stats.live_connections = 1; + server.free_count = 0; + slot.writer.open(0, true, false); + try slot.writer.begin(200, "text/plain", null); + try slot.writer.write("ab"); + slot.action = slot.writer.flush(); + slot.phase.store(.stream_ready, .release); + try server.serviceSlot(0); + try std.testing.expect(slot.cancelled.load(.acquire)); + try std.testing.expectEqual(Phase.running, slot.phase.load(.acquire)); + try std.testing.expect(slot.in_use and slot.writer.frozen); + try std.testing.expectEqual(@as(usize, 3), slot.logical_written); + try std.testing.expectEqual(@as(usize, 0), server.stats.live_operations); + try std.testing.expectEqual(@as(u64, 0), server.stats.bytes_sent); + } + test "worker response headers retain their exclusive dispatch snapshot" { const server = try Server.init(std.testing.allocator, .{ .execution = .workers,