lib/quic/src/connection/stream/receive.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const quic = @import("../../root.zig");
3
4 const stream = quic.connection.stream;
5
6 /// One stretch of arrived bytes for the caller sizing the range table, held as the offset it starts
7 /// at and the offset it ends at. Every stretch sits at or above the read offset, because a read
8 /// trims away what it takes.
9 pub const Range = struct {
10 start: u62,
11 end: u62,
12
13 pub fn empty() Range {
14 return .{ .start = 0, .end = 0 };
15 }
16 };
17
18 /// Result of one read, recording how many bytes the call copied and whether it carried the read
19 /// offset to the final size. `read` returns this result so an application learns from the FIN flag
20 /// that the stream ended.
21 pub const ReadResult = struct {
22 bytes: usize,
23 fin: bool,
24 };
25
26 pub const ReceiveError = error{ FlowControl, FinalSize, RangesFull };
27
28 /// The half of one stream that takes data in, held for stream 0 by the connection and driven from
29 /// arriving frames, working out of a ring and a range table the caller supplies. This half holds
30 /// that ring and table, the stream's window, the state, the read offset, the final size, and the
31 /// codes a reset and a stop carried. `init` wants both the ring and the table nonempty, and a
32 /// window no wider than the ring.
33 pub const Receive = struct {
34 bytes: []u8,
35 ranges: []Range,
36 window: stream.Window,
37 state: stream.ReceiveState = .recv,
38 count: u16 = 0,
39 head: usize = 0,
40 read_offset: u62 = 0,
41 final_size: ?u62 = null,
42 reset_code: ?u62 = null,
43 stop_code: ?u62 = null,
44 stop_pending: bool = false,
45
46 pub fn init(bytes: []u8, ranges: []Range, window: u62) Receive {
47 std.debug.assert(bytes.len > 0);
48 std.debug.assert(ranges.len > 0);
49 std.debug.assert(ranges.len <= std.math.maxInt(u16));
50 std.debug.assert(window <= bytes.len);
51 for (ranges) |*range| range.* = Range.empty();
52 return .{ .bytes = bytes, .ranges = ranges, .window = stream.Window.init(window) };
53 }
54
55 /// Takes in one arriving STREAM frame's data for the stream, putting up with bytes that repeat
56 /// or overlap what has arrived already. The bytes new to the half are charged against this
57 /// stream's window and against the connection window handed in. Data past either window answers
58 /// `FlowControl`, and data that moves a settled final size or runs beyond one answers
59 /// `FinalSize`. Data ending past the highest offset a stream can reach answers `FlowControl` as
60 /// well. A packet that would want one stretch more than the table holds answers `RangesFull`.
61 /// Once the half has everything, or has been reset, later data is charged against both windows
62 /// and then thrown away. A frame carrying FIN settles the final size and moves the half to Size
63 /// Known.
64 pub fn receive(
65 self: *Receive,
66 connection: *stream.Window,
67 offset: u62,
68 data: []const u8,
69 fin: bool,
70 ) ReceiveError!void {
71 if (data.len > std.math.maxInt(u62) - offset) return error.FlowControl;
72 const end = offset + @as(u62, @intCast(data.len));
73 try self.checkFinal(end, fin);
74 const growth = try self.admit(connection, end);
75 if (self.discarding()) return;
76 const start = @max(offset, self.read_offset);
77 if (start < end) {
78 try self.insert(start, end);
79 const skipped: usize = @intCast(start - offset);
80 stream.ring.write(self.bytes, self.ringIndex(start), data[skipped..]);
81 }
82 self.window.record(end);
83 connection.record(connection.received + growth);
84 if (fin and self.final_size == null) {
85 self.final_size = end;
86 self.state = stream.receiveTransition(self.state, .receive_fin) catch unreachable;
87 }
88 self.observeComplete();
89 }
90
91 /// Counts the bytes waiting in order at the read offset, so the caller learns how much one read
92 /// can copy. The call answers zero after a reset, and zero while a hole sits at the read
93 /// offset.
94 pub fn readable(self: *const Receive) usize {
95 if (self.reset_code != null) return 0;
96 if (self.count == 0) return 0;
97 const first = self.ranges[0];
98 if (first.start != self.read_offset) return 0;
99 return @intCast(first.end - first.start);
100 }
101
102 /// Copies the waiting bytes in offset order for an application taking its bytes, and says
103 /// whether the read reached the final size. What the call copies is given up to this stream's
104 /// window and to the connection window, releasing room back to the peer so the peer earns more
105 /// room. A read that passes the final size moves the half to Data Read. On a stream the peer
106 /// reset, the call copies nothing and moves the half to Reset Read.
107 pub fn read(self: *Receive, connection: *stream.Window, output: []u8) ReadResult {
108 switch (self.state) {
109 .reset_recvd => {
110 self.state = stream.receiveTransition(.reset_recvd, .read_reset) catch unreachable;
111 return .{ .bytes = 0, .fin = false };
112 },
113 .reset_read => return .{ .bytes = 0, .fin = false },
114 .recv, .size_known, .data_recvd, .data_read => {},
115 }
116 const count = @min(output.len, self.readable());
117 stream.ring.read(self.bytes, self.head, output[0..count]);
118 self.consume(count);
119 connection.release(connection.consumed + @as(u62, @intCast(count)));
120 const fin = if (self.final_size) |final| self.read_offset == final else false;
121 if (fin and self.state == .data_recvd) {
122 self.state = stream.receiveTransition(.data_recvd, .read_all) catch unreachable;
123 }
124 return .{ .bytes = count, .fin = fin };
125 }
126
127 /// Applies one arriving RESET_STREAM frame for the stream once the final size and both windows
128 /// check out. A final size at odds with one already settled, or under what has arrived already,
129 /// answers `FinalSize`, and a final size past either window answers `FlowControl`. Every byte
130 /// up to the final size is given up to the connection window, the ones the application will
131 /// never see among them. The call throws away the arrived stretches, keeps the error code,
132 /// drops a STOP_SENDING that was waiting, and moves the half to Reset Recvd.
133 pub fn resetReceived(
134 self: *Receive,
135 connection: *stream.Window,
136 code: u62,
137 final: u62,
138 ) error{ FlowControl, FinalSize }!void {
139 if (self.final_size) |known| {
140 if (final != known) return error.FinalSize;
141 }
142 if (final < self.window.received) return error.FinalSize;
143 const growth = try self.admit(connection, final);
144 if (self.discarding()) return;
145 self.window.record(final);
146 connection.record(connection.received + growth);
147 connection.release(connection.consumed + (final - self.read_offset));
148 self.final_size = final;
149 self.reset_code = code;
150 self.stop_pending = false;
151 self.removeRanges(0, self.count);
152 self.state = stream.receiveTransition(self.state, .receive_reset) catch unreachable;
153 }
154
155 /// Puts a STOP_SENDING frame in the queue while the peer might still send, so an application
156 /// tells the peer to stop sending. The call answers false on a second call, and false once the
157 /// half holds everything or has been reset, so the returned flag says whether the request was
158 /// taken.
159 pub fn stop(self: *Receive, code: u62) bool {
160 if (self.stop_code != null) return false;
161 if (self.discarding()) return false;
162 self.stop_code = code;
163 self.stop_pending = true;
164 return true;
165 }
166
167 /// Puts STOP_SENDING back in the queue for the loss path after its packet was lost, so the
168 /// frame goes out again. The call queues nothing once the half holds everything or has been
169 /// reset, because the peer has stopped by then. A stop code is in hand when this function is
170 /// called.
171 pub fn loseStop(self: *Receive) void {
172 std.debug.assert(self.stop_code != null);
173 if (self.discarding()) return;
174 self.stop_pending = true;
175 }
176
177 /// Offers a larger MAX_STREAM_DATA figure while more data may yet arrive, so the send path can
178 /// decide whether a MAX_STREAM_DATA frame belongs in the packet it is building. Once the half
179 /// has left Recv the final size is settled, so more room would buy nothing and the call offers
180 /// none.
181 pub fn windowUpdate(self: *const Receive) ?u62 {
182 if (self.state != .recv) return null;
183 return self.window.update();
184 }
185
186 fn checkFinal(self: *const Receive, end: u62, fin: bool) error{FinalSize}!void {
187 if (self.final_size) |final| {
188 if (end > final) return error.FinalSize;
189 if (fin and end != final) return error.FinalSize;
190 return;
191 }
192 if (fin and end < self.window.received) return error.FinalSize;
193 }
194
195 fn admit(
196 self: *const Receive,
197 connection: *const stream.Window,
198 end: u62,
199 ) error{FlowControl}!u62 {
200 if (!self.window.allows(end)) return error.FlowControl;
201 const growth = end -| self.window.received;
202 if (!connection.allows(@as(u64, connection.received) + growth)) return error.FlowControl;
203 return growth;
204 }
205
206 fn discarding(self: *const Receive) bool {
207 return switch (self.state) {
208 .recv, .size_known => false,
209 .data_recvd, .data_read, .reset_recvd, .reset_read => true,
210 };
211 }
212
213 fn insert(self: *Receive, start: u62, end: u62) error{RangesFull}!void {
214 std.debug.assert(start < end);
215 std.debug.assert(start >= self.read_offset);
216 var first: usize = self.count;
217 for (0..self.count) |index| {
218 if (self.ranges[index].end >= start) {
219 first = index;
220 break;
221 }
222 }
223 var last = first;
224 for (first..self.count) |index| {
225 if (self.ranges[index].start > end) break;
226 last = index + 1;
227 }
228 if (first == last) return self.insertAt(first, .{ .start = start, .end = end });
229 self.ranges[first] = .{
230 .start = @min(start, self.ranges[first].start),
231 .end = @max(end, self.ranges[last - 1].end),
232 };
233 self.removeRanges(first + 1, last - first - 1);
234 }
235
236 fn insertAt(self: *Receive, at: usize, range: Range) error{RangesFull}!void {
237 std.debug.assert(at <= self.count);
238 if (self.count == self.ranges.len) return error.RangesFull;
239 for (0..self.count - at) |step| {
240 const index = self.count - step;
241 self.ranges[index] = self.ranges[index - 1];
242 }
243 self.ranges[at] = range;
244 self.count += 1;
245 }
246
247 fn removeRanges(self: *Receive, at: usize, removed: usize) void {
248 std.debug.assert(at + removed <= self.count);
249 const remaining = self.count - removed;
250 for (at..remaining) |index| self.ranges[index] = self.ranges[index + removed];
251 for (remaining..self.count) |index| self.ranges[index] = Range.empty();
252 self.count = @intCast(remaining);
253 }
254
255 fn consume(self: *Receive, count: usize) void {
256 if (count == 0) return;
257 std.debug.assert(count <= self.readable());
258 const advance: u62 = @intCast(count);
259 self.read_offset += advance;
260 self.head = (self.head + count) % self.bytes.len;
261 self.ranges[0].start += advance;
262 if (self.ranges[0].start == self.ranges[0].end) self.removeRanges(0, 1);
263 self.window.release(self.read_offset);
264 }
265
266 fn observeComplete(self: *Receive) void {
267 if (self.state != .size_known) return;
268 const final = self.final_size.?;
269 const contiguous = self.read_offset + @as(u62, @intCast(self.readable()));
270 if (contiguous != final) return;
271 self.stop_pending = false;
272 self.state = stream.receiveTransition(.size_known, .all_received) catch unreachable;
273 }
274
275 fn ringIndex(self: *const Receive, offset: u62) usize {
276 std.debug.assert(offset >= self.read_offset);
277 const distance: usize = @intCast(offset - self.read_offset);
278 std.debug.assert(distance < self.bytes.len);
279 return (self.head + distance) % self.bytes.len;
280 }
281 };
282
283 test "RFC 9000 section 2.2 reassembly orders gaps, overlaps, duplicates, and reordering" {
284 var bytes: [16]u8 = undefined;
285 var ranges: [4]Range = undefined;
286 var connection = stream.Window.init(64);
287 var receive = Receive.init(&bytes, &ranges, 16);
288 try receive.receive(&connection, 6, "ghij", false);
289 try receive.receive(&connection, 2, "cd", false);
290 try std.testing.expectEqual(@as(usize, 0), receive.readable());
291 try std.testing.expectEqual(@as(u16, 2), receive.count);
292 try receive.receive(&connection, 3, "defg", false);
293 try std.testing.expectEqual(@as(u16, 1), receive.count);
294 try receive.receive(&connection, 6, "ghij", false);
295 try receive.receive(&connection, 0, "abc", false);
296 try std.testing.expectEqual(@as(u62, 10), connection.received);
297 var output: [16]u8 = undefined;
298 const result = receive.read(&connection, &output);
299 try std.testing.expectEqualStrings("abcdefghij", output[0..result.bytes]);
300 try std.testing.expect(!result.fin);
301 try std.testing.expectEqual(@as(u62, 10), connection.consumed);
302 receive.window.advertise(receive.windowUpdate() orelse return error.NoUpdate);
303 try receive.receive(&connection, 8, "ijklmnopqrst", false);
304 const tail = receive.read(&connection, &output);
305 try std.testing.expectEqualStrings("klmnopqrst", output[0..tail.bytes]);
306 }
307
308 test "RFC 9000 section 2.2 received range records maximum and maximum plus one" {
309 var bytes: [16]u8 = undefined;
310 var ranges: [3]Range = undefined;
311 var connection = stream.Window.init(64);
312 var receive = Receive.init(&bytes, &ranges, 16);
313 try receive.receive(&connection, 1, "b", false);
314 try receive.receive(&connection, 3, "d", false);
315 try receive.receive(&connection, 5, "f", false);
316 try std.testing.expectEqual(@as(u16, 3), receive.count);
317 try std.testing.expectError(error.RangesFull, receive.receive(&connection, 7, "h", false));
318 try std.testing.expectEqual(@as(u62, 6), receive.window.received);
319 try receive.receive(&connection, 2, "c", false);
320 try std.testing.expectEqual(@as(u16, 2), receive.count);
321 try receive.receive(&connection, 7, "h", false);
322 try std.testing.expectEqual(@as(u16, 3), receive.count);
323 }
324
325 test "RFC 9000 section 4.1 stream and connection windows accept the limit, not one byte past" {
326 var bytes: [16]u8 = undefined;
327 var ranges: [2]Range = undefined;
328 const data: [9]u8 = @splat('x');
329 var connection = stream.Window.init(64);
330 var receive = Receive.init(&bytes, &ranges, 8);
331 try receive.receive(&connection, 0, data[0..8], false);
332 try std.testing.expectError(error.FlowControl, receive.receive(&connection, 8, "x", false));
333 var narrow = stream.Window.init(8);
334 var other = Receive.init(&bytes, &ranges, 16);
335 try std.testing.expectError(error.FlowControl, other.receive(&narrow, 0, &data, false));
336 try other.receive(&narrow, 0, data[0..8], false);
337 try std.testing.expectEqual(@as(u62, 8), narrow.received);
338 }
339
340 test "RFC 9000 section 4.5 final size changes close with FINAL_SIZE_ERROR" {
341 var bytes: [16]u8 = undefined;
342 var ranges: [2]Range = undefined;
343 var connection = stream.Window.init(64);
344 var receive = Receive.init(&bytes, &ranges, 16);
345 try receive.receive(&connection, 0, "abcd", true);
346 try std.testing.expectError(error.FinalSize, receive.receive(&connection, 0, "abcde", false));
347 try std.testing.expectError(error.FinalSize, receive.receive(&connection, 0, "abc", true));
348 try std.testing.expectError(error.FinalSize, receive.resetReceived(&connection, 1, 5));
349 try receive.receive(&connection, 2, "cd", true);
350 var other_connection = stream.Window.init(64);
351 var unknown = Receive.init(&bytes, &ranges, 16);
352 try unknown.receive(&other_connection, 0, "abcd", false);
353 try std.testing.expectError(error.FinalSize, unknown.receive(&other_connection, 0, "ab", true));
354 try std.testing.expectError(error.FinalSize, unknown.resetReceived(&other_connection, 1, 3));
355 }
356
357 test "RFC 9000 section 3.2 RESET_STREAM discards unread data and reports the code" {
358 var bytes: [16]u8 = undefined;
359 var ranges: [2]Range = undefined;
360 var connection = stream.Window.init(64);
361 var receive = Receive.init(&bytes, &ranges, 16);
362 try std.testing.expect(receive.stop(7));
363 try std.testing.expect(!receive.stop(8));
364 try receive.receive(&connection, 0, "abcd", false);
365 try receive.resetReceived(&connection, 0x2a, 6);
366 try std.testing.expectEqual(stream.ReceiveState.reset_recvd, receive.state);
367 try std.testing.expectEqual(@as(?u62, 0x2a), receive.reset_code);
368 try std.testing.expect(!receive.stop_pending);
369 try std.testing.expectEqual(@as(u62, 6), connection.consumed);
370 try std.testing.expectEqual(@as(usize, 0), receive.readable());
371 var output: [8]u8 = undefined;
372 try std.testing.expectEqual(@as(usize, 0), receive.read(&connection, &output).bytes);
373 try std.testing.expectEqual(stream.ReceiveState.reset_read, receive.state);
374 try receive.receive(&connection, 4, "ef", false);
375 try std.testing.expectError(error.FinalSize, receive.receive(&connection, 6, "g", false));
376 }
377
378 test "RFC 9000 section 3.2 reading through the FIN reaches Data Read" {
379 var bytes: [16]u8 = undefined;
380 var ranges: [2]Range = undefined;
381 var connection = stream.Window.init(64);
382 var receive = Receive.init(&bytes, &ranges, 16);
383 try receive.receive(&connection, 1, "b", true);
384 try std.testing.expectEqual(stream.ReceiveState.size_known, receive.state);
385 try receive.receive(&connection, 0, "a", false);
386 try std.testing.expectEqual(stream.ReceiveState.data_recvd, receive.state);
387 var output: [1]u8 = undefined;
388 try std.testing.expect(!receive.read(&connection, &output).fin);
389 const last = receive.read(&connection, &output);
390 try std.testing.expect(last.fin);
391 try std.testing.expectEqual(@as(u8, 'b'), output[0]);
392 try std.testing.expectEqual(stream.ReceiveState.data_read, receive.state);
393 try std.testing.expect(!receive.stop(1));
394 }
395
396 test "RFC 9000 section 13.3 a lost STOP_SENDING goes out again until Data Recvd or Reset Recvd" {
397 var bytes: [16]u8 = undefined;
398 var ranges: [2]Range = undefined;
399 var connection = stream.Window.init(64);
400 var receive = Receive.init(&bytes, &ranges, 16);
401 try std.testing.expect(receive.stop(7));
402 receive.stop_pending = false;
403 try receive.receive(&connection, 1, "b", true);
404 try std.testing.expectEqual(stream.ReceiveState.size_known, receive.state);
405 receive.loseStop();
406 try std.testing.expect(receive.stop_pending);
407 try receive.receive(&connection, 0, "a", false);
408 try std.testing.expectEqual(stream.ReceiveState.data_recvd, receive.state);
409 try std.testing.expect(!receive.stop_pending);
410 receive.loseStop();
411 try std.testing.expect(!receive.stop_pending);
412 var reset = Receive.init(&bytes, &ranges, 16);
413 try std.testing.expect(reset.stop(8));
414 reset.stop_pending = false;
415 try reset.resetReceived(&connection, 8, 0);
416 reset.loseStop();
417 try std.testing.expect(!reset.stop_pending);
418 }