RT-Thread快速入门-消息邮箱

首发，公众号【一起学嵌入式】

前面几篇文章介绍了线程（任务）间的同步机制：信号量、互斥量、事件集。接下来我们学习线程（任务）之间的通信机制。

一般来说，RTOS 均会提供两种线程间通信的机制：消息邮箱和消息队列。RT-Thread 同样如此。

本篇文章介绍 RT-Thread 消息邮箱相关的内容。

邮箱的工作机制

1. 理解消息邮箱

邮箱是一种简单的线程间消息传递的方式，其特点是开销较低，效率较高。邮箱中的每一封邮件可以容纳固定大小的内容（针对 32 位处理器，可容纳 4 字节内容，所以一封邮件恰好可以容纳一个指针）。

邮箱的工作示意图如下，中断服务例程或者线程把一封 4 字节长度的邮件发送到邮箱中，一个或多个线程可以从邮箱中读取这些邮件并进行处理。

在中断服务例程中，只能用非阻塞的方式发送邮件。线程中可以设定发送超时时间，以阻塞的方式发送邮件。

当一个线程向邮箱发送邮件时，如果邮箱没满，将把邮件复制到邮箱中。如果邮箱已经满了，发送线程可以设置超时时间，选择等待挂起或直接返回 -RT_EFULL 。

接收邮件过程中，当邮箱中不存在邮件且超时时间不为 0 时，邮件收取过程将变成阻塞方式。此时，只能由线程进行邮件的收取。

2. 邮箱控制块

RT-Thread 中管理邮箱的数据结构为邮箱控制块，有结构体 struct rt_mailbox 表示。另外，rt_mailbox_t 表示的是邮箱的句柄，即指向邮箱控制块的指针。邮箱控制块结构体定义如下：

struct rt_mailbox
{
    struct rt_ipc_object parent;      /* 继承自 ipc_object 类 */

