Original from WeChat Official Account | Embedded Column
“Circular queue” and “message queue” are widely used in the embedded field, and experienced embedded software engineers are likely familiar with them.
However, beginners often have related questions. Today, I will share some content about “circular queues” and “message queues”.
A circular queue is a very useful data structure in practical programming. It is a FIFO data structure with its ends connected, using a linear array space. The data organization is simple, allowing quick checks for whether the queue is full or empty, and enabling fast data access.
Circular queues are typically used in communication fields such as UART, USB, CAN, and networks.
1. Implementation Principle of Circular Queue
There is no circular structure in memory; therefore, the circular queue is implemented using a linear array space. When data reaches the end, it wraps back to the 0 position for processing.
The logic of the circular queue is as follows: Connect the array elements q[0] and q[MAXN-1] to form a circular space for the queue.
To facilitate reading and writing, array indices are used to indicate the read and write positions, head/tail. Here, head points to the readable position, and tail points to the writable position.
The key to the circular queue is determining whether the queue is empty or full. The queue is full when tail catches up with head; it is empty when head catches up with tail. However, to know which is catching which, some auxiliary means are needed for judgment.
There are two methods to determine whether the circular queue is empty or full:
a. Add a flag variable tag
-
When head catches up with tail, the queue is empty, so set tag=0
-
When tail catches up with head, the queue is full, so set tag=1
b. Restrict tail from catching up with head, meaning there must be at least one element space between the tail node and the head node.
2. Implementation Principle with Additional Flag
a. The first circular queue has the following structure:
typedef struct ringq{ int head; /* Head, dequeue direction*/ int tail; /* Tail, enqueue direction*/ int tag ; int size ; /* Total size of the queue */ int space[RINGQ_MAX]; /* Queue space */}RINGQ;
Initialization state:
q->head = q->tail = q->tag = 0;
Queue empty:
( q->head == q->tail) && (q->tag == 0)
Queue full:
((q->head == q->tail) && (q->tag == 1))
Enqueue operation: if the queue is not full, write in:
q->tail = (q->tail + 1) % q->size ;
Dequeue operation: if the queue is not empty, read from head.
The next readable position is:
q->head = (q->head + 1) % q->size
#ifndef __RINGQ_H__#define __RINGQ_H__
#ifdef __cplusplusextern "C" {
#endif
#define QUEUE_MAX 20
typedef struct ringq{ int head; /* Head, dequeue direction*/ int tail; /* Tail, enqueue direction*/ int tag ; /* Flag for empty or full*/ int size ; /* Total size of the queue */ int space[QUEUE_MAX]; /* Queue space */}RINGQ;
/* First design method: when head == tail, tag = 0 means empty, = 1 means full.*/
extern int ringq_init(RINGQ * p_queue);
extern int ringq_free(RINGQ * p_queue);
/* Add data to the queue */extern int ringq_push(RINGQ * p_queue,int data);
/* Get data from the queue */extern int ringq_poll(RINGQ * p_queue,int *p_data);
#define ringq_is_empty(q) ( (q->head == q->tail) && (q->tag == 0))
#define ringq_is_full(q) ( (q->head == q->tail) && (q->tag == 1))
#define print_ringq(q) printf("ring head %d,tail %d,tag %d\n", q->head,q->tail,q->tag);
#ifdef __cplusplus}
#endif
#endif /* __RINGQ_H__ */
#include <stdio.h>#include "ringq.h"
int ringq_init(RINGQ * p_queue){ p_queue->size = QUEUE_MAX ;
p_queue->head = 0; p_queue->tail = 0;
p_queue->tag = 0;
return 0;}
int ringq_free(RINGQ * p_queue){ return 0;}
int ringq_push(RINGQ * p_queue,int data){ print_ringq(p_queue);
if(ringq_is_full(p_queue)) {
printf("ringq is full\n"); return -1; }
p_queue->space[p_queue->tail] = data;
p_queue->tail = (p_queue->tail + 1) % p_queue->size ;
/* At this point, the queue must be full */ if(p_queue->tail == p_queue->head) { p_queue->tag = 1; }
return p_queue->tag ; }
int ringq_poll(RINGQ * p_queue,int * p_data){ print_ringq(p_queue); if(ringq_is_empty(p_queue)) {
printf("ringq is empty\n"); return -1; }
*p_data = p_queue->space[p_queue->head];
p_queue->head = (p_queue->head + 1) % p_queue->size ;
/* At this point, the queue must be empty */ if(p_queue->tail == p_queue->head) { p_queue->tag = 0; } return p_queue->tag ;}
Seeing the source code, I believe everyone will understand the principle. In fact, there are methods that do not use a tag or other markers, but I won’t go into further detail here. Interested readers can research it themselves.
In RTOS, there is usually a message queue component, which is one of the most commonly used components.
1. Basic Concept of Message Queue
A message queue is a data structure commonly used for inter-task communication. It allows messages of variable lengths to be passed between tasks or between interrupts and tasks, enabling tasks to receive messages from other tasks or interrupts.
Through the message queue service, tasks or interrupt service routines can place one or more messages into the message queue. Similarly, one or more tasks can obtain messages from the message queue.
Using the message queue data structure enables asynchronous communication between tasks.
2. Characteristics of Message Queue
RTOS message queues have common characteristics:
-
Messages are queued in a first-in-first-out manner, supporting asynchronous read and write operations.
-
Both reading and writing queues support timeout mechanisms.
-
Messages can be queued in a last-in-first-out manner, sending messages to the head of the queue (LIFO).
-
Messages of varying lengths (not exceeding the maximum node value of the queue) can be allowed.
-
A task can receive and send messages from any message queue.
-
Multiple tasks can receive and send messages from the same message queue.
-
When the queue is no longer in use, it can be deleted using the delete queue function.
3. Principle of Message Queue
Here, I will explain using FreeRTOS as an example. The message queue control block in FreeRTOS consists of multiple elements. When the message queue is created, the system allocates corresponding memory space for the control block, which stores information such as the message storage location, head pointer pcHead, tail pointer pcTail, message size uxItemSize, and queue length uxLength.
For example, to create a message queue:
xQueue = xQueueCreate(QUEUE_LEN, QUEUE_SIZE);
Tasks or interrupt service routines can send messages to the message queue. When sending a message, if the queue is not full or allows overwrite, FreeRTOS will copy the message to the tail of the message queue. Otherwise, it will block according to the user-specified blocking timeout. During this time, if the queue does not allow enqueueing, the task will remain blocked waiting for the queue to allow enqueueing. When other tasks read data from their waiting queue (if the queue is not full), that task will automatically transition from the blocked state to the ready state. If the waiting time exceeds the specified blocking time, even if the queue still does not allow enqueueing, the task will automatically transition from the blocked state to the ready state. At this point, the task or interrupt program sending the message will receive an error code errQUEUE_FULL.
Sending urgent messages is almost the same process as sending regular messages, with the only difference being that urgent messages are sent to the head of the message queue instead of the tail, allowing the receiver to prioritize urgent messages for timely processing.
When a task attempts to read from a queue, it can specify a blocking timeout. During this time, if the queue is empty, the task will remain blocked waiting for the queue data to become valid. When other tasks or interrupt service routines write data into the waiting queue, that task will automatically transition from the blocked state to the ready state. If the waiting time exceeds the specified blocking time, even if the queue still does not have valid data, the task will automatically transition from the blocked state to the ready state.
When the message queue is no longer in use, it should be deleted to free system resources. Once the operation is completed, the message queue will be permanently deleted.
The operation process of the message queue is shown in the following diagram:
4. Blocking Mechanism of Message Queue
Blocking on Dequeue: A task can only read data when there is data in the message queue, and it can specify the blocking time to wait for data.
Blocking on Enqueue: A sender can only successfully send a message when the queue allows enqueueing; when there is no available message space in the queue, it indicates that the message queue is full, and the system will block the task according to the user-specified blocking timeout.
If multiple tasks are blocked on a message queue, these blocked tasks will be sorted by task priority, with higher-priority tasks gaining access to the queue first.
Similarities and Differences Between “Circular Queue” and “Message Queue”
Through the above analysis, you will find that there are many similarities between “circular queues” and “message queues”:
1. They are both data structures that contain information such as head, tail, and flags;
2. They both allocate a contiguous block of memory and can allocate multiple queues.
3. Their application scenarios are similar, particularly in situations where there is a large amount of data throughput, such as in communication fields.
Of course, they also have some differences:
1. “Circular queues” can be used independently or in conjunction with an operating system, while message queues depend on RTOS (some RTOS parameter information).
2. “Circular queues” occupy fewer resources and are more suitable for systems with limited resources.
3. “Message queues” are more flexible when combined with RTOS applications, such as for delays and interrupt data transmission.
Finally, both types of queues have broad applications, and I recommend taking the time to study both.
———— END ————
