Wait until we've seen all streams to start decoding. (#3616)

CHANGES.md

@@ -13,7 +13,9 @@ Changed:
 
 Fixed:
 
-- Pop/click at the end of fade out/in (#3318)
+- Fix pop/click at the end of fade out/in (#3318)
+- Fix audio/video synchronization issues when decoding
+  live streams using ffmpeg.
 
 ---
 

src/core/decoder/ffmpeg_decoder.ml

@@ -754,13 +754,45 @@ let mk_eof streams buffer =
     streams
 
 let mk_decoder ~streams ~target_position container =
-  let check_pts stream pts =
+  let streams_seen = Hashtbl.create 0 in
+  let position ~pts stream =
+    let { Avutil.num; den } = Av.get_time_base stream in
+    Int64.to_float pts *. float num /. float den
+  in
+  let decodable = ref [] in
+  let push (position, decode) =
+    decodable :=
+      (position, decode)
+      :: List.filter
+           (fun (p, _) ->
+             Float.abs (p -. position)
+             <= Ffmpeg_decoder_common.conf_max_interleave_duration#get)
+           !decodable
+  in
+  let flush position =
+    let d = List.sort (fun (p, _) (p', _) -> Float.compare p p') !decodable in
+    let min_position =
+      position -. Ffmpeg_decoder_common.conf_max_interleave_delta#get
+    in
+    List.iter (fun (p, decode) -> if min_position <= p then decode ()) d;
+    decodable := []
+  in
+  let check_pts ~decode stream pts =
     match (pts, !target_position) with
       | Some pts, Some target_position ->
-          let { Avutil.num; den } = Av.get_time_base stream in
-          let position = Int64.to_float pts *. float num /. float den in
-          target_position <= position
-      | _ -> true
+          if target_position <= position ~pts stream then decode ()
+      | Some pts, None ->
+          Hashtbl.replace streams_seen (Hashtbl.hash stream) true;
+          let position = position ~pts stream in
+          if Hashtbl.length streams_seen = Streams.cardinal streams then (
+            flush position;
+            decode ())
+          else push (position, decode)
+      | None, _ ->
+          log#important
+            "Got packet or frame with no timestamp! Synchronization issues may \
+             happen.";
+          decode ()
   in
   let audio_frame =
     Streams.fold
@@ -800,32 +832,40 @@ let mk_decoder ~streams ~target_position container =
           | `Audio_frame (i, frame) -> (
               match Streams.find_opt i streams with
                 | Some (`Audio_frame (s, decode)) ->
-                    if check_pts s (Avutil.Frame.pts frame) then
-                      decode ~buffer (`Frame frame)
+                    check_pts s
+                      ~decode:(fun () -> decode ~buffer (`Frame frame))
+                      (Avutil.Frame.pts frame)
                 | _ -> f ())
           | `Audio_packet (i, packet) -> (
               match Streams.find_opt i streams with
                 | Some (`Audio_packet (s, decode)) ->
-                    if check_pts s (Avcodec.Packet.get_pts packet) then
-                      decode ~buffer packet
+                    check_pts
+                      ~decode:(fun () -> decode ~buffer packet)
+                      s
+                      (Avcodec.Packet.get_pts packet)
                 | _ -> f ())
           | `Video_frame (i, frame) -> (
               match Streams.find_opt i streams with
                 | Some (`Video_frame (s, decode)) ->
-                    if check_pts s (Avutil.Frame.pts frame) then
-                      decode ~buffer (`Frame frame)
+                    check_pts
+                      ~decode:(fun () -> decode ~buffer (`Frame frame))
+                      s (Avutil.Frame.pts frame)
                 | _ -> f ())
           | `Video_packet (i, packet) -> (
               match Streams.find_opt i streams with
                 | Some (`Video_packet (s, decode)) ->
-                    if check_pts s (Avcodec.Packet.get_pts packet) then
-                      decode ~buffer packet
+                    check_pts
+                      ~decode:(fun () -> decode ~buffer packet)
+                      s
+                      (Avcodec.Packet.get_pts packet)
                 | _ -> f ())
           | `Data_packet (i, packet) -> (
               match Streams.find_opt i streams with
                 | Some (`Data_packet (s, decode)) ->
-                    if check_pts s (Avcodec.Packet.get_pts packet) then
-                      decode ~buffer packet
+                    check_pts
+                      ~decode:(fun () -> decode ~buffer packet)
+                      s
+                      (Avcodec.Packet.get_pts packet)
                 | _ -> f ())
           | _ -> ()
       with

src/core/decoder/ffmpeg_decoder_common.ml

@@ -32,6 +32,16 @@ let conf_ffmpeg_decoder =
 let conf_codecs =
   Dtools.Conf.unit ~p:(conf_ffmpeg_decoder#plug "codecs") "Codecs settings"
 
+let conf_max_interleave_duration =
+  Dtools.Conf.float
+    ~p:(conf_ffmpeg_decoder#plug "max_interleave_duration")
+    ~d:5. "Maximum data buffered while waiting for all streams."
+
+let conf_max_interleave_delta =
+  Dtools.Conf.float
+    ~p:(conf_ffmpeg_decoder#plug "max_interleave_delta")
+    ~d:0.04 "Maximum delay between interleaved streams."
+
 let conf_codecs =
   let codecs = Hashtbl.create 10 in
   List.iter