StreamSource: synchronize playback with frame presentation data

1 files changed, 75 insertions, 45 deletions

diff --git a/src/ffmpeg.zig b/src/ffmpeg.zig
index f5f0ee4..37b8d69 100644
--- a/src/ffmpeg.zig
+++ b/src/ffmpeg.zig
@@ -69,7 +69,7 @@ pub const Decoder = struct {
     vtable: VTable,
 
     pub const VTable = struct {
-        process_next_frame_fn: *const fn (self: *Decoder, texture: ?*const gl.Texture, z: i32) Errors!void,
+        process_next_frame_fn: *const fn (self: *Decoder, texture: ?*const gl.Texture, z: i32) Errors!?f64,
         init_texture_fn: *const fn (self: *Decoder, texture: *const gl.Texture) void,
         reset_fn: ?*const fn (self: *Decoder) void,
         deinit_fn: *const fn (self: *Decoder, allocator: std.mem.Allocator) void,
@@ -94,22 +94,20 @@ pub const Decoder = struct {
     }
 
     pub fn getNextPacket(self: *Decoder) Errors!void {
-        while (true) {
-            return switch (c.av_read_frame(self.format, self.packet)) {
-                c.AVERROR_EOF => error.EOF,
-                c.AVERROR(c.EAGAIN) => error.TryLater,
-                else => |res| {
-                    try check(res);
-
-                    if (self.packet.*.stream_index == self.stream.index) {
-                        return;
-                    } else {
-                        c.av_packet_unref(self.packet);
-                        continue;
-                    }
-                },
-            };
-        }
+        return res: switch (c.av_read_frame(self.format, self.packet)) {
+            c.AVERROR_EOF => error.EOF,
+            c.AVERROR(c.EAGAIN) => error.TryLater,
+            else => |res| {
+                try check(res);
+
+                if (self.packet.*.stream_index == self.stream.index) {
+                    break :res;
+                } else {
+                    c.av_packet_unref(self.packet);
+                    continue :res c.av_read_frame(self.format, self.packet);
+                }
+            },
+        };
     }
 
     fn processStream(
@@ -129,13 +127,14 @@ pub const Decoder = struct {
         try self.rewind();
         self.num_frames = 0;
         while (true) {
-            const res = self.vtable.process_next_frame_fn(self, texture, self.num_frames);
-            if (res != error.EOF) try res;
+            const res = self.processFrame(texture, self.num_frames);
+            if (res == error.EOF) break;
 
+            _ = try res;
             self.num_frames += 1;
             progress.setCompletedItems(@intCast(self.num_frames));
 
-            if (just_one or res == error.EOF) break;
+            if (just_one) break;
         }
     }
 
@@ -143,12 +142,8 @@ pub const Decoder = struct {
         self: *Decoder,
         texture: ?*const gl.Texture,
         z: i32,
-    ) Errors!void {
-        if (!self.opened) {
-            try self.getNextPacket();
-            self.opened = true;
-        }
-        try self.vtable.process_next_frame_fn(self, texture, z);
+    ) Errors!?f64 {
+        return try self.vtable.process_next_frame_fn(self, texture, z);
     }
 
     pub fn rewind(self: *Decoder) !void {
@@ -294,7 +289,7 @@ pub const AVDecoder = struct {
         }
     }
 
-    fn processNextFrame(decoder: *Decoder, texture: ?*const gl.Texture, z: i32) Errors!void {
+    fn processNextFrame(decoder: *Decoder, texture: ?*const gl.Texture, z: i32) Errors!?f64 {
         const self: *AVDecoder = @fieldParentPtr("decoder", decoder);
 
         try self.getNextFrame();
@@ -310,6 +305,9 @@ pub const AVDecoder = struct {
                 self.rgba_frame.*.data[0],
             );
         }
+
+        const pts: f64 = @floatFromInt(self.raw_frame.*.best_effort_timestamp);
+        return pts * c.av_q2d(self.decoder.stream.*.time_base);
     }
 };
 
@@ -373,7 +371,9 @@ pub const StreamSource = struct {
     decoder: *Decoder,
     format: *c.AVFormatContext,
     depth: i32,
+
     frame: i32,
+    start_time: ?f64, // last rendered glfw time
 
     thread: *gl.Thread,
 
@@ -435,12 +435,14 @@ pub const StreamSource = struct {
             .decoder = decoder,
             .format = format,
             .depth = depth,
+
             .frame = 0,
+            .start_time = null,
 
             .thread = undefined,
         };
 
-        _ = try self.update();
+        _ = try self.update(false);
         self.thread = try gl.Thread.init(allocator, constants, update_loop, .{self});
 
         return &self.source;
@@ -470,30 +472,58 @@ pub const StreamSource = struct {
     }
 
     fn update_loop(self: *StreamSource) !void {
+        var skip = false;
         while (!self.thread.quit) {
-            while (true) {
-                if (self.update()) {
-                    break;
-                } else |err| {
-                    switch (err) {
-                        error.TryLater => break,
-                        error.EOF => try self.decoder.rewind(),
-                        else => return err,
+            if (self.update(skip)) |maybe_pts| {
+                if (maybe_pts) |pts| {
+                    if (self.start_time) |start_time| {
+                        while (true) {
+                            const position = c.glfwGetTime() - start_time;
+                            const delta = pts - position;
+                            if (delta > 0.001) {
+                                // we are EARLY
+                                const nanos: u64 = @intFromFloat(delta * 1_000_000_000 - 100000);
+                                std.Thread.sleep(nanos);
+                                continue;
+                            } else if (delta < -0.5) {
+                                // we are LATE
+                                skip = true;
+                            }
+                            break;
+                        }
+                    } else {
+                        self.start_time = c.glfwGetTime() - pts;
                     }
                 }
-            }
 
-            c.glFlush();
-            try std.Thread.yield();
+                if (skip) continue;
+
+                c.glFlush();
+                try std.Thread.yield();
+            } else |err| {
+                switch (err) {
+                    error.TryLater => std.Thread.sleep(1000),
+                    error.EOF => {
+                        try self.decoder.rewind();
+                        self.start_time = null;
+                    },
+                    else => return err,
+                }
+            }
         }
     }
 
-    fn update(self: *StreamSource) Errors!void {
-        try self.decoder.processFrame(
-            if (self.flags.shouldStep()) &self.source.texture else null,
-            self.frame,
-        );
-        self.frame = @rem(self.frame + 1, self.depth);
+    fn update(self: *StreamSource, skip: bool) Errors!?f64 {
+        if (self.flags.shouldStep() and !skip) {
+            const new_pos = try self.decoder.processFrame(
+                &self.source.texture,
+                self.frame,
+            );
+            self.frame = @rem(self.frame + 1, self.depth);
+            return new_pos;
+        } else {
+            return try self.decoder.processFrame(null, 0);
+        }
     }
 
     fn deinit(source: *src.Source, allocator: std.mem.Allocator) void {