RT-Thread-RTT 源码分析笔记——互斥量篇RT-Thread问答社区

RTT 源码分析笔记——互斥量篇

发布于 2025-02-07 18:26:07 浏览：90 订阅该版

[tocm]

# 核心需求

在文档中心也提到过，互斥量不仅仅是单纯的二值信号量还需要解决优先级反转问题。

信号量代表着资源的多少，而互斥量代表着资源的所有权，

持有者可以递归请求。

互斥量只能由持有者释放。
除此之外实时性也必须保障。

# 最小可用接口集

RT thread版本：9afe6a51826fb84de229de220dab2077aa5734b2

![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20250207/915f92b37958f17855027ce07aded68e.png)

# 数据结构与其相关的不变式

```C++
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;
```

从IPC对象继承，而IPC对象就只是多了一个挂起队列（按优先级从高到低排序）的头结点（对头）。

```C++
struct rt_ipc_object
{
    struct rt_object parent;                            /**< inherit from rt_object */

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

`owner`不变式：当前持有线程的索引。

（不变式，invariant，意思是变量的意义）。
该状态主要用于当更高优先级的线程请求互斥量时，操作系统能根据请求的互斥量访问到持有线程状态，进而修改它的优先级。

`taken_list`，这里不要与父类ipc的挂起队列混淆，因为线程也可以持有多个锁，所以线程对象也会有队头节点，锁对象作为元素自然也需要个节点来被串起来。

一个线程可以持有多个锁，一个锁也可阻塞多个线程，在接下来的优先级继承部分会重点考虑这一点。

`hold`不变式：请求互斥量的次数。

`priority`不变式：等待该互斥量的线程中最高优先级线程的优先级。

这个状态不是非必要的，如果不考虑时间，每次更新优先级都可以先遍历一遍挂起队列。
实际上，互斥量模块的一大部分代码都是为了保持`priority`的不变式。比如辅助函数：`。_check_and_update_prio`。

由于`ceiling_priority`需要使用`rt_mutex_setprioceiling`接口后才有意义，在最小可用接口集里无关。
`spinlock`，在多线程程序中保护互斥量成员的状态。

互斥量模块还会影响持有线程的互斥量队列，等待对象。

```C++
#ifdef RT_USING_MUTEX
    /* object for IPC */
    rt_list_t                   taken_object_list;//? 线程拥有的资源？
    rt_object_t                 pending_object;
#endif /* RT_USING_MUTEX */
```

`taken_object_list`，队列队头，元素为已占有的IPC对象。

`pending_object`，等待IPC对象的索引。

不考虑其他IPC对象，只考虑互斥量。

线程的实时优先级等同于当前占有的互斥量中最高优先级的互斥量

锁的实时优先级等同于当前挂起的线程中最高优先级的线程。

所以在优先级修改的过程中会访问到`taken_object_list`和`pending_object`。

# 实现思路

## create

分配给的空间未初始化。需要初始化互斥量成员。

```C++
t_mutex_t rt_mutex_create(const char *name, rt_uint8_t flag)
{
    struct rt_mutex *mutex;

/* flag parameter has been obsoleted */
    RT_UNUSED(flag);

RT_DEBUG_NOT_IN_INTERRUPT;

/* allocate object */
    mutex = (rt_mutex_t)rt_object_allocate(RT_Object_Class_Mutex, name);
    if (mutex == RT_NULL)
        return mutex;

/* initialize ipc object */
    _ipc_object_init(&(mutex->parent));

mutex->owner    = RT_NULL;
    mutex->priority = 0xFF;
    mutex->hold     = 0;
    mutex->ceiling_priority = 0xFF;//?
    rt_list_init(&(mutex->taken_list));

/* flag can only be RT_IPC_FLAG_PRIO. RT_IPC_FLAG_FIFO cannot solve the unbounded priority inversion problem */
    mutex->parent.parent.flag = RT_IPC_FLAG_PRIO;
    rt_spin_lock_init(&(mutex->spinlock));

return mutex;
}
```

## take

`static rt_err_t _rt_mutex_take(rt_mutex_t mutex, rt_int32_t timeout, int suspend_flag)`

`thread = rt_thread_self();`

请求大概会有四种场景：

当前线程获取未被占用的互斥量。

持有者递归请求。

优先级比持有线程低的线程获取互斥量

优先级比持有线程高的线程获取互斥量

函数的开头自然需要`lock`互斥量，保持数据的一致性。

`rt_spin_lock(&(mutex->spinlock));`

当调用场景是获取未被占用的互斥量，则意味着互斥量被`thread`线程获取，及时更新到互斥量状态和线程的IPC对象队列上。

```C++
if (mutex->owner == RT_NULL)
        {
            /* set mutex owner and original priority */
            mutex->owner    = thread;
            mutex->priority = 0xff;//?
            mutex->hold     = 1;

if (mutex->ceiling_priority != 0xFF)//?
            {
                /* set the priority of thread to the ceiling priority */
                if (mutex->ceiling_priority < rt_sched_thread_get_curr_prio(mutex->owner))
                    _thread_update_priority(mutex->owner, mutex->ceiling_priority, suspend_flag);
                //?
            }

/* insert mutex to thread's taken object list */
            rt_list_insert_after(&thread->taken_object_list, &mutex->taken_list);
        }
