}
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;
GstBuffer *outbuf = NULL;
GstFlowReturn result;
guint16 seqnum;
- guint32 rtp_time;
- GstClockTime timestamp;
- guint64 exttimestamp;
+ GstClockTime timestamp, out_time;
priv = jitterbuffer->priv;
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,
* 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;
}
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);
}
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 */
}
/* 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. */
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;
* 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;
}
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
{
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;
*
* 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;
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);
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 {
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);
}
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
* 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
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);
* 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.
*/
projects / platform / upstream / gst-plugins-good.git / commitdiff