gst/rtpmanager/gstrtpsession.c: Implement forward and reverse reconsideration.

Original commit message from CVS:
* gst/rtpmanager/gstrtpsession.c: (rtcp_thread):
Implement forward and reverse reconsideration.
* gst/rtpmanager/rtpsession.c: (rtp_session_get_num_sources),
(rtp_session_get_num_active_sources), (rtp_session_process_sr),
(session_report_blocks):
* gst/rtpmanager/rtpsession.h:
Small cleanups.

3 files changed, 79 insertions, 23 deletions

diff --git a/gst/rtpmanager/gstrtpsession.c b/gst/rtpmanager/gstrtpsession.c
index 39c63e32..bc297235 100644
--- a/gst/rtpmanager/gstrtpsession.c
+++ b/gst/rtpmanager/gstrtpsession.c
@@ -293,6 +293,13 @@ rtcp_thread (GstRTPSession * rtpsession)
 {
   GstClock *clock;
   GstClockID id;
+  gdouble interval;
+  GstClockTime current_time;
+  GstClockTime next_rtcp_check_time;
+  GstClockTime new_rtcp_send_time;
+  GstClockTime last_rtcp_send_time;
+  GstClockTimeDiff jitter;
+  guint members, prev_members;
 
   clock = gst_element_get_clock (GST_ELEMENT_CAST (rtpsession));
   if (clock == NULL)
@@ -301,34 +308,83 @@ rtcp_thread (GstRTPSession * rtpsession)
   GST_DEBUG_OBJECT (rtpsession, "entering RTCP thread");
 
   GST_RTP_SESSION_LOCK (rtpsession);
+
+  /* get initial estimate */
+  interval = rtp_session_get_reporting_interval (rtpsession->priv->session);
+  current_time = gst_clock_get_time (clock);
+  last_rtcp_send_time = current_time;
+  next_rtcp_check_time = current_time + (GST_SECOND * interval);
+  /* we keep track of members before and after the timeout to do reverse
+   * reconsideration. */
+  prev_members = rtp_session_get_num_active_sources (rtpsession->priv->session);
+
+  GST_DEBUG_OBJECT (rtpsession, "first RTCP interval: %lf seconds", interval);
+
   while (!rtpsession->priv->stop_thread) {
-    gdouble timeout;
-    GstClockTime target;
     GstClockReturn res;
 
-    timeout = rtp_session_get_reporting_interval (rtpsession->priv->session);
-    GST_DEBUG_OBJECT (rtpsession, "next RTCP timeout: %lf", timeout);
+    GST_DEBUG_OBJECT (rtpsession, "next check time %" GST_TIME_FORMAT,
+        GST_TIME_ARGS (next_rtcp_check_time));
 
-    target = gst_clock_get_time (clock);
-    target += GST_SECOND * timeout;
+    id = rtpsession->priv->id =
+        gst_clock_new_single_shot_id (clock, next_rtcp_check_time);
+    GST_RTP_SESSION_UNLOCK (rtpsession);
 
+    res = gst_clock_id_wait (id, &jitter);
 
-    id = rtpsession->priv->id = gst_clock_new_single_shot_id (clock, target);
-    GST_RTP_SESSION_UNLOCK (rtpsession);
+    GST_RTP_SESSION_LOCK (rtpsession);
+    gst_clock_id_unref (id);
+    rtpsession->priv->id = NULL;
+
+    if (rtpsession->priv->stop_thread)
+      break;
+
+    if (res != GST_CLOCK_UNSCHEDULED)
+      if (jitter < 0)
+        current_time = next_rtcp_check_time;
+      else
+        current_time = next_rtcp_check_time - jitter;
+    else
+      current_time = gst_clock_get_time (clock);
+
+    GST_DEBUG_OBJECT (rtpsession, "unlocked %d, jitter %" G_GINT64_FORMAT
+        ", current %" GST_TIME_FORMAT, res, jitter,
+        GST_TIME_ARGS (current_time));
 
-    res = gst_clock_id_wait (id, NULL);
-    if (res != GST_CLOCK_UNSCHEDULED) {
-      GST_DEBUG_OBJECT (rtpsession, "got RTCP timeout");
+    members = rtp_session_get_num_active_sources (rtpsession->priv->session);
+
+    if (members < prev_members) {
+      GstClockTime time_remaining;
+
+      /* some members went away */
+      GST_DEBUG_OBJECT (rtpsession, "reverse reconsideration");
+      time_remaining = next_rtcp_check_time - current_time;
+      new_rtcp_send_time =
+          current_time + (time_remaining * members / prev_members);
+    } else {
+      interval = rtp_session_get_reporting_interval (rtpsession->priv->session);
+      GST_DEBUG_OBJECT (rtpsession, "forward reconsideration: %lf seconds",
+          interval);
+      new_rtcp_send_time = (interval * GST_SECOND) + last_rtcp_send_time;
+    }
+    prev_members = members;
+
+    if (current_time >= new_rtcp_send_time) {
+      GST_DEBUG_OBJECT (rtpsession, "sending RTCP now");
 
       /* make the session manager produce RTCP, we ignore the result. */
       rtp_session_perform_reporting (rtpsession->priv->session);
+
+      interval = rtp_session_get_reporting_interval (rtpsession->priv->session);
+
+      GST_DEBUG_OBJECT (rtpsession, "next RTCP interval: %lf seconds",
+          interval);
+      next_rtcp_check_time = (interval * GST_SECOND) + current_time;
+      last_rtcp_send_time = current_time;
     } else {
-      GST_DEBUG_OBJECT (rtpsession, "got unscheduled");
+      GST_DEBUG_OBJECT (rtpsession, "reconsider RTCP");
+      next_rtcp_check_time = new_rtcp_send_time;
     }
-
-    GST_RTP_SESSION_LOCK (rtpsession);
-    gst_clock_id_unref (id);
-    rtpsession->priv->id = NULL;
   }
   GST_RTP_SESSION_UNLOCK (rtpsession);
 
diff --git a/gst/rtpmanager/rtpsession.c b/gst/rtpmanager/rtpsession.c
index 27d6dabb..1f6e1ebc 100644
--- a/gst/rtpmanager/rtpsession.c
+++ b/gst/rtpmanager/rtpsession.c
@@ -703,10 +703,10 @@ rtp_session_add_source (RTPSession * sess, RTPSource * src)
  *
  * Returns: The number of sources in @sess.
  */
-gint
+guint
 rtp_session_get_num_sources (RTPSession * sess)
 {
-  gint result;
+  guint result;
 
   g_return_val_if_fail (RTP_IS_SESSION (sess), FALSE);
 
@@ -726,12 +726,12 @@ rtp_session_get_num_sources (RTPSession * sess)
  *
  * Returns: The number of active sources in @sess.
  */
-gint
+guint
 rtp_session_get_num_active_sources (RTPSession * sess)
 {
-  gint result;
+  guint result;
 
-  g_return_val_if_fail (RTP_IS_SESSION (sess), FALSE);
+  g_return_val_if_fail (RTP_IS_SESSION (sess), 0);
 
   RTP_SESSION_LOCK (sess);
   result = sess->stats.active_sources;
diff --git a/gst/rtpmanager/rtpsession.h b/gst/rtpmanager/rtpsession.h
index 493387fa..3554016f 100644
--- a/gst/rtpmanager/rtpsession.h
+++ b/gst/rtpmanager/rtpsession.h
@@ -216,8 +216,8 @@ gchar*          rtp_session_get_note               (RTPSession *sess);
 
 /* handling sources */
 gboolean        rtp_session_add_source             (RTPSession *sess, RTPSource *src);
-gint            rtp_session_get_num_sources        (RTPSession *sess);
-gint            rtp_session_get_num_active_sources (RTPSession *sess);
+guint           rtp_session_get_num_sources        (RTPSession *sess);
+guint           rtp_session_get_num_active_sources (RTPSession *sess);
 RTPSource*      rtp_session_get_source_by_ssrc     (RTPSession *sess, guint32 ssrc);
 RTPSource*      rtp_session_get_source_by_cname    (RTPSession *sess, const gchar *cname);
 RTPSource*      rtp_session_create_source          (RTPSession *sess);