c4ad90e2
refactor: one ngtcp2 egress loop, one expiry sum, one guarded read
a73x 2026-08-29 10:01
Commit message
src/quic.zig
| Old | New | ||
|---|---|---|---|
| @@ -565,6 +565,104 @@ pub fn accountWrite(out: *Egress, wrote: c.ngtcp2_ssize, n: c.ngtcp2_ssize) Writ | |||
| 565 | return .cont; | 565 | return .cont; |
| 566 | } | 566 | } |
| 567 | 567 | ||
| 568 | // --------------------------------------------------------------------------- | ||
| 569 | // The per-connection core. `Client` below and `Listener.Conn` across the seam | ||
| 570 | // are different objects — one dials, one accepts; one has a connected socket | ||
| 571 | // and dies of a refusal, the other has an unconnected one and cannot be told | ||
| 572 | // — but the ngtcp2 they drive is the same library run the same way. What | ||
| 573 | // follows is that: the parts where a difference would be a BUG, not a design. | ||
| 574 | // --------------------------------------------------------------------------- | ||
| 575 | |||
| 576 | /// Milliseconds until ngtcp2 next wants servicing on `conn`, capped. A | ||
| 577 | /// connection with nothing due, or none at all, wants the cap. Feeds a | ||
| 578 | /// poll timeout at both ends: no timerfd anywhere, deliberately. | ||
| 579 | pub fn expiryMs(conn: ?*c.ngtcp2_conn, cap_ms: i32, now: u64) i32 { | ||
| 580 | const cn = conn orelse return cap_ms; | ||
| 581 | const expiry = c.ngtcp2_conn_get_expiry(cn); | ||
| 582 | if (expiry == std.math.maxInt(u64)) return cap_ms; | ||
| 583 | if (expiry <= now) return 0; | ||
| 584 | return @intCast(@min(@as(u64, @intCast(cap_ms)), (expiry - now) / 1_000_000)); | ||
| 585 | } | ||
| 586 | |||
| 587 | /// One packet in, re-entrancy counter held across it. Nonzero is dead. | ||
| 588 | pub fn readPkt( | ||
| 589 | conn: *c.ngtcp2_conn, | ||
| 590 | path: *c.ngtcp2_path, | ||
| 591 | pkt: []const u8, | ||
| 592 | depth: *u8, | ||
| 593 | ) c.ngtcp2_ssize { | ||
| 594 | var pi: c.ngtcp2_pkt_info = .{ .ecn = 0 }; | ||
| 595 | depth.* += 1; | ||
| 596 | defer depth.* -= 1; | ||
| 597 | return c.ngtcp2_conn_read_pkt(conn, path, &pi, pkt.ptr, pkt.len, timestampNs()); | ||
| 598 | } | ||
| 599 | |||
| 600 | /// How a drain pass ended. `failed` is ngtcp2 refusing the connection: the | ||
| 601 | /// client marks itself dead, the listener discards it and leaves the verdict | ||
| 602 | /// to `tick`'s expiry, as it always has. A socket that would not take a | ||
| 603 | /// datagram ends the pass as `done`, because the next pass offers it again. | ||
| 604 | pub const DrainEnd = enum { done, failed }; | ||
| 605 | |||
| 606 | /// The egress loop: from the ring, into ngtcp2, onto the wire, repeat. | ||
| 607 | /// `sendPkt` is the seam where the two sockets differ. | ||
| 608 | pub fn drainConn( | ||
| 609 | conn: *c.ngtcp2_conn, | ||
| 610 | out: *Egress, | ||
| 611 | stream_id: i64, | ||
| 612 | ctx: anytype, | ||
| 613 | comptime sendPkt: fn (@TypeOf(ctx), []const u8) bool, | ||
| 614 | ) DrainEnd { | ||
| 615 | var buf: [max_udp]u8 = undefined; | ||
| 616 | // Set when the peer's stream window is full. Everything that is not | ||
| 617 | // stream data still has to leave. | ||
| 618 | var stream_blocked = false; | ||
| 619 | while (true) { | ||
| 620 | var ps: c.ngtcp2_path_storage = undefined; | ||
| 621 | c.ngtcp2_path_storage_zero(&ps); | ||
| 622 | var pi: c.ngtcp2_pkt_info = .{ .ecn = 0 }; | ||
| 623 | var wrote: c.ngtcp2_ssize = 0; | ||
| 624 | |||
| 625 | var vecs: [2]c.ngtcp2_vec = undefined; | ||
| 626 | var vcnt: usize = 0; | ||
| 627 | var sid: i64 = -1; | ||
| 628 | if (!stream_blocked and stream_id != -1) { | ||
| 629 | vcnt = out.vecs(&vecs); | ||
| 630 | if (vcnt > 0) sid = stream_id; | ||
| 631 | } | ||
| 632 | |||
| 633 | const n = c.ngtcp2_conn_writev_stream_versioned( | ||
| 634 | conn, | ||
| 635 | &ps.path, | ||
| 636 | c.NGTCP2_PKT_INFO_VERSION, | ||
| 637 | &pi, | ||
| 638 | &buf, | ||
| 639 | buf.len, | ||
| 640 | &wrote, | ||
| 641 | 0, | ||
| 642 | sid, | ||
| 643 | if (vcnt > 0) &vecs else null, | ||
| 644 | vcnt, | ||
| 645 | timestampNs(), | ||
| 646 | ); | ||
| 647 | if (n == c.NGTCP2_ERR_STREAM_DATA_BLOCKED or n == c.NGTCP2_ERR_STREAM_SHUT_WR) { | ||
| 648 | // A documented return, not a failure: the peer has no window | ||
| 649 | // left for this stream. Treating it as fatal abandoned the whole | ||
| 650 | // egress loop — including the ACKs and the flow-control updates | ||
| 651 | // that are how the peer's window reopens, which turns a moment | ||
| 652 | // of backpressure into a stall that never resolves. Retry the | ||
| 653 | // same iteration carrying no stream data. | ||
| 654 | stream_blocked = true; | ||
| 655 | continue; | ||
| 656 | } | ||
| 657 | switch (accountWrite(out, wrote, n)) { | ||
| 658 | .stop => return .failed, | ||
| 659 | .brk => return .done, | ||
| 660 | .cont => {}, | ||
| 661 | } | ||
| 662 | if (!sendPkt(ctx, buf[0..@intCast(n)])) return .done; | ||
| 663 | } | ||
| 664 | } | ||
| 665 | |||
| 568 | test "Egress: a byte does not move until it is acked, and the ring wraps" { | 666 | test "Egress: a byte does not move until it is acked, and the ring wraps" { |
| 569 | const alloc = std.testing.allocator; | 667 | const alloc = std.testing.allocator; |
| 570 | var e: Egress = .{ .buf = try alloc.alloc(u8, 8) }; | 668 | var e: Egress = .{ .buf = try alloc.alloc(u8, 8) }; |
| @@ -824,8 +922,18 @@ pub fn pathFrom( | |||
| 824 | // --------------------------------------------------------------------------- | 922 | // --------------------------------------------------------------------------- |
| 825 | 923 | ||
| 826 | /// wolfSSL's PSK callback carries no user pointer, so the key has to be | 924 | /// wolfSSL's PSK callback carries no user pointer, so the key has to be |
| 827 | /// reachable without one. A client process runs one connection at a time, | 925 | /// reachable without one. Process-global, and a mux client process is NOT |
| 828 | /// which makes this correct rather than merely convenient. | 926 | /// one dial at a time: every wall host poller and every tile pump dials on |
| 927 | /// its own thread, each with the key its own handoff announced, so two | ||
| 928 | /// concurrent dials with two different keys race on this — `connect` writes | ||
| 929 | /// it, then builds the ClientHello inside its own `drain()`, and the other | ||
| 930 | /// thread may write between those two. The symptom is a handshake that | ||
| 931 | /// never completes: one `unreachable` poll the wall rides out, with no log | ||
| 932 | /// line. The fix is a per-connection key — wolfSSL_set_app_data(ssl, self) | ||
| 933 | /// in `startTls`, wolfSSL_get_app_data(ssl) in `pskClientCb`, and this | ||
| 934 | /// global deleted — tracked as issue 92d0a0e2, and left out of the comment-only | ||
| 935 | /// change that wrote this paragraph because it is a code path with its own | ||
| 936 | /// pin to earn. | ||
| 829 | var g_key: ?Key = null; | 937 | var g_key: ?Key = null; |
| 830 | 938 | ||
| 831 | fn pskClientCb( | 939 | fn pskClientCb( |
| @@ -1068,15 +1176,10 @@ pub const Client = struct { | |||
| 1068 | /// Milliseconds until ngtcp2 next wants servicing, capped. Feeds the | 1176 | /// Milliseconds until ngtcp2 next wants servicing, capped. Feeds the |
| 1069 | /// client's existing poll timeout — no timer fd, same as the daemon. | 1177 | /// client's existing poll timeout — no timer fd, same as the daemon. |
| 1070 | pub fn timeoutMs(self: *Client, cap_ms: i32) i32 { | 1178 | pub fn timeoutMs(self: *Client, cap_ms: i32) i32 { |
| 1071 | // A negative cap means poll forever; the @intCast below would panic | 1179 | // A negative cap means poll forever; expiryMs's @intCast would panic |
| 1072 | // on it instead of honouring it. | 1180 | // on it instead of honouring it. |
| 1073 | if (cap_ms < 0) return cap_ms; | 1181 | if (cap_ms < 0) return cap_ms; |
| 1074 | const conn = self.conn orelse return cap_ms; | 1182 | return expiryMs(self.conn, cap_ms, timestampNs()); |
| 1075 | const expiry = c.ngtcp2_conn_get_expiry(conn); | ||
| 1076 | if (expiry == std.math.maxInt(u64)) return cap_ms; | ||
| 1077 | const now = timestampNs(); | ||
| 1078 | if (expiry <= now) return 0; | ||
| 1079 | return @intCast(@min(@as(u64, @intCast(cap_ms)), (expiry - now) / 1_000_000)); | ||
| 1080 | } | 1183 | } |
| 1081 | 1184 | ||
| 1082 | /// One service pass: read, run due timers, push ready egress. Safe to call | 1185 | /// One service pass: read, run due timers, push ready egress. Safe to call |
| @@ -1118,13 +1221,7 @@ pub const Client = struct { | |||
| 1118 | if (n == 0) continue; | 1221 | if (n == 0) continue; |
| 1119 | const conn = self.conn orelse return; | 1222 | const conn = self.conn orelse return; |
| 1120 | var path = pathFrom(&self.local, self.local_len, &self.remote, self.remote_len); | 1223 | var path = pathFrom(&self.local, self.local_len, &self.remote, self.remote_len); |
| 1121 | var pi: c.ngtcp2_pkt_info = .{ .ecn = 0 }; | 1224 | if (readPkt(conn, &path, buf[0..n], &self.ngtcp2_depth) != 0) { |
| 1122 | const rv = blk: { | ||
| 1123 | self.ngtcp2_depth += 1; | ||
| 1124 | defer self.ngtcp2_depth -= 1; | ||
| 1125 | break :blk c.ngtcp2_conn_read_pkt(conn, &path, &pi, &buf, n, timestampNs()); | ||
| 1126 | }; | ||
| 1127 | if (rv != 0) { | ||
| 1128 | self.dead = true; | 1225 | self.dead = true; |
| 1129 | return; | 1226 | return; |
| 1130 | } | 1227 | } |
| @@ -1141,62 +1238,21 @@ pub const Client = struct { | |||
| 1141 | } | 1238 | } |
| 1142 | 1239 | ||
| 1143 | fn drain(self: *Client) void { | 1240 | fn drain(self: *Client) void { |
| 1144 | // The listener learned this the hard way: writing to a connection | 1241 | // The write half of the invariant `readPkt` counts. A future caller |
| 1145 | // while ngtcp2 is reading a packet on it corrupts loss detection and | 1242 | // that drains from inside an ngtcp2 callback fails here |
| 1146 | // can abort outright. Enforced here so it stays true of this side. | 1243 | // deterministically in Debug rather than corrupting loss detection. |
| 1147 | std.debug.assert(self.ngtcp2_depth == 0); | 1244 | std.debug.assert(self.ngtcp2_depth == 0); |
| 1148 | const conn = self.conn orelse return; | 1245 | const conn = self.conn orelse return; |
| 1149 | var buf: [max_udp]u8 = undefined; | 1246 | const sendPkt = struct { |
| 1150 | var stream_blocked = false; | 1247 | fn f(cl: *Client, pkt: []const u8) bool { |
| 1151 | while (true) { | 1248 | _ = std.posix.send(cl.fd, pkt, 0) catch |err| { |
| 1152 | var ps: c.ngtcp2_path_storage = undefined; | 1249 | cl.sendRecvFailed(err); |
| 1153 | c.ngtcp2_path_storage_zero(&ps); | 1250 | return false; |
| 1154 | var pi: c.ngtcp2_pkt_info = .{ .ecn = 0 }; | 1251 | }; |
| 1155 | var wrote: c.ngtcp2_ssize = 0; | 1252 | return true; |
| 1156 | |||
| 1157 | var vecs: [2]c.ngtcp2_vec = undefined; | ||
| 1158 | var vcnt: usize = 0; | ||
| 1159 | var sid: i64 = -1; | ||
| 1160 | if (!stream_blocked and self.stream_id != -1) { | ||
| 1161 | vcnt = self.out.vecs(&vecs); | ||
| 1162 | if (vcnt > 0) sid = self.stream_id; | ||
| 1163 | } | ||
| 1164 | |||
| 1165 | const n = c.ngtcp2_conn_writev_stream_versioned( | ||
| 1166 | conn, | ||
| 1167 | &ps.path, | ||
| 1168 | c.NGTCP2_PKT_INFO_VERSION, | ||
| 1169 | &pi, | ||
| 1170 | &buf, | ||
| 1171 | buf.len, | ||
| 1172 | &wrote, | ||
| 1173 | 0, | ||
| 1174 | sid, | ||
| 1175 | if (vcnt > 0) &vecs else null, | ||
| 1176 | vcnt, | ||
| 1177 | timestampNs(), | ||
| 1178 | ); | ||
| 1179 | if (n == c.NGTCP2_ERR_STREAM_DATA_BLOCKED or n == c.NGTCP2_ERR_STREAM_SHUT_WR) { | ||
| 1180 | // Documented, not fatal. Retry the same iteration carrying | ||
| 1181 | // no stream data so ACKs and keepalives still leave: they | ||
| 1182 | // are how the peer's window reopens, and a client that | ||
| 1183 | // stopped here would stall exactly as the daemon did. | ||
| 1184 | stream_blocked = true; | ||
| 1185 | continue; | ||
| 1186 | } | ||
| 1187 | switch (accountWrite(&self.out, wrote, n)) { | ||
| 1188 | .stop => { | ||
| 1189 | self.dead = true; | ||
| 1190 | return; | ||
| 1191 | }, | ||
| 1192 | .brk => return, | ||
| 1193 | .cont => {}, | ||
| 1194 | } | 1253 | } |
| 1195 | _ = std.posix.send(self.fd, buf[0..@intCast(n)], 0) catch |err| { | 1254 | }.f; |
| 1196 | self.sendRecvFailed(err); | 1255 | if (drainConn(conn, &self.out, self.stream_id, self, sendPkt) == .failed) self.dead = true; |
| 1197 | return; | ||
| 1198 | }; | ||
| 1199 | } | ||
| 1200 | } | 1256 | } |
| 1201 | 1257 | ||
| 1202 | /// A short return is the caller's signal to keep the rest and offer it | 1258 | /// A short return is the caller's signal to keep the rest and offer it |
src/server/quic_server.zig
| Old | New | ||
|---|---|---|---|
| @@ -777,14 +777,7 @@ pub const Listener = struct { | |||
| 777 | const now = quic.timestampNs(); | 777 | const now = quic.timestampNs(); |
| 778 | for (self.conns) |slot| { | 778 | for (self.conns) |slot| { |
| 779 | const cn = slot orelse continue; | 779 | const cn = slot orelse continue; |
| 780 | const conn = cn.conn orelse continue; | 780 | best = @min(best, quic.expiryMs(cn.conn, cap_ms, now)); |
| 781 | const expiry = c.ngtcp2_conn_get_expiry(conn); | ||
| 782 | if (expiry == std.math.maxInt(u64)) continue; | ||
| 783 | const ms: i32 = if (expiry <= now) | ||
| 784 | 0 | ||
| 785 | else | ||
| 786 | @intCast(@min(@as(u64, @intCast(cap_ms)), (expiry - now) / 1_000_000)); | ||
| 787 | best = @min(best, ms); | ||
| 788 | } | 781 | } |
| 789 | return best; | 782 | return best; |
| 790 | } | 783 | } |
| @@ -863,12 +856,7 @@ pub const Listener = struct { | |||
| 863 | const cn = slot.* orelse return; | 856 | const cn = slot.* orelse return; |
| 864 | const conn = cn.conn orelse return; | 857 | const conn = cn.conn orelse return; |
| 865 | var path = quic.pathFrom(&cn.local_storage, cn.local_len, from, from_len); | 858 | var path = quic.pathFrom(&cn.local_storage, cn.local_len, from, from_len); |
| 866 | var pi: c.ngtcp2_pkt_info = .{ .ecn = 0 }; | 859 | const rv = quic.readPkt(conn, &path, pkt, &self.ngtcp2_depth); |
| 867 | const rv = blk: { | ||
| 868 | self.ngtcp2_depth += 1; | ||
| 869 | defer self.ngtcp2_depth -= 1; | ||
| 870 | break :blk c.ngtcp2_conn_read_pkt(conn, &path, &pi, pkt.ptr, pkt.len, quic.timestampNs()); | ||
| 871 | }; | ||
| 872 | // Reaped before anything else touches the table: a connection the | 860 | // Reaped before anything else touches the table: a connection the |
| 873 | // handler closed mid-callback is gone from here on. | 861 | // handler closed mid-callback is gone from here on. |
| 874 | self.reapClosing(); | 862 | self.reapClosing(); |
| @@ -897,66 +885,25 @@ pub const Listener = struct { | |||
| 897 | // one run in a few hundred somewhere else entirely. | 885 | // one run in a few hundred somewhere else entirely. |
| 898 | std.debug.assert(self.ngtcp2_depth == 0); | 886 | std.debug.assert(self.ngtcp2_depth == 0); |
| 899 | const conn = cn.conn orelse return; | 887 | const conn = cn.conn orelse return; |
| 900 | var buf: [quic.max_udp]u8 = undefined; | 888 | // No refusal check, unlike quic.Client's: this socket is |
| 901 | // Set when the peer's stream window is full. Everything that is not | 889 | // UNCONNECTED, so the kernel has no peer to attribute an ICMP |
| 902 | // stream data still has to leave. | 890 | // unreachable to and never delivers one. The asymmetry with the |
| 903 | var stream_blocked = false; | 891 | // client is the sockets', not drift — which is why `sendPkt` is the |
| 904 | while (true) { | 892 | // seam the shared loop is cut at. |
| 905 | var ps: c.ngtcp2_path_storage = undefined; | 893 | const To = struct { l: *Listener, cn: *Conn }; |
| 906 | c.ngtcp2_path_storage_zero(&ps); | 894 | const sendPkt = struct { |
| 907 | var pi: c.ngtcp2_pkt_info = .{ .ecn = 0 }; | 895 | fn f(ctx: To, pkt: []const u8) bool { |
| 908 | var wrote: c.ngtcp2_ssize = 0; | 896 | _ = std.posix.sendto( |
| 909 | 897 | ctx.l.fd, | |
| 910 | var vecs: [2]c.ngtcp2_vec = undefined; | 898 | pkt, |
| 911 | var vcnt: usize = 0; | 899 | 0, |
| 912 | var sid: i64 = -1; | 900 | @ptrCast(&ctx.cn.remote), |
| 913 | if (!stream_blocked and cn.stream_id != -1) { | 901 | ctx.cn.remote_len, |
| 914 | vcnt = cn.out.vecs(&vecs); | 902 | ) catch return false; |
| 915 | if (vcnt > 0) sid = cn.stream_id; | 903 | return true; |
| 916 | } | ||
| 917 | |||
| 918 | const n = c.ngtcp2_conn_writev_stream_versioned( | ||
| 919 | conn, | ||
| 920 | &ps.path, | ||
| 921 | c.NGTCP2_PKT_INFO_VERSION, | ||
| 922 | &pi, | ||
| 923 | &buf, | ||
| 924 | buf.len, | ||
| 925 | &wrote, | ||
| 926 | 0, | ||
| 927 | sid, | ||
| 928 | if (vcnt > 0) &vecs else null, | ||
| 929 | vcnt, | ||
| 930 | quic.timestampNs(), | ||
| 931 | ); | ||
| 932 | if (n == c.NGTCP2_ERR_STREAM_DATA_BLOCKED or n == c.NGTCP2_ERR_STREAM_SHUT_WR) { | ||
| 933 | // A documented return, not a failure: the peer has no window | ||
| 934 | // left for this stream. Treating it as fatal abandoned the | ||
| 935 | // whole egress loop — including the ACKs and the flow-control | ||
| 936 | // updates that are how the peer's window reopens, which turns | ||
| 937 | // a moment of backpressure into a stall that never resolves. | ||
| 938 | // Retry the same iteration carrying no stream data. | ||
| 939 | stream_blocked = true; | ||
| 940 | continue; | ||
| 941 | } | ||
| 942 | switch (quic.accountWrite(&cn.out, wrote, n)) { | ||
| 943 | .stop => return, | ||
| 944 | .brk => break, | ||
| 945 | .cont => {}, | ||
| 946 | } | 904 | } |
| 947 | 905 | }.f; | |
| 948 | // No refusal check here, unlike quic.Client's send: this socket | 906 | _ = quic.drainConn(conn, &cn.out, cn.stream_id, To{ .l = self, .cn = cn }, sendPkt); |
| 949 | // is UNCONNECTED, so the kernel has no peer to attribute an ICMP | ||
| 950 | // unreachable to and never delivers one. The asymmetry with the | ||
| 951 | // client is the sockets', not drift. | ||
| 952 | _ = std.posix.sendto( | ||
| 953 | self.fd, | ||
| 954 | buf[0..@intCast(n)], | ||
| 955 | 0, | ||
| 956 | @ptrCast(&cn.remote), | ||
| 957 | cn.remote_len, | ||
| 958 | ) catch break; | ||
| 959 | } | ||
| 960 | } | 907 | } |
| 961 | 908 | ||
| 962 | fn kill(self: *Listener, slot: *?*Conn) void { | 909 | fn kill(self: *Listener, slot: *?*Conn) void { |