videoaggregator: Pop out old buffers on timeout

This situation happens in the situation where an input stream has a framerate
exceeding the timeout latency (Ex: 1fps with a latency of 500ms) and an input
stream greater than output framerate (ex: 60fps in, 30 fps out).

The problem that would happen is that we would timeout, but then buffers from
the fast input stream would only be popped out one by one.... until a buffer
reaches the low-framerate input stream at which point they would quickly be
popped out/used. The resulting output would be "slow ... fast ... slow ... fast"
of that input fast stream.

In order to avoid this situation, whenever we detect a late buffer, check if
there's a next one and re-check with that one.

Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-plugins-base/-/merge_requests/991>

1 files changed, 12 insertions, 2 deletions

diff --git a/gst-libs/gst/video/gstvideoaggregator.c b/gst-libs/gst/video/gstvideoaggregator.c
index 84fa6a446..eb4170e1e 100644
--- a/gst-libs/gst/video/gstvideoaggregator.c
+++ b/gst-libs/gst/video/gstvideoaggregator.c
@@ -1515,7 +1515,7 @@ gst_video_aggregator_reset (GstVideoAggregator * vagg)
 static GstFlowReturn
 gst_video_aggregator_fill_queues (GstVideoAggregator * vagg,
     GstClockTime output_start_running_time,
-    GstClockTime output_end_running_time)
+    GstClockTime output_end_running_time, gboolean timeout)
 {
   GList *l;
   gboolean eos = TRUE;
@@ -1549,6 +1549,9 @@ gst_video_aggregator_fill_queues (GstVideoAggregator * vagg,
     if (buf) {
       GstClockTime start_time, end_time;
 
+    check_again:
+      GST_TRACE_OBJECT (pad, "Next buffer %" GST_PTR_FORMAT, buf);
+
       start_time = GST_BUFFER_TIMESTAMP (buf);
       if (start_time == -1) {
         gst_buffer_unref (buf);
@@ -1589,6 +1592,13 @@ gst_video_aggregator_fill_queues (GstVideoAggregator * vagg,
           gst_buffer_unref (buf);
           gst_aggregator_pad_drop_buffer (bpad);
           pad->priv->start_time = start_time;
+          if (timeout) {
+            /* If we're called for a timeout, we want to make sure we drain as
+             * much as possible any late data */
+            buf = gst_aggregator_pad_peek_buffer (bpad);
+            if (buf)
+              goto check_again;
+          }
           need_more_data = TRUE;
           continue;
         }
@@ -1996,7 +2006,7 @@ gst_video_aggregator_aggregate (GstAggregator * agg, gboolean timeout)
   } else {
     flow_ret =
         gst_video_aggregator_fill_queues (vagg, output_start_running_time,
-        output_end_running_time);
+        output_end_running_time, timeout);
   }
 
   if (flow_ret == GST_AGGREGATOR_FLOW_NEED_DATA && !timeout) {