FreeRTOS part 5: queues

This is the fifth post in my FreeRTOS series. So far I have introduced tasks, that simply exist performing on their own with little actual communication between them, while also yielding (moving to Blocked state) by using vTaskDelay, which is just a temporal event without any synchronization. This means that even though there would be something to process, the task is waiting until vTaskDelay timeout is reached.

What are queues really?

FreeRTOS documentation states that

Queues are the primary form of intertask communications. ¹

What does this mean? Let us not jump the queue, but rather take our place in line and work through it from the front.

A queue is (usually) a First-In, First-Out (FIFO) buffer (*)

QueueHandle_t xQueueCreate(UBaseType_t uxQueueLength, UBaseType_t uxItemSize);

When creating a queue a set count of data items (the queue length parameter) of predetermined size (the item size parameter) must be provided.

In its simplest form creating a queue looks like this:

QueueHandle_t queue = xQueueCreate(3, sizeof(int));

And this is how this simple queue operates as a FIFO buffer:

Table 1: Queue as a FIFO buffer.

Do note that the receive and removal are split into two steps here only to show the copy happening before the slot frees. The FreeRTOS kernel does both at once.

(*) It is also possible to make a Last-In, First-Out (LIFO) buffer by sending to the front of the queue and then receiving from the front. ²

Always a copy, sometimes a copy of a pointer

As seen in table 1 above, the values are always copied byte by byte on both send and receive. Therefore, when step 2 sends x = 10; and later step 3 sends x = 20;, the front value does not change from 10 to 20. This is also the reason why, when step 4 receives y == 10;, the value does not change to 20 on step 5 when the item is removed from the queue. As the queue handles the copying, there is no need to worry about values getting lost, e.g. when sending a value that is in local scope, nor about the variable used to send the data being changed afterwards.

Of course there are also use cases when copying is not preferred, such as when a large payload is being moved. In these cases a copy of a pointer’s value can be used instead. The ownership of the data must then be either (a) transferred to the receiving task, or (b) managed with some syncing mechanism so that once the receiver has processed the data, the sender can modify it again.

In its simplest form creating a queue that copies pointers looks like this:

QueueHandle_t queue = xQueueCreate(3, sizeof(int*));

Here is how data ownership changes using the new and delete operators:

Table 2: Queue holding copies of pointers, with ownership handed to the receiver.

Copy by value keeps the two tasks at arm’s length. Nothing is shared, so neither has to care what the other does next. Copy a pointer instead and that independence is gone: only an address rides the queue, so both tasks end up staring at the same chunk of memory, and now someone has to own it, free it, and agree on when. The queue still does its job perfectly, it just stops being the only thing you have to reason about. ²

Producers and consumers

When it comes to queues, there are two roles that a task can have. Producers, such as Task X in tables 1 and 2, send data to a queue. Tasks that then receive the data are called consumers, which is what Task Y in the same tables does. However, these roles can also overlap, so a task may also receive data from one queue, transform it somehow and then pass it on to another queue.

Queues are not direct task-to-task communication, meaning that tasks do not communicate directly between each other, only to a queue. Because of this, one queue can have multiple producers inserting data into it. This flexibility also expands to consumers, so multiple tasks may receive data from a queue. ²

From waiting on time to waiting on data

Now that we have established queues hold data, which is the intertask communication the quote describes, how do they provide synchronization?

Producers and consumers very rarely are in “perfect synchronization” where a producer drops data to a queue and consumer then picks it up almost immediately. There might be situations, such as tables 1 and 2 show, where producer places two items to a queue in burst and then consumer drains them both. But it is also possible that while consumer is working with a first item, producer already has a third item ready for the queue.

Blocking is the synchronization

Whilst a consumer is receiving from a queue, a time to wait value can also be specified. This value determines how long the task is kept in a Blocked state if no data is present. The task is then moved to Ready state once data is available or the time expires. In cases where multiple consumers are waiting in Ready state on the same queue, the task with higher priority goes to Running state and gets the data. If priorities are shared, then the task that was in Blocked state for the longest is picked. ²

This is what implements “from waiting on time to waiting on data”. Instead of using vTaskDelay to wait, for example 1000 milliseconds, and then checking if there is data available, a queue reduces the CPU cycles spent on “polling” the data. There also might be a situation where data is available 100 milliseconds after vTaskDelay, forcing consumer to wait 900 milliseconds before anything is even done with the data. Again using the queue instead makes the system much more reactive and responsive.

BaseType_t xQueueReceive( QueueHandle_t xQueue,
                          void * const pvBuffer,
                          TickType_t xTicksToWait )

Above is the API for reading from a queue. xQueue is the handle that was received when a queue was created, pvBuffer is the pointer to memory where data from the queue is placed and finally xTicksToWait is how long the task stays Blocked. There are two special values that can be used for the final parameter; portMAX_DELAY and 0. The former causes the task to wait indefinitely in the Blocked state if INCLUDE_vTaskSuspend is defined, and the latter disables the blocking completely. ³

