gst/rtpmanager/gstrtpjitterbuffer.c: Remove jitter correction code, it's now in the lower level object.

Original commit message from CVS:
* gst/rtpmanager/gstrtpjitterbuffer.c: (apply_latency),
(gst_rtp_jitter_buffer_loop), (gst_rtp_jitter_buffer_query):
Remove jitter correction code, it's now in the lower level object.
Use new -core method for doing a peer query.
* gst/rtpmanager/rtpjitterbuffer.c: (rtp_jitter_buffer_init),
(calculate_skew), (rtp_jitter_buffer_insert):
* gst/rtpmanager/rtpjitterbuffer.h:
Move jitter correction to the lowlevel jitterbuffer.
Increase the max window size.
When filling the window, already start estimating the skew using a
parabolic weighting factor so that we have a much better startup
behaviour that gets more accurate with the more samples we have.
Increase the default weighting factor for the steady state to get
smoother timestamps.

3 files changed, 105 insertions, 92 deletions

diff --git a/gst/rtpmanager/gstrtpjitterbuffer.c b/gst/rtpmanager/gstrtpjitterbuffer.c
index d27c2aa1..2155df89 100644
--- a/gst/rtpmanager/gstrtpjitterbuffer.c
+++ b/gst/rtpmanager/gstrtpjitterbuffer.c
@@ -943,25 +943,14 @@ duplicate:
 }
 
 static GstClockTime
-convert_rtptime_to_gsttime (GstRtpJitterBuffer * jitterbuffer,
-    guint64 exttimestamp)
+apply_latency (GstRtpJitterBuffer * jitterbuffer, GstClockTime timestamp)
 {
-  GstClockTime timestamp;
   GstRtpJitterBufferPrivate *priv;
 
   priv = jitterbuffer->priv;
 
-  /* construct a timestamp from the RTP timestamp now. We don't apply this
-   * timestamp to the outgoing buffer yet as the popped buffer might not be the
-   * one we need to push out right now. */
-  timestamp =
-      gst_util_uint64_scale_int (exttimestamp, GST_SECOND, priv->clock_rate);
-
-  /* apply first observed timestamp */
-  timestamp += priv->jbuf->base_time;
-
-  /* apply the current clock skew */
-  timestamp += priv->jbuf->skew;
+  if (timestamp == -1)
+    return -1;
 
   /* apply the timestamp offset */
   timestamp += priv->ts_offset;
@@ -987,9 +976,7 @@ gst_rtp_jitter_buffer_loop (GstRtpJitterBuffer * jitterbuffer)
   GstBuffer *outbuf = NULL;
   GstFlowReturn result;
   guint16 seqnum;
-  guint32 rtp_time;
-  GstClockTime timestamp;
-  guint64 exttimestamp;
+  GstClockTime timestamp, out_time;
 
   priv = jitterbuffer->priv;
 
@@ -1015,26 +1002,17 @@ again:
   outbuf = rtp_jitter_buffer_pop (priv->jbuf);
   seqnum = gst_rtp_buffer_get_seq (outbuf);
 
-  /* construct extended RTP timestamp from packet */
-  rtp_time = gst_rtp_buffer_get_timestamp (outbuf);
-  exttimestamp = gst_rtp_buffer_ext_timestamp (&priv->exttimestamp, rtp_time);
-
-  /* if no clock_base was given, take first ts as base */
-  if (priv->clock_base == -1) {
-    GST_DEBUG_OBJECT (jitterbuffer,
-        "no clock base, using exttimestamp %" G_GUINT64_FORMAT, exttimestamp);
-    priv->clock_base = exttimestamp;
-  }
-  /* subtract the base clock time so that we start counting from 0 */
-  exttimestamp -= priv->clock_base;
+  /* get the timestamp, this is already corrected for clock skew by the
+   * jitterbuffer */
+  timestamp = GST_BUFFER_TIMESTAMP (outbuf);
 
   GST_DEBUG_OBJECT (jitterbuffer,
-      "Popped buffer #%d, rtptime %u, exttime %" G_GUINT64_FORMAT
-      ", now %d left", seqnum, rtp_time, exttimestamp,
+      "Popped buffer #%d, timestamp %" GST_TIME_FORMAT ", now %d left",
+      seqnum, GST_TIME_ARGS (timestamp),
       rtp_jitter_buffer_num_packets (priv->jbuf));
 
-  /* convert the RTP timestamp to a gstreamer timestamp. */
-  timestamp = convert_rtptime_to_gsttime (jitterbuffer, exttimestamp);
+  /* apply our latency to the incomming buffer before syncing. */
+  out_time = apply_latency (jitterbuffer, timestamp);
 
   /* If we don't know what the next seqnum should be (== -1) we have to wait
    * because it might be possible that we are not receiving this buffer in-order,
@@ -1044,7 +1022,8 @@ again:
    * determine if we have missing a packet. If we have a missing packet (which
    * must be before this packet) we can wait for it until the deadline for this
    * packet expires. */
-  if (priv->next_seqnum == -1 || priv->next_seqnum != seqnum) {
+  if ((priv->next_seqnum == -1 || priv->next_seqnum != seqnum)
+      && out_time != -1) {
     GstClockID id;
     GstClockTime sync_time;
     GstClockReturn ret;
@@ -1071,10 +1050,10 @@ again:
     }
 
     GST_DEBUG_OBJECT (jitterbuffer, "sync to timestamp %" GST_TIME_FORMAT,
-        GST_TIME_ARGS (timestamp));
+        GST_TIME_ARGS (out_time));
 
     /* prepare for sync against clock */
-    sync_time = timestamp + GST_ELEMENT_CAST (jitterbuffer)->base_time;
+    sync_time = out_time + GST_ELEMENT_CAST (jitterbuffer)->base_time;
 
     /* create an entry for the clock */
     id = priv->clock_id = gst_clock_new_single_shot_id (clock, sync_time);
@@ -1113,9 +1092,8 @@ again:
       }
       goto again;
     }
