| Message ID | 20200227200407.490616-23-nicolas.dufresne@collabora.com |
|---|---|
| State | Accepted |
| Headers | show |
| Series |
|
| Related | show |
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@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; > } >
Le samedi 29 février 2020 à 16:54 +0200, Laurent Pinchart a écrit : > 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@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 ? I forgot why it was custom success initially, it's needed to make streaming work, I should probably try and squash this into previous commits indeed. > > > > > 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); > > ? I'm just not familiar enough with the data structure to really answer, what does erase do ? This basically map a stream and a buffer, if the buffer is out for delivery, isn't it logical to map it back to no buffer rather then removing the mapping? Will erasing the mapping will cause an free/alloc ? > > > + } > > + > > + 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. The _ style is probably what I dislike the most of libcamera coding style, so I was trying to get away with making Request a struct and using c-style :-), but here I thought it was not obvious if I didn't use this->. I can adhere to the style, I'll need to do that in all internal C++ structs. > > 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(); Must be locked, fast. > > + > > + 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); > > + } Must be locked, also fast. > > + > > + { > > + /* 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); > > + } > > + } This also really fast. So I think it's fine. Object locks should only be taken in GStreamer for short operation (in fact, if you try to do something fancy with these held, you will have a deadlock). So I don't think holding over the function is a concern. The entire thing here has been designed so this callback only queues the ready buffers in various places and wake up the pusher thread if needed. The pusher thread is that one dealing with stream locks, which is by design contentious for push back purpose. > > +} > > + > > 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). Ok, will add a \todo while at it. > > > + delete request; > > + request = NULL; > > s/NULL/nullptr/ Damn, I was sure I managed to catch them all. > > > + 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. I've guessed you mean "this loop". The flow combiner has a state, and remember the last flow that was into it. See the last argument of gst_flow_combiner_update_pad_flow(), but maybe you see something that I don't. > > > + } > > > > - 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). I'd use a bool indeed. Whenever I use TRUE/FALSE is when it's an external API using gboolean, cause gboolean is not the same size as bool. > > > + 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 ? This one was badly squashed, I'll fix. Was a typo I made, which can have disastrous effect, but surprisingly GCC didn't even warn about this (calling a constructor, but not storing the object). > > > 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 ? g_clear_pointer() is made for GObject dispose() function, which can be called multiple time if there is circular references. Short story, it's nul-safe. As for GObject instance, they are always initialized with 0, this is part of the API. > > > 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; > > } > >
Hi Nicolas, On Sat, Feb 29, 2020 at 10:19:02AM -0500, Nicolas Dufresne wrote: > Le samedi 29 février 2020 à 16:54 +0200, Laurent Pinchart a écrit : > > 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@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 ? > > I forgot why it was custom success initially, it's needed to make > streaming work, I should probably try and squash this into previous > commits indeed. > > > > > > > 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); > > > > ? > > I'm just not familiar enough with the data structure to really answer, > what does erase do ? This basically map a stream and a buffer, if the > buffer is out for delivery, isn't it logical to map it back to no > buffer rather then removing the mapping? Will erasing the mapping will > cause an free/alloc ? erase() will remove the element from the map completely. It will cause memory to be freed, so setting it to nullptr as you do here probably makes more sense to avoid constant reallocations. Please disregard my comment. > > > + } > > > + > > > + 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. > > The _ style is probably what I dislike the most of libcamera coding > style, so I was trying to get away with making Request a struct and > using c-style :-), but here I thought it was not obvious if I didn't > use this->. I can adhere to the style, I'll need to do that in all > internal C++ structs. I don't like it much either, but being able to tell members from non-members at a quick glance is nice. I dislike the 'm' prefix more, and haven't found a better option. > > 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(); > > Must be locked, fast. > > > > + > > > + 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); > > > + } > > Must be locked, also fast. > > > > + > > > + { > > > + /* 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); > > > + } > > > + } > > This also really fast. So I think it's fine. Object locks should only > be taken in GStreamer for short operation (in fact, if you try to do > something fancy with these held, you will have a deadlock). So I don't > think holding over the function is a concern. The entire thing here has > been designed so this callback only queues the ready buffers in various > places and wake up the pusher thread if needed. The pusher thread is > that one dealing with stream locks, which is by design contentious for > push back purpose. > > > > +} > > > + > > > 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). > > Ok, will add a \todo while at it. > > > > > > + delete request; > > > + request = NULL; > > > > s/NULL/nullptr/ > > Damn, I was sure I managed to catch them all. > > > > > > + 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. > > I've guessed you mean "this loop". The flow combiner has a state, and > remember the last flow that was into it. See the last argument of > gst_flow_combiner_update_pad_flow(), but maybe you see something that I > don't. So assigning ret in the gst_flow_combiner_update_pad_flow() line is for the sole purpose of getting return value of the last iteration, to be used below ? That makes sense. > > > + } > > > > > > - 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). > > I'd use a bool indeed. Whenever I use TRUE/FALSE is when it's an > external API using gboolean, cause gboolean is not the same size as > bool. > > > > + 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 ? > > This one was badly squashed, I'll fix. Was a typo I made, which can > have disastrous effect, but surprisingly GCC didn't even warn about > this (calling a constructor, but not storing the object). > > > > 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 ? > > g_clear_pointer() is made for GObject dispose() function, which can be > called multiple time if there is circular references. Short story, it's > nul-safe. As for GObject instance, they are always initialized with 0, > this is part of the API. So self is guaranteed to be 0-initialized, and g_clear_pointer() is null-safe. That's what I needed to know :-) > > > 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; > > > } > > >
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; 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); +} 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. */ + 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; + } + + 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)); + + 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()); + 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. */ + delete request; + request = NULL; + 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); + } - 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; + 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); GstLibcameraSrcState *state = self->state; GstFlowReturn flow_ret = GST_FLOW_OK; + gint ret = 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(); 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); 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; }
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@collabora.com> --- src/gstreamer/gstlibcamerapad.cpp | 11 +- src/gstreamer/gstlibcamerapad.h | 2 + src/gstreamer/gstlibcamerasrc.cpp | 192 +++++++++++++++++++++++++++++- 3 files changed, 202 insertions(+), 3 deletions(-)