Sending to the back, and jumping to the front

The blocking works the same in the other direction. When a queue is full, a producer waiting to send is moved to the Blocked state just as a consumer waits on an empty one, using the same xTicksToWait and the same portMAX_DELAY and 0 special values. What the sending side adds is a choice the receiving side never has: which end of the queue the data goes to. ²

BaseType_t xQueueSendToBack( QueueHandle_t xQueue,
                             const void * pvItemToQueue,
                             TickType_t xTicksToWait );

BaseType_t xQueueSendToFront( QueueHandle_t xQueue,
                              const void * pvItemToQueue,
                              TickType_t xTicksToWait );

Parameter pvItemToQueue is a pointer to data to be copied, other than that the parameters are the same as xQueueReceive. Which function you use sets the ordering: xQueueSendToBack gives FIFO, xQueueSendToFront gives LIFO. However, it is also possible to mix these, so for example, if there is some priority data that needs to be urgently handled, it can be placed in the front of a FIFO implementation. ³

Reacting to returns

Each queue function described above returns a pdPASS upon successfully sending data to / receiving data from a queue. However xQueueReceive returns errQUEUE_EMPTY upon returning without any data (when not waiting indefinitely). So pvBuffer holds no new data once the timeout is hit. In these situations xQueueReceive can be called again, since it might be perfectly normal that producer does not have any data yet. However if no data is ever received, there might be a need for some sort of reaction, e.g. sensor data is not coming, maybe a sensor is not working as expected.

The same goes for the sending functions, which both return errQUEUE_FULL if the data could not be placed in the queue before timeout. Once this happens it means that there is either too much data coming, or the queue size is too small. Either way sometimes it could also be okay to drop data, collect new data and try to place that into the queue. ³

Queue introspection

Queues can also be examined using uxQueueMessagesWaiting and uxQueueSpacesAvailable API functions. Both of them take a queue handle as a parameter and then return a UBaseType_t value (usually an unsigned integer) whose width is defined by the port. The former returns the number of items currently in the queue, the latter how many more can still be added. ⁴

Reading without removing

Back in table 1 the receive was split into two steps, the copy and then the removal, with a note that the kernel really does both when the xQueueReceive function is used. However, there is a function that does only the first half: xQueuePeek. It copies the item at the front, but leaves it on the queue, so the next receive or peek still finds it there. The parameters are identical to xQueueReceive:

BaseType_t xQueuePeek( QueueHandle_t xQueue,
                       void * const pvBuffer,
                       TickType_t xTicksToWait );

Blocking behavior is also identical, so if the queue is empty the task waits up to xTicksToWait, with the same portMAX_DELAY and 0 special values, and the same errQUEUE_EMPTY return when the wait runs out. The only difference is that a successful peek consumes nothing. That is handy when a task wants to inspect the next item before deciding whether to commit to it, and it is the read half of the mailbox pattern below. ⁵

A queue as a mailbox

Everything so far has treated the queue as a stream: each item is produced once and consumed once, and order matters. Sometimes you want the opposite, a single slot with only the latest value that can be read over and over without being emptied. Think of the most recent sensor reading or the current setpoint, where stale items are worthless and only the newest one counts. A single-slot queue gives you exactly that when paired with xQueueOverwrite:

BaseType_t xQueueOverwrite( QueueHandle_t xQueue,
                            const void * pvItemToQueue );

As it is impossible to block, there is no wait time. If the slot is already full it overwrites whatever value is sitting there. Do note that xQueueOverwrite is meant for single-slot queues only, so the pattern is to create the queue with a length of one and then overwrite it. A producer can call it as often as it likes and the queue always holds the latest value. Consumers then use xQueuePeek to sample that value without removing it, so it stays available for the next reader, and the next. ⁵

Getting rid of a queue

Finally, before moving to the code example, the last thing to note is that queues can be deleted with vQueueDelete. It returns nothing, it simply invalidates the handle and frees all the memory the queue allocated. Do note that this frees the queue’s own storage but not anything the items inside it point to. If items own heap memory, drain the queue and free them first, since deleting the queue will not chase those pointers. ⁴

Example: a button, a status reporter, and a consumer

To demonstrate FreeRTOS queues I have created a tag:

# Clone the tag
git clone \
  --depth 1 \
  --branch blog-freertos-queue --single-branch \
  https://gitlab.com/sorhanp/cortex-m7-qemu.git \
  cortex-m7-qemu-freertos-queue

# Go to folder
cd cortex-m7-qemu-freertos-queue

# Configure, build and run
cmake --preset arm-none-eabi-debug
cmake --build --preset arm-gcc-debug-build --target run-qemu

