7. Message Queues

 

A Message Queue is another kernel object that offers communication between tasks. This object differs from semaphores in many ways:

• Unlike semaphore, a Message carries a pointer to virtually any kind of data (including data structures). It is therefore possible to expose data from one task to another.
• Unlike semaphore, the Message queue hold history of sent messages within a buffer that is filled by senders, and unfilled by the receiving task when activated
• A common use of Message Queues involves several tasks sending messages to a unique destination task

In the previous tutorial, we have seen that using a Mutex provides a solution for the shared use of a common resource (i.e. the debug console). In this tutorial, we will address the same issue differently using a Message Queue. To achieve this, we create a third task 'Task_Console' which is the only one having control over the debug console. Task_1 and Task_2 then send messages to Task_Console every time there is something to be printed.

In this approach of managing an unique resource, Task_Console is called the 'gate-keeping' task.

The whole process is illustrated below. It is worth noting that what the message carries is the address (&) of a pointer to the message data (pm). The message data can be of any defined type, including data structures. In other words, you can share anything... In the example below we consider that the message data is a string, defined as an array of char. For now, keep in mind that the actual message data always belongs to the sending task, and is accessed by the receiving task by means of a pointer.

Let us go thru the practical example. First declare 3 tasks:

// FreeRTOS tasks
void vTask1 		(void *pvParameters);
void vTask2 		(void *pvParameters);
void vTaskConsole 	(void *pvParameters);

Then, as usual, the next step is to declare (as global) the Message Queue object:

// Kernel Objects
xQueueHandle	xConsoleQueue;

Second, we define the message data type (here, an array of 60 bytes)

// Define the message_t type as an array of 60 char
typedef uint8_t message_t[60];

 

Next, the main() function creates the Message Queue. The first parameter is the depth of the queue buffer (i.e. the maximum number of pending messages the queue can hold):

...
        // Create Queue to hold console messages
	xConsoleQueue = xQueueCreate(10, sizeof(message_t *));

	// Give a nice name to the Queue in the trace recorder
	vTraceSetQueueName(xConsoleQueue, "Console Queue");
...

Then the main() function creates the tasks, and start the scheduler:

...
	// Create Tasks
	xTaskCreate(vTask1, 		"Task_1", 	256, NULL, 2, NULL);
	xTaskCreate(vTask2, 		"Task_2", 	256, NULL, 3, NULL);
	xTaskCreate(vTaskConsole, 	"Task_Console", 256, NULL, 1, NULL);

	// Start the Scheduler
	vTaskStartScheduler();
...

Note that Task_Console has been given the lowest priority level.

Task_1 prints a long message every 20ms:

/*
 *	Task_1
 */
void vTask1 (void *pvParameters)
{
	message_t 	message;
	message_t	*pm;

	while(1)
	{
		// Prepare message
		my_sprintf((char *)message, "With great power comes great responsibility\r\n");

		// Send message to the Console Queue
		pm = &message;
		xQueueSendToBack(xConsoleQueue, &pm, 0);

		// Wait for 20ms
		vTaskDelay(20);
	}
}

Task_2 prints a short message every 2ms:

/*
 *	Task_2
 */
void vTask2 (void *pvParameters)
{
	message_t 	message;
	message_t	*pm;

	while(1)
	{
		// Prepare message
		my_sprintf((char *)message, "#");

		// Send message to Console Queue
		pm = &message;
		xQueueSendToBack(xConsoleQueue, &pm, 0);

		// Wait for 2ms
		vTaskDelay(2);
	}
}

And finally, Task_Console processes incoming messages:

/*
 * Task_Console
 */
void vTaskConsole (void *pvParameters)
{
	message_t *message;

	while(1)
	{
		// Wait for something in the message Queue
		xQueueReceive(xConsoleQueue, &message, portMAX_DELAY);

		// Send message to console
		my_printf((const char *)message);
	}
}

There are different ways to manage the Message Queue buffer. Using the xQueueSendToBack() function make the queue working as a FIFO (First In, First Out). Messages will be printed in the order of arrival.

The result:

As expected, we have ten '#' for one long message. So it seems no messages from Task_2 have been lost on the way!

The detail of Task_2 sending its message for a short print:

Every 20ms, both Task_1 and Task_2 are activated altogether. Task_2 has higher priority level, therefore it goes first, immediately followed by Task_1. When Task_Console gets active, 2 messages are waiting in the queue. The first one (from Task_2) is quickly processed because it is short. The second one (from Task_1) takes longer. In that period of time, 2 new messages from Task_2 come in the queue. Both are quickly processed by Task_Console as soon as the long message has finished printing, and before it gets blocked due to an now empty queue. Crystal clear!

 

By double-clicking on any Console Queue event tag, you can get the history of queue usage:

 

Commit name "Message queue"   Push onto Gitlab

Is that fine?

 

Well, it's not... There's a potential issue hidden there. To reveal this issue, let us number the Task_2 '#' messages with a one digit (to keep it short) index. In addition, let us record and print that index value using a user event channel:

/*
 *	Task_2
 */
void vTask2 (void *pvParameters)
{
	message_t 	message;
	message_t	*pm;

	uint8_t		index;

	index = 0;

	while(1)
	{
		// Prepare message
		my_sprintf((char *)message, "%d# ", index);

		// Send message to Console Queue
		pm = &message;
		xQueueSendToBack(xConsoleQueue, &pm, 0);

		// Reports index value into trace UEC
		vTracePrintF(ue1, "%d", index);

		// Increment index
		index++;
		if (index>9) index = 0;

		// Wait for 2ms
		vTaskDelay(2);
	}
}

What do we get? This:

Message '1#' never prints. Instead, we have a duplicate '2#'. Did you expect this? Well, you should...

Remember now that message only carries a pointer to the data to be processed, which is local to the sending task. After a message is sent to the queue, if that data changes before the message is processed by the receiving task, then the old data is lost, and will never be processed.

Take a look at the trace, and you should understand why we've got no '1#', and two '2#':

The solution to this issue is left to you as an exercise. When a task sends a lot of messages to a queue, and do not expect messages to be process quickly, then multiple message variables (local to the task) are necessary to protect data from overriding.