rtpssrcdemux: fix "data flow before segment event" crash

This crash could happen at any time a RTP and RTCP buffer arrived
simultaneously in ssrcdemux.

The problem was that sticky-event arriving while the rtp and rtcp pads
were being set up could arrive just too late to be included in the initial
forwarding.

The fix checks if the stickies have been sent on the srcpad about to be
pushed on, and if not sends them. It also blocks any stickes from
being forwarded *prior* to this happening, to avoid them arriving on
the srcpad multiple times.

Since the test loops 1000 times, this will make running under valgrind
take forever, so use the RUNNING_ON_VALGRIND variable to detect we
are running under valgrind, and reduce the loop-count to 2 in that case.

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

1 files changed, 22 insertions, 14 deletions

diff --git a/gst/rtpmanager/gstrtpssrcdemux.c b/gst/rtpmanager/gstrtpssrcdemux.c
index 7b8d0960d..132c52d10 100644
--- a/gst/rtpmanager/gstrtpssrcdemux.c
+++ b/gst/rtpmanager/gstrtpssrcdemux.c
@@ -83,6 +83,10 @@ GST_STATIC_PAD_TEMPLATE ("rtcp_src_%u",
 #define INTERNAL_STREAM_LOCK(obj)   (g_rec_mutex_lock (&(obj)->padlock))
 #define INTERNAL_STREAM_UNLOCK(obj) (g_rec_mutex_unlock (&(obj)->padlock))
 
+#define GST_PAD_FLAG_STICKIES_SENT (GST_PAD_FLAG_LAST << 0)
+#define GST_PAD_STICKIES_SENT(pad)  (GST_OBJECT_FLAG_IS_SET (pad, GST_PAD_FLAG_STICKIES_SENT))
+#define GST_PAD_SET_STICKIES_SENT(pad) (GST_OBJECT_FLAG_SET (pad, GST_PAD_FLAG_STICKIES_SENT))
+
 typedef enum
 {
   RTP_PAD,
@@ -244,13 +248,11 @@ forward_sticky_events (GstPad * pad, GstEvent ** event, gpointer user_data)
   GstEvent *newevent;
 
   newevent = add_ssrc_and_ref (*event, data->ssrc);
-
   gst_pad_push_event (data->pad, newevent);
 
   return TRUE;
 }
 
-/* With internal stream lock held */
 static void
 forward_initial_events (GstRtpSsrcDemux * demux, guint32 ssrc, GstPad * pad,
     PadType padtype)
@@ -318,9 +320,6 @@ find_or_create_demux_pad_for_ssrc (GstRtpSsrcDemux * demux, guint32 ssrc,
   dpads->rtp_pad = rtp_pad;
   dpads->rtcp_pad = rtcp_pad;
 
-  gst_pad_set_element_private (rtp_pad, dpads);
-  gst_pad_set_element_private (rtcp_pad, dpads);
-
   GST_OBJECT_LOCK (demux);
   demux->srcpads = g_slist_prepend (demux->srcpads, dpads);
   GST_OBJECT_UNLOCK (demux);
@@ -338,9 +337,6 @@ find_or_create_demux_pad_for_ssrc (GstRtpSsrcDemux * demux, guint32 ssrc,
   gst_pad_use_fixed_caps (rtcp_pad);
   gst_pad_set_active (rtcp_pad, TRUE);
 
-  forward_initial_events (demux, ssrc, rtp_pad, RTP_PAD);
-  forward_initial_events (demux, ssrc, rtcp_pad, RTCP_PAD);
-
   gst_element_add_pad (GST_ELEMENT_CAST (demux), rtp_pad);
   gst_element_add_pad (GST_ELEMENT_CAST (demux), rtcp_pad);
 
@@ -606,6 +602,13 @@ forward_event (GstPad * pad, gpointer user_data)
   GSList *walk = NULL;
   GstEvent *newevent = NULL;
 
+  /* special case for EOS */
+  if (GST_EVENT_TYPE (fdata->event) == GST_EVENT_EOS)
+    GST_PAD_SET_STICKIES_SENT (pad);
+
+  if (GST_EVENT_IS_STICKY (fdata->event) && !GST_PAD_STICKIES_SENT (pad))
+    return FALSE;
+
   GST_OBJECT_LOCK (fdata->demux);
   for (walk = fdata->demux->srcpads; walk; walk = walk->next) {
     GstRtpSsrcDemuxPads *dpads = (GstRtpSsrcDemuxPads *) walk->data;
@@ -668,6 +671,11 @@ gst_rtp_ssrc_demux_chain (GstPad * pad, GstObject * parent, GstBuffer * buf)
   if (srcpad == NULL)
     goto create_failed;
 
+  if (!GST_PAD_STICKIES_SENT (srcpad)) {
+    forward_initial_events (demux, ssrc, srcpad, RTP_PAD);
+    GST_PAD_SET_STICKIES_SENT (srcpad);
+  }
+
   /* push to srcpad */
   ret = gst_pad_push (srcpad, buf);
 
@@ -758,6 +766,11 @@ gst_rtp_ssrc_demux_rtcp_chain (GstPad * pad, GstObject * parent,
   if (srcpad == NULL)
     goto create_failed;
 
+  if (!GST_PAD_STICKIES_SENT (srcpad)) {
+    forward_initial_events (demux, ssrc, srcpad, RTCP_PAD);
+    GST_PAD_SET_STICKIES_SENT (srcpad);
+  }
+
   /* push to srcpad */
   ret = gst_pad_push (srcpad, buf);
 
@@ -944,17 +957,12 @@ gst_rtp_ssrc_demux_src_query (GstPad * pad, GstObject * parent,
       if ((res = gst_pad_peer_query (demux->rtp_sink, query))) {
         gboolean live;
         GstClockTime min_latency, max_latency;
-        GstRtpSsrcDemuxPads *dpads;
-
-        dpads = gst_pad_get_element_private (pad);
 
         gst_query_parse_latency (query, &live, &min_latency, &max_latency);
 
-        GST_DEBUG_OBJECT (demux, "peer min latency %" GST_TIME_FORMAT,
+        GST_DEBUG_OBJECT (pad, "peer min latency %" GST_TIME_FORMAT,
             GST_TIME_ARGS (min_latency));
 
-        GST_DEBUG_OBJECT (demux, "latency for SSRC %08x", dpads->ssrc);
-
         gst_query_set_latency (query, live, min_latency, max_latency);
       }
       break;