The example models a tiny appliance. A button gets pressed a few times, a status reporter emits health messages, and a main task consumes both streams and acts on them. All three talk through a single queue, and that one queue exercises everything from the sections above: copy by value, ownership handover, sending to the back and to the front, blocking and non-blocking sends, and back-pressure when the consumer cannot keep up.

Everything travels as one Event_t. The queue is created for sizeof(Event_t), so the whole struct is copied in and out by value, exactly like table 1.

typedef enum
{
  ButtonPress, // payload is an unsigned integer
  Status,      // payload is a heap buffer, the receiver takes ownership
  PowerOff,    // no payload
} EventType_t;

typedef struct
{
  const char* name;
  EventType_t type;
  Payload_t   payload;
} Event_t;

The Status event is the interesting one. Its payload is a pointer to a heap buffer, so even though the struct is copied by value, that copy carries an address. Here table 1 and table 2 are fused into a single queue: most of the event is copied wholesale, but one variant hands a pointer across, and with it the duty to free the memory.

The main task owns the lifecycle. It creates the queue, spawns the two producers, then drops into a consume loop.

constexpr UBaseType_t queueLength = 5;
queueHandle = xQueueCreate(queueLength, sizeof(Event_t));

Five slots, each one Event_t wide.

The button task is the impatient producer. Each press is sent to the back of the queue with a wait time of 0 (doNotWait), so the send never blocks. If the queue is full the send returns errQUEUE_FULL, and the task waits a moment and tries again rather than giving up.

while (SendToBackOfQueue(config, event, doNotWait) == false)
{
  vTaskDelay(pressInterval);
}

The off switch is different. It is urgent, so it jumps the queue: it goes to the front with portMAX_DELAY, blocking until there is room so it is never dropped.

SendToFrontOfQueue(config, off, portMAX_DELAY);

That is the priority case from the sending section made concrete. A normal press takes its place at the back, the off switch cuts to the front.

The status task is the patient producer. It allocates a small buffer, formats a message into it, and sends the event to the back with portMAX_DELAY. It would rather block than drop a status update, and once the event is on the queue the buffer belongs to whoever receives it.

uint8_t* message = static_cast<uint8_t*>(pvPortMalloc(messageMaxLen + 1));
// ... format the message into the buffer ...
SendToBackOfQueue(config, event, portMAX_DELAY);

From here on the status task must not touch that buffer again. Ownership has moved.

The consumer is deliberately slow. A 220 millisecond processing delay per event, against producers that emit every 40 to 90 milliseconds, so the queue fills up and the back-pressure becomes visible in the logs.

xQueueReceive(queueHandle, &event, portMAX_DELAY);
switch (event.type)
{
  case ButtonPress: // print the press number
    break;
  case Status: // print the message, then free the buffer we now own
    vPortFree(event.payload.as.buffer.bytes);
    break;
  case PowerOff: // end the loop
    poweredOff = true;
    break;
}

The receive blocks forever, so it never returns empty. The switch is where ownership is settled: a Status event is printed and then its buffer is freed, because the consumer now owns it. A button press carries nothing to free, and the off switch ends the loop.

Shutdown is where the vQueueDelete caveat earns its keep. The main task stops the producers, then drains whatever is still queued, freeing the buffer on every leftover Status event before it deletes the queue.

while (xQueueReceive(queueHandle, &leftover, doNotWait) == pdPASS)
{
  if (leftover.type == Status)
    vPortFree(leftover.payload.as.buffer.bytes);
}
vQueueDelete(queueHandle);

If it skipped the drain and deleted the queue straight away, every status buffer still sitting in it would leak, because deleting the queue frees the queue, not the things its items point to.

Run it and watch the queue status that every log line prints through uxQueueMessagesWaiting and uxQueueSpacesAvailable. Early on the producers run ahead of the slow consumer and the free count drops toward zero. Once the queue is full you will see the button task’s sends fail and retry, while the status task simply blocks until a slot frees. That is the whole point: the queue absorbs the speed mismatch, and each producer decided for itself how to react when it could not keep up.

Expected output, shortened:

Main task: started, queue status 0 used / 5 free
<Button task and Status task filling the queue>
Main task: waiting for an event, queue status 5 used / 0 free
<Button task and Status task filling the queue>
Event from Button task: button press #1
<Queue is full>
<Button task and Status task filling the queue>
Event from Button task: button press #2
<Continues normally until>
Event from Button task: OFF SWITCH pressed -- shutting down
Main task: stopping producers
Main task: draining queue
Main task: discarding unhandled status, freeing its buffer
Main task: discarding unhandled button press #4
Main task: discarding unhandled button press #5
Main task: discarding unhandled button press #6
Main task: exiting cleanly

In short: one queue, events copied by value, a pointer that changes hands, two producers with opposite patience, and a consumer slow enough to make all of it show.

Next

In Part 6 I will cover semaphores and mutexes. They look different on the surface, with a take/give API, but underneath they are queues doing exactly what you just saw.

-sorhanp

References