gst/rtpmanager/rtpjitterbuffer.c

   1 /* GStreamer
   2  * Copyright (C) <2007> Wim Taymans <wim.taymans@gmail.com>
   3  *
   4  * This library is free software; you can redistribute it and/or
   5  * modify it under the terms of the GNU Library General Public
   6  * License as published by the Free Software Foundation; either
   7  * version 2 of the License, or (at your option) any later version.
   8  *
   9  * This library is distributed in the hope that it will be useful,
  10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12  * Library General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU Library General Public
  15  * License along with this library; if not, write to the
  16  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  17  * Boston, MA 02111-1307, USA.
  18  */
  19 #include <string.h>
  20 #include <stdlib.h>
  21
  22 #include <gst/rtp/gstrtpbuffer.h>
  23 #include <gst/rtp/gstrtcpbuffer.h>
  24
  25 #include "rtpjitterbuffer.h"
  26
  27 GST_DEBUG_CATEGORY_STATIC (rtp_jitter_buffer_debug);
  28 #define GST_CAT_DEFAULT rtp_jitter_buffer_debug
  29
  30 #define MAX_WINDOW      RTP_JITTER_BUFFER_MAX_WINDOW
  31 #define MAX_TIME        (2 * GST_SECOND)
  32
  33 /* signals and args */
  34 enum
  35 {
  36   LAST_SIGNAL
  37 };
  38
  39 enum
  40 {
  41   PROP_0
  42 };
  43
  44 /* GObject vmethods */
  45 static void rtp_jitter_buffer_finalize (GObject * object);
  46
  47 /* static guint rtp_jitter_buffer_signals[LAST_SIGNAL] = { 0 }; */
  48
  49 G_DEFINE_TYPE (RTPJitterBuffer, rtp_jitter_buffer, G_TYPE_OBJECT);
  50
  51 static void
  52 rtp_jitter_buffer_class_init (RTPJitterBufferClass * klass)
  53 {
  54   GObjectClass *gobject_class;
  55
  56   gobject_class = (GObjectClass *) klass;
  57
  58   gobject_class->finalize = rtp_jitter_buffer_finalize;
  59
  60   GST_DEBUG_CATEGORY_INIT (rtp_jitter_buffer_debug, "rtpjitterbuffer", 0,
  61       "RTP Jitter Buffer");
  62 }
  63
  64 static void
  65 rtp_jitter_buffer_init (RTPJitterBuffer * jbuf)
  66 {
  67   jbuf->packets = g_queue_new ();
  68
  69   rtp_jitter_buffer_reset_skew (jbuf);
  70 }
  71
  72 static void
  73 rtp_jitter_buffer_finalize (GObject * object)
  74 {
  75   RTPJitterBuffer *jbuf;
  76
  77   jbuf = RTP_JITTER_BUFFER_CAST (object);
  78
  79   rtp_jitter_buffer_flush (jbuf);
  80   g_queue_free (jbuf->packets);
  81
  82   G_OBJECT_CLASS (rtp_jitter_buffer_parent_class)->finalize (object);
  83 }
  84
  85 /**
  86  * rtp_jitter_buffer_new:
  87  *
  88  * Create an #RTPJitterBuffer.
  89  *
  90  * Returns: a new #RTPJitterBuffer. Use g_object_unref() after usage.
  91  */
  92 RTPJitterBuffer *
  93 rtp_jitter_buffer_new (void)
  94 {
  95   RTPJitterBuffer *jbuf;
  96
  97   jbuf = g_object_new (RTP_TYPE_JITTER_BUFFER, NULL);
  98
  99   return jbuf;
 100 }
 101
 102 void
 103 rtp_jitter_buffer_reset_skew (RTPJitterBuffer * jbuf)
 104 {
 105   jbuf->base_time = -1;
 106   jbuf->base_rtptime = -1;
 107   jbuf->base_extrtp = -1;
 108   jbuf->ext_rtptime = -1;
 109   jbuf->window_pos = 0;
 110   jbuf->window_filling = TRUE;
 111   jbuf->window_min = 0;
 112   jbuf->skew = 0;
 113   jbuf->prev_send_diff = -1;
 114 }
 115
 116 /* For the clock skew we use a windowed low point averaging algorithm as can be
 117  * found in http://www.grame.fr/pub/TR-050601.pdf. The idea is that the jitter is
 118  * composed of:
 119  *
 120  *  J = N + n
 121  *
 122  *   N   : a constant network delay.
 123  *   n   : random added noise. The noise is concentrated around 0
 124  *
 125  * In the receiver we can track the elapsed time at the sender with:
 126  *
 127  *  send_diff(i) = (Tsi - Ts0);
 128  *
 129  *   Tsi : The time at the sender at packet i
 130  *   Ts0 : The time at the sender at the first packet
 131  *
 132  * This is the difference between the RTP timestamp in the first received packet
 133  * and the current packet.
 134  *
 135  * At the receiver we have to deal with the jitter introduced by the network.
 136  *
 137  *  recv_diff(i) = (Tri - Tr0)
 138  *
 139  *   Tri : The time at the receiver at packet i
 140  *   Tr0 : The time at the receiver at the first packet
 141  *
 142  * Both of these values contain a jitter Ji, a jitter for packet i, so we can
 143  * write:
 144  *
 145  *  recv_diff(i) = (Cri + D + ni) - (Cr0 + D + n0))
 146  *
 147  *    Cri    : The time of the clock at the receiver for packet i
 148  *    D + ni : The jitter when receiving packet i
 149  *
 150  * We see that the network delay is irrelevant here as we can elliminate D:
 151  *
 152  *  recv_diff(i) = (Cri + ni) - (Cr0 + n0))
 153  *
 154  * The drift is now expressed as:
 155  *
 156  *  Drift(i) = recv_diff(i) - send_diff(i);
 157  *
 158  * We now keep the W latest values of Drift and find the minimum (this is the
 159  * one with the lowest network jitter and thus the one which is least affected
 160  * by it). We average this lowest value to smooth out the resulting network skew.
 161  *
 162  * Both the window and the weighting used for averaging influence the accuracy
 163  * of the drift estimation. Finding the correct parameters turns out to be a
 164  * compromise between accuracy and inertia.
 165  *
 166  * We use a 2 second window or up to 512 data points, which is statistically big
 167  * enough to catch spikes (FIXME, detect spikes).
 168  * We also use a rather large weighting factor (125) to smoothly adapt. During
 169  * startup, when filling the window, we use a parabolic weighting factor, the
 170  * more the window is filled, the faster we move to the detected possible skew.
 171  *
 172  * Returns: @time adjusted with the clock skew.
 173  */
 174 static GstClockTime
 175 calculate_skew (RTPJitterBuffer * jbuf, guint32 rtptime, GstClockTime time,
 176     guint32 clock_rate)
 177 {
 178   guint64 ext_rtptime;
 179   guint64 send_diff, recv_diff;
 180   gint64 delta;
 181   gint64 old;
 182   gint pos, i;
 183   GstClockTime gstrtptime, out_time;
 184
 185   ext_rtptime = gst_rtp_buffer_ext_timestamp (&jbuf->ext_rtptime, rtptime);
 186
 187   gstrtptime = gst_util_uint64_scale_int (ext_rtptime, GST_SECOND, clock_rate);
 188
 189   /* first time, lock on to time and gstrtptime */
 190   if (G_UNLIKELY (jbuf->base_time == -1))
 191     jbuf->base_time = time;
 192   if (G_UNLIKELY (jbuf->base_rtptime == -1)) {
 193     jbuf->base_rtptime = gstrtptime;
 194     jbuf->base_extrtp = ext_rtptime;
 195   }
 196
 197   if (G_LIKELY (gstrtptime >= jbuf->base_rtptime))
 198     send_diff = gstrtptime - jbuf->base_rtptime;
 199   else {
 200     /* elapsed time at sender, timestamps can go backwards and thus be smaller
 201      * than our base time, take a new base time in that case. */
 202     GST_DEBUG ("backward timestamps at server, taking new base time");
 203     jbuf->base_time = time;
 204     jbuf->base_rtptime = gstrtptime;
 205     jbuf->base_extrtp = ext_rtptime;
 206     send_diff = 0;
 207   }
 208
 209   GST_DEBUG ("extrtp %" G_GUINT64_FORMAT ", gstrtp %" GST_TIME_FORMAT ", base %"
 210       GST_TIME_FORMAT ", send_diff %" GST_TIME_FORMAT, ext_rtptime,
 211       GST_TIME_ARGS (gstrtptime), GST_TIME_ARGS (jbuf->base_rtptime),
 212       GST_TIME_ARGS (send_diff));
 213
 214   /* we don't have an arrival timestamp so we can't do skew detection. we
 215    * should still apply a timestamp based on RTP timestamp and base_time */
 216   if (time == -1)
 217     goto no_skew;
 218
 219   /* elapsed time at receiver, includes the jitter */
 220   recv_diff = time - jbuf->base_time;
 221
 222   GST_DEBUG ("time %" GST_TIME_FORMAT ", base %" GST_TIME_FORMAT ", recv_diff %"
 223       GST_TIME_FORMAT, GST_TIME_ARGS (time), GST_TIME_ARGS (jbuf->base_time),
 224       GST_TIME_ARGS (recv_diff));
 225
 226   /* measure the diff */
 227   delta = ((gint64) recv_diff) - ((gint64) send_diff);
 228
 229   /* if the difference between the sender timeline and the receiver timeline
 230    * changed too quickly we have to resync because the server likely restarted
 231    * its timestamps. */
 232   if (ABS (delta - jbuf->skew) > GST_SECOND) {
 233     GST_DEBUG ("delta %" GST_TIME_FORMAT " too big, reset skew",
 234         delta - jbuf->skew);
 235     jbuf->base_time = time;
 236     jbuf->base_rtptime = gstrtptime;
 237     jbuf->base_extrtp = ext_rtptime;
 238     send_diff = 0;
 239     delta = 0;
 240   }
 241
 242   pos = jbuf->window_pos;
 243
 244   if (G_UNLIKELY (jbuf->window_filling)) {
 245     /* we are filling the window */
 246     GST_DEBUG ("filling %d, delta %" G_GINT64_FORMAT, pos, delta);
 247     jbuf->window[pos++] = delta;
 248     /* calc the min delta we observed */
 249     if (G_UNLIKELY (pos == 1 || delta < jbuf->window_min))
 250       jbuf->window_min = delta;
 251
 252     if (G_UNLIKELY (send_diff >= MAX_TIME || pos >= MAX_WINDOW)) {
 253       jbuf->window_size = pos;
 254
 255       /* window filled */
 256       GST_DEBUG ("min %" G_GINT64_FORMAT, jbuf->window_min);
 257
 258       /* the skew is now the min */
 259       jbuf->skew = jbuf->window_min;
 260       jbuf->window_filling = FALSE;
 261     } else {
 262       gint perc_time, perc_window, perc;
 263
 264       /* figure out how much we filled the window, this depends on the amount of
 265        * time we have or the max number of points we keep. */
 266       perc_time = send_diff * 100 / MAX_TIME;
 267       perc_window = pos * 100 / MAX_WINDOW;
 268       perc = MAX (perc_time, perc_window);
 269
 270       /* make a parabolic function, the closer we get to the MAX, the more value
 271        * we give to the scaling factor of the new value */
 272       perc = perc * perc;
 273
 274       /* quickly go to the min value when we are filling up, slowly when we are
 275        * just starting because we're not sure it's a good value yet. */
 276       jbuf->skew =
 277           (perc * jbuf->window_min + ((10000 - perc) * jbuf->skew)) / 10000;
 278       jbuf->window_size = pos + 1;
 279     }
 280   } else {
 281     /* pick old value and store new value. We keep the previous value in order
 282      * to quickly check if the min of the window changed */
 283     old = jbuf->window[pos];
 284     jbuf->window[pos++] = delta;
 285
 286     if (G_UNLIKELY (delta <= jbuf->window_min)) {
 287       /* if the new value we inserted is smaller or equal to the current min,
 288        * it becomes the new min */
 289       jbuf->window_min = delta;
 290     } else if (G_UNLIKELY (old == jbuf->window_min)) {
 291       gint64 min = G_MAXINT64;
 292
 293       /* if we removed the old min, we have to find a new min */
 294       for (i = 0; i < jbuf->window_size; i++) {
 295         /* we found another value equal to the old min, we can stop searching now */
 296         if (jbuf->window[i] == old) {
 297           min = old;
 298           break;
 299         }
 300         if (jbuf->window[i] < min)
 301           min = jbuf->window[i];
 302       }
 303       jbuf->window_min = min;
 304     }
 305     /* average the min values */
 306     jbuf->skew = (jbuf->window_min + (124 * jbuf->skew)) / 125;
 307     GST_DEBUG ("delta %" G_GINT64_FORMAT ", new min: %" G_GINT64_FORMAT,
 308         delta, jbuf->window_min);
 309   }
 310   /* wrap around in the window */
 311   if (G_UNLIKELY (pos >= jbuf->window_size))
 312     pos = 0;
 313   jbuf->window_pos = pos;
 314
 315 no_skew:
 316   /* the output time is defined as the base timestamp plus the RTP time
 317    * adjusted for the clock skew .*/
 318   out_time = jbuf->base_time + send_diff + jbuf->skew;
 319
 320   GST_DEBUG ("skew %" G_GINT64_FORMAT ", out %" GST_TIME_FORMAT,
 321       jbuf->skew, GST_TIME_ARGS (out_time));
 322
 323   return out_time;
 324 }
 325
 326 /**
 327  * rtp_jitter_buffer_insert:
 328  * @jbuf: an #RTPJitterBuffer
 329  * @buf: a buffer
 330  * @time: a running_time when this buffer was received in nanoseconds
 331  * @clock_rate: the clock-rate of the payload of @buf
 332  * @tail: TRUE when the tail element changed.
 333  *
 334  * Inserts @buf into the packet queue of @jbuf. The sequence number of the
 335  * packet will be used to sort the packets. This function takes ownerhip of
 336  * @buf when the function returns %TRUE.
 337  * @buf should have writable metadata when calling this function.
 338  *
 339  * Returns: %FALSE if a packet with the same number already existed.
 340  */
 341 gboolean
 342 rtp_jitter_buffer_insert (RTPJitterBuffer * jbuf, GstBuffer * buf,
 343     GstClockTime time, guint32 clock_rate, gboolean * tail)
 344 {
 345   GList *list;
 346   guint32 rtptime;
 347   guint16 seqnum;
 348
 349   g_return_val_if_fail (jbuf != NULL, FALSE);
 350   g_return_val_if_fail (buf != NULL, FALSE);
 351
 352   seqnum = gst_rtp_buffer_get_seq (buf);
 353
 354   /* loop the list to skip strictly smaller seqnum buffers */
 355   for (list = jbuf->packets->head; list; list = g_list_next (list)) {
 356     guint16 qseq;
 357     gint gap;
 358
 359     qseq = gst_rtp_buffer_get_seq (GST_BUFFER_CAST (list->data));
 360
 361     /* compare the new seqnum to the one in the buffer */
 362     gap = gst_rtp_buffer_compare_seqnum (seqnum, qseq);
 363
 364     /* we hit a packet with the same seqnum, notify a duplicate */
 365     if (G_UNLIKELY (gap == 0))
 366       goto duplicate;
 367
 368     /* seqnum > qseq, we can stop looking */
 369     if (G_LIKELY (gap < 0))
 370       break;
 371   }
 372
 373   /* do skew calculation by measuring the difference between rtptime and the
 374    * receive time, this function will retimestamp @buf with the skew corrected
 375    * running time. */
 376   rtptime = gst_rtp_buffer_get_timestamp (buf);
 377   time = calculate_skew (jbuf, rtptime, time, clock_rate);
 378   GST_BUFFER_TIMESTAMP (buf) = time;
 379
 380   if (G_LIKELY (list))
 381     g_queue_insert_before (jbuf->packets, list, buf);
 382   else
 383     g_queue_push_tail (jbuf->packets, buf);
 384
 385   /* tail was changed when we did not find a previous packet, we set the return
 386    * flag when requested. */
 387   if (G_UNLIKELY (tail))
 388     *tail = (list == NULL);
 389
 390   return TRUE;
 391
 392   /* ERRORS */
 393 duplicate:
 394   {
 395     GST_WARNING ("duplicate packet %d found", (gint) seqnum);
 396     return FALSE;
 397   }
 398 }
 399
 400 /**
 401  * rtp_jitter_buffer_pop:
 402  * @jbuf: an #RTPJitterBuffer
 403  *
 404  * Pops the oldest buffer from the packet queue of @jbuf. The popped buffer will
 405  * have its timestamp adjusted with the incomming running_time and the detected
 406  * clock skew.
 407  *
 408  * Returns: a #GstBuffer or %NULL when there was no packet in the queue.
 409  */
 410 GstBuffer *
 411 rtp_jitter_buffer_pop (RTPJitterBuffer * jbuf)
 412 {
 413   GstBuffer *buf;
 414
 415   g_return_val_if_fail (jbuf != NULL, FALSE);
 416
 417   buf = g_queue_pop_tail (jbuf->packets);
 418
 419   return buf;
 420 }
 421
 422 /**
 423  * rtp_jitter_buffer_peek:
 424  * @jbuf: an #RTPJitterBuffer
 425  *
 426  * Peek the oldest buffer from the packet queue of @jbuf. Register a callback
 427  * with rtp_jitter_buffer_set_tail_changed() to be notified when an older packet
 428  * was inserted in the queue.
 429  *
 430  * Returns: a #GstBuffer or %NULL when there was no packet in the queue.
 431  */
 432 GstBuffer *
 433 rtp_jitter_buffer_peek (RTPJitterBuffer * jbuf)
 434 {
 435   GstBuffer *buf;
 436
 437   g_return_val_if_fail (jbuf != NULL, FALSE);
 438
 439   buf = g_queue_peek_tail (jbuf->packets);
 440
 441   return buf;
 442 }
 443
 444 /**
 445  * rtp_jitter_buffer_flush:
 446  * @jbuf: an #RTPJitterBuffer
 447  *
 448  * Flush all packets from the jitterbuffer.
 449  */
 450 void
 451 rtp_jitter_buffer_flush (RTPJitterBuffer * jbuf)
 452 {
 453   GstBuffer *buffer;
 454
 455   g_return_if_fail (jbuf != NULL);
 456
 457   while ((buffer = g_queue_pop_head (jbuf->packets)))
 458     gst_buffer_unref (buffer);
 459 }
 460
 461 /**
 462  * rtp_jitter_buffer_num_packets:
 463  * @jbuf: an #RTPJitterBuffer
 464  *
 465  * Get the number of packets currently in "jbuf.
 466  *
 467  * Returns: The number of packets in @jbuf.
 468  */
 469 guint
 470 rtp_jitter_buffer_num_packets (RTPJitterBuffer * jbuf)
 471 {
 472   g_return_val_if_fail (jbuf != NULL, 0);
 473
 474   return jbuf->packets->length;
 475 }
 476
 477 /**
 478  * rtp_jitter_buffer_get_ts_diff:
 479  * @jbuf: an #RTPJitterBuffer
 480  *
 481  * Get the difference between the timestamps of first and last packet in the
 482  * jitterbuffer.
 483  *
 484  * Returns: The difference expressed in the timestamp units of the packets.
 485  */
 486 guint32
 487 rtp_jitter_buffer_get_ts_diff (RTPJitterBuffer * jbuf)
 488 {
 489   guint64 high_ts, low_ts;
 490   GstBuffer *high_buf, *low_buf;
 491   guint32 result;
 492
 493   g_return_val_if_fail (jbuf != NULL, 0);
 494
 495   high_buf = g_queue_peek_head (jbuf->packets);
 496   low_buf = g_queue_peek_tail (jbuf->packets);
 497
 498   if (!high_buf || !low_buf || high_buf == low_buf)
 499     return 0;
 500
 501   high_ts = gst_rtp_buffer_get_timestamp (high_buf);
 502   low_ts = gst_rtp_buffer_get_timestamp (low_buf);
 503
 504   /* it needs to work if ts wraps */
 505   if (high_ts >= low_ts) {
 506     result = (guint32) (high_ts - low_ts);
 507   } else {
 508     result = (guint32) (high_ts + G_MAXUINT32 + 1 - low_ts);
 509   }
 510   return result;
 511 }
 512
 513 /**
 514  * rtp_jitter_buffer_get_sync:
 515  * @jbuf: an #RTPJitterBuffer
 516  * @rtptime: result RTP time
 517  * @timestamp: result GStreamer timestamp
 518  *
 519  * Returns the relation between the RTP timestamp and the GStreamer timestamp
 520  * used for constructing timestamps.
 521  */
 522 void
 523 rtp_jitter_buffer_get_sync (RTPJitterBuffer * jbuf, guint64 * rtptime,
 524     guint64 * timestamp)
 525 {
 526   if (rtptime)
 527     *rtptime = jbuf->base_extrtp;
 528   if (timestamp)
 529     *timestamp = jbuf->base_time + jbuf->skew;
 530 }