zbus: Improve consistency with runtime observers

A previous PR merged (to remove runtime observers' dependency with heap)
added inconsistencies and compatibility breaks to the zbus. This commit
improves that by removing the inconsistencies and still attending to the
features requested by the community.

Signed-off-by: Rodrigo Peixoto <[email protected]>

Author: rodrigopex

include/zephyr/zbus/zbus.h

@@ -883,10 +883,11 @@ struct zbus_observer_node {
  * @retval 0 Observer added to the channel.
  * @retval -EALREADY The observer is already present in the channel's runtime observers list.
  * @retval -EAGAIN Waiting period timed out.
+ * @retval -ENOMEM No memory available for a new runtime observer node.
  * @retval -EINVAL Some parameter is invalid.
  */
 int zbus_chan_add_obs(const struct zbus_channel *chan, const struct zbus_observer *obs,
-		      struct zbus_observer_node *node, k_timeout_t timeout);
+		      k_timeout_t timeout);
 
 /**
  * @brief Remove an observer from a channel.

subsys/zbus/Kconfig

@@ -9,6 +9,10 @@ menuconfig ZBUS
 
 if ZBUS
 
+config ZBUS_PREFER_DYNAMIC_ALLOCATION
+	bool "Set zbus to work on static allocation only"
+	default y if HEAP_MEM_POOL_SIZE > 0
+
 config ZBUS_CHANNELS_SYS_INIT_PRIORITY
 	default 5
 	int "The priority used during the SYS_INIT procedure."
@@ -33,7 +37,8 @@ if ZBUS_MSG_SUBSCRIBER
 
 choice ZBUS_MSG_SUBSCRIBER_BUF_ALLOC
 	prompt "ZBus msg_subscribers buffer allocation"
-	default ZBUS_MSG_SUBSCRIBER_BUF_ALLOC_DYNAMIC
+	default ZBUS_MSG_SUBSCRIBER_BUF_ALLOC_DYNAMIC if ZBUS_PREFER_DYNAMIC_ALLOCATION
+	default ZBUS_MSG_SUBSCRIBER_BUF_ALLOC_STATIC
 
 config ZBUS_MSG_SUBSCRIBER_BUF_ALLOC_DYNAMIC
 	bool "Use heap to allocate msg_subscriber buffers data"
@@ -63,6 +68,28 @@ endif # ZBUS_MSG_SUBSCRIBER
 config ZBUS_RUNTIME_OBSERVERS
 	bool "Runtime observers support."
 
+if ZBUS_RUNTIME_OBSERVERS
+
+choice ZBUS_RUNTIME_OBSERVERS_NODE_ALLOC
+	prompt "ZBus runtime observers node allocation"
+	default ZBUS_RUNTIME_OBSERVERS_NODE_ALLOC_DYNAMIC if ZBUS_PREFER_DYNAMIC_ALLOCATION
+	default ZBUS_RUNTIME_OBSERVERS_NODE_ALLOC_STATIC
+
+config ZBUS_RUNTIME_OBSERVERS_NODE_ALLOC_DYNAMIC
+	bool "Use heap to allocate runtime observers node"
+
+config ZBUS_RUNTIME_OBSERVERS_NODE_ALLOC_STATIC
+	bool "Use a pool of runtime observers nodes"
+
+endchoice
+
+config ZBUS_RUNTIME_OBSERVERS_NODE_POOL_SIZE
+	int "Runtime observer pool size"
+	depends on ZBUS_RUNTIME_OBSERVERS_NODE_ALLOC_STATIC
+	default 8
+
+endif # ZBUS_RUNTIME_OBSERVERS
+
 config ZBUS_PRIORITY_BOOST
 	bool "ZBus priority boost algorithm"
 	default y

subsys/zbus/zbus_runtime_observers.c

@@ -8,17 +8,63 @@
 
 LOG_MODULE_DECLARE(zbus, CONFIG_ZBUS_LOG_LEVEL);
 
+#if defined(CONFIG_ZBUS_RUNTIME_OBSERVERS_NODE_ALLOC_DYNAMIC)
+
+static inline int _zbus_runtime_observer_node_alloc(struct zbus_observer_node **node,
+						    k_timeout_t timeout)
+{
+	ARG_UNUSED(timeout);
+
+	*node = k_malloc(sizeof(struct zbus_observer_node));
+
+	_ZBUS_ASSERT(*node != NULL, "could not allocate observer node the heap is full!");
+
+	if (*node == NULL) {
+		return -ENOMEM;
+	}
+
+	return 0;
+}
+
+static inline void _zbus_runtime_observer_node_free(struct zbus_observer_node *node)
+{
+	k_free(node);
+}
+#else
+
+K_MEM_SLAB_DEFINE_STATIC(_zbus_runtime_observers_slab, sizeof(struct zbus_observer_node),
+			 CONFIG_ZBUS_RUNTIME_OBSERVERS_NODE_POOL_SIZE, 8);
+
+static inline int _zbus_runtime_observer_node_alloc(struct zbus_observer_node **node,
+						    k_timeout_t timeout)
+{
+	int err = k_mem_slab_alloc(&_zbus_runtime_observers_slab, (void **)node, timeout);
+
+	_ZBUS_ASSERT(*node != NULL, "not enough runtime observer nodes in the pool. Increase the "
+				    "ZBUS_RUNTIME_OBSERVERS_NODE_POOL_SIZE");
+
+	return err;
+}
+
+static inline void _zbus_runtime_observer_node_free(struct zbus_observer_node *node)
+{
+	k_mem_slab_free(&_zbus_runtime_observers_slab, (void *)node);
+}
+#endif /* CONFIG_ZBUS_RUNTIME_OBSERVERS_NODE_ALLOC_DYNAMIC */
+
 int zbus_chan_add_obs(const struct zbus_channel *chan, const struct zbus_observer *obs,
-		      struct zbus_observer_node *node, k_timeout_t timeout)
+		      k_timeout_t timeout)
 {
 	int err;
+
 	struct zbus_observer_node *obs_nd, *tmp;
 	struct zbus_channel_observation *observation;
 
 	_ZBUS_ASSERT(!k_is_in_isr(), "ISR blocked");
 	_ZBUS_ASSERT(chan != NULL, "chan is required");
 	_ZBUS_ASSERT(obs != NULL, "obs is required");
-	_ZBUS_ASSERT(node != NULL, "node is required");
+
+	k_timepoint_t end_time = sys_timepoint_calc(timeout);
 
 	err = k_sem_take(&chan->data->sem, timeout);
 	if (err) {
@@ -47,9 +93,19 @@ int zbus_chan_add_obs(const struct zbus_channel *chan, const struct zbus_observe
 		}
 	}
 
-	node->obs = obs;
+	struct zbus_observer_node *new_obs_nd = NULL;
 
-	sys_slist_append(&chan->data->observers, &node->node);
+	err = _zbus_runtime_observer_node_alloc(&new_obs_nd, sys_timepoint_timeout(end_time));
+
+	if (err) {
+		k_sem_give(&chan->data->sem);
+
+		return err;
+	}
+
+	new_obs_nd->obs = obs;
+
+	sys_slist_append(&chan->data->observers, &new_obs_nd->node);
 
 	k_sem_give(&chan->data->sem);
 
@@ -76,6 +132,8 @@ int zbus_chan_rm_obs(const struct zbus_channel *chan, const struct zbus_observer
 		if (obs_nd->obs == obs) {
 			sys_slist_remove(&chan->data->observers, &prev_obs_nd->node, &obs_nd->node);
 
+			_zbus_runtime_observer_node_free(obs_nd);
+
 			k_sem_give(&chan->data->sem);
 
 			return 0;