原创 设计自己的嵌入式操作系统内核之七 ----- 消息队列实现

1、概述
同邮箱机制，使用消息队列也允许任务间互相发送消息。但与邮箱不同，消息队列发送消息时，如果队列为满，则发任务会阻塞；向邮箱发消息则不会。并且发送的消息，由原消息体将消息制到消息队列中，再由消息队列复制到收消息任务中。当然，这增加了内存的拷贝，带来了一定的性能损失，但可以解决前篇文章中提到的消息体管理问题。当任务发消息后，可以立即对原消息体写消息，再发送，不存在共享内存的问题。

eos提供的消息队列容许多条消息，并且不同的消息队列容许不同大小的消息，但同一消息队列内的消息则必须是同样大小的。

2、消息队列的实现
如下图所示，消息队列包含环形缓冲环、发消息阻塞队列、收消息阻塞队列。其中阻塞队列可配置为多级队列或单级队列。图中只显示了多级的阻塞队列。

消息队列的结构：
typedef struct _mqueue_t
{
? uint8 * start; /* 消息存储缓冲首址 */ ?
? uint8 * end; /* 消息存储缓冲末址 */
? uint8 * in; /* 消息存储写指针 */
? uint8 * out; /* 消息存储读指针 */
? uint8 entries; /* 消息队列中消息数 */
? uint8 size; /* 消息队列可容纳消息数 */
? uint16 msgsz; /* 消息队列中消息大小 */

#if OS_MQUEUE_SET_MLIST == 1
? mlist_t gwait_list; /* 读消息阻塞队列 */
#else
? slist_t gwait_list;
#endif

#if OS_MQUEUE_SET_WAIT_SEND == 1

#if OS_MQUEUE_SET_MLIST == 1
? mlist_t pwait_list; * 发消息队列阻塞队列 */
#else
? slist_t pwait_list;
#endif

#endif

}* mqueue_t;
结构体内包含.msgsz域的前半部分用于环形缓冲，后半部分描述了收发消息的阻塞队列。
.start 指示环形缓冲的首址。.end指示环形缓冲的末址。
.in 指示环形缓冲的写指针。.out指示环形缓冲的读指针
.entries 指示环形缓冲当前消息数。.size 指示环形缓冲可容纳总的消息数
.msgsz 指示消息队列中各消息占用存储空间大小(以字节计).

上图中的环形缓冲结构只是一逻辑示意图。实际的缓冲区只是一块连续的内存空间，类似于数组结构。但是在struct _mqueue_中各域的控制下，对该块内存的读写使得逻辑上类似于对环形缓冲的读写。

另外，还提供了mqueue_info_t 结构体用于查询消息队列信息。



typedef struct
{
? uint8 * msg_buf; /* 消息队列存储缓冲首址 */
? uint8 entries; /* 消息队列当前消息数 */
? uint8 size; /* 消息队列可容纳消息数 */
? uint16 msgsz; /* 收/发消息的消息大小 */
? uint8 avail; /* 消息队列空闲的储存量 */
}mqueue_info_t;

