PIP在Linux+glibc中的实现

Posted on 2019-02-03

优先级反转

优先级反转指一个高优先级任务因为资源被低优先级任务占用，而不得不等待。这个逻辑本身没有什么问题。优先级反转带来最直接的问题就是高优先级可能因为这次阻塞，而无法确定阻塞时延。如下图所示。

Task 9优先级较低，所以很容易被其他优先级的任务抢占运行，导致对资源S的锁无法释放，而高优先级的任务因为要持有该锁，不得不等待。

优先级继承

Priority Inheritance Protocol翻译为优先级继承，简称PIP。这是解决上述Unbounded inversion最直接的办法，就是将占有锁的线程优先级调整到在等待的任务的最高值。这个算法是贪婪的，局部最优，但不是整体最优解，在此之上又有PCP(优先级天花板)和SRP(基于栈的资源策略)。本文不做讨论。本文主要从源码的角度，阐述PIP在Linux和glibc中的具体实现。

Overview

Linux下的PIP实现分成以上3个部分：

glibc中的pthread_mutex_lock提供用户接口，通过FUTEX_LOCK_PI实现优先级调整
Linux的futex部分提供FUTEX_LOCK_PI的实现
最终的锁机制通过rt_mutex来实现

pthread_mutex

Types

pthread mutex是POSIX pthread库中定义的一种锁机制，与信号量不同的是，这个锁只有两种状态：locked或unlocked。与binary semaphore不同的是，mutex只能被lock的线程unlock，即只能被mutex owner unlock。

目前POSIX标准(1003.1-2008)中定义了4种类型的mutex:

PTHREAD_MUTEX_NORMAL
PTHREAD_MUTEX_ERRORCHECK
PTHREAD_MUTEX_RECURSIVE
PTHREAD_MUTEX_DEFAULT

这4种mutex不同之处如下

	attempting to relock this mutex without first unlocking	Attempting to unlock a mutex locked by a different thread
NORMAL	deadlock	undefined
ERRORCHECK	return error -EDEADLOCK	return error
RECURSIVE	succeed in locking	return error
DEFAULT	undefined	undefined

glibc对上述4种类型的mutex进行了扩展，例如：

    case PTHREAD_MUTEX_PI_RECURSIVE_NP:
    case PTHREAD_MUTEX_PI_ERRORCHECK_NP:
    case PTHREAD_MUTEX_PI_NORMAL_NP:
    case PTHREAD_MUTEX_PI_ADAPTIVE_NP:
    case PTHREAD_MUTEX_PI_ROBUST_RECURSIVE_NP:
    case PTHREAD_MUTEX_PI_ROBUST_ERRORCHECK_NP:
    case PTHREAD_MUTEX_PI_ROBUST_NORMAL_NP:
    case PTHREAD_MUTEX_PI_ROBUST_ADAPTIVE_NP:

主要多了这3种属性：

ADAPTIVE：在SMP架构下，并不直接进入wait，而是先调用spinlock等待一小段时间。musl里默认都采用了这种实现
PI：包含该属性的mutex，实现了优先级继承(Priority Inheritance, PI)
ROBUST：若线程退出时未释放锁，则以后对该锁操作都会返回-EOWNERDEAD，与之对应的是PTHREAD_MUTEX_STALLED(这是默认配置)，对该异常锁的操作会出现undefined行为。

再与原先的4种属性进行排列组合。

"_NP" suffix

Note: the _NP suffix indicates a non-portable extension to the POSIX specification. In fact the latest specification, 1003.1-2008 [2], includes most of the Linux additions so you can leave the _NP ending off. I have chosen not to because some older versions of the header files only have the _NP variants.

根据最新的POSIX标准1003.1-2008，这些扩展并没有加入。

本文只关注MUTEX_PI实现。

PTHREAD_MUTEX_PI

pthread_mutex_lock

重要数据结构如下：

typedef union
{
  struct __pthread_mutex_s __data;
  char __size[__SIZEOF_PTHREAD_MUTEX_T];
  long int __align;
} pthread_mutex_t;
struct __pthread_mutex_s
{
  int __lock __LOCK_ALIGNMENT;
  unsigned int __count;
  int __owner;
  ...
}

