oggdemux: fix a race in push mode when performing the duration seek

There may be two or more threads involved here however the important
interaction is the use of ogg->seeK_event_drop_till value that was only
set in the push-mode seek-event thread and could race with upstream
sending e.g. and EOS (or data).

Scenario is this:
1. oggdemux performs a seek to near the end of the file to try and find
   the duration. ogg->push_state is set to PUSH_DURATION.
2. Seek is picked up by the dedicated seek event thread and sets
   ogg->seek_event_drop_till to the seek event's seqnum.
3. Most operations are blocked or dropped waiting on the duration to
   be determined and processing continues until a duration is found.
4. Two branching options for how this ultimately plays out
4a. The source is too fast and we receive an EOS event which is dropped
    because ogg->push_state == PUSH_DURATION.  In this case everything
    works.
4b. We hit our 'almost at the end' check in
    gst_ogg_pad_handle_push_mode_state() and attempt to seek back to the
    beginning (or to a user-provided seek).  This seek is marshalled to
    the seek event thread without setting ogg->seek_event_drop_till but
    with change ogg->push_state = PUSH_PLAYING.  If an EOS event or
    e.g. buffers arrive from upstream before the seek event thread has
    picked up the seek event, then the EOS/data is processed as if it
    came as a result of the seek event.  This is the case that fails.

The fix is two-fold:
1. Preemptively set ogg->seek_event_drop_till when setting the seek
   event so that data and other events can be dropped correctly.
2. In addition to dropping and EOS events while ogg->push_state ==
   PUSH_DURATION, also drop any EOS events that are received before the
   seek event has been processed by also tracking the seqnum of the seek.

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

1 files changed, 18 insertions, 2 deletions

diff --git a/ext/ogg/gstoggdemux.c b/ext/ogg/gstoggdemux.c
index e88e325f1..2ce279dd7 100644
--- a/ext/ogg/gstoggdemux.c
+++ b/ext/ogg/gstoggdemux.c
@@ -1610,6 +1610,10 @@ gst_ogg_demux_seek_back_after_push_duration_check_unlock (GstOggDemux * ogg)
     event = gst_event_new_seek (1.0, GST_FORMAT_BYTES,
         GST_SEEK_FLAG_ACCURATE | GST_SEEK_FLAG_FLUSH,
         GST_SEEK_TYPE_SET, 1, GST_SEEK_TYPE_SET, GST_CLOCK_TIME_NONE);
+    /* drop everything until this seek event completed.  We can't wait until the
+     * seek thread sets this because there would be race between receiving e.g.
+     * an EOS or any data and the seek thread actually picking up the seek. */
+    ogg->seek_event_drop_till = gst_event_get_seqnum (event);
   }
   gst_event_replace (&ogg->seek_event, event);
   gst_event_unref (event);
@@ -2506,6 +2510,7 @@ gst_ogg_demux_sink_event (GstPad * pad, GstObject * parent, GstEvent * event)
       break;
     case GST_EVENT_EOS:
     {
+      gboolean drop = FALSE;
       GST_DEBUG_OBJECT (ogg, "got an EOS event");
       GST_PUSH_LOCK (ogg);
       if (ogg->push_state == PUSH_DURATION) {
@@ -2515,10 +2520,20 @@ gst_ogg_demux_sink_event (GstPad * pad, GstObject * parent, GstEvent * event)
           GST_DEBUG_OBJECT (ogg, "Error seeking back after duration check: %d",
               res);
         }
+        res = TRUE;
         break;
-      } else
+      } else {
+        if (ogg->seek_event_drop_till > 0) {
+          GST_DEBUG_OBJECT (ogg, "Dropping EOS (seqnum:%u) because we have "
+              "a pending seek (seqnum:%u)", gst_event_get_seqnum (event),
+              ogg->seek_event_drop_till);
+          drop = TRUE;
+        }
         GST_PUSH_UNLOCK (ogg);
-      res = gst_ogg_demux_send_event (ogg, event);
+        res = TRUE;
+      }
+      if (!drop)
+        res = gst_ogg_demux_send_event (ogg, event);
       if (ogg->current_chain == NULL) {
         GST_WARNING_OBJECT (ogg,
             "EOS while trying to retrieve chain, seeking disabled");
@@ -3719,6 +3734,7 @@ gst_ogg_demux_get_duration_push (GstOggDemux * ogg, int flags)
   sevent = gst_event_new_seek (1.0, GST_FORMAT_BYTES, flags, GST_SEEK_TYPE_SET,
       position, GST_SEEK_TYPE_SET, ogg->push_byte_length - 1);
   gst_event_replace (&ogg->seek_event, sevent);
+  ogg->seek_event_drop_till = gst_event_get_seqnum (sevent);
   gst_event_unref (sevent);
   g_mutex_lock (&ogg->seek_event_mutex);
   g_cond_broadcast (&ogg->seek_event_cond);