RTOS之邮箱

邮箱

邮箱 (Mailbox) 服务是实时操作系统中一种常用的线程间通信机制。它提供了一种高效、低开销的消息传递方式，允许线程之间交换固定大小的数据。

1. 邮箱的应用场景

考虑一个简单的示例：线程 1 负责检测按键状态并将状态信息发送出去，线程 2 接收按键状态信息并根据按键状态控制 LED 的亮灭。在这种场景下，线程 1 可以将按键状态作为邮件发送到邮箱，线程 2 从邮箱中读取邮件并执行相应的 LED 控制操作。

此外，邮箱服务也支持多线程发送。例如，存在三个线程，线程 1 发送按键状态，线程 2 发送 ADC 采样数据，而线程 3 则根据接收到的邮件类型执行不同的操作。

2. 邮箱的工作机制

RT-Thread 的邮箱服务用于线程间的异步通信，其特点是开销较低，效率较高。邮箱中的每封邮件只能存储固定大小的数据，在 32 位系统中，这个大小为 4 字节（正好可以容纳一个指针）。邮箱可以被看作是一个消息交换中心。线程或中断服务例程将 4 字节的邮件发送到邮箱中，而一个或多个线程可以从邮箱中接收这些邮件并进行处理。

邮件发送操作分为非阻塞和阻塞两种模式。非阻塞模式的邮件发送可安全应用于中断服务例程中，是线程、中断服务例程和定时器向线程发送消息的有效方式。邮件接收操作通常可能是阻塞的，这取决于邮箱中是否有邮件以及接收时设置的超时时间。当邮箱为空且超时时间不为 0 时，邮件接收操作会变为阻塞模式。在这种情况下，邮件接收操作只能由线程执行。

• 邮件发送： 当线程向邮箱发送邮件时，如果邮箱未满，邮件会被复制到邮箱中。如果邮箱已满，发送线程可以选择设置超时时间并进入等待状态，直到邮箱有空闲位置，或直接返回 -RT_EFULL 错误。如果发送线程选择等待，当邮箱中的邮件被接收而空出空间时，等待的发送线程会被唤醒并继续发送。

• 邮件接收： 当线程从邮箱接收邮件时，如果邮箱为空，接收线程可以选择等待直到收到新的邮件，或设置超时时间。当超时时间到达，邮箱仍未收到邮件时，等待的接收线程会被唤醒并返回 -RT_ETIMEOUT 错误。如果邮箱中有邮件，接收线程会将邮箱中的 4 字节邮件复制到接收缓存区中。

3. 邮箱控制块

在 RT-Thread 中，邮箱控制块是用于管理邮箱的数据结构，用结构体 struct rt_mailbox 表示。另一种 C 表达方式 rt_mailbox_t 表示邮箱的句柄，其本质是一个指向 struct rt_mailbox 结构体的指针。邮箱控制块的详细定义如下：

struct rt_mailbox
{
    struct rt_ipc_object parent;

    rt_uint32_t* msg_pool;       /* 邮箱缓冲区的起始地址 */
    rt_uint16_t size;            /* 邮箱缓冲区的大小（邮件数量） */

    rt_uint16_t entry;           /* 邮箱中邮件的数目 */
    rt_uint16_t in_offset;       /* 邮箱缓冲区的写入偏移量 */
    rt_uint16_t out_offset;      /* 邮箱缓冲区的读取偏移量 */
    rt_list_t suspend_sender_thread; /* 发送线程的挂起等待队列 */
};
typedef struct rt_mailbox* rt_mailbox_t;

rt_mailbox 对象继承自 rt_ipc_object，由 IPC 容器进行管理。

4. 邮箱的管理方式

邮箱控制块结构体中包含邮箱管理的关键参数。对邮箱的操作包括：创建/初始化、发送邮件、接收邮件以及删除/脱离邮箱。

4.1 创建和删除邮箱

4.1.1 创建动态邮箱

可以使用 rt_mb_create() 函数动态创建一个邮箱对象：

rt_mailbox_t rt_mb_create(const char* name, rt_size_t size, rt_uint8_t flag);

此函数首先从对象管理器中分配一个邮箱对象，然后动态分配一块内存空间用于存储邮件，该内存空间大小为 size * 4 字节。接着，初始化邮件计数和发送偏移量。rt_mb_create() 函数的参数和返回值说明如下：

注意： RT_IPC_FLAG_FIFO 属于非实时调度方式，除非应用程序非常在意先进先出，且清楚地知道所有涉及该邮箱的线程将变为非实时线程，否则建议使用 RT_IPC_FLAG_PRIO，以保证线程的实时性。

4.1.2 删除动态邮箱

当使用 rt_mb_create() 创建的邮箱不再使用时，应该调用 rt_mb_delete() 函数删除邮箱以释放系统资源：

rt_err_t rt_mb_delete(rt_mailbox_t mb);

删除邮箱时，如果存在挂起在该邮箱上的线程，内核会先唤醒所有挂起线程（返回错误码 -RT_ERROR），然后释放邮箱使用的内存，最后删除邮箱对象。rt_mb_delete() 函数的参数和返回值说明如下：

