kouu

Priority Inheritance，优先级继承，是解决优先级反转的一种办法。

一个经典的例子：A/B/C三个实时进程，优先级A>B>C。C持有a锁，而A等待a锁被挂起。原本C释放a锁之后，A进程就可以继续执行的，但是偏偏有个比C优先级高的B进程存在，导致C得不到运行，也就没法释放a锁，从而导致A进程一直挂起。从整体上看，进程B虽然比A优先级低，但它却成功的抢占掉了A。这就是所谓的优先级反转。

一种解决办法是优先级继承，C在持有a锁期间临时继承等待者A的优先级，那么B进程就无法从中捣乱了。

为了支持优先级继承，内核里面实现了一个叫rt_mutex的互斥锁。内核中的各个部分，包括本文要讨论的pi_futex，都可以使用rt_mutex来实现优先级继承。所以，要分析pi_futex，先得把rt_mutex弄清楚。

rt_mutex之所以叫"rt_"，是因为优先级继承主要是为实时进程服务的。我们知道，与普通进程"根据优先级瓜分CPU时间片"的理念不同，内核总是尽量满足最高优先级的实时进程的CPU需求，直到它挂起或退出。对于实时进程来说，优先级反转的后果尤为严重。比如上述例子，考虑在单核机器下，B进程"抢占"掉A进程之后，只要B进程一直处于RUNNING状态，C进程就得不到执行，A进程就得一直挂起。而如果是普通进程的话，B进程很快会用完时间片，就算"抢占"了A进程，那也只是一瞬间的事情。

要实现优先级继承，内核需要维护一条PI chain，保存进程持有锁和等待锁的依赖关系。PI chain其实是一个树型结构：一个task可能持有N个rt_mutex、每个rt_mutex只能被一个task持有。反过来，一个rt_mutex可能让M个task挂起、而每个task又只会挂起在一个rt_mutex上面。如：

rt_mutex_waiter结构是作为task在rt_mutex上挂起时的连接件而存在的，其实它逻辑上应该是task_struct的一部分，只不过这个部分只是在PI chain中才有用。rt_mutex_waiter.task和task_struct.pi_blocked_on相互指向。

task通过pi_waiters可以遍历到在它持有的每个rt_mutex上挂起的最高优先级的task（称作top waiter）。这个遍历逻辑都是由plist（有序链表）提供的，里面挂着的rt_mutex_waiter会以其所对应的task的优先级来排序。这样，一个task能继承到的优先级就是通过pi_waiters取到的第一个rt_mutex_waiter所对应的task（称作top pi waiter）的优先级。因为这里只需要关心最高优先级，所以每个rt_mutex只需要将自己的top waiter的提供上来就可以了，不必全部加进来。这个遍历关系是PI chain中最重要的一环，通过它来完成优先级继承。

rt_mutex则可以通过wait_list来遍历在它身上挂起的rt_mutex_waiter，从而遍历到挂起的task。这个遍历关系则是描述锁依赖关系的原始信息，为构建上一种遍历关系而存在。试想，当一个进程新持有一把锁时，rt_mutex.wait_list中的top waiter（rt_mutex_waiter）将加入到task_struct.pi_waiters链表中；当rt_mutex.wait_list中的进程优先级发生变化，导致top waiter改变时，需要将原来的top waiter从task_struct.pi_waiters里面移出，然后把新的top waiter加进去。为了这个过程的顺利进行，rt_mutex.wait_list也是一个按优先级排序的plist。

上述过程会导致task的pi_waiters发生变化，可能导致top pi waiter改变，从而重置进程优先级。

作为其中核心的rt_mutex，其实是一个很简单的结构，只有图中所列的owner和wait_list、外加一把spinlock。对它的上锁操作很简单，把owner指针指向持有者进程即可。而spinlock则用于保护lock/unlock过程，以及将进程挂入/移出PI chain的过程。

而如果不存在竞争，rt_mutex可以利用cmpxchg原子指令直接尝试lock/unlock：cmpxchg(rt_mutex.owner, NULL, current) / cmpxchg(rt_mutex.owner, current, NULL)。如果owner指针为NULL，原子性的修改为当前进程，则lock成功。否则说明已经被上锁，就需要上spinlock、然后修改PI chain；反过来，如果owner指针指向当前进程，原子性的修改为NULL，则unlock成功。否则说明锁有等待者（owner指针会打上flag，使得它不等于当前进程），需要上spinlock、修改PI chain、然后唤醒top waiter。