如果不考虑ROBUST情况的处理，一次针对实现了PI属性的pthread_mutex_lock操作如下：

atomic_compare_and_exchange_val_acq

这是一种原子操作，先比较再赋值，伪代码如下：
int compare_and_swap(int* reg, int oldval, int newval)
{
    ATOMIC();
    int old_reg_val = *reg;
    if (old_reg_val == oldval)
      *reg = newval;
    END_ATOMIC();
    return old_reg_val;
}
针对代码中的具体情况：

oldval == 0: 表示mutex->__data.__lock == 0，即该lock无owner，可以获得。

oldval != 0: 表示该mutex被lock了，从而进入kernel来处理锁的情况。

pthread_mutex_lock的逻辑比较简单，就是通过一个原子操作判断锁是否已经被锁住。如果没锁住就更新owner，以及user计数。如果已经被锁住，则陷入内核调用FUTEX_LOCK_PI系统调用完成上锁的操作。

pthread_mutex_unlock

这个接口基本上就是直接调用futex_unlock_pi SYSCALL来完成工作：

      /* Unlock the mutex using a CAS unless there are futex waiters or our
›    TID is not the value of __lock anymore, in which case we let the
›    kernel take care of the situation.  Use release MO in the CAS to
›    synchronize with acquire MO in lock acquisitions.  */
      int l = atomic_load_relaxed (&mutex->__data.__lock);
      do
›   {
›     if (((l & FUTEX_WAITERS) != 0)
›         || (l != THREAD_GETMEM (THREAD_SELF, tid)))
›       {
›         INTERNAL_SYSCALL_DECL (__err);
›         INTERNAL_SYSCALL (futex, __err, 2, &mutex->__data.__lock,
›   ›   ›   ›   __lll_private_flag (FUTEX_UNLOCK_PI, private));
›         break;
›       }
›   }
      while (!atomic_compare_exchange_weak_release (&mutex->__data.__lock,
›   ›   ›   ›   ›   ›       &l, 0));

futex

futex是一种内核的等待机制，等待某个用户态数据被设置成某个值。针对优先级继承的futex略有不同。它不再等待用户传入的地址上的值，而是基于内核rt_mutex机制实现等待，上锁和解锁的功能。看下面glibc和Linux的接口对接：

// glibc: pthread_mutex_lock.c
int e = INTERNAL_SYSCALL (futex, __err, 4, &mutex->__data.__lock,
›   ›         __lll_private_flag (FUTEX_LOCK_PI,
›   ›   ›   ›   ›     private), 1, 0);

// Linux: futex.c
long do_futex(u32 __user *uaddr, int op, u32 val, ktime_t *timeout,
›   ›   u32 __user *uaddr2, u32 val2, u32 val3)
{
...
	case FUTEX_LOCK_PI:
›   	return futex_lock_pi(uaddr, flags, NULL, 1);    
... 
}

可见：

uaddr = &mutex->__data.__lock
flags = __lll_private_flag (FUTEX_LOCK_PI, private)
val = 1, timeout = NULL

关于lock/unlock的叙述会简化掉time相关部分，以及PI无关的ROBUST等特性。

重要数据结构

/*
 * Hash buckets are shared by all the futex_keys that hash to the same
 * location.  Each key may have multiple futex_q structures, one for each task
 * waiting on a futex.
 */
struct futex_hash_bucket {
›   atomic_t waiters;
›   spinlock_t lock;
›   struct plist_head chain;
} ____cacheline_aligned_in_smp;

/**
 * struct futex_q - The hashed futex queue entry, one per waiting task
 * @list:›  ›   priority-sorted list of tasks waiting on this futex
 * @task:›  ›   the task waiting on the futex
 * @lock_ptr:›  ›   the hash bucket lock
 * @key:›   ›   the key the futex is hashed on
 * @pi_state:›  ›   optional priority inheritance state
 * @rt_waiter:› ›   rt_waiter storage for use with requeue_pi
 * @requeue_pi_key:›the requeue_pi target futex key
 * @bitset:››   bitset for the optional bitmasked wakeup
 *
 * We use this hashed waitqueue, instead of a normal wait_queue_entry_t, so
 * we can wake only the relevant ones (hashed queues may be shared).
 *
 * A futex_q has a woken state, just like tasks have TASK_RUNNING.
 * It is considered woken when plist_node_empty(&q->list) || q->lock_ptr == 0.
 * The order of wakeup is always to make the first condition true, then
 * the second.
 *
 * PI futexes are typically woken before they are removed from the hash list via
 * the rt_mutex code. See unqueue_me_pi().
 */
