[libcamera-devel] [PATCH v5 3/4] android: post_processor: Make post processing async

[Umang Jain]

Hello,

On 10/20/21 4:12 PM, Umang Jain wrote:
> Introduce a dedicated worker class derived from libcamera::Thread.
> The worker class maintains a queue for post-processing requests
> and waits for a post-processing request to become available.
> It will process them as per FIFO before de-queuing it from the
> queue.
>
> To get access to the source and destination buffers in the worker
> thread, we also need to save a pointer to them in the
> Camera3RequestDescriptor.
>
> This patch also implements a flush() for the PostProcessorWorker
> class which is responsible to purge post-processing requests
> queued up while a camera is stopping/flushing. It is hooked with
> CameraStream::flush(), which isn't used currently but will be
> used when we handle flush/stop scnearios in greater detail
> subsequently (in a different patchset).
>
> The libcamera request completion handler CameraDevice::requestComplete()
> assumes that the request that has just completed is at the front of the
> queue. Now that the post-processor runs asynchronously, this isn't true
> anymore, a request being post-processed will stay in the queue and a new
> libcamera request may complete. Remove that assumption, and use the
> request cookie to obtain the Camera3RequestDescriptor.
>
> Signed-off-by: Umang Jain <umang.jain at ideasonboard.com>
> Signed-off-by: Laurent Pinchart <laurent.pinchart at ideasonboard.com>
> ---
>   src/android/camera_device.cpp |  46 +++++----------
>   src/android/camera_request.h  |   4 ++
>   src/android/camera_stream.cpp | 107 +++++++++++++++++++++++++++++++---
>   src/android/camera_stream.h   |  38 +++++++++++-
>   4 files changed, 156 insertions(+), 39 deletions(-)
>
> diff --git a/src/android/camera_device.cpp b/src/android/camera_device.cpp
> index 541c2c81..b14416ce 100644
> --- a/src/android/camera_device.cpp
> +++ b/src/android/camera_device.cpp
> @@ -1020,29 +1020,8 @@ int CameraDevice::processCaptureRequest(camera3_capture_request_t *camera3Reques
>   
>   void CameraDevice::requestComplete(Request *request)
>   {
> -	Camera3RequestDescriptor *descriptor;
> -	{
> -		MutexLocker descriptorsLock(descriptorsMutex_);
> -		ASSERT(!descriptors_.empty());
> -		descriptor = descriptors_.front().get();
> -	}
> -
> -	if (descriptor->request_->cookie() != request->cookie()) {
> -		/*
> -		 * \todo Clarify if the Camera has to be closed on
> -		 * ERROR_DEVICE.
> -		 */
> -		LOG(HAL, Error)
> -			<< "Out-of-order completion for request "
> -			<< utils::hex(request->cookie());
> -
> -		MutexLocker descriptorsLock(descriptorsMutex_);
> -		descriptors_.pop();
> -
> -		notifyError(0, nullptr, CAMERA3_MSG_ERROR_DEVICE);
> -
> -		return;
> -	}
> +	Camera3RequestDescriptor *descriptor =
> +		reinterpret_cast<Camera3RequestDescriptor *>(request->cookie());
>   
>   	/*
>   	 * Prepare the capture result for the Android camera stack.
> @@ -1120,12 +1099,7 @@ void CameraDevice::requestComplete(Request *request)
>   	bool needsPostProcessing = false;
>   	Camera3RequestDescriptor::Status processingStatus =
>   		Camera3RequestDescriptor::Status::Pending;
> -	/*
> -	 * \todo Apply streamsProcessMutex_ when post-processing is adapted to run
> -	 * asynchronously. If we introduce the streamsProcessMutex_ right away, the
> -	 * lock will be held forever since it is synchronous at this point
> -	 * (see streamProcessingComplete).
> -	 */
> +
>   	for (auto &buffer : descriptor->buffers_) {
>   		CameraStream *stream = buffer.stream;
>   
> @@ -1145,7 +1119,13 @@ void CameraDevice::requestComplete(Request *request)
>   			buffer.internalBuffer = src;
>   
>   		needsPostProcessing = true;
> -		int ret = stream->process(*src, buffer, descriptor);
> +
> +		int ret;
> +		{
> +			MutexLocker locker(descriptor->streamsProcessMutex_);
> +			ret = stream->process(*src, buffer, descriptor);
> +		}
> +
>   		if (ret) {
>   			setBufferStatus(stream, buffer, descriptor,
>   					Camera3RequestDescriptor::Status::Error);
> @@ -1156,6 +1136,12 @@ void CameraDevice::requestComplete(Request *request)
>   	if (needsPostProcessing) {
>   		if (processingStatus == Camera3RequestDescriptor::Status::Error) {
>   			descriptor->status_ = processingStatus;
> +			/*
> +			 * \todo This is problematic when some streams are
> +			 * queued successfully, but some fail to get queued.
> +			 * We might end up with use-after-free situation for
> +			 * descriptor in streamProcessingComplete().
> +			 */
>   			sendCaptureResults();
>   		}
>   
> diff --git a/src/android/camera_request.h b/src/android/camera_request.h
> index 3a2774e0..85274414 100644
> --- a/src/android/camera_request.h
> +++ b/src/android/camera_request.h
> @@ -18,6 +18,7 @@
>   
>   #include <hardware/camera3.h>
>   
> +#include "camera_buffer.h"
>   #include "camera_metadata.h"
>   #include "camera_worker.h"
>   
> @@ -39,6 +40,9 @@ public:
>   		int fence;
>   		Status status;
>   		libcamera::FrameBuffer *internalBuffer = nullptr;
> +		std::unique_ptr<CameraBuffer> destBuffer;
> +		const libcamera::FrameBuffer *srcBuffer;
> +		Camera3RequestDescriptor *request = nullptr;
>   	};
>   
>   	Camera3RequestDescriptor(libcamera::Camera *camera,
> diff --git a/src/android/camera_stream.cpp b/src/android/camera_stream.cpp
> index b3cb06a2..a29ce33b 100644
> --- a/src/android/camera_stream.cpp
> +++ b/src/android/camera_stream.cpp
> @@ -99,6 +99,7 @@ int CameraStream::configure()
>   		if (ret)
>   			return ret;
>   
> +		worker_ = std::make_unique<PostProcessorWorker>(postProcessor_.get());
>   		postProcessor_->processComplete.connect(
>   			this, [&](Camera3RequestDescriptor *request, PostProcessor::Status status) {
>   				Camera3RequestDescriptor::Status bufferStatus =
> @@ -110,6 +111,7 @@ int CameraStream::configure()
>   				cameraDevice_->streamProcessingComplete(this, request,
>   									bufferStatus);
>   			});
> +		worker_->start();
>   	}
>   
>   	if (type_ == Type::Internal) {
> @@ -179,23 +181,31 @@ int CameraStream::process(const FrameBuffer &source,
>   	if (!postProcessor_)
>   		return 0;
>   
> -	/*
> -	 * \todo Buffer mapping and processing should be moved to a
> -	 * separate thread.
> -	 */
>   	const StreamConfiguration &output = configuration();
> -	CameraBuffer destBuffer(*dest.camera3Buffer, output.pixelFormat,
> -				output.size, PROT_READ | PROT_WRITE);
> -	if (!destBuffer.isValid()) {
> +	dest.destBuffer = std::make_unique<CameraBuffer>(
> +		*dest.camera3Buffer, output.pixelFormat, output.size, PROT_READ | PROT_WRITE);
> +	if (!dest.destBuffer->isValid()) {
>   		LOG(HAL, Error) << "Failed to create destination buffer";
>   		return -EINVAL;
>   	}
>   
> -	postProcessor_->process(source, &destBuffer, request);
> +	dest.srcBuffer = &source;
> +	dest.request = request;
> +
> +	/* Push the postProcessor request to the worker queue. */
> +	worker_->queueRequest(&dest);
>   
>   	return 0;
>   }
>   
> +void CameraStream::flush()
> +{
> +	if (!postProcessor_)
> +		return;
> +
> +	worker_->flush();
> +}
> +
>   FrameBuffer *CameraStream::getBuffer()
>   {
>   	if (!allocator_)
> @@ -223,3 +233,84 @@ void CameraStream::putBuffer(FrameBuffer *buffer)
>   
>   	buffers_.push_back(buffer);
>   }
> +
> +CameraStream::PostProcessorWorker::PostProcessorWorker(PostProcessor *postProcessor)
> +	: postProcessor_(postProcessor)
> +{
> +}
> +
> +CameraStream::PostProcessorWorker::~PostProcessorWorker()
> +{
> +	{
> +		libcamera::MutexLocker lock(mutex_);
> +		state_ = State::Stopped;
> +	}
> +
> +	cv_.notify_one();
> +	wait();
> +}
> +
> +void CameraStream::PostProcessorWorker::start()
> +{
> +	{
> +		libcamera::MutexLocker lock(mutex_);
> +		state_ = State::Running;
> +	}
> +
> +	Thread::start();
> +}
> +
> +void CameraStream::PostProcessorWorker::queueRequest(Camera3RequestDescriptor::StreamBuffer *dest)
> +{
> +	{
> +		MutexLocker lock(mutex_);
> +		ASSERT(state_ == State::Running);
> +		requests_.push(dest);
> +	}
> +
> +	cv_.notify_one();
> +}
> +
> +void CameraStream::PostProcessorWorker::run()
> +{
> +	MutexLocker locker(mutex_);
> +
> +	while (1) {
> +		cv_.wait(locker, [&] {
> +			return state_ != State::Running || !requests_.empty();
> +		});
> +
> +		if (state_ != State::Running)
> +			break;
> +
> +		Camera3RequestDescriptor::StreamBuffer *stream = requests_.front();
> +		requests_.pop();
> +		locker.unlock();
> +
> +		postProcessor_->process(*stream->srcBuffer, stream->destBuffer.get(),
> +					stream->request);


As I have received in reviews, processComplete should emit a 
Camera3RequestDescriptor::StreamBuffer pointer, instead of having 
emitted currently a pointer to Camera3RequestDescriptor

Now looking postProcessor_->process() call above, I think I should 
further migrate its signature to:

     PostProcessor::process(Camera3RequestDescriptor::StreamBuffer *stream)

and all the three elements can be accessed from there.

Laurent, do you agree with the design decision?

> +
> +		locker.lock();
> +	}
> +
> +	if (state_ == State::Flushing) {
> +		while (!requests_.empty()) {
> +			postProcessor_->processComplete.emit(
> +				requests_.front()->request,
> +				PostProcessor::Status::Error);
> +			requests_.pop();
> +		}
> +
> +		state_ = State::Stopped;
> +		locker.unlock();
> +	}
> +}
> +
> +void CameraStream::PostProcessorWorker::flush()
> +{
> +	libcamera::MutexLocker lock(mutex_);
> +	state_ = State::Flushing;
> +	lock.unlock();
> +
> +	cv_.notify_one();
> +}
> diff --git a/src/android/camera_stream.h b/src/android/camera_stream.h
> index f242336e..64e32c77 100644
> --- a/src/android/camera_stream.h
> +++ b/src/android/camera_stream.h
> @@ -7,12 +7,16 @@
>   #ifndef __ANDROID_CAMERA_STREAM_H__
>   #define __ANDROID_CAMERA_STREAM_H__
>   
> +#include <condition_variable>
>   #include <memory>
>   #include <mutex>
> +#include <queue>
>   #include <vector>
>   
>   #include <hardware/camera3.h>
>   
> +#include <libcamera/base/thread.h>
> +
>   #include <libcamera/camera.h>
>   #include <libcamera/framebuffer.h>
>   #include <libcamera/framebuffer_allocator.h>
> @@ -20,9 +24,9 @@
>   #include <libcamera/pixel_format.h>
>   
>   #include "camera_request.h"
> +#include "post_processor.h"
>   
>   class CameraDevice;
> -class PostProcessor;
>   
>   class CameraStream
>   {
> @@ -126,8 +130,38 @@ public:
>   		    Camera3RequestDescriptor *request);
>   	libcamera::FrameBuffer *getBuffer();
>   	void putBuffer(libcamera::FrameBuffer *buffer);
> +	void flush();
>   
>   private:
> +	class PostProcessorWorker : public libcamera::Thread
> +	{
> +	public:
> +		enum class State {
> +			Stopped,
> +			Running,
> +			Flushing,
> +		};
> +
> +		PostProcessorWorker(PostProcessor *postProcessor);
> +		~PostProcessorWorker();
> +
> +		void start();
> +		void queueRequest(Camera3RequestDescriptor::StreamBuffer *request);
> +		void flush();
> +
> +	protected:
> +		void run() override;
> +
> +	private:
> +		PostProcessor *postProcessor_;
> +
> +		libcamera::Mutex mutex_;
> +		std::condition_variable cv_;
> +
> +		std::queue<Camera3RequestDescriptor::StreamBuffer *> requests_;
> +		State state_;
> +	};
> +
>   	int waitFence(int fence);
>   
>   	CameraDevice *const cameraDevice_;
> @@ -144,6 +178,8 @@ private:
>   	 */
>   	std::unique_ptr<std::mutex> mutex_;
>   	std::unique_ptr<PostProcessor> postProcessor_;
> +
> +	std::unique_ptr<PostProcessorWorker> worker_;
>   };
>   
>   #endif /* __ANDROID_CAMERA_STREAM__ */