有了rt_mutex做支撑，pi_futex就可以登场了。三个耳熟能详的接口：

int futex_lock_pi(int *uaddr);

int futex_trylock_pi(int *uaddr);

int futex_unlock_pi(int *uaddr);

回想一下，futex的核心思想是什么（参阅《linux futex浅析》）？那就是用户态的锁变量与内核态futex机制的互动。pi_futex也不例外。

在锁没有竞争的情况下，用户态通过对锁变量的原子指令操作，依然可以完成pi_futex的上锁和解锁。但是锁变量的取值跟普通的mutex不太一样（普通mutex一般用0值表示未上锁、1是上锁、2是上锁且有进程挂起等待），pi_futex下，锁变量0值表示未上锁，上锁时写入owner进程的tid（全局的线程id），有进程挂起时在tid的基础上增加一个RT_MUTEX_HAS_WAITERS标记。

为什么要这样定呢？考虑这么一个场景，锁a处于未上锁状态(*uaddr==0)，然后进程A和B相继对其做lock操作。进程A在用户态就能lock成功了，而进程B则需要调用futex_lock_pi进入内核，然后被挂起。进程B挂起的时候，A和B应该会形成PI chain，但是A上锁的时候根本就没经过内核。换句话说，内核怎样得知进程A的信息，并构造出这条PI chain呢？就是通过记在*uaddr里面的tid。当然，这个tid是用户自己填的。如果用户乱填，是可以让自己的程序没法工作的:)（可能会让一个不相干的进程莫名其妙的成了rt_mutex的owner，或者futex_lock_pi找不到对应tid的进程，而返回失败。用户总是可以让自己的程序出问题的，不是么？）

有了A进程的tid，要构造PI chain还缺些什么呢？对啊，rt_mutex都还没创建出来。这就引出了pi_futex中的一个最重要的逻辑：uaddr跟rt_mutex的对应。

跟普通futex一样，uaddr在内核中会对应一个"等待队列"（其实是一个全局的队列）。挂起的进程在等待队列中对应一个futex_q结构：

struct futex_q {

    struct plist_node list;             // 链入等待队列

    struct task_struct *task;           // 挂起的进程本身

    spinlock_t *lock_ptr;               // 保存等待队列的锁，便于操作

    union futex_key key;                // 唯一标识uaddr的key值

    struct futex_pi_state *pi_state;    // 进程正在等待的锁

    struct rt_mutex_waiter *rt_waiter;  // 进程对应的rt_waiter

    union futex_key *requeue_pi_key;    // 等待被requeue的key

    u32 bitset;                         // futex_XXX_bitset时使用

};

其中的pi_state其实就是rt_mutex。由于存在于pi_futex体系中需要额外的关联信息，所以使用pi_state对rt_mutex做了包装：

struct futex_pi_state {

    struct list_head list;      // 链入持有者进程的list

    struct rt_mutex pi_mutex;   // 锁本身

    struct task_struct *owner;  // 锁的持有者进程

    atomic_t refcount;          // 引用计数

    union futex_key key;        // 唯一标识uaddr的key值

};

futex_pi_state中的key和pi_mutex就把uaddr跟rt_mutex的绑定了起来。

引入这两个结构之后，数据模型变成了这样：

这样一来，持有uaddr对应的锁的进程，就持有了内核中对应的rt_mutex，从而能够获得相应的优先级继承；而在uaddr对应的锁上挂起的进程，也就挂起到了相应的rt_mutex上，从而优先级被继承。

task通过pi_state_list可以遍历到它所持有的所有rt_mutex（via pi_state），在进程地址空间销毁（比如exit、exec）时需要释放这些rt_mutex（这个链表是关联在pi_state上的，如果不是使用pi_futex，而是内核中直接使用rt_mutex呢？就不存在这个自动销毁的逻辑了。不过rt_mutex只能内核代码才能直接使用，使用时注意管理好进程的生命周期就好了）。而既然是为销毁而准备的，这个链表只需要普通的list就足够了，不需要使用有序的plist。

