[libcamera-devel,v1,10/23] gst: libcamerasrc: Start/Stop the streaming thread

2744 diff mbox series

From: Nicolas Dufresne <nicolas.dufresne@collabora.com>

Use a GstTask as our internal streaming thread. Unlike GstBaseSrc, we
will be running an streaming thread at the element level rather then
per pad. This is needed to combine buffer request for multiple pads.

Signed-off-by: Nicolas Dufresne <nicolas.dufresne@collabora.com>
---
 src/gstreamer/gstlibcamerasrc.cpp | 45 +++++++++++++++++++++++++++++++
 1 file changed, 45 insertions(+)

Hi Nicolas,

Thank you for the patch.

The commit message mentions start/stop, but the patch doesn't actually
start or stop the task. Should it be rewritten to instead say "Add a
task for the streaming thread", or something similar ?

On Tue, Jan 28, 2020 at 10:31:57PM -0500, Nicolas Dufresne wrote:
> From: Nicolas Dufresne <nicolas.dufresne@collabora.com>
> 
> Use a GstTask as our internal streaming thread. Unlike GstBaseSrc, we
> will be running an streaming thread at the element level rather then

s/an/a/
s/then/than/

> per pad. This is needed to combine buffer request for multiple pads.
> 
> Signed-off-by: Nicolas Dufresne <nicolas.dufresne@collabora.com>
> ---
>  src/gstreamer/gstlibcamerasrc.cpp | 45 +++++++++++++++++++++++++++++++
>  1 file changed, 45 insertions(+)
> 
> diff --git a/src/gstreamer/gstlibcamerasrc.cpp b/src/gstreamer/gstlibcamerasrc.cpp
> index abb376b..4d0c7d0 100644
> --- a/src/gstreamer/gstlibcamerasrc.cpp
> +++ b/src/gstreamer/gstlibcamerasrc.cpp
> @@ -26,7 +26,11 @@ struct GstLibcameraSrcState {
>  
>  struct _GstLibcameraSrc {
>  	GstElement parent;
> +
> +	GRecMutex stream_lock;
> +	GstTask *task;
>  	GstPad *srcpad;
> +
>  	gchar *camera_name;
>  
>  	GstLibcameraSrcState *state;
> @@ -112,6 +116,30 @@ gst_libcamera_src_open(GstLibcameraSrc *self)
>  	return true;
>  }
>  
> +static void
> +gst_libcamera_src_task_run(gpointer user_data)
> +{
> +	GstLibcameraSrc *self = GST_LIBCAMERA_SRC(user_data);
> +
> +	GST_DEBUG_OBJECT(self, "Streaming thread it now capturing");

s/it/is/ ?

> +}
> +
> +static void
> +gst_libcamera_src_task_enter(GstTask *task, GThread *thread, gpointer user_data)
> +{
> +	GstLibcameraSrc *self = GST_LIBCAMERA_SRC(user_data);
> +
> +	GST_DEBUG_OBJECT(self, "Streaming thread has started");
> +}
> +
> +static void
> +gst_libcamera_src_task_leave(GstTask *task, GThread *thread, gpointer user_data)
> +{
> +	GstLibcameraSrc *self = GST_LIBCAMERA_SRC(user_data);
> +
> +	GST_DEBUG_OBJECT(self, "Streaming thread is about to stop");
> +}
> +
>  static void
>  gst_libcamera_src_close(GstLibcameraSrc *self)
>  {
> @@ -182,9 +210,18 @@ gst_libcamera_src_change_state(GstElement *element, GstStateChange transition)
>  			return GST_STATE_CHANGE_FAILURE;
>  		break;
>  	case GST_STATE_CHANGE_READY_TO_PAUSED:
> +		/* This needs to be called after pads activation */

s/activation/activation./

(I'll stop mentioning this too, could you please apply it to the other
patches too ?)

> +		if (!gst_task_pause(self->task))
> +			return GST_STATE_CHANGE_FAILURE;
> +		ret = GST_STATE_CHANGE_NO_PREROLL;
> +		break;
>  	case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
>  		ret = GST_STATE_CHANGE_NO_PREROLL;
>  		break;
> +	case GST_STATE_CHANGE_PAUSED_TO_READY:
> +		/* FIXME maybe this should happen before pad deactivation ? */

This can be fixed later if needed, but I'm curious to know what this
means :-) Is there anything that prevents fixing this now ? Or is it
just a matter of investigating whether this is required or not ? A
"\todo Investigate if the task should be joined before pad deactivation"
would be better in that case.

> +		gst_task_join(self->task);
> +		break;
>  	case GST_STATE_CHANGE_READY_TO_NULL:
>  		gst_libcamera_src_close(self);
>  		break;
> @@ -201,6 +238,8 @@ gst_libcamera_src_finalize(GObject *object)
>  	GObjectClass *klass = G_OBJECT_CLASS(gst_libcamera_src_parent_class);
>  	GstLibcameraSrc *self = GST_LIBCAMERA_SRC(object);
>  
> +	g_rec_mutex_clear(&self->stream_lock);
> +	g_clear_object(&self->task);
>  	g_free(self->camera_name);
>  	delete self->state;
>  
> @@ -213,6 +252,12 @@ gst_libcamera_src_init(GstLibcameraSrc *self)
>  	GstLibcameraSrcState *state = new GstLibcameraSrcState();
>  	GstPadTemplate *templ = gst_element_get_pad_template(GST_ELEMENT(self), "src");
>  
> +	g_rec_mutex_init(&self->stream_lock);
> +	self->task = gst_task_new(gst_libcamera_src_task_run, self, nullptr);
> +	gst_task_set_enter_callback(self->task, gst_libcamera_src_task_enter, self, nullptr);
> +	gst_task_set_leave_callback(self->task, gst_libcamera_src_task_leave, self, nullptr);
> +	gst_task_set_lock(self->task, &self->stream_lock);
> +
>  	self->srcpad = gst_pad_new_from_template(templ, "src");
>  	gst_element_add_pad(GST_ELEMENT(self), self->srcpad);
>  	self->state = state;

On mar, 2020-02-11 at 21:50 +0200, Laurent Pinchart wrote:
> Hi Nicolas,
> 
> Thank you for the patch.
> 
> The commit message mentions start/stop, but the patch doesn't actually
> start or stop the task. Should it be rewritten to instead say "Add a
> task for the streaming thread", or something similar ?

We do gst_task_pause(), which effectively start the thread (enter callback is
called), and gst_task_join() which do gst_task_stop() and wait (leave callback
is called). Your suggestion is good though, it's an implementation detail that
it does start and stop a thread.

> 
> On Tue, Jan 28, 2020 at 10:31:57PM -0500, Nicolas Dufresne wrote:
> > From: Nicolas Dufresne <nicolas.dufresne@collabora.com>
> > 
> > Use a GstTask as our internal streaming thread. Unlike GstBaseSrc, we
> > will be running an streaming thread at the element level rather then
> 
> s/an/a/
> s/then/than/
> 
> > per pad. This is needed to combine buffer request for multiple pads.
> > 
> > Signed-off-by: Nicolas Dufresne <nicolas.dufresne@collabora.com>
> > ---
> >  src/gstreamer/gstlibcamerasrc.cpp | 45 +++++++++++++++++++++++++++++++
> >  1 file changed, 45 insertions(+)
> > 
> > diff --git a/src/gstreamer/gstlibcamerasrc.cpp
> > b/src/gstreamer/gstlibcamerasrc.cpp
> > index abb376b..4d0c7d0 100644
> > --- a/src/gstreamer/gstlibcamerasrc.cpp
> > +++ b/src/gstreamer/gstlibcamerasrc.cpp
> > @@ -26,7 +26,11 @@ struct GstLibcameraSrcState {
> >  
> >  struct _GstLibcameraSrc {
> >  	GstElement parent;
> > +
> > +	GRecMutex stream_lock;
> > +	GstTask *task;
> >  	GstPad *srcpad;
> > +
> >  	gchar *camera_name;
> >  
> >  	GstLibcameraSrcState *state;
> > @@ -112,6 +116,30 @@ gst_libcamera_src_open(GstLibcameraSrc *self)
> >  	return true;
> >  }
> >  
> > +static void
> > +gst_libcamera_src_task_run(gpointer user_data)
> > +{
> > +	GstLibcameraSrc *self = GST_LIBCAMERA_SRC(user_data);
> > +
> > +	GST_DEBUG_OBJECT(self, "Streaming thread it now capturing");
> 
> s/it/is/ ?
> 
> > +}
> > +
> > +static void
> > +gst_libcamera_src_task_enter(GstTask *task, GThread *thread, gpointer
> > user_data)
> > +{
> > +	GstLibcameraSrc *self = GST_LIBCAMERA_SRC(user_data);
> > +
> > +	GST_DEBUG_OBJECT(self, "Streaming thread has started");
> > +}
> > +
> > +static void
> > +gst_libcamera_src_task_leave(GstTask *task, GThread *thread, gpointer
> > user_data)
> > +{
> > +	GstLibcameraSrc *self = GST_LIBCAMERA_SRC(user_data);
> > +
> > +	GST_DEBUG_OBJECT(self, "Streaming thread is about to stop");
> > +}
> > +
> >  static void
> >  gst_libcamera_src_close(GstLibcameraSrc *self)
> >  {
> > @@ -182,9 +210,18 @@ gst_libcamera_src_change_state(GstElement *element,
> > GstStateChange transition)
> >  			return GST_STATE_CHANGE_FAILURE;
> >  		break;
> >  	case GST_STATE_CHANGE_READY_TO_PAUSED:
> > +		/* This needs to be called after pads activation */
> 
> s/activation/activation./
> 
> (I'll stop mentioning this too, could you please apply it to the other
> patches too ?)
> 
> > +		if (!gst_task_pause(self->task))
> > +			return GST_STATE_CHANGE_FAILURE;
> > +		ret = GST_STATE_CHANGE_NO_PREROLL;
> > +		break;
> >  	case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
> >  		ret = GST_STATE_CHANGE_NO_PREROLL;
> >  		break;
> > +	case GST_STATE_CHANGE_PAUSED_TO_READY:
> > +		/* FIXME maybe this should happen before pad deactivation ? */
> 
> This can be fixed later if needed, but I'm curious to know what this
> means :-) Is there anything that prevents fixing this now ? Or is it
> just a matter of investigating whether this is required or not ? A
> "\todo Investigate if the task should be joined before pad deactivation"
> would be better in that case.

I might be able to remove it in V2. For sources using BaseSrc, we often do
poll() or select() in the create() function. Create function is basically the
run function of a GstTask (with special BaseSrc wrapper). What one needs to know
here is that this transition is responsible of deactivating pads. To deactivate
pads, you need to take the stream lock (a recursive mutex on the pads). This
pads is taken each time the run function of a task is called. If you don't
cancel any blocking waits, the pad deactivation will block, slowing down the
deactivation, or blocking forever in some cases. To unblock these things in the
state transition, you have to do so before chaining to the base class
(GstElement), hence "before pad deactivation".

There is effectively a upper and a lowe part to the state transition virtual
function implementation. In this case, it seems I only need the lower part. For
threaded filters, when handling a flush-start/stop sequence, it is much more
involving.

Now I had no idea what it would look like in the end when I wrote this. What it
ended up is that libcamera calls us from it's own thread. In that thread we
queue and wakeup the task (take it out of pause). All I had to do is to make
sure it's not possible to wakup the task after we have decided to shut it down.
I think this is covered in later patch, and so there is nothing to unlock ever.
And this comment can be removed.

> 
> > +		gst_task_join(self->task);
> > +		break;
> >  	case GST_STATE_CHANGE_READY_TO_NULL:
> >  		gst_libcamera_src_close(self);
> >  		break;
> > @@ -201,6 +238,8 @@ gst_libcamera_src_finalize(GObject *object)
> >  	GObjectClass *klass = G_OBJECT_CLASS(gst_libcamera_src_parent_class);
> >  	GstLibcameraSrc *self = GST_LIBCAMERA_SRC(object);
> >  
> > +	g_rec_mutex_clear(&self->stream_lock);
> > +	g_clear_object(&self->task);
> >  	g_free(self->camera_name);
> >  	delete self->state;
> >  
> > @@ -213,6 +252,12 @@ gst_libcamera_src_init(GstLibcameraSrc *self)
> >  	GstLibcameraSrcState *state = new GstLibcameraSrcState();
> >  	GstPadTemplate *templ = gst_element_get_pad_template(GST_ELEMENT(self),
> > "src");
> >  
> > +	g_rec_mutex_init(&self->stream_lock);
> > +	self->task = gst_task_new(gst_libcamera_src_task_run, self, nullptr);
> > +	gst_task_set_enter_callback(self->task, gst_libcamera_src_task_enter,
> > self, nullptr);
> > +	gst_task_set_leave_callback(self->task, gst_libcamera_src_task_leave,
> > self, nullptr);
> > +	gst_task_set_lock(self->task, &self->stream_lock);
> > +
> >  	self->srcpad = gst_pad_new_from_template(templ, "src");
> >  	gst_element_add_pad(GST_ELEMENT(self), self->srcpad);
> >  	self->state = state;