struct futex_q {
›   struct plist_node list;

›   struct task_struct *task;
›   spinlock_t *lock_ptr;
›   union futex_key key;
›   struct futex_pi_state *pi_state;
›   struct rt_mutex_waiter *rt_waiter;
›   union futex_key *requeue_pi_key;
›   u32 bitset;
} __randomize_layout;

/*
 * Priority Inheritance state:
 */
struct futex_pi_state {
›   /*
›    * list of 'owned' pi_state instances - these have to be
›    * cleaned up in do_exit() if the task exits prematurely:
›    */
›   struct list_head list;

›   /*
›    * The PI object:
›    */
›   struct rt_mutex pi_mutex;

›   struct task_struct *owner;
›   atomic_t refcount;

›   union futex_key key;
} __randomize_layout;

futex_hash_bucket对应一个用户态的uaddr。通过用户态传入的uaddr地址，来获得对应的futex结构体。用hash的原因是传入的是uaddr地址值，数字很大，分布很散，所以用hash可以有效索引。

futex_q是一个临时变量，对应了一个futex的waiter线程。它通常被用来传递各种数据结构。

pi_state是task_struct内部的一个变量，在futex_lock_pi调用refill_pi_state_cache里面做分配。

static int refill_pi_state_cache(void)
{
...
	pi_state = kzalloc(sizeof(*pi_state), GFP_KERNEL);
    current->pi_state_cache = pi_state;
...
}

futex_lock_pi

这个函数的逻辑还是比较直线的，同大多数锁的实现一样，先尝试原子锁，如果不行，则调用rt_mutex的实现来处理优先级继承的机制。以下捡取比较重要的函数做阐述，也可以参考参考文献[3], [4]中的描述。

queue_lock, queue_unlock

对hash_bucket进行锁操作。他们操作的是hash_bucket下的lock，同时也是futex_q下的lock_ptr。这里锁的出现比较混乱，有时调用上面的这两个函数，有时直接对lock指针操作。需要小心。

__queue_me, unqueue_me

这里将申请futex的线程在hash_bucket的chain上做添加或移除操作。还会涉及锁的操作。

以上4个函数更多的与futex的实现相关，与优先级继承并无关系，所以可以略过。

futex_lock_pi_atomic

这个函数做的就是原子操作来请求锁，

如果futex没有waiter，则直接获得锁并返回1，表示成功获得锁
如果有waiter，就返回0，表示ready to wait

这个函数不同的是会做一些与PI相关的处理：

有waiter的情况：
- 调用attach_to_pi_state来获得该top_waiter的pi_state
没有waiter的情况
- 调用attach_to_pi_owner来取用本线程里分配的pi_state结构体，并更新数据。之后，会被queue到hash_bucket的队列中。

rt_mutex接口

futex_lock_pi中使用到如下这些rt_mutex接口：

rt_mutex_init_waiter：初始化waiter数据结构，其实为一个红黑树的节点。这个waiter对应当前线程，并反映线程在某个锁上的等待关系。
__rt_mutex_start_proxy_lock：这个是PI的核心函数，用来对各waiters数据进行操作，以及遍历lock链来检查死锁。这个函数与之前的原子检查类似，如果返回1表示已经获得锁，如果返回0，则ready to wait。
rt_mutex_wait_proxy_lock：通过__rt_mutex_start_proxy_lock处理之后，如果还没有获得锁，那就可以开始等待了，这个函数将线程切换为TASK_INTERRUPTIBLE状态，并开始等待。这其间会处理基于时间的等待，且会触发调度。
rt_mutex_cleanup_proxy_lock：现在已经阻塞结束了，如果还没获得锁，要么是超时，要么是出错，总之要将本线程从waiters中移除，这个函数就是起这个作用。

fixup_owner

从注释上看，是为了处理lock steal。而lock steal似乎多发生在futex requeue操作上，目前尚未仔细研究。

futex_unlock_pi