消息队列的接口函数：
主要包含消息队列的创建、销毁、收／发消息以及查询消息队列。
接口函数列表如下：
uint8 mqueue_create( mqueue_t * q, uint8 * buf, uint8 size, uint16 msgsz ); ?
uint8 mqueue_recv( mqueue_t q, void * msg, uint16 ticks ); /* 从消息队列收消息 */
? uint8 mqueue_destroy( mqueue_t q ); /* 销毁消息队列 */
uint8 mqueue_try_recv( mqueue_t q, void * msg ); /* 非阻塞队列取消息 */
uint8 mqueue_send( mqueue_t q, void * msg, uint16 ticks, uint8 opt );/* 向消息队列发消息 */
uint8 mqueue_send( mqueue_t q, void * msg, uint16 ticks ); ?
uint8 mqueue_try_send( mqueue_t q, void * msg, uint8 opt );/* 非阻塞向发消息 */
uint8 mqueue_try_send( mqueue_t q, void * msg );
uint8 mqueue_info( mqueue_t q, mqueue_info_t * info );/* 查询消息队列信息 */
uint8 mqueue_flush( mqueue_t q ); /* 清空消息队列?

各接口函数操作流程如下：
mqueue_create()操作流程：
参数检查
调用mpool_get()分配消息队列控制块struct _mqueue_t
初始化消息队列环形缓冲结构
调用xlist_init(),初始化取消息阻塞队列
调用xlist_init(),初始化发消息阻塞队列
返回OS_ERR_OK

mqueue_destroy()操作流程：
参数检查
检查取消息队列是否为空
否，
调用event_xlist_del()销毁取消息队列，将所有任务置为就绪态

检查发消息队列是否为空
否，
调用event_xlist_del()销毁发消息队列，将所有任务置为就绪态

检查是否需要调度
是，
调用scheduler()进行调度

调用mpool_free()回收消息队列控制
返回OS_ERR_OK

注：不像邮箱销毁函数mbox_destroy()那样，消息队列中即使有消息，也是容许删除的。因为消息是通过复制的方式存许于消息队列内。

mqueue_recv()操作流程：
参数检查
变量write = FALSE
消息队列中是否有消息
是，
变量write = TRUE
否，
调用event_xlist_wait()将当前任务挂起
调用scheduler()进行调度，切换至其它任务

返回时，检查当前任务控制块的.wait_state是否为OS_ERR_OK
是，
消息队列中已有消息，置write = TRUE

write的值是否为TRUE
是，
从消息队列复制数据消息到参数msg指定缓冲区中
调整读指针.out
发消息队列是否有任务阻塞
是，
调用event_xlist_rdy()唤醒发消息阻塞队列中某任务
调用scheduler()进行调度

返回.wait_state值。

注：在对消息进行复制时，要据消息的大小，选择了不同的复制方式。对于只有一字节的大小的消息体，直接使用字节赋值的方式；而对于较大的消息体，采用memcpy()进行复制。而且，同一消息体内的消息体大小总是相同的。

mqueue_try_recv()操作流程：
参数检查
消息队列是否有消息
否，
返回OS_ERR_MQUEUE_EMPTY
是，
从消息队列取消息并复制到参数.msg指定的缓冲区中。
发消息队列是否有任务阻塞
是，
调用event_xlist_rdy()唤醒发消息阻塞队列中某任务
调用scheduler()进行调度
返回OS_ERR_OK

mqueue_send()操作流程：
参数检查
取消息阻塞队列是有任务
是，
调用event_xlist_rdy()唤醒其中某任务
置变量sched为TRUE

消息队列消息是否已满
否，
置write = TRUE, wait_state = OS_ERR_OK
是，
检查是否在中断中或调度器上锁是调用
是，
立即返回，退出函数

调用event_xlist_wait()将自己置入发消息阻塞队列
调用scheduler()进行调度，切换至其它任务

当任务再次被唤醒执行时，wait_state = OSTaskCur->wait_state
检查wait_state 是否为OS_ERR_OK，即可以往队列写消息
是，
置write = TRUE

检查write 是否为TRUE
如果函数参数为OS_MQUEUE_SEND_LIFO
是，
按LIFO方式写入消息
否，
按FIFO方式写入消息

检查sched 是否为TRUE
是，
调用scheduler()进行调度

注：按LIFO方式发送的消息，会相对于其它消息首先被收消息任务查收。



mqueue_try_send()操作流程：
参数检查
取消息阻塞队列是有任务
是，
调用event_xlist_rdy()唤醒其中某任务
置变量sched为TRUE
消息队列消息是否已满
否，
如果函数参数为OS_MQUEUE_SEND_LIFO
是，
按LIFO方式写入消息
否，
按FIFO方式写入消息

检查sched 是否为TRUE
是，
调用scheduler()进行调度

mqueue_info()操作流程：
参数检查
从struct _mqueue_t中取信息复制到mqueue_info_t 结构中。

mqueue_flush()操作流程：
参数检查
重置缓冲环结构。

从某种程度上来说，这里的消息队列结构有些类似于管道结构。但是同时允许多个读者和写者，并且每次读写的消息大小都是相同的。

3、简单的演示
提供了一个简单的测试实例 app_mqueuec 演示了消息队列的操作。

4、问题与思考
是否可以减少消息的拷贝次数以提高效率?可以考虑在任务发消息时,如果阻塞队列中有任务,发消息的任务可直接将消息复制到阻塞队列中取消息任务的接收消息的缓冲区中.该缓冲区的指针,可以在接收任务的消息阻塞时,保存在任务的task_struct的.event域中.

示例文件app_mqueue.c演示了使用消息队列进行串口收发的程序。串口操作的支持文件为uart.c，其中各函数调用了消息队列的接口，这样任务和中断处理程序通过消息队列进行通信。创建了两个消息队列，串口接收数据队列和发数据队列，各队列中可容纳的消息数量只设置为１，这样每次只能从消息队列收发一个字节的数据。即使在这种情况下，工作还比较正常。对于任务和中断服务程序而言，两个队列类似于管道结构。对发消息队列，任务向管道依次写入消息，每次写入一字节，ISR每次从管理取一字节消息，由于消息队列的结构，使得ISR取得的数据字节顺序能与任务写入的一致。

在运行app_mqueue.c中应用时，经常性的出现系统死机问题。但是原因只有一个，即堆栈溢出。而导致堆栈溢出的原因除了堆栈深度设置太小外，另一个隐藏较深的原因是在任务上下文保存时，对中断的处理不当！对于后者，因为问题较复杂，在该篇文档中不作说明。在以前移植的uc/os代码中，对任务上下文的保存采用了类似的方法，且也写了类似的串口收发程序，但并未出现问题！原因尚未明确！

五、相关源码
core.c / core.h ---------- 内核支持函数文件
mqueue.c/ipc.h ----------- 消息队列实现文件