【RTOS】RT-Thread 进程间通信IPC源码级分析详解

本文会更加聚焦RTT下各种IPC机制的实现方法，各种IPC机制具体是怎么用参见
https://blog.csdn.net/weixin_45434953/article/details/146117181?spm=1011.2415.3001.5331

在开始，我们先看看诸多IPC的爹——rt_ipc_object父类

/*** 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 */
};

可以看到RT Thread的源码写得相当的面向对象——它使用了类似继承的机制，其中一个是OOP的根类Obejct类rt_object，然后剩下的一个是一个列表，表示的是等待这个IPC资源的threads 组成的队列。

信号量rt_semaphore

首先是信号量机制，我们知道信号量是一个值，可以进行P操作和V操作来增加或减少它。信号量的两大核心用途是实现互斥（防止多个执行单元同时访问共享资源）和实现同步（协调多个执行单元的执行顺序）。

struct rt_semaphore
{struct rt_ipc_object parent;                        /**< inherit from ipc_object */rt_uint16_t          value;                         /**< value of semaphore. */rt_uint16_t          max_value;struct rt_spinlock   spinlock;
};
typedef struct rt_semaphore *rt_sem_t;

• value：rt_uint16_t 类型，代表信号量的当前值。信号量的值表示可用资源的数量，线程获取资源时该值减 1，释放资源时该值加 1。当 value 为 0 时，请求资源的线程会被挂起。挂起的线程会被塞入parent父类的suspend_thread队列中
• max_value：rt_uint16_t 类型，是信号量的最大值。它限制了信号量值的上限，确保信号量的值不会超过这个最大值，避免资源过度分配。
• spinlock：rt_spinlock 类型，是一个自旋锁。自旋锁用于在多处理器环境下保护信号量的内部数据结构，确保对信号量操作（如获取、释放）的原子性，防止多个线程同时修改信号量的值导致数据不一致。对信号量的读取都非常快速，所以可以使用spinlock。

首先看获取信号量的过程，rtt使用_rt_sem_take函数来尝试获取信号量，相当于P操作，这里放上简化版代码

static rt_err_t _rt_sem_take(rt_sem_t sem, rt_int32_t timeout, int suspend_flag){level = rt_spin_lock_irqsave(&(sem->spinlock));	// 关闭中断并获取信号量自身的自旋锁，保证操作的原子性。// 若信号量的值大于 0，表示信号量可用，将信号量的值减 1，然后恢复中断并释放自旋锁。if (sem->value > 0){/* semaphore is available */sem->value --;rt_spin_unlock_irqrestore(&(sem->spinlock), level);}else{// 若 timeout 为 0，表示不等待，直接恢复中断并释放自旋锁，返回超时错误。if (timeout == 0){rt_spin_unlock_irqrestore(&(sem->spinlock), level);return -RT_ETIMEOUT;}else{// 获取当前线程，调用 rt_thread_suspend_to_list 函数将当前线程挂起，并把它添加到信号量的挂起线程列表中。thread = rt_thread_self();ret = rt_thread_suspend_to_list(thread, &(sem->parent.suspend_thread),sem->parent.parent.flag, suspend_flag);// 挂起失败则解除自旋锁，返回错误码if (ret != RT_EOK){rt_spin_unlock_irqrestore(&(sem->spinlock), level);return ret;}// 检查超时时间是否大于 0。若大于 0，意味着线程需要等待一段时间，而非永久等待。if (timeout > 0){// 调用该函数控制线程的定时器，设置定时器的超时时间，并且启动线程的定时器// 定时器开始计时。当计时达到 timeout 设定的时间，会触发相应的超时处理逻辑。rt_timer_control(&(thread->thread_timer),RT_TIMER_CTRL_SET_TIME,&timeout);rt_timer_start(&(thread->thread_timer));}// 恢复之前被关闭的中断，并释放自旋锁rt_spin_unlock_irqrestore(&(sem->spinlock), level);// 调用线程调度函数，重新选择一个合适的线程运行。由于当前线程已被挂起，调度器会从就绪队列中选择其他线程执行。rt_schedule();}}
}

然后就是释放信号量的V操作，RTT中使用rt_sem_release来释放一个信号量。下面是简化代码：

rt_err_t rt_sem_release(rt_sem_t sem){// 用于保存中断状态，在关闭和恢复中断时使用。rt_base_t level;rt_bool_t need_schedule = RT_FALSE; // 布尔类型变量，用于标记是否需要进行线程调度// 关闭中断并获取自旋锁，保证操作的原子性。level = rt_spin_lock_irqsave(&(sem->spinlock));// 如果当前信号量的等待列表非空，表示有进程在等待当前的信号量if (!rt_list_isempty(&sem->parent.suspend_thread)){// 调用 rt_susp_list_dequeue 函数从挂起列表中取出一个线程并使其就绪// 同时将 need_schedule 置为 RT_TRUE，表示需要进行线程调度rt_susp_list_dequeue(&(sem->parent.suspend_thread), RT_EOK);need_schedule = RT_TRUE;}else{/*若挂起线程列表为空，检查信号量的值是否小于最大值。若小于最大值，将信号量的值加 1；若等于最大值，恢复中断并释放自旋锁，返回 -RT_EFULL 表示信号量值溢出。*/if(sem->value < sem->max_value){sem->value ++; /* increase value */}else{rt_spin_unlock_irqrestore(&(sem->spinlock), level);return -RT_EFULL; /* value overflowed */}rt_spin_unlock_irqrestore(&(sem->spinlock), level); // 恢复之前被关闭的中断，并释放自旋锁。/* 若 need_schedule 为 RT_TRUE，表示有线程被唤醒，调用 rt_schedule 函数进行线程调度 */if (need_schedule == RT_TRUE)rt_schedule();return RT_EOK;}
}

