IOT-OS之RT-Thread（六）--- 线程间同步与线程间通信

文章目录

之前的博客UCOS任务间同步与通信介绍了信号量、互斥量等任务间同步机制，消息邮箱、消息队列等任务间通信机制的实现原理，本文主要从RT-Thread与UCOS的对比与差异看RT-Thread线程间同步与通信对象管理的实现。

一、IPC对象管理

再回顾下内核对象的派生和继承关系：
内核对象派生继承关系

1.1 IPC对象控制块

前面已经介绍过直接继承自基对象rt_object的定时器对象rt_timer、内存池对象rt_mempool、线程对象rt_thread，下面要介绍线程间的同步与通信，线程间同步对象rt_sem / rt_mutex / rt_event和线程间通信对象rt_mb / rt_mq都直接继承自rt_ipc_object，而IPC对象又继承自基对象rt_object，在介绍线程间同步与通信对象前先介绍派生IPC对象rt_ipc_object，其数据结构如下：

// rt-thread-4.0.1\include\rtdef.h
/**
 * Base structure of IPC object
 */
struct rt_ipc_object
{
    struct rt_object parent;                            /**< inherit from rt_object */

    rt_list_t        suspend_thread;                    /**< threads pended on this resource */
};

/**
 * IPC flags and control command definitions
 */
#define RT_IPC_FLAG_FIFO                0x00            /**< FIFOed IPC. @ref IPC. */
#define RT_IPC_FLAG_PRIO                0x01            /**< PRIOed IPC. @ref IPC. */

#define RT_IPC_CMD_UNKNOWN              0x00            /**< unknown IPC command */
#define RT_IPC_CMD_RESET                0x01            /**< reset IPC object */

#define RT_WAITING_FOREVER              -1              /**< Block forever until get resource. */
#define RT_WAITING_NO                   0               /**< Non-block. */

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22

IPC对象rt_ipc_object继承自基对象rt_object，rt_ipc_object.parent.flag用以标识实际IPC对象的处理方式，根据上面的宏定义总结如下：

flag位01备注
bit0RT_IPC_FLAG_FIFO：按消息队列先进先出的方式处理.RT_IPC_FLAG_PRIO：按线程优先级的方式处理，即哪个线程的优先级高则哪个先操作IPC处理方式

rt_ipc_object继承自基对象rt_object的另外几个成员根据具体对象类型取值有所不同，等介绍具体对象类型时再介绍。

rt_ipc_object唯一的私有对象rt_ipc_object.suspend_thread是挂起等待线程链表节点，这些挂起等待线程共同构成了事件等待链表，事件挂起链表的组织顺序跟前面介绍的flag有关，如果设置RT_IPC_FLAG_FIFO则挂起线程链表按先入先出排序，如果设置RT_IPC_FLAG_PRIO则挂起线程按优先级排序。

1.2 IPC对象接口函数

派生IPC对象rt_ipc_object由于只有一个私有成员suspend_thread，对其的操作主要也就是链表初始化、链表节点插入/移除等，IPC对象的初始化函数与挂起线程函数实现代码如下：

// rt-thread-4.0.1\src\ipc.c

/**
 * This function will initialize an IPC object
 *
 * @param ipc the IPC object
 *
 * @return the operation status, RT_EOK on successful
 */
rt_inline rt_err_t rt_ipc_object_init(struct rt_ipc_object *ipc)
{
    /* init ipc object */
    rt_list_init(&(ipc->suspend_thread));

    return RT_EOK;
}

/**
 * This function will suspend a thread to a specified list. IPC object or some
 * double-queue object (mailbox etc.) contains this kind of list.
 *
 * @param list the IPC suspended thread list
 * @param thread the thread object to be suspended
 * @param flag the IPC object flag,
 *        which shall be RT_IPC_FLAG_FIFO/RT_IPC_FLAG_PRIO.
 *
 * @return the operation status, RT_EOK on successful
 */