-    /* After waiting, we might have a better estimate of skew, generate a new
-     * timestamp before pushing out the buffer */
-    timestamp = convert_rtptime_to_gsttime (jitterbuffer, exttimestamp);
+    /* Get new timestamp, latency might have changed */
+    out_time = apply_latency (jitterbuffer, timestamp);
   }
 push_buffer:
   /* check if we are pushing something unexpected */
@@ -1138,7 +1116,7 @@ push_buffer:
   }
 
   /* apply timestamp to buffer now */
-  GST_BUFFER_TIMESTAMP (outbuf) = timestamp;
+  GST_BUFFER_TIMESTAMP (outbuf) = out_time;
 
   /* now we are ready to push the buffer. Save the seqnum and release the lock
    * so the other end can push stuff in the queue again. */
@@ -1147,7 +1125,9 @@ push_buffer:
   JBUF_UNLOCK (priv);
 
   /* push buffer */
-  GST_DEBUG_OBJECT (jitterbuffer, "Pushing buffer %d", seqnum);
+  GST_DEBUG_OBJECT (jitterbuffer,
+      "Pushing buffer %d, timestamp %" GST_TIME_FORMAT, seqnum,
+      GST_TIME_ARGS (out_time));
   result = gst_pad_push (priv->srcpad, outbuf);
   if (result != GST_FLOW_OK)
     goto pause;
@@ -1208,39 +1188,35 @@ gst_rtp_jitter_buffer_query (GstPad * pad, GstQuery * query)
        * own */
       GstClockTime min_latency, max_latency;
       gboolean us_live;
-      GstPad *peer;
       GstClockTime our_latency;
 
