zbus.h

Go to the documentation of this file.

/*
 * Copyright (c) 2022 Rodrigo Peixoto <[email protected]>
 * SPDX-License-Identifier: Apache-2.0
 */
 
#ifndef ZEPHYR_INCLUDE_ZBUS_H_
#define ZEPHYR_INCLUDE_ZBUS_H_
 
#include <string.h>
 
#include <zephyr/kernel.h>
#include <zephyr/sys/iterable_sections.h>
 
#ifdef __cplusplus
extern "C" {
#endif
 
struct zbus_channel_data {
 int16_t observers_start_idx;
 
 int16_t observers_end_idx;
 
 struct k_sem sem;
 
#if defined(CONFIG_ZBUS_PRIORITY_BOOST)
 int highest_observer_priority;
#endif /* CONFIG_ZBUS_PRIORITY_BOOST */
 
#if defined(CONFIG_ZBUS_RUNTIME_OBSERVERS) || defined(__DOXYGEN__)
 sys_slist_t observers;
#endif /* CONFIG_ZBUS_RUNTIME_OBSERVERS */
};
 
struct zbus_channel {
#if defined(CONFIG_ZBUS_CHANNEL_NAME) || defined(__DOXYGEN__)
 const char *const name;
#endif
 void *const message;
 
 const size_t message_size;
 
 void *const user_data;
 
 bool (*const validator)(const void *msg, size_t msg_size);
 
 struct zbus_channel_data *const data;
};
 
enum __packed zbus_observer_type {
 ZBUS_OBSERVER_LISTENER_TYPE,
 ZBUS_OBSERVER_SUBSCRIBER_TYPE,
 ZBUS_OBSERVER_MSG_SUBSCRIBER_TYPE,
};
 
struct zbus_observer_data {
 bool enabled;
 
#if defined(CONFIG_ZBUS_PRIORITY_BOOST)
 int priority;
#endif /* CONFIG_ZBUS_PRIORITY_BOOST */
};
 
struct zbus_observer {
#if defined(CONFIG_ZBUS_OBSERVER_NAME) || defined(__DOXYGEN__)
 const char *const name;
#endif
 enum zbus_observer_type type;
 
 struct zbus_observer_data *const data;
 
 union {
 struct k_msgq *const queue;
 
 void (*const callback)(const struct zbus_channel *chan);
 
#if defined(CONFIG_ZBUS_MSG_SUBSCRIBER) || defined(__DOXYGEN__)
 struct k_fifo *const message_fifo;
#endif /* CONFIG_ZBUS_MSG_SUBSCRIBER */
 };
};
 
struct zbus_channel_observation_mask {
 bool enabled;
};
 
struct zbus_channel_observation {
 const struct zbus_channel *const chan;
 const struct zbus_observer *const obs;
};
 
#ifdef __cplusplus
#define _ZBUS_CPP_EXTERN extern
#else
#define _ZBUS_CPP_EXTERN
#endif /* __cplusplus */
 
#define ZBUS_MIN_THREAD_PRIORITY (CONFIG_NUM_PREEMPT_PRIORITIES - 1)
 
#if defined(CONFIG_ZBUS_ASSERT_MOCK)
#define _ZBUS_ASSERT(_cond, _fmt, ...) \
 do { \
 if (!(_cond)) { \
 printk("ZBUS ASSERT: "); \
 printk(_fmt, ##__VA_ARGS__); \
 printk("\n"); \
 return -EFAULT; \
 } \
 } while (0)
#else
#define _ZBUS_ASSERT(_cond, _fmt, ...) __ASSERT(_cond, _fmt, ##__VA_ARGS__)
#endif
 
#if defined(CONFIG_ZBUS_CHANNEL_NAME)
#define ZBUS_CHANNEL_NAME_INIT(_name) .name = #_name,
#define _ZBUS_CHAN_NAME(_chan) (_chan)->name
#else
#define ZBUS_CHANNEL_NAME_INIT(_name)
#define _ZBUS_CHAN_NAME(_chan) ""
#endif
 
#if defined(CONFIG_ZBUS_OBSERVER_NAME)
#define ZBUS_OBSERVER_NAME_INIT(_name) .name = #_name,
#define _ZBUS_OBS_NAME(_obs) (_obs)->name
#else
#define ZBUS_OBSERVER_NAME_INIT(_name)
#define _ZBUS_OBS_NAME(_obs) ""
#endif
 
#if defined(CONFIG_ZBUS_RUNTIME_OBSERVERS)
#define ZBUS_RUNTIME_OBSERVERS_LIST_DECL(_slist_name) static sys_slist_t _slist_name
#define ZBUS_RUNTIME_OBSERVERS_LIST_INIT(_slist_name) .runtime_observers = &_slist_name,
#else
#define ZBUS_RUNTIME_OBSERVERS_LIST_DECL(_slist_name)
#define ZBUS_RUNTIME_OBSERVERS_LIST_INIT(_slist_name) /* No runtime observers */
#endif
 
#define _ZBUS_OBS_EXTERN(_name) extern struct zbus_observer _name
 
#define _ZBUS_CHAN_EXTERN(_name) extern const struct zbus_channel _name
 
#define ZBUS_REF(_value) &(_value)
 
#define FOR_EACH_FIXED_ARG_NONEMPTY_TERM(F, sep, fixed_arg, ...) \
 COND_CODE_0(/* are there zero non-empty arguments ? */ \
 NUM_VA_ARGS_LESS_1( \
 LIST_DROP_EMPTY(__VA_ARGS__, _)), /* if so, expand to nothing */ \
 (), /* otherwise, expand to: */ \
 (FOR_EACH_IDX_FIXED_ARG( \
 F, sep, fixed_arg, \
 LIST_DROP_EMPTY(__VA_ARGS__)) /* plus a final terminator */ \
 __DEBRACKET sep))
 
#define _ZBUS_OBSERVATION_PREFIX(_idx) \
 GET_ARG_N(_idx, 00, 01, 02, 03, 04, 05, 06, 07, 08, 09, 10, 11, 12, 13, 14, 15, 16, 17, \
 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, \
 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, \
 58, 59, 60, 61, 62, 63)
 
#define _ZBUS_CHAN_OBSERVATION(_idx, _obs, _chan) \
 const STRUCT_SECTION_ITERABLE( \
 zbus_channel_observation, \
 _CONCAT(_chan, _ZBUS_OBSERVATION_PREFIX(UTIL_INC(_idx)))) = {.chan = &_chan, \
 .obs = &_obs}; \
 STRUCT_SECTION_ITERABLE(zbus_channel_observation_mask, \
 _CONCAT(_CONCAT(_chan, _ZBUS_OBSERVATION_PREFIX(UTIL_INC(_idx))), \
 _mask)) = {.enabled = false};
 
#if defined(CONFIG_ZBUS_RUNTIME_OBSERVERS) || defined(__DOXYGEN__)
#define _ZBUS_RUNTIME_OBSERVERS(_name) .observers = &(_CONCAT(_observers_, _name)),
#define _ZBUS_RUNTIME_OBSERVERS_DECL(_name) static sys_slist_t _CONCAT(_observers_, _name);
#else
#define _ZBUS_RUNTIME_OBSERVERS(_name)
#define _ZBUS_RUNTIME_OBSERVERS_DECL(_name)
#endif /* CONFIG_ZBUS_RUNTIME_OBSERVERS */
 
/* clang-format off */
#define ZBUS_CHAN_ADD_OBS_WITH_MASK(_chan, _obs, _masked, _prio) \
 const STRUCT_SECTION_ITERABLE(zbus_channel_observation, \
 _CONCAT(_CONCAT(_chan, zz), _CONCAT(_prio, _obs))) = { \
 .chan = &_chan, \
 .obs = &_obs, \
 }; \
 STRUCT_SECTION_ITERABLE(zbus_channel_observation_mask, \
 _CONCAT(_CONCAT(_CONCAT(_chan, zz), _CONCAT(_prio, _obs)), \
 _mask)) = {.enabled = _masked}
/* clang-format on */
 
#define ZBUS_CHAN_ADD_OBS(_chan, _obs, _prio) ZBUS_CHAN_ADD_OBS_WITH_MASK(_chan, _obs, false, _prio)
 
#define ZBUS_OBS_DECLARE(...) FOR_EACH_NONEMPTY_TERM(_ZBUS_OBS_EXTERN, (;), __VA_ARGS__)
 
#define ZBUS_CHAN_DECLARE(...) FOR_EACH(_ZBUS_CHAN_EXTERN, (;), __VA_ARGS__)
 
#define ZBUS_OBSERVERS_EMPTY
 
#define ZBUS_OBSERVERS(...) __VA_ARGS__
 
/* clang-format off */
#define ZBUS_CHAN_DEFINE(_name, _type, _validator, _user_data, _observers, _init_val) \
 static _type _CONCAT(_zbus_message_, _name) = _init_val; \
 static struct zbus_channel_data _CONCAT(_zbus_chan_data_, _name) = { \
 .observers_start_idx = -1, \
 .observers_end_idx = -1, \
 IF_ENABLED(CONFIG_ZBUS_PRIORITY_BOOST, ( \
 .highest_observer_priority = ZBUS_MIN_THREAD_PRIORITY, \
 )) \
 }; \
 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 */ \
 .message = &_CONCAT(_zbus_message_, _name), \
 .message_size = sizeof(_type), \
 .user_data = _user_data, \
 .validator = _validator, \
 .data = &_CONCAT(_zbus_chan_data_, _name), \
 }; \
 /* Extern declaration of observers */ \
 ZBUS_OBS_DECLARE(_observers); \
 /* Create all channel observations from observers list */ \
 FOR_EACH_FIXED_ARG_NONEMPTY_TERM(_ZBUS_CHAN_OBSERVATION, (;), _name, _observers)
/* clang-format on */
 
#define ZBUS_MSG_INIT(_val, ...) \
 { \
 _val, ##__VA_ARGS__ \
 }
 
/* clang-format off */
#define ZBUS_SUBSCRIBER_DEFINE_WITH_ENABLE(_name, _queue_size, _enable) \
 K_MSGQ_DEFINE(_zbus_observer_queue_##_name, \
 sizeof(const struct zbus_channel *), \
 _queue_size, sizeof(const struct zbus_channel *) \
 ); \
 static struct zbus_observer_data _CONCAT(_zbus_obs_data_, _name) = { \
 .enabled = _enable, \
 IF_ENABLED(CONFIG_ZBUS_PRIORITY_BOOST, ( \
 .priority = ZBUS_MIN_THREAD_PRIORITY, \
 )) \
 }; \
 STRUCT_SECTION_ITERABLE(zbus_observer, _name) = { \
 ZBUS_OBSERVER_NAME_INIT(_name) /* Name field */ \
 .type = ZBUS_OBSERVER_SUBSCRIBER_TYPE, \
 .data = &_CONCAT(_zbus_obs_data_, _name), \
 .queue = &_zbus_observer_queue_##_name, \
 }
/* clang-format on */
 
#define ZBUS_SUBSCRIBER_DEFINE(_name, _queue_size) \
 ZBUS_SUBSCRIBER_DEFINE_WITH_ENABLE(_name, _queue_size, true)
 
/* clang-format off */
#define ZBUS_LISTENER_DEFINE_WITH_ENABLE(_name, _cb, _enable) \
 static struct zbus_observer_data _CONCAT(_zbus_obs_data_, _name) = { \
 .enabled = _enable, \
 IF_ENABLED(CONFIG_ZBUS_PRIORITY_BOOST, ( \
 .priority = ZBUS_MIN_THREAD_PRIORITY, \
 )) \
 }; \
 STRUCT_SECTION_ITERABLE(zbus_observer, _name) = { \
 ZBUS_OBSERVER_NAME_INIT(_name) /* Name field */ \
 .type = ZBUS_OBSERVER_LISTENER_TYPE, \
 .data = &_CONCAT(_zbus_obs_data_, _name), \
 .callback = (_cb) \
 }
/* clang-format on */
 
#define ZBUS_LISTENER_DEFINE(_name, _cb) ZBUS_LISTENER_DEFINE_WITH_ENABLE(_name, _cb, true)
 
/* clang-format off */
#define ZBUS_MSG_SUBSCRIBER_DEFINE_WITH_ENABLE(_name, _enable) \
 static K_FIFO_DEFINE(_zbus_observer_fifo_##_name); \
 static struct zbus_observer_data _CONCAT(_zbus_obs_data_, _name) = { \
 .enabled = _enable, \
 IF_ENABLED(CONFIG_ZBUS_PRIORITY_BOOST, ( \
 .priority = ZBUS_MIN_THREAD_PRIORITY, \
 )) \
 }; \
 STRUCT_SECTION_ITERABLE(zbus_observer, _name) = { \
 ZBUS_OBSERVER_NAME_INIT(_name) /* Name field */ \
 .type = ZBUS_OBSERVER_MSG_SUBSCRIBER_TYPE, \
 .data = &_CONCAT(_zbus_obs_data_, _name), \
 .message_fifo = &_zbus_observer_fifo_##_name, \
 }
/* clang-format on */
 
#define ZBUS_MSG_SUBSCRIBER_DEFINE(_name) ZBUS_MSG_SUBSCRIBER_DEFINE_WITH_ENABLE(_name, true)
int zbus_chan_pub(const struct zbus_channel *chan, const void *msg, k_timeout_t timeout);
 
int zbus_chan_read(const struct zbus_channel *chan, void *msg, k_timeout_t timeout);
 
int zbus_chan_claim(const struct zbus_channel *chan, k_timeout_t timeout);
 
int zbus_chan_finish(const struct zbus_channel *chan);
 
int zbus_chan_notify(const struct zbus_channel *chan, k_timeout_t timeout);
 
#if defined(CONFIG_ZBUS_CHANNEL_NAME) || defined(__DOXYGEN__)
 
static inline const char *zbus_chan_name(const struct zbus_channel *chan)
{
 __ASSERT(chan != NULL, "chan is required");
 
 return chan->name;
}
 
#endif
 
static inline void *zbus_chan_msg(const struct zbus_channel *chan)
{
 __ASSERT(chan != NULL, "chan is required");
 
 return chan->message;
}
 
static inline const void *zbus_chan_const_msg(const struct zbus_channel *chan)
{
 __ASSERT(chan != NULL, "chan is required");
 
 return chan->message;
}
 
static inline uint16_t zbus_chan_msg_size(const struct zbus_channel *chan)
{
 __ASSERT(chan != NULL, "chan is required");
 
 return chan->message_size;
}
 
static inline void *zbus_chan_user_data(const struct zbus_channel *chan)
{
 __ASSERT(chan != NULL, "chan is required");
 
 return chan->user_data;
}
 
#if defined(CONFIG_ZBUS_RUNTIME_OBSERVERS) || defined(__DOXYGEN__)
 
int zbus_chan_add_obs(const struct zbus_channel *chan, const struct zbus_observer *obs,
 k_timeout_t timeout);
 
int zbus_chan_rm_obs(const struct zbus_channel *chan, const struct zbus_observer *obs,
 k_timeout_t timeout);
 
struct zbus_observer_node {
 sys_snode_t node;
 const struct zbus_observer *obs;
};
 
#endif /* CONFIG_ZBUS_RUNTIME_OBSERVERS */
 
int zbus_obs_set_enable(struct zbus_observer *obs, bool enabled);
 
static inline int zbus_obs_is_enabled(struct zbus_observer *obs, bool *enable)
{
 _ZBUS_ASSERT(obs != NULL, "obs is required");
 _ZBUS_ASSERT(enable != NULL, "enable is required");
 
 *enable = obs->data->enabled;
 
 return 0;
}
 
int zbus_obs_set_chan_notification_mask(const struct zbus_observer *obs,
 const struct zbus_channel *chan, bool masked);
 
int zbus_obs_is_chan_notification_masked(const struct zbus_observer *obs,
 const struct zbus_channel *chan, bool *masked);
 
#if defined(CONFIG_ZBUS_OBSERVER_NAME) || defined(__DOXYGEN__)
 
static inline const char *zbus_obs_name(const struct zbus_observer *obs)
{
 __ASSERT(obs != NULL, "obs is required");
 
 return obs->name;
}
 
#endif
 
#if defined(CONFIG_ZBUS_PRIORITY_BOOST) || defined(__DOXYGEN__)
 
int zbus_obs_attach_to_thread(const struct zbus_observer *obs);
 
int zbus_obs_detach_from_thread(const struct zbus_observer *obs);
 
#endif /* CONFIG_ZBUS_PRIORITY_BOOST */
 
int zbus_sub_wait(const struct zbus_observer *sub, const struct zbus_channel **chan,
 k_timeout_t timeout);
 
#if defined(CONFIG_ZBUS_MSG_SUBSCRIBER) || defined(__DOXYGEN__)
 
int zbus_sub_wait_msg(const struct zbus_observer *sub, const struct zbus_channel **chan, void *msg,
 k_timeout_t timeout);
 
#endif /* CONFIG_ZBUS_MSG_SUBSCRIBER */
 
bool zbus_iterate_over_channels(bool (*iterator_func)(const struct zbus_channel *chan));
bool zbus_iterate_over_channels_with_user_data(
 bool (*iterator_func)(const struct zbus_channel *chan, void *user_data), void *user_data);
 
bool zbus_iterate_over_observers(bool (*iterator_func)(const struct zbus_observer *obs));
bool zbus_iterate_over_observers_with_user_data(
 bool (*iterator_func)(const struct zbus_observer *obs, void *user_data), void *user_data);
 
#ifdef __cplusplus
}
#endif
 
#endif /* ZEPHYR_INCLUDE_ZBUS_H_ */