[PATCH v2 5/5] libcamera: software_isp: Modify dispatching messages on stop

[Milan Zamazal]

Hi Laurent,

thank you for review.

Laurent Pinchart <laurent.pinchart at ideasonboard.com> writes:

> Hi Milan,
>
> Thank you for the patch.
>
> On Mon, Feb 24, 2025 at 07:52:35PM +0100, Milan Zamazal wrote:
>> There may be pending messages in SoftwareIsp message queue when
>> SoftwareIsp stops.  The call to IPAProxySoft::stop() will dispatch them
>> before SoftwareIsp::stop() finishes.  But this is dependent on
>> IPAProxySoft::stop() implementation and we can do better to ensure they
>> are dispatched before SoftwareIsp::stop() finishes.
>> 
>> Let's introduce new `receiver' argument to Thread::dispatchMessages(),
>> limiting dispatching the messages to the given receiver.  Now we can
>> flush the messages destined for the SoftwareIsp instance explicitly.
>> And the IPA proxy can flush just the messages destined for itself.
>> Other messages of the given thread remain queued and will be handled
>> elsewhere as appropriate.
>> 
>> Signed-off-by: Milan Zamazal <mzamazal at redhat.com>
>> ---
>>  include/libcamera/base/thread.h                   |  3 ++-
>>  src/libcamera/base/thread.cpp                     | 15 ++++++++++-----
>
> Could you split this to a separate patch ? Something like
>
> libcamera: base: thread: Support dispatching messages for specific receiver
>
> The Thread::dispatchMessage() function supports filtering messages based
> on their type. It can be useful to also dispatch only messages posted
> for a specific receiver. Add an optional receiver argument to the
> dispatchMessage() function to do so. When set to null (the default
> value), the behaviour of the function is not changed.
>
>>  src/libcamera/software_isp/software_isp.cpp       |  3 +++
>>  .../libcamera_templates/proxy_functions.tmpl      |  2 +-
>
> prxoy_function.tmpl can then also be split to a separate patch.
>
> utils: ipc: Only dispatch messages for proxy when stopping thread
>
> When stopping the proxy thread, all messages of InvokeMessage type
> posted to the pipeline handler thread are dispatched, to ensure that all
> signals emitted from the proxy thread and queued for delivery to the
> proxy are delivered synchronously. This unnecessarily delivers queued
> signals for other objects in the pipeline handler thread, possibly
> delaying processing of higher priority events.
>
> Improve the implementation by limiting synchronous delivery to messages
> posted for the proxy.
>
>
>
> In case a problem is found later with the IPA proxy, it will be easier
> to revert this specific change.
>
>>  4 files changed, 16 insertions(+), 7 deletions(-)
>> 
>> diff --git a/include/libcamera/base/thread.h b/include/libcamera/base/thread.h
>> index 3cbf6398..b9284c2c 100644
>> --- a/include/libcamera/base/thread.h
>> +++ b/include/libcamera/base/thread.h
>> @@ -48,7 +48,8 @@ public:
>>  
>>  	EventDispatcher *eventDispatcher();
>>  
>> -	void dispatchMessages(Message::Type type = Message::Type::None);
>> +	void dispatchMessages(Message::Type type = Message::Type::None,
>> +			      Object *receiver = nullptr);
>>  
>>  protected:
>>  	int exec();
>> diff --git a/src/libcamera/base/thread.cpp b/src/libcamera/base/thread.cpp
>> index 02128f23..6f22f4ed 100644
>> --- a/src/libcamera/base/thread.cpp
>> +++ b/src/libcamera/base/thread.cpp
>> @@ -603,6 +603,8 @@ void Thread::removeMessages(Object *receiver)
>>  /**
>>   * \brief Dispatch posted messages for this thread
>>   * \param[in] type The message type
>> + * \param[in] receiver If not null, dispatch only messages for the given
>> + *    receiver
>>   *
>>   * This function immediately dispatches all the messages previously posted for
>>   * this thread with postMessage() that match the message \a type. If the \a type
>
> Let's expand this paragraph to explain the receiver parameter:
>
>  * This function immediately dispatches all the messages of the given \a type
>  * previously posted to this thread for the \a receiver with postMessage(). If
>  * the \a type is Message::Type::None, all messages types are dispatched. If the
>  * \a receiver is null, messages to all receivers are dispatched.
>
> I think you can then shorten the documentation of the parameter:
>
>  * \param[in] receiver The receiver whose messages to dispatch
>
>> @@ -616,7 +618,7 @@ void Thread::removeMessages(Object *receiver)
>>   * same thread from an object's message handler. It guarantees delivery of
>>   * messages in the order they have been posted in all cases.
>>   */
>> -void Thread::dispatchMessages(Message::Type type)
>> +void Thread::dispatchMessages(Message::Type type, Object *receiver)
>>  {
>>  	ASSERT(data_ == ThreadData::current());
>>  
>> @@ -633,6 +635,9 @@ void Thread::dispatchMessages(Message::Type type)
>>  		if (type != Message::Type::None && msg->type() != type)
>>  			continue;
>>  
>> +		if (receiver && receiver != msg->receiver_)
>> +			continue;
>> +
>>  		/*
>>  		 * Move the message, setting the entry in the list to null. It
>>  		 * will cause recursive calls to ignore the entry, and the erase
>> @@ -640,12 +645,12 @@ void Thread::dispatchMessages(Message::Type type)
>>  		 */
>>  		std::unique_ptr<Message> message = std::move(msg);
>>  
>> -		Object *receiver = message->receiver_;
>> -		ASSERT(data_ == receiver->thread()->data_);
>> -		receiver->pendingMessages_--;
>> +		Object *messageReceiver = message->receiver_;
>> +		ASSERT(data_ == messageReceiver->thread()->data_);
>> +		messageReceiver->pendingMessages_--;
>>  
>>  		locker.unlock();
>> -		receiver->message(message.get());
>> +		messageReceiver->message(message.get());
>>  		message.reset();
>>  		locker.lock();
>>  	}
>> diff --git a/src/libcamera/software_isp/software_isp.cpp b/src/libcamera/software_isp/software_isp.cpp
>> index 3a605ab2..8f5ee774 100644
>> --- a/src/libcamera/software_isp/software_isp.cpp
>> +++ b/src/libcamera/software_isp/software_isp.cpp
>> @@ -13,6 +13,8 @@
>>  #include <sys/types.h>
>>  #include <unistd.h>
>>  
>> +#include <libcamera/base/thread.h>
>> +
>>  #include <libcamera/controls.h>
>>  #include <libcamera/formats.h>
>>  #include <libcamera/stream.h>
>> @@ -339,6 +341,7 @@ void SoftwareIsp::stop()
>>  	ispWorkerThread_.wait();
>>  
>>  	running_ = false;
>> +	Thread::current()->dispatchMessages(Message::Type::InvokeMessage, this);
>
> I'd add a blank line here.
>
> I think you can also swap the two lines. Dispatching the messages first
> will ensure that any queued signal that indicates completion of a buffer
> will be delivered to the pipeline handler. That may push one extra frame
> to the application. On the other hand, as we're stopping the ISP, maybe
> the application doesn't care.
>
> I wonder if we could actually drop the running_ variable now. 

I think we can and it works for me.  I'll do it in v3.

> By dispatching all pending messages, we ensure that no new calls to
> the IPA will be made from signal handlers invoked from the
> dispatchMessages() call from within ipa_->stop().
>
>>  	ipa_->stop();
>>  
>>  	for (auto buffer : queuedOutputBuffers_) {
>> diff --git a/utils/codegen/ipc/generators/libcamera_templates/proxy_functions.tmpl b/utils/codegen/ipc/generators/libcamera_templates/proxy_functions.tmpl
>> index b5797b14..25476990 100644
>> --- a/utils/codegen/ipc/generators/libcamera_templates/proxy_functions.tmpl
>> +++ b/utils/codegen/ipc/generators/libcamera_templates/proxy_functions.tmpl
>> @@ -34,7 +34,7 @@
>>  	thread_.exit();
>>  	thread_.wait();
>>  
>> -	Thread::current()->dispatchMessages(Message::Type::InvokeMessage);
>> +	Thread::current()->dispatchMessages(Message::Type::InvokeMessage, this);
>>  
>>  	state_ = ProxyStopped;
>>  {%- endmacro -%}