From 2b1f49a26e1879b41329e9198f65826fdce960a4 Mon Sep 17 00:00:00 2001 From: Wim Taymans Date: Fri, 28 Sep 2007 11:17:35 +0000 Subject: [PATCH] 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. --- gst/rtpmanager/gstrtpjitterbuffer.c | 108 +++++++++++----------------- gst/rtpmanager/rtpjitterbuffer.c | 86 +++++++++++++++------- gst/rtpmanager/rtpjitterbuffer.h | 3 +- 3 files changed, 105 insertions(+), 92 deletions(-) diff --git a/gst/rtpmanager/gstrtpjitterbuffer.c b/gst/rtpmanager/gstrtpjitterbuffer.c index d27c2aa1ce..2155df8911 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 3285c879d1..0f52949e00 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 1db070595f..fdc8d48868 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;