2 * Async Jitter Queue based on g_async_queue
3 * This code is GST RTP smart and deals with timestamps
5 * Farsight Voice+Video library
6 * Copyright 2007 Collabora Ltd,
7 * Copyright 2007 Nokia Corporation
8 * @author: Philippe Khalaf <philippe.khalaf@collabora.co.uk>.
10 * This is an async queue that has a buffering mecanism based on the set low
11 * and high threshold. When the lower threshold is reached, the queue will
12 * fill itself up until the higher threshold is reached before allowing any
13 * pops to occur. This allows a jitterbuffer of at least min threshold items
17 /* GLIB - Library of useful routines for C programming
18 * Copyright (C) 1995-1997 Peter Mattis, Spencer Kimball and Josh MacDonald
20 * GAsyncQueue: asynchronous queue implementation, based on Gqueue.
21 * Copyright (C) 2000 Sebastian Wilhelmi; University of Karlsruhe
23 * This library is free software; you can redistribute it and/or
24 * modify it under the terms of the GNU Lesser General Public
25 * License as published by the Free Software Foundation; either
26 * version 2 of the License, or (at your option) any later version.
28 * This library is distributed in the hope that it will be useful,
29 * but WITHOUT ANY WARRANTY; without even the implied warranty of
30 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
31 * Lesser General Public License for more details.
33 * You should have received a copy of the GNU Lesser General Public
34 * License along with this library; if not, write to the
35 * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
36 * Boston, MA 02111-1307, USA.
45 #include "async_jitter_queue.h"
48 #include <gst/rtp/gstrtpbuffer.h>
50 #define DEFAULT_LOW_THRESHOLD 0.1
51 #define DEFAULT_HIGH_THRESHOLD 0.9
53 struct _AsyncJitterQueue
58 guint waiting_threads;
61 gfloat high_threshold;
62 guint32 max_queue_length;
64 gboolean pop_flushing;
65 gboolean pop_blocking;
67 guint32 tail_buffer_duration;
71 * async_jitter_queue_new:
73 * Creates a new asynchronous queue with the initial reference count of 1.
75 * Return value: the new #AsyncJitterQueue.
78 async_jitter_queue_new (void)
80 AsyncJitterQueue *retval = g_new (AsyncJitterQueue, 1);
82 retval->mutex = g_mutex_new ();
83 retval->cond = g_cond_new ();
84 retval->queue = g_queue_new ();
85 retval->waiting_threads = 0;
86 retval->ref_count = 1;
87 retval->low_threshold = DEFAULT_LOW_THRESHOLD;
88 retval->high_threshold = DEFAULT_HIGH_THRESHOLD;
89 retval->buffering = TRUE; /* we need to buffer initially */
90 retval->pop_flushing = TRUE;
91 retval->pop_blocking = TRUE;
92 retval->pops_remaining = 0;
93 retval->tail_buffer_duration = 0;
97 /* checks buffering state and wakes up waiting pops */
99 signal_waiting_threads (AsyncJitterQueue * queue)
101 if (async_jitter_queue_length_ts_units_unlocked (queue) >=
102 queue->high_threshold * queue->max_queue_length) {
103 queue->buffering = FALSE;
106 if (queue->waiting_threads > 0) {
107 if (!queue->buffering) {
108 g_cond_signal (queue->cond);
114 * async_jitter_queue_ref:
115 * @queue: a #AsyncJitterQueue.
117 * Increases the reference count of the asynchronous @queue by 1. You
118 * do not need to hold the lock to call this function.
120 * Returns: the @queue that was passed in (since 2.6)
123 async_jitter_queue_ref (AsyncJitterQueue * queue)
125 g_return_val_if_fail (queue, NULL);
126 g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
128 g_atomic_int_inc (&queue->ref_count);
134 * async_jitter_queue_ref_unlocked:
135 * @queue: a #AsyncJitterQueue.
137 * Increases the reference count of the asynchronous @queue by 1.
140 async_jitter_queue_ref_unlocked (AsyncJitterQueue * queue)
142 g_return_if_fail (queue);
143 g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
145 g_atomic_int_inc (&queue->ref_count);
149 * async_jitter_queue_set_low_threshold:
150 * @queue: a #AsyncJitterQueue.
151 * @threshold: the lower threshold (fraction of max size)
153 * Sets the low threshold on the queue. This threshold indicates the minimum
154 * number of items allowed in the queue before we refill it up to the set
158 async_jitter_queue_set_low_threshold (AsyncJitterQueue * queue,
161 g_return_if_fail (queue);
162 g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
164 queue->low_threshold = threshold;
168 * async_jitter_queue_set_max_threshold:
169 * @queue: a #AsyncJitterQueue.
170 * @threshold: the higher threshold (fraction of max size)
172 * Sets the high threshold on the queue. This threshold indicates the amount of
173 * items to fill in the queue before releasing any blocking pop calls. This
174 * blocking mecanism is only triggered when we reach the low threshold and must
178 async_jitter_queue_set_high_threshold (AsyncJitterQueue * queue,
181 g_return_if_fail (queue);
182 g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
184 queue->high_threshold = threshold;
187 /* set the maximum queue length in RTP timestamp units */
189 async_jitter_queue_set_max_queue_length (AsyncJitterQueue * queue,
192 g_return_if_fail (queue);
193 g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
195 queue->max_queue_length = max_length;
199 async_jitter_queue_get_g_queue (AsyncJitterQueue * queue)
201 g_return_val_if_fail (queue, NULL);
207 calculate_ts_diff (guint32 high_ts, guint32 low_ts)
209 /* it needs to work if ts wraps */
210 if (high_ts >= low_ts) {
211 return high_ts - low_ts;
213 return high_ts + G_MAXUINT32 + 1 - low_ts;
217 /* this function returns the length of the queue in timestamp units. It will
218 * also add the duration of the last buffer in the queue */
219 /* FIXME This function wrongly assumes that there are no missing packets inside
220 * the buffer, in reality it needs to check for gaps and subsctract those from
223 async_jitter_queue_length_ts_units_unlocked (AsyncJitterQueue * queue)
231 g_return_val_if_fail (queue, 0);
233 if (queue->queue->length < 2) {
237 tail = g_queue_peek_tail (queue->queue);
238 head = g_queue_peek_head (queue->queue);
240 if (!GST_IS_BUFFER (tail) || !GST_IS_BUFFER (head))
243 tail_ts = gst_rtp_buffer_get_timestamp (tail);
244 head_ts = gst_rtp_buffer_get_timestamp (head);
246 ret = calculate_ts_diff (head_ts, tail_ts);
248 /* let's add the duration of the tail buffer */
249 ret += queue->tail_buffer_duration;
255 * async_jitter_queue_unref_and_unlock:
256 * @queue: a #AsyncJitterQueue.
258 * Decreases the reference count of the asynchronous @queue by 1 and
259 * releases the lock. This function must be called while holding the
260 * @queue's lock. If the reference count went to 0, the @queue will be
261 * destroyed and the memory allocated will be freed.
264 async_jitter_queue_unref_and_unlock (AsyncJitterQueue * queue)
266 g_return_if_fail (queue);
267 g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
269 g_mutex_unlock (queue->mutex);
270 async_jitter_queue_unref (queue);
274 * async_jitter_queue_unref:
275 * @queue: a #AsyncJitterQueue.
277 * Decreases the reference count of the asynchronous @queue by 1. If
278 * the reference count went to 0, the @queue will be destroyed and the
279 * memory allocated will be freed. So you are not allowed to use the
280 * @queue afterwards, as it might have disappeared. You do not need to
281 * hold the lock to call this function.
284 async_jitter_queue_unref (AsyncJitterQueue * queue)
286 g_return_if_fail (queue);
287 g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
289 if (g_atomic_int_dec_and_test (&queue->ref_count)) {
290 g_return_if_fail (queue->waiting_threads == 0);
291 g_mutex_free (queue->mutex);
293 g_cond_free (queue->cond);
294 g_queue_free (queue->queue);
300 * async_jitter_queue_lock:
301 * @queue: a #AsyncJitterQueue.
303 * Acquires the @queue's lock. After that you can only call the
304 * <function>async_jitter_queue_*_unlocked()</function> function variants on that
305 * @queue. Otherwise it will deadlock.
308 async_jitter_queue_lock (AsyncJitterQueue * queue)
310 g_return_if_fail (queue);
311 g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
313 g_mutex_lock (queue->mutex);
317 * async_jitter_queue_unlock:
318 * @queue: a #AsyncJitterQueue.
320 * Releases the queue's lock.
323 async_jitter_queue_unlock (AsyncJitterQueue * queue)
325 g_return_if_fail (queue);
326 g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
328 g_mutex_unlock (queue->mutex);
332 * async_jitter_queue_push:
333 * @queue: a #AsyncJitterQueue.
334 * @data: @data to push into the @queue.
336 * Pushes the @data into the @queue. @data must not be %NULL.
339 async_jitter_queue_push (AsyncJitterQueue * queue, gpointer data)
341 g_return_if_fail (queue);
342 g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
343 g_return_if_fail (data);
345 g_mutex_lock (queue->mutex);
346 async_jitter_queue_push_unlocked (queue, data);
347 g_mutex_unlock (queue->mutex);
351 * async_jitter_queue_push_unlocked:
352 * @queue: a #AsyncJitterQueue.
353 * @data: @data to push into the @queue.
355 * Pushes the @data into the @queue. @data must not be %NULL. This
356 * function must be called while holding the @queue's lock.
359 async_jitter_queue_push_unlocked (AsyncJitterQueue * queue, gpointer data)
361 g_return_if_fail (queue);
362 g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
363 g_return_if_fail (data);
365 g_queue_push_head (queue->queue, data);
367 signal_waiting_threads (queue);
371 * async_jitter_queue_push_sorted:
372 * @queue: a #AsyncJitterQueue
373 * @data: the @data to push into the @queue
374 * @func: the #GCompareDataFunc is used to sort @queue. This function
375 * is passed two elements of the @queue. The function should return
376 * 0 if they are equal, a negative value if the first element
377 * should be higher in the @queue or a positive value if the first
378 * element should be lower in the @queue than the second element.
379 * @user_data: user data passed to @func.
381 * Inserts @data into @queue using @func to determine the new
384 * This function requires that the @queue is sorted before pushing on
387 * This function will lock @queue before it sorts the queue and unlock
388 * it when it is finished.
390 * For an example of @func see async_jitter_queue_sort().
395 async_jitter_queue_push_sorted (AsyncJitterQueue * queue,
396 gpointer data, GCompareDataFunc func, gpointer user_data)
398 g_return_val_if_fail (queue != NULL, FALSE);
401 g_mutex_lock (queue->mutex);
402 ret = async_jitter_queue_push_sorted_unlocked (queue, data, func, user_data);
403 g_mutex_unlock (queue->mutex);
409 * async_jitter_queue_push_sorted_unlocked:
410 * @queue: a #AsyncJitterQueue
411 * @data: the @data to push into the @queue
412 * @func: the #GCompareDataFunc is used to sort @queue. This function
413 * is passed two elements of the @queue. The function should return
414 * 0 if they are equal, a negative value if the first element
415 * should be higher in the @queue or a positive value if the first
416 * element should be lower in the @queue than the second element.
417 * @user_data: user data passed to @func.
419 * Inserts @data into @queue using @func to determine the new
422 * This function requires that the @queue is sorted before pushing on
425 * If @GCompareDataFunc returns 0, this function does not insert @data and
428 * This function is called while holding the @queue's lock.
430 * For an example of @func see async_jitter_queue_sort().
435 async_jitter_queue_push_sorted_unlocked (AsyncJitterQueue * queue,
436 gpointer data, GCompareDataFunc func, gpointer user_data)
439 gint func_ret = TRUE;
441 g_return_val_if_fail (queue != NULL, FALSE);
443 list = queue->queue->head;
444 while (list && (func_ret = func (list->data, data, user_data)) < 0)
451 g_queue_insert_before (queue->queue, list, data);
453 g_queue_push_tail (queue->queue, data);
456 signal_waiting_threads (queue);
461 async_jitter_queue_insert_after_unlocked (AsyncJitterQueue * queue,
462 GList * sibling, gpointer data)
464 g_return_if_fail (queue != NULL);
466 g_queue_insert_before (queue->queue, sibling, data);
468 signal_waiting_threads (queue);
472 async_jitter_queue_pop_intern_unlocked (AsyncJitterQueue * queue)
475 GstBuffer *tail_buffer = NULL;
477 if (queue->pop_flushing)
480 while (queue->pop_blocking) {
481 queue->waiting_threads++;
482 g_cond_wait (queue->cond, queue->mutex);
483 queue->waiting_threads--;
484 if (queue->pop_flushing)
488 if (async_jitter_queue_length_ts_units_unlocked (queue) <=
489 queue->low_threshold * queue->max_queue_length
490 && queue->pops_remaining == 0) {
491 if (!queue->buffering) {
492 queue->buffering = TRUE;
493 queue->pops_remaining = queue->queue->length;
495 while (!g_queue_peek_tail (queue->queue) || queue->pop_blocking) {
496 queue->waiting_threads++;
497 g_cond_wait (queue->cond, queue->mutex);
498 queue->waiting_threads--;
499 if (queue->pop_flushing)
505 retval = g_queue_pop_tail (queue->queue);
506 if (queue->pops_remaining)
507 queue->pops_remaining--;
509 tail_buffer = g_queue_peek_tail (queue->queue);
511 if (!GST_IS_BUFFER (tail_buffer) || !GST_IS_BUFFER (retval)) {
512 queue->tail_buffer_duration = 0;
513 } else if (gst_rtp_buffer_get_seq (tail_buffer)
514 - gst_rtp_buffer_get_seq (retval) == 1) {
515 queue->tail_buffer_duration =
516 calculate_ts_diff (gst_rtp_buffer_get_timestamp (tail_buffer),
517 gst_rtp_buffer_get_timestamp (retval));
519 /* There is a sequence number gap -> we can't calculate the duration
520 * let's just set it to 0 */
521 queue->tail_buffer_duration = 0;
531 * async_jitter_queue_pop:
532 * @queue: a #AsyncJitterQueue.
534 * Pops data from the @queue. This function blocks until data become
535 * available. If pop is disabled, tis function return NULL.
537 * Return value: data from the queue.
540 async_jitter_queue_pop (AsyncJitterQueue * queue)
544 g_return_val_if_fail (queue, NULL);
545 g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
547 g_mutex_lock (queue->mutex);
548 retval = async_jitter_queue_pop_intern_unlocked (queue);
549 g_mutex_unlock (queue->mutex);
555 * async_jitter_queue_pop_unlocked:
556 * @queue: a #AsyncJitterQueue.
558 * Pops data from the @queue. This function blocks until data become
559 * available. This function must be called while holding the @queue's
562 * Return value: data from the queue.
565 async_jitter_queue_pop_unlocked (AsyncJitterQueue * queue)
567 g_return_val_if_fail (queue, NULL);
568 g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, NULL);
570 return async_jitter_queue_pop_intern_unlocked (queue);
574 * async_jitter_queue_length:
575 * @queue: a #AsyncJitterQueue.
577 * Returns the length of the queue
578 * Return value: the length of the @queue.
581 async_jitter_queue_length (AsyncJitterQueue * queue)
585 g_return_val_if_fail (queue, 0);
586 g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, 0);
588 g_mutex_lock (queue->mutex);
589 retval = queue->queue->length;
590 g_mutex_unlock (queue->mutex);
596 * async_jitter_queue_length_unlocked:
597 * @queue: a #AsyncJitterQueue.
599 * Returns the length of the queue.
601 * Return value: the length of the @queue.
604 async_jitter_queue_length_unlocked (AsyncJitterQueue * queue)
606 g_return_val_if_fail (queue, 0);
607 g_return_val_if_fail (g_atomic_int_get (&queue->ref_count) > 0, 0);
609 return queue->queue->length;
613 * async_jitter_queue_set_flushing_unlocked:
614 * @queue: a #AsyncJitterQueue.
615 * @free_func: a function to call to free the elements
616 * @user_data: user data passed to @free_func
618 * This function is used to set/unset flushing. If flushing is set any
619 * waiting/blocked pops will be unblocked. Any subsequent calls to pop will
620 * return NULL. Flushing is set by default.
623 async_jitter_queue_set_flushing_unlocked (AsyncJitterQueue * queue,
624 GFunc free_func, gpointer user_data)
626 g_return_if_fail (queue);
627 g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
629 queue->pop_flushing = TRUE;
630 /* let's unblock any remaining pops */
631 if (queue->waiting_threads > 0)
632 g_cond_broadcast (queue->cond);
633 /* free data from queue */
634 g_queue_foreach (queue->queue, free_func, user_data);
638 * async_jitter_queue_unset_flushing_unlocked:
639 * @queue: a #AsyncJitterQueue.
640 * @free_func: a function to call to free the elements
641 * @user_data: user data passed to @free_func
643 * This function is used to set/unset flushing. If flushing is set any
644 * waiting/blocked pops will be unblocked. Any subsequent calls to pop will
645 * return NULL. Flushing is set by default.
648 async_jitter_queue_unset_flushing_unlocked (AsyncJitterQueue * queue)
650 g_return_if_fail (queue);
651 g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
653 queue->pop_flushing = FALSE;
654 /* let's unblock any remaining pops */
655 if (queue->waiting_threads > 0)
656 g_cond_broadcast (queue->cond);
660 * async_jitter_queue_set_blocking_unlocked:
661 * @queue: a #AsyncJitterQueue.
662 * @enabled: a boolean to enable/disable blocking
664 * This function is used to enable/disable blocking. If blocking is enabled any
665 * pops will be blocked until the queue is unblocked. The queue is blocked by
669 async_jitter_queue_set_blocking_unlocked (AsyncJitterQueue * queue,
672 g_return_if_fail (queue);
673 g_return_if_fail (g_atomic_int_get (&queue->ref_count) > 0);
675 queue->pop_blocking = blocking;
676 /* let's unblock any remaining pops */
677 if (queue->waiting_threads > 0)
678 g_cond_broadcast (queue->cond);