Commit 09982c3c authored by Sebastian Dröge's avatar Sebastian Dröge

dataqueue/queuearray: Make public API again

These are actually used outside of coreelements nowadays.
Also hide lots of internals and add padding and documentation.
parent ebbce545
......@@ -45,6 +45,8 @@
<xi:include href="xml/gstbytewriter.xml" />
<xi:include href="xml/gstcollectpads.xml" />
<xi:include href="xml/gsttypefindhelper.xml" />
<xi:include href="xml/gstdataqueue.xml" />
<xi:include href="xml/gstqueuearray.xml" />
</chapter>
<chapter id="gstreamer-control">
......
......@@ -711,6 +711,53 @@ gst_type_find_helper_get_range_ext
<SUBSECTION Private>
</SECTION>
<SECTION>
<FILE>gstdataqueue</FILE>
<TITLE>GstDataQueue</TITLE>
<INCLUDE>gst/base/gstdataqueue.h</INCLUDE>
GstDataQueue
GstDataQueueSize
GstDataQueueCheckFullFunction
GstDataQueueItem
GstDataQueueEmptyCallback
GstDataQueueFullCallback
gst_data_queue_new
gst_data_queue_push
gst_data_queue_pop
gst_data_queue_flush
gst_data_queue_set_flushing
gst_data_queue_drop_head
gst_data_queue_is_full
gst_data_queue_is_empty
gst_data_queue_get_level
gst_data_queue_limits_changed
<SUBSECTION Standard>
GstDataQueueClass
GST_DATA_QUEUE
GST_IS_DATA_QUEUE
GST_TYPE_DATA_QUEUE
GST_DATA_QUEUE_CLASS
GST_IS_DATA_QUEUE_CLASS
<SUBSECTION Private>
gst_data_queue_get_type
</SECTION>
<SECTION>
<FILE>gstqueuearray</FILE>
<TITLE>GstQueueArray</TITLE>
<INCLUDE>gst/base/gstqueuearray.h</INCLUDE>
GstQueueArray
gst_queue_array_new
gst_queue_array_free
gst_queue_array_get_length
gst_queue_array_pop_head
gst_queue_array_peek_head
gst_queue_array_push_tail
gst_queue_array_is_empty
gst_queue_array_drop_element
gst_queue_array_find
</SECTION>
# net
<SECTION>
......
......@@ -12,7 +12,9 @@ libgstbase_@GST_API_VERSION@_la_SOURCES = \
gstbytereader.c \
gstbytewriter.c \
gstcollectpads.c \
gstdataqueue.c \
gstpushsrc.c \
gstqueuearray.c \
gsttypefindhelper.c
libgstbase_@GST_API_VERSION@_la_CFLAGS = $(GST_OBJ_CFLAGS)
......@@ -32,7 +34,9 @@ libgstbase_@GST_API_VERSION@include_HEADERS = \
gstbytereader.h \
gstbytewriter.h \
gstcollectpads.h \
gstdataqueue.h \
gstpushsrc.h \
gstqueuearray.h \
gsttypefindhelper.h
noinst_HEADERS = \
......
......@@ -24,7 +24,6 @@
#define __GST_DATA_QUEUE_H__
#include <gst/gst.h>
#include "gstqueuearray.h"
G_BEGIN_DECLS
#define GST_TYPE_DATA_QUEUE \
......@@ -41,6 +40,7 @@ typedef struct _GstDataQueue GstDataQueue;
typedef struct _GstDataQueueClass GstDataQueueClass;
typedef struct _GstDataQueueSize GstDataQueueSize;
typedef struct _GstDataQueueItem GstDataQueueItem;
typedef struct _GstDataQueuePrivate GstDataQueuePrivate;
/**
* GstDataQueueItem:
......@@ -66,6 +66,9 @@ struct _GstDataQueueItem
/* user supplied destroy function */
GDestroyNotify destroy;
/* < private > */
gpointer _gst_reserved[GST_PADDING];
};
/**
......@@ -113,24 +116,8 @@ struct _GstDataQueue
GObject object;
/*< private >*/
/* the array of data we're keeping our grubby hands on */
GstQueueArray queue;
GstDataQueueSize cur_level; /* size of the queue */
GstDataQueueCheckFullFunction checkfull; /* Callback to check if the queue is full */
gpointer *checkdata;
GMutex qlock; /* lock for queue (vs object lock) */
gboolean waiting_add;
GCond item_add; /* signals buffers now available for reading */
gboolean waiting_del;
GCond item_del; /* signals space now available for writing */
gboolean flushing; /* indicates whether conditions where signalled because
* of external flushing */
GstDataQueueFullCallback fullcallback;
GstDataQueueEmptyCallback emptycallback;
/* gpointer _gst_reserved[GST_PADDING]; */
GstDataQueuePrivate *priv;
gpointer _gst_reserved[GST_PADDING];
};
struct _GstDataQueueClass
......@@ -141,47 +128,32 @@ struct _GstDataQueueClass
void (*empty) (GstDataQueue * queue);
void (*full) (GstDataQueue * queue);
/* gpointer _gst_reserved[GST_PADDING]; */
gpointer _gst_reserved[GST_PADDING];
};
G_GNUC_INTERNAL
GType gst_data_queue_get_type (void);
G_GNUC_INTERNAL
GstDataQueue * gst_data_queue_new (GstDataQueueCheckFullFunction checkfull,
gpointer checkdata) G_GNUC_MALLOC;
G_GNUC_INTERNAL
GstDataQueue * gst_data_queue_new_full (GstDataQueueCheckFullFunction checkfull,
GstDataQueueFullCallback fullcallback,
GstDataQueueEmptyCallback emptycallback,
gpointer checkdata) G_GNUC_MALLOC;
G_GNUC_INTERNAL
gboolean gst_data_queue_push (GstDataQueue * queue, GstDataQueueItem * item);
G_GNUC_INTERNAL
gboolean gst_data_queue_pop (GstDataQueue * queue, GstDataQueueItem ** item);
G_GNUC_INTERNAL
void gst_data_queue_flush (GstDataQueue * queue);
G_GNUC_INTERNAL
void gst_data_queue_set_flushing (GstDataQueue * queue, gboolean flushing);
G_GNUC_INTERNAL
gboolean gst_data_queue_drop_head (GstDataQueue * queue, GType type);
G_GNUC_INTERNAL
gboolean gst_data_queue_is_full (GstDataQueue * queue);
G_GNUC_INTERNAL
gboolean gst_data_queue_is_empty (GstDataQueue * queue);
G_GNUC_INTERNAL
void gst_data_queue_get_level (GstDataQueue * queue, GstDataQueueSize *level);
G_GNUC_INTERNAL
void gst_data_queue_limits_changed (GstDataQueue * queue);
G_END_DECLS
......
......@@ -19,31 +19,82 @@
* Boston, MA 02111-1307, USA.
*/
/**
* SECTION:gstqueuearray
* @short_description: Array based queue object
*
* #GstQueueArray is an object that provides standard queue functionality
* based on an array instead of linked lists. This reduces the overhead
* caused by memory managment by a large factor.
*/
#include <string.h>
#include <gst/gst.h>
#include "gstqueuearray.h"
void
gst_queue_array_init (GstQueueArray * array, guint initial_size)
struct _GstQueueArray
{
/* < private > */
gpointer *array;
guint size;
guint head;
guint tail;
guint length;
};
/**
* gst_queue_array_new:
* @initial_size: Initial size of the new queue
*
* Allocates a new #GstQueueArray object with an initial
* queue size of @initial_size.
*
* Returns: a new #GstQueueArray object
*
* Since: 1.2.0
*/
GstQueueArray *
gst_queue_array_new (guint initial_size)
{
GstQueueArray *array;
array = g_slice_new (GstQueueArray);
array->size = initial_size;
array->array = g_new0 (gpointer, initial_size);
array->head = 0;
array->tail = 0;
array->length = 0;
return array;
}
GstQueueArray *
gst_queue_array_new (guint initial_size)
{
GstQueueArray *array;
array = g_new (GstQueueArray, 1);
gst_queue_array_init (array, initial_size);
return array;
/**
* gst_queue_array_free:
* @array: a #GstQueueArray object
*
* Frees queue @array and all memory associated to it.
*
* Since: 1.2.0
*/
void
gst_queue_array_free (GstQueueArray * array)
{
g_free (array->array);
g_slice_free (GstQueueArray, array);
}
/**
* gst_queue_array_pop_head:
* @array: a #GstQueueArray object
*
* Returns and head of the queue @array and removes
* it from the queue.
*
* Returns: The head of the queue
*
* Since: 1.2.0
*/
gpointer
gst_queue_array_pop_head (GstQueueArray * array)
{
......@@ -59,6 +110,35 @@ gst_queue_array_pop_head (GstQueueArray * array)
return ret;
}
/**
* gst_queue_array_pop_head:
* @array: a #GstQueueArray object
*
* Returns and head of the queue @array and does not
* remove it from the queue.
*
* Returns: The head of the queue
*
* Since: 1.2.0
*/
gpointer
gst_queue_array_peek_head (GstQueueArray * array)
{
/* empty array */
if (G_UNLIKELY (array->length == 0))
return NULL;
return array->array[array->head];
}
/**
* gst_queue_array_push_tail:
* @array: a #GstQueueArray object
* @data: object to push
*
* Pushes @data to the tail of the queue @array.
*
* Since: 1.2.0
*/
void
gst_queue_array_push_tail (GstQueueArray * array, gpointer data)
{
......@@ -102,43 +182,56 @@ gst_queue_array_push_tail (GstQueueArray * array, gpointer data)
array->length++;
}
/**
* gst_queue_array_is_empty:
* @array: a #GstQueueArray object
*
* Checks if the queue @array is empty.
*
* Returns: %TRUE if the queue @array is empty
*
* Since: 1.2.0
*/
gboolean
gst_queue_array_is_empty (GstQueueArray * array)
{
return (array->length == 0);
}
void
gst_queue_array_clear (GstQueueArray * array)
{
g_free (array->array);
}
void
gst_queue_array_free (GstQueueArray * array)
{
gst_queue_array_clear (array);
g_free (array);
}
void
/**
* gst_queue_array_drop_element:
* @array: a #GstQueueArray object
* @idx: index to drop
*
* Drops the queue element at position @idx from queue @array.
*
* Returns: the dropped element
*
* Since: 1.2.0
*/
gpointer
gst_queue_array_drop_element (GstQueueArray * array, guint idx)
{
gpointer element;
if (idx == array->head) {
/* just move the head */
element = array->array[idx];
array->head++;
array->head %= array->size;
return;
return element;
}
if (idx == array->tail - 1) {
/* just move the tail */
element = array->array[idx];
array->tail = (array->tail - 1 + array->size) % array->size;
return;
return element;
}
/* drop the element #idx... and readjust the array */
if (array->head < array->tail) {
/* Make sure it's within the boundaries */
g_assert (array->head < idx && idx <= array->tail);
element = array->array[idx];
/* ends not wrapped */
/* move head-idx to head+1 */
memcpy (&array->array[array->head + 1],
......@@ -147,21 +240,39 @@ gst_queue_array_drop_element (GstQueueArray * array, guint idx)
} else {
/* ends are wrapped */
if (idx < array->tail) {
element = array->array[idx];
/* move idx-tail backwards one */
memcpy (&array->array[idx - 1],
&array->array[idx], (array->tail - idx) * sizeof (gpointer));
array->tail--;
} else if (idx >= array->head) {
element = array->array[idx];
/* move head-idx forwards one */
memcpy (&array->array[array->head],
&array->array[array->head + 1],
(idx - array->head) * sizeof (gpointer));
array->head++;
} else
} else {
g_assert_not_reached ();
element = NULL;
}
}
return element;
}
/**
* gst_queue_array_find:
* @array: a #GstQueueArray object
* @func: comparison function
* @data: data for comparison function
*
* Finds an element in the queue @array by comparing every element
* with @func and returning the index of the found element.
*
* Returns: Index of the found element or -1 if nothing was found.
*
* Since: 1.2.0
*/
guint
gst_queue_array_find (GstQueueArray * array, GCompareFunc func, gpointer data)
{
......@@ -173,3 +284,19 @@ gst_queue_array_find (GstQueueArray * array, GCompareFunc func, gpointer data)
return i;
return -1;
}
/**
* gst_queue_array_get_length:
* @array: a #GstQueueArray object
*
* Returns the length of the queue @array
*
* Returns: the length of the queue @array.
*
* Since: 1.2.0
*/
guint
gst_queue_array_get_length (GstQueueArray * array)
{
return array->length;
}
......@@ -26,36 +26,25 @@
typedef struct _GstQueueArray GstQueueArray;
struct _GstQueueArray
{
gpointer *array;
guint size;
guint head;
guint tail;
guint length;
};
GstQueueArray * gst_queue_array_new (guint initial_size);
G_GNUC_INTERNAL void gst_queue_array_init (GstQueueArray * array,
guint initial_size);
void gst_queue_array_free (GstQueueArray * array);
G_GNUC_INTERNAL void gst_queue_array_clear (GstQueueArray * array);
gpointer gst_queue_array_pop_head (GstQueueArray * array);
gpointer gst_queue_array_peek_head (GstQueueArray * array);
G_GNUC_INTERNAL GstQueueArray * gst_queue_array_new (guint initial_size);
void gst_queue_array_push_tail (GstQueueArray * array,
gpointer data);
G_GNUC_INTERNAL gpointer gst_queue_array_pop_head (GstQueueArray * array);
gboolean gst_queue_array_is_empty (GstQueueArray * array);
G_GNUC_INTERNAL void gst_queue_array_push_tail (GstQueueArray * array,
gpointer data);
gpointer gst_queue_array_drop_element (GstQueueArray * array,
guint idx);
G_GNUC_INTERNAL gboolean gst_queue_array_is_empty (GstQueueArray * array);
guint gst_queue_array_find (GstQueueArray * array,
GCompareFunc func,
gpointer data);
G_GNUC_INTERNAL void gst_queue_array_free (GstQueueArray * array);
G_GNUC_INTERNAL void gst_queue_array_drop_element (GstQueueArray * array,
guint idx);
G_GNUC_INTERNAL guint gst_queue_array_find (GstQueueArray * array,
GCompareFunc func,
gpointer data);
guint gst_queue_array_get_length (GstQueueArray * array);
#endif
......@@ -15,9 +15,7 @@ libgstcoreelements_la_SOURCES = \
gstidentity.c \
gstinputselector.c \
gstoutputselector.c \
gstdataqueue.c \
gstmultiqueue.c \
gstqueuearray.c \
gstqueue.c \
gstqueue2.c \
gsttee.c \
......@@ -43,9 +41,7 @@ noinst_HEADERS = \
gstidentity.h \
gstinputselector.h \
gstoutputselector.h \
gstdataqueue.h \
gstmultiqueue.h \
gstqueuearray.h \
gstqueue.h \
gstqueue2.h \
gsttee.h \
......
......@@ -1983,7 +1983,7 @@ gst_single_queue_new (GstMultiQueue * mqueue, guint id)
sq->mqueue = mqueue;
sq->srcresult = GST_FLOW_FLUSHING;
sq->pushed = FALSE;
sq->queue = gst_data_queue_new_full ((GstDataQueueCheckFullFunction)
sq->queue = gst_data_queue_new ((GstDataQueueCheckFullFunction)
single_queue_check_full,
(GstDataQueueFullCallback) single_queue_overrun_cb,
(GstDataQueueEmptyCallback) single_queue_underrun_cb, sq);
......
......@@ -24,7 +24,7 @@
#define __GST_MULTI_QUEUE_H__
#include <gst/gst.h>
#include "gstdataqueue.h"
#include <gst/base/gstdataqueue.h>
G_BEGIN_DECLS
......
......@@ -92,7 +92,7 @@ GST_DEBUG_CATEGORY_STATIC (queue_dataflow);
queue->cur_level.time, \
queue->min_threshold.time, \
queue->max_size.time, \
queue->queue.length)
gst_queue_array_get_length (queue->queue))
/* Queue signals and args */
enum
......@@ -417,7 +417,7 @@ gst_queue_init (GstQueue * queue)
g_cond_init (&queue->item_add);
g_cond_init (&queue->item_del);
gst_queue_array_init (&queue->queue, DEFAULT_MAX_SIZE_BUFFERS * 3 / 2);
queue->queue = gst_queue_array_new (DEFAULT_MAX_SIZE_BUFFERS * 3 / 2);
queue->sinktime = GST_CLOCK_TIME_NONE;
queue->srctime = GST_CLOCK_TIME_NONE;
......@@ -440,13 +440,13 @@ gst_queue_finalize (GObject * object)
GST_DEBUG_OBJECT (queue, "finalizing queue");
while (!gst_queue_array_is_empty (&queue->queue)) {
data = gst_queue_array_pop_head (&queue->queue);
while (!gst_queue_array_is_empty (queue->queue)) {
data = gst_queue_array_pop_head (queue->queue);
/* FIXME: if it's a query, shouldn't we unref that too? */
if (!GST_IS_QUERY (data))
gst_mini_object_unref (data);
}
gst_queue_array_clear (&queue->queue);
gst_queue_array_free (queue->queue);
g_mutex_clear (&queue->qlock);
g_cond_clear (&queue->item_add);
......@@ -556,8 +556,8 @@ gst_queue_locked_flush (GstQueue * queue)
{
GstMiniObject *data;
while (!gst_queue_array_is_empty (&queue->queue)) {
data = gst_queue_array_pop_head (&queue->queue);
while (!gst_queue_array_is_empty (queue->queue)) {
data = gst_queue_array_pop_head (queue->queue);
/* Then lose another reference because we are supposed to destroy that
data when flushing */
if (!GST_IS_QUERY (data))
......@@ -590,7 +590,7 @@ gst_queue_locked_enqueue_buffer (GstQueue * queue, gpointer item)
apply_buffer (queue, buffer, &queue->sink_segment, TRUE, TRUE);
if (item)
gst_queue_array_push_tail (&queue->queue, item);
gst_queue_array_push_tail (queue->queue, item);
GST_QUEUE_SIGNAL_ADD (queue);
}
......@@ -611,7 +611,7 @@ gst_queue_locked_enqueue_event (GstQueue * queue, gpointer item)
case GST_EVENT_SEGMENT:
apply_segment (queue, event, &queue->sink_segment, TRUE);
/* if the queue is empty, apply sink segment on the source */
if (queue->queue.length == 0) {
if (gst_queue_array_is_empty (queue->queue)) {
GST_CAT_LOG_OBJECT (queue_dataflow, queue, "Apply segment on srcpad");
apply_segment (queue, event, &queue->src_segment, FALSE);
queue->newseg_applied_to_src = TRUE;
......@@ -625,7 +625,7 @@ gst_queue_locked_enqueue_event (GstQueue * queue, gpointer item)
}
if (item)
gst_queue_array_push_tail (&queue->queue, item);
gst_queue_array_push_tail (queue->queue, item);
GST_QUEUE_SIGNAL_ADD (queue);
}
......@@ -635,7 +635,7 @@ gst_queue_locked_dequeue (GstQueue * queue)
{
GstMiniObject *item;
item = gst_queue_array_pop_head (&queue->queue);
item = gst_queue_array_pop_head (queue->queue);
if (item == NULL)
goto no_item;
......@@ -792,9 +792,9 @@ gst_queue_handle_sink_query (GstPad * pad, GstObject * parent, GstQuery * query)
GST_QUEUE_MUTEX_LOCK_CHECK (queue, out_flushing);
GST_LOG_OBJECT (queue, "queuing query %p (%s)", query,
GST_QUERY_TYPE_NAME (query));
gst_queue_array_push_tail (&queue->queue, query);
gst_queue_array_push_tail (queue->queue, query);
GST_QUEUE_SIGNAL_ADD (queue);
while (queue->queue.length != 0) {
while (!gst_queue_array_is_empty (queue->queue)) {
/* for as long as the queue has items, we know the query is
* not handled yet */
GST_QUEUE_WAIT_DEL_CHECK (queue, out_flushing);
......@@ -822,14 +822,14 @@ gst_queue_is_empty (GstQueue * queue)
{
GstMiniObject *head;
if (queue->queue.length == 0)
if (gst_queue_array_is_empty (queue->queue))
return TRUE;
/* Only consider the queue empty if the minimum thresholds
* are not reached and data is at the queue head. Otherwise
* we would block forever on serialized queries.
*/
head = queue->queue.array[queue->queue.head];
head = gst_queue_array_peek_head (queue->queue);
if (!GST_IS_BUFFER (head) && !GST_IS_BUFFER_LIST (head))
return FALSE;
......
......@@ -25,7 +25,7 @@
#define __GST_QUEUE_H__
#include <gst/gst.h>
#include "gstqueuearray.h"
#include <gst/base/gstqueuearray.h>
G_BEGIN_DECLS
......@@ -108,7 +108,7 @@ struct _GstQueue {
gboolean eos;