rt_inline rt_err_t rt_ipc_list_suspend(rt_list_t        *list,
                                       struct rt_thread *thread,
                                       rt_uint8_t        flag)
{
    /* suspend thread */
    rt_thread_suspend(thread);

    switch (flag)
    {
    case RT_IPC_FLAG_FIFO:
        rt_list_insert_before(list, &(thread->tlist));
        break;

    case RT_IPC_FLAG_PRIO:
        {
            struct rt_list_node *n;
            struct rt_thread *sthread;

            /* find a suitable position */
            for (n = list->next; n != list; n = n->next)
            {
                sthread = rt_list_entry(n, struct rt_thread, tlist);

                /* find out */
                if (thread->current_priority < sthread->current_priority)
                {
                    /* insert this thread before the sthread */
                    rt_list_insert_before(&(sthread->tlist), &(thread->tlist));
                    break;
                }
            }

            /*
             * not found a suitable position,
             * append to the end of suspend_thread list
             */
            if (n == list)
                rt_list_insert_before(list, &(thread->tlist));
        }
        break;
    }

    return RT_EOK;
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72

函数rt_ipc_list_suspend有一个参数flag正好表示前面介绍的IPC处理方式，RT_IPC_FLAG_FIFO比较简单快速，RT_IPC_FLAG_PRIO能实现更好的实时性，用户根据需求选择传入的flag值。

有挂起自然有唤醒，考虑到IPC对象可能会被删除或脱离，此时挂起在该IPC对象上的所有线程都需要被唤醒，因此还提供了唤醒全部挂起线程的函数，实现代码如下：

// rt-thread-4.0.1\src\ipc.c

/**
 * This function will resume the first thread in the list of a IPC object:
 * - remove the thread from suspend queue of IPC object
 * - put the thread into system ready queue
 *
 * @param list the thread list
 *
 * @return the operation status, RT_EOK on successful
 */
rt_inline rt_err_t rt_ipc_list_resume(rt_list_t *list)
{
    struct rt_thread *thread;

    /* get thread entry */
    thread = rt_list_entry(list->next, struct rt_thread, tlist);

    RT_DEBUG_LOG(RT_DEBUG_IPC, ("resume thread:%s\n", thread->name));

    /* resume it */
    rt_thread_resume(thread);

    return RT_EOK;
}

/**
 * This function will resume all suspended threads in a list, including
 * suspend list of IPC object and private list of mailbox etc.
 *
 * @param list of the threads to resume
 *
 * @return the operation status, RT_EOK on successful
 */
rt_inline rt_err_t rt_ipc_list_resume_all(rt_list_t *list)
{
    struct rt_thread *thread;
    register rt_ubase_t temp;

    /* wakeup all suspend threads */
    while (!rt_list_isempty(list))
    {
        /* disable interrupt */
        temp = rt_hw_interrupt_disable();

        /* get next suspend thread */
        thread = rt_list_entry(list->next, struct rt_thread, tlist);
        /* set error code to RT_ERROR */
        thread->error = -RT_ERROR;

        /*
         * resume thread
         * In rt_thread_resume function, it will remove current thread from
         * suspend list
         */
        rt_thread_resume(thread);

        /* enable interrupt */
        rt_hw_interrupt_enable(temp);
    }

    return RT_EOK;
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63

二、线程间同步对象管理

同步是指按预定的先后次序进行运行，线程同步是指多个线程通过特定的机制（如互斥量，事件对象，临界区）来控制线程之间的执行顺序，也可以说是在线程之间通过同步建立起执行顺序的关系，如果没有同步，那线程之间将是无序的。

线程的同步方式有很多种，其核心思想都是：在访问临界区的时候只允许一个 (或一类) 线程运行。进入 / 退出临界区的方式有很多种：

  1. 调用 rt_hw_interrupt_disable() 进入临界区，调用 rt_hw_interrupt_enable()
    退出临界区；
  2. 调用 rt_enter_critical() 进入临界区，调用 rt_exit_critical() 退出临界区

2.1 信号量对象管理

信号量是一种轻型的用于解决线程间同步问题的内核对象，线程可以获取或释放它，从而达到同步或互斥的目的。

信号量工作示意图如下图所示，每个信号量对象都有一个信号量值和一个线程等待队列，信号量的值对应了信号量对象的实例数目、资源数目，假如信号量值为 5，则表示共有 5 个信号量实例（资源）可以被使用，当信号量实例数目为零时，再申请该信号量的线程就会被挂起在该信号量的等待队列上，等待可用的信号量实例（资源）。
信号量工作示意图

  • 信号量控制块

在 RT-Thread 中，信号量控制块是操作系统用于管理信号量的一个数据结构，由结构体 struct rt_semaphore 表示。另外一种 C 表达方式 rt_sem_t，表示的是信号量的句柄，在 C 语言中的实现是指向信号量控制块的指针。信号量控制块结构的详细定义如下：

// rt-thread-4.0.1\include\rtdef.h
/**
 * Semaphore structure
 */
struct rt_semaphore
{
    struct rt_ipc_object parent;                        /**< inherit from ipc_object */

    rt_uint16_t          value;                         /**< value of semaphore. */
    rt_uint16_t          reserved;                      /**< reserved field */
};
typedef struct rt_semaphore *rt_sem_t;

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12

rt_semaphore 对象从 rt_ipc_object 中派生，rt_semaphore.parent.parent.type值为RT_Object_Class_Semaphor，rt_semaphore.parent.parent.list为已初始化/创建信号量链表节点，组织成一个信号量链表（IPC对象不像定时器或线程对象分好几种状态，只维护一个IPC对象链表即可）。

rt_semaphore对象有两个私有成员，但一个保留未用，相当于只有一个私有成员rt_semaphore.value，信号量的最大值是 65535，有成员类型rt_uint16_t决定。

  • 信号量接口函数

信号量控制块中含有信号量相关的重要参数，在信号量各种状态间起到纽带的作用。信号量相关接口如下图所示，对一个信号量的操作包含：创建 / 初始化信号量、获取信号量、释放信号量、删除 / 脱离信号量。
信号量接口函数
先看信号量的构造/析构函数原型（依然分静态对象与动态对象）：

// rt-thread-4.0.1\include\rtdef.h
/**
 * This function will initialize a semaphore and put it under control of
 * resource management.
 *
 * @param sem the semaphore object
 * @param name the name of semaphore
 * @param value the init value of semaphore
 * @param flag the flag of semaphore
 *
 * @return the operation status, RT_EOK on successful
 */
rt_err_t rt_sem_init(rt_sem_t    sem,
                     const char *name,
                     rt_uint32_t value,
                     rt_uint8_t  flag);

/**
 * This function will detach a semaphore from resource management
 *
 * @param sem the semaphore object
 *
 * @return the operation status, RT_EOK on successful
 *
 * @see rt_sem_delete
 */
rt_err_t rt_sem_detach(rt_sem_t sem);

/**
 * This function will create a semaphore from system resource
 *
 * @param name the name of semaphore
 * @param value the init value of semaphore
 * @param flag the flag of semaphore
 *
 * @return the created semaphore, RT_NULL on error happen
 *
 * @see rt_sem_init
 */
rt_sem_t rt_sem_create(const char *name, rt_uint32_t value, rt_uint8_t flag);

/**
 * This function will delete a semaphore object and release the memory
 *
 * @param sem the semaphore object
 *
 * @return the error code
 *
 * @see rt_sem_detach
 */
rt_err_t rt_sem_delete(rt_sem_t sem);

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51

再看信号量的获取rt_sem_take与释放rt_sem_release函数原型：

// rt-thread-4.0.1\include\rtdef.h
/**
 * This function will take a semaphore, if the semaphore is unavailable, the
 * thread shall wait for a specified time.
 *
 * @param sem the semaphore object
 * @param time the waiting time
 *
 * @return the error code
 */
rt_err_t rt_sem_take(rt_sem_t sem, rt_int32_t time);

/**
 * This function will try to take a semaphore and immediately return
 *
 * @param sem the semaphore object
 *
 * @return the error code
 */
rt_err_t rt_sem_trytake(rt_sem_t sem);

/**
 * This function will release a semaphore, if there are threads suspended on
 * semaphore, it will be waked up.
 *
 * @param sem the semaphore object
 *
 * @return the error code
 */
rt_err_t rt_sem_release(rt_sem_t sem);

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30

最后看信号量控制函数rt_sem_control，前面介绍IPC对象时也提到了两个命令宏定义，IPC对象支持的控制函数命令只有一种RT_IPC_CMD_RESET，对于RT_IPC_CMD_UNKNOWN可能是保留备用吧。rt_sem_control的作用是重新设置信号量值，函数原型如下：

// rt-thread-4.0.1\include\rtdef.h
/**
 * This function can get or set some extra attributions of a semaphore object.
 *
 * @param sem the semaphore object
 * @param cmd the execution command
 * @param arg the execution argument
 *
 * @return the error code
 */
rt_err_t rt_sem_control(rt_sem_t sem, int cmd, void *arg);

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11

信号量是一种非常灵活的同步方式，可以运用在多种场合中。形成锁、同步、资源计数等关系，也能方便的用于线程与线程、中断与线程间的同步中。中断与线程间的互斥不能采用信号量（锁）的方式，而应采用开关中断的方式。

一般资源计数类型多是混合方式的线程间同步，因为对于单个的资源处理依然存在线程的多重访问，这就需要对一个单独的资源进行访问、处理，并进行锁方式的互斥操作。

2.2 互斥量对象管理

互斥量又叫相互排斥的信号量，是一种特殊的二值信号量。互斥量和信号量不同的是：拥有互斥量的线程拥有互斥量的所有权，互斥量支持递归访问且能防止线程优先级翻转；并且互斥量只能由持有线程释放，而信号量则可以由任何线程释放。

互斥量的状态只有两种，开锁或闭锁（两种状态值）。当有线程持有它时，互斥量处于闭锁状态，由这个线程获得它的所有权。相反，当这个线程释放它时，将对互斥量进行开锁，失去它的所有权。当一个线程持有互斥量时，其他线程将不能够对它进行开锁或持有它，持有该互斥量的线程也能够再次获得这个锁而不被挂起，如下图时所示。这个特性与一般的二值信号量有很大的不同：在信号量中，因为已经不存在实例，线程递归持有会发生主动挂起（最终形成死锁）。
互斥量工作机制
使用信号量会导致的另一个潜在问题是线程优先级翻转问题，优先级反转在博客UCOS任务间同步与通信中介绍过，这里不再赘述了，RT-Thread与UCOS类似，都是采用优先级继承算法来解决优先级翻转问题的。在获得互斥量后，请尽快释放互斥量，并且在持有互斥量的过程中，不得再行更改持有互斥量线程的优先级。

  • 互斥量控制块

在 RT-Thread 中，互斥量控制块是操作系统用于管理互斥量的一个数据结构，由结构体 struct rt_mutex 表示。另外一种 C 表达方式 rt_mutex_t，表示的是互斥量的句柄，在 C 语言中的实现是指互斥量控制块的指针。互斥量控制块结构的详细定义请见以下代码：

// rt-thread-4.0.1\include\rtdef.h
/**
 * Mutual exclusion (mutex) structure
 */
struct rt_mutex
{
    struct rt_ipc_object parent;                        /**< inherit from ipc_object */

    rt_uint16_t          value;                         /**< value of mutex */

    rt_uint8_t           original_priority;             /**< priority of last thread hold the mutex */
    rt_uint8_t           hold;                          /**< numbers of thread hold the mutex */

    struct rt_thread    *owner;                         /**< current owner of mutex */
};
typedef struct rt_mutex *rt_mutex_t;

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16

rt_mutex 对象从 rt_ipc_object 中派生，rt_mutex.parent.parent.type值为RT_Object_Class_Mutex，rt_mutex.parent.parent.list互斥量对象链表节点，所有互斥量组织成一个互斥量对象链表。

rt_mutex 对象私有成员比信号量多，rt_mutex.value为互斥量值，一般为0或1；rt_mutex.original_priority为持有线程的原始优先级，用于优先级继承算法；rt_mutex.hold为持有线程的持有次数，互斥量支持持有线程对其多次嵌套持有；
rt_mutex.owner指向当前拥有互斥量的线程地址。

  • 互斥量接口函数

互斥量控制块中含有互斥相关的重要参数，在互斥量功能的实现中起到重要的作用。互斥量相关接口如下图所示，对一个互斥量的操作包含：创建 / 初始化互斥量、获取互斥量、释放互斥量、删除 / 脱离互斥量。
互斥量接口函数
先看互斥量构造、析构函数原型：

// rt-thread-4.0.1\src\ipc.c
/**
 * This function will initialize a mutex and put it under control of resource
 * management.
 *
 * @param mutex the mutex object
 * @param name the name of mutex
 * @param flag the flag of mutex
 *
 * @return the operation status, RT_EOK on successful
 */
rt_err_t rt_mutex_init(rt_mutex_t mutex, const char *name, rt_uint8_t flag);

/**
 * This function will detach a mutex from resource management
 *
 * @param mutex the mutex object
 *
 * @return the operation status, RT_EOK on successful
 *
 * @see rt_mutex_delete
 */
rt_err_t rt_mutex_detach(rt_mutex_t mutex);

/**
 * This function will create a mutex from system resource
 *
 * @param name the name of mutex
 * @param flag the flag of mutex
 *
 * @return the created mutex, RT_NULL on error happen
 *
 * @see rt_mutex_init
 */
rt_mutex_t rt_mutex_create(const char *name, rt_uint8_t flag);

/**
 * This function will delete a mutex object and release the memory
 *
 * @param mutex the mutex object
 *
 * @return the error code
 *
 * @see rt_mutex_detach
 */
rt_err_t rt_mutex_delete(rt_mutex_t mutex);

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46

接下来看互斥量的获取rt_mutex_take与释放rt_mutex_release函数原型：

// rt-thread-4.0.1\src\ipc.c
/**
 * This function will take a mutex, if the mutex is unavailable, the
 * thread shall wait for a specified time.
 *
 * @param mutex the mutex object
 * @param time the waiting time
 *
 * @return the error code
 */
rt_err_t rt_mutex_take(rt_mutex_t mutex, rt_int32_t time);

/**
 * This function will release a mutex, if there are threads suspended on mutex,
 * it will be waked up.
 *
 * @param mutex the mutex object
 *
 * @return the error code
 */
rt_err_t rt_mutex_release(rt_mutex_t mutex);

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21

由于互斥量的获取与释放一般在同一个线程内部，互斥量值在初始化时设定为1，所以互斥量构造函数不需要传入互斥量值参数，互斥量控制函数也就变得没有意义，所以互斥量并没有实现控制命令。

互斥量的使用比较单一，因为它是信号量的一种，并且它是以锁的形式存在。在初始化的时候，互斥量永远都处于开锁的状态，而被线程持有的时候则立刻转为闭锁的状态，需要切记的是互斥量不能在中断服务例程中使用（中断服务例程使用开关中断达到互斥目的）。互斥量更适合于：

  1. 线程多次持有互斥量的情况下。这样可以避免同一线程多次递归持有而造成死锁的问题。
  2. 可能会由于多线程同步而造成优先级翻转的情况。

2.3 事件集对象管理

事件集主要用于线程间的同步，与信号量不同，它的特点是可以实现一对多，多对多的同步。即一个线程与多个事件的关系可设置为：其中任意一个事件唤醒线程，或几个事件都到达后才唤醒线程进行后续的处理；同样，事件也可以是多个线程同步多个事件。这种多个事件的集合可以用一个 32 位无符号整型变量来表示，变量的每一位代表一个事件，线程通过 “逻辑与” 或“逻辑或”将一个或多个事件关联起来，形成事件组合。事件的 “逻辑或” 也称为是独立型同步，指的是线程与任何事件之一发生同步；事件 “逻辑与” 也称为是关联型同步，指的是线程与若干事件都发生同步。

RT-Thread 定义的事件集有以下特点：

  1. 事件只与线程相关，事件间相互独立：每个线程可拥有 32 个事件标志，采用一个 32 bit 无符号整型数进行记录，每一个 bit 代表一个事件；
  2. 事件仅用于同步，不提供数据传输功能；
  3. 事件无排队性，即多次向线程发送同一事件 (如果线程还未来得及读走)，其效果等同于只发送一次。

在 RT-Thread 中，每个线程都拥有一个事件信息标记，它有三个属性，分别是 RT_EVENT_FLAG_AND(逻辑与)，RT_EVENT_FLAG_OR(逻辑或）以及 RT_EVENT_FLAG_CLEAR(清除标记）。当线程等待事件同步时，可以通过 32 个事件标志和这个事件信息标记来判断当前接收的事件是否满足同步条件。
事件标志发送与接收
如上图所示，线程 #1 的事件标志中第 1 位和第 30 位被置位，如果事件信息标记位设为逻辑与，则表示线程 #1 只有在事件 1 和事件 30 都发生以后才会被触发唤醒，如果事件信息标记位设为逻辑或，则事件 1 或事件 30 中的任意一个发生都会触发唤醒线程 #1。如果信息标记同时设置了清除标记位，则当线程 #1 唤醒后将主动把事件 1 和事件 30 清为零，否则事件标志将依然存在（即置 1）。

  • 事件集控制块

在 RT-Thread 中，事件集控制块是操作系统用于管理事件的一个数据结构，由结构体 struct rt_event 表示。另外一种 C 表达方式 rt_event_t，表示的是事件集的句柄，在 C 语言中的实现是事件集控制块的指针。事件集控制块结构的详细定义请见以下代码：

// rt-thread-4.0.1\include\rtdef.h
/*
 * event structure
 */
struct rt_event
{
    struct rt_ipc_object parent;                        /**< inherit from ipc_object */

    rt_uint32_t          set;                           /**< event set */
};
typedef struct rt_event *rt_event_t;

/**
 * flag defintions in event
 */
#define RT_EVENT_FLAG_AND               0x01            /**< logic and */
#define RT_EVENT_FLAG_OR                0x02            /**< logic or */
#define RT_EVENT_FLAG_CLEAR             0x04            /**< clear flag */

/**
 * Thread structure
 */
struct rt_thread
{
	......
#if defined(RT_USING_EVENT)
    /* thread event */
    rt_uint32_t event_set;
    rt_uint8_t  event_info;
#endif
	......
};
typedef struct rt_thread *rt_thread_t;

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33

rt_event 对象从 rt_ipc_object 中派生，rt_event.parent.parent.type值为RT_Object_Class_Event，rt_event.parent.parent.list为事件集对象链表节点，所有事件集对象组织成一个链表。

rt_event 对象也只有一个私有成员，rt_event.set表示一个32位的事件集合，每个位标识一个事件，bit值可以标记某事件是否发生。

由于事件集要实现一对多、多对多的线程间同步，只靠事件集对象很难满足需求，还要求线程对象配合管理自己所需要的事件集。前面介绍线程管理时提到被条件宏RT_USING_EVENT包含的两个成员rt_thread.event_set和rt_thread.event_info就是线程对象配合事件集实现同步的关键。rt_thread.event_set表示该线程等待的事件集；rt_thread.event_info表示该线程事件集中多个事件的逻辑组合，取值跟上面flag定义一致，可以是RT_EVENT_FLAG_AND / OR / CLEAR。

  • 事件集接口函数

先看事件集的构造、析构函数原型：

// rt-thread-4.0.1\src\ipc.c
/**
 * This function will initialize an event and put it under control of resource
 * management.
 *
 * @param event the event object
 * @param name the name of event
 * @param flag the flag of event
 *
 * @return the operation status, RT_EOK on successful
 */
rt_err_t rt_event_init(rt_event_t event, const char *name, rt_uint8_t flag);

/**
 * This function will detach an event object from resource management
 *
 * @param event the event object
 *
 * @return the operation status, RT_EOK on successful
 */
rt_err_t rt_event_detach(rt_event_t event);

/**
 * This function will create an event object from system resource
 *
 * @param name the name of event
 * @param flag the flag of event
 *
 * @return the created event, RT_NULL on error happen
 */
rt_event_t rt_event_create(const char *name, rt_uint8_t flag);

/**
 * This function will delete an event object and release the memory
 *
 * @param event the event object
 *
 * @return the error code
 */
rt_err_t rt_event_delete(rt_event_t event);

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40

接下来看事件集的发送rt_event_send与接收rt_event_recv函数原型：

// rt-thread-4.0.1\src\ipc.c
/**
 * This function will send an event to the event object, if there are threads
 * suspended on event object, it will be waked up.
 *
 * @param event the event object
 * @param set the event set
 *
 * @return the error code
 */
rt_err_t rt_event_send(rt_event_t event, rt_uint32_t set);

/**
 * This function will receive an event from event object, if the event is
 * unavailable, the thread shall wait for a specified time.
 *
 * @param event the fast event object
 * @param set the interested event set
 * @param option the receive option, either RT_EVENT_FLAG_AND or
 *        RT_EVENT_FLAG_OR should be set.
 * @param timeout the waiting time
 * @param recved the received event, if you don't care, RT_NULL can be set.
 *
 * @return the error code
 */
rt_err_t rt_event_recv(rt_event_t   event,
                       rt_uint32_t  set,
                       rt_uint8_t   option,
                       rt_int32_t   timeout,
                       rt_uint32_t *recved);

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30

事件集接收函数rt_event_recv先根据期望接收的事件集set和事件逻辑组合关系option确定要想接收的事件组合，再到事件集对象event中查找看期望的事件组合是否已发送，若已发送则立即返回，若未完全发送，则将当前线程期望接收的事件集set和逻辑组合关系option分别保存到当前线程的event_set和event_info成员中，然后将当前线程挂起等待。

事件集发送函数rt_event_send先将发送的事件集set保存到事件集对象event中，再遍历已挂起线程链表，根据各线程等待的事件组合判断该线程是否已满足唤醒条件，若某挂起线程等待的事件组合均已收到，则唤醒该线程。

事件集控制函数实际上就是重置事件集对象值，也即把事件集对象的事件集rt_event.set重置为0（事件集对象构造时也将其置为0），rt_event_control的函数原型如下：

// rt-thread-4.0.1\src\ipc.c
/**
 * This function can get or set some extra attributions of an event object.
 *
 * @param event the event object
 * @param cmd the execution command
 * @param arg the execution argument
 *
 * @return the error code
 */
rt_err_t rt_event_control(rt_event_t event, int cmd, void *arg);

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11

事件集可使用于多种场合，它能够在一定程度上替代信号量，用于线程间同步。一个线程或中断服务例程发送一个事件给事件集对象，而后等待的线程被唤醒并对相应的事件进行处理。但是它与信号量不同的是，事件的发送操作在事件未清除前，是不可累计的，而信号量的释放动作是累计的。事件的另一个特性是，接收线程可等待多种事件，即多个事件对应一个线程或多个线程。同时按照线程等待的参数，可选择是 “逻辑或” 触发还是 “逻辑与” 触发。这个特性也是信号量等所不具备的，信号量只能识别单一的释放动作，而不能同时等待多种类型的释放。如下图所示为多事件接收示意图：
多事件接收示意图
一个事件集中包含 32 个事件，特定线程只等待、接收它关注的事件。可以是一个线程等待多个事件的到来（线程 1、2 均等待多个事件，事件间可以使用 “与” 或者 “或” 逻辑触发线程），也可以是多个线程等待一个事件的到来（事件 25）。当有它们关注的事件发生时，线程将被唤醒并进行后续的处理动作。

三、线程间通信对象管理

在裸机编程中，经常会使用全局变量进行功能间的通信，如某些功能可能由于一些操作而改变全局变量的值，另一个功能对此全局变量进行读取，根据读取到的全局变量值执行相应的动作，达到通信协作的目的。RT-Thread 中则提供了更多的工具帮助在不同的线程中间传递信息，比如邮箱、消息队列、信号。

3.1 邮箱对象管理

RT-Thread 操作系统的邮箱用于线程间通信，特点是开销比较低，效率较高。邮箱中的每一封邮件只能容纳固定的 4 字节内容（针对 32 位处理系统，指针的大小即为 4 个字节，所以一封邮件恰好能够容纳一个指针）。典型的邮箱也称作交换消息，如下图所示，线程或中断服务例程把一封 4 字节长度的邮件发送到邮箱中，而一个或多个线程可以从邮箱中接收这些邮件并进行处理。
邮箱工作机制
非阻塞方式的邮件发送过程能够安全的应用于中断服务中，是线程、中断服务、定时器向线程发送消息的有效手段。通常来说，邮件收取过程可能是阻塞的，这取决于邮箱中是否有邮件，以及收取邮件时设置的超时时间。当邮箱中不存在邮件且超时时间不为 0 时，邮件收取过程将变成阻塞方式。在这类情况下，只能由线程进行邮件的收取。

当一个线程向邮箱发送邮件时，如果邮箱没满，将把邮件复制到邮箱中。如果邮箱已经满了，发送线程可以设置超时时间，选择等待挂起或直接返回 - RT_EFULL。如果发送线程选择挂起等待，那么当邮箱中的邮件被收取而空出空间来时，等待挂起的发送线程将被唤醒继续发送。

当一个线程从邮箱中接收邮件时，如果邮箱是空的，接收线程可以选择是否等待挂起直到收到新的邮件而唤醒，或可以设置超时时间。当达到设置的超时时间，邮箱依然未收到邮件时，这个选择超时等待的线程将被唤醒并返回 - RT_ETIMEOUT。如果邮箱中存在邮件，那么接收线程将复制邮箱中的 4 个字节邮件到接收缓存中。

  • 邮箱控制块

在 RT-Thread 中，邮箱控制块是操作系统用于管理邮箱的一个数据结构，由结构体 struct rt_mailbox 表示。另外一种 C 表达方式 rt_mailbox_t，表示的是邮箱的句柄，在 C 语言中的实现是邮箱控制块的指针。邮箱控制块结构的详细定义请见以下代码：

// rt-thread-4.0.1\include\rtdef.h
/**
 * mailbox structure
 */
struct rt_mailbox
{
    struct rt_ipc_object parent;                        /**< inherit from ipc_object */

    rt_ubase_t          *msg_pool;                      /**< start address of message buffer */

    rt_uint16_t          size;                          /**< size of message pool */

    rt_uint16_t          entry;                         /**< index of messages in msg_pool */
    rt_uint16_t          in_offset;                     /**< input offset of the message buffer */
    rt_uint16_t          out_offset;                    /**< output offset of the message buffer */

    rt_list_t            suspend_sender_thread;         /**< sender thread suspended on this mailbox */
};
typedef struct rt_mailbox *rt_mailbox_t;

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19

rt_mailbox 对象从 rt_ipc_object 中派生，rt_mailbox.parent.parent.type值为RT_Object_Class_MailBox，rt_mailbox.parent.parent.list为邮箱对象链表节点，所有邮箱组织成一个双向链表。

rt_mailbox 对象的私有成员比较多，rt_mailbox.msg_pool指向邮箱缓冲区的开始地址，rt_mailbox.size为邮箱缓冲区的大小，rt_mailbox.entry为邮箱中已有邮件的数目，rt_mailbox.in_offset与rt_mailbox.out_offset为邮箱缓冲区的入口便宜点和出口偏移点（相对于缓冲区开始地址的偏移，可以看作缓冲区邮件数组的索引）。

rt_mailbox 对象最后一个私有成员rt_mailbox.suspend_sender_thread为邮件发送线程的挂起等待队列，之前介绍IPC对象rt_ipc_object也有一个私有成员rt_ipc_object.suspend_thread是等待IPC对象线程的挂起等待队列，邮箱对象为何又增加了一个挂起线程等待队列呢？一般进程间同步或通信都是在等待获取IPC对象时线程才会挂起，rt_ipc_object.suspend_thread保存的也正是获取/接收IPC对象线程的挂起等待队列；rt_mailbox对象除了在邮箱为空时等待接收邮件的线程会挂起，在邮箱已满时等待发送邮件的线程也会挂起，IPC对象已经提供了保存等待接收邮件挂起线程链表的成员，rt_mailbox对象又提供了保存等待发送邮件挂起线程链表的私有成员。

  • 邮箱接口函数

邮箱控制块是一个结构体，其中含有事件相关的重要参数，在邮箱的功能实现中起重要的作用。邮箱的相关接口如下图所示，对一个邮箱的操作包含：创建 / 初始化邮箱、发送邮件、接收邮件、删除 / 脱离邮箱。
邮箱接口函数
先看邮箱对象的构造、析构函数原型：

// rt-thread-4.0.1\src\ipc.c
/**
 * This function will initialize a mailbox and put it under control of resource
 * management.
 *
 * @param mb the mailbox object
 * @param name the name of mailbox
 * @param msgpool the begin address of buffer to save received mail
 * @param size the size of mailbox
 * @param flag the flag of mailbox
 *
 * @return the operation status, RT_EOK on successful
 */
rt_err_t rt_mb_init(rt_mailbox_t mb,
                    const char  *name,
                    void        *msgpool,
                    rt_size_t    size,
                    rt_uint8_t   flag);

/**
 * This function will detach a mailbox from resource management
 *
 * @param mb the mailbox object
 *
 * @return the operation status, RT_EOK on successful
 */
rt_err_t rt_mb_detach(rt_mailbox_t mb);

/**
 * This function will create a mailbox object from system resource
 *
 * @param name the name of mailbox
 * @param size the size of mailbox
 * @param flag the flag of mailbox
 *
 * @return the created mailbox, RT_NULL on error happen
 */
rt_mailbox_t rt_mb_create(const char *name, rt_size_t size, rt_uint8_t flag);

/**
 * This function will delete a mailbox object and release the memory
 *
 * @param mb the mailbox object
 *
 * @return the error code
 */
rt_err_t rt_mb_delete(rt_mailbox_t mb);

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47

接下来看邮箱对象的发送rt_mb_send与接收rt_mb_recv函数原型，由于邮箱对象单独提供了邮件发送线程的挂起等待链表节点成员，邮箱对象还提供了发送等待函数接口，原型如下：

// rt-thread-4.0.1\src\ipc.c
/**
 * This function will send a mail to mailbox object. If the mailbox is full,
 * current thread will be suspended until timeout.
 *
 * @param mb the mailbox object
 * @param value the mail
 * @param timeout the waiting time
 *
 * @return the error code
 */
rt_err_t rt_mb_send_wait(rt_mailbox_t mb,
                         rt_ubase_t   value,
                         rt_int32_t   timeout);

/**
 * This function will send a mail to mailbox object, if there are threads
 * suspended on mailbox object, it will be waked up. This function will return
 * immediately, if you want blocking send, use rt_mb_send_wait instead.
 *
 * @param mb the mailbox object
 * @param value the mail
 *
 * @return the error code
 */
rt_err_t rt_mb_send(rt_mailbox_t mb, rt_ubase_t value);

/**
 * This function will receive a mail from mailbox object, if there is no mail
 * in mailbox object, the thread shall wait for a specified time.
 *
 * @param mb the mailbox object
 * @param value the received mail will be saved in
 * @param timeout the waiting time
 *
 * @return the error code
 */
rt_err_t rt_mb_recv(rt_mailbox_t mb, rt_ubase_t *value, rt_int32_t timeout);

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38

邮箱控制函数rt_mb_control提供了邮箱重置命令，实际上就是将挂起线程唤醒，邮件数目、出入口便宜点清零等，函数原型如下：

// rt-thread-4.0.1\src\ipc.c
/**
 * This function can get or set some extra attributions of a mailbox object.
 *
 * @param mb the mailbox object
 * @param cmd the execution command
 * @param arg the execution argument
 *
 * @return the error code
 */
rt_err_t rt_mb_control(rt_mailbox_t mb, int cmd, void *arg);

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11

邮箱是一种简单的线程间消息传递方式，特点是开销比较低，效率较高。在 RT-Thread 操作系统的实现中能够一次传递一个 4 字节大小的邮件，并且邮箱具备一定的存储功能，能够缓存一定数量的邮件数 (邮件数由创建、初始化邮箱时指定的容量决定)。邮箱中一封邮件的最大长度是 4 字节，所以邮箱能够用于不超过 4 字节的消息传递。由于在 32 系统上 4 字节的内容恰好可以放置一个指针，因此当需要在线程间传递比较大的消息时，可以把指向一个缓冲区的指针作为邮件发送到邮箱中，即邮箱也可以传递指针。

3.2 消息队列对象管理

消息队列是另一种常用的线程间通讯方式，是邮箱的扩展。可以应用在多种场合：线程间的消息交换、使用串口接收不定长数据等。

消息队列能够接收来自线程或中断服务例程中不固定长度的消息，并把消息缓存在自己的内存空间中。其他线程也能够从消息队列中读取相应的消息，而当消息队列是空的时候，可以挂起读取线程。当有新的消息到达时，挂起的线程将被唤醒以接收并处理消息。消息队列是一种异步的通信方式。

如下图所示，线程或中断服务例程可以将一条或多条消息放入消息队列中。同样，一个或多个线程也可以从消息队列中获得消息。当有多个消息发送到消息队列时，通常将先进入消息队列的消息先传给线程，也就是说，线程先得到的是最先进入消息队列的消息，即先进先出原则 (FIFO)。
消息队列工作方式
RT-Thread 操作系统的消息队列对象由多个元素组成，当消息队列被创建时，它就被分配了消息队列控制块：消息队列名称、内存缓冲区、消息大小以及队列长度等。同时每个消息队列对象中包含着多个消息框，每个消息框可以存放一条消息；消息队列中的第一个和最后一个消息框被分别称为消息链表头和消息链表尾，对应于消息队列控制块中的 msg_queue_head 和 msg_queue_tail；有些消息框可能是空的，它们通过 msg_queue_free 形成一个空闲消息框链表。所有消息队列中的消息框总数即是消息队列的长度，这个长度可在消息队列创建时指定。

  • 消息队列控制块

在 RT-Thread 中，消息队列控制块是操作系统用于管理消息队列的一个数据结构，由结构体 struct rt_messagequeue 表示。另外一种 C 表达方式 rt_mq_t，表示的是消息队列的句柄，在 C 语言中的实现是消息队列控制块的指针。消息队列控制块结构的详细定义请见以下代码：

rt-thread-4.0.1\include\rtdef.h
/**
 * message queue structure
 */
struct rt_messagequeue
{
    struct rt_ipc_object parent;                        /**< inherit from ipc_object */

    void                *msg_pool;                      /**< start address of message queue */

    rt_uint16_t          msg_size;                      /**< message size of each message */
    rt_uint16_t          max_msgs;                      /**< max number of messages */

    rt_uint16_t          entry;                         /**< index of messages in the queue */

    void                *msg_queue_head;                /**< list head */
    void                *msg_queue_tail;                /**< list tail */
    void                *msg_queue_free;                /**< pointer indicated the free node of queue */
};
typedef struct rt_messagequeue *rt_mq_t;

struct rt_mq_message
{
    struct rt_mq_message *next;
};

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25

rt_messagequeue 对象从 rt_ipc_object 中派生，rt_messagequeue.parent.parent.type值为RT_Object_Class_MessageQueue，rt_messagequeue.parent.parent.list为消息队列对象链表节点，所有消息队列组织成一个双向链表。

rt_messagequeue 对象的私有成员也不少，rt_messagequeue.msg_pool指向消息队列缓冲区的开始地址，rt_messagequeue.msg_size表示每个消息的长度（消息队列中消息的长度不固定，不像邮箱中邮件的长度固定），rt_messagequeue.max_msgs表示该消息队列最大能够容纳的消息数量，rt_messagequeue. entry表示队列中已有的消息数量。

rt_mq_message是消息队列中的消息头，队列中的消息靠消息头组织成一个单向链表。邮箱中邮件长度固定，所以邮箱中的邮件可以以数组形式管理，消息队列中的消息长度不固定，所以每个消息需要一个消息头组织成链表方便管理。rt_messagequeue. msg_queue_head消息链表头与rt_messagequeue.msg_queue_tail消息链表尾则作为消息链表的首尾标识，方便遍历链表以免越界访问。初始化后未使用的空白消息也被组织成链表的形式便于管理，rt_messagequeue.msg_queue_free便是这些空白消息链表的首节点。

  • 消息队列接口函数

消息队列控制块是一个结构体，其中含有消息队列相关的重要参数，在消息队列的功能实现中起重要的作用。消息队列的相关接口如下图所示，对一个消息队列的操作包含：创建消息队列 - 发送消息 - 接收消息 - 删除消息队列。
消息队列接口函数
首先看消息队列的构造、析构函数原型：

// rt-thread-4.0.1\src\ipc.c
/**
 * This function will initialize a message queue and put it under control of
 * resource management.
 *
 * @param mq the message object
 * @param name the name of message queue
 * @param msgpool the beginning address of buffer to save messages
 * @param msg_size the maximum size of message
 * @param pool_size the size of buffer to save messages
 * @param flag the flag of message queue
 *
 * @return the operation status, RT_EOK on successful
 */
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);

/**
 * This function will detach a message queue object from resource management
 *
 * @param mq the message queue object
 *
 * @return the operation status, RT_EOK on successful
 */
rt_err_t rt_mq_detach(rt_mq_t mq);

/**
 * This function will create a message queue object from system resource
 *
 * @param name the name of message queue
 * @param msg_size the size of message
 * @param max_msgs the maximum number of message in queue
 * @param flag the flag of message queue
 *
 * @return the created message queue, RT_NULL on error happen
 */
rt_mq_t rt_mq_create(const char *name,
                     rt_size_t   msg_size,
                     rt_size_t   max_msgs,
                     rt_uint8_t  flag);

/**
 * This function will delete a message queue object and release the memory
 *
 * @param mq the message queue object
 *
 * @return the error code
 */
rt_err_t rt_mq_delete(rt_mq_t mq);

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53

消息队列中每个消息的占用空间除了rt_messagequeue.msg_size还包括消息头占用的空间sizeof(rt_mq_message)。所以创建消息队列静态对象时根据传入的参数，计算得到消息数目mq->max_msgs = pool_size / (mq->msg_size + sizeof(struct rt_mq_message))，创建消息队列动态对象时根据传入的参数，计算得到要分配的空间mq->msg_pool = RT_KERNEL_MALLOC((mq->msg_size + sizeof(struct rt_mq_message)) * mq->max_msgs)。

接下来看消息队列的消息发送rt_mq_send和接收rt_mq_recv函数原型，由于消息队列都是先入先出FIFO的，有时候传递的消息比较紧急，需要尽快传送过去以保证实时性，消息队列针对这种紧急消息提供了后进先出LIFO的处理，即将紧急消息插入队列头部，下次接收消息时直接从队列头部取出放入的紧急消息，发送紧急消息的函数rt_mq_urgent，这些函数的原型如下：

// rt-thread-4.0.1\src\ipc.c
/**
 * This function will send a message to message queue object, if there are
 * threads suspended on message queue object, it will be waked up.
 *
 * @param mq the message queue object
 * @param buffer the message
 * @param size the size of buffer
 *
 * @return the error code
 */
rt_err_t rt_mq_send(rt_mq_t mq, void *buffer, rt_size_t size);

/**
 * This function will send an urgent message to message queue object, which
 * means the message will be inserted to the head of message queue. If there
 * are threads suspended on message queue object, it will be waked up.
 *
 * @param mq the message queue object
 * @param buffer the message
 * @param size the size of buffer
 *
 * @return the error code
 */
rt_err_t rt_mq_urgent(rt_mq_t mq, void *buffer, rt_size_t size);

/**
 * This function will receive a message from message queue object, if there is
 * no message in message queue object, the thread shall wait for a specified
 * time.
 *
 * @param mq the message queue object
 * @param buffer the received message will be saved in
 * @param size the size of buffer
 * @param timeout the waiting time
 *
 * @return the error code
 */
rt_err_t rt_mq_recv(rt_mq_t    mq,
                    void      *buffer,
                    rt_size_t  size,
                    rt_int32_t timeout);

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42

消息队列控制函数rt_mq_control提供了消息队列重置命令，实际上就是将挂起线程唤醒，将队列内的消息清空后重新组织到空白链表中，函数原型如下：

// rt-thread-4.0.1\src\ipc.c
/**
 * This function can get or set some extra attributions of a message queue
 * object.
 *  * @param mq the message queue object
 * @param cmd the execution command
 * @param arg the execution argument
 *  * @return the error code
 */
rt_err_t rt_mq_control(rt_mq_t mq, int cmd, void *arg);

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10

消息队列可以应用于发送不定长消息的场合，包括线程与线程间的消息交换，以及中断服务例程中给线程发送消息（中断服务例程不能接收消息）。下面分发送消息和同步消息两部分来介绍消息队列的使用。

  • 发送消息：消息队列和邮箱的明显不同是消息的长度并不限定在 4 个字节以内；另外，消息队列也包括了一个发送紧急消息的函数接口。但是当创建的是一个所有消息的最大长度是 4 字节的消息队列时，消息队列对象将蜕化成邮箱。消息队列是直接的数据内容复制，相当于邮箱加内存池的作用，邮箱如果想传递超过4字节的消息还需要另外申请并维护内存池空间，消息队列则可以免去用户额外为消息动态分配内存的烦恼，也不用担心后续的消息内存空间释放问题；
  • 同步消息：在一般的系统设计中会经常遇到要发送同步消息的问题，这个时候就可以根据当时状态的不同选择相应的实现：两个线程间可以采用[消息队列 + 信号量或邮箱]的形式实现。发送线程通过消息发送的形式发送相应的消息给消息队列，发送完毕后希望获得接收线程的收到确认，工作示意图如下图所示： 消息队列同步机制
    邮箱作为确认标志，代表着接收线程能够通知一些状态值给发送线程；而信号量作为确认标志只能够单一的通知发送线程，消息已经确认接收。

3.3 信号对象管理

信号（又称为软中断信号），在软件层次上是对中断机制的一种模拟，在原理上，一个线程收到一个信号与处理器收到一个中断请求可以说是类似的。

信号在 RT-Thread 中用作异步通信，POSIX 标准定义了 sigset_t 类型来定义一个信号集，然而 sigset_t 类型在不同的系统可能有不同的定义方式，在 RT-Thread 中，将 sigset_t 定义成了 unsigned long 型，并命名为 rt_sigset_t，应用程序能够使用的信号为 SIGUSR1和 SIGUSR2。

信号本质是软中断，用来通知线程发生了异步事件，用做线程之间的异常通知、应急处理。一个线程不必通过任何操作来等待信号的到达，事实上，线程也不知道信号到底什么时候到达，线程之间可以互相通过调用 rt_thread_kill() 发送软中断信号。

收到信号的线程对各种信号有不同的处理方法，处理方法可以分为三类：

  1. 类似中断的处理程序，对于需要处理的信号，线程可以指定处理函数，由该函数来处理；
  2. 忽略某个信号，对该信号不做任何处理，就像未发生过一样；
  3. 对该信号的处理保留系统的默认值。

如下图所示，假设线程 1 需要对信号进行处理，首先线程 1 安装一个信号并解除阻塞，并在安装的同时设定了对信号的异常处理方式；然后其他线程可以给线程 1 发送信号，触发线程 1 对该信号的处理。
信号处理过程
当信号被传递给线程 1 时，如果它正处于挂起状态，那会把状态改为就绪状态去处理对应的信号。如果它正处于运行状态，那么会在它当前的线程栈基础上建立新栈帧空间去处理对应的信号，需要注意的是使用的线程栈大小也会相应增加。

  • 信号数据结构

信号数据结构跟前面的五种内核对象并不类似，没有继承自IPC对象，也没有对应的控制块管理，主要是靠线程对象rt_thread内的几个成员来实现信号管理的目的。下面先看看rt_thread中跟信号相关的成员有哪些：

// rt-thread-4.0.1\include\rtdef.h
/**
 * @addtogroup Signal
 */
#ifdef RT_USING_SIGNALS
#include 
typedef unsigned long rt_sigset_t;
typedef void (*rt_sighandler_t)(int signo);
typedef siginfo_t rt_siginfo_t;

#define RT_SIG_MAX          32
#endif

/**
 * Thread structure
 */
struct rt_thread
{
	......
#if defined(RT_USING_SIGNALS)
    rt_sigset_t     sig_pending;                        /**< the pending signals */
    rt_sigset_t     sig_mask;                           /**< the mask bits of signal */

#ifndef RT_USING_SMP
    void            *sig_ret;                           /**< the return stack pointer from signal */
#endif
    rt_sighandler_t *sig_vectors;                       /**< vectors of signal handler */
    void            *si_list;                           /**< the signal infor list */
#endif
	......
};
typedef struct rt_thread *rt_thread_t;

#define RT_THREAD_STAT_SIGNAL           0x10                /**< task hold signals */
#define RT_THREAD_STAT_SIGNAL_READY     (RT_THREAD_STAT_SIGNAL | RT_THREAD_READY)
#define RT_THREAD_STAT_SIGNAL_WAIT      0x20                /**< task is waiting for signals */
#define RT_THREAD_STAT_SIGNAL_PENDING   0x40                /**< signals is held and it has not been procressed */
#define RT_THREAD_STAT_SIGNAL_MASK      0xf0

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38

对于非SMP对称多核处理器来说，rt_thread被条件宏RT_USING_SIGNALS包含的成员共四个：rt_thread.sig_pending表示悬起的信号，可参考中断中介绍的PendSV来理解；
rt_thread.sig_mask表示信号的屏蔽位，可参考中断屏蔽寄存器组（PRIMASK, FAULTMASK, BASEPRI）来理解；rt_thread.sig_vectors保存信号/软中断处理程序的地址，可参考中断向量表来理解；rt_thread.si_list是一个信号信息链表节点，保存信号信息结构体地址，一个信号都有哪些信息呢，看下面信号信息结构体的源码：

// rt-thread-4.0.1\include\libc\libc_signal.h
struct siginfo
{
    rt_uint16_t si_signo;
    rt_uint16_t si_code;

    union sigval si_value;
};
typedef struct siginfo siginfo_t;

union sigval 
{
    int    sival_int;    /* Integer signal value */
    void  *sival_ptr;    /* Pointer signal value */
};

#define SI_USER     0x01    /* Signal sent by kill(). */
#define SI_QUEUE    0x02    /* Signal sent by sigqueue(). */
#define SI_TIMER    0x03    /* Signal generated by expiration of a 
                               timer set by timer_settime(). */
......
/* #define SIGUSR1     25 */
/* #define SIGUSR2     26 */
#define SIGRTMIN    27
#define SIGRTMAX    31
#define NSIG        32

// rt-thread-4.0.1\src\signal.c

struct siginfo_node
{
    siginfo_t si;
    struct rt_slist_node list;
};



  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36

信号信息siginfo有三个成员：siginfo.si_signo为信号编号，可以类比中断号来理解，用户应用程序能够使用的信号为SIGUSR1与SIGUSR2（对于ARMGCC / ARMCLANG编译器，这两个信号编号被注释了，不清楚RT-Thread开发者出于什么考虑）；siginfo.si_code为信号的编码类型，用户应用程序能使用的编码类型为SI_USER；siginfo.si_value为信号值，可以是整型或指针型，根据需要决定是否使用。

为了将信号信息siginfo组织成链表，又引入了信号信息节点结构体siginfo_node，该结构体将siginfo组织成一个单向链表，前面介绍的线程对象成员rt_thread.si_list正是指向信号链表节点siginfo_node的指针。

线程对象如果使用信号的话，rt_thread.stat又增加了四种状态辅助进行信号管理：RT_THREAD_STAT_SIGNAL表示线程持有信号；RT_THREAD_STAT_SIGNAL_READY表示线程持有信号且处于就绪状态；RT_THREAD_STAT_SIGNAL_WAIT表示线程处于等待信号状态；
RT_THREAD_STAT_SIGNAL_PENDING表示线程持有信号处于悬起暂未处理状态。

  • 信号对象接口函数

由于信号数据结构并没有继承自rt_ipc_object或rt_object，接口函数跟前面五个对象也不类似，首先看信号的系统初始化函数：

// rt-thread-4.0.1\src\signal.c

int rt_system_signal_init(void)
{
    _rt_siginfo_pool = rt_mp_create("signal", RT_SIG_INFO_MAX, sizeof(struct siginfo_node));
    if (_rt_siginfo_pool == RT_NULL)
    {
        LOG_E("create memory pool for signal info failed.");
        RT_ASSERT(0);
    }
    return 0;
}

void rt_thread_alloc_sig(rt_thread_t tid);
void rt_thread_free_sig(rt_thread_t tid);
void rt_thread_handle_sig(rt_bool_t clean_state);

// rt-thread-4.0.1\src\components.c

int rtthread_startup(void)
{
	......
#ifdef RT_USING_SIGNALS
    /* signal system initialization */
    rt_system_signal_init();
#endif
	......
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28

信号的系统初始化函数rt_system_signal_init在系统启动函数rtthread_startup内被调用，且在初始化函数中完成了信号内存池的创建。信号的分配、释放、处理由其它系统接口函数调用，并不需要我们调用，接下来看用户常用的信号操作接口函数（主要有以下几种：安装信号、阻塞信号、阻塞解除、信号发送、信号等待）：
信号操作接口函数
安装信号：如果线程要处理某一信号，那么就要在线程中安装该信号。安装信号主要用来确定信号编号及线程针对该信号编号的动作（信号处理程序）之间的映射关系，即线程将要处理哪个信号，该信号被传递给线程时，将执行何种操作，在该函数中也会调用前面的信号分配函数rt_thread_alloc_sig。该函数的原型如下：

// rt-thread-4.0.1\src\signal.c
/**
 * This function can establish the mapping between signal number and signal handler.
 * 
 * @param signo: signal number, Only SIGUSR1 and SIGUSR2 are open to the user.
 * @param handler: signal handler, Three processing methods: user-defined processing function,  
 * 		SIG_IGN(ignore a signal),SIG_DFL(call _signal_default_handler());
 * 
 * @return The handler value before the signal is installed or SIG_ERR.
 */
rt_sighandler_t rt_signal_install(int signo, rt_sighandler_t handler);

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11

屏蔽信号/解除信号屏蔽：类似中断的屏蔽和解除屏蔽，软件模拟中断的信号也有屏蔽和解除屏蔽的接口函数，函数原型如下：

// rt-thread-4.0.1\src\signal.c
/**
 * This function can mask the specified signal.
 * @param signo: signal number, Only SIGUSR1 and SIGUSR2 are open to the user.
 */
void rt_signal_mask(int signo);

/**
 * This function can unmask the specified signal.
 * @param signo: signal number, Only SIGUSR1 and SIGUSR2 are open to the user.
 */
void rt_signal_unmask(int signo);

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12

发送信号/等待信号：当需要进行异常处理时，可以给设定了处理异常的线程发送信号，调用 rt_thread_kill() 可以用来向任何线程发送信号，rt_thread_kill最后会调用私有函数_signal_deliver(tid)，并由_signal_deliver调用rt_thread_handle_sig完成信号处理；调用rt_signal_wait等待 set 信号的到来，如果没有等到这个信号，则将线程挂起，直到等到这个信号或者等待时间超过指定的超时时间 timeout，如果等到了该信号，则将指向该信号体的指针存入 si，这两个函数的原型如下：

// rt-thread-4.0.1\src\signal.c
/**
 * This function can send a signal to the specified thread.
 * 
 * @param tid: thread receiving signal;
 * @param sig: signal number to be sent;
 * 
 * @return RT_EOK or -RT_EINVAL.
 */
int rt_thread_kill(rt_thread_t tid, int sig);

/**
 * This function will wait for a signal set.
 * 
 * @param set: specify the set of signals to wait for;
 * @param si: pointer to store waiting for signal information;
 * @param timeout: specified waiting time;
 * 
 * @return RT_EOK or -RT_ETIMEOUT or -RT_EINVAL.
 */
int rt_signal_wait(const rt_sigset_t *set, rt_siginfo_t *si, rt_int32_t timeout);

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21

四、IPC对象管理示例

我们拿比较经典的生产者消费者模型来示例，首先创建生产者、消费者两个线程，生产者线程与消费者线程的数据资源数量可以用信号量计数，当信号量资源减至0时再次尝试获取信号量资源则线程挂起等待，也即如果生产者生产的数据被消费完则消费者挂起等待新的数据被生产出来。但一般储存数据资源的空间有限，一个信号量可以告诉消费者资源已空来限制数据资源消费速度，但难以控制生产者线程的数据资源生产速度，因此还需要第二个信号量告诉生产者资源空间已满来限制数据资源生产速度。

既然两个信号量都用于资源计数，对共享数据资源的互斥访问也需要保护，可以使用互斥量或二值信号量来实现对共享资源的互斥访问，这里使用互斥量来实现。

在projects\stm32l475_kernel_sample\applications目录下新建文件ipc1_sample.c，并在ipc1_sample.c内编辑还示例程序代码如下：

// projects\stm32l475_kernel_sample\applications\ipc1_sample.c

#include 

#define THREAD_PRIORITY       6
#define THREAD_STACK_SIZE     512
#define THREAD_TIMESLICE      5

/* 定义最大 5 个元素能够被产生 */
#define MAXSEM 5

/* 用于放置生产的整数数组 */
static rt_uint32_t array[MAXSEM];

/* 指向生产者、消费者在 array 数组中的读写位置 */
static rt_uint32_t set, get;

/* 指向线程控制块的指针 */
static rt_thread_t producer_tid = RT_NULL;
static rt_thread_t consumer_tid = RT_NULL;

static struct rt_mutex      mutex_lock;
static struct rt_semaphore  sem_empty;
static rt_sem_t             sem_full;

/* 生产者线程入口 */
static void producer_thread_entry(void *parameter)
{
    int cnt = 0;

    /* 运行 10 次 */
    while (cnt < 10)
    {
        /* 获取一个空位 */
        rt_sem_take(&sem_empty, RT_WAITING_FOREVER);

        /* 修改 array 内容，上锁 */
        rt_mutex_take(&mutex_lock, RT_WAITING_FOREVER);
        array[set % MAXSEM] = cnt + 1;
        rt_kprintf("the producer generates a number: %d\n", array[set % MAXSEM]);
        set++;
        rt_mutex_release(&mutex_lock);

        /* 发布一个满位 */
        rt_sem_release(sem_full);
        cnt++;

        /* 暂停一段时间 */
        rt_thread_mdelay(20);
    }

    rt_kprintf("the producer exit!\n");
}

/* 消费者线程入口 */
static void consumer_thread_entry(void *parameter)
{
    rt_uint32_t sum = 0;

    while (1)
    {
        /* 获取一个满位 */
        rt_sem_take(sem_full, RT_WAITING_FOREVER);

        /* 临界区，上锁进行操作 */
        rt_mutex_take(&mutex_lock, RT_WAITING_FOREVER);
        sum += array[get % MAXSEM];
        rt_kprintf("the consumer[%d] get a number: %d\n", (get % MAXSEM), array[get % MAXSEM]);
        get++;
        rt_mutex_release(&mutex_lock);

        /* 释放一个空位 */
        rt_sem_release(&sem_empty);

        /* 生产者生产到 10 个数目，停止，消费者线程相应停止 */
        if (get == 10)
            break;

        /* 暂停一小会时间 */
        rt_thread_mdelay(50);
    }

    rt_kprintf("the consumer sum is: %d\n", sum);
    rt_kprintf("the consumer exit!\n");
}

static int producer_consumer(void)
{
    set = 0;
    get = 0;

    /* 初始化 1个互斥量 */
    rt_mutex_init(&mutex_lock, "lock",         RT_IPC_FLAG_FIFO);
    /* 初始化 1 个信号量 */
    rt_sem_init(&sem_empty, "empty",   MAXSEM, RT_IPC_FLAG_FIFO);
    /* 创建   1 个信号量 */
    sem_full = rt_sem_create("full",     0,    RT_IPC_FLAG_FIFO);
    if(sem_full == RT_NULL){
        rt_kprintf("create full semaphore failed.\n");
        return -1;
    }

    /* 创建生产者线程 */
    producer_tid = rt_thread_create("producer",
                                    producer_thread_entry, RT_NULL,
                                    THREAD_STACK_SIZE,
                                    THREAD_PRIORITY - 1,
                                    THREAD_TIMESLICE);
    if (producer_tid != RT_NULL)
    {
        rt_thread_startup(producer_tid);
    }
    else
    {
        rt_kprintf("create thread producer failed.\n");
        return -1;
    }

    /* 创建消费者线程 */
    consumer_tid = rt_thread_create("consumer",
                                    consumer_thread_entry, RT_NULL,
                                    THREAD_STACK_SIZE,
                                    THREAD_PRIORITY + 1,
                                    THREAD_TIMESLICE);
    if (consumer_tid != RT_NULL)
    {
        rt_thread_startup(consumer_tid);
    }
    else
    {
        rt_kprintf("create thread consumer failed.\n");
        return -1;
    }

    return 0;
}

/* 导出到 msh 命令列表中 */
MSH_CMD_EXPORT(producer_consumer, producer_consumer sample);

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92
  • 93
  • 94
  • 95
  • 96
  • 97
  • 98
  • 99
  • 100
  • 101
  • 102
  • 103
  • 104
  • 105
  • 106
  • 107
  • 108
  • 109
  • 110
  • 111
  • 112
  • 113
  • 114
  • 115
  • 116
  • 117
  • 118
  • 119
  • 120
  • 121
  • 122
  • 123
  • 124
  • 125
  • 126
  • 127
  • 128
  • 129
  • 130
  • 131
  • 132
  • 133
  • 134
  • 135
  • 136
  • 137
  • 138
  • 139

在env内执行scons --target=mdk5编译生成Keil MDK V5工程，使用Keil MDK V5打开工程编译无错误，烧录到我们的STM32L475潘多拉开发板内，运行结果如下：
生产者消费者例程运行结果
最开始资源为空，消费者等待生产者生产数据资源，后面生产者的生产速度超过消费者的消费速度，当资源已满时生产者等待消费者消费资源。示例程序中创建的两个信号量和一个互斥量对象的详细信息也可以通过finsh组件提供的命令查看，对程序调试和观察工程运行情况还是挺方便的。

  • 多生产者-消费者模型扩展

前面的示例程序中使用数组来保存生产者与消费者的共享数据资源，还需要用户管理数组元素的出入口偏移get/set，同时还需要互斥量保护数组元素的互斥访问。前面介绍的邮箱和消息队列对象自带数据出入口管理功能，下面我们用邮箱和消息队列分别实现上面的生产者消费者模型。

使用邮箱对象传递整型数据，使用消息队列对象传递字符消息，也即有两个生产者分别生产整型数据和字符消息；消费者使用事件集逻辑与同时处理两个生产者生产的数据资源，若生产者线程关闭，则消费者也相应停止。

在projects\stm32l475_kernel_sample\applications目录下新建文件ipc2_sample.c，在文件ipc2_sample.c内编写该示例实现代码：

// projects\stm32l475_kernel_sample\applications\ipc2_sample.c

#include 

#define THREAD_PRIORITY       6
#define THREAD_STACK_SIZE     512
#define THREAD_TIMESLICE      5

/* 定义最大 5 个元素能够被产生 */
#define MAXNUM                5

/* 定义两个事件 */
#define EVENT_FLAG1 (1 << 1)
#define EVENT_FLAG2 (1 << 2)

/* 指向IPC控制块的指针 */
static rt_mailbox_t mb;
static rt_mq_t      mq;
static rt_event_t   event;

/* 指向线程控制块的指针 */
static rt_thread_t producer1_tid = RT_NULL;
static rt_thread_t producer2_tid = RT_NULL;
static rt_thread_t consumer_tid =  RT_NULL;

/* 生产者1线程入口 */
static void producer1_thread_entry(void *parameter)
{
    int cnt = 0;

    /* 运行 8 次 */
    while (cnt < 8)
    {
        /* 向邮箱发送 1 个数据 */
        if(rt_mb_send_wait(mb, cnt, RT_WAITING_FOREVER) == RT_EOK)
        {           
            rt_kprintf("the producer1 generates a number: %d\n", cnt);
            /* 向事件集发送事件1 */
            rt_event_send(event, EVENT_FLAG1);

            cnt++;
        }
        /* 暂停一段时间 */
        rt_thread_mdelay(20);
    }
    rt_kprintf("the producer1 exit!\n");
}

/* 生产者2线程入口 */
static void producer2_thread_entry(void *parameter)
{
    char buf = 'A';

    /* 运行 8 次 */
    while (buf < 'I')
    {
        /* 向消息队列发送 1 个数据 */
        if(rt_mq_send(mq, &buf, sizeof(buf)) == RT_EOK)
        {           
            rt_kprintf("the producer2 generates a message: %c\n", buf);
            /* 向事件集发送事件2 */
            rt_event_send(event, EVENT_FLAG2);

            buf++;
        }
        /* 暂停一段时间 */
        rt_thread_mdelay(30);
    }
    rt_kprintf("the producer2 exit!\n");
}

/* 消费者线程入口 */
static void consumer_thread_entry(void *parameter)
{
    rt_uint32_t sum = 0;
    rt_uint8_t  str[9] = {0};
    rt_uint8_t  i = 0;
    rt_uint32_t cnt;
    rt_uint8_t  buf;
    rt_uint32_t e;

    while (1)
    {
        /* 接收事件1与事件2，事件组合逻辑与，接收后清除事件 */
        if(rt_event_recv(event, (EVENT_FLAG1 | EVENT_FLAG2), 
                        RT_EVENT_FLAG_AND | RT_EVENT_FLAG_CLEAR, 
                        5000, &e) == RT_EOK)
        {
            /* 从邮箱接收 1 个数据，并从消息队列接收一个消息 */
            if(rt_mb_recv(mb, (rt_ubase_t *)&cnt, RT_WAITING_FOREVER) == RT_EOK &&
                rt_mq_recv(mq, &buf, sizeof(buf), RT_WAITING_FOREVER) == RT_EOK)
            {
                sum += cnt;
                str[i++] = buf;
                rt_kprintf("the consumer get a number and a message: (%d, %c)\n", cnt, buf);
            }        
        }

        /* 如果生产者线程退出，消费者线程相应停止 */
        if (producer1_tid->stat == RT_THREAD_CLOSE || producer2_tid->stat == RT_THREAD_CLOSE)
        {
            str[i] = '\0';
            break;
        }          
        /* 暂停一小会时间 */
        rt_thread_mdelay(50);
    }
    /* 输出消费者数据和消息处理结果 */
    rt_kprintf("the consumer sum is: %d\n", sum);
    rt_kprintf("the consumer str is: %s\n", str);
    rt_kprintf("the consumer exit!\n");
}

static int producer_consumer_ex(void)
{
    /* 创建  1 个邮箱 */
    mb = rt_mb_create("mailbox", MAXNUM, RT_IPC_FLAG_FIFO);
    if(mb == RT_NULL){
        rt_kprintf("create mailbox failed.\n");
        return -1;
    }

    /* 创建  1 个消息队列 */
    mq = rt_mq_create("messagequeue",1 , MAXNUM, RT_IPC_FLAG_FIFO);
    if(mq == RT_NULL){
        rt_kprintf("create messagequeue failed.\n");
        return -1;
    }

    /* 创建  1 个事件集 */
    event = rt_event_create("event", RT_IPC_FLAG_FIFO);
    if(event == RT_NULL){
        rt_kprintf("create event failed.\n");
        return -1;
    }

    /* 创建生产者1线程 */
    producer1_tid = rt_thread_create("producer1",
                                    producer1_thread_entry, RT_NULL,
                                    THREAD_STACK_SIZE,
                                    THREAD_PRIORITY - 1,
                                    THREAD_TIMESLICE);
    if (producer1_tid != RT_NULL){
        rt_thread_startup(producer1_tid);
    }else{
        rt_kprintf("create thread producer1 failed.\n");
        return -1;
    }

    /* 创建生产者2线程 */
    producer2_tid = rt_thread_create("producer2",
                                    producer2_thread_entry, RT_NULL,
                                    THREAD_STACK_SIZE,
                                    THREAD_PRIORITY,
                                    THREAD_TIMESLICE);
    if (producer2_tid != RT_NULL){
        rt_thread_startup(producer2_tid);
    }else{
        rt_kprintf("create thread producer2 failed.\n");
        return -1;
    }

    /* 创建消费者线程 */
    consumer_tid = rt_thread_create("consumer",
                                    consumer_thread_entry, RT_NULL,
                                    THREAD_STACK_SIZE,
                                    THREAD_PRIORITY + 1,
                                    THREAD_TIMESLICE);
    if (consumer_tid != RT_NULL){
        rt_thread_startup(consumer_tid);
    }else{
        rt_kprintf("create thread consumer failed.\n");
        return -1;
    }

    return 0;
}

/* 导出到 msh 命令列表中 */
MSH_CMD_EXPORT_ALIAS(producer_consumer_ex, pro_con_ex, producer_consumer_ex sample);

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92
  • 93
  • 94
  • 95
  • 96
  • 97
  • 98
  • 99
  • 100
  • 101
  • 102
  • 103
  • 104
  • 105
  • 106
  • 107
  • 108
  • 109
  • 110
  • 111
  • 112
  • 113
  • 114
  • 115
  • 116
  • 117
  • 118
  • 119
  • 120
  • 121
  • 122
  • 123
  • 124
  • 125
  • 126
  • 127
  • 128
  • 129
  • 130
  • 131
  • 132
  • 133
  • 134
  • 135
  • 136
  • 137
  • 138
  • 139
  • 140
  • 141
  • 142
  • 143
  • 144
  • 145
  • 146
  • 147
  • 148
  • 149
  • 150
  • 151
  • 152
  • 153
  • 154
  • 155
  • 156
  • 157
  • 158
  • 159
  • 160
  • 161
  • 162
  • 163
  • 164
  • 165
  • 166
  • 167
  • 168
  • 169
  • 170
  • 171
  • 172
  • 173
  • 174
  • 175
  • 176
  • 177
  • 178
  • 179
  • 180

在env内执行scons --target=mdk5编译生成Keil MDK V5工程，使用Keil MDK V5打开工程编译无错误，烧录到我们的STM32L475潘多拉开发板内，运行结果如下：
生产者消费者扩展例程运行结果
两个IPC对象管理示例程序下载地址：https://github.com/StreamAI/RT-Thread_Projects/tree/master/projects/stm32l475_kernel_sample

更多文章：

注：本文转载自blog.csdn.net的流云IoT的文章"https://blog.csdn.net/m0_37621078/article/details/101082972"。版权归原作者所有，此博客不拥有其著作权，亦不承担相应法律责任。如有侵权，请联系我们删除。
复制链接
相关推荐
发表评论

评论记录：

未查询到任何数据！