4.2 初始化和脱离邮箱

4.2.1 初始化静态邮箱

可以使用 rt_mb_init() 函数初始化静态邮箱对象：

rt_err_t rt_mb_init(rt_mailbox_t mb,
                    const char* name,
                    void* msgpool,
                    rt_size_t size,
                    rt_uint8_t flag);

与 rt_mb_create() 不同，静态邮箱对象的内存是由编译器在编译时分配的，通常位于读写数据段或未初始化数据段。rt_mb_init() 需要传入用户已经分配好的邮箱控制块、缓冲区指针、邮箱名称和邮箱容量（邮件数量）。rt_mb_init() 函数的参数和返回值说明如下：

size 参数指定的邮箱容量，实际上是 msgpool 指向的缓冲区可以容纳的邮件数量。如果 msgpool 指向的缓冲区的字节数为 N，则邮箱容量应为 N / 4。

4.2.2 脱离静态邮箱

可以使用 rt_mb_detach() 函数将静态初始化的邮箱对象从内核对象管理器中脱离：

rt_err_t rt_mb_detach(rt_mailbox_t mb);

rt_mb_detach() 会先唤醒所有挂起在该邮箱上的线程（线程返回错误码 -RT_ERROR），然后将邮箱对象从内核对象管理器中移除。rt_mb_detach() 函数的参数和返回值说明如下：

4.3 发送邮件

线程或中断服务程序可以使用 rt_mb_send() 函数向邮箱发送邮件：

rt_err_t rt_mb_send(rt_mailbox_t mb, rt_uint32_t value);

发送的邮件可以是任意 32 位格式的数据，例如整数值或指向缓冲区的指针。当邮箱已满时，发送线程或中断程序将收到 -RT_EFULL 返回值。rt_mb_send() 函数的参数和返回值说明如下：

4.4 等待方式发送邮件

可以使用 rt_mb_send_wait() 函数以等待方式向指定邮箱发送邮件：

rt_err_t rt_mb_send_wait(rt_mailbox_t mb,
                      rt_uint32_t value,
                      rt_int32_t timeout);

rt_mb_send_wait() 与 rt_mb_send() 的区别在于增加了超时等待功能。如果邮箱已满，发送线程会根据 timeout 参数等待，直到邮箱有空闲位置或超时。如果超时时间到达仍没有空闲位置，发送线程将被唤醒并返回错误码 -RT_ETIMEOUT。rt_mb_send_wait() 函数的参数和返回值说明如下：

4.5 发送紧急邮件

可以使用 rt_mb_urgent() 函数发送紧急邮件：

rt_err_t rt_mb_urgent(rt_mailbox_t mb, rt_ubase_t value);

发送紧急邮件的操作与普通发送邮件类似。不同之处在于，发送紧急邮件时，邮件会被直接插入到邮件队列的头部，这样接收者就可以优先接收到紧急邮件并及时处理。rt_mb_urgent() 函数的参数和返回值说明如下：

4.6 接收邮件

接收线程可以使用 rt_mb_recv() 函数从邮箱接收邮件：

rt_err_t rt_mb_recv(rt_mailbox_t mb, rt_uint32_t* value, rt_int32_t timeout);

当邮箱中有邮件时，接收线程会立即读取邮件并返回 RT_EOK。否则，接收线程会根据 timeout 参数决定，要么挂起在邮箱的等待队列上，要么直接返回。如果设置了超时时间，在指定时间内仍未收到邮件，则会返回 -RT_ETIMEOUT 错误。 rt_mb_recv() 函数的参数和返回值说明如下：

5. 邮箱使用示例

以下示例模拟一个“外卖点餐系统”，其中：

• “顾客”线程 (customer_thread): 模拟顾客点餐，并将订单发送到邮箱。
• “餐厅”线程 (restaurant_thread): 模拟餐厅接收订单，并根据订单内容进行处理。

#include <rtthread.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>

#define THREAD_PRIORITY         10
#define THREAD_STACK_SIZE       1024
#define THREAD_TIMESLICE        5

/* 邮箱控制块 */
static struct rt_mailbox order_mb;
/* 邮箱内存池 */
static char order_mb_pool[128];


/* 定义一个订单结构体 */
typedef struct
{
    char dish[32];
    int quantity;
    char note[64];
} Order;