这里面最难理解的地方还在于pi_state.owner和rt_mutex.owner。它们不是同一个东西么？为什么会有两个owner指针？难道一个锁能被多个进程持有？这个东西的存在主要是为了解决pi_futex执行过程中的一些中间状态，可能pi_state.owner指向某一进程，而rt_mutex.owner为空（未上锁）。此时，被pi_state.owner指向的进程被称作是pending owner。来看看具体的场景：

对于futex_unlock_pi来说，它不能像普通的unlock操作那样，将*uaddr置为0，然后唤醒一个等待进程就完事。当rt_mutex还有进程在挂起等待的时候，绝不能将*uaddr设为0，因为这将使得新的owner直接在用户态就lock成功，没法建立起与waiter们的继承关系（如果没有waiter，在用户态完成lock就无所谓，但是现在情况不一样）。

实现上，unlock会保持*uaddr的上锁状态，但是会将对应的rt_mutex unlock掉（否则被唤醒的进程不可能lock成功）。并且，将*uaddr改成继任者的tid+flags，pi_state.owner也已经指向继任者，这时的继任者成为pending owner。

另一方面，被唤醒的进程在离开futex_lock_pi之前（注意，它是在futex_lock_pi里面进入睡眠的，醒来的时候还在futex_lock_pi里面），futex_lock_pi需要为它获取到锁，让它真正成为锁的owner（获取到已经unlock的rt_mutex，并修改PI chain）。

当然，唤醒之后futex_lock_pi获取rt_mutex不一定能成功，可能有新来的进程参与竞争，或者之前等待队列中的进程被信号唤醒了，也来重新参与竞争。不过没关系，这将导致被唤醒的进程继续被挂起，并不会从futex_lock_pi返回。（如果pending owner醒来后锁被别的进程抢走了，称作lock stealing。）

且说这时可能会有多个进程同时竞争一个rt_mutex，既然PI chain已经保存了进程的优先级关系，竞争的结果也应该是让优先级最高的进程获得锁。futex_unlock_pi会选择唤醒rt_mutex下的top waiter进程，并将其置为pending owner。如果多个waiter进程优先级相同，则遵循先来后到的原则，先来的被认为是top waiter。于是，当进程futex_lock_pi时，能否获得rt_mutex，不是看谁下手快，而是看谁有这个资格：如果进程是futex_unlock_pi选中的pending owner，那么应该让它获得锁；如果进程是新来的，并且优先级比top waiter进程还高（高于。如果只是等于，则应该去排队），那么也应该让它获得锁（lock stealing）。如果多个进程都有资格，那只好看谁下手快了。然后被唤醒的进程并不会从PI chain中移出，因为这个进程被唤醒之后要么得不到锁（lock stealing），会继续等待；要么会获得锁，那么应该届时再调整PI chain。

为什么要搞这么复杂呢？简单一点：unlock时直接将锁释放、将被唤醒的进程移出PI chain；然后被唤醒时的futex_lock_pi直接返回，让用户态去重试。这样不行么？乍一看，虽然被唤醒的进程要被从PI chain移出移进，有些冗余，但是似乎还是可以工作的，只要每次唤醒的是最高优先级的进程，它重新去lock的时候肯定应该是能够获得锁的。不过考虑一下rt_mutex的前几个waiter拥有相同优先级的情况，被唤醒的进程如果脱离了PI chain，重新去lock，那它就成了新来的。按规矩，它应该遵循先来后到原则，排在其他相同优先级的waiter之后，而不能获得锁。如此一来，就没有人能获得锁了。所以pending owner这么一个中间状态是有必要的，以区别被唤醒的进程和新来的进程。

那么，futex_lock_pi如果不在返回之前获得锁行不行呢？返回用户态，等用户态去重新调用futex_lock_pi，这个看上去貌似也没问题。不过考虑一下futex_lock_pi返回用户态后，进程挂掉的情况。被唤醒的top waiter竟然挂了，再也不会去尝试持有锁了，队列里的waiters又该怎么办呢？（注意，如果futex_lock_pi先获得锁，返回用户态后再挂掉，内核会有回收机制，参见前面提到过的task_struct.pi_state_list。）（再注意，普通futex其实并没有这么强的保护。对于普通futex来说，内核的功能非常轻，仅仅是给用户态提供一个睡眠唤醒的机制。只要进程退出，futex等待队列中维护的对应节点自然会被清理。用户再怎么玩都无所谓。而pi_futex在内核里面则是非常重，内核需要保证功能的完整性。）

