rtpsession: fix a race condition during the EOS event in gstrtpsession.c

This patch prevents a possible race condition from taking place between the EOS event handling and rtcp send
function/thread.

The condition starts by getting the GST_EVENT_EOS event on the send_rtp_sink pad, which causes two core things
to happen -- the event gets pushed down to the send_rtp_src pad and all sessions get marked "bye" prior to
completion of the event handler. In another thread the rtp_session_on_timeout function gets called after an
expiration of gst_clock_id_wait in the rtcp_thread function. This results in a call to the
ess->callbacks.send_rtcp(), which is configured as a function pointer to gst_rtp_session_send_rtcp via the
RTPSessionCallbacks structure passed to rtp_session_set_callbacks in the gst_rtp_session_init function.

In the race condition, the call to gst_rtp_session_send_rtcp can have the all_sources_bye boolean set to true
while GST_PAD_IS_EOS(rtpsession->send_rtp_sink) evaluates to false. This is the result of gst_rtp_session_send_rtcp
running before the send_rtp_sink's GST_EVENT_EOS handler completes. The exact point at which this condition occurs
is if there's a context switch to the rtcp_thread right after the call to rtp_session_mark_all_bye in the
GET_EVENT_EOS handler, but before the handler returns.

Normally, this would not be an issue because the rtcp_thread continues to run and indirectly call
gst_rtp_session_send_rtcp. However, the call to rtp_source_reset sets the sent_bye boolean to false, which ends up
causing rtp_session_are_all_sources_bye to return false. This gets passed to gst_rtp_session_send_rtcp and the EOS
event never gets sent.

The race condition results in the EOS event never getting passed to the rtcp_src pad, which prevents the bin and
pipeline from ever completing with EOS.

Part-of: <https://gitlab.freedesktop.org/gstreamer/gstreamer/-/merge_requests/3798>

subprojects/gst-plugins-good/gst/rtpmanager/gstrtpsession.c

@@ -289,6 +289,8 @@ struct _GstRtpSessionPrivate
    * pushed a buffer list.
    */
   GstBufferList *processed_list;
+
+  gboolean send_rtp_sink_eos;
 };
 
 /* callbacks to handle actions from the session manager */
@@ -925,6 +927,8 @@ gst_rtp_session_init (GstRtpSession * rtpsession)
   rtpsession->priv->sent_rtx_req_count = 0;
 
   rtpsession->priv->ntp_time_source = DEFAULT_NTP_TIME_SOURCE;
+
+  rtpsession->priv->send_rtp_sink_eos = FALSE;
 }
 
 static void
@@ -1343,6 +1347,9 @@ gst_rtp_session_change_state (GstElement * element, GstStateChange transition)
       break;
     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
     case GST_STATE_CHANGE_PAUSED_TO_READY:
+      GST_RTP_SESSION_LOCK (rtpsession);
+      rtpsession->priv->send_rtp_sink_eos = FALSE;
+      GST_RTP_SESSION_UNLOCK (rtpsession);
       /* no need to join yet, we might want to continue later. Also, the
        * dataflow could block downstream so that a join could just block
        * forever. */
@@ -1545,9 +1552,12 @@ gst_rtp_session_send_rtcp (RTPSession * sess, RTPSource * src,
 
     /* Forward send an EOS on the RTCP sink if we received an EOS on the
      * send_rtp_sink. We don't need to check the recv_rtp_sink since in this
-     * case the EOS event would already have been sent */
+     * case the EOS event would already have been sent. Also, prevent a
-    if (all_sources_bye && rtpsession->send_rtp_sink &&
+     * race condition between the EOS event handling and rtcp send
-        GST_PAD_IS_EOS (rtpsession->send_rtp_sink)) {
+     * function/thread  by using send_rtp_sink_eos directly instead of
+     * GST_PAD_IS_EOS*/
+    GST_RTP_SESSION_LOCK (rtpsession);
+    if (all_sources_bye && rtpsession->priv->send_rtp_sink_eos) {
       GstEvent *event;
 
       GST_LOG_OBJECT (rtpsession, "sending EOS");
@@ -1556,6 +1566,7 @@ gst_rtp_session_send_rtcp (RTPSession * sess, RTPSource * src,
       gst_event_set_seqnum (event, rtpsession->recv_rtcp_segment_seqnum);
       gst_pad_push_event (rtcp_src, event);
     }
+    GST_RTP_SESSION_UNLOCK (rtpsession);
     gst_object_unref (rtcp_src);
   } else {
     GST_RTP_SESSION_UNLOCK (rtpsession);
@@ -1754,6 +1765,9 @@ gst_rtp_session_event_recv_rtp_sink (GstPad * pad, GstObject * parent,
       gst_segment_init (&rtpsession->recv_rtp_seg, GST_FORMAT_UNDEFINED);
       rtpsession->recv_rtcp_segment_seqnum = GST_SEQNUM_INVALID;
       ret = gst_pad_push_event (rtpsession->recv_rtp_src, event);
+      GST_RTP_SESSION_LOCK (rtpsession);
+      rtpsession->priv->send_rtp_sink_eos = FALSE;
+      GST_RTP_SESSION_UNLOCK (rtpsession);
       break;
     case GST_EVENT_SEGMENT:
     {
@@ -2253,6 +2267,9 @@ gst_rtp_session_event_send_rtp_sink (GstPad * pad, GstObject * parent,
        * because we stop sending. */
       ret = gst_pad_push_event (rtpsession->send_rtp_src, event);
       current_time = gst_clock_get_time (rtpsession->priv->sysclock);
+      GST_RTP_SESSION_LOCK (rtpsession);
+      rtpsession->priv->send_rtp_sink_eos = TRUE;
+      GST_RTP_SESSION_UNLOCK (rtpsession);
 
       GST_DEBUG_OBJECT (rtpsession, "scheduling BYE message");
       rtp_session_mark_all_bye (rtpsession->priv->session, "End Of Stream");