diff --git a/frameworks/zix/.gitignore b/frameworks/zix/.gitignore new file mode 100644 index 000000000..0e6a877fe --- /dev/null +++ b/frameworks/zix/.gitignore @@ -0,0 +1,5 @@ +.zig-cache +zig-out +zig-package +vendor +/zig* diff --git a/frameworks/zix/Dockerfile b/frameworks/zix/Dockerfile index 692929f64..cf27650fa 100644 --- a/frameworks/zix/Dockerfile +++ b/frameworks/zix/Dockerfile @@ -4,66 +4,68 @@ FROM alpine:3.20 AS build ARG RETRY=6 ARG TARGETARCH ARG RETRY_DELAY=3 +ARG TIMEOUT_SEC=180 ARG ZIG_VERSION=0.16.0 -ARG ZIX_VERSION=0.4.x-rc3 -RUN apk add --no-cache ca-certificates curl git tar xz +ARG ZIX_VERSION=0.5.x-rc2 +RUN apk add --no-cache ca-certificates curl git tar xz openssl +WORKDIR /server RUN set -eu; \ case "${TARGETARCH:-amd64}" in \ amd64) ZIG_ARCH=x86_64 ;; \ arm64) ZIG_ARCH=aarch64 ;; \ *) echo "unsupported arch: ${TARGETARCH}" >&2; exit 1 ;; \ esac; \ - curl -fsSL "https://ziglang.org/download/${ZIG_VERSION}/zig-${ZIG_ARCH}-linux-${ZIG_VERSION}.tar.xz" \ + curl -fSL -m ${TIMEOUT_SEC} "https://ziglang.org/download/${ZIG_VERSION}/zig-${ZIG_ARCH}-linux-${ZIG_VERSION}.tar.xz" \ | tar -xJ -C /opt; \ mv "/opt/zig-${ZIG_ARCH}-linux-${ZIG_VERSION}" /opt/zig ENV PATH="/opt/zig:${PATH}" -# Vendor zix X.Y.Z, separate layer so source-only rebuilds skip the fetch. -# The Http1 raw engine work this image needs (large-body drain plus the per-worker -# response cache used by the /json endpoint) must be present on the X.Y.Z branch. -# Four ordered attempts before giving up: curl the archive tarball from github then -# codeberg, then a shallow git clone from github then codeberg. The github archive -# redirects to codeload.github.com (which the benchmark runner may not resolve), so -# curl can fall through to the codeberg tarball, and git clone talks to github.com -# and codeberg.org directly as the deeper fallback. RETRY and RETRY_DELAY bound -# every attempt. +COPY build.zig ./ +COPY build.zig.zon.template ./build.zig.zon +COPY src ./src + +# Resolve zix with zig fetch git+https RUN set -eu; \ - fetch() { \ - rm -rf /src/vendor/zix; mkdir -p /src/vendor/zix; \ - curl -fsSL --retry ${RETRY} --retry-delay ${RETRY_DELAY} --retry-all-errors "$1" -o /tmp/zix.tar.gz \ - && tar -xz --strip-components=1 -C /src/vendor/zix -f /tmp/zix.tar.gz; \ - }; \ - clone() { \ - attempt=0; \ - while [ "${attempt}" -lt "${RETRY}" ]; do \ - rm -rf /src/vendor/zix; \ - git clone --depth 1 --branch "${ZIX_VERSION}" "$1" /src/vendor/zix && return 0; \ + fetch_zix() { \ + url="$1"; attempt=1; \ + while [ "${attempt}" -le "${RETRY}" ]; do \ + if $(zig fetch --save ${url}); then \ + return 0; \ + fi; \ + echo "zix: ${url} attempt ${attempt}/${RETRY} failed" >&2; \ attempt=$((attempt + 1)); \ - sleep "${RETRY_DELAY}"; \ + [ "${attempt}" -le "${RETRY}" ] && sleep "${RETRY_DELAY}"; \ done; \ return 1; \ }; \ - fetch "https://github.com/prothegee/zix/archive/refs/heads/${ZIX_VERSION}.tar.gz" \ - || { echo "FAILED: curl ${RETRY} times from github" >&2; \ - fetch "https://codeberg.org/prothegee/zix/archive/${ZIX_VERSION}.tar.gz" \ - || { echo "FAILED: curl ${RETRY} times from codeberg" >&2; \ - clone "https://github.com/prothegee/zix.git" \ - || { echo "FAILED: git clone ${RETRY} times from github" >&2; \ - clone "https://codeberg.org/prothegee/zix.git" \ - || { echo "FAILED: git clone ${RETRY} times from codeberg" >&2; exit 1; }; }; }; } + fetch_zix "git+https://codeberg.org/prothegee/zix#${ZIX_VERSION}" \ + || { echo "zix: codeberg exhausted ${RETRY} attempts" >&2; \ + fetch_zix "git+https://github.com/prothegee/zix#${ZIX_VERSION}"; } \ + || { echo "zix: github exhausted ${RETRY} attempts" >&2; exit 1; } -WORKDIR /src -COPY build.zig build.zig.zon ./ -COPY src ./src +# +aes+pclmul: x86_64_v3 omits AES-NI / PCLMUL, +# so zix TLS would compile the ~40x slower software AES-GCM. +# +adx speeds the RSA / Montgomery path. Every x86_64_v3 CPU has them, so it is safe. RUN set -eu; \ case "${TARGETARCH:-amd64}" in \ amd64) ZIG_TARGET=x86_64-linux-musl; ZIG_CPU=x86_64_v3 ;; \ arm64) ZIG_TARGET=aarch64-linux-musl; ZIG_CPU=baseline ;; \ esac; \ - zig build -Dtarget="${ZIG_TARGET}" -Dcpu="${ZIG_CPU}" --release=fast + zig build \ + -Dtarget="${ZIG_TARGET}" \ + -Dcpu="${ZIG_CPU}+aes+pclmul+adx" \ + --release=fast --summary line; + +# Self-signed Ed25519 cert generated at image build, baked at /etc/zix-tls. +RUN set -eu; \ + mkdir -p /etc/zix-tls; \ + openssl genpkey -algorithm ED25519 -out /etc/zix-tls/server.key; \ + openssl req -new -x509 -key /etc/zix-tls/server.key -out /etc/zix-tls/server.cert \ + -days 3650 -subj "/CN=localhost" FROM alpine:3.20 -COPY --from=build /src/zig-out/bin/zix /zix -EXPOSE 8080 -ENTRYPOINT ["/zix"] +COPY --from=build /server/zig-out/bin/zix-http1 /zix-http1 +COPY --from=build /etc/zix-tls /etc/zix-tls +EXPOSE 8080 8081 +ENTRYPOINT ["/zix-http1"] diff --git a/frameworks/zix/build.zig b/frameworks/zix/build.zig index 5276d7f96..97f1f57a1 100644 --- a/frameworks/zix/build.zig +++ b/frameworks/zix/build.zig @@ -8,7 +8,7 @@ pub fn build(b: *std.Build) void { const zix_mod = zix_dep.module("zix"); const exe = b.addExecutable(.{ - .name = "zix", + .name = "zix-http1", .root_module = b.createModule(.{ .root_source_file = b.path("src/main.zig"), .target = target, diff --git a/frameworks/zix/build.zig.zon.template b/frameworks/zix/build.zig.zon.template new file mode 100644 index 000000000..cdde7a12c --- /dev/null +++ b/frameworks/zix/build.zig.zon.template @@ -0,0 +1,12 @@ +.{ + .name = .zix_arena, + .version = "0.1.0", + .fingerprint = 0x84cceaddc218682f, + .minimum_zig_version = "0.16.0", + .paths = .{ + "build.zig", + "build.zig.zon", + "src", + }, + .dependencies = .{}, +} diff --git a/frameworks/zix/meta.json b/frameworks/zix/meta.json index b031b6910..1b18a76bf 100644 --- a/frameworks/zix/meta.json +++ b/frameworks/zix/meta.json @@ -2,17 +2,23 @@ "display_name": "zix", "language": "Zig", "type": "engine", - "engine": "zix", - "description": "Zig HTTP/1.1 server on the zix.Http1 raw engine (no std.http). Shared-nothing: each worker runs its own SO_REUSEPORT multishot accept plus io_uring completion loop and owns its connections. The /json endpoint serves from the per-worker response cache, and request bodies larger than the read buffer are drained rather than buffered.", + "engine": "zix.Http1 URING Dispatch Model", + "description": "Zig HTTP/1.1 server on the zix.Http1 raw engine, .URING dispatch: per-worker SO_REUSEPORT io_uring loops, comptime Router, staged send sink. Dual listener: cleartext 8080, TLS 1.3 on 8081. DB endpoints run on engine-worker lanes: each worker pumps its own pipelined postgrez connections on its ring, named prepared statements, cache-aside single-item crud reads (200 ms TTL).", "repo": "https://github.com/prothegee/zix", "enabled": true, "tests": [ "baseline", "pipelined", "limited-conn", - "json", "upload", - "static" + "static", + "json", + "json-comp", + "json-tls", + "api-4", + "api-16", + "async-db", + "crud" ], "maintainers": ["prothegee"] } diff --git a/frameworks/zix/src/handlers/asyncdb.zig b/frameworks/zix/src/handlers/asyncdb.zig new file mode 100644 index 000000000..73f722871 --- /dev/null +++ b/frameworks/zix/src/handlers/asyncdb.zig @@ -0,0 +1,35 @@ +//! GET /async-db?min=..&max=..&limit=.. : price-range item scan. Queues a +//! Job on the worker's postgres lane (dbpg.zig), answered async. + +const std = @import("std"); +const zix = @import("zix"); + +const dbpg = @import("../shared/dbpg.zig"); +const response = @import("../shared/response.zig"); + +// --------------------------------------------------------- // + +pub const PATH = "/async-db"; + +const ASYNC_DB_LIMIT_MAX: i64 = 50; + +// --------------------------------------------------------- // + +pub fn RESPONSE(req: *zix.Http1.Request, _: *zix.Http1.Response, _: *zix.Http1.Context) !void { + const head = req.head; + const fd = req.fd; + + const min: i64 = if (zix.Http1.queryParam(head, "min")) |raw| std.fmt.parseInt(i64, raw, 10) catch 0 else 0; + const max: i64 = if (zix.Http1.queryParam(head, "max")) |raw| std.fmt.parseInt(i64, raw, 10) catch 0 else 0; + var limit: i64 = if (zix.Http1.queryParam(head, "limit")) |raw| std.fmt.parseInt(i64, raw, 10) catch 10 else 10; + if (limit < 1) limit = 1; + if (limit > ASYNC_DB_LIMIT_MAX) limit = ASYNC_DB_LIMIT_MAX; + + const queued = dbpg.submitJob(head, .{ .ASYNC_DB = .{ + .fd = fd, + .min = min, + .max = max, + .limit = limit, + } }); + if (!queued) response.serviceUnavailable(fd); +} diff --git a/frameworks/zix/src/handlers/baseline.zig b/frameworks/zix/src/handlers/baseline.zig new file mode 100644 index 000000000..2b9de88c1 --- /dev/null +++ b/frameworks/zix/src/handlers/baseline.zig @@ -0,0 +1,61 @@ +//! GET/POST /baseline11?a=..&b=.. : sum the query values, plus the POST body +//! as an integer. Returns the sum as text/plain. + +const std = @import("std"); +const zix = @import("zix"); + +// --------------------------------------------------------- // + +pub const PATH = "/baseline11"; + +// --------------------------------------------------------- // + +fn sumQuery(query: []const u8) i64 { + var sum: i64 = 0; + var it = std.mem.tokenizeScalar(u8, query, '&'); + while (it.next()) |pair| { + if (std.mem.indexOfScalar(u8, pair, '=')) |eq| { + sum += std.fmt.parseInt(i64, pair[eq + 1 ..], 10) catch 0; + } + } + return sum; +} + +fn parseIntLoose(s: []const u8) i64 { + var i: usize = 0; + while (i < s.len and (s[i] == ' ' or s[i] == '\t' or s[i] == '\r' or s[i] == '\n')) i += 1; + + var neg = false; + if (i < s.len and s[i] == '-') { + neg = true; + i += 1; + } + + var n: i64 = 0; + while (i < s.len and s[i] >= '0' and s[i] <= '9') : (i += 1) { + n = n * 10 + (s[i] - '0'); + } + + return if (neg) -n else n; +} + +// --------------------------------------------------------- // + +pub fn RESPONSE(req: *zix.Http1.Request, _: *zix.Http1.Response, _: *zix.Http1.Context) !void { + const head = req.head; + const body = try req.body(); + const fd = req.fd; + + var sum: i64 = sumQuery(head.query); + + if (req.method() == .POST and body.len > 0) { + sum += parseIntLoose(body); + } + + var body_buf: [32]u8 = undefined; + const out = std.fmt.bufPrint(&body_buf, "{d}", .{sum}) catch return; + + zix.Http1.sendSimpleFD(fd, 200, "text/plain", out) catch { + try zix.Http1.sendSimpleFD(fd, @intFromEnum(zix.Http1.Status.Code.INTERNAL_SERVER_ERROR), zix.Http1.Content.Type.TEXT_PLAIN.asString(), zix.Http1.Status.Code.INTERNAL_SERVER_ERROR.asString()); + }; +} diff --git a/frameworks/zix/src/handlers/crud.zig b/frameworks/zix/src/handlers/crud.zig new file mode 100644 index 000000000..e9a12c867 --- /dev/null +++ b/frameworks/zix/src/handlers/crud.zig @@ -0,0 +1,154 @@ +//! /crud/items : GET lists (paged, category filter) or reads one id, POST +//! creates, PUT updates. A single-item cache hit (crudcache.zig) answers +//! right away, everything else queues a Job on the worker's postgres lane +//! (dbpg.zig). + +const std = @import("std"); +const zix = @import("zix"); + +const dbpg = @import("../shared/dbpg.zig"); + +const response = @import("../shared/response.zig"); +const crudcache = @import("../shared/crudcache.zig"); + +// --------------------------------------------------------- // + +pub const PATH = "/crud/items"; + +const CRUD_LIST_LIMIT_DEFAULT: i64 = 10; +const CRUD_LIST_LIMIT_MAX: i64 = 100; + +// Per-worker scratch for the engine-side cache reads. +threadlocal var tl_db_body_buf: [32 * 1024]u8 = undefined; + +// Per-engine-worker arena for the crud POST/PUT JSON body parse. +threadlocal var tl_crud_arena: ?std.heap.ArenaAllocator = null; + +const CrudBody = struct { + id: i64 = 0, + name: []const u8 = "", + category: []const u8 = "", + price: i64 = 0, + quantity: i64 = 0, +}; + +// --------------------------------------------------------- // + +fn crudList(head: *const zix.Http1.ParsedHead, fd: std.posix.fd_t) void { + const category = zix.Http1.queryParam(head, "category") orelse ""; + var page: i64 = if (zix.Http1.queryParam(head, "page")) |raw| std.fmt.parseInt(i64, raw, 10) catch 1 else 1; + var limit: i64 = if (zix.Http1.queryParam(head, "limit")) |raw| std.fmt.parseInt(i64, raw, 10) catch CRUD_LIST_LIMIT_DEFAULT else CRUD_LIST_LIMIT_DEFAULT; + if (page < 1) page = 1; + if (limit < 1) limit = CRUD_LIST_LIMIT_DEFAULT; + if (limit > CRUD_LIST_LIMIT_MAX) limit = CRUD_LIST_LIMIT_MAX; + if (category.len > dbpg.CATEGORY_MAX) return response.badRequest(fd); + + var job: dbpg.Job = .{ .CRUD_LIST = .{ + .fd = fd, + .page = page, + .limit = limit, + .category_len = @intCast(category.len), + .category_buf = undefined, + } }; + @memcpy(job.CRUD_LIST.category_buf[0..category.len], category); + + if (!dbpg.submitJob(head, job)) response.serviceUnavailable(fd); +} + +fn parseCrudBody(body: []const u8) ?CrudBody { + if (tl_crud_arena == null) tl_crud_arena = std.heap.ArenaAllocator.init(std.heap.smp_allocator); + + const arena = &tl_crud_arena.?; + _ = arena.reset(.retain_capacity); + + return std.json.parseFromSliceLeaky(CrudBody, arena.allocator(), body, .{ + .ignore_unknown_fields = true, + }) catch null; +} + +fn crudCreate(head: *const zix.Http1.ParsedHead, body: []const u8, fd: std.posix.fd_t) void { + const item = parseCrudBody(body) orelse return response.badRequest(fd); + if (item.id < 1 or item.name.len == 0) return response.badRequest(fd); + if (item.name.len > dbpg.NAME_MAX or item.category.len > dbpg.CATEGORY_MAX) return response.badRequest(fd); + + var job: dbpg.Job = .{ .CRUD_CREATE = .{ + .fd = fd, + .id = item.id, + .price = item.price, + .quantity = item.quantity, + .name_len = @intCast(item.name.len), + .category_len = @intCast(item.category.len), + .name_buf = undefined, + .category_buf = undefined, + } }; + @memcpy(job.CRUD_CREATE.name_buf[0..item.name.len], item.name); + @memcpy(job.CRUD_CREATE.category_buf[0..item.category.len], item.category); + + if (!dbpg.submitJob(head, job)) response.serviceUnavailable(fd); +} + +/// 200 JSON response with the X-Cache header the crud cache check reads. +fn sendCrudBody(fd: std.posix.fd_t, body: []const u8, cache_state: []const u8) void { + var hdr_buf: [128]u8 = undefined; + const hdr = std.fmt.bufPrint(&hdr_buf, "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nX-Cache: {s}\r\nContent-Length: {d}\r\n\r\n", .{ cache_state, body.len }) catch return; + + zix.Http1.writeAllFD(fd, hdr) catch return; + zix.Http1.writeAllFD(fd, body) catch {}; +} + +fn crudUpdate(head: *const zix.Http1.ParsedHead, id: i64, body: []const u8, fd: std.posix.fd_t) void { + const item = parseCrudBody(body) orelse return response.badRequest(fd); + if (item.name.len > dbpg.NAME_MAX or item.category.len > dbpg.CATEGORY_MAX) return response.badRequest(fd); + + var job: dbpg.Job = .{ .CRUD_UPDATE = .{ + .fd = fd, + .id = id, + .price = item.price, + .quantity = item.quantity, + .name_len = @intCast(item.name.len), + .category_len = @intCast(item.category.len), + .name_buf = undefined, + .category_buf = undefined, + } }; + @memcpy(job.CRUD_UPDATE.name_buf[0..item.name.len], item.name); + @memcpy(job.CRUD_UPDATE.category_buf[0..item.category.len], item.category); + + if (!dbpg.submitJob(head, job)) response.serviceUnavailable(fd); +} + +// --------------------------------------------------------- // + +pub fn RESPONSE(req: *zix.Http1.Request, _: *zix.Http1.Response, _: *zix.Http1.Context) !void { + const head = req.head; + const body = try req.body(); + const fd = req.fd; + + const sub = head.path[PATH.len..]; + + if (sub.len == 0) { + if (std.mem.eql(u8, head.method, "GET")) return crudList(head, fd); + if (std.mem.eql(u8, head.method, "POST")) return crudCreate(head, body, fd); + + return response.notFound(fd); + } + + if (sub[0] != '/' or sub.len == 1) return response.notFound(fd); + const id = std.fmt.parseInt(i64, sub[1..], 10) catch return response.badRequest(fd); + + if (std.mem.eql(u8, head.method, "GET")) { + // in-process cache first: a HIT answers on the engine worker and + // never becomes a job + if (crudcache.get(id, &tl_db_body_buf)) |len| { + sendCrudBody(fd, tl_db_body_buf[0..len], "HIT"); + + return; + } + + if (!dbpg.submitJob(head, .{ .CRUD_GET = .{ .fd = fd, .id = id } })) response.serviceUnavailable(fd); + + return; + } + if (std.mem.eql(u8, head.method, "PUT")) return crudUpdate(head, id, body, fd); + + return response.notFound(fd); +} diff --git a/frameworks/zix/src/handlers/json.zig b/frameworks/zix/src/handlers/json.zig new file mode 100644 index 000000000..3e1e2c2f6 --- /dev/null +++ b/frameworks/zix/src/handlers/json.zig @@ -0,0 +1,124 @@ +//! GET /json/{count}?m=M : render count dataset items, total = price*qty*m. +//! Every response serializes per request from the loaded dataset, gzip +//! compresses per request when the client accepts it. + +const std = @import("std"); +const zix = @import("zix"); + +const response = @import("../shared/response.zig"); +const dataset = @import("../shared/dataset.zig"); +const util = @import("../shared/util.zig"); + +// --------------------------------------------------------- // + +pub const PATH = "/json"; + +/// Must initialize in main (the loaded dataset the bodies render from). +pub var g_dataset: dataset.Dataset = undefined; + +// Space in front of the body for the header, so header plus body go out as +// one slice. +const JSON_HDR_RESERVE: usize = 96; +threadlocal var json_resp_buf: [JSON_HDR_RESERVE + 32 * 1024]u8 = undefined; + +// --------------------------------------------------------- // + +fn sendJsonGzipFD(fd: std.posix.fd_t, json_body: []const u8) void { + zix.Http1.sendGzipFD(fd, 200, "application/json", json_body) catch {}; +} + +// --------------------------------------------------------- // + +/// Largest body one render can produce, measured at init. Sizes the reserve. +var g_body_max: usize = 0; + +pub fn init(aa: std.mem.Allocator) !void { + var dataset_path_buf: [512]u8 = undefined; + const data_dir = "/data"; + const dataset_path = + try std.fmt.bufPrint(&dataset_path_buf, "{s}/dataset.json", .{data_dir}); + g_dataset = try dataset.load(aa, dataset_path); + + const full = renderItems(json_resp_buf[JSON_HDR_RESERVE..], dataset.ItemCount, 1); + g_body_max = full + (dataset.ItemCount + 1) * 20; +} + +/// Render the items body for (count, m) into buf, returning the body length. +fn renderItems(buf: []u8, count: u8, m: u64) usize { + var pos: usize = 0; + + pos = util.appendStr(buf, pos, "{\"items\":["); + var i: usize = 0; + while (i < count) : (i += 1) { + if (i > 0) { + buf[pos] = ','; + pos += 1; + } + const item = g_dataset.items[i]; + @memcpy(buf[pos..][0..item.prefix.len], item.prefix); + pos += item.prefix.len; + pos = util.appendStr(buf, pos, ",\"total\":"); + pos = util.appendInt(buf, pos, item.pq * m); + buf[pos] = '}'; + pos += 1; + } + pos = util.appendStr(buf, pos, "],\"count\":"); + pos = util.appendInt(buf, pos, count); + buf[pos] = '}'; + pos += 1; + + return pos; +} + +pub fn RESPONSE(req: *zix.Http1.Request, _: *zix.Http1.Response, _: *zix.Http1.Context) !void { + const head = req.head; + const fd = req.fd; + + const accept = zix.Http1.acceptEncoding(head) orelse ""; + const want_gzip = std.mem.indexOf(u8, accept, "gzip") != null; + + // The PREFIX route also matches a bare /json (no trailing slash), which + // would slice out of bounds below. + if (head.path.len < "/json/".len) return response.badRequest(fd); + + const count_str = head.path["/json/".len..]; + const count = + std.fmt.parseInt(u8, count_str, 10) catch return response.badRequest(fd); + if (count < 1 or count > dataset.ItemCount) return response.badRequest(fd); + + const m: u64 = + if (zix.Http1.queryParam(head, "m")) |s| std.fmt.parseInt(u64, s, 10) catch + 1 else 1; + + // No-gzip path: render the body straight into the engine's send buffer, + // the engine puts the header in front. The body is written once, no copy. + if (!want_gzip) { + if (zix.Http1.responseReserve(fd, g_body_max)) |region| { + const body_len = renderItems(region, count, m); + + try zix.Http1.responseCommit(fd, 200, "application/json", body_len); + return; + } + } + + const buf = json_resp_buf[JSON_HDR_RESERVE..]; + const pos = renderItems(buf, count, m); + + if (want_gzip) { + sendJsonGzipFD(fd, buf[0..pos]); + return; + } + + // The header is built right before the body, so both go out as one slice. + var hdr_buf: [JSON_HDR_RESERVE]u8 = undefined; + const hdr = std.fmt.bufPrint(&hdr_buf, "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {d}\r\n\r\n", .{pos}) catch { + zix.Http1.sendJsonFD(fd, 200, buf[0..pos]) catch {}; + return; + }; + const start = JSON_HDR_RESERVE - hdr.len; + @memcpy(json_resp_buf[start..JSON_HDR_RESERVE], hdr); + + zix.Http1.writeAllFD(fd, json_resp_buf[start .. JSON_HDR_RESERVE + pos]) catch { + try zix.Http1.sendSimpleFD(fd, @intFromEnum(zix.Http1.Status.Code.INTERNAL_SERVER_ERROR), zix.Http1.Content.Type.TEXT_PLAIN.asString(), zix.Http1.Status.Code.INTERNAL_SERVER_ERROR.asString()); + }; +} diff --git a/frameworks/zix/src/handlers/pipeline.zig b/frameworks/zix/src/handlers/pipeline.zig new file mode 100644 index 000000000..9e6381340 --- /dev/null +++ b/frameworks/zix/src/handlers/pipeline.zig @@ -0,0 +1,21 @@ +//! GET /pipeline : fixed tiny response. writeAllFD appends to the engine's +//! staged sink, a pipelined batch leaves in request order as one send. + +const zix = @import("zix"); + +// --------------------------------------------------------- // + +pub const PATH = "/pipeline"; + +const PIPELINE_RESP: []const u8 = + "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: 2\r\n\r\nok"; + +// --------------------------------------------------------- // + +pub fn RESPONSE(req: *zix.Http1.Request, _: *zix.Http1.Response, _: *zix.Http1.Context) !void { + const fd = req.fd; + + zix.Http1.writeAllFD(req.fd, PIPELINE_RESP) catch { + try zix.Http1.sendSimpleFD(fd, @intFromEnum(zix.Http1.Status.Code.INTERNAL_SERVER_ERROR), zix.Http1.Content.Type.TEXT_PLAIN.asString(), zix.Http1.Status.Code.INTERNAL_SERVER_ERROR.asString()); + }; +} diff --git a/frameworks/zix/src/handlers/static.zig b/frameworks/zix/src/handlers/static.zig new file mode 100644 index 000000000..f604c5d6c --- /dev/null +++ b/frameworks/zix/src/handlers/static.zig @@ -0,0 +1,253 @@ +//! GET /static/{file} : serve from /data/static. Picks .br then .gz when the +//! client accepts them, else the plain file. One header send coalesced with +//! a zero-copy sendfile body. Resolved names are cached: open fd, size, and +//! a pre-built header per name. + +const std = @import("std"); +const zix = @import("zix"); + +const response = @import("../shared/response.zig"); +const encoding = @import("../shared/encoding.zig"); + +// --------------------------------------------------------- // + +pub const PATH = "/static"; + +/// Static cache name cap. Fixture names are short, anything longer is a 404. +pub const STATIC_NAME_MAX: usize = 96; +/// 20 fixtures times their (.br, .gz, identity) candidates plus headroom. +pub const STATIC_CACHE_MAX: usize = 128; +/// Name index over g_static_entries: name hash to (entry index + 1), 0 = +/// empty. Twice the entry capacity so a probe always ends at its entry or an +/// empty slot. Entries are never removed. +const STATIC_INDEX_SLOTS: usize = STATIC_CACHE_MAX * 2; + +var g_static_index: [STATIC_INDEX_SLOTS]u16 = @splat(0); +var g_static_entries: [STATIC_CACHE_MAX]StaticEntry = undefined; +var g_static_count: usize = 0; +var g_static_lock: std.atomic.Value(bool) = .init(false); + +pub const g_static_base: []const u8 = "/data/static/"; + +/// One servable static variant: a process-lifetime fd, its size, and the +/// pre-rendered response header. +const StaticVariant = struct { + fd: std.posix.fd_t, + size: u64, + hdr_len: u16, + hdr_buf: [192]u8, +}; +/// Cache slot for one resolved static name. A null variant caches a miss +/// so a bad name is probed only once. +const StaticEntry = struct { + name_len: u16, + // rel plus a 3-char precompressed suffix (".br" or ".gz") + name_buf: [STATIC_NAME_MAX + 3]u8, + variant: ?StaticVariant, +}; +/// Content type plus content encoding for a cached static name. +const StaticMeta = struct { + content_type: []const u8, + content_encoding: []const u8, +}; + +// --------------------------------------------------------- // + +fn staticNameHash(name: []const u8) u64 { + return std.hash.Wyhash.hash(0, name); +} + +fn staticLookup(name: []const u8) ?*const StaticEntry { + var slot: usize = @intCast(staticNameHash(name) & (STATIC_INDEX_SLOTS - 1)); + while (true) { + const tag = @atomicLoad(u16, &g_static_index[slot], .acquire); + if (tag == 0) return null; + + const entry = &g_static_entries[tag - 1]; + if (std.mem.eql(u8, entry.name_buf[0..entry.name_len], name)) return entry; + + slot = (slot + 1) & (STATIC_INDEX_SLOTS - 1); + } +} + +fn openStatic(name: []const u8) ?std.posix.fd_t { + var path_buf: [512]u8 = undefined; + const file_path = + std.fmt.bufPrint(&path_buf, "{s}{s}", .{ g_static_base, name }) catch + return null; + if (file_path.len >= path_buf.len) return null; + + path_buf[file_path.len] = 0; + + return std.posix.openatZ(std.posix.AT.FDCWD, @ptrCast(&path_buf), .{ .ACCMODE = .RDONLY }, 0) catch null; +} + +fn buildVariant(name: []const u8) ?StaticVariant { + const file_fd = openStatic(name) orelse return null; + + var stx: std.os.linux.Statx = undefined; + const stat_rc = + std.os.linux.statx(file_fd, "", std.os.linux.AT.EMPTY_PATH, .{ .SIZE = true }, &stx); + if (std.posix.errno(stat_rc) != .SUCCESS) { + _ = std.posix.system.close(file_fd); + return null; + } + + const size: u64 = stx.size; + const meta = staticMeta(name); + + var v: StaticVariant = .{ .fd = file_fd, .size = size, .hdr_len = 0, .hdr_buf = undefined }; + const hdr = (if (meta.content_encoding.len > 0) + std.fmt.bufPrint(&v.hdr_buf, "HTTP/1.1 200 OK\r\nContent-Type: {s}\r\nContent-Length: {d}\r\nContent-Encoding: {s}\r\n\r\n", .{ meta.content_type, size, meta.content_encoding }) + else + std.fmt.bufPrint(&v.hdr_buf, "HTTP/1.1 200 OK\r\nContent-Type: {s}\r\nContent-Length: {d}\r\n\r\n", .{ meta.content_type, size })) catch { + _ = std.posix.system.close(file_fd); + return null; + }; + v.hdr_len = @intCast(hdr.len); + + return v; +} + +fn staticInsert(name: []const u8) ?*const StaticEntry { + while (g_static_lock.swap(true, .acquire)) std.atomic.spinLoopHint(); + defer g_static_lock.store(false, .release); + + if (staticLookup(name)) |e| return e; + const count = @atomicLoad(usize, &g_static_count, .acquire); + if (count == STATIC_CACHE_MAX) return null; + + const e = &g_static_entries[count]; + e.name_len = @intCast(name.len); + @memcpy(e.name_buf[0..name.len], name); + e.variant = buildVariant(name); + @atomicStore(usize, &g_static_count, count + 1, .release); + + // Publish the index slot last (release), so a lock-free reader that sees + // the slot always sees the finished entry bytes behind it. + var slot: usize = @intCast(staticNameHash(name) & (STATIC_INDEX_SLOTS - 1)); + while (@atomicLoad(u16, &g_static_index[slot], .monotonic) != 0) + slot = (slot + 1) & (STATIC_INDEX_SLOTS - 1); + @atomicStore(u16, &g_static_index[slot], @intCast(count + 1), .release); + + return e; +} + +fn contentType(rel: []const u8) []const u8 { + const ext = std.fs.path.extension(rel); + const ext_no_dot = if (ext.len > 0) ext[1..] else ext; + + return zix.Http1.Content.fromExtension(ext_no_dot); +} + +fn waitWritable(fd: std.posix.fd_t) error{BrokenPipe}!void { + var pfd = + [_]std.posix.pollfd{.{ .fd = fd, .events = std.posix.POLL.OUT, .revents = 0 }}; + + _ = std.posix.poll(&pfd, -1) catch return error.BrokenPipe; +} + +fn sendMoreFD(fd: std.posix.fd_t, data: []const u8) error{BrokenPipe}!void { + const linux = std.os.linux; + + var rem = data; + while (rem.len > 0) { + const rc = linux.sendto(fd, rem.ptr, rem.len, linux.MSG.MORE, null, 0); + switch (std.posix.errno(rc)) { + .SUCCESS => { + const n: usize = @intCast(rc); + if (n == 0) return error.BrokenPipe; + + rem = rem[n..]; + }, + .INTR => {}, + .AGAIN => try waitWritable(fd), + else => return error.BrokenPipe, + } + } +} + +fn sendfileAll(sock: std.posix.fd_t, file_fd: std.posix.fd_t, size: u64) error{BrokenPipe}!void { + const linux = std.os.linux; + + var off: i64 = 0; + while (@as(u64, @intCast(off)) < size) { + const remaining: usize = @intCast(size - @as(u64, @intCast(off))); + const rc = linux.sendfile(sock, file_fd, &off, remaining); + switch (std.posix.errno(rc)) { + .SUCCESS => { + if (rc == 0) return error.BrokenPipe; + }, + .INTR => {}, + .AGAIN => try waitWritable(sock), + else => return error.BrokenPipe, + } + } +} + +// --------------------------------------------------------- // + +pub fn resolveStatic(name: []const u8) ?*const StaticEntry { + const entry = staticLookup(name) orelse staticInsert(name) orelse return null; + if (entry.variant == null) return null; + + return entry; +} + +pub fn staticMeta(name: []const u8) StaticMeta { + if (std.mem.endsWith(u8, name, ".br")) { + return .{ .content_type = contentType(name[0 .. name.len - ".br".len]), .content_encoding = "br" }; + } + if (std.mem.endsWith(u8, name, ".gz")) { + return .{ .content_type = contentType(name[0 .. name.len - ".gz".len]), .content_encoding = "gzip" }; + } + + return .{ .content_type = contentType(name), .content_encoding = "" }; +} + +pub fn RESPONSE(req: *zix.Http1.Request, _: *zix.Http1.Response, _: *zix.Http1.Context) !void { + const head = req.head; + const fd = req.fd; + + // The PREFIX route also matches a bare /static (no trailing slash), which + // would slice out of bounds below. + if (head.path.len < "/static/".len) return response.notFound(fd); + + const rel = head.path["/static/".len..]; + if (rel.len == 0 or + rel.len > STATIC_NAME_MAX or + std.mem.indexOf(u8, rel, "..") != null or + rel[0] == '/') return response.notFound(fd); + + const accept = encoding.accept(head); + + // Candidates "{rel}.br" / "{rel}.gz" / "{rel}". + var cand_buf: [STATIC_NAME_MAX + 3]u8 = undefined; + var entry: ?*const StaticEntry = null; + + if (accept.prefers_br) { + const cand = + std.fmt.bufPrint(&cand_buf, "{s}.br", .{rel}) catch + return response.notFound(fd); + entry = resolveStatic(cand); + } + if (entry == null and accept.accepts_gzip) { + const cand = std.fmt.bufPrint(&cand_buf, "{s}.gz", .{rel}) catch + return response.notFound(fd); + entry = resolveStatic(cand); + } + if (entry == null) { + entry = resolveStatic(rel); + } + + const served = entry orelse return response.notFound(fd); + const variant: *const StaticVariant = + if (served.variant) |*v| v else return response.notFound(fd); + + // Raw fd writes below: flush engine-staged responses first so the wire + // order matches the request order under pipelining. + zix.Http1.flushPending(fd); + + sendMoreFD(fd, variant.hdr_buf[0..variant.hdr_len]) catch return; + sendfileAll(fd, variant.fd, variant.size) catch {}; +} diff --git a/frameworks/zix/src/handlers/upload.zig b/frameworks/zix/src/handlers/upload.zig new file mode 100644 index 000000000..cd7f4b1a8 --- /dev/null +++ b/frameworks/zix/src/handlers/upload.zig @@ -0,0 +1,24 @@ +//! POST /upload : return how many body bytes were read off the socket +//! (bodyReceived), not the Content-Length value. + +const std = @import("std"); +const zix = @import("zix"); + +// --------------------------------------------------------- // + +pub const PATH = "/upload"; + +// --------------------------------------------------------- // + +pub fn RESPONSE(req: *zix.Http1.Request, _: *zix.Http1.Response, _: *zix.Http1.Context) !void { + const fd = req.fd; + + const n: u64 = req.bodyReceived(); + + var body_buf: [24]u8 = undefined; + const out = std.fmt.bufPrint(&body_buf, "{d}", .{n}) catch return; + + zix.Http1.sendSimpleFD(fd, @intFromEnum(zix.Http1.Status.Code.OK), zix.Http1.Content.Type.TEXT_PLAIN.asString(), out) catch { + try zix.Http1.sendSimpleFD(fd, @intFromEnum(zix.Http1.Status.Code.INTERNAL_SERVER_ERROR), zix.Http1.Content.Type.TEXT_PLAIN.asString(), zix.Http1.Status.Code.INTERNAL_SERVER_ERROR.asString()); + }; +} diff --git a/frameworks/zix/src/main.zig b/frameworks/zix/src/main.zig index 90fe1bd22..950753dbb 100644 --- a/frameworks/zix/src/main.zig +++ b/frameworks/zix/src/main.zig @@ -1,515 +1,86 @@ //! HttpArena: zix //! -//! zix HttpArena HTTP/1.1 entry point. -//! -//! Intent: demonstrate zix.Http1 (EPOLL dispatch model) against the HttpArena -//! HTTP/1.1 benchmark suite (baseline, pipelined, short-lived). -//! -//! Design choices: -//! - rawIntercept: called before any header parsing for each EPOLL request. -//! Handles /pipeline with zero parse overhead (direct byte-match + sink write), -//! direct byte-match before any parsing, avoiding the header scan loop. Routes that fall -//! through are handled by the Router dispatch with full parsing. -//! - Router: comptime route table (StaticStringMap for EXACT, inline for PREFIX). -//! - PIPELINE_RESP: precomputed response bytes; one sink.append per request, no -//! header build overhead. +//! zix.Http1 (.URING), Router-only: every request goes through the engine's +//! parser and the comptime Router, one handler module per route +//! (src/handlers/). TLS rides tls_port (dual listener, one worker fleet). +//! async-db and crud run over engine-worker lanes (dbpg.zig): each worker +//! owns a pipelined postgres connection on its own ring. crud reads also +//! check the in-process cache (crudcache.zig). const std = @import("std"); const zix = @import("zix"); -const dataset = @import("dataset.zig"); - -// --------------------------------------------------------- // - -const PORT: u16 = 8080; -const LISTEN_IP: []const u8 = "::"; -const DISPATCH_MODEL: zix.Http1.DispatchModel = .URING; -const KERNEL_BACKLOG: u31 = 16 * 1024; -/// Per-machine tuning profile (ADR-041 increment 5). The dev box is 12 threads -/// / 32 GB (lean, memory-bound), the competition box is 64 cores / 251 GB -/// (throughput, RAM-abundant). Only the recv buffer differs: workers auto-scale -/// (WORKERS = 0), and the backlog and cache are already sized for both. Select -/// .throughput for the 64-core deployment. -const Profile = enum { lean, throughput }; -const PROFILE: Profile = .lean; - -/// Per-connection recv buffer, heap-allocated once at accept time. .lean keeps -/// 4 KiB to hold the working set small (benchmark requests are under 300 bytes, -/// and large upload bodies are drained by the engine in chunks, so only headers, -/// always < 4 KiB, must fit). .throughput raises it to 16 KiB for deeper -/// pipelined batches per recv, which the RAM-abundant box absorbs. -const MAX_RECV_BUF: usize = switch (PROFILE) { - .lean => 4 * 1024, - .throughput => 16 * 1024, -}; -const MAX_HEADERS: u8 = 16; -const WORKERS: usize = 0; - -// Response cache (ADR-036), used by the /json endpoint only. The /json body is -// deterministic in (count, m) and large enough to clear the cache crossover -// (~4 KiB), so a hit replays the full response with zero serialization. The -// other endpoints stay below the crossover (baseline, pipeline, upload) or use -// their own zero-copy sendfile cache (static), so none of them enable it. -const CACHE_MAX_ENTRIES: u32 = 64; -/// Per-slot cap. A /json/50 response is near 12 KiB, so 32 KiB leaves headroom. -const CACHE_MAX_VALUE_BYTES: u32 = 32 * 1024; -/// Freshness window. The dataset is immutable for the process lifetime, so a -/// long TTL means each key is built once and replayed for the whole run. -const CACHE_TTL_MS: u32 = 60 * 1000; - -// Data directory, overridable via the ARENA_DATA env var (default /data, the -// container mount point). Lets the same binary run against a local data tree. -var g_static_base: []const u8 = "/data/static/"; -var g_static_base_buf: [256]u8 = undefined; - -// Per-worker scratch. The JSON body (count up to 50) tops out near 12 KiB; the -// assembled response (status line + headers + body) sits a little above it. -threadlocal var json_body_buf: [32 * 1024]u8 = undefined; -threadlocal var json_resp_buf: [32 * 1024]u8 = undefined; - -// --------------------------------------------------------- // - -var g_dataset: dataset.Dataset = undefined; - -// --------------------------------------------------------- // - -fn notFound(fd: std.posix.fd_t) void { - zix.Http1.writeSimple(fd, 404, "text/plain", "Not Found") catch {}; -} - -fn badRequest(fd: std.posix.fd_t) void { - zix.Http1.writeSimple(fd, 400, "text/plain", "bad request") catch {}; -} - -// --------------------------------------------------------- // - -// GET/POST /baseline11?a=..&b=.. : sum the query values, plus the POST body as -// an integer. Returns the sum as text/plain. -fn baselineHandler(head: *const zix.Http1.ParsedHead, body: []const u8, fd: std.posix.fd_t) void { - var sum: i64 = sumQuery(head.query); - - if (std.mem.eql(u8, head.method, "POST") and body.len > 0) { - sum += parseIntLoose(body); - } - - var body_buf: [32]u8 = undefined; - const out = std.fmt.bufPrint(&body_buf, "{d}", .{sum}) catch return; - - zix.Http1.writeSimple(fd, 200, "text/plain", out) catch {}; -} - -// Precomputed response for the pipeline endpoint: one memcpy per request into the -// response sink. No header build overhead on the hot path. -const PIPELINE_RESP: []const u8 = "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: 2\r\n\r\nok"; - -// GET /pipeline : fixed tiny response, the pipelined-throughput endpoint. -fn pipelineHandler(head: *const zix.Http1.ParsedHead, body: []const u8, fd: std.posix.fd_t) void { - _ = head; - _ = body; - - zix.Http1.fdWriteAll(fd, PIPELINE_RESP) catch {}; -} - -// Raw-request interceptor for the EPOLL dispatch model. Called before any header -// parsing on each inbound request. Handles /pipeline with zero parse overhead: -// byte-matches the path directly on rem, then appends PIPELINE_RESP to the -// coalescing RespSink. Unknown -// routes return null and fall through to the Router dispatch with full parsing. -// -// This is intentional benchmark infrastructure, not general HTTP parsing. It -// exploits knowledge that /pipeline is always a bare GET with no body, so the -// consumed length is always header_end + 4 ("\r\n\r\n"). -fn rawIntercept(rem: []const u8, header_end: usize, fd: std.posix.fd_t) ?usize { - // Must start with "GET /p" to qualify for this fast path. - if (rem.len < 24 or rem[0] != 'G' or rem[4] != '/' or rem[5] != 'p') return null; - // "GET /pipeline " is 15 bytes. Verify without scanning the request line. - if (!std.mem.eql(u8, rem[4..15], "/pipeline ")) return null; +const baseline = @import("handlers/baseline.zig"); +const pipeline = @import("handlers/pipeline.zig"); +const upload = @import("handlers/upload.zig"); +const static = @import("handlers/static.zig"); +const json = @import("handlers/json.zig"); +const asyncdb = @import("handlers/asyncdb.zig"); +const crud = @import("handlers/crud.zig"); - zix.Http1.fdWriteAll(fd, PIPELINE_RESP) catch {}; - - return header_end + 4; -} - -// GET /json/{count}?m=M : render count dataset items, total = price*qty*M. -// -// Response-cache aware. The body is deterministic in (count, m) and big enough -// to clear the cache crossover, so the full response is the ideal cache value. -// The cache key is hash(method, path, query), so every distinct /json/{count}?m=M -// caches under its own slot. A hit skips the whole build loop and replays the -// stored bytes; a miss builds the response and stores it on the way out. When -// the cache is disabled or full the path still works, it just always rebuilds. -fn jsonHandler(head: *const zix.Http1.ParsedHead, body: []const u8, fd: std.posix.fd_t) void { - _ = body; - - if (zix.Http1.cacheLookup(head)) |cached| { - zix.Http1.fdWriteAll(fd, cached) catch {}; - return; - } - - const count_str = head.path["/json/".len..]; - const count = std.fmt.parseInt(u8, count_str, 10) catch return badRequest(fd); - if (count < 1 or count > dataset.ItemCount) return badRequest(fd); - - const m: u64 = if (zix.Http1.queryParam(head, "m")) |s| std.fmt.parseInt(u64, s, 10) catch 1 else 1; - - const buf = &json_body_buf; - var pos: usize = 0; - - pos = appendStr(buf, pos, "{\"items\":["); - var i: usize = 0; - while (i < count) : (i += 1) { - if (i > 0) { - buf[pos] = ','; - pos += 1; - } - const item = g_dataset.items[i]; - @memcpy(buf[pos..][0..item.prefix.len], item.prefix); - pos += item.prefix.len; - pos = appendStr(buf, pos, ",\"total\":"); - pos = appendInt(buf, pos, item.pq * m); - buf[pos] = '}'; - pos += 1; - } - pos = appendStr(buf, pos, "],\"count\":"); - pos = appendInt(buf, pos, count); - buf[pos] = '}'; - pos += 1; - - // Assemble the full response so it can be cached and replayed verbatim. The - // header matches the engine's writeJson output (send_date_header is off, so - // there is no time-varying field to freeze in the cache). - const resp = &json_resp_buf; - const hdr = std.fmt.bufPrint(resp, "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {d}\r\n\r\n", .{pos}) catch { - zix.Http1.writeJson(fd, 200, buf[0..pos]) catch {}; - return; - }; - @memcpy(resp[hdr.len..][0..pos], buf[0..pos]); - - zix.Http1.writeWithCache(fd, head, resp[0 .. hdr.len + pos], CACHE_TTL_MS) catch {}; -} - -// POST /upload : return the received byte count. The Content-Length header is -// authoritative (curl/clients always send it here), and the engine drains the -// body for sizes beyond the read buffer, so this never touches the bytes. -fn uploadHandler(head: *const zix.Http1.ParsedHead, body: []const u8, fd: std.posix.fd_t) void { - const n: u64 = if (head.content_length > 0) head.content_length else body.len; - - var body_buf: [24]u8 = undefined; - const out = std.fmt.bufPrint(&body_buf, "{d}", .{n}) catch return; - - zix.Http1.writeSimple(fd, 200, "text/plain", out) catch {}; -} +const dbpg = @import("shared/dbpg.zig"); // --------------------------------------------------------- // -fn contentType(rel: []const u8) []const u8 { - if (std.mem.endsWith(u8, rel, ".css")) return "text/css"; - if (std.mem.endsWith(u8, rel, ".js")) return "application/javascript"; - if (std.mem.endsWith(u8, rel, ".json")) return "application/json"; - if (std.mem.endsWith(u8, rel, ".html")) return "text/html"; - if (std.mem.endsWith(u8, rel, ".svg")) return "image/svg+xml"; - if (std.mem.endsWith(u8, rel, ".woff2")) return "font/woff2"; - if (std.mem.endsWith(u8, rel, ".webp")) return "image/webp"; - - return "application/octet-stream"; -} - -fn openVariant(rel: []const u8, suffix: []const u8) ?std.posix.fd_t { - var path_buf: [512]u8 = undefined; - const path = std.fmt.bufPrint(&path_buf, "{s}{s}{s}", .{ g_static_base, rel, suffix }) catch return null; - if (path.len >= path_buf.len) return null; - - path_buf[path.len] = 0; - - return std.posix.openatZ(std.posix.AT.FDCWD, @ptrCast(&path_buf), .{ .ACCMODE = .RDONLY }, 0) catch null; -} - -// --------------------------------------------------------- // - -/// Static cache name cap. Fixture names are short, anything longer is a 404. -const STATIC_NAME_MAX = 96; -/// Static cache capacity: 20 fixtures plus room for cached 404 lookups. -const STATIC_CACHE_MAX = 64; - -/// One servable variant of a static file: a fd kept open for the process -/// lifetime, its size, and the pre-rendered response header. -const StaticVariant = struct { - fd: std.posix.fd_t, - size: u64, - hdr_len: u16, - hdr_buf: [192]u8, -}; - -/// Cache slot for one /static/{name} path. All-null variants cache a 404. -const StaticEntry = struct { - name_len: u16, - name_buf: [STATIC_NAME_MAX]u8, - identity: ?StaticVariant, - br: ?StaticVariant, - gz: ?StaticVariant, -}; - -// Append-only cache: readers scan 0..count lock-free (count is published -// with release ordering after the slot is fully written), the spinlock only -// serializes inserts (rare: one per distinct path, all during warmup). -var g_static_entries: [STATIC_CACHE_MAX]StaticEntry = undefined; -var g_static_count: usize = 0; -var g_static_lock: std.atomic.Mutex = .unlocked; - -/// Probe one variant on disk and build its cache record: open, fstat, and -/// pre-render the header so serving it later is send + sendfile only. -fn buildStaticVariant(rel: []const u8, suffix: []const u8, encoding: []const u8) ?StaticVariant { - const file_fd = openVariant(rel, suffix) orelse return null; - - var stx: std.os.linux.Statx = undefined; - const stat_rc = std.os.linux.statx(file_fd, "", std.os.linux.AT.EMPTY_PATH, .{ .SIZE = true }, &stx); - if (std.posix.errno(stat_rc) != .SUCCESS) { - _ = std.posix.system.close(file_fd); - return null; - } - - const size: u64 = stx.size; - const ct = contentType(rel); - - var v: StaticVariant = .{ .fd = file_fd, .size = size, .hdr_len = 0, .hdr_buf = undefined }; - const hdr = (if (encoding.len > 0) - std.fmt.bufPrint(&v.hdr_buf, "HTTP/1.1 200 OK\r\nContent-Type: {s}\r\nContent-Length: {d}\r\nContent-Encoding: {s}\r\n\r\n", .{ ct, size, encoding }) - else - std.fmt.bufPrint(&v.hdr_buf, "HTTP/1.1 200 OK\r\nContent-Type: {s}\r\nContent-Length: {d}\r\n\r\n", .{ ct, size })) catch { - _ = std.posix.system.close(file_fd); - return null; - }; - v.hdr_len = @intCast(hdr.len); - - return v; -} - -fn staticLookup(rel: []const u8, count: usize) ?*const StaticEntry { - for (g_static_entries[0..count]) |*e| { - if (std.mem.eql(u8, e.name_buf[0..e.name_len], rel)) return e; - } - - return null; -} - -/// Slow path on first request for a path: probe all variants and publish the -/// slot. Returns null only when the cache is full. -fn staticInsert(rel: []const u8) ?*const StaticEntry { - while (!g_static_lock.tryLock()) std.atomic.spinLoopHint(); - defer g_static_lock.unlock(); - const count = @atomicLoad(usize, &g_static_count, .acquire); - if (staticLookup(rel, count)) |e| return e; - if (count == STATIC_CACHE_MAX) return null; - - const e = &g_static_entries[count]; - e.name_len = @intCast(rel.len); - @memcpy(e.name_buf[0..rel.len], rel); - e.identity = buildStaticVariant(rel, "", ""); - e.br = buildStaticVariant(rel, ".br", "br"); - e.gz = buildStaticVariant(rel, ".gz", "gzip"); - - @atomicStore(usize, &g_static_count, count + 1, .release); - - return e; -} - -/// Block until fd is writable again. Used by the static send path to ride -/// out a full socket buffer, mirroring fdWriteAll's EAGAIN handling. -fn waitWritable(fd: std.posix.fd_t) error{BrokenPipe}!void { - var pfd = [_]std.posix.pollfd{.{ .fd = fd, .events = std.posix.POLL.OUT, .revents = 0 }}; - - _ = std.posix.poll(&pfd, -1) catch return error.BrokenPipe; -} - -/// Send with MSG_MORE so the header coalesces into the same packets as the -/// sendfile body that follows instead of leaving as its own small packet. -fn fdSendMore(fd: std.posix.fd_t, data: []const u8) error{BrokenPipe}!void { - const linux = std.os.linux; - - var rem = data; - while (rem.len > 0) { - const rc = linux.sendto(fd, rem.ptr, rem.len, linux.MSG.MORE, null, 0); - switch (std.posix.errno(rc)) { - .SUCCESS => { - const n: usize = @intCast(rc); - if (n == 0) return error.BrokenPipe; - - rem = rem[n..]; - }, - .INTR => {}, - .AGAIN => try waitWritable(fd), - else => return error.BrokenPipe, - } - } -} - -/// Zero-copy file body: kernel pages straight to the socket, no userspace -/// bounce buffer. A local offset keeps the shared cached fd position -/// untouched, so one fd serves all workers concurrently. -fn sendfileAll(sock: std.posix.fd_t, file_fd: std.posix.fd_t, size: u64) error{BrokenPipe}!void { - const linux = std.os.linux; - - var off: i64 = 0; - while (@as(u64, @intCast(off)) < size) { - const remaining: usize = @intCast(size - @as(u64, @intCast(off))); - const rc = linux.sendfile(sock, file_fd, &off, remaining); - switch (std.posix.errno(rc)) { - .SUCCESS => { - if (rc == 0) return error.BrokenPipe; - }, - .INTR => {}, - .AGAIN => try waitWritable(sock), - else => return error.BrokenPipe, - } - } -} - -/// Cache-full fallback: probe, serve, close. Keeps rarely-hit paths correct -/// without evicting anything. -fn staticServeUncached(rel: []const u8, want_br: bool, want_gzip: bool, fd: std.posix.fd_t) void { - const variant: StaticVariant = blk: { - if (want_br) { - if (buildStaticVariant(rel, ".br", "br")) |v| break :blk v; - } - if (want_gzip) { - if (buildStaticVariant(rel, ".gz", "gzip")) |v| break :blk v; - } - break :blk buildStaticVariant(rel, "", "") orelse return notFound(fd); - }; - defer _ = std.posix.system.close(variant.fd); - - // Raw fd writes below: flush engine-staged responses first to keep the - // wire order matching the request order under pipelining. - zix.Http1.flushPending(fd); - - fdSendMore(fd, variant.hdr_buf[0..variant.hdr_len]) catch return; - sendfileAll(fd, variant.fd, variant.size) catch {}; -} - -// GET /static/{file} : serve from /data/static with content negotiation. -// Prefers a precompressed .br then .gz variant when the client accepts it, -// falling back to the identity file. Content-Type is by extension. The first -// request for a path probes the disk and caches fd + size + pre-rendered -// header, every later request is one header send plus one zero-copy sendfile. -fn staticHandler(head: *const zix.Http1.ParsedHead, body: []const u8, fd: std.posix.fd_t) void { - _ = body; - - const rel = head.path["/static/".len..]; - if (rel.len == 0 or rel.len > STATIC_NAME_MAX or std.mem.indexOf(u8, rel, "..") != null or rel[0] == '/') return notFound(fd); - - const accept_encoding = zix.Http1.getHeader(head, "accept-encoding") orelse ""; - const want_br = std.mem.indexOf(u8, accept_encoding, "br") != null; - const want_gzip = std.mem.indexOf(u8, accept_encoding, "gzip") != null; - - const count = @atomicLoad(usize, &g_static_count, .acquire); - const entry = staticLookup(rel, count) orelse staticInsert(rel) orelse - return staticServeUncached(rel, want_br, want_gzip, fd); - - const variant: *const StaticVariant = blk: { - if (want_br) { - if (entry.br) |*v| break :blk v; - } - if (want_gzip) { - if (entry.gz) |*v| break :blk v; - } - if (entry.identity) |*v| break :blk v; - - return notFound(fd); - }; - - // This path writes to the fd directly (raw send + sendfile), so any - // engine-staged responses from earlier pipelined requests go first. - zix.Http1.flushPending(fd); - - fdSendMore(fd, variant.hdr_buf[0..variant.hdr_len]) catch return; - sendfileAll(fd, variant.fd, variant.size) catch {}; -} - -// --------------------------------------------------------- // - -// Comptime route table. EXACT routes use a StaticStringMap (O(1) hash lookup), -// PREFIX routes match on startsWith with a boundary check (longest match wins). -// rawIntercept handles /pipeline before this dispatch is reached for that route. const Routes = zix.Http1.Router(&[_]zix.Http1.Route{ - .{ .path = "/baseline11", .handler = baselineHandler }, - .{ .path = "/pipeline", .handler = pipelineHandler }, - .{ .path = "/upload", .handler = uploadHandler }, - .{ .path = "/json", .handler = jsonHandler, .kind = .PREFIX }, - .{ .path = "/static", .handler = staticHandler, .kind = .PREFIX }, + .{ .path = baseline.PATH, .handler = baseline.RESPONSE }, + .{ .path = pipeline.PATH, .handler = pipeline.RESPONSE }, + .{ .path = upload.PATH, .handler = upload.RESPONSE }, + .{ .path = asyncdb.PATH, .handler = asyncdb.RESPONSE }, + .{ .path = static.PATH, .handler = static.RESPONSE, .kind = .PREFIX }, + .{ .path = json.PATH, .handler = json.RESPONSE, .kind = .PREFIX }, + .{ .path = crud.PATH, .handler = crud.RESPONSE, .kind = .PREFIX }, }); -// --------------------------------------------------------- // - -fn sumQuery(query: []const u8) i64 { - var sum: i64 = 0; - var it = std.mem.tokenizeScalar(u8, query, '&'); - while (it.next()) |pair| { - if (std.mem.indexOfScalar(u8, pair, '=')) |eq| { - sum += std.fmt.parseInt(i64, pair[eq + 1 ..], 10) catch 0; - } - } - return sum; -} - -fn parseIntLoose(s: []const u8) i64 { - var i: usize = 0; - while (i < s.len and (s[i] == ' ' or s[i] == '\t' or s[i] == '\r' or s[i] == '\n')) i += 1; - - var neg = false; - if (i < s.len and s[i] == '-') { - neg = true; - i += 1; - } - - var n: i64 = 0; - while (i < s.len and s[i] >= '0' and s[i] <= '9') : (i += 1) { - n = n * 10 + (s[i] - '0'); - } - - return if (neg) -n else n; -} - -fn appendStr(out: []u8, pos: usize, s: []const u8) usize { - @memcpy(out[pos..][0..s.len], s); - return pos + s.len; -} - -fn appendInt(out: []u8, pos: usize, n: u64) usize { - var tmp: [24]u8 = undefined; - const s = std.fmt.bufPrint(&tmp, "{d}", .{n}) catch unreachable; - @memcpy(out[pos..][0..s.len], s); - return pos + s.len; -} - -// --------------------------------------------------------- // - pub fn main(process: std.process.Init) !void { - // Elevate scheduling priority (setpriority -19). Fails silently when the - // process lacks CAP_SYS_NICE, so no special capability is required for correctness. - _ = std.os.linux.syscall3(.setpriority, 0, 0, @as(usize, @bitCast(@as(isize, -19)))); - - const data_dir = process.environ_map.get("ARENA_DATA") orelse "/data"; - g_static_base = std.fmt.bufPrint(&g_static_base_buf, "{s}/static/", .{data_dir}) catch "/data/static/"; - - var dataset_path_buf: [512]u8 = undefined; - const dataset_path = try std.fmt.bufPrint(&dataset_path_buf, "{s}/dataset.json", .{data_dir}); + var json_alloc = std.heap.ArenaAllocator.init(std.heap.smp_allocator); + defer json_alloc.deinit(); + + var tls_alloc = std.heap.ArenaAllocator.init(std.heap.smp_allocator); + defer tls_alloc.deinit(); + + try json.init(json_alloc.allocator()); + + // DB endpoints: lanes open lazily per engine worker once DATABASE_URL is + // present (non-DB profiles open nothing, DB routes answer 503). + dbpg.init(process); + dbpg.start(); + + var tls = zix.Tls.Context.init(tls_alloc.allocator(), process.io, .{ + .cert_path = "/etc/zix-tls/server.cert", + .key_path = "/etc/zix-tls/server.key", + .alpn = &.{.HTTP_1_1}, + .min_version = .TLS_1_3, + }) catch |e| { + return e; + }; - g_dataset = try dataset.load(std.heap.smp_allocator, dataset_path); + // Park ring sized to peak conns per worker: 16384c is the deepest + // scenario and workers = 0 spawns one worker per CPU. + const cpus = std.Thread.getCpuCount() catch 8; + const park_len = @max(512, 16 * 1024 / cpus); - var server = zix.Http1.Server.initRaw(Routes.dispatch, rawIntercept, .{ + var server = zix.Http1.Server.init(Routes.dispatch, .{ .io = process.io, - .ip = LISTEN_IP, - .port = PORT, - .dispatch_model = DISPATCH_MODEL, - .kernel_backlog = KERNEL_BACKLOG, - .max_recv_buf = MAX_RECV_BUF, - .max_headers = MAX_HEADERS, - .workers = WORKERS, + .ip = "::", + .port = 8080, + .workers = 0, + .dispatch_model = .URING, + .tls = &tls, + .tls_port = 8081, + // .send_date_header = false, - .response_cache = true, - .cache_max_entries = CACHE_MAX_ENTRIES, - .cache_max_value_bytes = CACHE_MAX_VALUE_BYTES, - .cache_ttl_ms = CACHE_TTL_MS, + .max_response_headers = .{ .CUSTOM = 8 }, + // + .compress = true, + // + .kernel_backlog = 16 * 1024, + .max_recv_buf = 8 * 1024, + // + .uring_send_buf_size = 16 * 1024, + .uring_idle_pool_floor = 16, + .uring_idle_pool_ceiling = 1 * 1024, + .process_queue_len = park_len, }); defer server.deinit(); diff --git a/frameworks/zix/src/shared/crudcache.zig b/frameworks/zix/src/shared/crudcache.zig new file mode 100644 index 000000000..8e57aa6bf --- /dev/null +++ b/frameworks/zix/src/shared/crudcache.zig @@ -0,0 +1,98 @@ +//! In-process cache for crud single-item reads: direct-mapped slots with +//! per-slot spinlocks, X-Cache MISS/HIT, writes invalidate their id. + +const std = @import("std"); + +// --------------------------------------------------------- // + +/// Power of two, covers the 1..50000 bench id keyspace so every id owns its +/// slot (no collision eviction inside the TTL). +const SLOT_COUNT = 65536; +/// Rendered single-item JSON cap, a larger body skips the cache. The seed +/// data bounds a rendered item near 210 bytes (name <= 19, category <= 11, +/// tags <= 48). +const BODY_MAX = 256; + +/// Absolute item TTL, the crud profile specifies 200 ms. Writes also +/// invalidate their id. +const ITEM_TTL_MS: i64 = 200; + +const Slot = struct { + lock_flag: std.atomic.Value(bool) = .init(false), + id: i64 = 0, + expires_ms: i64 = 0, + len: u16 = 0, + body: [BODY_MAX]u8 = undefined, +}; + +var g_slots: [SLOT_COUNT]Slot = @splat(.{}); + +fn slotFor(id: i64) ?*Slot { + if (id < 1) return null; + + return &g_slots[@as(usize, @intCast(id)) & (SLOT_COUNT - 1)]; +} + +fn lock(flag: *std.atomic.Value(bool)) void { + while (flag.swap(true, .acquire)) std.atomic.spinLoopHint(); +} + +fn unlock(flag: *std.atomic.Value(bool)) void { + flag.store(false, .release); +} + +fn nowMs() i64 { + var ts: std.os.linux.timespec = undefined; + _ = std.os.linux.clock_gettime(.MONOTONIC_COARSE, &ts); + + return @as(i64, ts.sec) * 1000 + @divTrunc(@as(i64, ts.nsec), 1_000_000); +} + +/// Copy a fresh cached body for id into out. +/// +/// Return: +/// - the body length (out[0..len] is the item JSON) +/// - null on a miss (absent, expired, or a colliding id) +pub fn get(id: i64, out: []u8) ?usize { + const slot = slotFor(id) orelse return null; + + lock(&slot.lock_flag); + defer unlock(&slot.lock_flag); + + if (slot.id != id or slot.len == 0) return null; + if (nowMs() >= slot.expires_ms) return null; + + const len: usize = slot.len; + if (len > out.len) return null; + @memcpy(out[0..len], slot.body[0..len]); + + return len; +} + +/// Store a rendered body for id, TTL-bounded. An oversized body is skipped. +pub fn put(id: i64, body: []const u8) void { + if (body.len > BODY_MAX) return; + + const slot = slotFor(id) orelse return; + + lock(&slot.lock_flag); + defer unlock(&slot.lock_flag); + + slot.id = id; + slot.expires_ms = nowMs() + ITEM_TTL_MS; + slot.len = @intCast(body.len); + @memcpy(slot.body[0..body.len], body); +} + +/// Drop the cached body for id (write invalidation). +pub fn remove(id: i64) void { + const slot = slotFor(id) orelse return; + + lock(&slot.lock_flag); + defer unlock(&slot.lock_flag); + + if (slot.id == id) { + slot.len = 0; + slot.expires_ms = 0; + } +} diff --git a/frameworks/zix/src/dataset.zig b/frameworks/zix/src/shared/dataset.zig similarity index 95% rename from frameworks/zix/src/dataset.zig rename to frameworks/zix/src/shared/dataset.zig index 4d12be803..e3d94d56d 100644 --- a/frameworks/zix/src/dataset.zig +++ b/frameworks/zix/src/shared/dataset.zig @@ -1,5 +1,3 @@ -//! HttpArena: zix -//! //! Dataset loader for the /json endpoint. //! //! Loads the fixed 50-item benchmark dataset once at startup and pre-renders @@ -22,10 +20,6 @@ pub const Item = struct { pub const Dataset = struct { items: []Item, arena: std.heap.ArenaAllocator, - - pub fn deinit(self: *Dataset) void { - self.arena.deinit(); - } }; pub fn load(gpa: std.mem.Allocator, path: []const u8) !Dataset { @@ -69,7 +63,8 @@ fn readFileAlloc(aa: std.mem.Allocator, path: []const u8, max: usize) ![]u8 { if (path.len >= path_z.len) return error.NameTooLong; @memcpy(path_z[0..path.len], path); path_z[path.len] = 0; - const fd = try std.posix.openatZ(std.posix.AT.FDCWD, @ptrCast(&path_z), .{ .ACCMODE = .RDONLY }, 0); + const fd = + try std.posix.openatZ(std.posix.AT.FDCWD, @ptrCast(&path_z), .{ .ACCMODE = .RDONLY }, 0); defer _ = std.posix.system.close(fd); var buf: std.ArrayList(u8) = .empty; @@ -104,7 +99,7 @@ fn renderItemPrefix(buf: *std.ArrayList(u8), aa: std.mem.Allocator, obj: std.jso try buf.append(aa, ':'); try writeValue(buf, aa, kv.value_ptr.*); } - // Intentionally no closing `}` — caller appends `,"total":N}`. + // Intentionally no closing `}`: the caller appends `,"total":N}`. } fn writeValue(buf: *std.ArrayList(u8), aa: std.mem.Allocator, v: std.json.Value) !void { diff --git a/frameworks/zix/src/shared/dbpg.zig b/frameworks/zix/src/shared/dbpg.zig new file mode 100644 index 000000000..1f76828ae --- /dev/null +++ b/frameworks/zix/src/shared/dbpg.zig @@ -0,0 +1,1211 @@ +//! PostgreSQL for the DB endpoints (async-db, crud) over engine-worker +//! lanes: each engine worker owns one pipelined connection +//! (postgrez.dispatch.Line) pumped by its own ring (zix.Http1.uringWatchFd), +//! so a reply decodes, renders, and writes on the core that owns the client +//! socket. Handlers build a Job and call submitJob, replies come back +//! through onLaneReply on the same worker. + +const std = @import("std"); +const zix = @import("zix"); + +const crudcache = @import("crudcache.zig"); + +const postgrez = zix.Driver.postgrez; +const frontend = postgrez.frontend; +const backend = postgrez.backend; +const row = postgrez.row; + +// --------------------------------------------------------- // + +pub const NAME_MAX = 96; +pub const CATEGORY_MAX = 48; + +/// Every query returns the nine item columns, 16 bounds the decode scratch. +const MAX_COLUMNS = 16; + +/// Pipeline depth per connection (driver default). +const WINDOW = postgrez.dispatch.DEFAULT_WINDOW; + +/// Bytes one rendered DB body may reach before renderDbRow sheds it. A crud +/// list of 100 rows tops out near 21 KiB. +const DB_BODY_MAX = 32 * 1024; + +/// In-flight ASYNC_DB scans one lane allows: line count times this. Caps how +/// many concurrent price-range scans can pile onto the server at once. +/// Swept on the isolate bench: 8 reads better than 4 on both DB cells. +const SCAN_CAP_PER_CONN = 8; + +// SQL: column order is fixed here, the renderers decode cells by position. +const SQL_ASYNC_DB = "SELECT id, name, category, price, quantity, active, tags, rating_score, rating_count FROM items WHERE price BETWEEN $1 AND $2 LIMIT $3"; +const SQL_CRUD_LIST = "SELECT id, name, category, price, quantity, active, tags, rating_score, rating_count FROM items WHERE category = $1 ORDER BY id LIMIT $2 OFFSET $3"; // total = returned page size per spec, no count(*) OVER() +const SQL_CRUD_GET = "SELECT id, name, category, price, quantity, active, tags, rating_score, rating_count FROM items WHERE id = $1"; +const SQL_CRUD_UPSERT = "INSERT INTO items (id, name, category, price, quantity, active, tags, rating_score, rating_count) VALUES ($1, $2, $3, $4, $5, true, '[]', 0, 0) ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, category = EXCLUDED.category, price = EXCLUDED.price, quantity = EXCLUDED.quantity"; +const SQL_CRUD_UPDATE = "UPDATE items SET name = $2, category = $3, price = $4, quantity = $5 WHERE id = $1"; + +// Named prepared statements, one per SQL. Prepared on every connection at +// open, so per request only Bind plus Execute is sent. +const STMT_ASYNC_DB = "zix_async_db"; +const STMT_CRUD_LIST = "zix_crud_list"; +const STMT_CRUD_GET = "zix_crud_get"; +const STMT_CRUD_UPSERT = "zix_crud_upsert"; +const STMT_CRUD_UPDATE = "zix_crud_update"; + +const Prepared = struct { + name: []const u8, + sql: []const u8, +}; + +const STATEMENTS = [_]Prepared{ + .{ .name = STMT_ASYNC_DB, .sql = SQL_ASYNC_DB }, + .{ .name = STMT_CRUD_LIST, .sql = SQL_CRUD_LIST }, + .{ .name = STMT_CRUD_GET, .sql = SQL_CRUD_GET }, + .{ .name = STMT_CRUD_UPSERT, .sql = SQL_CRUD_UPSERT }, + .{ .name = STMT_CRUD_UPDATE, .sql = SQL_CRUD_UPDATE }, +}; + +// --------------------------------------------------------- // + +/// One parsed DB request. Strings are copied into fixed buffers because the +/// engine reuses its receive buffer the moment the handler returns. +pub const Job = union(enum) { + ASYNC_DB: struct { + fd: std.posix.fd_t, + min: i64, + max: i64, + limit: i64, + }, + CRUD_LIST: struct { + fd: std.posix.fd_t, + page: i64, + limit: i64, + category_len: u8, + category_buf: [CATEGORY_MAX]u8, + }, + CRUD_GET: struct { + fd: std.posix.fd_t, + id: i64, + }, + CRUD_CREATE: struct { + fd: std.posix.fd_t, + id: i64, + price: i64, + quantity: i64, + name_len: u8, + category_len: u8, + name_buf: [NAME_MAX]u8, + category_buf: [CATEGORY_MAX]u8, + }, + CRUD_UPDATE: struct { + fd: std.posix.fd_t, + id: i64, + price: i64, + quantity: i64, + name_len: u8, + category_len: u8, + name_buf: [NAME_MAX]u8, + category_buf: [CATEGORY_MAX]u8, + }, +}; + +fn jobFd(job: Job) std.posix.fd_t { + return switch (job) { + inline else => |request| request.fd, + }; +} + +/// Close-marked requests must be answered before the handler returns (the +/// engine closes the fd on return). submit pumps the lane until done is set. +const Completion = struct { + done: std.atomic.Value(bool) = .init(false), +}; + +/// One queued Job plus its optional close-marked completion signal. +const QueuedJob = struct { + job: Job, + completion: ?*Completion = null, +}; + +// --------------------------------------------------------- // + +/// In-flight Jobs one lane tracks, above its lines times WINDOW ceiling. +const SLOT_CAP = 1024; + +/// Client fds one worker can have parked mid-response at once. +const PENDING_CAP = 64; + +const Slot = struct { + job: Job = undefined, + completion: ?*Completion = null, +}; + +/// One stalled client write: the unsent remainder of a response, flushed +/// non-blocking every loop pass. len zero marks the slot free. +const PendingWrite = struct { + fd: std.posix.fd_t = 0, + sent: usize = 0, + len: usize = 0, + buf: [HEAD_MAX + DB_BODY_MAX]u8 = undefined, +}; + +/// Parked client writes for one worker: the unsent remainder of responses +/// whose socket was full, flushed non-blocking on every lane turn. +const WritePool = struct { + pw_slots: [PENDING_CAP]PendingWrite = @splat(.{}), + pw_active: usize = 0, +}; + +// Set once in init before start, read-only afterwards. +var g_io: std.Io = undefined; +var g_config: postgrez.Config = undefined; +var g_enabled: bool = false; +var g_conns: usize = 8; +var g_open: bool = false; + +// Prepared-statement bytes, encoded once in start(). Module scope because the +// lane mode opens connections lazily per worker, long after start() returned. +var g_prepare_buf: [4096]u8 = undefined; +var g_prepares: [STATEMENTS.len][]const u8 = undefined; + +/// The pending-write pool owned by the current worker, set at lane creation +/// so the send helpers reach it without threading a parameter through every +/// renderer. +threadlocal var tl_pool: ?*WritePool = null; + +/// Set around a close-marked reply: its delivery must block until fully +/// written because the engine worker closes the fd the moment it returns. +threadlocal var tl_write_blocking: bool = false; + +// Per-worker scratch. One reply is decoded at a time on a worker, so a +// per-thread buffer set is safe. +threadlocal var db_body_buf: [DB_BODY_MAX]u8 = undefined; +threadlocal var request_buf: [8 * 1024]u8 = undefined; +threadlocal var param_scratch: [8][24]u8 = undefined; + +// --------------------------------------------------------- // + +/// Total DB connections for a CPU budget when DATABASE_MAX_CONN is absent. +/// The arena postgres runs max_connections=256, 64 stays well inside it. +/// Small CPU budgets (the api profiles) double up: one postgres backend per +/// worker serializes the scans there, two keeps them parallel. Larger budgets +/// stay at one per worker, where two contends (isolate-swept). +fn connsFor(cpu: usize) usize { + if (cpu <= 4) return cpu * 2; + + return std.math.clamp(cpu, 4, 64); +} + +/// A crud read may have been filled between the engine-worker miss and this +/// dequeue. Answer from the cache when so, the database is not hit twice. +/// +/// Return: +/// - true when answered from the cache (Job consumed) +/// - false when the Job still needs the database +fn serveFromCache(item: QueuedJob) bool { + tl_write_blocking = item.completion != null; + defer tl_write_blocking = false; + + switch (item.job) { + .CRUD_GET => |request| { + if (crudcache.get(request.id, &db_body_buf)) |len| { + sendCrudBody(request.fd, db_body_buf[0..len], "HIT"); + signal(item); + + return true; + } + }, + else => {}, + } + + return false; +} + +/// Shed a Job the lane could not accept: 503 the fd and release any +/// close-request waiter. +fn shed(item: QueuedJob) void { + tl_write_blocking = item.completion != null; + defer tl_write_blocking = false; + + send503(jobFd(item.job)); + signal(item); +} + +/// Release a close-request waiter, no-op for a keep-alive Job. +fn signal(item: QueuedJob) void { + if (item.completion) |completion| completion.done.store(true, .release); +} + +/// Route one Job to its prepared statement name. +fn stmtName(job: Job) []const u8 { + return switch (job) { + .ASYNC_DB => STMT_ASYNC_DB, + .CRUD_LIST => STMT_CRUD_LIST, + .CRUD_GET => STMT_CRUD_GET, + .CRUD_CREATE => STMT_CRUD_UPSERT, + .CRUD_UPDATE => STMT_CRUD_UPDATE, + }; +} + +/// Encode one Job as Bind plus Describe plus Execute plus Sync against its +/// named prepared statement. Params bind as text, results come back binary. +/// The returned slice lives in request_buf, consumed by line.submit. +/// +/// Return: +/// - []const u8 request bytes +/// - error.OutOfMemory when the encoding overruns request_buf +fn buildRequest(job: Job) ![]const u8 { + var out: std.ArrayList(u8) = .empty; + + var fixed = std.heap.FixedBufferAllocator.init(&request_buf); + const allocator = fixed.allocator(); + + var params: [5]?[]const u8 = undefined; + switch (job) { + .ASYNC_DB => |request| { + params[0] = intParam(0, request.min); + params[1] = intParam(1, request.max); + params[2] = intParam(2, request.limit); + }, + .CRUD_LIST => |request| { + const offset = (request.page - 1) * request.limit; + params[0] = request.category_buf[0..request.category_len]; + params[1] = intParam(1, request.limit); + params[2] = intParam(2, offset); + }, + .CRUD_GET => |request| { + params[0] = intParam(0, request.id); + }, + .CRUD_CREATE => |request| { + params[0] = intParam(0, request.id); + params[1] = request.name_buf[0..request.name_len]; + params[2] = request.category_buf[0..request.category_len]; + params[3] = intParam(3, request.price); + params[4] = intParam(4, request.quantity); + }, + .CRUD_UPDATE => |request| { + params[0] = intParam(0, request.id); + params[1] = request.name_buf[0..request.name_len]; + params[2] = request.category_buf[0..request.category_len]; + params[3] = intParam(3, request.price); + params[4] = intParam(4, request.quantity); + }, + } + + const param_count: usize = switch (job) { + .ASYNC_DB, .CRUD_LIST => 3, + .CRUD_GET => 1, + .CRUD_CREATE, .CRUD_UPDATE => 5, + }; + + try frontend.bind(allocator, &out, "", stmtName(job), &.{}, params[0..param_count], &[_]frontend.Format{.BINARY}); + try frontend.describePortal(allocator, &out, ""); + try frontend.execute(allocator, &out, "", 0); + try frontend.sync(allocator, &out); + + return out.items; +} + +/// Text-encode an i64 parameter into slot `index` of the scratch, returning a +/// slice valid until the next buildRequest on this thread. +fn intParam(index: usize, value: i64) []const u8 { + return std.fmt.bufPrint(¶m_scratch[index], "{d}", .{value}) catch unreachable; +} + +// --------------------------------------------------------- // + +const RowShape = enum { ASYNC_DB, CRUD_LIST, CRUD_GET }; + +const RenderMeta = struct { + id: i64 = 0, + page: i64 = 0, + limit: i64 = 0, + category: []const u8 = "", +}; + +/// Walk a SELECT-shaped reply: capture the RowDescription columns, render each +/// DataRow with renderDbRow, and write the shaped JSON. A server error sheds +/// 503, a missing single item answers 404. +fn renderRows(reply: []const u8, fd: std.posix.fd_t, shape: RowShape, meta: RenderMeta) void { + var columns: [MAX_COLUMNS]row.ColumnInfo = undefined; + var column_count: usize = 0; + + var cells: [MAX_COLUMNS]?[]const u8 = undefined; + + const prefix = "{\"items\":["; + @memcpy(db_body_buf[0..prefix.len], prefix); + var pos: usize = prefix.len; + + var rows: usize = 0; + + var walk = MessageWalk{ .bytes = reply }; + while (walk.next() catch { + send503(fd); + return; + }) |message| { + switch (message) { + .error_response => { + send503(fd); + return; + }, + .row_description => |description| { + column_count = 0; + var it = description.iterator(); + while (it.next() catch null) |column| { + if (column_count >= MAX_COLUMNS) break; + + columns[column_count] = .{ .name = column.name, .type_oid = column.type_oid, .format = column.format }; + column_count += 1; + } + }, + .data_row => |data| { + var count: usize = 0; + var it = data.iterator(); + while (it.next() catch null) |cell| { + if (count >= MAX_COLUMNS) break; + + cells[count] = cell; + count += 1; + } + + if (shape == .CRUD_GET) { + const len = renderDbRow(&db_body_buf, 0, columns[0..column_count], cells[0..count]) catch { + send503(fd); + return; + }; + finishCrudGet(meta.id, db_body_buf[0..len], fd); + + return; + } + + if (rows > 0) { + db_body_buf[pos] = ','; + pos += 1; + } + pos = renderDbRow(&db_body_buf, pos, columns[0..column_count], cells[0..count]) catch { + send503(fd); + return; + }; + rows += 1; + }, + else => {}, + } + } + + if (shape == .CRUD_GET) { + send404(fd); + return; + } + + pos = appendStr(&db_body_buf, pos, "],"); + if (shape == .CRUD_LIST) { + pos = appendStr(&db_body_buf, pos, "\"total\":"); + pos = appendInt(&db_body_buf, pos, rows); + pos = appendStr(&db_body_buf, pos, ",\"page\":"); + pos = appendI64(&db_body_buf, pos, meta.page); + } else { + pos = appendStr(&db_body_buf, pos, "\"count\":"); + pos = appendInt(&db_body_buf, pos, rows); + } + db_body_buf[pos] = '}'; + pos += 1; + + sendJson(fd, db_body_buf[0..pos]); +} + +const WriteKind = enum { CREATE, UPDATE }; + +/// Drive a row-less write reply (insert or update) to ReadyForQuery. On +/// success the cached id is invalidated and a small JSON status answers, a +/// server error sheds 503. +fn finishWrite(reply: []const u8, fd: std.posix.fd_t, id: i64, kind: WriteKind) void { + var walk = MessageWalk{ .bytes = reply }; + while (walk.next() catch { + send503(fd); + return; + }) |message| { + if (message == .error_response) { + send503(fd); + return; + } + } + + invalidateCrud(id); + + switch (kind) { + .CREATE => sendStatus(fd, 201, "{\"status\":\"created\"}"), + .UPDATE => sendStatus(fd, 200, "{\"status\":\"ok\"}"), + } +} + +/// Iterator over the backend messages in one framed reply (up to and +/// including ReadyForQuery). +const MessageWalk = struct { + bytes: []const u8, + pos: usize = 0, + + fn next(self: *MessageWalk) !?backend.BackendMessage { + if (self.pos + 5 > self.bytes.len) return null; + + const tag = self.bytes[self.pos]; + const length = std.mem.readInt(u32, self.bytes[self.pos + 1 ..][0..4], .big); + if (length < 4) return error.BadMessage; + + const total = 1 + @as(usize, length); + if (self.pos + total > self.bytes.len) return null; + + const payload = self.bytes[self.pos + 5 .. self.pos + total]; + self.pos += total; + + return try backend.decode(tag, payload); + } +}; + +// --------------------------------------------------------- // + +/// MISS response plus the in-process cache fill. +fn finishCrudGet(id: i64, body: []const u8, fd: std.posix.fd_t) void { + crudcache.put(id, body); + + sendCrudBody(fd, body, "MISS"); +} + +/// Drop the cached crud body on every write. +fn invalidateCrud(id: i64) void { + crudcache.remove(id); +} + +// --------------------------------------------------------- // + +fn cellInt(columns: []const row.ColumnInfo, cells: []const ?[]const u8, index: usize) !i64 { + const bytes = cells[index] orelse return error.BadCell; + const column = columns[index]; + + return row.rawDecode(i64, @enumFromInt(column.type_oid), column.format, bytes); +} + +fn cellBool(columns: []const row.ColumnInfo, cells: []const ?[]const u8, index: usize) !bool { + const bytes = cells[index] orelse return error.BadCell; + const column = columns[index]; + + return row.rawDecode(bool, @enumFromInt(column.type_oid), column.format, bytes); +} + +fn cellStr(columns: []const row.ColumnInfo, cells: []const ?[]const u8, index: usize) ![]const u8 { + const bytes = cells[index] orelse return error.BadCell; + const column = columns[index]; + + return row.rawDecode([]const u8, @enumFromInt(column.type_oid), column.format, bytes); +} + +fn appendStr(out: []u8, pos: usize, text: []const u8) usize { + @memcpy(out[pos..][0..text.len], text); + + return pos + text.len; +} + +fn appendInt(out: []u8, pos: usize, value: u64) usize { + var tmp: [24]u8 = undefined; + const rendered = std.fmt.bufPrint(&tmp, "{d}", .{value}) catch unreachable; + @memcpy(out[pos..][0..rendered.len], rendered); + + return pos + rendered.len; +} + +fn appendI64(out: []u8, pos: usize, value: i64) usize { + var tmp: [24]u8 = undefined; + const rendered = std.fmt.bufPrint(&tmp, "{d}", .{value}) catch unreachable; + @memcpy(out[pos..][0..rendered.len], rendered); + + return pos + rendered.len; +} + +/// Append a JSON string body (quotes are the caller's), escaped. Worst case +/// is 6x the input, the renderDbRow budget accounts for it. +fn appendJsonStr(out: []u8, begin: usize, value: []const u8) usize { + const HEX = "0123456789abcdef"; + + var pos = begin; + for (value) |char| { + switch (char) { + '"', '\\' => { + out[pos] = '\\'; + out[pos + 1] = char; + pos += 2; + }, + 0x00...0x1f => { + out[pos..][0..4].* = "\\u00".*; + out[pos + 4] = HEX[char >> 4]; + out[pos + 5] = HEX[char & 0xf]; + pos += 6; + }, + else => { + out[pos] = char; + pos += 1; + }, + } + } + + return pos; +} + +/// Render one items row as a JSON object, cells by SQL column order. tags is +/// jsonb text, emitted raw. +fn renderDbRow(out: []u8, begin: usize, columns: []const row.ColumnInfo, cells: []const ?[]const u8) !usize { + const name = try cellStr(columns, cells, 1); + const category = try cellStr(columns, cells, 2); + const tags = try cellStr(columns, cells, 6); + if (begin + name.len * 6 + category.len * 6 + tags.len + 192 > out.len) return error.NoSpaceLeft; + + var pos = begin; + pos = appendStr(out, pos, "{\"id\":"); + pos = appendI64(out, pos, try cellInt(columns, cells, 0)); + pos = appendStr(out, pos, ",\"name\":\""); + pos = appendJsonStr(out, pos, name); + pos = appendStr(out, pos, "\",\"category\":\""); + pos = appendJsonStr(out, pos, category); + pos = appendStr(out, pos, "\",\"price\":"); + pos = appendI64(out, pos, try cellInt(columns, cells, 3)); + pos = appendStr(out, pos, ",\"quantity\":"); + pos = appendI64(out, pos, try cellInt(columns, cells, 4)); + pos = appendStr(out, pos, ",\"active\":"); + pos = appendStr(out, pos, if (try cellBool(columns, cells, 5)) "true" else "false"); + pos = appendStr(out, pos, ",\"tags\":"); + pos = appendStr(out, pos, tags); + pos = appendStr(out, pos, ",\"rating\":{\"score\":"); + pos = appendI64(out, pos, try cellInt(columns, cells, 7)); + pos = appendStr(out, pos, ",\"count\":"); + pos = appendI64(out, pos, try cellInt(columns, cells, 8)); + pos = appendStr(out, pos, "}}"); + + return pos; +} + +// --------------------------------------------------------- // + +/// Bytes one response head may reach (status line, Content-Type, +/// Content-Length, Date). +const HEAD_MAX = 192; + +/// Status lines for the lane-side responses, byte-matching the engine. +fn statusLine(status: u16) []const u8 { + return switch (status) { + 200 => "HTTP/1.1 200 OK\r\n", + 201 => "HTTP/1.1 201 Created\r\n", + 404 => "HTTP/1.1 404 Not Found\r\n", + 503 => "HTTP/1.1 503 Service Unavailable\r\n", + else => "HTTP/1.1 500 Internal Server Error\r\n", + }; +} + +/// Build a response head byte-matching what the engine fd writers emit +/// (Content-Type, Content-Length, Date). +fn buildHead(buf: []u8, status: u16, content_type: []const u8, body_len: usize) []const u8 { + var pos: usize = 0; + pos = appendStr(buf, pos, statusLine(status)); + pos = appendStr(buf, pos, "Content-Type: "); + pos = appendStr(buf, pos, content_type); + pos = appendStr(buf, pos, "\r\nContent-Length: "); + pos = appendInt(buf, pos, body_len); + pos = appendStr(buf, pos, "\r\nDate: "); + pos = appendStr(buf, pos, httpDate()); + pos = appendStr(buf, pos, "\r\n\r\n"); + + return buf[0..pos]; +} + +threadlocal var date_secs: u64 = 0; +threadlocal var date_len: usize = 0; +threadlocal var date_tick: u8 = 0; +threadlocal var date_buf: [40]u8 = undefined; + +/// Cached RFC 7231 date, refreshed at most once per 256 responses (the same +/// pacing as the engine's date cache). +fn httpDate() []const u8 { + date_tick +%= 1; + if (date_tick == 0 or date_len == 0) { + var ts: std.os.linux.timespec = undefined; + _ = std.os.linux.clock_gettime(.REALTIME, &ts); + + const secs: u64 = if (ts.sec >= 0) @intCast(ts.sec) else 0; + if (secs != date_secs or date_len == 0) { + date_len = formatHttpDate(secs, &date_buf).len; + date_secs = secs; + } + } + + return date_buf[0..date_len]; +} + +fn formatHttpDate(secs: u64, buf: []u8) []u8 { + const ep = std.time.epoch; + const es = ep.EpochSeconds{ .secs = secs }; + const epoch_day = es.getEpochDay(); + const year_day = epoch_day.calculateYearDay(); + const month_day = year_day.calculateMonthDay(); + const day_secs = es.getDaySeconds(); + const day_names = [_][]const u8{ "Sun", "Mon", "Tue", "Wed", "Thu", "Fri", "Sat" }; + const month_names = [_][]const u8{ "Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec" }; + const dow = (@as(u64, epoch_day.day) % 7 + 4) % 7; + + return std.fmt.bufPrint(buf, "{s}, {d:0>2} {s} {d} {d:0>2}:{d:0>2}:{d:0>2} GMT", .{ + day_names[dow], + @as(u32, month_day.day_index) + 1, + month_names[@intFromEnum(month_day.month) - 1], + year_day.year, + day_secs.getHoursIntoDay(), + day_secs.getMinutesIntoHour(), + day_secs.getSecondsIntoMinute(), + }) catch buf[0..0]; +} + +/// Write one response (head then body) without letting a stalled client +/// block the worker: send non-blocking, park the remainder on a full +/// socket buffer, keep per-fd order by appending behind a parked response. +/// +/// Note: +/// - a close-marked reply (tl_write_blocking) writes blocking instead, the +/// engine worker closes the fd the moment it returns. +fn deliver(fd: std.posix.fd_t, head: []const u8, body: []const u8) void { + const pool = tl_pool orelse { + deliverBlocking(fd, head, body, 0); + return; + }; + + if (tl_write_blocking) { + if (pool.pw_active > 0) { + if (findPending(pool, fd)) |slot| flushSlotBlocking(pool, slot); + } + + deliverBlocking(fd, head, body, 0); + + return; + } + + if (pool.pw_active > 0) { + if (findPending(pool, fd)) |slot| { + if (slot.len + head.len + body.len <= slot.buf.len) { + @memcpy(slot.buf[slot.len..][0..head.len], head); + @memcpy(slot.buf[slot.len + head.len ..][0..body.len], body); + slot.len += head.len + body.len; + + return; + } + + flushSlotBlocking(pool, slot); + } + } + + const total = head.len + body.len; + var sent: usize = 0; + while (sent < total) { + var iovs: [2]std.posix.iovec_const = undefined; + var iov_count: usize = 0; + if (sent < head.len) { + iovs[0] = .{ .base = head[sent..].ptr, .len = head.len - sent }; + iovs[1] = .{ .base = body.ptr, .len = body.len }; + iov_count = 2; + } else { + iovs[0] = .{ .base = body[sent - head.len ..].ptr, .len = total - sent }; + iov_count = 1; + } + + const msg = std.os.linux.msghdr_const{ + .name = null, + .namelen = 0, + .iov = &iovs, + .iovlen = iov_count, + .control = null, + .controllen = 0, + .flags = 0, + }; + const rc = std.os.linux.sendmsg(fd, &msg, std.os.linux.MSG.DONTWAIT | std.os.linux.MSG.NOSIGNAL); + switch (std.posix.errno(rc)) { + .SUCCESS => { + const sent_now: usize = @intCast(rc); + if (sent_now == 0) return; + + sent += sent_now; + }, + .INTR => continue, + .AGAIN => { + park(pool, fd, head, body, sent); + return; + }, + else => return, + } + } +} + +/// Blocking tail of a delivery, from `sent` bytes into head plus body. +fn deliverBlocking(fd: std.posix.fd_t, head: []const u8, body: []const u8, sent: usize) void { + if (sent < head.len) { + zix.Http1.writeAllFD(fd, head[sent..]) catch return; + zix.Http1.writeAllFD(fd, body) catch {}; + + return; + } + + zix.Http1.writeAllFD(fd, body[sent - head.len ..]) catch {}; +} + +/// Park the unsent remainder of a response in the worker's pending pool. A +/// full pool finishes blocking instead (the pre-parking behavior). +fn park(pool: *WritePool, fd: std.posix.fd_t, head: []const u8, body: []const u8, sent: usize) void { + const slot = allocPending(pool, fd) orelse { + deliverBlocking(fd, head, body, sent); + return; + }; + + var len: usize = 0; + if (sent < head.len) { + const head_rest = head[sent..]; + @memcpy(slot.buf[0..head_rest.len], head_rest); + len = head_rest.len; + @memcpy(slot.buf[len..][0..body.len], body); + len += body.len; + } else { + const body_rest = body[sent - head.len ..]; + @memcpy(slot.buf[0..body_rest.len], body_rest); + len = body_rest.len; + } + + slot.sent = 0; + slot.len = len; + pool.pw_active += 1; +} + +fn findPending(pool: *WritePool, fd: std.posix.fd_t) ?*PendingWrite { + for (&pool.pw_slots) |*slot| { + if (slot.len > 0 and slot.fd == fd) return slot; + } + + return null; +} + +fn allocPending(pool: *WritePool, fd: std.posix.fd_t) ?*PendingWrite { + for (&pool.pw_slots) |*slot| { + if (slot.len == 0) { + slot.fd = fd; + return slot; + } + } + + return null; +} + +/// Drain one parked slot blocking (order guard before a same-fd write). +fn flushSlotBlocking(pool: *WritePool, slot: *PendingWrite) void { + zix.Http1.writeAllFD(slot.fd, slot.buf[slot.sent..slot.len]) catch {}; + + slot.len = 0; + slot.sent = 0; + pool.pw_active -= 1; +} + +/// One non-blocking pass over the worker's parked responses. +/// +/// Return: +/// - true when any parked bytes moved or a slot freed +fn flushPendingWrites(pool: *WritePool) bool { + var progressed = false; + + var remaining = pool.pw_active; + for (&pool.pw_slots) |*slot| { + if (remaining == 0) break; + if (slot.len == 0) continue; + + remaining -= 1; + + while (slot.sent < slot.len) { + const rc = std.os.linux.sendto(slot.fd, slot.buf[slot.sent..].ptr, slot.len - slot.sent, std.os.linux.MSG.DONTWAIT | std.os.linux.MSG.NOSIGNAL, null, 0); + switch (std.posix.errno(rc)) { + .SUCCESS => { + const sent_now: usize = @intCast(rc); + if (sent_now == 0) break; + + slot.sent += sent_now; + progressed = true; + }, + .INTR => continue, + .AGAIN => break, + else => { + // client gone: drop the remainder + slot.sent = slot.len; + progressed = true; + }, + } + } + + if (slot.sent >= slot.len) { + slot.len = 0; + slot.sent = 0; + pool.pw_active -= 1; + } + } + + return progressed; +} + +fn sendJson(fd: std.posix.fd_t, body: []const u8) void { + var head_buf: [HEAD_MAX]u8 = undefined; + + deliver(fd, buildHead(&head_buf, 200, "application/json", body.len), body); +} + +fn sendStatus(fd: std.posix.fd_t, status: u16, body: []const u8) void { + var head_buf: [HEAD_MAX]u8 = undefined; + + deliver(fd, buildHead(&head_buf, status, "application/json", body.len), body); +} + +fn send503(fd: std.posix.fd_t) void { + const body = "Service Unavailable"; + var head_buf: [HEAD_MAX]u8 = undefined; + + deliver(fd, buildHead(&head_buf, 503, "text/plain", body.len), body); +} + +fn send404(fd: std.posix.fd_t) void { + const body = "Not Found"; + var head_buf: [HEAD_MAX]u8 = undefined; + + deliver(fd, buildHead(&head_buf, 404, "text/plain", body.len), body); +} + +/// 200 JSON response with the X-Cache header the crud cache check reads. +fn sendCrudBody(fd: std.posix.fd_t, body: []const u8, cache_state: []const u8) void { + var hdr_buf: [128]u8 = undefined; + const hdr = std.fmt.bufPrint(&hdr_buf, "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nX-Cache: {s}\r\nContent-Length: {d}\r\n\r\n", .{ cache_state, body.len }) catch return; + + deliver(fd, hdr, body); +} + +// --------------------------------------------------------- // + +/// Read DATABASE_URL and DATABASE_MAX_CONN once at startup. Absent or +/// malformed DATABASE_URL leaves the DB endpoints answering 503. +pub fn init(process: std.process.Init) void { + const url_text = process.environ_map.get("DATABASE_URL") orelse return; + + g_config = postgrez.parseUrl(url_text) catch return; + g_config.tls = .OFF; + g_config.dispatch_model = .URING; + g_io = process.io; + g_enabled = true; + + const cpu = std.Thread.getCpuCount() catch 8; + + if (process.environ_map.get("DATABASE_MAX_CONN")) |max_text| { + if (std.fmt.parseInt(usize, max_text, 10)) |parsed| { + if (parsed > 0) g_conns = parsed; + } else |_| {} + } else { + g_conns = connsFor(cpu); + } +} + +/// Encode the prepared-statement bytes and mark the lanes ready to open. +/// Does nothing when DATABASE_URL was absent, so non-DB profiles touch +/// nothing. +pub fn start() void { + if (!g_enabled) return; + + // Pre-encode one Parse plus Sync per named statement, handed to Line.open + // so every connection prepares them before the pipelined loop runs. + var fixed = std.heap.FixedBufferAllocator.init(&g_prepare_buf); + const allocator = fixed.allocator(); + + inline for (STATEMENTS, 0..) |spec, index| { + var out: std.ArrayList(u8) = .empty; + frontend.parse(allocator, &out, spec.name, spec.sql, &.{}) catch { + g_enabled = false; + return; + }; + frontend.sync(allocator, &out) catch { + g_enabled = false; + return; + }; + + g_prepares[index] = out.items; + } + + g_open = true; +} + +// Engine-worker lanes: one Lane per worker, single-threaded (every field is +// touched only by the owning worker). The embedded pool supplies the +// pending-write slots the shared send helpers park into through tl_pool. +const LANE_LINES_MAX = 4; + +const Lane = struct { + pool: WritePool = .{}, + lines: [LANE_LINES_MAX]*postgrez.dispatch.Line = undefined, + line_count: usize = 0, + slots: [SLOT_CAP]Slot = undefined, + slot_line: [SLOT_CAP]u8 = undefined, + slot_live: [SLOT_CAP]bool = @splat(false), + free: [SLOT_CAP]usize = undefined, + free_count: usize = 0, + queue: [SLOT_CAP]QueuedJob = undefined, + q_head: usize = 0, + q_tail: usize = 0, + scan_inflight: usize = 0, + scan_cap: usize = 0, +}; + +threadlocal var tl_lane: ?*Lane = null; + +/// This worker's lane, opened lazily on its first DB request (blocking +/// connect + prepare warm-up, once per worker). +fn laneGet() ?*Lane { + if (tl_lane) |lane| return lane; + if (!g_open) return null; + + const lane = std.heap.smp_allocator.create(Lane) catch return null; + lane.* = .{}; + + const cpu = std.Thread.getCpuCount() catch 8; + const want = std.math.clamp(g_conns / cpu, 1, LANE_LINES_MAX); + + var opened: usize = 0; + while (opened < want) : (opened += 1) { + lane.lines[opened] = postgrez.dispatch.Line.open(std.heap.smp_allocator, g_io, g_config, .{ + .conns = 1, + .window = WINDOW, + .context = lane, + .on_reply = onLaneReply, + .prepare = &g_prepares, + }) catch break; + } + + if (opened == 0) { + std.heap.smp_allocator.destroy(lane); + + return null; + } + + lane.line_count = opened; + lane.scan_cap = opened * SCAN_CAP_PER_CONN; + for (0..SLOT_CAP) |index| lane.free[index] = SLOT_CAP - 1 - index; + lane.free_count = SLOT_CAP; + + tl_lane = lane; + tl_pool = &lane.pool; + + zix.Http1.setExternalHandler(onLaneReadable); + var armed = false; + for (lane.lines[0..opened]) |line| { + if (zix.Http1.uringWatchFd(line.fd())) armed = true; + } + + if (!armed) { + for (lane.lines[0..opened]) |line| line.deinit(); + std.heap.smp_allocator.destroy(lane); + tl_lane = null; + tl_pool = null; + + return null; + } + + return lane; +} + +fn laneQueueFull(lane: *const Lane) bool { + return lane.q_tail -% lane.q_head >= SLOT_CAP; +} + +fn lanePush(lane: *Lane, item: QueuedJob) void { + lane.queue[lane.q_tail & (SLOT_CAP - 1)] = item; + lane.q_tail +%= 1; +} + +/// The least loaded line takes the next request. +fn laneLineFor(lane: *const Lane) usize { + var best: usize = 0; + for (lane.lines[1..lane.line_count], 1..) |line, index| { + if (line.pending() < lane.lines[best].pending()) best = index; + } + + return best; +} + +/// Read replies from every line that owes some. A dead connection sheds +/// everything it owes and leaves the rotation. +fn lanePump(lane: *Lane) void { + var index: usize = 0; + while (index < lane.line_count) { + const line = lane.lines[index]; + + if (line.pending() > 0) { + _ = line.pump() catch { + laneDropLine(lane, index); + continue; + }; + } + + index += 1; + } +} + +/// Shed every request a dead connection owes (503) and drop it from the +/// rotation, so a lost backend degrades to shedding instead of hangs. +fn laneDropLine(lane: *Lane, index: usize) void { + const line = lane.lines[index]; + + for (0..SLOT_CAP) |slot_index| { + if (!lane.slot_live[slot_index] or lane.slot_line[slot_index] != index) continue; + + const slot = lane.slots[slot_index]; + if (slot.job == .ASYNC_DB) lane.scan_inflight -= 1; + shed(.{ .job = slot.job, .completion = slot.completion }); + lane.slot_live[slot_index] = false; + lane.free[lane.free_count] = slot_index; + lane.free_count += 1; + } + + const last = lane.line_count - 1; + lane.lines[index] = lane.lines[last]; + for (0..SLOT_CAP) |slot_index| { + if (lane.slot_live[slot_index] and lane.slot_line[slot_index] == last) { + lane.slot_line[slot_index] = @intCast(index); + } + } + + lane.line_count -= 1; + line.deinit(); +} + +/// Drain queued jobs into the lines until a window fills, then flush any +/// parked client writes. Runs on the owning engine worker only. +fn laneDrain(lane: *Lane) void { + while (lane.line_count > 0 and lane.q_head != lane.q_tail) { + const item = lane.queue[lane.q_head & (SLOT_CAP - 1)]; + + if (item.job == .ASYNC_DB and lane.scan_inflight >= lane.scan_cap) break; + if (lane.free_count == 0) break; + + if (serveFromCache(item)) { + lane.q_head +%= 1; + continue; + } + + lane.free_count -= 1; + const tag = lane.free[lane.free_count]; + lane.slots[tag] = .{ .job = item.job, .completion = item.completion }; + + const request = buildRequest(item.job) catch { + lane.free_count += 1; + lane.q_head +%= 1; + shed(item); + + continue; + }; + + const line_index = laneLineFor(lane); + if (!lane.lines[line_index].submit(request, tag)) { + lane.free_count += 1; + + break; + } + + lane.slot_line[tag] = @intCast(line_index); + lane.slot_live[tag] = true; + if (item.job == .ASYNC_DB) lane.scan_inflight += 1; + lane.q_head +%= 1; + } + + for (lane.lines[0..lane.line_count]) |line| line.flush(); + + if (lane.pool.pw_active > 0) _ = flushPendingWrites(&lane.pool); +} + +/// Reply sink for the lane: render and send, then release the slot. +fn onLaneReply(context: ?*anyopaque, tag: u64, reply: []const u8) void { + const lane: *Lane = @ptrCast(@alignCast(context.?)); + + const slot = &lane.slots[@intCast(tag)]; + const job = slot.job; + const completion = slot.completion; + + tl_write_blocking = completion != null; + defer tl_write_blocking = false; + + switch (job) { + .ASYNC_DB => |request| renderRows(reply, request.fd, .ASYNC_DB, .{}), + .CRUD_LIST => |request| renderRows(reply, request.fd, .CRUD_LIST, .{ + .page = request.page, + .limit = request.limit, + .category = request.category_buf[0..request.category_len], + }), + .CRUD_GET => |request| renderRows(reply, request.fd, .CRUD_GET, .{ .id = request.id }), + .CRUD_CREATE => |request| finishWrite(reply, request.fd, request.id, .CREATE), + .CRUD_UPDATE => |request| finishWrite(reply, request.fd, request.id, .UPDATE), + } + + if (completion) |signal_ptr| signal_ptr.done.store(true, .release); + + if (job == .ASYNC_DB) lane.scan_inflight -= 1; + lane.slot_live[@intCast(tag)] = false; + lane.free[lane.free_count] = @intCast(tag); + lane.free_count += 1; +} + +/// Ring readable callback for a lane fd: pump and refill. The watch is +/// multishot, the engine keeps it armed. +fn onLaneReadable(fd: std.posix.fd_t) void { + _ = fd; + + const lane = tl_lane orelse return; + + lanePump(lane); + laneDrain(lane); +} + +/// Lane-mode submit. For a close request the same thread owns the pump, so +/// waiting means pumping. +fn laneSubmit(job: Job, keep_alive: bool) bool { + const lane = laneGet() orelse return false; + if (lane.line_count == 0) return false; + + lanePump(lane); + + if (keep_alive) { + if (laneQueueFull(lane)) return false; + + lanePush(lane, .{ .job = job }); + laneDrain(lane); + + return true; + } + + var completion: Completion = .{}; + if (laneQueueFull(lane)) return false; + + lanePush(lane, .{ .job = job, .completion = &completion }); + laneDrain(lane); + while (!completion.done.load(.acquire)) { + lanePump(lane); + laneDrain(lane); + } + + return true; +} + +/// Queue a Job on this worker's lane. +/// +/// Note: +/// - keep_alive false marks a close request: the call pumps the lane until +/// the response is written (the engine closes the fd on return). +/// +/// Return: +/// - true when the response is written or owned by the lane +/// - false when the lane is down or full (caller sheds 503) +pub fn submit(job: Job, keep_alive: bool) bool { + return laneSubmit(job, keep_alive); +} + +/// Queue a Job on this worker's lane. For a close request submit blocks +/// until the response is written, so a deferred write never races the fd +/// close. +pub fn submitJob(head: *const zix.Http1.ParsedHead, job: Job) bool { + return submit(job, head.keep_alive); +} diff --git a/frameworks/zix/src/shared/encoding.zig b/frameworks/zix/src/shared/encoding.zig new file mode 100644 index 000000000..3fb4fd602 --- /dev/null +++ b/frameworks/zix/src/shared/encoding.zig @@ -0,0 +1,27 @@ +//! Accept-Encoding parsing shared by the static and json handlers: a +//! substring scan for br/gzip tokens, no q-value parsing. + +const std = @import("std"); +const zix = @import("zix"); + +// --------------------------------------------------------- // + +/// Accept-Encoding tokens the client advertised (substring scan, the bench +/// header needs no q-value parsing). +pub const AcceptEncoding = struct { + prefers_br: bool, + accepts_gzip: bool, +}; + +// --------------------------------------------------------- // + +pub fn accept(head: *const zix.Http1.ParsedHead) AcceptEncoding { + const value = + zix.Http1.acceptEncoding(head) orelse + return .{ .prefers_br = false, .accepts_gzip = false }; + + return .{ + .prefers_br = std.mem.indexOf(u8, value, "br") != null, + .accepts_gzip = std.mem.indexOf(u8, value, "gzip") != null, + }; +} diff --git a/frameworks/zix/src/shared/response.zig b/frameworks/zix/src/shared/response.zig new file mode 100644 index 000000000..18a0c6288 --- /dev/null +++ b/frameworks/zix/src/shared/response.zig @@ -0,0 +1,18 @@ +//! Shared error responders (400/404/503) for the handlers. + +const std = @import("std"); +const zix = @import("zix"); + +// --------------------------------------------------------- // + +pub fn badRequest(fd: std.posix.fd_t) void { + zix.Http1.sendSimpleFD(fd, 400, "text/plain", "Bad Request") catch {}; +} + +pub fn notFound(fd: std.posix.fd_t) void { + zix.Http1.sendSimpleFD(fd, 404, "text/plain", "Not Found") catch {}; +} + +pub fn serviceUnavailable(fd: std.posix.fd_t) void { + zix.Http1.sendSimpleFD(fd, 503, "text/plain", "Service Unavailable") catch {}; +} diff --git a/frameworks/zix/src/shared/util.zig b/frameworks/zix/src/shared/util.zig new file mode 100644 index 000000000..8df9fa381 --- /dev/null +++ b/frameworks/zix/src/shared/util.zig @@ -0,0 +1,17 @@ +//! Tiny byte-buffer appenders (string, integer) shared by the json handler. + +const std = @import("std"); + +// --------------------------------------------------------- // + +pub fn appendStr(out: []u8, pos: usize, s: []const u8) usize { + @memcpy(out[pos..][0..s.len], s); + return pos + s.len; +} + +pub fn appendInt(out: []u8, pos: usize, n: u64) usize { + var tmp: [24]u8 = undefined; + const s = std.fmt.bufPrint(&tmp, "{d}", .{n}) catch unreachable; + @memcpy(out[pos..][0..s.len], s); + return pos + s.len; +} diff --git a/site/data/frameworks.json b/site/data/frameworks.json index 07438c545..500cc3b53 100644 --- a/site/data/frameworks.json +++ b/site/data/frameworks.json @@ -1019,10 +1019,10 @@ }, "zix": { "dir": "zix", - "description": "Zig HTTP/1.1 server on the zix.Http1 raw engine (no std.http). Shared-nothing: each worker runs its own SO_REUSEPORT multishot accept plus io_uring completion loop and owns its connections. The /json endpoint serves from the per-worker response cache, and request bodies larger than the read buffer are drained rather than buffered.", + "description": "Zig HTTP/1.1 server on the zix.Http1 raw engine, .URING dispatch: per-worker SO_REUSEPORT io_uring loops, comptime Router, staged send sink. Dual listener: cleartext 8080, TLS 1.3 on 8081. DB endpoints run on engine-worker lanes: each worker pumps its own pipelined postgrez connections on its ring, named prepared statements, cache-aside single-item crud reads (200 ms TTL).", "repo": "https://github.com/prothegee/zix", "type": "engine", - "engine": "zix", + "engine": "zix.Http1 URING Dispatch Model", "variants": [ { "dir": "zix-http2", diff --git a/site/data/results/zix.json b/site/data/results/zix.json index 67822789b..e25c59358 100644 --- a/site/data/results/zix.json +++ b/site/data/results/zix.json @@ -1,22 +1,94 @@ { "framework": "zix", "results": { + "api-16-1024": { + "framework": "zix", + "language": "Zig", + "rps": 243615, + "avg_latency": "1.25ms", + "p99_latency": "13.50ms", + "cpu": "958.6%", + "memory": "118MiB", + "connections": 1024, + "threads": 64, + "duration": "5s", + "pipeline": 1, + "bandwidth": "1.19GB/s", + "input_bw": "13.71MB/s", + "reconnects": 730807, + "status_2xx": 3654227, + "status_3xx": 0, + "status_4xx": 0, + "status_5xx": 0, + "tpl_baseline": 1370460, + "tpl_json": 1371308, + "tpl_db": 0, + "tpl_upload": 0, + "tpl_static": 0, + "tpl_async_db": 912459 + }, + "api-4-256": { + "framework": "zix", + "language": "Zig", + "rps": 67725, + "avg_latency": "1.15ms", + "p99_latency": "4.55ms", + "cpu": "215.0%", + "memory": "41MiB", + "connections": 256, + "threads": 64, + "duration": "5s", + "pipeline": 1, + "bandwidth": "339.91MB/s", + "input_bw": "3.81MB/s", + "reconnects": 203172, + "status_2xx": 1015889, + "status_3xx": 0, + "status_4xx": 0, + "status_5xx": 0, + "tpl_baseline": 380995, + "tpl_json": 381218, + "tpl_db": 0, + "tpl_upload": 0, + "tpl_static": 0, + "tpl_async_db": 253676 + }, + "async-db-1024": { + "framework": "zix", + "language": "Zig", + "rps": 406353, + "avg_latency": "1.94ms", + "p99_latency": "18.00ms", + "cpu": "3230.5%", + "memory": "325MiB", + "connections": 1024, + "threads": 64, + "duration": "5s", + "pipeline": 1, + "bandwidth": "1.53GB/s", + "input_bw": "27.13MB/s", + "reconnects": 162078, + "status_2xx": 4063536, + "status_3xx": 0, + "status_4xx": 0, + "status_5xx": 0 + }, "baseline-4096": { "framework": "zix", "language": "Zig", - "rps": 4492613, - "avg_latency": "911us", - "p99_latency": "1.19ms", - "cpu": "6413.1%", + "rps": 4398710, + "avg_latency": "930us", + "p99_latency": "1.26ms", + "cpu": "6426.3%", "memory": "186MiB", "connections": 4096, "threads": 64, "duration": "5s", "pipeline": 1, - "bandwidth": "282.68MB/s", - "input_bw": "347.04MB/s", + "bandwidth": "276.74MB/s", + "input_bw": "339.79MB/s", "reconnects": 0, - "status_2xx": 22463065, + "status_2xx": 21993554, "status_3xx": 0, "status_4xx": 0, "status_5xx": 0 @@ -24,19 +96,19 @@ "baseline-512": { "framework": "zix", "language": "Zig", - "rps": 4217982, - "avg_latency": "120us", - "p99_latency": "231us", - "cpu": "6348.6%", - "memory": "130MiB", + "rps": 4018752, + "avg_latency": "126us", + "p99_latency": "300us", + "cpu": "6491.1%", + "memory": "128MiB", "connections": 512, "threads": 64, "duration": "5s", "pipeline": 1, - "bandwidth": "265.42MB/s", - "input_bw": "325.83MB/s", + "bandwidth": "252.88MB/s", + "input_bw": "310.44MB/s", "reconnects": 0, - "status_2xx": 21089911, + "status_2xx": 20093764, "status_3xx": 0, "status_4xx": 0, "status_5xx": 0 @@ -60,22 +132,121 @@ "status_4xx": 0, "status_5xx": 0 }, + "crud-4096": { + "framework": "zix", + "language": "Zig", + "rps": 834511, + "avg_latency": "4.64ms", + "p99_latency": "22.70ms", + "cpu": "3304.5%", + "memory": "394MiB", + "connections": 4096, + "threads": 64, + "duration": "5s", + "pipeline": 1, + "bandwidth": "244.17MB/s", + "input_bw": "71.63MB/s", + "reconnects": 60650, + "status_2xx": 12517672, + "status_3xx": 0, + "status_4xx": 0, + "status_5xx": 0 + }, "json-4096": { "framework": "zix", "language": "Zig", - "rps": 2348626, - "avg_latency": "614us", - "p99_latency": "2.75ms", - "cpu": "5365.1%", - "memory": "290MiB", + "rps": 2405756, + "avg_latency": "719us", + "p99_latency": "2.97ms", + "cpu": "6061.2%", + "memory": "242MiB", + "connections": 4096, + "threads": 64, + "duration": "5s", + "pipeline": 1, + "bandwidth": "8.05GB/s", + "input_bw": "114.72MB/s", + "reconnects": 481346, + "status_2xx": 12028784, + "status_3xx": 0, + "status_4xx": 0, + "status_5xx": 0 + }, + "json-comp-16384": { + "framework": "zix", + "language": "Zig", + "rps": 1110217, + "avg_latency": "14.63ms", + "p99_latency": "21.70ms", + "cpu": "6821.1%", + "memory": "519MiB", + "connections": 16384, + "threads": 64, + "duration": "5s", + "pipeline": 1, + "bandwidth": "1.90GB/s", + "input_bw": "82.59MB/s", + "reconnects": 214287, + "status_2xx": 5551085, + "status_3xx": 0, + "status_4xx": 0, + "status_5xx": 0 + }, + "json-comp-4096": { + "framework": "zix", + "language": "Zig", + "rps": 1139676, + "avg_latency": "3.55ms", + "p99_latency": "7.79ms", + "cpu": "6627.7%", + "memory": "208MiB", + "connections": 4096, + "threads": 64, + "duration": "5s", + "pipeline": 1, + "bandwidth": "1.95GB/s", + "input_bw": "84.78MB/s", + "reconnects": 226110, + "status_2xx": 5698383, + "status_3xx": 0, + "status_4xx": 0, + "status_5xx": 0 + }, + "json-comp-512": { + "framework": "zix", + "language": "Zig", + "rps": 1024427, + "avg_latency": "468us", + "p99_latency": "1.85ms", + "cpu": "6249.5%", + "memory": "118MiB", + "connections": 512, + "threads": 64, + "duration": "5s", + "pipeline": 1, + "bandwidth": "1.75GB/s", + "input_bw": "76.20MB/s", + "reconnects": 204897, + "status_2xx": 5122139, + "status_3xx": 0, + "status_4xx": 0, + "status_5xx": 0 + }, + "json-tls-4096": { + "framework": "zix", + "language": "Zig", + "rps": 1922974, + "avg_latency": "1.13ms", + "p99_latency": "40.39ms", + "cpu": "5024.6%", + "memory": "273MiB", "connections": 4096, "threads": 64, "duration": "5s", "pipeline": 1, - "bandwidth": "7.86GB/s", - "input_bw": "111.99MB/s", - "reconnects": 468990, - "status_2xx": 11743130, + "bandwidth": "6.43GB", + "reconnects": 0, + "status_2xx": 9804595, "status_3xx": 0, "status_4xx": 0, "status_5xx": 0 @@ -83,19 +254,19 @@ "limited-conn-4096": { "framework": "zix", "language": "Zig", - "rps": 2751069, - "avg_latency": "1.35ms", - "p99_latency": "1.89ms", - "cpu": "5783.9%", - "memory": "237MiB", + "rps": 2698000, + "avg_latency": "1.38ms", + "p99_latency": "1.87ms", + "cpu": "5831.0%", + "memory": "229MiB", "connections": 4096, "threads": 64, "duration": "5s", "pipeline": 1, - "bandwidth": "173.09MB/s", - "input_bw": "212.51MB/s", - "reconnects": 1375655, - "status_2xx": 13755345, + "bandwidth": "169.75MB/s", + "input_bw": "208.41MB/s", + "reconnects": 1349624, + "status_2xx": 13490000, "status_3xx": 0, "status_4xx": 0, "status_5xx": 0 @@ -103,19 +274,19 @@ "limited-conn-512": { "framework": "zix", "language": "Zig", - "rps": 2740199, - "avg_latency": "170us", - "p99_latency": "403us", - "cpu": "5474.7%", - "memory": "150MiB", + "rps": 2640887, + "avg_latency": "178us", + "p99_latency": "438us", + "cpu": "5649.7%", + "memory": "122MiB", "connections": 512, "threads": 64, "duration": "5s", "pipeline": 1, - "bandwidth": "172.36MB/s", - "input_bw": "211.67MB/s", - "reconnects": 1369714, - "status_2xx": 13700995, + "bandwidth": "166.11MB/s", + "input_bw": "204.00MB/s", + "reconnects": 1320451, + "status_2xx": 13204438, "status_3xx": 0, "status_4xx": 0, "status_5xx": 0 @@ -123,18 +294,18 @@ "pipelined-4096": { "framework": "zix", "language": "Zig", - "rps": 55979164, - "avg_latency": "1.17ms", - "p99_latency": "1.79ms", - "cpu": "6454.4%", + "rps": 57794272, + "avg_latency": "1.13ms", + "p99_latency": "1.59ms", + "cpu": "6420.1%", "memory": "186MiB", "connections": 4096, "threads": 64, "duration": "5s", "pipeline": 16, - "bandwidth": "3.44GB/s", + "bandwidth": "3.55GB/s", "reconnects": 0, - "status_2xx": 279895824, + "status_2xx": 288971360, "status_3xx": 0, "status_4xx": 0, "status_5xx": 0 @@ -142,18 +313,18 @@ "pipelined-512": { "framework": "zix", "language": "Zig", - "rps": 53609380, - "avg_latency": "152us", - "p99_latency": "344us", - "cpu": "6399.9%", - "memory": "134MiB", + "rps": 53079958, + "avg_latency": "153us", + "p99_latency": "317us", + "cpu": "6387.0%", + "memory": "124MiB", "connections": 512, "threads": 64, "duration": "5s", "pipeline": 16, - "bandwidth": "3.29GB/s", + "bandwidth": "3.26GB/s", "reconnects": 0, - "status_2xx": 268046903, + "status_2xx": 265399791, "status_3xx": 0, "status_4xx": 0, "status_5xx": 0 @@ -161,18 +332,18 @@ "static-1024": { "framework": "zix", "language": "Zig", - "rps": 2025521, - "avg_latency": "300.92us", - "p99_latency": "3.21ms", - "cpu": "5488.4%", - "memory": "140MiB", + "rps": 2022673, + "avg_latency": "303.60us", + "p99_latency": "2.79ms", + "cpu": "5577.6%", + "memory": "141MiB", "connections": 1024, "threads": 64, "duration": "5s", "pipeline": 1, - "bandwidth": "30.86GB", + "bandwidth": "30.81GB", "reconnects": 0, - "status_2xx": 10331115, + "status_2xx": 10315518, "status_3xx": 0, "status_4xx": 0, "status_5xx": 0 @@ -180,18 +351,18 @@ "static-4096": { "framework": "zix", "language": "Zig", - "rps": 2036680, - "avg_latency": "1.05ms", - "p99_latency": "7.16ms", - "cpu": "5503.2%", - "memory": "195MiB", + "rps": 2018326, + "avg_latency": "1.06ms", + "p99_latency": "4.74ms", + "cpu": "5590.9%", + "memory": "205MiB", "connections": 4096, "threads": 64, "duration": "5s", "pipeline": 1, - "bandwidth": "31.03GB", + "bandwidth": "30.75GB", "reconnects": 0, - "status_2xx": 10386093, + "status_2xx": 10293028, "status_3xx": 0, "status_4xx": 0, "status_5xx": 0 @@ -199,18 +370,18 @@ "static-6800": { "framework": "zix", "language": "Zig", - "rps": 1986456, - "avg_latency": "1.75ms", - "p99_latency": "5.18ms", - "cpu": "5269.1%", - "memory": "241MiB", + "rps": 1976393, + "avg_latency": "1.76ms", + "p99_latency": "7.78ms", + "cpu": "5443.8%", + "memory": "265MiB", "connections": 6800, "threads": 64, "duration": "5s", "pipeline": 1, - "bandwidth": "30.26GB", + "bandwidth": "30.11GB", "reconnects": 0, - "status_2xx": 10111205, + "status_2xx": 10080305, "status_3xx": 0, "status_4xx": 0, "status_5xx": 0 @@ -237,19 +408,19 @@ "upload-256": { "framework": "zix", "language": "Zig", - "rps": 6661, - "avg_latency": "38.12ms", - "p99_latency": "45.10ms", - "cpu": "943.2%", - "memory": "130MiB", + "rps": 6469, + "avg_latency": "34.83ms", + "p99_latency": "78.10ms", + "cpu": "889.0%", + "memory": "128MiB", "connections": 256, "threads": 64, "duration": "5s", "pipeline": 1, - "bandwidth": "463.46KB/s", - "input_bw": "52.83GB/s", - "reconnects": 6688, - "status_2xx": 33441, + "bandwidth": "450.10KB/s", + "input_bw": "51.31GB/s", + "reconnects": 6519, + "status_2xx": 32479, "status_3xx": 0, "status_4xx": 0, "status_5xx": 0 @@ -257,19 +428,19 @@ "upload-32": { "framework": "zix", "language": "Zig", - "rps": 8185, - "avg_latency": "3.88ms", - "p99_latency": "10.90ms", - "cpu": "1065.4%", - "memory": "124MiB", + "rps": 8944, + "avg_latency": "3.55ms", + "p99_latency": "9.98ms", + "cpu": "1048.6%", + "memory": "121MiB", "connections": 32, "threads": 64, "duration": "5s", "pipeline": 1, - "bandwidth": "569.93KB/s", - "input_bw": "64.92GB/s", - "reconnects": 8199, - "status_2xx": 41007, + "bandwidth": "622.86KB/s", + "input_bw": "70.94GB/s", + "reconnects": 8961, + "status_2xx": 44812, "status_3xx": 0, "status_4xx": 0, "status_5xx": 0 diff --git a/site/static/logs/api-16/1024/zix.log b/site/static/logs/api-16/1024/zix.log new file mode 100644 index 000000000..e69de29bb diff --git a/site/static/logs/api-4/256/zix.log b/site/static/logs/api-4/256/zix.log new file mode 100644 index 000000000..e69de29bb diff --git a/site/static/logs/async-db/1024/zix.log b/site/static/logs/async-db/1024/zix.log new file mode 100644 index 000000000..e69de29bb diff --git a/site/static/logs/crud/4096/zix.log b/site/static/logs/crud/4096/zix.log new file mode 100644 index 000000000..e69de29bb diff --git a/site/static/logs/json-comp/16384/zix.log b/site/static/logs/json-comp/16384/zix.log new file mode 100644 index 000000000..e69de29bb diff --git a/site/static/logs/json-comp/4096/zix.log b/site/static/logs/json-comp/4096/zix.log new file mode 100644 index 000000000..e69de29bb diff --git a/site/static/logs/json-comp/512/zix.log b/site/static/logs/json-comp/512/zix.log new file mode 100644 index 000000000..e69de29bb diff --git a/site/static/logs/json-tls/4096/zix.log b/site/static/logs/json-tls/4096/zix.log new file mode 100644 index 000000000..e69de29bb