```

当调用场景是持有者递归请求：

则意味着互斥量的持有次数加一，及时更新互斥量的`hold`即可。

```C++
    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`的挂起队列。
更新当前线程的`pending_object`和mutex的挂起队列。

最后调度。

```C++
                rt_sched_lock_level_t slvl;
                rt_uint8_t priority;

/* mutex is unavailable, push to suspend list */
                LOG_D("mutex_take: suspend thread: %s",
                      thread->parent.name);

/* suspend current thread */
                ret = rt_thread_suspend_to_list(thread, &(mutex->parent.suspend_thread),
                                                mutex->parent.parent.flag, suspend_flag);
                if (ret != RT_EOK)
                {
                    rt_spin_unlock(&(mutex->spinlock));
                    return ret;
                }

/* set pending object in thread to this mutex */
                thread->pending_object = &(mutex->parent.parent);//?

rt_sched_lock(&slvl);                
```

当调用场景是优先级比持有线程高的线程获取互斥量：

除了以上操作外，还需要保持`mutex`的`priority`不变式，提升`mutex`的持有线程的优先级。
优先级继承的逻辑由`_thread_update_priority`实现

最后调度。

```C++
/* update the priority level of mutex */
                if (priority < mutex->priority)
                {
                    mutex->priority = priority;
                    if (mutex->priority < rt_sched_thread_get_curr_prio(mutex->owner))
                    {
                        _thread_update_priority(mutex->owner, priority, RT_UNINTERRUPTIBLE); /* TODO */
                    }
                }
