[libcamera-devel] [PATCH v2 22/27] gst: libcamerasrc: Implement initial streaming
Nicolas Dufresne
nicolas.dufresne at collabora.com
Thu Feb 27 21:04:02 CET 2020
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;
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;
}
--
2.24.1
More information about the libcamera-devel
mailing list