also looking at this
feat(media): add bounded PCM capture, owned snapshots and A/V streaming
#917
11 files
+683
-0
Review threads live on the whole diff, not on one commit, so none are shown here - a thread's line means something in the branch's final form, and painting it into an intermediate step would put it on code it is not about.
| @@ -56,3 +56,10 @@ Consumers can align an encoded keyframe ring to the PCM retention floor and | ||
| 56 | 56 | replay separate snapshots without consuming live audio. If eviction passes a |
| 57 | 57 | live consumer's watermark, report loss rather than silently producing a gap. |
| 58 | 58 | No encoder, thread or file I/O is performed by these history methods. |
| 59 | ||
| 60 | `Queue.cloneWindow(allocator, minimum, cut)` freezes just the selected sample | |
| 61 | window into independent storage. It preserves capture timestamps and sample phase, | |
| 62 | handles a wrapped source block even when all 4096 descriptors are occupied, and | |
| 63 | frees partial allocations on failure. The source is unchanged. Serialize cloning | |
| 64 | with source mutation; after return the copy may move to an independent worker and | |
| 65 | outlive the source. No audio encoder or worker runtime is imported by this module. | |
| @@ -102,6 +102,27 @@ pub const Queue = struct { | ||
| 102 | 102 | } |
| 103 | 103 | return null; |
| 104 | 104 | } |
| 105 | /// Freeze a selected sample window into independent, minimally sized storage. | |
| 106 | /// Caller serializes source mutation; destination may then move to a worker. | |
| 107 | pub fn cloneWindow(s: *const Queue, a: std.mem.Allocator, minimum: u64, cut: u64) !Queue { | |
| 108 | var cursor: Cursor = .{}; | |
| 109 | var frames: usize = 0; | |
| 110 | var blocks: usize = 0; | |
| 111 | while (try s.window(minimum, cut, &cursor)) |chunk| { | |
| 112 | frames += chunk.frames; | |
| 113 | blocks += 1; | |
| 114 | } | |
| 115 | // One wrapped source block can become two contiguous descriptors. | |
| 116 | // A full source therefore needs at most entries.len + 1 descriptors. | |
| 117 | const data = try a.alloc(u8, @max(frames, 1) * s.channels * 2); | |
| 118 | errdefer a.free(data); | |
| 119 | const entries = try a.alloc(Entry, @max(blocks, 1)); | |
| 120 | errdefer a.free(entries); | |
| 121 | var result = Queue{ .allocator = a, .data = data, .entries = entries, .rate = s.rate, .channels = s.channels, .capacity = @max(frames, 1) }; | |
| 122 | cursor = .{}; | |
| 123 | while (try s.window(minimum, cut, &cursor)) |chunk| try result.push(chunk.data, chunk.timestamp); | |
| 124 | return result; | |
| 125 | } | |
| 105 | 126 | /// Emit samples with time >= minimum and < cut. Wrap is returned as separate |
| 106 | 127 | /// contiguous chunks; never duplicate a sample at a segment boundary. |
| 107 | 128 | pub fn take(s: *Queue, minimum: u64, cut: u64) !?Chunk { |
| @@ -204,3 +225,38 @@ test "rolling PCM metadata limit and window clipping preserve gaps" { | ||
| 204 | 225 | try std.testing.expectEqual(@as(u64, 1250000), chunk.end_ns); |
| 205 | 226 | try std.testing.expect((try q.window(1125000, 2000000, &cursor)) == null); |
| 206 | 227 | } |
| 228 | ||
| 229 | test "frozen PCM windows own wrapped clipped samples and clean up allocation failures" { | |
| 230 | const Check = struct { | |
| 231 | fn run(a: std.mem.Allocator) !void { | |
| 232 | var q = try Queue.init(a, 44100, 1, 6, 3); | |
| 233 | defer q.deinit(); | |
| 234 | try q.pushRolling(&.{ 1, 0, 2, 0, 3, 0, 4, 0 }, 1000); | |
| 235 | try q.pushRolling(&.{ 5, 0, 6, 0, 7, 0, 8, 0 }, try at(1000, 4, 44100)); | |
| 236 | const begin = try at(try at(1000, 4, 44100), 1, 44100); | |
| 237 | var frozen = try q.cloneWindow(a, begin, std.math.maxInt(u64)); | |
| 238 | defer frozen.deinit(); | |
| 239 | try q.pushRolling(&.{ 9, 0, 10, 0, 11, 0, 12, 0 }, q.last_end.?); | |
| 240 | var cursor: Cursor = .{}; | |
| 241 | var frames: usize = 0; | |
| 242 | var values: [6]u8 = undefined; | |
| 243 | while (try frozen.window(0, std.math.maxInt(u64), &cursor)) |chunk| { | |
| 244 | @memcpy(values[frames * 2 ..][0..chunk.data.len], chunk.data); | |
| 245 | frames += chunk.frames; | |
| 246 | } | |
| 247 | try std.testing.expectEqualSlices(u8, &.{ 6, 0, 7, 0, 8, 0 }, &values); | |
| 248 | } | |
| 249 | }; | |
| 250 | try std.testing.checkAllAllocationFailures(std.testing.allocator, Check.run, .{}); | |
| 251 | } | |
| 252 | ||
| 253 | test "freeze supports a full descriptor queue with a wrapped final block" { | |
| 254 | var q = try Queue.init(std.testing.allocator, 48000, 1, 8193, 4096); | |
| 255 | defer q.deinit(); | |
| 256 | for (0..4097) |i| try q.pushRolling(&.{ 1, 0, 2, 0 }, try at(0, i * 2, 48000)); | |
| 257 | var frozen = try q.cloneWindow(std.testing.allocator, 0, std.math.maxInt(u64)); | |
| 258 | defer frozen.deinit(); | |
| 259 | try std.testing.expectEqual(@as(usize, 4096), q.count); | |
| 260 | try std.testing.expectEqual(@as(usize, 4097), frozen.count); | |
| 261 | try std.testing.expectEqual(q.used, frozen.used); | |
| 262 | } | |