自旋锁Spinlock

自旋锁是一种用于保护共享资源的同步机制，在多线程或多任务环境中，确保同一时间只有一个线程或任务能访问共享资源。

struct rt_spinlock
{
#ifdef RT_USING_DEBUGrt_uint32_t critical_level;
#endif /* RT_USING_DEBUG */rt_ubase_t lock;
};

• rt_uint32_t critical_level：该成员用于记录临界区的嵌套级别。在调试模式下，通过记录临界区的嵌套级别，开发者可以更好地跟踪和调试自旋锁的使用情况，判断是否存在死锁或锁竞争等问题
• rt_ubase_t lock：rt_ubase_t 是无符号的 CPU 相关数据类型，该成员用于表示自旋锁的状态。通常，0 表示锁处于解锁状态，非 0 表示锁处于锁定状态。线程或任务在尝试获取自旋锁时，会检查该成员的值，若为 0 则将其置为非 0 以获取锁，若为非 0 则会持续等待，直到锁被释放。

rt_spin_lock()

互斥锁mutex

/*** Mutual exclusion (mutex) structure*/
struct rt_mutex
{struct rt_ipc_object parent;                        /**< inherit from ipc_object */rt_uint8_t           ceiling_priority;              /**< the priority ceiling of mutexe */rt_uint8_t           priority;                      /**< the maximal priority for pending thread */rt_uint8_t           hold;                          /**< numbers of thread hold the mutex */rt_uint8_t           reserved;                      /**< reserved field */struct rt_thread    *owner;                         /**< current owner of mutex */rt_list_t            taken_list;                    /**< the object list taken by thread */struct rt_spinlock   spinlock;
};
typedef struct rt_mutex *rt_mutex_t;

• ceiling_priority：互斥锁的优先级天花板，用于解决优先级反转问题。优先级天花板是一个预先设定的优先级，当线程持有互斥锁时，其优先级会被提升到这个值。
• priority：等待该互斥锁的线程中的最高优先级。通过记录这个值，系统可以在互斥锁被释放时，优先唤醒优先级最高的线程。
• hold：记录当前持有该互斥锁的线程对互斥锁的持有次数。对于递归互斥锁，同一线程可以多次获取互斥锁，每次获取 hold 值加 1，释放时减 1。
• owner：指向当前持有该互斥锁的线程的指针。通过这个指针，系统可以知道哪个线程正在使用互斥锁，也能在释放锁时进行相关检查。
• taken_list：当前线程持有的所有 IPC 对象列表。线程可能会同时持有多个互斥锁或其他 IPC 对象，该列表用于记录这些对象。
• spinlock：自旋锁对象，用于保护互斥锁自身的数据结构。在对互斥锁的属性进行读写操作时，使用自旋锁保证操作的原子性，防止多线程同时访问导致数据不一致。

我们可以看到rtt显式著名Mutex不允许在中断上下文使用，因为这会导致线程睡眠。

rt_mutex_take()

照例，我们分析rt_mutex的关键函数，比如获取锁的rt_mutex_take()，这个函数真的又臭又长，我们聚焦关键部分：

	thread = rt_thread_self();if (mutex->owner == thread){if (mutex->hold < RT_MUTEX_HOLD_MAX){/* it's the same thread */mutex->hold ++;}else{rt_spin_unlock(&(mutex->spinlock));return -RT_EFULL; /* value overflowed */}}

这一段是实现递归mutex的关键代码，当发现是持有者再次申请持有mutex，会直接将hold +1，这一般会在递归中出现：在同一个线程中递归上锁n次的mutex也需要递归解锁n次

反之，如果mutex->owner == thread为False，表示是一个其他线程尝试take mutex。那么有两种可能：owner==NULL的时候表示没占用，反之则表示已有其他线程占用。当mutex无人占用的时候，会触发优先级天花板机制，详见源码

// 目前mutex没有被占用
if (mutex->owner == RT_NULL)
{mutex->owner    = thread;          /* 设置持有者 */mutex->priority = 0xff;            /* 重置优先级记录 */mutex->hold     = 1;               /* 初始化持有计数 *//* 应用天花板优先级, 如果mutex的天花板ceiling_priority比当前申请加锁的进程的优先级更高，则将当前进程的优先级提高到ceiling_priority*/if (mutex->ceiling_priority != 0xFF && mutex->ceiling_priority < rt_sched_thread_get_curr_prio(thread)){_thread_update_priority(thread, mutex->ceiling_priority, suspend_flag);}/* 将互斥锁加入线程的持有对象列表 */rt_list_insert_after(&thread->taken_object_list, &mutex->taken_list);
}

反之就是互斥锁已被占用，当前进程等待释放，等待释放主要有以下几步：

1. 用rt_thread_suspend_to_list将进程加入到mutex的阻塞列表，队列采用优先级排序
2. 优先级继承：当高优先级线程等待低优先级线程持有的互斥锁时，通过_thread_update_priority将持有线程优先级临时提升至当前线程优先级，防止优先级反转，这个地方有一个调度器锁，使用rt_sched_lock/unlock保护优先级更新的原子性
3. 初始化线程私有定时器：若操作在timeout时间内未完成（如未获取到互斥量），定时器会触发并唤醒线程，避免线程永久阻塞。这是处理死锁的一把好手，破坏“持有并等待”条件，系统也不会直接死锁掉不动，而是超时向上抛出错误，同时通过观察超时比例可以初步观察是否出现死锁
4. 当前进程已经完成挂起的一系列操作，释放spinlock，调用rt_schedule()让调度器选择就绪队列的进程进行调度。