zbus: Introduce Async Listener

doc/services/zbus/index.rst

@@ -69,6 +69,8 @@ Another essential aspect of zbus is the observers. There are three types of obse
 
 * Listeners, a callback that the event dispatcher executes every time an observed channel is
   published or notified;
+* Async Listeners, a callback that the event dispatcher schedules to execute in a work
+  queue (system work queue by default) every time an observed channel is published or notified;
 * Subscriber, a thread-based observer that relies internally on a message queue where the event
   dispatcher puts a changed channel's reference every time an observed channel is published or
   notified. Note this kind of observer does not receive the message itself. It should read the
@@ -386,26 +388,27 @@ kind of need.
 Delivery guarantees
 -------------------
 
-ZBus always delivers the messages to the listeners and message subscribers. However, there are no
+ZBus always delivers the messages to the listeners, message subscribers, and async listeners. However, there are no
 message delivery guarantees for subscribers because zbus only sends the notification, but the
 message reading depends on the subscriber's implementation. It is possible to increase the delivery
 rate by following design tips:
 
-* Keep the listeners quick-as-possible (deal with them as ISRs). If some processing is needed,
-  consider submitting a work item to a work-queue;
-* Try to give producers a high priority to avoid losses;
-* Leave spare CPU for observers to consume data produced;
+* Keep the listeners as quick as possible (deal with them as ISRs). If time-consuming processing is
+  required, consider offloading some or all of it to a work queue using async listeners.
+* Try to give consumers a high priority to avoid losses.
+* Leave spare CPU for observers to consume data produced.
 * Consider using message queues or pipes for intensive byte transfers.
 
 .. warning::
    ZBus uses :zephyr_file:`include/zephyr/net_buf.h` (network buffers) to exchange data with message
    subscribers. Thus, choose carefully the configurations
    :kconfig:option:`CONFIG_ZBUS_MSG_SUBSCRIBER_NET_BUF_POOL_SIZE` and
-   :kconfig:option:`CONFIG_HEAP_MEM_POOL_ADD_SIZE_ZBUS`. They are crucial to a proper VDED execution
-   (delivery guarantee) considering message subscribers. If you want to keep an isolated pool for a
-   specific set of channels, you can use
+   :kconfig:option:`CONFIG_HEAP_MEM_POOL_ADD_SIZE_ZBUS` are crucial to a proper VDED execution
+   (delivery guarantee) considering message subscribers and async listeners. If you want to keep an
+   isolated pool for a specific set of channels, you can use
    :kconfig:option:`CONFIG_ZBUS_MSG_SUBSCRIBER_NET_BUF_POOL_ISOLATION` with a dedicated pool. Look
-   at the :zephyr:code-sample:`zbus-msg-subscriber` to see the isolation in action.
+   at the :zephyr:code-sample:`zbus-msg-subscriber` and :zephyr:code-sample:`zbus-async-listeners`
+   to see the isolation in action.
 
 .. warning::
    Subscribers will receive only the reference of the changing channel. A data loss may be perceived
@@ -434,10 +437,12 @@ The message delivery will follow the precedence:
 Usage
 *****
 
-ZBus operation depends on channels and observers. Therefore, it is necessary to determine its
-message and observers list during the channel definition. A message is a regular C struct; the
-observer can be a subscriber (asynchronous), a message subscriber (asynchronous), or a listener
-(synchronous).
+ZBus operation depends on channels and observers. Therefore, it is necessary to determine the
+channel's message during the channel definition and its list of observers, which can be either
+statically (in the channel definition), using the :c:macro:`ZBUS_CHAN_ADD_OBS` or at runtime (see
+`runtime observers`_). A message is a regular C struct; the observer can be a listener
+(synchronous), an async listener (asynchronous), a subscriber (asynchronous), or a message
+subscriber (asynchronous).
 
 The following code defines and initializes a regular channel and its dependencies. This channel
 exchanges accelerometer data, for example.
@@ -784,6 +789,39 @@ message.
             }
     }
 
