How to use Zephyr zbus to Communicate Between Threads

If you’re not using zbus, you probably should be. As the name indicates, this is the Zephyr bus, a built-in system for publishing, reading, subscribing to, and observing messages in a thread-safe way. It delivers peace of mind with minimal effort.

Think of zbus as a collection of data producers and data consumers. One thread might be periodically taking sensor readings (producer). When that thread publishes to a zbus channel, all listeners (consumers) have a callback that runs to process the new message. You could make Golioth Stream service one of your consumers, pushing sensor data to the cloud each time a new zbus message is received. At the same time, another consumer could update your device’s screen with data from each new message. This example may seem trivial, but as your project grows in complexity you’ll be glad you used zbus.

Zbus overview

Your Zbus experience begins by defining a channel. Each channel has a set message type (a struct), a starting message value, and a list of observers. Optionally each channel may specify a function to validate each message before publishing, and a pointer to user data. In addition to the observers that were set up when the channel was defined, a zbus channel may be read from at any time, making the most recently published message data available.

Zbus breaks down “observers” in a few different categories. Listeners receive a callback with data from each new message. Subscribers may choose to receive a notification of new message. The subscriber is responsible for choosing when to read the message after receiving the notification. There is also an option for subscribers to receive a copy of each message via a FIFO.

There are several benefits to the zbus system. Chief among them is thread-safe data handling. Each channel stores its own message and arbitrates access to the data so that you never have a situation where data is changed while it is being read my another thread. Another notable benefit is the ability for many consumers to read data from a single producer. If you added more threads or functions that need to access a channel message, the new code can read messages from the zbus subsystem directly without needing to change existing code.

The rest of this post will walk through how the Golioth Coldchain Reference Design to uses zbus to:

1. Publish sensor readings on the zbus at 1Hz
2. When recording location data, read the latest sensor value from zbus

Enable the zbus subsystem

As with all Zephyr libraries, use Kconfig to enable zbus by adding the following to your project’s prj.conf file:

CONFIG_ZBUS=y

Then just add the appropriate header file whenever you need to call zbus functions:

#include <zephyr/zbus/zbus.h>

Define a zbus channel

Each zbus channel has a defined message type so this definition includes a struct to use as the message. The state of the message must be initialized so I fill my sensor readings with obviously invalid data that may be checked for by the consumers.

struct weather_data {
    struct sensor_value tem;
    struct sensor_value pre;
    struct sensor_value hum;
};

#define ERROR_VAL1 999
#define ERROR_VAL2 999999

ZBUS_CHAN_DEFINE(weather_chan, struct weather_data, NULL, NULL, ZBUS_OBS_DECLARE(),
         ZBUS_MSG_INIT(.tem.val1 = ERROR_VAL1, .tem.val2 = ERROR_VAL2,
                   .pre.val1 = ERROR_VAL1, .pre.val2 = ERROR_VAL2,
                   .hum.val1 = ERROR_VAL1, .hum.val2 = ERROR_VAL2));

In the code snippet above I’ve named my new zbus channel weather_chan. The message type for this channel is struct weather_data, an agreed upon format that both the producers and consumers will have access to.

I am not using a validator function or user data so both of those params have been set to NULL. In this example I don’t have any observers so I added an empty initializer. The final param is a macro that initializes the channel with my obviously invalid data that indicates a sensor error. This will be replaced as soon as the first message is published.

Publishing a zbus message

Publishing to zbus is a simple one-liner with an error check.

err = zbus_chan_pub(&weather_chan, &reading, K_MSEC(100));
if (err != 0) {
    LOG_ERR("Failed to publish sensor data: %d", err);
    return;
}

The publish command begins with the channel name &weather_chan which we defined in the previous section. The message itself (&reading) is the address of a weather_data struct containing the sensor readings I want to publish. The zbus subsystem will copy this struct data so its fine to free that memory after calling the publish function. Since zbus arbitrates access to read/write a message, the final param is a timeout value in case another thread is currently accessing this zbus channel.

Reading a zbus message

Reading from zbus is very similar to publishing to it.

struct weather_data w_data;

err = zbus_chan_read(&weather_chan, &w_data, K_MSEC(50));
if (err) {
    LOG_ERR("Cannot access weather data: %d", err);
}

Just as with the publish operation, we begin a ready by passing the channel name &weather_chan. The address of a struct matching the zbus channel type is given to receive the data from the zbus message. And finally we use a timeout value to account for any concurrent access issues.

Keep going with observers

Today’s post is just a taste of what zbus can deliver. To get a full picture I encourage you to try out the different observer options. You can register callbacks, receive notifications that a new message is available, or receive a copy of each new message in a FIFO buffer, depending on what type of observer you use.

For more on this, check out the Zbus Concepts section of the Zephyr docs.