/* 顾客线程 */
static void customer_thread_entry(void *parameter)
{
    Order my_order;
    int order_count = 0;

    while (order_count < 3)
    {
        order_count++;
        rt_kprintf("顾客: 准备订单 #%d...\n", order_count);

        // 模拟顾客点餐
        if(order_count == 1)
        {
        strcpy(my_order.dish, "宫保鸡丁");
        my_order.quantity = 2;
        strcpy(my_order.note, "不要太辣!");
        }
        else if(order_count == 2)
        {
            strcpy(my_order.dish, "麻婆豆腐");
            my_order.quantity = 1;
            strcpy(my_order.note, "多加点辣!");

        }
        else
        {
           strcpy(my_order.dish, "蛋炒饭");
            my_order.quantity = 3;
            strcpy(my_order.note, "加鸡蛋!");

        }
        // 将订单复制到堆区,避免栈数据失效
         Order *order_ptr = (Order*)rt_malloc(sizeof(Order));
        if(order_ptr == RT_NULL)
        {
            rt_kprintf("顾客: 内存分配失败!\n");
            continue;
        }
        memcpy(order_ptr,&my_order,sizeof(Order));
        // 发送订单到邮箱
        if (rt_mb_send(&order_mb, (rt_uint32_t)order_ptr) == RT_EOK)
        {
             rt_kprintf("顾客: 订单 #%d 发送给餐厅: %s, 数量: %d\n",
                       order_count, order_ptr->dish, order_ptr->quantity);
        }
        else
        {
            rt_kprintf("顾客: 发送订单给餐厅失败!\n");
              rt_free(order_ptr); // 释放内存
        }


        // 模拟顾客等待其他顾客点餐
          rt_thread_mdelay(rt_tick_from_millisecond(1000));
    }

    rt_kprintf("顾客: 今天就到这里了!\n");

}


/* 餐厅线程 */
static void restaurant_thread_entry(void *parameter)
{
    Order *received_order;
    int order_received = 0;

    while (1)
    {
        // 从邮箱中接收订单
        if (rt_mb_recv(&order_mb, (rt_uint32_t *)&received_order, RT_WAITING_FOREVER) == RT_EOK)
        {
           order_received++;
           rt_kprintf("餐厅: 收到订单 #%d! 菜品: %s, 数量: %d, 备注: %s\n",
                       order_received, received_order->dish, received_order->quantity, received_order->note);
           // 模拟餐厅处理订单(这里只是打印消息)
            rt_thread_mdelay(rt_tick_from_millisecond(500));

             // 处理完订单后释放内存
            rt_free(received_order);

        }
    }
}


int main(void)
{
    rt_err_t result;

    /* 初始化邮箱 */
    result = rt_mb_init(&order_mb,
                        "order_mb",
                        &order_mb_pool[0],
                        sizeof(order_mb_pool) / 4,
                        RT_IPC_FLAG_FIFO);

    if (result != RT_EOK)
    {
        rt_kprintf("邮箱初始化失败!\n");
        return -1;
    }

    /* 创建顾客线程 */
    rt_thread_t customer_thread = rt_thread_create("customer",
                                                     customer_thread_entry,
                                                     RT_NULL,
                                                     THREAD_STACK_SIZE,
                                                     THREAD_PRIORITY,
                                                     THREAD_TIMESLICE);

    if (customer_thread != RT_NULL)
    {
        rt_thread_startup(customer_thread);
    }

    /* 创建餐厅线程 */
    rt_thread_t restaurant_thread = rt_thread_create("restaurant",
                                                      restaurant_thread_entry,
                                                      RT_NULL,
                                                      THREAD_STACK_SIZE,
                                                      THREAD_PRIORITY,
                                                      THREAD_TIMESLICE);

    if (restaurant_thread != RT_NULL)
    {
        rt_thread_startup(restaurant_thread);
    }

    return 0;
}

实验现象：

6. 邮箱的应用场合

邮箱是一种简单高效的线程间消息传递方式。在 RT-Thread 中，邮箱可以传递一个 4 字节大小的邮件，并且邮箱具有一定的缓冲功能。邮箱的容量决定了它可以缓存的邮件数量。

由于邮箱每次只能传递 4 字节数据，所以它适用于传递小于等于 4 字节的消息。在 32 位系统中，4 字节恰好可以容纳一个指针，因此可以利用邮箱传递指向缓冲区的指针，从而间接实现传递较大的消息。例如：

struct msg
{
    rt_uint8_t *data_ptr;
    rt_uint32_t data_size;
};

当一个线程需要传递以上结构体 msg 时，可以先动态分配结构体，再将指向数据的指针 data_ptr 和数据块长度 data_size 写入，然后将指向这个结构体的指针作为邮件发送到邮箱中：

struct msg* msg_ptr;

msg_ptr = (struct msg*)rt_malloc(sizeof(struct msg));
msg_ptr->data_ptr = ...; /* 指向相应的数据块地址 */
msg_ptr->data_size = len; /* 数据块的长度 */
/* 发送这个消息指针给 mb 邮箱 */
rt_mb_send(mb, (rt_uint32_t)msg_ptr);

接收线程在接收邮件时，需要将接收到的指针转换为 struct msg* 类型，使用完成后，需要释放动态分配的内存：

struct msg* msg_ptr;
if (rt_mb_recv(mb, (rt_uint32_t*)&msg_ptr) == RT_EOK)
{
    /* 在接收线程处理完毕后，需要释放相应的内存块 */
    rt_free(msg_ptr);
}

好的，这次的内容就到这里啦

感谢你的阅读，欢迎点赞、关注、转发

我们，下次再见！

本文使用 markdown.com.cn 排版