    rt_ubase_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 容器管理。结构体 rt_ipc_object 定义如下：

struct rt_object
{
    char       name[RT_NAME_MAX]; /* 内核对象名称 */
    rt_uint8_t type;              /* 内核对象类型 */
    rt_uint8_t flag;              /* 内核对象的参数 */

#ifdef RT_USING_MODULE
    void      *module_id;  /* 应用程序模块 ID */
#endif
    rt_list_t  list;       /* 内核对象管理链表 */
};

struct rt_ipc_object
{
    struct rt_object parent;          /* 继承自 rt_object */
    rt_list_t        suspend_thread;  /* 挂起的线程链表 */
};

结构体定义中，继承关系一目了然，不再赘述。

管理邮箱

RT-Thread 邮箱相关的操作函数如下所示，包含：创建 / 初始化邮箱、发送邮件、接收邮件、删除 / 脱离邮
箱。

本文只重点介绍几种常用的接口函数。

1. 创建邮箱

RT-Thread 创建一个邮箱有两种方式：动态创建、静态初始化。

动态创建一个邮箱的系统函数如下，调用这个函数创建一个邮箱时，内核会先从对象管理器中分配一个邮箱对象，然后创建一个邮箱控制块，接着对邮箱控制块进行初始化，包括邮箱缓冲区地址、邮件数目、发送邮件在邮箱中的偏移等。

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

在对邮箱控制块初始化期间，内核会动态分配一块内存空间用来存放消息邮件，这块内存的大小等于邮件大小（4 字节）与邮箱容量的乘积。

rt_mb_create()函数的参数，name 为邮箱名称；size 表示邮箱容量；flag 为邮箱的标志，取值为 RT_IPC_FLAG_FIFO 或 RT_IPC_FLAG_PRIO。

邮箱创建成功，则返回邮箱控制块指针；创建失败，则返回 RT_NULL。

静态方式创建邮箱需要两步：（1）定义一个邮箱控制块和一段存放邮件的缓冲区（2）对邮箱控制块进行初始化。

邮箱控制块初始化函数接口如下：

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

参数 mb 为邮箱控制块的指针；name 为邮箱名称；msgpool 为邮箱缓冲区指针；size 为邮箱容量；flag 为邮箱标志，与 rt_mb_create() 相同。

这里的 size 参数指定的是邮箱的容量，即如果 msgpool 指向的缓冲区的字节数是 N，那么邮箱容量
应该是 N/4 。

函数rt_mb_init() 的返回值为 RT_EOK。

创建邮箱的标志变量取值有两种：

RT_IPC_FLAG_FIFO，等待邮箱的线程按照先进先出的方式进行排列。
RT_IPC_FLAG_PRIO，等待邮箱的线程按照优先级的方式进行排列。

2. 发送邮件

RT-Thread 提供的发送邮件接口函数有两种：一种是无等待超时接口，一种是有等待超时。

线程或者中断服务程序可以通过邮箱给其他线程发送消息，发送邮件的函数接口如下，此函数没有等待超时参数。

rt_err_t rt_mb_send (rt_mailbox_t mb, rt_ubase_t value)

参数 mb 为邮箱对象的句柄；value 为邮件内容。

发送成功，函数返回 RT_EOK；发送失败，返回 -RT_EFULL，表示邮箱已经满了。

等待方式发送邮件的函数接口如下，这个函数有等待超时参数：

rt_err_t rt_mb_send_wait (rt_mailbox_t mb,
                          rt_ubase_t value,
                          rt_int32_t timeout)

此函数的参数 timeout 为发送等待超时时间，单位为系统时钟节拍。其他参数与 rt_mb_send() 相同。

如果邮箱已经满了，发送线程会根据设定的 timeout 参数等待邮箱中因为收取邮件而空出空间。若超时时间到达依然没有空出空间，则发送线程将会被唤醒并返回错误码。

返回 RT_EOK 表示发送成功；返回 -RT_ETIMEOUT 表示超时；返回 -RT_ERROR 表示发送失败。

邮件的内容可以是 32 位任意格式的数据，一个整型值或一个指向某个缓冲区的指针。可以根据自己的实际应用进行设定。

注意：在中断服务例程中发送邮件时，应该采用无等待延时的方式发送，直接使用 rt_mb_send() 或者等待超时设定为 0 的函数rt_mb_send_wait()。

3.接收邮件

线程接收邮件的函数接口如下，线程接收邮件时，需要指定接收邮件的邮箱句柄、邮件存放位置以及等待的超时时间。

rt_err_t rt_mb_recv (rt_mailbox_t mb, rt_ubase_t *value, rt_int32_t timeout)

参数 mb 为邮箱的句柄；value 为邮箱消息存储地址；timeout 为等待超时时间。

接收成功，则返回 RT_EOK；接收超时，则返回 -RT_ETIMEOUT；接收失败，返回 -RT_ERROR。

只有当邮箱中有邮件时，接收者才能立即取到邮件并返回 RT_EOK；否则接收线程会根据设定的超时时间，挂起在等待线程队列或者立即返回（超时时间设定为 0）。

实战演练

举例来说明邮箱操作函数的用法，代码如下。动态创建两个线程，一个线程往邮箱中发送邮件，一个线程从邮箱中收取邮件。

#include <rtthread.h>

#define THREAD_PRIORITY 8
#define THREAD_TIMESLICE 5

/* 邮 箱 控 制 块 */
rt_mailbox_t mb_handle;

static char mb_str1[] = "I'm a mail!";
static char mb_str2[] = "this is another mail!";
static char mb_str3[] = "over";

/* 线程 1 入口 */
static void thread1_entry(void *parameter)
{
    char *str;

    while (1)
    {
        rt_kprintf("thread1: try to recv a mail\n");
        
        /* 从邮箱中收取邮件 */
        if (rt_mb_recv(mb_handle, (rt_ubase_t *)&str, RT_WAITING_FOREVER) == RT_EOK)
        {
            rt_kprintf("thread1: get a mail from mailbox, the content:%s\n", str);
            if (str == mb_str3)
            {
                break;
            }
            /* 延时 100ms */
            rt_thread_mdelay(100);
        }    
    }
}

/* 线程 2 入口 */
static void thread2_entry(void *parameter)
{
    rt_uint8_t count = 0;
    
    while (count < 10)
    {
        count ++;
        if (count & 0x1)
        {
            /* 发送 mb_str1 地址到邮箱中 */
            rt_mb_send(mb_handle, (rt_uint32_t)&mb_str1);
        }
        else
        {
            /* 发送 mb_str2 地址到邮箱中 */
            rt_mb_send(mb_handle, (rt_uint32_t)&mb_str2);
        }
        /* 延 时 200ms */
        rt_thread_mdelay(200);
    }
    /* 发送邮件告诉线程 1， 线程 2 已经运行结束 */
    rt_mb_send(mb_handle, (rt_uint32_t)&mb_str3);
}

int main()
{
    /* 线程控制块指针 */
    rt_thread_t thread1 = RT_NULL;
    rt_thread_t thread2 = RT_NULL;    

    /* 创建一个邮箱 */
    mb_handle = rt_mb_create("mt", 32, RT_IPC_FLAG_FIFO);
    if (mb_handle == RT_NULL)
    {
        rt_kprintf("create mailbox failed.\n");
        return -1;
    }

    /* 动态创建线程1 */
    thread1 = rt_thread_create("thread1", thread1_entry, RT_NULL,
                    1024, THREAD_PRIORITY - 1, THREAD_TIMESLICE);
    
    if(thread1 != RT_NULL)
    {
        /* 启动线程 */
        rt_thread_startup(thread1);
    }

    /* 动态创建线程2 */
    thread2 = rt_thread_create("thread2", thread2_entry, RT_NULL,
                    1024, THREAD_PRIORITY, THREAD_TIMESLICE);
    if(thread2 != RT_NULL)
    {
        /* 启动线程 */
        rt_thread_startup(thread2);
    }    
}

编译，运行结果如下：