[libcamera-devel] [PATCH 10/13] gstreamer: Fix pads locking

[Nicolas Dufresne]

Le vendredi 24 juin 2022 à 02:22 +0300, Laurent Pinchart a écrit :
> The srcpads_ vector is protected by two different locks, the GstObject
> lock of the libcamerasrc element, and the stream_lock that covers the
> run function of the thread. This isn't correct. Use the stream_lock
> consistently to protect the pads.
> 
> Signed-off-by: Laurent Pinchart <laurent.pinchart at ideasonboard.com>
> ---
>  src/gstreamer/gstlibcamerasrc.cpp | 68 ++++++++++++++++---------------
>  1 file changed, 35 insertions(+), 33 deletions(-)
> 
> diff --git a/src/gstreamer/gstlibcamerasrc.cpp b/src/gstreamer/gstlibcamerasrc.cpp
> index b85ba39fb808..58a322b251c7 100644
> --- a/src/gstreamer/gstlibcamerasrc.cpp
> +++ b/src/gstreamer/gstlibcamerasrc.cpp
> @@ -112,7 +112,8 @@ struct GstLibcameraSrcState {
>  	std::shared_ptr<CameraManager> cm_;
>  	std::shared_ptr<Camera> cam_;
>  	std::unique_ptr<CameraConfiguration> config_;
> -	std::vector<GstPad *> srcpads_;
> +
> +	std::vector<GstPad *> srcpads_; /* Protected by stream_lock */
>  
>  	Mutex lock_;
>  	std::queue<std::unique_ptr<RequestWrap>> queuedRequests_
> @@ -349,36 +350,34 @@ gst_libcamera_src_task_run(gpointer user_data)
>  							srcpad, ret);
>  	}
>  
> +	if (ret != GST_FLOW_OK) {
> +		if (ret == GST_FLOW_EOS) {
> +			g_autoptr(GstEvent) eos = gst_event_new_eos();
> +			guint32 seqnum = gst_util_seqnum_next();
> +			gst_event_set_seqnum(eos, seqnum);
> +			for (GstPad *srcpad : state->srcpads_)
> +				gst_pad_push_event(srcpad, gst_event_ref(eos));
> +		} else if (ret != GST_FLOW_FLUSHING) {
> +			GST_ELEMENT_FLOW_ERROR(self, ret);
> +		}
> +		gst_task_stop(self->task);
> +		return;
> +	}
> +
> +	/*
> +	 * Here we need to decide if we want to pause. This needs to
> +	 * happen in lock step with the callback thread which may want
> +	 * to resume the task and might push pending buffers.
> +	 */
> +	bool do_pause;
> +
>  	{
> -		if (ret != GST_FLOW_OK) {
> -			if (ret == GST_FLOW_EOS) {
> -				g_autoptr(GstEvent) eos = gst_event_new_eos();
> -				guint32 seqnum = gst_util_seqnum_next();
> -				gst_event_set_seqnum(eos, seqnum);
> -				for (GstPad *srcpad : state->srcpads_)
> -					gst_pad_push_event(srcpad, gst_event_ref(eos));
> -			} else if (ret != GST_FLOW_FLUSHING) {
> -				GST_ELEMENT_FLOW_ERROR(self, ret);
> -			}
> -			gst_task_stop(self->task);
> -			return;
> -		}
> -
> -		/*
> -		 * Here we need to decide if we want to pause. This needs to
> -		 * happen in lock step with the callback thread which may want
> -		 * to resume the task and might push pending buffers.
> -		 */
> -		bool do_pause;
> -
> -		{
> -			MutexLocker locker(state->lock_);
> -			do_pause = state->completedRequests_.empty();
> -		}
> -
> -		if (do_pause)
> -			gst_task_pause(self->task);
> +		MutexLocker locker(state->lock_);
> +		do_pause = state->completedRequests_.empty();

As you introduce locking order restrictions here, you should also add a comment
next to the state->lock_ documetation that reminds users what order must be
respected. If my reading is correct, the recursive streaming lock must be taken
before this lock (if both are to be taken).

We need to be careful with pads streaming lock, since serialized queries and
even can be re-entrant. So we could be called back with the pad stream lock
held, which needs to be considered in our lock ordering scenarios. So to be
avoided:

  StreamLock
  Lock
  gst_pad_query(...)
  -> May be re-entrant, non-recursive lock held + locking order complexity
  ...

I would simply disallow the request lock from being held while calling into
other element in the graph.

>  	}
> +
> +	if (do_pause)
> +		gst_task_pause(self->task);

This is still racy, I believe you fix that in patch 13. What I would normally do
is prevent concurrent request throughout the patchset, and only enable it in the
last patch. This is just to stay "bisect" friendly, but not strictly required.

>  }
>  
>  static void
> @@ -532,8 +531,11 @@ gst_libcamera_src_task_leave([[maybe_unused]] GstTask *task,
>  		state->completedRequests_ = {};
>  	}
>  
> -	for (GstPad *srcpad : state->srcpads_)
> -		gst_libcamera_pad_set_pool(srcpad, nullptr);
> +	{
> +		GLibRecLocker locker(&self->stream_lock);
> +		for (GstPad *srcpad : state->srcpads_)
> +			gst_libcamera_pad_set_pool(srcpad, nullptr);
> +	}
>  
>  	g_clear_object(&self->allocator);
>  	g_clear_pointer(&self->flow_combiner,
> @@ -692,7 +694,7 @@ gst_libcamera_src_request_new_pad(GstElement *element, GstPadTemplate *templ,
>  	g_object_ref_sink(pad);
>  
>  	if (gst_element_add_pad(element, pad)) {
> -		GLibLocker lock(GST_OBJECT(self));
> +		GLibRecLocker lock(&self->stream_lock);
>  		self->state->srcpads_.push_back(reinterpret_cast<GstPad *>(g_object_ref(pad)));
>  	} else {
>  		GST_ELEMENT_ERROR(element, STREAM, FAILED,
> @@ -712,7 +714,7 @@ gst_libcamera_src_release_pad(GstElement *element, GstPad *pad)
>  	GST_DEBUG_OBJECT(self, "Pad %" GST_PTR_FORMAT " being released", pad);
>  
>  	{
> -		GLibLocker lock(GST_OBJECT(self));
> +		GLibRecLocker lock(&self->stream_lock);
>  		std::vector<GstPad *> &pads = self->state->srcpads_;
>  		auto begin_iterator = pads.begin();
>  		auto end_iterator = pads.end();

With lock order documented:

Reviewed-by: Nicolas Dufresne <nicolas.dufresne at collabora.com>