mirrors/gstreamer
mirrors/gstreamer
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gstreamer.git synced 2025-04-26 04:36:20 +00:00
Code Issues 1 Projects Releases Wiki Activity

libs/gst/base/gstbasesrc.c: Rework the locking of basesrc in a similar fashion to basesink. We basically have one loc...

Browse source 
Original commit message from CVS:
* libs/gst/base/gstbasesrc.c: (gst_base_src_wait_playing),
(gst_base_src_set_live), (gst_base_src_is_live),
(gst_base_src_query_latency), (gst_base_src_perform_seek),
(gst_base_src_default_event), (gst_base_src_wait),
(gst_base_src_do_sync), (gst_base_src_get_range),
(gst_base_src_pad_get_range), (gst_base_src_loop),
(gst_base_src_unlock), (gst_base_src_unlock_stop),
(gst_base_src_set_flushing), (gst_base_src_set_playing),
(gst_base_src_activate_push), (gst_base_src_activate_pull),
(gst_base_src_change_state):
Rework the locking of basesrc in a similar fashion to basesink. We
basically have one lock (LIVE_LOCK) protecting the dataflow. This allows
us to handle live sources and semi live ones much better.
Simplify flushing.
Fix unlocking when seeking, shutting down and pausing in live sources.
This commit is contained in:
Wim Taymans 2007-10-05 14:40:06 +00:00
parent a2e299f3ef
commit 04fb0735f9
2 changed files with 219 additions and 109 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
ChangeLog
View file

@ -1,3 +1,21 @@
2007-10-05 Wim Taymans <wim.taymans@gmail.com>
* libs/gst/base/gstbasesrc.c: (gst_base_src_wait_playing),
(gst_base_src_set_live), (gst_base_src_is_live),
(gst_base_src_query_latency), (gst_base_src_perform_seek),
(gst_base_src_default_event), (gst_base_src_wait),
(gst_base_src_do_sync), (gst_base_src_get_range),
(gst_base_src_pad_get_range), (gst_base_src_loop),
(gst_base_src_unlock), (gst_base_src_unlock_stop),
(gst_base_src_set_flushing), (gst_base_src_set_playing),
(gst_base_src_activate_push), (gst_base_src_activate_pull),
(gst_base_src_change_state):
Rework the locking of basesrc in a similar fashion to basesink. We
basically have one lock (LIVE_LOCK) protecting the dataflow. This allows
us to handle live sources and semi live ones much better.
Simplify flushing.
Fix unlocking when seeking, shutting down and pausing in live sources.
2007-10-05 Wim Taymans <wim.taymans@gmail.com> 2007-10-05 Wim Taymans <wim.taymans@gmail.com>
* tests/check/pipelines/simple-launch-lines.c: (run_pipeline): * tests/check/pipelines/simple-launch-lines.c: (run_pipeline):

310
libs/gst/base/gstbasesrc.c
View file