+
+Async listeners message access
+------------------------------
+
+Async listeners are implemented by utilizing the established message subscriber infrastructure. They
+are executed in a work queue context rather than in the publisher's context. When using the system
+work queue, the user will experience similar behavior to the regular listeners. Since the system
+work queue is, by default, a cooperative thread with a priority of -1, async listeners would run
+before all other application threads (usually preemptive) after the VDED execution.
+
+Async listeners can access the receiving channel's message directly via the message copy reference
+passed to the callback. Be aware that the message copy is freed just after the async listener
+execution. To access the channel's message, the async listener should only cast to the proper
+constant message format. The following example demonstrates how to access the message from an async
+listener.
+
+
+.. code-block:: c
+
+   static void async_listener_callback(const struct zbus_channel *chan, const void *message)
+   {
+           if (chan != &chan_event) {
+                   LOG_ERR("Unexpected channel");
+                   return;
+           }
+
+           const struct msg_event *msg = message;
+
+           LOG_INF("From async listener -> Evt=%d | %s", msg->type,
+                   k_thread_name_get(k_current_get()));
+   }
+
+
 User Data
 ---------
 It is possible to pass custom data into the channel's ``user_data`` for various purposes, such as
@@ -844,6 +882,7 @@ The following code has the exact behavior of the code in :ref:`reading from a ch
             zbus_chan_finish(&acc_chan);
     }
 
+.. _runtime observers:
 
 Runtime observer registration
 -----------------------------
@@ -889,6 +928,7 @@ available:
   Note there is an example of using a work queue instead of executing the listener as an execution
   option;
 * :zephyr:code-sample:`zbus-msg-subscriber` illustrates how to use message subscribers;
+* :zephyr:code-sample:`zbus-async-listeners` illustrates how to use async listeners;
 * :zephyr:code-sample:`zbus-dyn-channel` demonstrates how to use dynamically allocated exchanging
   data in zbus;
 * :zephyr:code-sample:`zbus-uart-bridge` shows an example of sending the operation of the channel to
@@ -926,12 +966,15 @@ For enabling zbus, it is necessary to enable the :kconfig:option:`CONFIG_ZBUS` o
 Related configuration options:
 
 * :kconfig:option:`CONFIG_ZBUS_PRIORITY_BOOST` zbus Highest Locker Protocol implementation;
+
 * :kconfig:option:`CONFIG_ZBUS_CHANNELS_SYS_INIT_PRIORITY` determine the :c:macro:`SYS_INIT`
   priority used by zbus to organize the channels observations by channel;
 * :kconfig:option:`CONFIG_ZBUS_CHANNEL_NAME` enables the name of channels to be available inside the
   channels metadata. The log uses this information to show the channels' names;
 * :kconfig:option:`CONFIG_ZBUS_OBSERVER_NAME` enables the name of observers to be available inside
   the channels metadata;
+* :kconfig:option:`CONFIG_ZBUS_PREFER_DYNAMIC_ALLOCATION` instructs zbus to
+  use dynamic allocation for its internals. That can be disabled by the user and tunned later;
 * :kconfig:option:`CONFIG_ZBUS_MSG_SUBSCRIBER` enables the message subscriber observer type;
 * :kconfig:option:`CONFIG_ZBUS_MSG_SUBSCRIBER_BUF_ALLOC_DYNAMIC` uses the heap to allocate message
   buffers;
@@ -945,6 +988,7 @@ Related configuration options:
   channels to be transported into a message buffer;
 * :kconfig:option:`CONFIG_HEAP_MEM_POOL_ADD_SIZE_ZBUS` the reserved heap size for ZBus in a whole
   including message buffer allocation;
+* :kconfig:option:`CONFIG_ZBUS_ASYNC_LISTENER` enables the async listener observer type;
 * :kconfig:option:`CONFIG_ZBUS_RUNTIME_OBSERVERS` enables the runtime observer registration;
 * :kconfig:option:`CONFIG_ZBUS_RUNTIME_OBSERVERS_NODE_ALLOC_DYNAMIC` allocate the runtime observers
   dynamically using the heap;

include/zephyr/zbus/zbus.h

@@ -9,6 +9,7 @@
 #include <string.h>
 
 #include <zephyr/kernel.h>
+#include <zephyr/sys/check.h>
 #include <zephyr/sys/iterable_sections.h>
 
 #ifdef __cplusplus
@@ -60,7 +61,8 @@ struct zbus_channel_data {
 #endif /* CONFIG_ZBUS_RUNTIME_OBSERVERS */
 
 #if defined(CONFIG_ZBUS_MSG_SUBSCRIBER_NET_BUF_POOL_ISOLATION) || defined(__DOXYGEN__)
-	/** Net buf pool for message subscribers. It can be either the global or a separated one.
+	/** Net buf pool for message subscribers and async listeners. It can be either the global or
+	 * a separated one.
 	 */
 	struct net_buf_pool *msg_subscriber_pool;
 #endif /* ZBUS_MSG_SUBSCRIBER_NET_BUF_POOL_ISOLATION */
@@ -120,6 +122,7 @@ enum __packed zbus_observer_type {
 	ZBUS_OBSERVER_LISTENER_TYPE,
 	ZBUS_OBSERVER_SUBSCRIBER_TYPE,
 	ZBUS_OBSERVER_MSG_SUBSCRIBER_TYPE,
+	ZBUS_OBSERVER_ASYNC_LISTENER_TYPE,
 };
 
 struct zbus_observer_data {
@@ -171,6 +174,13 @@ struct zbus_observer {
 		 */
 		struct k_fifo *message_fifo;
 #endif /* CONFIG_ZBUS_MSG_SUBSCRIBER */
+
+#if defined(CONFIG_ZBUS_ASYNC_LISTENER) || defined(__DOXYGEN__)
+		/** Observer work. It turns the observer into an async listener. It only
+		 * exists if the @kconfig{CONFIG_ZBUS_ASYNC_LISTENER} is enabled.
+		 */
+		struct k_work *work;
+#endif /* CONFIG_ZBUS_ASYNC_LISTENER */
 	};
 };
 
@@ -286,7 +296,6 @@ struct zbus_channel_observation {
 			   (.observers = SYS_SLIST_STATIC_INIT(                                    \
 				&_CONCAT(_zbus_chan_data_, _name).observers),))                    \
 	};                                                                                         \
-	static K_MUTEX_DEFINE(_CONCAT(_zbus_mutex_, _name));                                       \
 	_ZBUS_CPP_EXTERN const STRUCT_SECTION_ITERABLE(zbus_channel, _name) = {                    \
 		ZBUS_CHANNEL_NAME_INIT(_name) /* Maybe removed */                                  \
 		IF_ENABLED(CONFIG_ZBUS_CHANNEL_ID, (.id = _id,))                                   \
@@ -555,6 +564,114 @@ struct zbus_channel_observation {
  * @param[in] _name The subscriber's name.
  */
 #define ZBUS_MSG_SUBSCRIBER_DEFINE(_name) ZBUS_MSG_SUBSCRIBER_DEFINE_WITH_ENABLE(_name, true)
+
+#if defined(CONFIG_ZBUS_ASYNC_LISTENER)
+/** @cond INTERNAL_HIDDEN */
+struct zbus_async_listener_work {
+	struct k_work work;
+	struct k_fifo *message_fifo;
+	struct k_work_q *queue;
+	void (*callback)(const struct zbus_channel *chan, const void *msg);
+};
+
+void async_listener_work_handler(struct k_work *item);
+/** @endcond */
+
+/* clang-format off */
+/**
+ * @brief Define and initialize an async listener.
+ *
+ * This macro defines an observer of @ref ZBUS_OBSERVER_ASYNC_LISTENER_TYPE type. It defines an
+ * async listener work item that will handle a FIFO where messages are received asynchronously
+ * and initialize the @ref zbus_observer defining the async listener.
+ *
+ * @kconfig_dep{CONFIG_ZBUS_ASYNC_LISTENER}
+ *
+ * @param[in] _name The async listener's name.
+ * @param[in] _cb The async listener's callback function.
+ * @param[in] _enable The async listener's initial state.
+ */
+#define ZBUS_ASYNC_LISTENER_DEFINE_WITH_ENABLE(_name, _cb, _enable)               \
+	static K_FIFO_DEFINE(_zbus_observer_work_fifo_##_name);                   \
+	static struct zbus_async_listener_work _zbus_observer_work_##_name = {    \
+		.work = Z_WORK_INITIALIZER(async_listener_work_handler),          \
+		.message_fifo = &_CONCAT(_zbus_observer_work_fifo_, _name),       \
+		.queue = &k_sys_work_q,                                           \
+		.callback = _cb,                                                  \
+	};                                                                        \
+	static struct zbus_observer_data _CONCAT(_zbus_obs_data_, _name) = {      \
+		.enabled = _enable,                                               \
+		IF_ENABLED(CONFIG_ZBUS_PRIORITY_BOOST, (                          \
+			.priority = ZBUS_MIN_THREAD_PRIORITY,                     \
+		))                                                                \
+	};                                                                        \
+	_ZBUS_CPP_EXTERN const STRUCT_SECTION_ITERABLE(zbus_observer, _name) = {  \
+		ZBUS_OBSERVER_NAME_INIT(_name) /* Name field */                   \
+		.type = ZBUS_OBSERVER_ASYNC_LISTENER_TYPE,                        \
+		.data = &_CONCAT(_zbus_obs_data_, _name),                         \
+		.work = &_CONCAT(_zbus_observer_work_, _name).work,               \
+	}
+
+/**
+ *
+ * @brief Define and initialize an enabled async listener.
+ *
+ * This macro defines an observer of async listener type. The async listeners are defined in the
+ * enabled state with this macro.
+ *
+ * @kconfig_dep{CONFIG_ZBUS_ASYNC_LISTENER}
+ *
+ * @param[in] _name The async listener's name.
+ * @param[in] _cb The async listener's callback function.
+ */
+#define ZBUS_ASYNC_LISTENER_DEFINE(_name, _cb)                    \
+	ZBUS_ASYNC_LISTENER_DEFINE_WITH_ENABLE(_name, _cb, true)
+/* clang-format on */
+
+/**
+ * @brief Set the work queue for an async listener.
+ *
+ * This routine sets the work queue that will be used to execute the async listener's
+ * callback function when a notification is received.
+ *
+ * @kconfig_dep{CONFIG_ZBUS_ASYNC_LISTENER}
+ *
+ * @param[in] obs The async listener observer's reference.
+ * @param[in] queue The work queue to be used for executing the async listener.
+ *
+ * @retval 0 Work queue set successfully.
+ * @retval -ENOENT The observer reference is NULL.
+ * @retval -EINVAL The observer type is not valid.
+ * @retval -EBADF The work queue reference is NULL.
+ */
+static inline int zbus_async_listener_set_work_queue(const struct zbus_observer *obs,
+						     struct k_work_q *queue)
+{
+	CHECKIF(obs == NULL) {
+		return -EINVAL;
+	}
+
+	CHECKIF(obs->type != ZBUS_OBSERVER_ASYNC_LISTENER_TYPE) {
+		return -EINVAL;
+	}
+
+	CHECKIF(queue == NULL) {
+		return -EINVAL;
+	}
+
+	static struct k_spinlock zbus_async_listener_slock;
+
+	K_SPINLOCK(&zbus_async_listener_slock) {
+		struct zbus_async_listener_work *async_listener =
+			CONTAINER_OF(obs->work, struct zbus_async_listener_work, work);
+
+		async_listener->queue = queue;
+	}
+	return 0;
+}
+
+#endif /* CONFIG_ZBUS_ASYNC_LISTENER */
+
 /**
  *
  * @brief Publish to a channel
@@ -883,7 +1000,7 @@ static inline void zbus_chan_pub_stats_update(const struct zbus_channel *chan)
 #if defined(CONFIG_ZBUS_RUNTIME_OBSERVERS) || defined(__DOXYGEN__)
 
 /**
- * @brief Structure used to register runtime obeservers
+ * @brief Structure used to register runtime observers
  *
  */
 struct zbus_observer_node {

samples/subsys/zbus/async_listeners/CMakeLists.txt

@@ -0,0 +1,8 @@
+# SPDX-License-Identifier: Apache-2.0
+cmake_minimum_required(VERSION 3.20.0)
+
+find_package(Zephyr REQUIRED HINTS $ENV{ZEPHYR_BASE})
+project(async_listener)
+
+FILE(GLOB app_sources src/*.c)
+target_sources(app PRIVATE ${app_sources})

samples/subsys/zbus/async_listeners/README.rst

@@ -0,0 +1,60 @@
+.. zephyr:code-sample:: zbus-async-listeners
+   :name: zbus Async Listeners
+   :relevant-api: zbus_apis
+
+   Demonstrate the usage of zbus async listeners.
+
+Overview
+********
+This sample demonstrates how to use zbus asynchronous listeners by implementing a simple
+application. The main thread submits an event to a channel that has three listeners: (i) a
+conventional listener, (ii) an asynchronous listener that utilizes the system's work queue, and
+(iii) an asynchronous listener that employs an isolated work queue.
+
+
+Building and Running
+********************
+
+This project outputs to the console.  It can be built and executed
+on QEMU as follows:
+
+.. zephyr-app-commands::
+   :zephyr-app: samples/subsys/zbus/async_listeners/
+   :host-os: unix
+   :board: qemu_x86
+   :goals: run
+
+Sample Output
+=============
+
+.. code-block:: console
+
+   I: Output                       | Thread Context
+   I: =============================================
+   I: From listener -> Evt=0       | main
+   I: From async listener -> Evt=0 | sysworkq
+   I: From async listener -> Evt=0 | My work queue
+   I: From listener -> Evt=1       | main
+   I: From async listener -> Evt=1 | sysworkq
+   I: From async listener -> Evt=1 | My work queue
+   I: From listener -> Evt=2       | main
+   I: From async listener -> Evt=2 | sysworkq
+   I: From async listener -> Evt=2 | My work queue
+   I: From listener -> Evt=3       | main
+   I: From async listener -> Evt=3 | sysworkq
+   I: From async listener -> Evt=3 | My work queue
+   I: =============================================
+   I: From listener -> Evt=0       | main
+   I: From async listener -> Evt=0 | sysworkq
+   I: From async listener -> Evt=0 | My work queue
+   I: From listener -> Evt=1       | main
+   I: From async listener -> Evt=1 | sysworkq
+   I: From async listener -> Evt=1 | My work queue
+   I: From listener -> Evt=2       | main
+   I: From async listener -> Evt=2 | sysworkq
+   I: From async listener -> Evt=2 | My work queue
+   I: From listener -> Evt=3       | main
+   I: From async listener -> Evt=3 | sysworkq
+   I: From async listener -> Evt=3 | My work queue
+
+Exit QEMU by pressing :kbd:`CTRL+A` :kbd:`x`.

samples/subsys/zbus/async_listeners/prj.conf

@@ -0,0 +1,12 @@
+CONFIG_LOG=y
+CONFIG_THREAD_NAME=y
+CONFIG_ASSERT=y
+CONFIG_LOG_MODE_MINIMAL=y
+CONFIG_BOOT_BANNER=n
+CONFIG_MAIN_THREAD_PRIORITY=5
+CONFIG_ZBUS=y
+CONFIG_ZBUS_LOG_LEVEL_INF=y
+CONFIG_ZBUS_CHANNEL_NAME=y
+CONFIG_ZBUS_OBSERVER_NAME=y
+CONFIG_ZBUS_ASYNC_LISTENER=y
+CONFIG_HEAP_MEM_POOL_SIZE=512