mirrors/gstreamer
mirrors/gstreamer
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gstreamer.git synced 2024-11-15 22:01:27 +00:00
Code Issues 1 Projects Releases Wiki Activity

queue2: do not post buffering messages holding the lock

Browse source 
It might cause deadlocks to post messages while holding the queue2
lock. To avoid this a new boolean flag is set whenever a new
buffering percent is found. The message is posted after the lock
is released.

To make sure the buffering messages are posted in the right order, messages
are posted holding another lock. This prevents 2 threads trying to post
messages at the same time.

https://bugzilla.gnome.org/show_bug.cgi?id=736969
This commit is contained in:
Thiago Santos 2014-09-19 12:02:46 -03:00
parent 23205f04f1
commit 7630ada89a
2 changed files with 57 additions and 31 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

81
plugins/elements/gstqueue2.c
View file

@ -206,6 +206,16 @@ enum
} \
} G_STMT_END
#define SET_PERCENT(q, perc) G_STMT_START { \
if (perc != q->buffering_percent) { \
q->buffering_percent = perc; \
q->percent_changed = TRUE; \
GST_DEBUG_OBJECT (q, "buffering %d percent", perc); \
get_buffering_stats (q, perc, &q->mode, &q->avg_in, &q->avg_out, \
&q->buffering_left); \
} \
} G_STMT_END
#define _do_init \
GST_DEBUG_CATEGORY_INIT (queue_debug, "queue2", 0, "queue element"); \
GST_DEBUG_CATEGORY_INIT (queue_dataflow, "queue2_dataflow", 0, \
@ -254,6 +264,7 @@ static gboolean gst_queue2_is_filled (GstQueue2 * queue);
static void update_cur_level (GstQueue2 * queue, GstQueue2Range * range);
static void update_in_rates (GstQueue2 * queue);
static void gst_queue2_post_buffering (GstQueue2 * queue);
typedef enum
{
@ -449,6 +460,7 @@ gst_queue2_init (GstQueue2 * queue)
g_cond_init (&queue->query_handled);
queue->last_query = FALSE;
g_mutex_init (&queue->buffering_post_lock);
queue->buffering_percent = 100;
/* tempfile related */
@ -482,6 +494,7 @@ gst_queue2_finalize (GObject * object)
queue->last_query = FALSE;
g_queue_clear (&queue->queue);
g_mutex_clear (&queue->qlock);
g_mutex_clear (&queue->buffering_post_lock);
g_cond_clear (&queue->item_add);
g_cond_clear (&queue->item_del);
g_cond_clear (&queue->query_handled);
@ -868,11 +881,36 @@ get_buffering_stats (GstQueue2 * queue, gint percent, GstBufferingMode * mode,
}
}
static void
gst_queue2_post_buffering (GstQueue2 * queue)
{
GstMessage *msg = NULL;
g_mutex_lock (&queue->buffering_post_lock);
GST_QUEUE2_MUTEX_LOCK (queue);
if (queue->percent_changed) {
gint percent = queue->buffering_percent;
queue->percent_changed = FALSE;
GST_DEBUG_OBJECT (queue, "Going to post buffering: %d%%", percent);
msg = gst_message_new_buffering (GST_OBJECT_CAST (queue), percent);
gst_message_set_buffering_stats (msg, queue->mode, queue->avg_in,
queue->avg_out, queue->buffering_left);
}
GST_QUEUE2_MUTEX_UNLOCK (queue);
if (msg != NULL)
gst_element_post_message (GST_ELEMENT_CAST (queue), msg);
g_mutex_unlock (&queue->buffering_post_lock);
}
static void
update_buffering (GstQueue2 * queue)
{
gint percent;
gboolean post = FALSE;
/* Ensure the variables used to calculate buffering state are up-to-date. */
if (queue->current)
@ -883,42 +921,19 @@ update_buffering (GstQueue2 * queue)
return;
if (queue->is_buffering) {
post = TRUE;
/* if we were buffering see if we reached the high watermark */
if (percent >= queue->high_percent)
queue->is_buffering = FALSE;
SET_PERCENT (queue, percent);
} else {
/* we were not buffering, check if we need to start buffering if we drop
* below the low threshold */
if (percent < queue->low_percent) {
queue->is_buffering = TRUE;
post = TRUE;
SET_PERCENT (queue, percent);
}
}
if (post) {
if (percent == queue->buffering_percent)
post = FALSE;
else
queue->buffering_percent = percent;
}
if (post) {
GstMessage *message;
GstBufferingMode mode;
gint avg_in, avg_out;
gint64 buffering_left;
get_buffering_stats (queue, percent, &mode, &avg_in, &avg_out,
&buffering_left);
message = gst_message_new_buffering (GST_OBJECT_CAST (queue),
(gint) percent);
gst_message_set_buffering_stats (message, mode,
avg_in, avg_out, buffering_left);
gst_element_post_message (GST_ELEMENT_CAST (queue), message);
}
}
static void
@ -2310,6 +2325,7 @@ gst_queue2_handle_sink_event (GstPad * pad, GstObject * parent,
goto out_eos;
gst_queue2_locked_enqueue (queue, event, GST_QUEUE2_ITEM_TYPE_EVENT);
GST_QUEUE2_MUTEX_UNLOCK (queue);
gst_queue2_post_buffering (queue);
} else {
/* non-serialized events are passed upstream. */
ret = gst_pad_push_event (queue->srcpad, event);
@ -2386,6 +2402,7 @@ gst_queue2_handle_sink_query (GstPad * pad, GstObject * parent,
res = FALSE;
}
GST_QUEUE2_MUTEX_UNLOCK (queue);
gst_queue2_post_buffering (queue);
} else {
res = gst_pad_query_default (pad, parent, query);
}
@ -2486,6 +2503,7 @@ gst_queue2_chain_buffer_or_buffer_list (GstQueue2 * queue,
/* put buffer in queue now */
gst_queue2_locked_enqueue (queue, item, item_type);
GST_QUEUE2_MUTEX_UNLOCK (queue);
gst_queue2_post_buffering (queue);
return GST_FLOW_OK;
@ -2626,6 +2644,7 @@ next:
item_type == GST_QUEUE2_ITEM_TYPE_BUFFER ||
item_type == GST_QUEUE2_ITEM_TYPE_BUFFER_LIST);
GST_QUEUE2_MUTEX_UNLOCK (queue);
gst_queue2_post_buffering (queue);
if (item_type == GST_QUEUE2_ITEM_TYPE_BUFFER) {
GstBuffer *buffer;
@ -2745,6 +2764,7 @@ gst_queue2_loop (GstPad * pad)
goto out_flushing;
GST_QUEUE2_MUTEX_UNLOCK (queue);
gst_queue2_post_buffering (queue);
return;
@ -3105,6 +3125,7 @@ gst_queue2_get_range (GstPad * pad, GstObject * parent, guint64 offset,
/* FIXME - function will block when the range is not yet available */
ret = gst_queue2_create_read (queue, offset, length, buffer);
GST_QUEUE2_MUTEX_UNLOCK (queue);
gst_queue2_post_buffering (queue);
return ret;
@ -3430,13 +3451,10 @@ gst_queue2_set_property (GObject * object,
case PROP_USE_BUFFERING:
queue->use_buffering = g_value_get_boolean (value);
if (!queue->use_buffering && queue->is_buffering) {
GstMessage *msg = gst_message_new_buffering (GST_OBJECT_CAST (queue),
100);
GST_DEBUG_OBJECT (queue, "Disabled buffering while buffering, "
"posting 100%% message");
SET_PERCENT (queue, 100);
queue->is_buffering = FALSE;
gst_element_post_message (GST_ELEMENT_CAST (queue), msg);
}
if (queue->use_buffering) {
@ -3468,6 +3486,7 @@ gst_queue2_set_property (GObject * object,
}
GST_QUEUE2_MUTEX_UNLOCK (queue);
gst_queue2_post_buffering (queue);
}
static void

7
plugins/elements/gstqueue2.h
View file

@ -153,6 +153,13 @@ struct _GstQueue2
guint8 * ring_buffer;
volatile gint downstream_may_block;
GstBufferingMode mode;
gint64 buffering_left;
gint avg_in;
gint avg_out;
gboolean percent_changed;
GMutex buffering_post_lock; /* assures only one posted at a time */
};
struct _GstQueue2Class