futex_unlock_pi释放futex以及更新相应的pi_state和rt_mutex的owner字段。函数调用层次比较深，基本上都是每一层做一些基本的更新工作，最终调用到__rt_mutex_futex_unlock来重新调整本线程的优先级，并将top waiter放到wake queue上。在rt_mutex_postunlock调用的rt_mutex_postunlock将新线程唤醒。

如果没有waiter，则使用cmpxchg_futex_value_locked来将用户态地址uaddr上的值设为0。因为锁没有waiter了，所以优先级也不需要恢复了，或者已经恢复过了。

如果有waiter，就使用下面几个函数来切换lock owner，并恢复优先级继承的优先级。

wake_futex_pi

这个函数做了下面几件事：

用cmpxchg_futex_value_locked将用户态传入的uaddr处的数值更新为新的rt_mutex_next_owner() | FUTEX_WAITERS
list_del_init(&pi_state->list)：将list从原先的表上删除，并重新初始化，准备移到其他的表上。
list_add(&pi_state->list, &new_owner->pi_state_list)
pi_state->owner = new_owner
__rt_mutex_futex_unlock
rt_mutex_postunlock

mark_wakeup_next_waiter

注意：此时pi_state->owner已经设置过了，但是pi_state->pi_mutex->owner还没有设置。这个函数主要做了下面几件事情：

rt_mutex_dequeue_pi(current, waiter);
rt_mutex_adjust_prio(current):在这里调整了当前线程的优先级，完成了整个优先级继承协议。
wake_q_add(wake_q, waiter->task)

put_pi_state

这个函数更多的是清理工作:

调用了rt_mutex_proxy_unlock来释放pi_state->pi_mutex.
释放了current->pi_state_cache空间

cmpxchg_futex_value_locked原子操作

static int cmpxchg_futex_value_locked(u32 *uval, u32 __user *uaddr,
›   ›   ›   ›         u32 oldval, u32 newval)
{
ATOMIC();
int val = *uaddr;
if (val == oldval) {
*uaddr = newval;
}
*uval = val;
END_ATOMIC();
return 0;
}

其中ATOMIC操作通过page_disable()和preempt_disable()保证。

__rt_mutex_start_proxy_lock

从上面futex_lock_pi用到的rt_mutex接口介绍中可知，这个函数是整个PI处理的核心，这个函数又主要调用task_blocks_on_rt_mutex来处理PI。

rt_mutex_adjust_prio(owner)

这个函数就是调整阻塞线程的优先级，从而实现优先级继承的。实现方式也很简单：

if (task_has_pi_waiters(p))
›   pi_task = task_top_pi_waiter(p)->task;

rt_mutex_setprio(p, pi_task);

把owner线程设置为owner线程所有waiter(可能来自多个锁)的最高优先级。

rt_mutex_adjust_prio_chain

为什么有这个链式调整？

这里有两张表格：

每个rt_mutex有张waiters表格，表示所有等待的线程
每个task也有张表格，表格里的每一项对应owner线程占有的锁，每一项是一个rt_mutex_waiter，表示申请该锁的最高优先级waiter

假设Task1就是owner，Task0对Lock0的申请导致Task1提升优先级，如果Task1同时被Task2持有的某个锁阻塞，则Task1优先级的改变，有可能会导致Task2 pi_waiters列表改变，同时可能导致Task2优先级改变。依此类推，就会导致上面的这种链式反应。

链处理流程

整个chain的处理流程大致如下：

如果Task0是Lock0的top waiter，则激发该链式调整。
因为Task0成为Lock0的topwaiter，则Task1的pi_waiters需要调整，则Task1的优先级可能需要调整。退出可能：
- 起点锁释放了
- 链的当前环节被更改( next_lock ≠waiter->lock)
- Task1没有占锁
如果Task1也wait在某个lock的时候，则该lock的waiters需要调整，这里有若干种退出可能。

Task1没有被block
Task1 pi_waiters中有优先级高于Task0的waiter，换句话说，lock0的top_waiter不是Task1的top pi_waiter，优先级继承不需要继续传递
如果Task1在申请Lock1 的时候的优先级（这个记录在waiter->prio里）与当前调整后的优先级相等，则后续没必要在继续调整优先级
检测到DEADLOCK