所以内核里面的条原则，不能让pi_futex在有waiter而没有owner的情况下离开内核态，否则可能就没人能成为owner了，一堆waiter将无人解救。

另外一个问题是关于futex_trylock_pi。既然是futex，trylock操作不是应该由用户态的原子操作来实现么？为什么还要有trylock系统调用？的确，pi_futex系列刚诞生的时候是没有futex_trylock_pi的。

用户态的trylock只能在未上锁的情况下才能成功，而futex_trylock_pi会进入内核，如果rt_mutex暂时没有owner（比如旧的持有者已经unlock了，被唤醒的pending owner还尚未lock），且当前进程优先级更高，则可以上锁成功（lock stealing）。

另一方面，进程有时候可能不知道自己有没有持有pi_futex（比如futex_lock_pi被信号打断时，在信号处理函数中拿不到futex_lock_pi的返回值，不知道是否上锁成功了）。这时可以调用futex_trylock_pi去检查（如果直接调用futex_lock_pi可能会死锁）。通过*uaddr==gettid()去检查可以么？因为*uaddr上会被内核打上其他标记，直接检查很可能通不过，而用户态又不应该去识别这些标记。

除了最普通的lock/trylock/unlock三个接口，pi_futex还有两个接口：

int futex_wait_requeue_pi(int *uaddr1, int val1, int *uaddr2);

int futex_cmp_requeue_pi(int *uaddr1, int n1, int *uaddr2, int n2, int val);

futex_cmp_requeue_pi是pi版本的futex_cmp_requeue，可以用于带pi的phread_cond_broadcast。有个问题，pi是针对mutex而言的，那么既然是phread_cond，跟pi又有什么关系呢？的确，phread_cond是没有优先级继承这回事情的，但是phread_cond需要跟mutex配合使用，而mutex则可能是带pi的。回想一下，phread_cond_broadcast为了避免"惊群"，会使用futex_cmp_requeue唤醒一个waiter，然后将其他waiter移动到mutex的等待队列上去。那么，如果mutex是带pi的，简单的将其移动到等待队列显然不够，需要调用futex_cmp_requeue_pi，完成PI chain的构建。

于是，随着futex_cmp_requeue_pi被调用，一个pthread_cond的waiter被唤醒，其他waiter被移动到rt_mutex的等待队列中，组建PI chain。然后被唤醒的进程会去调用futex_lock_pi尝试加锁……稍等，是不是有什么问题？这不是又跟futex_unlock_pi时的情形一样了么？不能够放任rt_mutex没有owner，而让被唤醒的进程自己去上锁的呀！

所以，futex_cmp_requeue_pi不能只是简单的将进程唤醒，还需要将其置为pending owner。另一方面，pthread_cond_wait的进程不能只是使用futex_wait去等待pthread_cond，那样的话没法保证返回用户态时已经获得pi_futex的锁。而等待pthread_cond的场景又跟futex_lock_pi不一样，于是futex_wait_requeue_pi就诞生了，它需要等待普通futex（futex_wait的前半部分），并且在被唤醒后获取pi_futex（futex_lock_pi的后半部分）。

futex_cmp_requeue_pi唤醒进程的做法跟futex_unlock_pi不太一样，后者原本是锁的owner，所以可以由它来指定继任者pending owner；而前者并不是owner，它所唤醒的进程不应该成为pending owner（owner都没说话呢）。所以futex_cmp_requeue_pi会调用rt_mutex的trylock操作，尝试为被唤醒的进程获得锁。如果成功，那么被唤醒的进程将直接晋升为owner；否则说明锁的owner还在，那就不用唤醒了，直接加入等待队列去。

另一方面，futex_wait_requeue_pi可能会面对两种情况，它可能被futex_cmp_requeue_pi直接唤醒；或者是被放入等待队列，再被随后的futex_unlock_pi唤醒。如前所述，第一种情况下被唤醒时进程已经是锁的owner；而第二种情况则还是pending owner，还需要为成为owner再努一把力。futex_wait_requeue_pi需要能handle这两种情况。

