[libcamera-devel] [PATCH v2 22/27] gst: libcamerasrc: Implement initial streaming
Laurent Pinchart
laurent.pinchart at ideasonboard.com
Sat Feb 29 15:54:58 CET 2020
Hi Nicolas,
Thank you for the patch.
On Thu, Feb 27, 2020 at 03:04:02PM -0500, Nicolas Dufresne wrote:
> With this patch, the element is now able to push buffers to the next
> element in the graph. The buffers are currently missing any metadata
> like timestamp, sequence number. This will be added in the next commit.
>
> Signed-off-by: Nicolas Dufresne <nicolas.dufresne at collabora.com>
> ---
> src/gstreamer/gstlibcamerapad.cpp | 11 +-
> src/gstreamer/gstlibcamerapad.h | 2 +
> src/gstreamer/gstlibcamerasrc.cpp | 192 +++++++++++++++++++++++++++++-
> 3 files changed, 202 insertions(+), 3 deletions(-)
>
> diff --git a/src/gstreamer/gstlibcamerapad.cpp b/src/gstreamer/gstlibcamerapad.cpp
> index 8662c92..2cf1630 100644
> --- a/src/gstreamer/gstlibcamerapad.cpp
> +++ b/src/gstreamer/gstlibcamerapad.cpp
> @@ -18,6 +18,7 @@ struct _GstLibcameraPad {
> StreamRole role;
> GstLibcameraPool *pool;
> GQueue pending_buffers;
> + GstClockTime latency;
> };
>
> enum {
> @@ -159,7 +160,15 @@ gst_libcamera_pad_push_pending(GstPad *pad)
> }
>
> if (!buffer)
> - return GST_FLOW_CUSTOM_SUCCESS;
> + return GST_FLOW_OK;
Does this belong to the previous patch ?
>
> return gst_pad_push(pad, buffer);
> }
> +
> +bool
> +gst_libcamera_pad_has_pending(GstPad *pad)
> +{
> + auto *self = GST_LIBCAMERA_PAD(pad);
> + GLibLocker lock(GST_OBJECT(self));
> + return (self->pending_buffers.length > 0);
No need for the outer parentheses.
> +}
> diff --git a/src/gstreamer/gstlibcamerapad.h b/src/gstreamer/gstlibcamerapad.h
> index d928570..eb24000 100644
> --- a/src/gstreamer/gstlibcamerapad.h
> +++ b/src/gstreamer/gstlibcamerapad.h
> @@ -30,4 +30,6 @@ void gst_libcamera_pad_queue_buffer(GstPad *pad, GstBuffer *buffer);
>
> GstFlowReturn gst_libcamera_pad_push_pending(GstPad *pad);
>
> +bool gst_libcamera_pad_has_pending(GstPad *pad);
> +
> #endif /* __GST_LIBCAMERA_PAD_H__ */
> diff --git a/src/gstreamer/gstlibcamerasrc.cpp b/src/gstreamer/gstlibcamerasrc.cpp
> index 83f93cc..70f2048 100644
> --- a/src/gstreamer/gstlibcamerasrc.cpp
> +++ b/src/gstreamer/gstlibcamerasrc.cpp
> @@ -18,8 +18,10 @@
> #include "gstlibcamerapool.h"
> #include "gstlibcamerasrc.h"
>
> +#include <gst/base/base.h>
> #include <libcamera/camera.h>
> #include <libcamera/camera_manager.h>
> +#include <queue>
> #include <vector>
>
> using namespace libcamera;
> @@ -27,12 +29,71 @@ using namespace libcamera;
> GST_DEBUG_CATEGORY_STATIC(source_debug);
> #define GST_CAT_DEFAULT source_debug
>
> +struct RequestWrap {
> + RequestWrap(Request *request);
> + ~RequestWrap();
> +
> + void attachBuffer(GstBuffer *buffer);
> + GstBuffer *detachBuffer(Stream *stream);
> +
> + /* For ptr comparision only. */
s/comparision/comparison/
> + Request *request_;
> + std::map<Stream *, GstBuffer *> buffers_;
> +};
> +
> +RequestWrap::RequestWrap(Request *request)
> + : request_(request)
> +{
> +}
> +
> +RequestWrap::~RequestWrap()
> +{
> + for (std::pair<Stream *const, GstBuffer *> &item : buffers_) {
> + if (item.second)
> + gst_buffer_unref(item.second);
> + }
> +}
> +
> +void RequestWrap::attachBuffer(GstBuffer *buffer)
> +{
> + FrameBuffer *fb = gst_libcamera_buffer_get_frame_buffer(buffer);
> + Stream *stream = gst_libcamera_buffer_get_stream(buffer);
> +
> + request_->addBuffer(stream, fb);
> +
> + auto item = buffers_.find(stream);
> + if (item != buffers_.end()) {
> + gst_buffer_unref(item->second);
> + item->second = buffer;
> + } else {
> + buffers_[stream] = buffer;
> + }
> +}
> +
> +GstBuffer *RequestWrap::detachBuffer(Stream *stream)
> +{
> + GstBuffer *buffer = nullptr;
> +
> + auto item = buffers_.find(stream);
> + if (item != buffers_.end()) {
> + buffer = item->second;
> + item->second = nullptr;
Do you really want to set second to nullptr here, or should you
buffers_.erase(item);
?
> + }
> +
> + return buffer;
> +}
> +
> /* Used for C++ object with destructors. */
> struct GstLibcameraSrcState {
> + GstLibcameraSrc *src;
> +
> std::unique_ptr<CameraManager> cm;
> std::shared_ptr<Camera> cam;
> std::unique_ptr<CameraConfiguration> config;
> std::vector<GstPad *> srcpads;
> + std::queue<std::unique_ptr<RequestWrap>> requests;
> +
> + void requestCompleted(Request *request);
> };
>
> struct _GstLibcameraSrc {
> @@ -45,6 +106,7 @@ struct _GstLibcameraSrc {
>
> GstLibcameraSrcState *state;
> GstLibcameraAllocator *allocator;
> + GstFlowCombiner *flow_combiner;
> };
>
> enum {
> @@ -68,6 +130,41 @@ GstStaticPadTemplate request_src_template = {
> "src_%s", GST_PAD_SRC, GST_PAD_REQUEST, TEMPLATE_CAPS
> };
>
> +void
> +GstLibcameraSrcState::requestCompleted(Request *request)
> +{
> + GLibLocker lock(GST_OBJECT(this->src));
This is a member function, so you can drop the "this->" and write "src".
Same for everything below. You may want to suffix the data members with
_ to avoid confusion though.
Do we need to lock the whole function ? The requestComplete slot is
called from the pipeline handler thread, so we should minimize the time
we spent here, especially with locks held. Are there any expensive
operation below, is there a risk of delay due to lock contention ?
> +
> + GST_DEBUG_OBJECT(this->src, "buffers are ready");
> +
> + std::unique_ptr<RequestWrap> wrap = std::move(this->requests.front());
> + this->requests.pop();
> +
> + g_return_if_fail(wrap->request_ == request);
> +
> + if ((request->status() == Request::RequestCancelled)) {
> + GST_DEBUG_OBJECT(this->src, "Request was cancelled");
> + return;
> + }
> +
> + GstBuffer *buffer;
> + for (GstPad *srcpad : this->srcpads) {
> + Stream *stream = gst_libcamera_pad_get_stream(srcpad);
> + buffer = wrap->detachBuffer(stream);
> + gst_libcamera_pad_queue_buffer(srcpad, buffer);
> + }
> +
> + {
> + /* We only want to resume the task if it's paused. */
> + GstTask *task = this->src->task;
> + GLibLocker lock(GST_OBJECT(task));
> + if (GST_TASK_STATE(task) == GST_TASK_PAUSED) {
> + GST_TASK_STATE(task) = GST_TASK_STARTED;
> + GST_TASK_SIGNAL(task);
> + }
> + }
> +}
> +
> static bool
> gst_libcamera_src_open(GstLibcameraSrc *self)
> {
> @@ -120,6 +217,8 @@ gst_libcamera_src_open(GstLibcameraSrc *self)
> return false;
> }
>
> + cam->requestCompleted.connect(self->state, &GstLibcameraSrcState::requestCompleted);
> +
> /* No need to lock here, we didn't start our threads yet. */
> self->state->cm = std::move(cm);
> self->state->cam = cam;
> @@ -131,17 +230,85 @@ static void
> gst_libcamera_src_task_run(gpointer user_data)
> {
> GstLibcameraSrc *self = GST_LIBCAMERA_SRC(user_data);
> + GstLibcameraSrcState *state = self->state;
> +
> + Request *request = new Request(state->cam.get());
You should use cam->createRequest(). Looks like we forgot to make the
Request constructor private.
> + auto wrap = std::make_unique<RequestWrap>(request);
> + for (GstPad *srcpad : state->srcpads) {
> + GstLibcameraPool *pool = gst_libcamera_pad_get_pool(srcpad);
> + GstBuffer *buffer;
> + GstFlowReturn ret;
> +
> + ret = gst_buffer_pool_acquire_buffer(GST_BUFFER_POOL(pool),
> + &buffer, nullptr);
> + if (ret != GST_FLOW_OK) {
> + /* RequestWrap does not take ownership, and we won't be
> + * queueing this one due to lack of buffers. */
/*
* ...
*/
and we should fix this in libcamera, the API makes request ownership
awkward (not that the implementation is broken, it's just the API).
> + delete request;
> + request = NULL;
s/NULL/nullptr/
> + break;
> + }
> +
> + wrap->attachBuffer(buffer);
> + }
> +
> + if (request) {
> + GLibLocker lock(GST_OBJECT(self));
> + GST_TRACE_OBJECT(self, "Requesting buffers");
> + state->cam->queueRequest(request);
> + state->requests.push(std::move(wrap));
> + }
> +
> + GstFlowReturn ret = GST_FLOW_OK;
> + gst_flow_combiner_reset(self->flow_combiner);
> + for (GstPad *srcpad : state->srcpads) {
> + ret = gst_libcamera_pad_push_pending(srcpad);
> + ret = gst_flow_combiner_update_pad_flow(self->flow_combiner,
> + srcpad, ret);
We have a single pad so that's no an issue yet, but this looks ignores
errors from any pad but the last.
> + }
>
> - GST_DEBUG_OBJECT(self, "Streaming thread is now capturing");
> + {
> + /*
> + * Here we need to decide if we want to pause or stop the task. This
> + * needs to happen in lock step with the callback thread which may want
> + * to resume the task.
> + */
> + GLibLocker lock(GST_OBJECT(self));
> + 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;
> + }
> +
> + gboolean do_pause = true;
I would replace this with a bool, or use TRUE here (and FALSE below).
> + for (GstPad *srcpad : state->srcpads) {
> + if (gst_libcamera_pad_has_pending(srcpad)) {
> + do_pause = false;
> + break;
> + }
> + }
> +
> + if (do_pause)
> + gst_task_pause(self->task);
> + }
> }
>
> static void
> gst_libcamera_src_task_enter(GstTask *task, GThread *thread, gpointer user_data)
> {
> GstLibcameraSrc *self = GST_LIBCAMERA_SRC(user_data);
> - GLibRecLocker(&self->stream_lock);
> + GLibRecLocker lock(&self->stream_lock);
This this fix a compilation error in a previous patch ?
> GstLibcameraSrcState *state = self->state;
> GstFlowReturn flow_ret = GST_FLOW_OK;
> + gint ret = 0;
I don't think you need to initialize this to 0.
>
> GST_DEBUG_OBJECT(self, "Streaming thread has started");
>
> @@ -230,12 +397,23 @@ gst_libcamera_src_task_enter(GstTask *task, GThread *thread, gpointer user_data)
> return;
> }
>
> + self->flow_combiner = gst_flow_combiner_new();
If an error occurs above and flow_combiner isn't set there, won't the
g_clear_pointer() call below operate on an unitialized variable ?
> for (gsize i = 0; i < state->srcpads.size(); i++) {
> GstPad *srcpad = state->srcpads[i];
> const StreamConfiguration &stream_cfg = state->config->at(i);
> GstLibcameraPool *pool = gst_libcamera_pool_new(self->allocator,
> stream_cfg.stream());
> gst_libcamera_pad_set_pool(srcpad, pool);
> + gst_flow_combiner_add_pad(self->flow_combiner, srcpad);
> + }
> +
> + ret = state->cam->start();
> + if (ret) {
> + GST_ELEMENT_ERROR(self, RESOURCE, SETTINGS,
> + ("Failed to start the camera: %s", g_strerror(-ret)),
> + ("Camera.start() failed with error code %i", ret));
> + gst_task_stop(task);
> + return;
> }
>
> done:
> @@ -257,10 +435,14 @@ gst_libcamera_src_task_leave(GstTask *task, GThread *thread, gpointer user_data)
>
> GST_DEBUG_OBJECT(self, "Streaming thread is about to stop");
>
> + state->cam->stop();
> +
> for (GstPad *srcpad : state->srcpads)
> gst_libcamera_pad_set_pool(srcpad, NULL);
s/NULL/nullptr/
>
> g_clear_object(&self->allocator);
> + g_clear_pointer(&self->flow_combiner,
> + (GDestroyNotify)gst_flow_combiner_free);
> }
>
> static void
> @@ -340,6 +522,9 @@ gst_libcamera_src_change_state(GstElement *element, GstStateChange transition)
> return GST_STATE_CHANGE_FAILURE;
> ret = GST_STATE_CHANGE_NO_PREROLL;
> break;
> + case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
> + gst_task_start(self->task);
> + break;
> case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
> ret = GST_STATE_CHANGE_NO_PREROLL;
> break;
> @@ -389,6 +574,9 @@ gst_libcamera_src_init(GstLibcameraSrc *self)
>
> state->srcpads.push_back(gst_pad_new_from_template(templ, "src"));
> gst_element_add_pad(GST_ELEMENT(self), state->srcpads[0]);
> +
> + /* C-style friend. */
> + state->src = self;
> self->state = state;
> }
>
--
Regards,
Laurent Pinchart
More information about the libcamera-devel
mailing list