Thread Communication: Message Queues and Mailboxes

Message queue is a FIFO queue that copies fixed-size elements between producers and consumers in a concurrency-safe manner.

Suppose one thread reads a temperature sensor and another displays the value on a screen. The first one needs to pass a float or a structure to the second one.

If we use a global variable protected with a Mutex, it works, but it’s inelegant and error-prone if the data frequency is high. Zephyr offers us much more powerful mechanisms specifically designed for moving data: Message Queues and Mailboxes.

Message queues (k_msgq)

Message Queues are by far the most used mechanism. They work like a FIFO (First-In, First-Out) pipe.

The most important thing about Message Queues in Zephyr is that they copy the data. When you put a variable into the queue, the system makes an exact copy of its value in the internal buffer. This means you can reuse or destroy your local variable immediately after sending it without worry.

Practical Implementation

Let’s pass a complete data structure from one thread to another.

#include <zephyr/kernel.h>
#include <zephyr/logging/log.h>

LOG_MODULE_REGISTER(demo_msgq, LOG_LEVEL_INF);

/* 1. Define the data type to be passed */
struct data_item_t {
    uint32_t id;
    float temperatura;
    float humedad;
};

/* 2. Define the Queue statically */
/* K_MSGQ_DEFINE(name, item_size, max_capacity, alignment) */
/* Here we create a queue capable of holding up to 10 readings */
K_MSGQ_DEFINE(my_msgq, sizeof(struct data_item_t), 10, 1);

/* Producer Thread (Simulates a sensor) */
void producer_thread(void *p1, void *p2, void *p3)
{
    struct data_item_t data;
    data.id = 0;

    while (1) {
        /* Simulate reading */
        data.id++;
        data.temperatura = 20.0f + (data.id % 5); // Invented value
        data.humedad = 50.0f;

        /* 3. Send data to the queue */
        /* If the queue is full, K_NO_WAIT makes it fail immediately.
           We could use K_FOREVER to wait for space to become available. */
        if (k_msgq_put(&my_msgq, &data, K_NO_WAIT) != 0) {
            LOG_WRN("Queue full! Losing data...");
        } else {
            LOG_INF("Sent data ID %u", data.id);
        }

        k_msleep(1000);
    }
}

/* Consumer Thread (Processes or displays data) */
void consumer_thread(void *p1, void *p2, void *p3)
{
    struct data_item_t received_data;

    while (1) {
        /* 4. Read from the queue */
        /* K_FOREVER: The thread sleeps until something arrives. */
        k_msgq_get(&my_msgq, &received_data, K_FOREVER);

        LOG_INF("Received -> Temp: %.1f, Hum: %.1f", 
                received_data.temperatura, received_data.humedad);
    }
}

K_THREAD_DEFINE(producer_id, 1024, producer_thread, NULL, NULL, NULL, 7, 0, 0);
K_THREAD_DEFINE(consumer_id, 1024, consumer_thread, NULL, NULL, NULL, 7, 0, 0);

Each element is delivered to a single consumer. If multiple threads are waiting on the same queue, they share the messages; they do not each receive a copy.

Mailboxes (k_mbox)

If Message Queues are so good, why do we need Mailboxes?

Queues have a limitation: the message size is fixed. If you sometimes want to send a single byte and other times a 1KB image, the queue is inefficient (you would have to dimension it for the largest case).

Mailboxes are more flexible and allow:

Mailboxes offer more control, at the cost of a more complex API and message lifecycle. Furthermore, they only allow exchanges between threads, not from an ISR.

Clearing the Queue (k_msgq_purge)

Sometimes, the queue fills up with old data that is no longer useful (for example, if the consumer thread was hung for a while). We can empty it instantly with:

k_msgq_purge(&my_msgq);

This discards all elements and resets the counters, allowing a fresh start.

In the next article, we’ll see how to delegate work from interrupts using workqueues.