```

再注意，`priority`不变式是等待该互斥量的线程中最高优先级线程的优先级。
线程的实时优先级等同于当前占有的互斥量中最高优先级的互斥量。
所以`mutex`的`priority`先升高，持有线程的优先级再升高。

### 辅助函数：`_thread_update_priority`

```C++
rt_inline void _thread_update_priority(struct rt_thread *thread, rt_uint8_t priority, int suspend_flag)
{
    rt_err_t ret = -RT_ERROR;
    struct rt_object* pending_obj = RT_NULL;

LOG_D("thread:%s priority -> %d", thread->parent.name, priority);

/* change priority of the thread */
    ret = rt_sched_thread_change_priority(thread, priority);

while ((ret == RT_EOK) && rt_sched_thread_is_suspended(thread))//* 注意优先级传导问题
    {
        /* whether change the priority of taken mutex */
        pending_obj = thread->pending_object;

if (pending_obj && rt_object_get_type(pending_obj) == RT_Object_Class_Mutex)
        {
            rt_uint8_t mutex_priority = 0xff;
            struct rt_mutex* pending_mutex = (struct rt_mutex *)pending_obj;

/* re-insert thread to suspended thread list to resort priority list */
            rt_list_remove(&RT_THREAD_LIST_NODE(thread));

ret = rt_susp_list_enqueue(
                &(pending_mutex->parent.suspend_thread), thread,
                pending_mutex->parent.parent.flag);
            
            if (ret == RT_EOK)
            {
                /* update priority */
                _mutex_update_priority(pending_mutex);
                /* change the priority of mutex owner thread */
                LOG_D("mutex: %s priority -> %d", pending_mutex->parent.parent.name,
                        pending_mutex->priority);

mutex_priority = _thread_get_mutex_priority(pending_mutex->owner);
                if (mutex_priority != rt_sched_thread_get_curr_prio(pending_mutex->owner))
                {
                    thread = pending_mutex->owner;

ret = rt_sched_thread_change_priority(thread, mutex_priority);
                }
                else
                {
                    ret = -RT_ERROR;
                }
            }
        }
        else
        {
            ret = -RT_ERROR;
        }
    }
}
```

正常状态下直接赋给`thread`的优先级即可。

但是也有可能`thread`，即`mutex`的持有线程A也处于阻塞状态。而可能不只这一个线程处于阻塞，整个场景是一个请求链。

比如持有互斥量1的线程A死循环空等，持有互斥量2的线程B请求互斥量1，持有互斥量3的线程C请求互斥量2，持有互斥量4的线程D请求互斥量3。

![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20250207/ce4a10a31720a836958536dffd28045f.png.webp)

`while ((ret == RT_EOK) && rt_sched_thread_is_suspended(thread))：`这个循环块的内容则是将请求链上的每个互斥量优先级提升至请求线程的优先级（也就是线程D的优先级）。也就意味着所有的持有线程的优先级状态都需要更新。
具体部分有点绕：
代入一个具体场景更好理解。

当线程D请求互斥量3时，`_thread_update_priority`的`thread`为线程C，等待对象为锁2，线程B是锁2的等待对象。

那么会先提升锁2的优先级，再提升线程B的优先级。
此时`thread`索引是线程B，依次循环。

![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20250207/7c899a7763b846af6890dd493253ab9f.png.webp)

### 辅助函数 _mutex_update_priority，

```C++
rt_inline rt_uint8_t _mutex_update_priority(struct rt_mutex *mutex)
{
    struct rt_thread *thread;

if (!rt_list_isempty(&mutex->parent.suspend_thread))
    {
        thread = RT_THREAD_LIST_NODE_ENTRY(mutex->parent.suspend_thread.next);
        mutex->priority = rt_sched_thread_get_curr_prio(thread);
    }
    else
    {
        mutex->priority = 0xff;
    }

return mutex->priority;
}
```

通常在被`mutex`挂起的线程优先级发生改变后，调用此函数进行更新。

## release

调用场景：

持有者释放。

非持有者调用。

当持有者释放时：
若持有次数不为零，则更新互斥量的`hold`状态即可。

若为零，则做以下操作：
将互斥量转交给等待互斥量的第一个线程A（最优先）

降低当前线程的优先级。`_check_and_update_prio`实现
A线程应移到调度器的就绪队列，更新互斥量的状态（`hold，priority，owner`）和线程A控制块的状态（`pending_object,taken_objects`）。

```C++
if (mutex->hold == 0)
    {
        rt_sched_lock(&slvl);

/* remove mutex from thread's taken list */
        rt_list_remove(&mutex->taken_list);

/* whether change the thread priority */
        need_schedule = _check_and_update_prio(thread, mutex);
        //* 线程优先级改回去，很有可能优先级降低。需要调度到优先级更高的线程

/* wakeup suspended thread */
        if (!rt_list_isempty(&mutex->parent.suspend_thread))
        {
            struct rt_thread *next_thread;
            do
            {
                /* get the first suspended thread */
                next_thread = RT_THREAD_LIST_NODE_ENTRY(mutex->parent.suspend_thread.next);

RT_ASSERT(rt_sched_thread_is_suspended(next_thread));

/* remove the thread from the suspended list of mutex */
                rt_list_remove(&RT_THREAD_LIST_NODE(next_thread));

/* resume thread to ready queue */
                if (rt_sched_thread_ready(next_thread) != RT_EOK)
                {
                    /**
                     * a timeout timer had triggered while we try. So we skip
                     * this thread and try again.
                     */
                    //? 怎么会有这种情况？
                    next_thread = RT_NULL;
                }
            } while (!next_thread && !rt_list_isempty(&mutex->parent.suspend_thread));

if (next_thread)
            {
                LOG_D("mutex_release: resume thread: %s",
                    next_thread->parent.name);

/* set new owner and put mutex into taken list of thread */
                mutex->owner = next_thread;
                mutex->hold  = 1;
                rt_list_insert_after(&next_thread->taken_object_list, &mutex->taken_list);

/* cleanup pending object */
                next_thread->pending_object = RT_NULL;

/* update mutex priority */
                if (!rt_list_isempty(&(mutex->parent.suspend_thread)))
                {
                    struct rt_thread *th;

th = RT_THREAD_LIST_NODE_ENTRY(mutex->parent.suspend_thread.next);
                    mutex->priority = rt_sched_thread_get_curr_prio(th);
                }
                else
                {
                    mutex->priority = 0xff;
                }

need_schedule = RT_TRUE;
            }
            else
            {
                /* no waiting thread is woke up, clear owner */
                mutex->owner = RT_NULL;
                mutex->priority = 0xff;
            }

rt_sched_unlock(slvl);
        }
        else
        {
            rt_sched_unlock(slvl);

/* clear owner */
            mutex->owner    = RT_NULL;
            mutex->priority = 0xff;
        }
    }
