前言
前两篇文章记录了 IPC 基础层、信号量和互斥量。这篇文章继续记录 Yuan RTOS 中的消息队列。
消息队列用于在任务之间传递数据。相比信号量只能表达“有没有资源”或“资源数量”，队列可以真正携带一段消息内容，因此非常适合生产者消费者模型。
完整代码请访问我的Github仓库，网址为：yuan-rtos。
值得注意是本文的示例代码在你阅读时可能已经进行过修改了，具体实现以仓库中代码为主。
队列结构
Yuan RTOS 的队列使用用户提供的静态缓冲区，并在内部维护一个环形队列：
typedef struct yr_queue_t {
yr_ipc_base_t send_ipc;
yr_ipc_base_t receive_ipc;
yr_uint8_t *buffer;
yr_uint32_t capacity;
yr_uint32_t item_size;
yr_uint32_t item_count;
yr_uint32_t head;
yr_uint32_t tail;
} yr_queue_t;
这里有两个 IPC 基础对象：
send_ipc：队列满时，发送任务阻塞在这里receive_ipc：队列空时，接收任务阻塞在这里
为什么队列需要两个阻塞队列？
因为发送者和接收者等待的是两种相反的条件。发送者等待“队列有空位”，接收者等待“队列有数据”。如果混在一个阻塞链表里，唤醒逻辑会很别扭，也容易唤醒错误类型的任务。
初始化队列
队列初始化函数如下：
yr_err_t yr_queue_init( yr_queue_t *queue,
yr_uint32_t item_size,
yr_uint8_t *buffer,
yr_uint32_t buffer_size,
yr_uint32_t flag)
参数中比较关键的是 item_size、buffer 和 buffer_size。
Yuan RTOS 的队列并不会自己动态分配内存，而是由用户传入一块静态缓冲区。buffer_size 必须是 item_size 的整数倍：
capacity = buffer_size / item_size;
这样队列最多可以保存 capacity 个元素。
初始化时还会把读写游标和元素数量清零：
static void __queue_reset( yr_queue_t *queue)
{
queue->head = 0;
queue->tail = 0;
queue->item_count = 0;
}
环形缓冲区
队列内部使用 head 和 tail 两个游标实现环形缓冲区：
head：下一次读取的位置tail：下一次写入的位置item_count：当前队列中已有的元素数量
写入时推进 tail，读取时推进 head。当游标到达 capacity 后，再回到 0。
这种结构的好处是实现简单，不需要移动已有数据。只要更新游标即可完成入队和出队。
写入元素
写入一个元素的内部函数如下：
static void __queue_write_item( yr_queue_t *queue,
const void *item )
{
yr_uint8_t *dst = queue->buffer +
queue->tail * queue->item_size;
memcpy( dst, item, queue->item_size );
queue->tail++;
if( queue->tail >= queue->capacity )
queue->tail = 0;
queue->item_count++;
}
这里用 memcpy() 拷贝固定大小的元素，因此队列可以保存任意类型的数据，只要发送方和接收方对 item_size 的理解一致即可。
读取元素
读取一个元素的内部函数如下：
static void __queue_read_item( yr_queue_t *queue,
void *item )
{
yr_uint8_t *src = queue->buffer +
queue->head * queue->item_size;
memcpy( item, src, queue->item_size );
queue->head++;
if( queue->head >= queue->capacity )
queue->head = 0;
queue->item_count--;
}
读取完成后，head 向后移动，并减少 item_count。
发送消息
普通任务上下文下发送消息使用：
yr_err_t yr_queue_send( yr_queue_t *queue,
void *item,
yr_uint32_t wait_ticks)
如果队列没满，就直接写入数据：
if( queue->item_count < queue->capacity ) {
__queue_write_item( queue, item );
if( !yr_list_isempty( &queue->receive_ipc.blocked_list ) ) {
__queue_resume_one( &queue->receive_ipc,
current_task,
&need_switch );
}
return YR_OK;
}
如果此时有任务阻塞在接收队列上，说明之前有接收者在等数据。发送成功后，就可以唤醒一个接收任务。
如果队列已满，并且调用者不愿意等待，则直接返回错误。
如果调用者允许等待，则当前任务会阻塞在 send_ipc 上：
yr_task_set_msg( current_task,
(void*)&queue->send_ipc,
NULL,
YR_TASK_MR_IPC,
YR_TASK_MN_NONE );
yr_ipc_block_task( &queue->send_ipc, current_task );
如果设置了有限等待时间，还会启动任务自己的定时器。
接收消息
普通任务上下文下接收消息使用：
yr_err_t yr_queue_receive( yr_queue_t *queue,
void *item,
yr_uint32_t wait_ticks)
如果队列非空，就直接读取数据：
if( queue->item_count > 0 ) {
__queue_read_item( queue, item );
if( !yr_list_isempty( &queue->send_ipc.blocked_list ) ) {
__queue_resume_one( &queue->send_ipc,
current_task,
&need_switch );
}
return YR_OK;
}
如果此时有任务阻塞在发送队列上，说明之前有发送者在等空位。接收成功后，队列空位增加，就可以唤醒一个发送任务。
如果队列为空，并且调用者不愿意等待，则直接返回错误。
如果调用者允许等待，则当前任务会阻塞在 receive_ipc 上，并可根据 wait_ticks 启动超时定时器。
队列不是直接交付资源
这里需要特别说明一下，队列的唤醒语义和前面文章中的信号量、互斥量并不完全一样。
信号量和互斥量更接近“直接交付”。例如信号量释放时，如果有任务正在等待，就不会简单地让 current_count++，而是直接唤醒等待队列中策略上最靠前的任务，并把等待结果设置为 YR_TASK_MN_WAIT_OK。互斥量释放时也类似，如果有任务等待锁，就会把 mutex->owner 直接设置为被唤醒的任务。
这样做的原因是，信号量和互斥量的资源语义比较明确：释放的就是一个计数资源或一把锁。如果已经有任务在等，那么把资源直接交给等待者，可以避免资源释放后又被其它任务抢走，语义也更接近“谁在等，就优先满足谁”。
但队列这里不是这样。队列发送或接收成功后，只是改变了队列状态：
- 发送成功：队列中多了一个元素，接收条件变为可能满足
- 接收成功：队列中少了一个元素，发送条件变为可能满足
随后内核会从对应的阻塞队列中选择策略上最靠前的任务，将它唤醒，让它重新尝试发送或接收。被唤醒的任务并不等于已经拿到了某个队列元素，也不等于已经获得了某个空位。
这也是为什么 yr_queue_send() 和 yr_queue_receive() 都写成了循环结构。任务被唤醒后，会回到循环开头重新检查队列状态，而不是直接认为操作成功。
为什么队列不做直接交付？主要有两个原因。
第一，队列保存的是一段缓冲区状态，而不是一个单独的“资源令牌”。发送者写入的是队列公共缓冲区，接收者读取的也是队列公共缓冲区。如果要做严格的直接交付，就需要把“唤醒哪个任务”和“把哪一条消息交给哪个任务”绑定起来，队列内部语义会复杂很多。
第二，被唤醒任务真正运行之前，系统可能发生任务切换、中断、其它高优先级任务抢占等情况。比如一个接收任务被唤醒后，还没来得及运行，另一个更高优先级接收任务先执行并取走了队列中的数据，那么原来被唤醒的任务再次检查时仍然可能发现队列为空。
因此 Yuan RTOS 当前选择的是更简单、也更通用的方式：队列操作只负责改变队列状态，并唤醒一个策略上最应该被唤醒的任务；被唤醒任务是否真的能完成操作，要靠恢复运行后再次检查队列状态来决定。
这种设计的好处是实现简单，和超时等待、优先级/FIFO 阻塞队列都比较容易配合。缺点是被唤醒任务不一定一次就能成功完成操作，可能会出现“被唤醒后重新竞争失败，然后再次阻塞”的情况。不过这个代价是可以接受的，因为队列本身就是一个共享缓冲区，重新检查状态也能让语义更稳妥。
等待时间的处理
队列的发送和接收都使用了一个循环结构。原因是任务被唤醒后，仍然需要重新检查队列状态。
例如发送任务因为队列满而阻塞，之后被接收任务唤醒。它恢复运行后，不是直接认为发送成功，而是回到循环开头，再次检查队列是否有空位。
同时，如果 wait_ticks 是有限时间，函数还会记录已经等待的时间：
now_ticks = yr_get_current_ticks();
elapsed_ticks = now_ticks - start_ticks;
if( (yr_int32_t)(elapsed_ticks - wait_ticks) >= 0 )
return YR_ERR;
wait_ticks -= elapsed_ticks;
start_ticks = now_ticks;
这样可以避免任务被多次唤醒、重新阻塞时，等待时间被错误地重置。
中断接口
队列提供了两个中断接口：
yr_queue_send_from_isr(...)
yr_queue_receive_from_isr(...)
中断上下文下仍然遵循一个原则：不能阻塞。
因此：
- 中断中发送时，如果队列满了，直接返回错误
- 中断中接收时，如果队列空了，直接返回错误
- 如果操作成功，并唤醒了更高优先级任务，则通过
need_switch通知外部
这样可以让 ISR 中的逻辑保持较轻，不在中断里做复杂的阻塞等待。
删除与重置
队列删除时，需要唤醒所有阻塞在发送队列和接收队列上的任务：
if( !yr_list_isempty( &queue->send_ipc.blocked_list) ) {
yr_ipc_resume_all( &queue->send_ipc);
}
if( !yr_list_isempty( &queue->receive_ipc.blocked_list) ) {
yr_ipc_resume_all( &queue->receive_ipc);
}
这些任务被唤醒后，会得到 YR_TASK_MN_WAIT_IPC_DELETED 的通知。
队列也支持 reset，但 reset 只允许在没有任务等待队列状态变化时调用：
if( !yr_list_isempty( &queue->send_ipc.blocked_list ) ||
!yr_list_isempty( &queue->receive_ipc.blocked_list ) ) {
return YR_ERR;
}
这是为了避免唤醒语义混乱。比如有任务正在等数据，此时直接 reset 队列，任务应该被唤醒还是继续等？为了不引入这种模糊语义，当前实现直接禁止这种情况。
总结
Yuan RTOS 的队列实现可以概括为：
- 使用用户提供的静态缓冲区保存数据
- 内部通过
head、tail和item_count实现环形队列 - 队列满时发送任务阻塞在
send_ipc - 队列空时接收任务阻塞在
receive_ipc - 发送成功后改变队列状态，并可能唤醒接收任务重新竞争
- 接收成功后改变队列状态，并可能唤醒发送任务重新竞争
- 支持普通任务上下文和中断上下文两类接口
- 有限等待时间通过任务定时器实现超时
队列是任务间传递数据最常用的 IPC 机制之一。相比信号量和互斥量，它不仅能同步任务，还能携带实际的数据内容，因此在实际项目中使用频率通常会很高。