mirrors/gstreamer
mirrors/gstreamer
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gstreamer.git synced 2024-11-30 21:51:09 +00:00
Code Issues 1 Projects Releases Wiki Activity

inputselector: Properly sync when changing streams

Browse source 
This adds properties to use the clock time for deciding when
to drop buffers for inactive pads and a property to buffer all
not rendered buffers for the active pad to allow pad switching
without losing any buffers at all.
This commit is contained in:
Andre Moreira Magalhaes (andrunko) 2012-05-28 14:29:00 -03:00 committed by Sebastian Dröge
parent 732238b66b
commit cd798a0a52
2 changed files with 558 additions and 126 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

670
plugins/elements/gstinputselector.c
View file

@ -61,9 +61,30 @@
#include "gst/glib-compat-private.h"
#define DEBUG_CACHED_BUFFERS 0
GST_DEBUG_CATEGORY_STATIC (input_selector_debug);
#define GST_CAT_DEFAULT input_selector_debug
#define GST_TYPE_INPUT_SELECTOR_SYNC_MODE (gst_input_selector_sync_mode_get_type())
static GType
gst_input_selector_sync_mode_get_type (void)
{
static GType type = 0;
static const GEnumValue data[] = {
{GST_INPUT_SELECTOR_SYNC_MODE_ACTIVE_SEGMENT,
"Sync using the current active segment",
"active-segment"},
{GST_INPUT_SELECTOR_SYNC_MODE_CLOCK, "Sync using the clock", "clock"},
{0, NULL, NULL},
};
if (!type) {
type = g_enum_register_static ("GstInputSelectorSyncMode", data);
}
return type;
}
#if GLIB_CHECK_VERSION(2, 26, 0)
#define NOTIFY_MUTEX_LOCK()
#define NOTIFY_MUTEX_UNLOCK()
@ -98,10 +119,14 @@ enum
PROP_0,
PROP_N_PADS,
PROP_ACTIVE_PAD,
PROP_SYNC_STREAMS
PROP_SYNC_STREAMS,
PROP_SYNC_MODE,
PROP_CACHE_BUFFERS
};
#define DEFAULT_SYNC_STREAMS FALSE
#define DEFAULT_SYNC_MODE GST_INPUT_SELECTOR_SYNC_MODE_ACTIVE_SEGMENT
#define DEFAULT_CACHE_BUFFERS FALSE
#define DEFAULT_PAD_ALWAYS_OK TRUE
@ -123,6 +148,8 @@ enum
};
static guint gst_input_selector_signals[LAST_SIGNAL] = { 0 };
static void gst_input_selector_active_pad_changed (GstInputSelector * sel,
GParamSpec * pspec, gpointer user_data);
static inline gboolean gst_input_selector_is_active_sinkpad (GstInputSelector *
sel, GstPad * pad);
static GstPad *gst_input_selector_activate_sinkpad (GstInputSelector * sel,
@ -145,6 +172,7 @@ static GstPad *gst_input_selector_get_linked_pad (GstInputSelector * sel,
typedef struct _GstSelectorPad GstSelectorPad;
typedef struct _GstSelectorPadClass GstSelectorPadClass;
typedef struct _GstSelectorPadCachedBuffer GstSelectorPadCachedBuffer;
struct _GstSelectorPad
{
@ -157,12 +185,23 @@ struct _GstSelectorPad
gboolean discont; /* after switching we create a discont */
gboolean flushing; /* set after flush-start and before flush-stop */
gboolean always_ok;
gboolean segment_update;
GstSegment segment; /* the current segment on the pad */
GstTagList *tags; /* last tags received on the pad */
gboolean sending_cached_buffers;
GQueue *cached_buffers;
gboolean segment_pending;
};
struct _GstSelectorPadCachedBuffer
{
GstBuffer *buffer;
GstSegment segment;
gboolean segment_update;
};
struct _GstSelectorPadClass
{
GstPadClass parent;
@ -180,6 +219,10 @@ static GstPadClass *selector_pad_parent_class = NULL;
static gint64 gst_selector_pad_get_running_time (GstSelectorPad * pad);
static void gst_selector_pad_reset (GstSelectorPad * pad);
static void gst_selector_pad_cache_buffer (GstSelectorPad * selpad,
GstBuffer * buffer);
static void gst_selector_pad_free_cached_buffers (GstSelectorPad * selpad);
static gboolean gst_selector_pad_event (GstPad * pad, GstEvent * event);
static GstCaps *gst_selector_pad_getcaps (GstPad * pad);
static gboolean gst_selector_pad_acceptcaps (GstPad * pad, GstCaps * caps);
@ -262,6 +305,7 @@ gst_selector_pad_finalize (GObject * object)
if (pad->tags)
gst_tag_list_free (pad->tags);
gst_selector_pad_free_cached_buffers (pad);
G_OBJECT_CLASS (selector_pad_parent_class)->finalize (object);
}
@ -335,12 +379,13 @@ gst_selector_pad_get_running_time (GstSelectorPad * pad)
}
GST_OBJECT_UNLOCK (pad);
GST_DEBUG_OBJECT (pad, "running time: %" GST_TIME_FORMAT,
GST_TIME_ARGS (ret));
GST_DEBUG_OBJECT (pad, "running time: %" GST_TIME_FORMAT
" segment: %" GST_SEGMENT_FORMAT, GST_TIME_ARGS (ret), &pad->segment);
return ret;
}
/* must be called with the SELECTOR_LOCK */
static void
gst_selector_pad_reset (GstSelectorPad * pad)
{
@ -353,9 +398,57 @@ gst_selector_pad_reset (GstSelectorPad * pad)
pad->discont = FALSE;
pad->flushing = FALSE;
gst_segment_init (&pad->segment, GST_FORMAT_UNDEFINED);
pad->segment_update = FALSE;
pad->sending_cached_buffers = FALSE;
gst_selector_pad_free_cached_buffers (pad);
GST_OBJECT_UNLOCK (pad);
}
static GstSelectorPadCachedBuffer *
gst_selector_pad_new_cached_buffer (GstSelectorPad * selpad, GstBuffer * buffer)
{
GstSelectorPadCachedBuffer *cached_buffer =
g_slice_new (GstSelectorPadCachedBuffer);
cached_buffer->buffer = buffer;
cached_buffer->segment = selpad->segment;
cached_buffer->segment_update = selpad->segment_update;
return cached_buffer;
}
static void
gst_selector_pad_free_cached_buffer (GstSelectorPadCachedBuffer * cached_buffer)
{
gst_buffer_unref (cached_buffer->buffer);
g_slice_free (GstSelectorPadCachedBuffer, cached_buffer);
}
/* must be called with the SELECTOR_LOCK */
static void
gst_selector_pad_cache_buffer (GstSelectorPad * selpad, GstBuffer * buffer)
{
GST_DEBUG_OBJECT (selpad, "Caching buffer %p", buffer);
if (!selpad->cached_buffers)
selpad->cached_buffers = g_queue_new ();
g_queue_push_tail (selpad->cached_buffers,
gst_selector_pad_new_cached_buffer (selpad, buffer));
}
/* must be called with the SELECTOR_LOCK */
static void
gst_selector_pad_free_cached_buffers (GstSelectorPad * selpad)
{
GstSelectorPadCachedBuffer *cached_buffer;
if (!selpad->cached_buffers)
return;
GST_DEBUG_OBJECT (selpad, "Freeing cached buffers");
while ((cached_buffer = g_queue_pop_head (selpad->cached_buffers)))
gst_selector_pad_free_cached_buffer (cached_buffer);
g_queue_free (selpad->cached_buffers);
selpad->cached_buffers = NULL;
}
/* strictly get the linked pad from the sinkpad. If the pad is active we return
* the srcpad else we return NULL */
static GstIterator *
@ -396,13 +489,11 @@ gst_selector_pad_event (GstPad * pad, GstEvent * event)
return FALSE;
}
selpad = GST_SELECTOR_PAD_CAST (pad);
GST_DEBUG_OBJECT (selpad, "received event %" GST_PTR_FORMAT, event);
GST_INPUT_SELECTOR_LOCK (sel);
prev_active_sinkpad = sel->active_sinkpad;
active_sinkpad = gst_input_selector_activate_sinkpad (sel, pad);
/* only forward if we are dealing with the active sinkpad */
forward = (pad == active_sinkpad);
GST_INPUT_SELECTOR_UNLOCK (sel);
if (prev_active_sinkpad != active_sinkpad && pad == active_sinkpad) {
@ -411,19 +502,22 @@ gst_selector_pad_event (GstPad * pad, GstEvent * event)
NOTIFY_MUTEX_UNLOCK ();
}
GST_INPUT_SELECTOR_LOCK (sel);
active_sinkpad = gst_input_selector_activate_sinkpad (sel, pad);
/* only forward if we are dealing with the active sinkpad */
forward = (pad == active_sinkpad);
switch (GST_EVENT_TYPE (event)) {
case GST_EVENT_FLUSH_START:
/* Unblock the pad if it's waiting */
GST_INPUT_SELECTOR_LOCK (sel);
selpad->flushing = TRUE;
GST_INPUT_SELECTOR_BROADCAST (sel);
GST_INPUT_SELECTOR_UNLOCK (sel);
break;
case GST_EVENT_FLUSH_STOP:
GST_INPUT_SELECTOR_LOCK (sel);
gst_selector_pad_reset (selpad);
sel->pending_close = FALSE;
GST_INPUT_SELECTOR_UNLOCK (sel);
GST_INPUT_SELECTOR_BROADCAST (sel);
break;
case GST_EVENT_NEWSEGMENT:
{
@ -441,11 +535,11 @@ gst_selector_pad_event (GstPad * pad, GstEvent * event)
"%" G_GINT64_FORMAT " -- %" G_GINT64_FORMAT ", time %"
G_GINT64_FORMAT, update, rate, arate, format, start, stop, time);
GST_INPUT_SELECTOR_LOCK (sel);
GST_OBJECT_LOCK (selpad);
gst_segment_set_newsegment_full (&selpad->segment, update,
rate, arate, format, start, stop, time);
GST_OBJECT_UNLOCK (selpad);
/* keep the segment update value to be used when resending cached
* buffers if in sync-mode */
selpad->segment_update = update;
/* If we aren't forwarding the event because the pad is not the
* active_sinkpad, then set the flag on the pad
@ -455,8 +549,6 @@ gst_selector_pad_event (GstPad * pad, GstEvent * event)
*/
if (!forward)
selpad->segment_pending = TRUE;
GST_INPUT_SELECTOR_UNLOCK (sel);
break;
}
case GST_EVENT_TAG:
@ -465,7 +557,6 @@ gst_selector_pad_event (GstPad * pad, GstEvent * event)
gst_event_parse_tag (event, &tags);
GST_OBJECT_LOCK (selpad);
oldtags = selpad->tags;
newtags = gst_tag_list_merge (oldtags, tags, GST_TAG_MERGE_REPLACE);
@ -473,7 +564,6 @@ gst_selector_pad_event (GstPad * pad, GstEvent * event)
if (oldtags)
gst_tag_list_free (oldtags);
GST_DEBUG_OBJECT (pad, "received tags %" GST_PTR_FORMAT, newtags);
GST_OBJECT_UNLOCK (selpad);
g_object_notify (G_OBJECT (selpad), "tags");
break;
@ -484,7 +574,7 @@ gst_selector_pad_event (GstPad * pad, GstEvent * event)
if (forward) {
selpad->eos_sent = TRUE;
} else {
GstSelectorPad *tmp;
GstSelectorPad *active_selpad;
/* If the active sinkpad is in EOS state but EOS
* was not sent downstream this means that the pad
@ -492,18 +582,16 @@ gst_selector_pad_event (GstPad * pad, GstEvent * event)
* the previously active pad got EOS after it was
* active
*/
GST_INPUT_SELECTOR_LOCK (sel);
active_sinkpad = gst_input_selector_activate_sinkpad (sel, pad);
tmp = GST_SELECTOR_PAD (active_sinkpad);
forward = (tmp->eos && !tmp->eos_sent);
tmp->eos_sent = TRUE;
GST_INPUT_SELECTOR_UNLOCK (sel);
active_selpad = GST_SELECTOR_PAD (active_sinkpad);
forward = (active_selpad->eos && !active_selpad->eos_sent);
active_selpad->eos_sent = TRUE;
}
GST_DEBUG_OBJECT (pad, "received EOS");
break;
default:
break;
}
GST_INPUT_SELECTOR_UNLOCK (sel);
if (forward) {
GST_DEBUG_OBJECT (pad, "forwarding event");
res = gst_pad_push_event (sel->srcpad, event);
@ -602,7 +690,7 @@ not_active:
/* unselected pad, perform fallback alloc or return unlinked when
* asked */
GST_OBJECT_LOCK (selpad);
if (selpad->always_ok || !active_pad_pushed) {
if (selpad->always_ok || !active_pad_pushed || sel->cache_buffers) {
GST_DEBUG_OBJECT (pad, "Not selected, performing fallback allocation");
*buf = NULL;
result = GST_FLOW_OK;
@ -629,105 +717,267 @@ gst_input_selector_wait (GstInputSelector * self, GstSelectorPad * pad)
return self->flushing;
}
/* must be called with the SELECTOR_LOCK, will block until the running time
/* must be called without the SELECTOR_LOCK, will wait until the running time
* of the active pad is after this pad or return TRUE when flushing */
static gboolean
gst_input_selector_wait_running_time (GstInputSelector * sel,
GstSelectorPad * pad, GstBuffer * buf)
GstSelectorPad * selpad, GstBuffer * buf)
{
GstPad *active_sinkpad;
GstSelectorPad *active_selpad;
GstSegment *seg, *active_seg;
GstClockTime running_time, active_running_time = GST_CLOCK_TIME_NONE;
GstSegment *seg;
seg = &pad->segment;
active_sinkpad =
gst_input_selector_activate_sinkpad (sel, GST_PAD_CAST (pad));
active_selpad = GST_SELECTOR_PAD_CAST (active_sinkpad);
active_seg = &active_selpad->segment;
/* We can only sync if the segments are in time format or
* if the active pad had no newsegment event yet */
if (seg->format != GST_FORMAT_TIME ||
(active_seg->format != GST_FORMAT_TIME
&& active_seg->format != GST_FORMAT_UNDEFINED))
return FALSE;
GST_DEBUG_OBJECT (selpad, "entering wait for buffer %p", buf);
/* If we have no valid timestamp we can't sync this buffer */
if (!GST_BUFFER_TIMESTAMP_IS_VALID (buf))
if (!GST_BUFFER_TIMESTAMP_IS_VALID (buf)) {
GST_DEBUG_OBJECT (selpad, "leaving wait for buffer with "
"invalid timestamp");
return FALSE;
}
running_time = GST_BUFFER_TIMESTAMP (buf);
/* If possible try to get the running time at the end of the buffer */
if (GST_BUFFER_DURATION_IS_VALID (buf))
running_time += GST_BUFFER_DURATION (buf);
if (running_time > seg->stop)
running_time = seg->stop;
running_time =
gst_segment_to_running_time (seg, GST_FORMAT_TIME, running_time);
/* If this is outside the segment don't sync */
if (running_time == -1)
return FALSE;
/* Get active pad's running time, if no configured segment yet keep at -1 */
if (active_seg->format == GST_FORMAT_TIME)
active_running_time =
gst_segment_to_running_time (active_seg, GST_FORMAT_TIME,
active_seg->last_stop);
seg = &selpad->segment;
/* Wait until
* a) this is the active pad
* b) the pad or the selector is flushing
* c) the selector is not blocked
* d) the active pad has no running time or the active
* pad's running time is before this running time
* e) the active pad has a non-time segment
* f) the active pad changed and has not pushed anything
* d) the buffer running time is before the current running time
* (either active-seg or clock, depending on sync-mode)
*/
while (pad != active_selpad && !sel->flushing && !pad->flushing
&& active_selpad->pushed && (sel->blocked || active_running_time == -1
|| running_time >= active_running_time)) {
if (!sel->blocked)
GST_DEBUG_OBJECT (pad,
"Waiting for active streams to advance. %" GST_TIME_FORMAT " >= %"
GST_TIME_FORMAT, GST_TIME_ARGS (running_time),
GST_TIME_ARGS (active_running_time));
GST_INPUT_SELECTOR_WAIT (sel);
GST_INPUT_SELECTOR_LOCK (sel);
while (TRUE) {
GstPad *active_sinkpad;
GstSelectorPad *active_selpad;
GstClock *clock;
gint64 cur_running_time;
GstClockTime running_time;
/* Get new active pad, it might have changed */
active_sinkpad =
gst_input_selector_activate_sinkpad (sel, GST_PAD_CAST (pad));
gst_input_selector_activate_sinkpad (sel, GST_PAD_CAST (selpad));
active_selpad = GST_SELECTOR_PAD_CAST (active_sinkpad);
active_seg = &active_selpad->segment;
/* If the active segment is configured but not to time format
* we can't do any syncing at all */
if (active_seg->format != GST_FORMAT_TIME
&& active_seg->format != GST_FORMAT_UNDEFINED)
running_time = GST_BUFFER_TIMESTAMP (buf);
/* If possible try to get the running time at the end of the buffer */
if (GST_BUFFER_DURATION_IS_VALID (buf))
running_time += GST_BUFFER_DURATION (buf);
/* Only use the segment to convert to running time if the segment is
* in TIME format, otherwise do our best to try to sync */
if (seg->format == GST_FORMAT_TIME && GST_CLOCK_TIME_IS_VALID (seg->stop)) {
if (running_time > seg->stop) {
running_time = seg->stop;
}
running_time =
gst_segment_to_running_time (seg, GST_FORMAT_TIME, running_time);
/* If this is outside the segment don't sync */
if (running_time == -1) {
GST_INPUT_SELECTOR_UNLOCK (sel);
return FALSE;
}
}
cur_running_time = GST_CLOCK_TIME_NONE;
if (sel->sync_mode == GST_INPUT_SELECTOR_SYNC_MODE_CLOCK) {
clock = gst_element_get_clock (GST_ELEMENT_CAST (sel));
if (clock) {
GstClockTime base_time;
cur_running_time = gst_clock_get_time (clock);
base_time = gst_element_get_base_time (GST_ELEMENT_CAST (sel));
if (base_time <= cur_running_time)
cur_running_time -= base_time;
else
cur_running_time = 0;
}
} else {
GstSegment *active_seg;
active_seg = &active_selpad->segment;
/* Get active pad's running time, if no configured segment yet keep at -1 */
if (active_seg->format == GST_FORMAT_TIME)
cur_running_time = gst_segment_to_running_time (active_seg,
GST_FORMAT_TIME, active_seg->last_stop);
}
if (selpad != active_selpad && !sel->flushing && !selpad->flushing &&
(sel->cache_buffers || active_selpad->pushed) &&
(sel->blocked || cur_running_time == -1
|| running_time >= cur_running_time)) {
if (!sel->blocked)
GST_DEBUG_OBJECT (selpad,
"Waiting for active streams to advance. %" GST_TIME_FORMAT " >= %"
GST_TIME_FORMAT, GST_TIME_ARGS (running_time),
GST_TIME_ARGS (cur_running_time));
GST_INPUT_SELECTOR_WAIT (sel);
} else {
GST_INPUT_SELECTOR_UNLOCK (sel);
break;
/* Get the new active pad running time */
if (active_seg->format == GST_FORMAT_TIME)
active_running_time =
gst_segment_to_running_time (active_seg, GST_FORMAT_TIME,
active_seg->last_stop);
else
active_running_time = -1;
if (!sel->blocked)
GST_DEBUG_OBJECT (pad,
"Waited for active streams to advance. %" GST_TIME_FORMAT " >= %"
GST_TIME_FORMAT, GST_TIME_ARGS (running_time),
GST_TIME_ARGS (active_running_time));
}
}
/* Return TRUE if the selector or the pad is flushing */
return (sel->flushing || pad->flushing);
return (sel->flushing || selpad->flushing);
}
#if DEBUG_CACHED_BUFFERS
static void
gst_input_selector_debug_cached_buffers (GstInputSelector * sel)
{
GList *walk;
for (walk = GST_ELEMENT_CAST (sel)->sinkpads; walk; walk = g_list_next (walk)) {
GstSelectorPad *selpad;
GString *timestamps;
gchar *str;
int i;
selpad = GST_SELECTOR_PAD_CAST (walk->data);
if (!selpad->cached_buffers) {
GST_DEBUG_OBJECT (selpad, "Cached buffers timestamps: <none>");
continue;
}
timestamps = g_string_new ("Cached buffers timestamps:");
for (i = 0; i < selpad->cached_buffers->length; ++i) {
GstSelectorPadCachedBuffer *cached_buffer;
cached_buffer = g_queue_peek_nth (selpad->cached_buffers, i);
g_string_append_printf (timestamps, " %" GST_TIME_FORMAT,
GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (cached_buffer->buffer)));
}
str = g_string_free (timestamps, FALSE);
GST_DEBUG_OBJECT (selpad, str);
g_free (str);
}
}
#endif
/* must be called with the SELECTOR_LOCK */
static void
gst_input_selector_cleanup_old_cached_buffers (GstInputSelector * sel,
GstPad * pad)
{
GstSelectorPad *selpad;
GstSegment *seg;
GstClock *clock;
gint64 cur_running_time;
GList *walk;
selpad = GST_SELECTOR_PAD_CAST (pad);
seg = &selpad->segment;
cur_running_time = GST_CLOCK_TIME_NONE;
if (sel->sync_mode == GST_INPUT_SELECTOR_SYNC_MODE_CLOCK) {
clock = gst_element_get_clock (GST_ELEMENT_CAST (sel));
if (clock) {
GstClockTime base_time;
cur_running_time = gst_clock_get_time (clock);
base_time = gst_element_get_base_time (GST_ELEMENT_CAST (sel));
if (base_time <= cur_running_time)
cur_running_time -= base_time;
else
cur_running_time = 0;
}
} else {
GstPad *active_sinkpad;
GstSelectorPad *active_selpad;
GstSegment *active_seg;
active_sinkpad = gst_input_selector_activate_sinkpad (sel, pad);
active_selpad = GST_SELECTOR_PAD_CAST (active_sinkpad);
active_seg = &active_selpad->segment;
/* Get active pad's running time, if no configured segment yet keep at -1 */
if (active_seg->format == GST_FORMAT_TIME)
cur_running_time = gst_segment_to_running_time (active_seg,
GST_FORMAT_TIME, active_seg->last_stop);
}
if (!GST_CLOCK_TIME_IS_VALID (cur_running_time))
return;
GST_DEBUG_OBJECT (sel, "Cleaning up old cached buffers");
for (walk = GST_ELEMENT_CAST (sel)->sinkpads; walk; walk = g_list_next (walk)) {
GstSelectorPad *selpad = GST_SELECTOR_PAD_CAST (walk->data);
GstSelectorPadCachedBuffer *cached_buffer;
GSList *maybe_remove;
guint queue_position;
if (!selpad->cached_buffers)
continue;
maybe_remove = NULL;
queue_position = 0;
while ((cached_buffer = g_queue_peek_nth (selpad->cached_buffers,
queue_position))) {
GstBuffer *buffer = cached_buffer->buffer;
GstClockTime running_time;
GSList *l;
/* If we have no valid timestamp we can't sync this buffer */
if (!GST_BUFFER_TIMESTAMP_IS_VALID (buffer)) {
maybe_remove = g_slist_append (maybe_remove, cached_buffer);
queue_position = g_slist_length (maybe_remove);
continue;
}
/* the buffer is still valid if its duration is valid and the
* timestamp + duration is >= time, or if its duration is invalid
* and the timestamp is >= time */
running_time = GST_BUFFER_TIMESTAMP (buffer);
/* If possible try to get the running time at the end of the buffer */
if (GST_BUFFER_DURATION_IS_VALID (buffer))
running_time += GST_BUFFER_DURATION (buffer);
/* Only use the segment to convert to running time if the segment is
* in TIME format, otherwise do our best to try to sync */
if (seg->format == GST_FORMAT_TIME && GST_CLOCK_TIME_IS_VALID (seg->stop)) {
if (running_time > seg->stop) {
running_time = seg->stop;
}
running_time =
gst_segment_to_running_time (seg, GST_FORMAT_TIME, running_time);
}
GST_DEBUG_OBJECT (selpad,
"checking if buffer %p running time=%" GST_TIME_FORMAT
" >= stream time=%" GST_TIME_FORMAT, buffer,
GST_TIME_ARGS (running_time), GST_TIME_ARGS (cur_running_time));
if (running_time >= cur_running_time) {
break;
}
GST_DEBUG_OBJECT (selpad, "Removing old cached buffer %p", buffer);
g_queue_pop_nth (selpad->cached_buffers, queue_position);
gst_selector_pad_free_cached_buffer (cached_buffer);
for (l = maybe_remove; l != NULL; l = g_slist_next (l)) {
/* A buffer after some invalid buffers was removed, it means the invalid buffers
* are old, lets also remove them */
cached_buffer = l->data;
g_queue_remove (selpad->cached_buffers, cached_buffer);
gst_selector_pad_free_cached_buffer (cached_buffer);
}
g_slist_free (maybe_remove);
maybe_remove = NULL;
queue_position = 0;
}
g_slist_free (maybe_remove);
maybe_remove = NULL;
if (g_queue_is_empty (selpad->cached_buffers)) {
g_queue_free (selpad->cached_buffers);
selpad->cached_buffers = NULL;
}
}
#if DEBUG_CACHED_BUFFERS
gst_input_selector_debug_cached_buffers (sel);
#endif
}
static GstFlowReturn
gst_selector_pad_chain (GstPad * pad, GstBuffer * buf)
@ -746,10 +996,16 @@ gst_selector_pad_chain (GstPad * pad, GstBuffer * buf)
selpad = GST_SELECTOR_PAD_CAST (pad);
seg = &selpad->segment;
GST_DEBUG_OBJECT (selpad,
"entering chain for buf %p with timestamp %" GST_TIME_FORMAT, buf,
GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buf)));
GST_INPUT_SELECTOR_LOCK (sel);
/* wait or check for flushing */
if (gst_input_selector_wait (sel, selpad))
if (gst_input_selector_wait (sel, selpad)) {
GST_INPUT_SELECTOR_UNLOCK (sel);
goto flushing;
}
GST_LOG_OBJECT (pad, "getting active pad");
@ -758,13 +1014,61 @@ gst_selector_pad_chain (GstPad * pad, GstBuffer * buf)
/* In sync mode wait until the active pad has advanced
* after the running time of the current buffer */
if (sel->sync_streams && active_sinkpad != pad) {
if (gst_input_selector_wait_running_time (sel, selpad, buf))
goto flushing;
}
if (sel->sync_streams) {
/* call chain for each cached buffer if we are not the active pad
* or if we are the active pad but didn't push anything yet. */
if (active_sinkpad != pad || !selpad->pushed) {
/* no need to check for sel->cache_buffers as selpad->cached_buffers
* will only be valid if cache_buffers is TRUE */
if (selpad->cached_buffers && !selpad->sending_cached_buffers) {
GstSelectorPadCachedBuffer *cached_buffer;
GstSegment saved_segment;
gboolean saved_segment_update;
/* Might have changed while waiting */
active_sinkpad = gst_input_selector_activate_sinkpad (sel, pad);
saved_segment = selpad->segment;
saved_segment_update = selpad->segment_update;
selpad->sending_cached_buffers = TRUE;
while (!sel->flushing && !selpad->flushing &&
(cached_buffer = g_queue_pop_head (selpad->cached_buffers))) {
GST_DEBUG_OBJECT (pad, "Cached buffers found, "
"invoking chain for cached buffer %p", cached_buffer->buffer);
selpad->segment = cached_buffer->segment;
selpad->segment_update = cached_buffer->segment_update;
selpad->segment_pending = TRUE;
GST_INPUT_SELECTOR_UNLOCK (sel);
gst_selector_pad_chain (pad, cached_buffer->buffer);
GST_INPUT_SELECTOR_LOCK (sel);
/* we may have cleaned up the queue in the meantime because of
* old buffers */
if (!selpad->cached_buffers) {
break;
}
}
selpad->sending_cached_buffers = FALSE;
/* all cached buffers sent, restore segment for current buffer */
selpad->segment = saved_segment;
selpad->segment_update = saved_segment_update;
selpad->segment_pending = TRUE;
/* Might have changed while calling chain for cached buffers */
active_sinkpad = gst_input_selector_activate_sinkpad (sel, pad);
}
}
if (active_sinkpad != pad) {
GST_INPUT_SELECTOR_UNLOCK (sel);
if (gst_input_selector_wait_running_time (sel, selpad, buf))
goto flushing;
GST_INPUT_SELECTOR_LOCK (sel);
}
/* Might have changed while waiting */
active_sinkpad = gst_input_selector_activate_sinkpad (sel, pad);
}
/* update the segment on the srcpad */
start_time = GST_BUFFER_TIMESTAMP (buf);
@ -807,17 +1111,27 @@ gst_selector_pad_chain (GstPad * pad, GstBuffer * buf)
/* if we have a pending segment, push it out now */
if (G_UNLIKELY (selpad->segment_pending)) {
if (G_UNLIKELY (seg->format == GST_FORMAT_UNDEFINED)) {
GST_ERROR_OBJECT (pad, "Buffers arrived before NEWSEGMENT event");
GST_ERROR_OBJECT (pad, "buffers arrived before NEWSEGMENT event");
} else {
gboolean update = FALSE;
if (sel->sync_streams && sel->cache_buffers) {
/* if we didn't push anything on this pad after it became active, send an
* update FALSE to reset the segments, otherwise use whatever is in the
* selpad current segment update value */
update = selpad->pushed ? selpad->segment_update : FALSE;
}
GST_DEBUG_OBJECT (pad,
"pushing pending NEWSEGMENT update %d, rate %lf, applied rate %lf, "
"format %d, "
"%" G_GINT64_FORMAT " -- %" G_GINT64_FORMAT ", time %"
G_GINT64_FORMAT, FALSE, seg->rate, seg->applied_rate, seg->format,
G_GINT64_FORMAT, update,
seg->rate, seg->applied_rate, seg->format,
seg->start, seg->stop, seg->time);
start_event = gst_event_new_new_segment_full (FALSE, seg->rate,
start_event = gst_event_new_new_segment_full (update, seg->rate,
seg->applied_rate, seg->format, seg->start, seg->stop, seg->time);
selpad->segment_pending = FALSE;
}
@ -845,15 +1159,34 @@ gst_selector_pad_chain (GstPad * pad, GstBuffer * buf)
}
/* forward */
GST_LOG_OBJECT (pad, "Forwarding buffer %p", buf);
GST_LOG_OBJECT (pad, "Forwarding buffer %p with timestamp %" GST_TIME_FORMAT,
buf, GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buf)));
if ((caps = GST_BUFFER_CAPS (buf))) {
if (GST_PAD_CAPS (sel->srcpad) != caps)
gst_pad_set_caps (sel->srcpad, caps);
}
res = gst_pad_push (sel->srcpad, buf);
selpad->pushed = TRUE;
res = gst_pad_push (sel->srcpad, gst_buffer_ref (buf));
GST_LOG_OBJECT (pad, "Buffer %p forwarded result=%d", buf, res);
GST_INPUT_SELECTOR_LOCK (sel);
if (sel->sync_streams && sel->cache_buffers) {
/* Might have changed while pushing */
active_sinkpad = gst_input_selector_activate_sinkpad (sel, pad);
/* only set pad to pushed if we are still the active pad */
if (active_sinkpad == pad)
selpad->pushed = TRUE;
/* cache buffer as we may need it again if we change pads */
gst_selector_pad_cache_buffer (selpad, buf);
gst_input_selector_cleanup_old_cached_buffers (sel, pad);
} else {
selpad->pushed = TRUE;
gst_buffer_unref (buf);
}
GST_INPUT_SELECTOR_UNLOCK (sel);
done:
gst_object_unref (sel);
@ -864,7 +1197,8 @@ ignore:
{
gboolean active_pad_pushed = GST_SELECTOR_PAD_CAST (active_sinkpad)->pushed;
GST_DEBUG_OBJECT (pad, "Pad not active, discard buffer %p", buf);
GST_DEBUG_OBJECT (pad, "Pad not active or buffer timestamp is invalid, "
"discard buffer %p", buf);
/* when we drop a buffer, we're creating a discont on this pad */
selpad->discont = TRUE;
GST_INPUT_SELECTOR_UNLOCK (sel);
@ -883,7 +1217,6 @@ ignore:
flushing:
{
GST_DEBUG_OBJECT (pad, "We are flushing, discard buffer %p", buf);
GST_INPUT_SELECTOR_UNLOCK (sel);
gst_buffer_unref (buf);
res = GST_FLOW_WRONG_STATE;
goto done;
@ -1026,16 +1359,59 @@ gst_input_selector_class_init (GstInputSelectorClass * klass)
* GstInputSelector:sync-streams
*
* If set to %TRUE all inactive streams will be synced to the
* running time of the active stream. This makes sure that no
* buffers are dropped by input-selector that might be needed
* when switching the active pad.
* running time of the active stream or to the current clock.
*
* To make sure no buffers are dropped by input-selector
* that might be needed when switching the active pad,
* sync-mode should be set to "clock" and cache-buffers to TRUE.
*
* Since: 0.10.36
*/
g_object_class_install_property (gobject_class, PROP_SYNC_STREAMS,
g_param_spec_boolean ("sync-streams", "Sync Streams",
"Synchronize inactive streams to the running time of the active stream",
DEFAULT_SYNC_STREAMS, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
"Synchronize inactive streams to the running time of the active "
"stream or to the current clock",
DEFAULT_SYNC_STREAMS,
G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS |
GST_PARAM_MUTABLE_READY));
/**
* GstInputSelector:sync-mode
*
* Select how input-selector will sync buffers when in sync-streams mode.
*
* Note that when using the "active-segment" mode, the "active-segment" may
* be ahead of current clock time when switching the active pad, as the current
* active pad may have pushed more buffers than what was displayed/consumed,
* which may cause delays and some missing buffers.
*
* Since: 0.10.36
*/
g_object_class_install_property (gobject_class, PROP_SYNC_MODE,
g_param_spec_enum ("sync-mode", "Sync mode",
"Behavior in sync-streams mode", GST_TYPE_INPUT_SELECTOR_SYNC_MODE,
DEFAULT_SYNC_MODE,
G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS |
GST_PARAM_MUTABLE_READY));
/**
* GstInputSelector:cache-buffers
*
* If set to %TRUE and GstInputSelector:sync-streams is also set to %TRUE,
* the active pad will cache the buffers still considered valid (after current
* running time, see sync-mode) to avoid missing frames if/when the pad is
* reactivated.
*
* The active pad may push more buffers than what is currently displayed/consumed
* and when changing pads those buffers will be discarded and the only way to
* reactivate that pad without loosing the already consumed buffers is to enable cache.
*/
g_object_class_install_property (gobject_class, PROP_CACHE_BUFFERS,
g_param_spec_boolean ("cache-buffers", "Cache Buffers",
"Cache buffers for active-pad",
DEFAULT_CACHE_BUFFERS,
G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS |
GST_PARAM_MUTABLE_READY));
/**
* GstInputSelector::block:
@ -1134,6 +1510,12 @@ gst_input_selector_init (GstInputSelector * sel,
sel->lock = g_mutex_new ();
sel->cond = g_cond_new ();
sel->blocked = FALSE;
/* lets give a change for downstream to do something on
* active-pad change before we start pushing new buffers */
g_signal_connect_data (sel, "notify::active-pad",
(GCallback) gst_input_selector_active_pad_changed, NULL,
NULL, G_CONNECT_AFTER);
}
static void
@ -1222,7 +1604,7 @@ gst_input_selector_set_active_pad (GstInputSelector * self,
if (old && old->active && !self->pending_close && stop_time >= 0) {
/* schedule a last_stop update if one isn't already scheduled, and a
segment has been pushed before. */
memcpy (&self->segment, &old->segment, sizeof (self->segment));
self->segment = old->segment;
GST_DEBUG_OBJECT (self, "setting stop_time to %" GST_TIME_FORMAT,
GST_TIME_ARGS (stop_time));
@ -1233,6 +1615,9 @@ gst_input_selector_set_active_pad (GstInputSelector * self,
old->pushed = FALSE;
if (new && new->active && start_time >= 0) {
/* schedule a new segment push */
if (self->sync_streams && self->cache_buffers)
start_time = gst_selector_pad_get_running_time (new);
GST_DEBUG_OBJECT (self, "setting start_time to %" GST_TIME_FORMAT,
GST_TIME_ARGS (start_time));
/* schedule a new segment push */
@ -1245,11 +1630,6 @@ gst_input_selector_set_active_pad (GstInputSelector * self,
active_pad_p = &self->active_sinkpad;
gst_object_replace ((GstObject **) active_pad_p, GST_OBJECT_CAST (pad));
/* Wake up all non-active pads in sync mode, they might be
* the active pad now */
if (self->sync_streams)
GST_INPUT_SELECTOR_BROADCAST (self);
GST_DEBUG_OBJECT (self, "New active pad is %" GST_PTR_FORMAT,
self->active_sinkpad);
@ -1270,24 +1650,52 @@ gst_input_selector_set_property (GObject * object, guint prop_id,
pad = g_value_get_object (value);
GST_INPUT_SELECTOR_LOCK (sel);
#if DEBUG_CACHED_BUFFERS
gst_input_selector_debug_cached_buffers (sel);
#endif
gst_input_selector_set_active_pad (sel, pad,
GST_CLOCK_TIME_NONE, GST_CLOCK_TIME_NONE);
#if DEBUG_CACHED_BUFFERS
gst_input_selector_debug_cached_buffers (sel);
#endif
GST_INPUT_SELECTOR_UNLOCK (sel);
break;
}
case PROP_SYNC_STREAMS:
{
GST_INPUT_SELECTOR_LOCK (sel);
sel->sync_streams = g_value_get_boolean (value);
GST_INPUT_SELECTOR_UNLOCK (sel);
break;
}
case PROP_SYNC_MODE:
GST_INPUT_SELECTOR_LOCK (sel);
sel->sync_mode = g_value_get_enum (value);
GST_INPUT_SELECTOR_UNLOCK (sel);
break;
case PROP_CACHE_BUFFERS:
GST_INPUT_SELECTOR_LOCK (object);
sel->cache_buffers = g_value_get_boolean (value);
GST_INPUT_SELECTOR_UNLOCK (object);
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
break;
}
}
static void
gst_input_selector_active_pad_changed (GstInputSelector * sel,
GParamSpec * pspec, gpointer user_data)
{
/* Wake up all non-active pads in sync mode, they might be
* the active pad now */
if (sel->sync_streams)
GST_INPUT_SELECTOR_BROADCAST (sel);
}
static void
gst_input_selector_get_property (GObject * object, guint prop_id,
GValue * value, GParamSpec * pspec)
@ -1310,6 +1718,16 @@ gst_input_selector_get_property (GObject * object, guint prop_id,
g_value_set_boolean (value, sel->sync_streams);
GST_INPUT_SELECTOR_UNLOCK (object);
break;
case PROP_SYNC_MODE:
GST_INPUT_SELECTOR_LOCK (object);
g_value_set_enum (value, sel->sync_mode);
GST_INPUT_SELECTOR_UNLOCK (object);
break;
case PROP_CACHE_BUFFERS:
GST_INPUT_SELECTOR_LOCK (object);
g_value_set_boolean (value, sel->cache_buffers);
GST_INPUT_SELECTOR_UNLOCK (object);
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
break;

14
plugins/elements/gstinputselector.h
View file

@ -48,6 +48,18 @@ typedef struct _GstInputSelectorClass GstInputSelectorClass;
GST_INPUT_SELECTOR_GET_LOCK(sel)))
#define GST_INPUT_SELECTOR_BROADCAST(sel) (g_cond_broadcast (GST_INPUT_SELECTOR_GET_COND(sel)))
/**
* GstInputSelectorSyncMode:
* @GST_INPUT_SELECTOR_SYNC_MODE_ACTIVE_SEGMENT: Sync using the current active segment.
* @GST_INPUT_SELECTOR_SYNC_MODE_CLOCK: Sync using the clock.
*
* The different ways that input-selector can behave when in sync-streams mode.
*/
typedef enum {
GST_INPUT_SELECTOR_SYNC_MODE_ACTIVE_SEGMENT,
GST_INPUT_SELECTOR_SYNC_MODE_CLOCK
} GstInputSelectorSyncMode;
struct _GstInputSelector {
GstElement element;
@ -57,6 +69,8 @@ struct _GstInputSelector {
guint n_pads;
guint padcount;
gboolean sync_streams;
GstInputSelectorSyncMode sync_mode;
gboolean cache_buffers;
GstSegment segment; /* the output segment */
gboolean pending_close; /* if we should push a close first */