题外话，最近futex曝出了提权漏洞，代号CVE-2014-3153（见：https://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2014-3153）。

问题出在哪里呢？回顾一下之前的数据结构图，当一个进程要在rt_mutex上挂起时，加入等待队列，需要一个futex_q结构、加入PI chain，需要一个rt_mutex_waiter结构。由于这两个东西都只是在挂起的时候会用到，所以将它们定义在栈上。

在调用futex_wait_requeue_pi的时候，情况比较特殊。此时wait的是一个普通futex，而期望被requeue的地方是一个pi_futex。尽管当前wait的地方不涉及PI chain，不需要用到rt_mutex_waiter，但这个东西可能在被requeue的时候用到，所以进程挂起之前得一并定义好：

struct rt_mutex_waiter rt_waiter;

struct futex_q q = futex_q_init;

q.rt_waiter = &rt_waiter;

前面说过，futex_wait_requeue_pi可能有两个结局，要么成为rt_mutex的owner被直接唤醒，要么在rt_mutex上挂起等待。前一种情况下rt_waiter是不需要使用的，后者则不然。

为了区别直接被futex_cmp_requeue_pi唤醒，和挂起后再被futex_unlock_pi唤醒两种情况，futex_cmp_requeue_pi直接唤醒时会将q.rt_waiter置NULL，表示rt_waiter根本没有使用过，可以不用管了。而被futex_unlock_pi唤醒时，由于被唤醒的进程还只是pending owner，还挂在PI chain中，所以将q.rt_waiter保留，等回到futex_wait_requeue_pi真正成为owner以后，再将rt_waiter移出PI chain。

这时，骇客的破解程序构造了一种反常的场景，让调用futex_wait_requeue_pi的进程被唤醒时看到q.rt_waiter被置NULL，但rt_waiter却还挂在PI chain中（并非预想的那样没有被使用）。于是，futex_wait_requeue_pi看到q.rt_waiter为NULL，会认为rt_waiter并未使用，并不会去清理PI chain。等futex_wait_requeue_pi返回之后，PI chain里的rt_waiter就成了野指针（注意它原本是在栈上分配的）。然后，通过骇客对栈上数据的精心构造，可以让内核在处理PI chain的时候调用到栈上预埋下的破解代码……

骇客怎么在栈上预埋破解代码，那是另外一个story（有兴趣可以参阅：https://tinyhack.com/2014/07/07/exploiting-the-futex-bug-and-uncovering-towelroot/）。我们现在关心的是，骇客如何构造出q.rt_waiter为NULL，但rt_waiter却还挂在PI chain中的场景？

这样想，futex_cmp_requeue_pi会认为uaddr1代表的是普通futex，uaddr2是pi_futex，所以如果waiter被直接唤醒，那么它肯定不会使用rt_mutex_waiter。

如果我们让uaddr1代表pi_futex，想要的场景不就构造出来了么？futex_cmp_requeue_pi认为rt_mutex_waiter没有使用，但它却已经被挂在了uaddr1对应的PI chain中。

好了，现在的问题是如何让uaddr1代表pi_futex。直接调用futex_wait_requeue_pi是不行的，它不会把uaddr1当作pi_futex，所以不会将rt_mutex_waiter挂上去。解决办法是先让futex_wait_requeue_pi的调用者被requeue一次（先调用一次futex_cmp_requeue_pi将其requeue，需要构造另一个被直接唤醒的炮灰进程），从而使得rt_mutex_waiter被挂入uaddr2对应的PI chain。

然后，此时再调用futex_cmp_requeue_pi(uaddr2, uaddr3)，让waiter被直接唤醒，那不就大功告成了么！

先别高兴得太早，回顾一下之前的数据结构，futex_q有一个成员叫做requeue_pi_key，它会记录下调用futex_wait_requeue_pi时的uaddr2。此时你想将它requeue到uaddr3上，那是糊弄不过去的。futex_cmp_requeue_pi会检查requeue的目标一定等于futex_q.requeue_pi_key。

但是，看似内核已经把路给堵死了，谁曾想居然还有这么一招：futex_cmp_requeue_pi(uaddr2, uaddr2)。由于futex_cmp_requeue_pi的BUG，没有去检查两个uaddr是否不同，还真的就此被蒙混过关，成就了一个影响甚广的漏洞。