[libcamera-devel] [PATCH v4 5/7] android: post_processor: Make post processing async

[Umang Jain]

Hi me,

On 10/11/21 1:05 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.
>
> 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 |  25 +-------
>   src/android/camera_device.h   |   3 +
>   src/android/camera_stream.cpp | 108 +++++++++++++++++++++++++++++++---
>   src/android/camera_stream.h   |  39 +++++++++++-
>   4 files changed, 144 insertions(+), 31 deletions(-)
>
> diff --git a/src/android/camera_device.cpp b/src/android/camera_device.cpp
> index eba370ea..61b902ad 100644
> --- a/src/android/camera_device.cpp
> +++ b/src/android/camera_device.cpp
> @@ -239,6 +239,7 @@ Camera3RequestDescriptor::Camera3RequestDescriptor(
>   	/* Clone the controls associated with the camera3 request. */
>   	settings_ = CameraMetadata(camera3Request->settings);
>   
> +	dest_.reset();
>   	/*
>   	 * Create the CaptureRequest, stored as a unique_ptr<> to tie its
>   	 * lifetime to the descriptor.
> @@ -1094,28 +1095,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 and possibly demote the Fatal to simple
> -		 * Error.
> -		 */
> -		notifyError(0, nullptr, CAMERA3_MSG_ERROR_DEVICE);
> -		LOG(HAL, Fatal)
> -			<< "Out-of-order completion for request "
> -			<< utils::hex(request->cookie());
> -
> -		MutexLocker descriptorsLock(descriptorsMutex_);
> -		descriptors_.pop();
> -		return;
> -	}
> +	Camera3RequestDescriptor *descriptor =
> +		reinterpret_cast<Camera3RequestDescriptor *>(request->cookie());
>   
>   	/*
>   	 * Prepare the capture result for the Android camera stack.
> diff --git a/src/android/camera_device.h b/src/android/camera_device.h
> index eee97516..725a0618 100644
> --- a/src/android/camera_device.h
> +++ b/src/android/camera_device.h
> @@ -59,6 +59,9 @@ struct Camera3RequestDescriptor {
>   	std::unique_ptr<CameraMetadata> resultMetadata_;
>   	libcamera::FrameBuffer *internalBuffer_;
>   
> +	std::unique_ptr<CameraBuffer> dest_;
> +	const libcamera::FrameBuffer *src_;
> +
>   	camera3_capture_result_t captureResult_ = {};
>   	Status status_ = Status::Pending;
>   };
> diff --git a/src/android/camera_stream.cpp b/src/android/camera_stream.cpp
> index cec07269..818ef948 100644
> --- a/src/android/camera_stream.cpp
> +++ b/src/android/camera_stream.cpp
> @@ -94,10 +94,12 @@ int CameraStream::configure()
>   		if (ret)
>   			return ret;
>   
> +		worker_ = std::make_unique<PostProcessorWorker>(postProcessor_.get());
>   		postProcessor_->processComplete.connect(
>   			this, [&](Camera3RequestDescriptor *request, PostProcessor::Status status) {
>   				cameraDevice_->streamProcessingComplete(this, request, status);
>   			});
> +		worker_->start();
>   	}
>   
>   	if (type_ == Type::Internal) {
> @@ -167,19 +169,26 @@ void CameraStream::process(const FrameBuffer &source,
>   	if (!postProcessor_)
>   		return;
>   
> -	/*
> -	 * \todo Buffer mapping and processing should be moved to a
> -	 * separate thread.
> -	 */
>   	const StreamConfiguration &output = configuration();
> -	CameraBuffer dest(*camera3Dest.buffer, output.pixelFormat, output.size,
> -			  PROT_READ | PROT_WRITE);
> -	if (!dest.isValid()) {
> +	request->dest_ = std::make_unique<CameraBuffer>(
> +		*camera3Dest.buffer, output.pixelFormat, output.size, PROT_READ | PROT_WRITE);
> +	if (!request->dest_->isValid()) {
>   		LOG(HAL, Error) << "Failed to create destination buffer";
>   		return;
>   	}
>   
> -	postProcessor_->process(source, &dest, request);
> +	request->src_ = &source;
> +
> +	/* Push the postProcessor request to the worker queue. */
> +	worker_->queueRequest(request);
> +}
> +
> +void CameraStream::flush()
> +{
> +	if (!postProcessor_)
> +		return;
> +
> +	worker_->flush();
>   }
>   
>   FrameBuffer *CameraStream::getBuffer()
> @@ -209,3 +218,86 @@ 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 *request)
> +{
> +	{
> +		MutexLocker lock(mutex_);
> +		ASSERT(state_ == State::Running);
> +		requests_.push(request);
> +	}
> +	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 *descriptor = requests_.front();
> +		requests_.pop();
> +		locker.unlock();
> +
> +		postProcessor_->process(*descriptor->src_, descriptor->dest_.get(),
> +					descriptor);
> +
> +		locker.lock();
> +	}
> +
> +	if (state_ == State::Flushing) {
> +		while (!requests_.empty()) {
> +			postProcessor_->processComplete.emit(requests_.front(),
> +							     PostProcessor::Status::Error);
> +			requests_.pop();
> +		}
> +		state_ = State::Stopped;
> +		locker.unlock();
> +		cv_.notify_one();
> +	}
> +}
> +
> +void CameraStream::PostProcessorWorker::flush()
> +{
> +	libcamera::MutexLocker lock(mutex_);
> +	state_ = State::Flushing;
> +	lock.unlock();
> +	cv_.notify_one();
> +
> +	lock.lock();
> +	cv_.wait(lock, [&] {


Mental note:

This can add a lot of latency to the flush op since it's happening on 
the main thread. I think the queues (post-processing one and 
descriptors) can be flushed out independently without waiting for 
one-another.

One still needs to set error state on descriptors/buffers, so that will 
require some thinking (since you can't iterate over a std::queue but 
set-pop() only, even for error setting). This might introduce another 
process_capture_results() callback outside sendCaptureResults (which is 
a divergence from ideal but maybe need to bite that bullet). The goal 
here is to be as fast as possible on flush op, to clear the queues. As 
an experiment, one can measure asctual latency on flush() as in master 
to have a common base figure can compare with the latency of this series


> +		return state_ == State::Stopped;
> +	});
> +}
> diff --git a/src/android/camera_stream.h b/src/android/camera_stream.h
> index a0c5f166..e410f35d 100644
> --- a/src/android/camera_stream.h
> +++ b/src/android/camera_stream.h
> @@ -7,21 +7,26 @@
>   #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>
>   #include <libcamera/geometry.h>
>   #include <libcamera/pixel_format.h>
>   
> +#include "post_processor.h"
> +
>   class CameraDevice;
>   class CameraMetadata;
> -class PostProcessor;
>   
>   struct Camera3RequestDescriptor;
>   
> @@ -125,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 *request);
> +		void flush();
> +
> +	protected:
> +		void run() override;
> +
> +	private:
> +		PostProcessor *postProcessor_;
> +
> +		libcamera::Mutex mutex_;
> +		std::condition_variable cv_;
> +
> +		std::queue<Camera3RequestDescriptor *> requests_;
> +		State state_;
> +	};
> +
>   	int waitFence(int fence);
>   
>   	CameraDevice *const cameraDevice_;
> @@ -143,6 +178,8 @@ private:
>   	 */
>   	std::unique_ptr<std::mutex> mutex_;
>   	std::unique_ptr<PostProcessor> postProcessor_;
> +
> +	std::unique_ptr<PostProcessorWorker> worker_;
>   };
>   
>   #endif /* __ANDROID_CAMERA_STREAM__ */