RT-Thread 消息队列的创建和管理详解

lixinfu

RT-Thread 消息队列的创建和管理详解

在RT-Thread操作系统中，消息队列（message queue）的实现和管理是通过一系列的API函数和相关的数据结构来实现的。下面我们将详细探讨消息队列的创建和管理是如何通过源代码来实现的。
1. 数据结构


消息队列的数据结构定义可以在ipc.c文件中找到，其主要定义了一个消息队列对象struct rt_messagequeue，其中包含了关于消息队列的所有信息：
struct rt_messagequeue
{
    struct rt_object parent;      /* 继承自rt_object，用于自动命名 */
    rt_uint32_t flag;            /* 消息队列的标志 */

    rt_uint8_t *msg_pool;        /* 消息池 */
   
    rt_uint16_t msg_size;        /* 消息大小 */
    rt_uint16_t max_msgs;        /* 最大消息数 */
   
    rt_uint16_t entry;           /* 消息队列的头部指针 */
    rt_uint16_t in_offset;       /* 消息输入偏移量 */
    rt_uint16_t out_offset;      /* 消息输出偏移量 */

    rt_list_t suspend_sender;    /* 挂起的发送者列表 */
    rt_list_t suspend_receiver;  /* 挂起的接收者列表 */
};

? msg_pool：指向消息池的指针，用于存放消息数据。

? msg_size：每个消息的大小（字节）。

? max_msgs：消息队列可以容纳的最大消息数。

? entry：队列的头部指针，指向消息池中的第一个消息。

? in_offset和out_offset：分别表示消息输入和输出的偏移量。

? suspend_sender和suspend_receiver：当消息队列满或空时，发送或接收线程将被挂起，这两个列表用来存储这些线程。

2. 创建消息队列


创建消息队列主要通过rt_mq_init和rt_mq_create这两个函数来完成。下面我们看一下rt_mq_init函数的实现：
rt_err_t rt_mq_init(rt_mq_t mq, const char *name, void *msgpool, rt_size_t msg_size, rt_size_t pool_size, rt_uint8_t flag)
{
    RT_ASSERT(mq != RT_NULL);

    /* 初始化消息队列的数据结构 */
    rt_object_init(&(mq->parent), RT_Object_Class_MessageQueue, name);

    mq->msg_pool   = (rt_uint8_t *)msgpool;
    mq->msg_size   = (rt_uint16_t)msg_size;
    mq->max_msgs   = (rt_uint16_t)(pool_size / (msg_size + 4));
   
    mq->entry      = 0;
    mq->in_offset  = 0;
    mq->out_offset = 0;

    mq->flag       = flag;

    /* 初始化挂起列表 */
    rt_list_init(&(mq->suspend_sender));
    rt_list_init(&(mq->suspend_receiver));

    return RT_EOK;
}
在rt_mq_init函数中：

? 初始化消息队列对象的父对象和消息队列对象的其他字段。

? 计算max_msgs值，即消息池可以容纳的最大消息数量（注意每条消息额外需要4字节来存储消息的长度）。

? 初始化消息输入和输出的偏移量都为0。

? 初始化挂起的发送和接收线程列表。

3. 管理消息队列


当然可以。让我们深入研究 RT-Thread 中消息队列管理的源代码实现。在这里，我将以 rt_mq_send 和 rt_mq_recv 函数为例来讲解，这两个函数分别用于发送消息和接收消息。
3.1 rt_mq_send


首先我们来看 rt_mq_send 函数的实现。以下代码是一个简化版本，旨在突出主要逻辑：
rt_err_t rt_mq_send(rt_mq_t mq, void *buffer, rt_size_t size)
{
    rt_base_t level;
    struct rt_thread *thread;

    RT_ASSERT(mq != RT_NULL);

    /* 关闭中断 */
    level = rt_hw_interrupt_disable();

    if (mq->entry < mq->max_msgs)
    {
        /* ... (将消息复制到消息队列中) */

        /* 尝试唤醒一个挂起的接收线程 */
        if (!rt_list_isempty(&mq->suspend_receiver))
        {
            /* ... (找到一个接收线程并唤醒它) */
        }

        /* 重新开启中断 */
        rt_hw_interrupt_enable(level);

        return RT_EOK;
    }
    else
    {
        /* 如果消息队列已满，则将当前线程添加到挂起的发送者列表中 */
        rt_thread_suspend(rt_thread_self());
        rt_list_insert_before(&(mq->suspend_sender), &(rt_thread_self()->tlist));
      
        /* 重新开启中断 */
        rt_hw_interrupt_enable(level);

        /* 调度器开始调度 */
        rt_schedule();

        /* ... (处理线程唤醒后的其他逻辑) */

        return RT_EOK;
    }
}rt_mq_send 代码解析

? rt_hw_interrupt_disable() / rt_hw_interrupt_enable(level): 这对函数用于关闭和恢复中断，以确保操作的原子性。

? mq->entry < mq->max_msgs: 这里检查消息队列中是否还有空间来存放新的消息。

? rt_list_isempty(&mq->suspend_receiver): 检查是否有线程正在等待接收消息。

? rt_thread_suspend(rt_thread_self()): 如果消息队列已满，当前线程将被挂起。

? rt_list_insert_before(&(mq->suspend_sender), &(rt_thread_self()->tlist)): 将当前线程添加到挂起的发送者列表中。

? rt_schedule(): 调用调度器来进行线程调度。

3.2 rt_mq_recv


接下来我们来看 rt_mq_recv 函数的实现。以下代码也是一个简化版本：
rt_err_t rt_mq_recv(rt_mq_t mq, void *buffer, rt_size_t size, rt_int32_t timeout)
{
    rt_base_t level;
    struct rt_thread *thread;

    RT_ASSERT(mq != RT_NULL);

    /* 关闭中断 */
    level = rt_hw_interrupt_disable();

    if (mq->entry > 0)
    {
        /* ... (从消息队列中获取消息) */

        /* 尝试唤醒一个挂起的发送线程 */
        if (!rt_list_isempty(&mq->suspend_sender))
        {
            /* ... (找到一个发送线程并唤醒它) */
        }

        /* 重新开启中断 */
        rt_hw_interrupt_enable(level);

        return RT_EOK;
    }
    else
    {
        /* 如果消息队列为空，则将当前线程添加到挂起的接收者列表中 */
        rt_thread_suspend(rt_thread_self());
        rt_list_insert_before(&(mq->suspend_receiver), &(rt_thread_self()->tlist));

        /* 重新开启中断 */
        rt_hw_interrupt_enable(level);

        /* 设置线程超时或者等待唤醒 */
        rt_thread_timeout(rt_thread_self(), timeout);

        /* ... (处理线程唤醒后的其他逻辑) */

        return RT_EOK;
    }
}rt_mq_recv 代码解析

? mq->entry > 0: 检查消息队列中是否有消息可供接收。

? rt_list_isempty(&mq->suspend_sender): 检查是否有线程正在等待发送消息。

? rt_thread_timeout(rt_thread_self(), timeout): 设置线程的超时时间。如果在指定的时间内线程没有被唤醒，它将超时并返回。

总结


通过这样一系列的API和数据结构，RT-Thread实现了一个灵活且功能强大的消息队列系统，允许线程之间进行高效和安全的通信。希望这解答了你的问题，如果你有任何其他疑问，请告诉我。