@ -243,6 +243,7 @@ struct _GstBaseSrcPrivate
gboolean last_sent_eos; /* last thing we did was send an EOS (we set this gboolean last_sent_eos; /* last thing we did was send an EOS (we set this
* to avoid the sending of two EOS in some cases) */ * to avoid the sending of two EOS in some cases) */
gboolean discont; gboolean discont;
gboolean flushing;
/* two segments to be sent in the streaming thread with STREAM_LOCK */ /* two segments to be sent in the streaming thread with STREAM_LOCK */
GstEvent *close_segment; GstEvent *close_segment;
@ -314,8 +315,11 @@ static gboolean gst_base_src_default_query (GstBaseSrc * src, GstQuery * query);
static gboolean gst_base_src_default_prepare_seek_segment (GstBaseSrc * src, static gboolean gst_base_src_default_prepare_seek_segment (GstBaseSrc * src,
GstEvent * event, GstSegment * segment); GstEvent * event, GstSegment * segment);
static gboolean gst_base_src_unlock (GstBaseSrc * basesrc); static gboolean gst_base_src_set_flushing (GstBaseSrc * basesrc,
static gboolean gst_base_src_unlock_stop (GstBaseSrc * basesrc); gboolean flushing, gboolean live_play, gboolean unlock);
static gboolean gst_base_src_unlock (GstBaseSrc * basesrc, gboolean unlock);
static gboolean gst_base_src_unlock_stop (GstBaseSrc * basesrc,
gboolean unlock);
static gboolean gst_base_src_start (GstBaseSrc * basesrc); static gboolean gst_base_src_start (GstBaseSrc * basesrc);
static gboolean gst_base_src_stop (GstBaseSrc * basesrc); static gboolean gst_base_src_stop (GstBaseSrc * basesrc);
@ -483,7 +487,6 @@ GstFlowReturn
gst_base_src_wait_playing (GstBaseSrc * src) gst_base_src_wait_playing (GstBaseSrc * src)
{ {
/* block until the state changes, or we get a flush, or something */ /* block until the state changes, or we get a flush, or something */
GST_LIVE_LOCK (src);
if (src->is_live) { if (src->is_live) {
while (G_UNLIKELY (!src->live_running)) { while (G_UNLIKELY (!src->live_running)) {
GST_DEBUG ("live source signal waiting"); GST_DEBUG ("live source signal waiting");
@ -492,23 +495,16 @@ gst_base_src_wait_playing (GstBaseSrc * src)
GST_LIVE_WAIT (src); GST_LIVE_WAIT (src);
GST_DEBUG ("live source unlocked"); GST_DEBUG ("live source unlocked");
} }
/* FIXME, use another variable to signal stopping so that we don't if (src->priv->flushing)
* have to grab another lock. */
GST_OBJECT_LOCK (src->srcpad);
if (G_UNLIKELY (GST_PAD_IS_FLUSHING (src->srcpad)))
goto flushing; goto flushing;
GST_OBJECT_UNLOCK (src->srcpad);
} }
GST_LIVE_UNLOCK (src);
return GST_FLOW_OK; return GST_FLOW_OK;
/* ERRORS */ /* ERRORS */
flushing: flushing:
{ {
GST_DEBUG_OBJECT (src, "pad is flushing"); GST_DEBUG_OBJECT (src, "we are flushing");
GST_OBJECT_UNLOCK (src->srcpad);
GST_LIVE_UNLOCK (src);
return GST_FLOW_WRONG_STATE; return GST_FLOW_WRONG_STATE;
} }
} }
@ -530,9 +526,9 @@ flushing:
void void
gst_base_src_set_live (GstBaseSrc * src, gboolean live) gst_base_src_set_live (GstBaseSrc * src, gboolean live)
{ {
GST_LIVE_LOCK (src); GST_OBJECT_LOCK (src);
src->is_live = live; src->is_live = live;
GST_LIVE_UNLOCK (src); GST_OBJECT_UNLOCK (src);
} }
/** /**
@ -548,9 +544,9 @@ gst_base_src_is_live (GstBaseSrc * src)
{ {
gboolean result; gboolean result;
GST_LIVE_LOCK (src); GST_OBJECT_LOCK (src);
result = src->is_live; result = src->is_live;
GST_LIVE_UNLOCK (src); GST_OBJECT_UNLOCK (src);
return result; return result;
} }
@ -598,7 +594,7 @@ gst_base_src_query_latency (GstBaseSrc * src, gboolean * live,
{ {
GstClockTime min; GstClockTime min;
GST_LIVE_LOCK (src); GST_OBJECT_LOCK (src);
if (live) if (live)
*live = src->is_live; *live = src->is_live;
@ -618,7 +614,7 @@ gst_base_src_query_latency (GstBaseSrc * src, gboolean * live,
GST_LOG_OBJECT (src, "latency: live %d, min %" GST_TIME_FORMAT GST_LOG_OBJECT (src, "latency: live %d, min %" GST_TIME_FORMAT
", max %" GST_TIME_FORMAT, src->is_live, GST_TIME_ARGS (min), ", max %" GST_TIME_FORMAT, src->is_live, GST_TIME_ARGS (min),
GST_TIME_ARGS (-1)); GST_TIME_ARGS (-1));
GST_LIVE_UNLOCK (src); GST_OBJECT_UNLOCK (src);
return TRUE; return TRUE;
} }
@ -1032,7 +1028,10 @@ gst_base_src_prepare_seek_segment (GstBaseSrc * src, GstEvent * event,
* acquire the STREAM_LOCK which is taken when we are in the * acquire the STREAM_LOCK which is taken when we are in the
* _loop() function or when a getrange() is called. Normally * _loop() function or when a getrange() is called. Normally
* we will not receive a seek if we are operating in pull mode * we will not receive a seek if we are operating in pull mode
* though. * though. When we operate as a live source we might block on the live
* cond, which does not release the STREAM_LOCK. Therefore we will try
* to grab the LIVE_LOCK instead of the STREAM_LOCK to make sure it is
* safe to perform the seek.
* *
* When we are in the loop() function, we might be in the middle * When we are in the loop() function, we might be in the middle
* of pushing a buffer, which might block in a sink. To make sure * of pushing a buffer, which might block in a sink. To make sure
@ -1045,10 +1044,11 @@ gst_base_src_prepare_seek_segment (GstBaseSrc * src, GstEvent * event,
* blocks on the sample we might wait a very long time until the sink * blocks on the sample we might wait a very long time until the sink
* unblocks the sample. In any case we acquire the STREAM_LOCK and * unblocks the sample. In any case we acquire the STREAM_LOCK and
* can continue the seek. A non-flushing seek is normally done in a * can continue the seek. A non-flushing seek is normally done in a
* running pipeline to perform seamless playback. * running pipeline to perform seamless playback, this means that the sink is
* PLAYING and will return from its chain function.
* In the case of a non-flushing seek we need to make sure that the * In the case of a non-flushing seek we need to make sure that the
* data we output after the seek is continuous with the previous data, * data we output after the seek is continuous with the previous data,
* this is because a non-flushing seek does not reset the stream-time * this is because a non-flushing seek does not reset the running-time
* to 0. We do this by closing the currently running segment, ie. sending * to 0. We do this by closing the currently running segment, ie. sending
* a new_segment event with the stop position set to the last processed * a new_segment event with the stop position set to the last processed
* position. * position.
@ -1119,16 +1119,15 @@ gst_base_src_perform_seek (GstBaseSrc * src, GstEvent * event, gboolean unlock)
gst_pad_pause_task (src->srcpad); gst_pad_pause_task (src->srcpad);
/* unblock streaming thread */ /* unblock streaming thread */
if (unlock) gst_base_src_unlock (src, unlock);
gst_base_src_unlock (src);
/* grab streaming lock, this should eventually be possible, either /* grab live lock, this should eventually be possible, either
* because the task is paused or our streaming thread stopped * because the task is paused, our streaming thread stopped
* because our peer is flushing. */ * because our peer is flushing or because we are blocked in the live
GST_PAD_STREAM_LOCK (src->srcpad); * cond. */
GST_LIVE_LOCK (src);
if (unlock) gst_base_src_unlock_stop (src, unlock);
gst_base_src_unlock_stop (src);
/* If we configured the seeksegment above, don't overwrite it now. Otherwise /* If we configured the seeksegment above, don't overwrite it now. Otherwise
* copy the current segment info into the temp segment that we can actually * copy the current segment info into the temp segment that we can actually
@ -1232,7 +1231,7 @@ gst_base_src_perform_seek (GstBaseSrc * src, GstEvent * event, gboolean unlock)
src->srcpad); src->srcpad);
/* and release the lock again so we can continue streaming */ /* and release the lock again so we can continue streaming */
GST_PAD_STREAM_UNLOCK (src->srcpad); GST_LIVE_UNLOCK (src);
return res; return res;
@ -1388,10 +1387,10 @@ gst_base_src_default_event (GstBaseSrc * src, GstEvent * event)
case GST_EVENT_FLUSH_START: case GST_EVENT_FLUSH_START:
/* cancel any blocking getrange, is normally called /* cancel any blocking getrange, is normally called
* when in pull mode. */ * when in pull mode. */
result = gst_base_src_unlock (src); result = gst_base_src_set_flushing (src, TRUE, FALSE, TRUE);
break; break;
case GST_EVENT_FLUSH_STOP: case GST_EVENT_FLUSH_STOP:
result = gst_base_src_unlock_stop (src); result = gst_base_src_set_flushing (src, FALSE, TRUE, TRUE);
break; break;
default: default:
result = TRUE; result = TRUE;
@ -1500,12 +1499,12 @@ gst_base_src_wait (GstBaseSrc * basesrc, GstClock * clock, GstClockTime time)
id = gst_clock_new_single_shot_id (clock, time); id = gst_clock_new_single_shot_id (clock, time);
basesrc->clock_id = id; basesrc->clock_id = id;
/* release the object lock while waiting */ /* release the live lock while waiting */
GST_OBJECT_UNLOCK (basesrc); GST_LIVE_UNLOCK (basesrc);
ret = gst_clock_id_wait (id, NULL); ret = gst_clock_id_wait (id, NULL);
GST_OBJECT_LOCK (basesrc); GST_LIVE_LOCK (basesrc);
gst_clock_id_unref (id); gst_clock_id_unref (id);
basesrc->clock_id = NULL; basesrc->clock_id = NULL;
@ -1646,9 +1645,9 @@ gst_base_src_do_sync (GstBaseSrc * basesrc, GstBuffer * buffer)
"waiting for clock, base time %" GST_TIME_FORMAT "waiting for clock, base time %" GST_TIME_FORMAT
", stream_start %" GST_TIME_FORMAT, ", stream_start %" GST_TIME_FORMAT,
GST_TIME_ARGS (base_time), GST_TIME_ARGS (start)); GST_TIME_ARGS (base_time), GST_TIME_ARGS (start));
GST_OBJECT_UNLOCK (basesrc);
result = gst_base_src_wait (basesrc, clock, start + base_time); result = gst_base_src_wait (basesrc, clock, start + base_time);
GST_OBJECT_UNLOCK (basesrc);
GST_LOG_OBJECT (basesrc, "clock entry done: %d", result); GST_LOG_OBJECT (basesrc, "clock entry done: %d", result);
@ -1739,6 +1738,7 @@ unexpected_length:
} }
} }
/* must be called with LIVE_LOCK */
static GstFlowReturn static GstFlowReturn
gst_base_src_get_range (GstBaseSrc * src, guint64 offset, guint length, gst_base_src_get_range (GstBaseSrc * src, guint64 offset, guint length,
GstBuffer ** buf) GstBuffer ** buf)
@ -1785,6 +1785,11 @@ gst_base_src_get_range (GstBaseSrc * src, guint64 offset, guint length,
/* now sync before pushing the buffer */ /* now sync before pushing the buffer */
status = gst_base_src_do_sync (src, *buf); status = gst_base_src_do_sync (src, *buf);
/* waiting for the clock could have made us flushing */
if (src->priv->flushing)
goto flushing;
switch (status) { switch (status) {
case GST_CLOCK_EARLY: case GST_CLOCK_EARLY:
/* the buffer is too late. We currently don't drop the buffer. */ /* the buffer is too late. We currently don't drop the buffer. */
@ -1851,6 +1856,13 @@ reached_num_buffers:
GST_DEBUG_OBJECT (src, "sent all buffers"); GST_DEBUG_OBJECT (src, "sent all buffers");
return GST_FLOW_UNEXPECTED; return GST_FLOW_UNEXPECTED;
} }
flushing:
{
GST_DEBUG_OBJECT (src, "we are flushing");
gst_buffer_unref (*buf);
*buf = NULL;
return GST_FLOW_WRONG_STATE;
}
} }
static GstFlowReturn static GstFlowReturn
@ -1862,11 +1874,26 @@ gst_base_src_pad_get_range (GstPad * pad, guint64 offset, guint length,
src = GST_BASE_SRC (gst_pad_get_parent (pad)); src = GST_BASE_SRC (gst_pad_get_parent (pad));
GST_LIVE_LOCK (src);
if (src->priv->flushing)
goto flushing;
res = gst_base_src_get_range (src, offset, length, buf); res = gst_base_src_get_range (src, offset, length, buf);
done:
GST_LIVE_UNLOCK (src);
gst_object_unref (src); gst_object_unref (src);
return res; return res;
/* ERRORS */
flushing:
{
GST_DEBUG_OBJECT (src, "we are flushing");
res = GST_FLOW_WRONG_STATE;
goto done;
}
} }
static gboolean static gboolean
@ -1941,6 +1968,10 @@ gst_base_src_loop (GstPad * pad)
src = GST_BASE_SRC (gst_pad_get_parent (pad)); src = GST_BASE_SRC (gst_pad_get_parent (pad));
GST_LIVE_LOCK (src);
if (src->priv->flushing)
goto flushing;
src->priv->last_sent_eos = FALSE; src->priv->last_sent_eos = FALSE;
/* if we operate in bytes, we can calculate an offset */ /* if we operate in bytes, we can calculate an offset */
@ -1953,6 +1984,7 @@ gst_base_src_loop (GstPad * pad)
if (G_UNLIKELY (ret != GST_FLOW_OK)) { if (G_UNLIKELY (ret != GST_FLOW_OK)) {
GST_INFO_OBJECT (src, "pausing after gst_base_src_get_range() = %s", GST_INFO_OBJECT (src, "pausing after gst_base_src_get_range() = %s",
gst_flow_get_name (ret)); gst_flow_get_name (ret));
GST_LIVE_UNLOCK (src);
goto pause; goto pause;
} }
/* this should not happen */ /* this should not happen */
@ -2012,6 +2044,7 @@ gst_base_src_loop (GstPad * pad)
GST_BUFFER_FLAG_SET (buf, GST_BUFFER_FLAG_DISCONT); GST_BUFFER_FLAG_SET (buf, GST_BUFFER_FLAG_DISCONT);
src->priv->discont = FALSE; src->priv->discont = FALSE;
} }
GST_LIVE_UNLOCK (src);
ret = gst_pad_push (pad, buf); ret = gst_pad_push (pad, buf);
if (G_UNLIKELY (ret != GST_FLOW_OK)) { if (G_UNLIKELY (ret != GST_FLOW_OK)) {
@ -2031,6 +2064,13 @@ done:
return; return;
/* special cases */ /* special cases */
flushing:
{
GST_DEBUG_OBJECT (src, "we are flushing");
GST_LIVE_UNLOCK (src);
ret = GST_FLOW_WRONG_STATE;
goto pause;
}
pause: pause:
{ {
const gchar *reason = gst_flow_get_name (ret); const gchar *reason = gst_flow_get_name (ret);
@ -2065,11 +2105,9 @@ null_buffer:
{ {
GST_ELEMENT_ERROR (src, STREAM, FAILED, GST_ELEMENT_ERROR (src, STREAM, FAILED,
(_("Internal data flow error.")), ("element returned NULL buffer")); (_("Internal data flow error.")), ("element returned NULL buffer"));
GST_LIVE_UNLOCK (src);
/* we finished the segment on error */ /* we finished the segment on error */
src->data.ABI.running = FALSE; ret = GST_FLOW_ERROR;
gst_pad_pause_task (pad);
gst_pad_push_event (pad, gst_event_new_eos ());
src->priv->last_sent_eos = TRUE;
goto done; goto done;
} }
} }
@ -2078,26 +2116,33 @@ null_buffer:
* resources allocated by start() and freed from stop(). This needs to be added * resources allocated by start() and freed from stop(). This needs to be added
* to the docs at some point. */ * to the docs at some point. */
static gboolean static gboolean
gst_base_src_unlock (GstBaseSrc * basesrc) gst_base_src_unlock (GstBaseSrc * basesrc, gboolean unlock)
{ {
GstBaseSrcClass *bclass; GstBaseSrcClass *bclass;
gboolean result = TRUE; gboolean result = TRUE;
GST_DEBUG ("unlock"); GST_DEBUG ("unlock");
/* unblock whatever the subclass is doing */
bclass = GST_BASE_SRC_GET_CLASS (basesrc); bclass = GST_BASE_SRC_GET_CLASS (basesrc);
if (bclass->unlock) /* unblock whatever the subclass is doing if we were asked to */
result = bclass->unlock (basesrc); if (unlock) {
if (bclass->unlock)
result = bclass->unlock (basesrc);
}
GST_LIVE_LOCK (basesrc);
GST_DEBUG ("unschedule clock"); GST_DEBUG ("unschedule clock");
/* and unblock the clock as well, if any */ /* and unblock the clock as well, if any */
GST_OBJECT_LOCK (basesrc);
if (basesrc->clock_id) { if (basesrc->clock_id) {
gst_clock_id_unschedule (basesrc->clock_id); gst_clock_id_unschedule (basesrc->clock_id);
} }
GST_OBJECT_UNLOCK (basesrc);
if (unlock) {
if (bclass->unlock_stop)
result = bclass->unlock_stop (basesrc);
}
GST_DEBUG ("unlock done"); GST_DEBUG ("unlock done");
GST_LIVE_UNLOCK (basesrc);
return result; return result;
} }
@ -2106,7 +2151,7 @@ gst_base_src_unlock (GstBaseSrc * basesrc)
* resources allocated by start() and freed from stop(). This needs to be added * resources allocated by start() and freed from stop(). This needs to be added
* to the docs at some point. */ * to the docs at some point. */
static gboolean static gboolean
gst_base_src_unlock_stop (GstBaseSrc * basesrc) gst_base_src_unlock_stop (GstBaseSrc * basesrc, gboolean unlock)
{ {
GstBaseSrcClass *bclass; GstBaseSrcClass *bclass;
gboolean result = TRUE; gboolean result = TRUE;
@ -2115,9 +2160,11 @@ gst_base_src_unlock_stop (GstBaseSrc * basesrc)
/* Finish a previous unblock request, allowing subclasses to flush command /* Finish a previous unblock request, allowing subclasses to flush command
* queues or whatever they need to do */ * queues or whatever they need to do */
bclass = GST_BASE_SRC_GET_CLASS (basesrc); if (unlock) {
if (bclass->unlock_stop) bclass = GST_BASE_SRC_GET_CLASS (basesrc);
result = bclass->unlock_stop (basesrc); if (bclass->unlock_stop)
result = bclass->unlock_stop (basesrc);
}
GST_DEBUG_OBJECT (basesrc, "unlock stop done"); GST_DEBUG_OBJECT (basesrc, "unlock stop done");
@ -2329,25 +2376,97 @@ gst_base_src_stop (GstBaseSrc * basesrc)
} }
static gboolean static gboolean
gst_base_src_deactivate (GstBaseSrc * basesrc, GstPad * pad) gst_base_src_set_flushing (GstBaseSrc * basesrc,
gboolean flushing, gboolean live_play, gboolean unlock)
{ {
gboolean result; GstBaseSrcClass *bclass;
bclass = GST_BASE_SRC_GET_CLASS (basesrc);
if (flushing && unlock) {
/* unlock any subclasses, we need to do this before grabbing the
* LIVE_LOCK since we hold this lock before going into ::create. We pass an
* unlock to the params because of backwards compat (see seek handler)*/
if (bclass->unlock)
bclass->unlock (basesrc);
}
/* the live lock is released when we are blocked, waiting for playing or
* when we sync to the clock. */
GST_LIVE_LOCK (basesrc); GST_LIVE_LOCK (basesrc);
basesrc->live_running = TRUE; basesrc->priv->flushing = flushing;
if (flushing) {
/* if we are locked in the live lock, signal it to make it flush */
basesrc->live_running = TRUE;
/* step 1, now that we have the LIVE lock, clear our unlock request */
if (bclass->unlock_stop)
bclass->unlock_stop (basesrc);
/* step 2, unblock clock sync (if any) or any other blocking thing */
if (basesrc->clock_id)
gst_clock_id_unschedule (basesrc->clock_id);
} else {
/* signal the live source that it can start playing */
basesrc->live_running = live_play;
}
GST_LIVE_SIGNAL (basesrc); GST_LIVE_SIGNAL (basesrc);
GST_LIVE_UNLOCK (basesrc); GST_LIVE_UNLOCK (basesrc);
/* step 1, unblock clock sync (if any) */ return TRUE;
result = gst_base_src_unlock (basesrc); }
/* step 2, make sure streaming finishes */ /* the purpose of this function is to make sure that a live source blocks in the
result &= gst_pad_stop_task (pad); * LIVE lock or leaves the LIVE lock and continues playing. */
static gboolean
gst_base_src_set_playing (GstBaseSrc * basesrc, gboolean live_play)
{
GstBaseSrcClass *bclass;
/* step 3, clear the unblock condition */ bclass = GST_BASE_SRC_GET_CLASS (basesrc);
result &= gst_base_src_unlock_stop (basesrc);
return result; /* unlock subclasses locked in ::create, we only do this when we stop playing. */
if (!live_play) {
GST_DEBUG_OBJECT (basesrc, "unlock");
if (bclass->unlock)
bclass->unlock (basesrc);
}
/* we are now able to grab the LIVE lock, when we get it, we can be
* waiting for PLAYING while blocked in the LIVE cond or we can be waiting
* for the clock. */
GST_LIVE_LOCK (basesrc);
/* unblock clock sync (if any) */
if (basesrc->clock_id)
gst_clock_id_unschedule (basesrc->clock_id);
/* configure what to do when we get to the LIVE lock. */
basesrc->live_running = live_play;
if (live_play) {
gboolean start;
/* clear our unlock request when going to PLAYING */
GST_DEBUG_OBJECT (basesrc, "unlock stop");
if (bclass->unlock_stop)
bclass->unlock_stop (basesrc);
/* for live sources we restart the timestamp correction */
basesrc->priv->latency = -1;
/* have to restart the task in case it stopped because of the unlock when
* we went to PAUSED. Only do this if we operating in push mode. */
GST_OBJECT_LOCK (basesrc->srcpad);
start = (GST_PAD_ACTIVATE_MODE (basesrc->srcpad) == GST_ACTIVATE_PUSH);
GST_OBJECT_UNLOCK (basesrc->srcpad);
if (start)
gst_pad_start_task (basesrc->srcpad, (GstTaskFunction) gst_base_src_loop,
basesrc->srcpad);
}
GST_LIVE_SIGNAL (basesrc);
GST_LIVE_UNLOCK (basesrc);
return TRUE;
} }
static gboolean static gboolean
@ -2369,6 +2488,8 @@ gst_base_src_activate_push (GstPad * pad, gboolean active)
goto error_start; goto error_start;
basesrc->priv->last_sent_eos = FALSE; basesrc->priv->last_sent_eos = FALSE;
basesrc->priv->discont = TRUE;
gst_base_src_set_flushing (basesrc, FALSE, FALSE, FALSE);
/* do initial seek, which will start the task */ /* do initial seek, which will start the task */
GST_OBJECT_LOCK (basesrc); GST_OBJECT_LOCK (basesrc);
@ -2386,10 +2507,10 @@ gst_base_src_activate_push (GstPad * pad, gboolean active)
gst_event_unref (event); gst_event_unref (event);
} else { } else {
GST_DEBUG_OBJECT (basesrc, "Deactivating in push mode"); GST_DEBUG_OBJECT (basesrc, "Deactivating in push mode");
/* call the unlock function and stop the task */ /* flush all */
if (G_UNLIKELY (!gst_base_src_deactivate (basesrc, pad))) gst_base_src_set_flushing (basesrc, TRUE, FALSE, TRUE);
goto deactivate_failed; /* stop the task */
gst_pad_stop_task (pad);
/* now we can stop the source */ /* now we can stop the source */
if (G_UNLIKELY (!gst_base_src_stop (basesrc))) if (G_UNLIKELY (!gst_base_src_stop (basesrc)))
goto error_stop; goto error_stop;
@ -2415,11 +2536,6 @@ seek_failed:
gst_event_unref (event); gst_event_unref (event);
return FALSE; return FALSE;
} }
deactivate_failed:
{
GST_ERROR_OBJECT (basesrc, "Failed to deactivate in push mode");
return FALSE;
}
error_stop: error_stop:
{ {
GST_DEBUG_OBJECT (basesrc, "Failed to stop in push mode"); GST_DEBUG_OBJECT (basesrc, "Failed to stop in push mode");
@ -2443,11 +2559,14 @@ gst_base_src_activate_pull (GstPad * pad, gboolean active)
/* if not random_access, we cannot operate in pull mode for now */ /* if not random_access, we cannot operate in pull mode for now */
if (G_UNLIKELY (!gst_base_src_check_get_range (basesrc))) if (G_UNLIKELY (!gst_base_src_check_get_range (basesrc)))
goto no_get_range; goto no_get_range;
/* stop flushing now but for live sources, still block in the LIVE lock when
* we are not yet PLAYING */
gst_base_src_set_flushing (basesrc, FALSE, FALSE, FALSE);
} else { } else {
GST_DEBUG_OBJECT (basesrc, "Deactivating in pull mode"); GST_DEBUG_OBJECT (basesrc, "Deactivating in pull mode");
/* call the unlock function. We have no task to stop. */ /* flush all, there is no task to stop */
if (G_UNLIKELY (!gst_base_src_deactivate (basesrc, pad))) gst_base_src_set_flushing (basesrc, TRUE, FALSE, TRUE);
goto deactivate_failed;
/* don't send EOS when going from PAUSED => READY when in pull mode */ /* don't send EOS when going from PAUSED => READY when in pull mode */
basesrc->priv->last_sent_eos = TRUE; basesrc->priv->last_sent_eos = TRUE;
@ -2469,11 +2588,6 @@ no_get_range:
gst_base_src_stop (basesrc); gst_base_src_stop (basesrc);
return FALSE; return FALSE;
} }
deactivate_failed:
{
GST_ERROR_OBJECT (basesrc, "Failed to deactivate in pull mode");
return FALSE;
}
error_stop: error_stop:
{ {
GST_ERROR_OBJECT (basesrc, "Failed to stop in pull mode"); GST_ERROR_OBJECT (basesrc, "Failed to stop in pull mode");
@ -2494,36 +2608,13 @@ gst_base_src_change_state (GstElement * element, GstStateChange transition)
case GST_STATE_CHANGE_NULL_TO_READY: case GST_STATE_CHANGE_NULL_TO_READY:
break; break;
case GST_STATE_CHANGE_READY_TO_PAUSED: case GST_STATE_CHANGE_READY_TO_PAUSED:
GST_LIVE_LOCK (element); no_preroll = gst_base_src_is_live (basesrc);
basesrc->priv->latency = -1;
if (basesrc->is_live) {
no_preroll = TRUE;
basesrc->live_running = FALSE;
}
basesrc->priv->last_sent_eos = FALSE;
basesrc->priv->discont = TRUE;
GST_LIVE_UNLOCK (element);
break; break;
case GST_STATE_CHANGE_PAUSED_TO_PLAYING: case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
GST_LIVE_LOCK (element); if (gst_base_src_is_live (basesrc)) {
if (basesrc->is_live) { /* now we can start playback */
gboolean start; gst_base_src_set_playing (basesrc, TRUE);
gst_base_src_unlock_stop (basesrc);
/* for live sources we restart the timestamp correction */
basesrc->priv->latency = -1;
basesrc->live_running = TRUE;
GST_LIVE_SIGNAL (element);
/* have to restart the task in case it stopped because of the unlock when
* we went to PAUSED. Only do this if we operating in push mode. */
GST_OBJECT_LOCK (basesrc->srcpad);
start = (GST_PAD_ACTIVATE_MODE (basesrc->srcpad) == GST_ACTIVATE_PUSH);
GST_OBJECT_UNLOCK (basesrc->srcpad);
if (start)
gst_pad_start_task (basesrc->srcpad,
(GstTaskFunction) gst_base_src_loop, basesrc->srcpad);
} }
GST_LIVE_UNLOCK (element);
break; break;
default: default:
break; break;
@ -2536,18 +2627,19 @@ gst_base_src_change_state (GstElement * element, GstStateChange transition)
switch (transition) { switch (transition) {
case GST_STATE_CHANGE_PLAYING_TO_PAUSED: case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
GST_LIVE_LOCK (element); if (gst_base_src_is_live (basesrc)) {
if (basesrc->is_live) { /* make sure we block in the live lock in PAUSED */
gst_base_src_unlock (basesrc); gst_base_src_set_playing (basesrc, FALSE);
no_preroll = TRUE; no_preroll = TRUE;
basesrc->live_running = FALSE;
} }
GST_LIVE_UNLOCK (element);
break; break;
case GST_STATE_CHANGE_PAUSED_TO_READY: case GST_STATE_CHANGE_PAUSED_TO_READY:
{ {
GstEvent **event_p; GstEvent **event_p;
/* we don't need to unblock anything here, the pad deactivation code
* already did this */
/* FIXME, deprecate this behaviour, it is very dangerous. /* FIXME, deprecate this behaviour, it is very dangerous.
* the prefered way of sending EOS downstream is by sending * the prefered way of sending EOS downstream is by sending
* the EOS event to the element */ * the EOS event to the element */