-      if ((peer = gst_pad_get_peer (priv->sinkpad))) {
-        if ((res = gst_pad_query (peer, query))) {
-          gst_query_parse_latency (query, &us_live, &min_latency, &max_latency);
+      if ((res = gst_pad_peer_query (priv->sinkpad, query))) {
+        gst_query_parse_latency (query, &us_live, &min_latency, &max_latency);
 
-          GST_DEBUG_OBJECT (jitterbuffer, "Peer latency: min %"
-              GST_TIME_FORMAT " max %" GST_TIME_FORMAT,
-              GST_TIME_ARGS (min_latency), GST_TIME_ARGS (max_latency));
+        GST_DEBUG_OBJECT (jitterbuffer, "Peer latency: min %"
+            GST_TIME_FORMAT " max %" GST_TIME_FORMAT,
+            GST_TIME_ARGS (min_latency), GST_TIME_ARGS (max_latency));
 
-          /* store this so that we can safely sync on the peer buffers. */
-          JBUF_LOCK (priv);
-          priv->peer_latency = min_latency;
-          our_latency = ((guint64) priv->latency_ms) * GST_MSECOND;
-          JBUF_UNLOCK (priv);
+        /* store this so that we can safely sync on the peer buffers. */
+        JBUF_LOCK (priv);
+        priv->peer_latency = min_latency;
+        our_latency = ((guint64) priv->latency_ms) * GST_MSECOND;
+        JBUF_UNLOCK (priv);
 
-          GST_DEBUG_OBJECT (jitterbuffer, "Our latency: %" GST_TIME_FORMAT,
-              GST_TIME_ARGS (our_latency));
+        GST_DEBUG_OBJECT (jitterbuffer, "Our latency: %" GST_TIME_FORMAT,
+            GST_TIME_ARGS (our_latency));
 
-          min_latency += our_latency;
-          /* max_latency can be -1, meaning there is no upper limit for the
-           * latency. */
-          if (max_latency != -1)
-            max_latency += our_latency * GST_MSECOND;
+        min_latency += our_latency;
+        /* max_latency can be -1, meaning there is no upper limit for the
+         * latency. */
+        if (max_latency != -1)
+          max_latency += our_latency;
 
-          GST_DEBUG_OBJECT (jitterbuffer, "Calculated total latency : min %"
-              GST_TIME_FORMAT " max %" GST_TIME_FORMAT,
-              GST_TIME_ARGS (min_latency), GST_TIME_ARGS (max_latency));
+        GST_DEBUG_OBJECT (jitterbuffer, "Calculated total latency : min %"
+            GST_TIME_FORMAT " max %" GST_TIME_FORMAT,
+            GST_TIME_ARGS (min_latency), GST_TIME_ARGS (max_latency));
 
-          gst_query_set_latency (query, TRUE, min_latency, max_latency);
-        }
-        gst_object_unref (peer);
+        gst_query_set_latency (query, TRUE, min_latency, max_latency);
       }
       break;
     }
diff --git a/gst/rtpmanager/rtpjitterbuffer.c b/gst/rtpmanager/rtpjitterbuffer.c
index 3285c879..0f52949e 100644
--- a/gst/rtpmanager/rtpjitterbuffer.c
+++ b/gst/rtpmanager/rtpjitterbuffer.c
@@ -27,6 +27,9 @@
 GST_DEBUG_CATEGORY_STATIC (rtp_jitter_buffer_debug);
 #define GST_CAT_DEFAULT rtp_jitter_buffer_debug
 
