ReviewOS
Reload

also looking at this

home-lang/home

feat(video): add bounded packet history, delayed FIFO and owned snapshots

#916
Merged chris wants to merge feat/keyframe-packet-ring into main
3 files +229 -0
packages/video/docs/packet-ring.mdadded+36-0
@@ -0,0 +1,36 @@
1# Keyframe-aligned packet history
2
3`core/packet_ring.zig` (exported as `video.packet_ring`) adapts the existing video
4ring-buffer structure to encoded pre-roll: storage consists of one fixed byte
5arena and a fixed entry array. Push copies borrowed packets without allocating
6or blocking. The caller serializes all access and chooses timestamp units.
7
8Byte capacity, frame capacity and duration are hard upper bounds. Eviction
9removes dependent packets until the oldest retained packet is a keyframe.
10If the limits cannot hold a GOP, history can be empty until the next key.
11Oversized packets and non-increasing timestamps are rejected without modifying
12retained history. Parameter sets/codec configuration belong outside the ring.
13`entry` returns metadata; `copy` explicitly copies a retained payload to caller
14storage, including when it wraps in the byte arena. Views expire on mutation.
15
16Three standalone regression tests cover arena wrap, byte/frame/time eviction,
17keyframe gating, rejected-input atomicity, zero capacities and allocator failure
18cleanup. Tested on Zig 0.15.2 and the installed 0.17 development toolchain. This
19focused validation does not establish correctness of the wider video graph.
20
21zig-camera uses this module with Home JPEG/Matroska and x264/x265/fragmented MP4.
22It retains compressed access units, without retaining camera DMA leases or raw
23RGB frames. Explicit snapshots need scratch space for one largest retained
24packet; remuxing starts with fresh codec headers and resets file timestamps.
25
26For time-shifted streaming, `pushQueued` uses the same owned arena as a bounded
27FIFO: it never evicts or filters dependent frames, and rejects full/oversized or
28non-monotonic input without mutation. Use `ready(now, delay)` then `copy(0, scratch)`
29and `pop()` to forward matured packets; `ready(null, delay)` flushes EOF. Queue
30capacity must cover the delay. The caller controls keyframe starts and output
31errors; a full FIFO must not silently discard encoded dependencies.
32
33`cloneFrom(allocator, start)` owns a minimally sized copy of a selected suffix.
34Serialize it with source mutation; the resulting data can move to a worker and
35outlive the source. It copies wrapped bytes directly into a contiguous arena and
36cleans up partial allocation failures. The module creates no threads or file I/O.
packages/video/src/core/packet_ring.zigadded+191-0
@@ -0,0 +1,191 @@
1//! Owned encoded-packet history. One caller serializes all access. Fixed byte
2//! arena and entry array; push never allocates, blocks, or keeps borrowed input.
3//! Eviction drops dependent packets until the oldest retained packet is a key.
4const std = ;
5pub const Entry = struct { offset: usize, length: usize, timestamp: u64, key: bool };
6pub const Ring = struct {
7 allocator: std.mem.Allocator,
8 bytes: []u8,
9 entries: []Entry,
10 duration: u64,
11 head: usize = 0,
12 count: usize = 0,
13 used: usize = 0,
14 write_pos: usize = 0,
15 last: ?u64 = null,
16 evicted: u64 = 0,
17 skipped: u64 = 0,
18 pub fn init(allocator: std.mem.Allocator, byte_capacity: usize, frame_capacity: usize, duration: u64) !Ring {
19 if (byte_capacity == 0 or frame_capacity == 0 or frame_capacity > std.math.maxInt(usize) / 2 or duration == 0) return error.InvalidConfiguration;
20 const bytes = try allocator.alloc(u8, byte_capacity);
21 errdefer allocator.free(bytes);
22 const entries = try allocator.alloc(Entry, frame_capacity);
23 return .{ .allocator = allocator, .bytes = bytes, .entries = entries, .duration = duration };
24 }
25 pub fn deinit(s: *Ring) void {
26 s.allocator.free(s.bytes);
27 s.allocator.free(s.entries);
28 }
29 fn drop(s: *Ring) void {
30 s.used -= s.entries[s.head].length;
31 s.head = (s.head + 1) % s.entries.len;
32 s.count -= 1;
33 s.evicted += 1;
34 }
35 pub fn entry(s: *const Ring, index: usize) ?Entry {
36 if (index >= s.count) return null;
37 return s.entries[(s.head + index) % s.entries.len];
38 }
39 /// False means waiting for a key after eviction. Limits are upper bounds;
40 /// history can be empty when a GOP is larger than the retention limits.
41 /// Oversized/invalid input is rejected before modifying retained history.
42 pub fn push(s: *Ring, packet: []const u8, timestamp: u64, key: bool) !bool {
43 if (packet.len == 0 or packet.len > s.bytes.len) return error.PacketTooLarge;
44 if (s.last) |last| if (timestamp <= last) return error.InvalidTimestamp;
45 while (s.count > 0 and (s.count == s.entries.len or packet.len > s.bytes.len - s.used or timestamp - s.entry(0).?.timestamp > s.duration)) s.drop();
46 while (s.count > 0 and !s.entry(0).?.key) s.drop();
47 s.last = timestamp;
48 if (s.count == 0 and !key) {
49 s.skipped += 1;
50 return false;
51 }
52 s.store(packet, timestamp, key);
53 return true;
54 }
55 fn store(s: *Ring, packet: []const u8, timestamp: u64, key: bool) void {
56 s.last = timestamp;
57 const first = @min(packet.len, s.bytes.len - s.write_pos);
58 @memcpy(s.bytes[s.write_pos..][0..first], packet[0..first]);
59 @memcpy(s.bytes[0 .. packet.len - first], packet[first..]);
60 s.entries[(s.head + s.count) % s.entries.len] = .{ .offset = s.write_pos, .length = packet.len, .timestamp = timestamp, .key = key };
61 s.write_pos = if (packet.len - first != 0) packet.len - first else if (s.write_pos + first == s.bytes.len) 0 else s.write_pos + first;
62 s.count += 1;
63 s.used += packet.len;
64 }
65 /// FIFO mode: no implicit eviction or GOP filtering. A consumer must pop
66 /// mature entries before enqueueing; overflow leaves all state unchanged.
67 pub fn pushQueued(s: *Ring, packet: []const u8, timestamp: u64, key: bool) !void {
68 if (packet.len == 0 or packet.len > s.bytes.len) return error.PacketTooLarge;
69 if (s.last) |last| if (timestamp <= last) return error.InvalidTimestamp;
70 if (s.count == s.entries.len or packet.len > s.bytes.len - s.used) return error.BufferFull;
71 s.store(packet, timestamp, key);
72 }
73 pub fn pop(s: *Ring) void {
74 if (s.count != 0) s.drop();
75 }
76 /// Null clock flushes EOF; subtraction avoids overflowing timestamp+delay.
77 pub fn ready(s: *const Ring, now: ?u64, delay: u64) bool {
78 const e = s.entry(0) orelse return false;
79 const clock = now orelse return true;
80 return clock >= e.timestamp and clock - e.timestamp >= delay;
81 }
82 /// Own only the selected suffix. Borrowed source data never escapes.
83 pub fn cloneFrom(s: *const Ring, a: std.mem.Allocator, start: usize) !Ring {
84 if (start >= s.count) return error.InvalidIndex;
85 var bytes: usize = 0;
86 for (start..s.count) |i| bytes = try std.math.add(usize, bytes, s.entry(i).?.length);
87 var result = try Ring.init(a, bytes, s.count - start, s.duration);
88 errdefer result.deinit();
89 // Destination is contiguous in this clone; copy straight into its arena.
90 for (start..s.count) |i| {
91 const e = s.entry(i).?;
92 _ = try s.copy(i, result.bytes[result.used..]);
93 result.entries[result.count] = .{ .offset = result.used, .length = e.length, .timestamp = e.timestamp, .key = e.key };
94 result.count += 1;
95 result.used += e.length;
96 result.last = e.timestamp;
97 }
98 result.write_pos = 0;
99 return result;
100 }
101 /// Explicit owned snapshot copy; data and metadata views otherwise expire
102 /// on the next mutation. Returns the active destination slice.
103 pub fn copy(s: *const Ring, index: usize, out: []u8) ![]const u8 {
104 const e = s.entry(index) orelse return error.InvalidIndex;
105 if (out.len < e.length) return error.BufferTooSmall;
106 const first = @min(e.length, s.bytes.len - e.offset);
107 @memcpy(out[0..first], s.bytes[e.offset..][0..first]);
108 @memcpy(out[first..e.length], s.bytes[0 .. e.length - first]);
109 return out[0..e.length];
110 }
111};
112test "byte wrap and GOP eviction keep only independently decodable history" {
113 var ring = try Ring.init(std.testing.allocator, 10, 4, 100);
114 defer ring.deinit();
115 try std.testing.expect(try ring.push("aaaa", 0, true));
116 try std.testing.expect(try ring.push("bb", 1, false));
117 try std.testing.expect(try ring.push("ccc", 2, true));
118 try std.testing.expect(try ring.push("dddd", 3, false));
119 try std.testing.expectEqual(@as(usize, 2), ring.count);
120 try std.testing.expectEqual(@as(usize, 7), ring.used);
121 var out: [10]u8 = undefined;
122 try std.testing.expectEqualStrings("ccc", try ring.copy(0, &out));
123 try std.testing.expectEqualStrings("dddd", try ring.copy(1, &out));
124 try std.testing.expect(!try ring.push("12345678", 4, false));
125 try std.testing.expectEqual(@as(usize, 0), ring.count);
126 try std.testing.expect(try ring.push("key", 5, true));
127 try std.testing.expectEqualStrings("key", try ring.copy(0, &out));
128}
129test "frame and time limits; rejected inputs preserve existing payload" {
130 var ring = try Ring.init(std.testing.allocator, 10, 2, 3);
131 defer ring.deinit();
132 try std.testing.expect(!try ring.push("p", 0, false));
133 try std.testing.expect(try ring.push("a", 1, true));
134 try std.testing.expect(try ring.push("b", 2, false));
135 try std.testing.expect(!try ring.push("c", 3, false));
136 try std.testing.expect(try ring.push("d", 4, true));
137 try std.testing.expectError(error.PacketTooLarge, ring.push("01234567890", 5, true));
138 try std.testing.expectError(error.InvalidTimestamp, ring.push("x", 4, true));
139 var out: [1]u8 = undefined;
140 try std.testing.expectEqualStrings("d", try ring.copy(0, &out));
141 try std.testing.expect(!try ring.push("e", 8, false));
142 try std.testing.expectEqual(@as(usize, 0), ring.count);
143 try std.testing.expectError(error.InvalidIndex, ring.copy(0, &out));
144}
145test "initialization rejects zero limits and frees partial allocations" {
146 try std.testing.expectError(error.InvalidConfiguration, Ring.init(std.testing.allocator, 0, 1, 1));
147 const Check = struct {
148 fn run(a: std.mem.Allocator) !void {
149 var r = try Ring.init(a, 10, 2, 3);
150 defer r.deinit();
151 }
152 };
153 try std.testing.checkAllAllocationFailures(std.testing.allocator, Check.run, .{});
154}
155
156test "FIFO delay preserves dependencies and rejects overflow atomically" {
157 var r = try Ring.init(std.testing.allocator, 8, 3, 1);
158 defer r.deinit();
159 try r.pushQueued("key", 10, true);
160 try r.pushQueued("p", 20, false);
161 try std.testing.expect(!r.ready(19, 10));
162 try std.testing.expect(r.ready(20, 10));
163 r.pop();
164 try std.testing.expect(!r.entry(0).?.key);
165 try r.pushQueued("next", 30, false);
166 try std.testing.expectError(error.BufferFull, r.pushQueued("full", 40, true));
167 try std.testing.expectEqual(@as(u64, 30), r.last.?);
168 try std.testing.expectEqual(@as(usize, 2), r.count);
169 try std.testing.expect(r.ready(null, std.math.maxInt(u64)));
170 try std.testing.expect(!r.ready(0, 0));
171 var clone = try r.cloneFrom(std.testing.allocator, 0);
172 defer clone.deinit();
173 r.pop();
174 r.pop();
175 var out: [8]u8 = undefined;
176 try std.testing.expectEqualStrings("next", try clone.copy(1, &out));
177}
178test "suffix cloning frees partial allocations and preserves source" {
179 const Check = struct {
180 fn run(a: std.mem.Allocator) !void {
181 var r = try Ring.init(a, 9, 3, 10);
182 defer r.deinit();
183 _ = try r.push("first", 0, true);
184 _ = try r.push("next", 1, true);
185 var frozen = try r.cloneFrom(a, 1);
186 defer frozen.deinit();
187 try std.testing.expectEqual(@as(usize, 2), r.count);
188 }
189 };
190 try std.testing.checkAllAllocationFailures(std.testing.allocator, Check.run, .{});
191}
packages/video/src/video.zigmodified+2-0
@@ -842,6 +842,7 @@ pub const applyAudioGain = sample_convert.applyGain;
842842pub const normalizeAudio = sample_convert.normalize;
843843pub const AudioChannelLayout = sample_convert.ChannelLayout;
844844
845pub const packet_ring = ;
845846pub const pixel_convert = ;
846847pub const PixelConvertFormat = pixel_convert.PixelFormat;
847848pub const ConvertColorSpace = pixel_convert.ColorSpace;
@@ -1218,6 +1219,7 @@ test "Video library imports" {
12181219 _ = dash;
12191220 _ = subtitle_convert;
12201221 _ = sample_convert;
1222 _ = packet_ring;
12211223 _ = pixel_convert;
12221224 _ = nal_convert;
12231225 _ = remux;