```

出现更高优先级的线程时(当前优先级降低和新的线程移入就绪队列)，进行调度。

### 辅助函数：_check_and_update_prio

```C++
//* 因持有锁而被提升的优先级应该被降低，降低为thread持有的锁中最高优先级
static rt_bool_t _check_and_update_prio(rt_thread_t thread, rt_mutex_t mutex)//* owner_thread,mutex
{
    RT_SCHED_DEBUG_IS_LOCKED;
    rt_bool_t do_sched = RT_FALSE;

//? 这个判断条件不懂
    if ((mutex->ceiling_priority != 0xFF) || (rt_sched_thread_get_curr_prio(thread) == mutex->priority))
    {
        //* 
        rt_uint8_t priority = 0xff;

/* get the highest priority in the taken list of thread */
        priority = _thread_get_mutex_priority(thread);

rt_sched_thread_change_priority(thread, priority);

/**
         * notify a pending reschedule. Since scheduler is locked, we will not
         * really do a re-schedule at this point
         */
        do_sched = RT_TRUE;
    }
    return do_sched;
}
```

## delete

删除`mutex`之前要做的事：

`mutex`从持有线程的队列移出。

将`mutex`挂起的线程释放，移动到调度器的就绪队列。

释放占用的内存。

注意：非持有者居然也可以删除`mutex`？？

```C++
static void _mutex_before_delete_detach(rt_mutex_t mutex)
{
    rt_sched_lock_level_t slvl;
    rt_bool_t need_schedule = RT_FALSE;

rt_spin_lock(&(mutex->spinlock));
    /* wakeup all suspended threads */
    rt_susp_list_resume_all(&(mutex->parent.suspend_thread), RT_ERROR);

rt_sched_lock(&slvl);

/* remove mutex from thread's taken list */
    rt_list_remove(&mutex->taken_list);

/* whether change the thread priority */
    if (mutex->owner)
    {
        need_schedule = _check_and_update_prio(mutex->owner, mutex);
    }

if (need_schedule)
    {
        rt_sched_unlock_n_resched(slvl);
    }
    else
    {
        rt_sched_unlock(slvl);
    }