+#define MAX_WINDOW	RTP_JITTER_BUFFER_MAX_WINDOW
+#define MAX_TIME	(2 * GST_SECOND)
+
 /* signals and args */
 enum
 {
@@ -61,18 +64,11 @@ rtp_jitter_buffer_class_init (RTPJitterBufferClass * klass)
 static void
 rtp_jitter_buffer_init (RTPJitterBuffer * jbuf)
 {
-  gint i;
-
   jbuf->packets = g_queue_new ();
   jbuf->base_time = -1;
   jbuf->base_rtptime = -1;
   jbuf->ext_rtptime = -1;
-
-  for (i = 0; i < 100; i++) {
-    jbuf->window[i] = 0;
-  }
   jbuf->window_pos = 0;
-  jbuf->window_size = 100;
   jbuf->window_filling = TRUE;
   jbuf->window_min = 0;
   jbuf->skew = 0;
@@ -183,9 +179,17 @@ rtp_jitter_buffer_get_clock_rate (RTPJitterBuffer * jbuf)
  *
  * Both the window and the weighting used for averaging influence the accuracy
  * of the drift estimation. Finding the correct parameters turns out to be a
- * compromise between accuracy and inertia.
+ * compromise between accuracy and inertia. 
+ *
+ * We use a 2 second window or up to 512 data points, which is statistically big
+ * enough to catch spikes (FIXME, detect spikes).
+ * We also use a rather large weighting factor (125) to smoothly adapt. During
+ * startup, when filling the window) we use a parabolic weighting factor, the
+ * more the window is filled, the faster we move to the detected possible skew.
+ *
+ * Returns: @time adjusted with the clock skew.
  */
-static void
+static GstClockTime
 calculate_skew (RTPJitterBuffer * jbuf, guint32 rtptime, GstClockTime time)
 {
   guint64 ext_rtptime;
@@ -193,7 +197,12 @@ calculate_skew (RTPJitterBuffer * jbuf, guint32 rtptime, GstClockTime time)
   gint64 delta;
   gint64 old;
   gint pos, i;
-  GstClockTime gstrtptime;
+  GstClockTime gstrtptime, out_time;
+
+  /* we don't have an arrival timestamp so we can't do skew detection. FIXME, we
+   * should still apply a timestamp based on RTP timestamp and base_time */
+  if (time == -1)
+    return -1;
 
   ext_rtptime = gst_rtp_buffer_ext_timestamp (&jbuf->ext_rtptime, rtptime);
 
@@ -218,25 +227,39 @@ calculate_skew (RTPJitterBuffer * jbuf, guint32 rtptime, GstClockTime time)
 
   if (jbuf->window_filling) {
     /* we are filling the window */
-    GST_DEBUG ("filling %d %" G_GINT64_FORMAT ", diff %" G_GUINT64_FORMAT, pos,
-        delta, send_diff);
+    GST_DEBUG ("filling %d %" G_GINT64_FORMAT ", send_diff %" G_GUINT64_FORMAT,
+        pos, delta, send_diff);
     jbuf->window[pos++] = delta;
     /* calc the min delta we observed */
     if (pos == 1 || delta < jbuf->window_min)
       jbuf->window_min = delta;
 
-    if (send_diff >= 2 * GST_SECOND || pos >= 100) {
+    if (send_diff >= MAX_TIME || pos >= MAX_WINDOW) {
       jbuf->window_size = pos;
 
-      /* window filled, fill window with min */
+      /* window filled */
       GST_DEBUG ("min %" G_GINT64_FORMAT, jbuf->window_min);
-      for (i = 0; i < jbuf->window_size; i++)
-        jbuf->window[i] = jbuf->window_min;
 
-      /* the skew is initially the min */
+      /* the skew is now the min */
       jbuf->skew = jbuf->window_min;
       jbuf->window_filling = FALSE;
     } else {
+      gint perc_time, perc_window, perc;
+
+      /* figure out how much we filled the window, this depends on the amount of
+       * time we have or the max number of points we keep. */
+      perc_time = send_diff * 100 / MAX_TIME;
+      perc_window = pos * 100 / MAX_WINDOW;
+      perc = MAX (perc_time, perc_window);
+
+      /* make a parabolic function, the closer we get to the MAX, the more value
+       * we give to the scaling factor of the new value */
+      perc = perc * perc;
+
+      /* quickly go to the min value when we are filling up, slowly when we are
+       * just starting because we're not sure it's a good value yet. */
+      jbuf->skew =
+          (perc * jbuf->window_min + ((10000 - perc) * jbuf->skew)) / 10000;
       jbuf->window_size = pos + 1;
     }
   } else {
@@ -265,7 +288,7 @@ calculate_skew (RTPJitterBuffer * jbuf, guint32 rtptime, GstClockTime time)
       jbuf->window_min = min;
     }
     /* average the min values */
-    jbuf->skew = (jbuf->window_min + (15 * jbuf->skew)) / 16;
+    jbuf->skew = (jbuf->window_min + (124 * jbuf->skew)) / 125;
     GST_DEBUG ("new min: %" G_GINT64_FORMAT ", skew %" G_GINT64_FORMAT,
         jbuf->window_min, jbuf->skew);
   }
@@ -273,6 +296,17 @@ calculate_skew (RTPJitterBuffer * jbuf, guint32 rtptime, GstClockTime time)
   if (pos >= jbuf->window_size)
     pos = 0;
   jbuf->window_pos = pos;
+
+  /* the output time is defined as the base timestamp plus the RTP time
+   * adjusted for the clock skew .*/
+  out_time = jbuf->base_time + send_diff + jbuf->skew;
+
+  GST_DEBUG ("base %" GST_TIME_FORMAT ", diff %" GST_TIME_FORMAT ", skew %"
+      G_GINT64_FORMAT ", out %" GST_TIME_FORMAT,
+      GST_TIME_ARGS (jbuf->base_time), GST_TIME_ARGS (send_diff),
+      jbuf->skew, GST_TIME_ARGS (out_time));
+
+  return out_time;
 }
 
 static gint
@@ -296,7 +330,7 @@ compare_seqnum (GstBuffer * a, GstBuffer * b, RTPJitterBuffer * jbuf)
  * rtp_jitter_buffer_insert:
  * @jbuf: an #RTPJitterBuffer
  * @buf: a buffer
- * @time: a timestamp when this buffer was received in nanoseconds
+ * @time: a running_time when this buffer was received in nanoseconds
  *
  * Inserts @buf into the packet queue of @jbuf. The sequence number of the
  * packet will be used to sort the packets. This function takes ownerhip of
@@ -327,11 +361,11 @@ rtp_jitter_buffer_insert (RTPJitterBuffer * jbuf, GstBuffer * buf,
     return FALSE;
 
   /* do skew calculation by measuring the difference between rtptime and the
-   * receive time */
-  if (time != -1) {
-    rtptime = gst_rtp_buffer_get_timestamp (buf);
-    calculate_skew (jbuf, rtptime, time);
-  }
+   * receive time, this function will retimestamp @buf with the skew corrected
+   * running time. */
+  rtptime = gst_rtp_buffer_get_timestamp (buf);
+  time = calculate_skew (jbuf, rtptime, time);
+  GST_BUFFER_TIMESTAMP (buf) = time;
 
   if (list)
     g_queue_insert_before (jbuf->packets, list, buf);
@@ -350,7 +384,9 @@ rtp_jitter_buffer_insert (RTPJitterBuffer * jbuf, GstBuffer * buf,
  * rtp_jitter_buffer_pop:
  * @jbuf: an #RTPJitterBuffer
  *
- * Pops the oldest buffer from the packet queue of @jbuf.
+ * Pops the oldest buffer from the packet queue of @jbuf. The popped buffer will
+ * have its timestamp adjusted with the incomming running_time and the detected
+ * clock skew.
  *
  * Returns: a #GstBuffer or %NULL when there was no packet in the queue.
  */
diff --git a/gst/rtpmanager/rtpjitterbuffer.h b/gst/rtpmanager/rtpjitterbuffer.h
index 1db07059..fdc8d488 100644
--- a/gst/rtpmanager/rtpjitterbuffer.h
+++ b/gst/rtpmanager/rtpjitterbuffer.h
@@ -43,6 +43,7 @@ typedef struct _RTPJitterBufferClass RTPJitterBufferClass;
  */
 typedef void (*RTPTailChanged) (RTPJitterBuffer *jbuf, gpointer user_data);
 
+#define RTP_JITTER_BUFFER_MAX_WINDOW 512
 /**
  * RTPJitterBuffer:
  *
@@ -59,7 +60,7 @@ struct _RTPJitterBuffer {
   GstClockTime   base_time;
   GstClockTime   base_rtptime;
   guint64        ext_rtptime;
-  gint64         window[100];
+  gint64         window[RTP_JITTER_BUFFER_MAX_WINDOW];
   guint          window_pos;
   guint          window_size;
   gboolean       window_filling;