Futex系统调用，Futex机制，及具体案例分析

Futex

1、背景
- 1.1 自己实现锁
- - 1.1.1 自旋锁
  - 1.1.2 sleep+自旋
  - 1.1.3 小结
- 1.2 futex
- - 1.2.1 什么是Futex
  - 1.2.2 futex诞生之前
  - 1.2.3 futex诞生之后
2、Futex系统调用
3、Futex机制
4、具体案例分析
- 4.1 在Bionic中的实现
- 4.2 C语言实现
5、参考及扩展阅读

首先要区分一下futex系统调用和futex机制。futex系统调用是操作系统提供给上层的系统调用接口。而futex机制是使用futex接口实现的一种锁。

1、背景

线程同步可以说在日常开发中是用的很多，但对于其内部如何实现的，一般人可能知道的并不多。本篇文章将从如何实现简单的锁开始，介绍linux中的锁实现futex的优点及原理。

1.1 自己实现锁

1.1.1 自旋锁

最容易想到可能是自旋：


volatile int status=0;void lock(){while(!compareAndSet(0,1)){}//get lock}void unlock(){status=0;
}boolean compareAndSet(int except,int newValue){//cas操作,修改status成功则返回true
}

上面的代码通过自旋和cas来实现一个最简单的锁。

这样实现的锁显然有个致命的缺点：耗费cpu资源。没有竞争到锁的线程会一直占用cpu资源进行cas操作，假如一个线程获得锁后要花费10s处理业务逻辑，那另外一个线程就会白白的花费10s的cpu资源。（假设系统中就只有这两个线程的情况）。

1.1.2 sleep+自旋

你可能从一开始就想到了，当竞争锁失败后，可以将用Thread.sleep将线程休眠，从而不占用cpu资源：


volatile int status=0;void lock(){while(!compareAndSet(0,1)){sleep(10);}//get lock}void unlock(){status=0;
}

上述方式我们可能见的比较多，通常用于实现上层锁。该方式不适合用于操作系统级别的锁，因为作为一个底层锁，其sleep时间很难设置。sleep的时间取决于同步代码块的执行时间，sleep时间如果太短了，会导致线程切换频繁（极端情况和yield方式一样）；sleep时间如果设置的过长，会导致线程不能及时获得锁。因此没法设置一个通用的sleep值。就算sleep的值由调用者指定也不能完全解决问题：有的时候调用锁的人也不知道同步块代码会执行多久。

1.1.3 小结

对于锁冲突不严重的情况，用自旋锁会更适合，试想每个线程获得锁后很短的一段时间内就释放锁，竞争锁的线程只要经历几次自旋运算后就能获得锁，那就没必要等待该线程了，因为等待线程意味着需要进入到内核态进行上下文切换，而上下文切换是有成本的并且还不低，如果锁很快就释放了，那上下文切换的开销将超过自旋。

目前操作系统中，一般是用自旋+等待结合的形式实现锁：在进入锁时先自旋一定次数，如果还没获得锁再进行等待。

1.2 futex

1.2.1 什么是Futex

Futex 是Fast Userspace muTexes的缩写，由Hubertus Franke, Matthew Kirkwood, Ingo Molnar and Rusty Russell共同设计完成。几位都是linux领域的专家，其中可能Ingo Molnar大家更熟悉一些，毕竟是O(1)调度器和CFS的实现者。

Futex按英文翻译过来就是快速用户空间互斥体。其设计思想其实不难理解，在传统的Unix系统中，System V IPC(inter process communication)，如 semaphores, msgqueues, sockets还有文件锁机制(flock())等进程间同步机制都是对一个内核对象操作来完成的，这个内核对象对要同步的进程都是可见的，其提供了共享的状态信息和原子操作。当进程间要同步的时候必须要通过系统调用(如semop())在内核中完成。可是经研究发现，很多同步是无竞争的，即某个进程进入互斥区，到再从某个互斥区出来这段时间，常常是没有进程也要进这个互斥区或者请求同一同步变量的。但是在这种情况下，这个进程也要陷入内核去看看有没有人和它竞争，退出的时侯还要陷入内核去看看有没有进程等待在同一同步变量上。这些不必要的系统调用(或者说内核陷入)造成了大量的性能开销。为了解决这个问题，Futex就应运而生，Futex是一种用户态和内核态混合的同步机制。首先，同步的进程间通过mmap共享一段内存，futex变量就位于这段共享的内存中且操作是原子的，当进程尝试进入互斥区或者退出互斥区的时候，先去查看共享内存中的futex变量，如果没有竞争发生，则只修改futex,而不用再执行系统调用了。当通过访问futex变量告诉进程有竞争发生，则还是得执行系统调用去完成相应的处理(wait 或者 wake up)。简单的说，futex就是通过在用户态的检查，（motivation）如果了解到没有竞争就不用陷入内核了，大大提高了low-contention时候的效率。 Linux从2.5.7开始支持Futex。

linux底层用futex实现锁，futex由一个内核层的队列和一个用户空间层的atomic integer构成。当获得锁时，尝试cas更改integer，如果integer原始值是0，则修改成功，该线程获得锁，否则就将当期线程放入到 wait queue中（即操作系统的等待队列）。
上述说法有些抽象，如果你没看明白也没关系。我们先看一下没有futex之前，linux是怎么实现锁的。

1.2.2 futex诞生之前

在futex诞生之前，linux下的同步机制可以归为两类：用户态的同步机制和内核同步机制。用户态的同步机制基本上就是利用原子指令实现的自旋锁。关于自旋锁其缺点也说过了，不适用于大的临界区（即锁占用时间比较长的情况）。

内核提供的同步机制，如semaphore等，使用的是上文说的自旋+等待的形式。它对于大小临界区和都适用。但是因为它是内核层的（释放cpu资源是内核级调用），所以每次lock与unlock都是一次系统调用，即使没有锁冲突，也必须要通过系统调用进入内核之后才能识别。

理想的同步机制应该是没有锁冲突时在用户态利用原子指令就解决问题，而需要挂起等待时再使用内核提供的系统调用进行睡眠与唤醒。换句话说，在用户态的自旋失败时，能不能让进程挂起，由持有锁的线程释放锁时将其唤醒？
如果你没有较深入地考虑过这个问题，很可能想当然的认为类似于这样就行了（伪代码）：

void lock(int lockval) {//trylock是用户级的自旋锁while(!trylock(lockval)) {wait();//释放cpu，并将当期线程加入等待队列，是系统调用}
}boolean trylock(int lockval){int i=0; //localval=1代表上锁成功while(!compareAndSet(lockval,0,1)){if(++i>10){return false;}}return true;
}void unlock(int lockval) {compareAndSet(lockval,1,0);notify();
}

上述代码的问题是trylock和wait两个调用之间存在一个窗口：如果一个线程trylock失败，在调用wait时持有锁的线程释放了锁，当前线程还是会调用wait进行等待，但之后就没有人再将该线程唤醒了。

1.2.3 futex诞生之后

  //uaddr指向一个地址，val代表这个地址期待的值，当*uaddr==val时，才会进行waitint futex_wait(int *uaddr, int val);//唤醒n个在uaddr指向的锁变量上挂起等待的进程int futex_wake(int *uaddr, int n);

futex_wait真正将进程挂起之前会检查addr指向的地址的值是否等于val，如果不相等则会立即返回，由用户态继续trylock。否则将当期线程插入到一个队列中去，并挂起。

futex内部维护了一个队列，在线程挂起前会线程插入到其中，同时对于队列中的每个节点都有一个标识，代表该线程关联锁的uaddr。这样，当用户态调用futex_wake时，只需要遍历这个等待队列，把带有相同uaddr的节点所对应的进程唤醒就行了。

另外，futex是支持多进程的，当使用futex在多进程间进行同步时，需要考虑同一个物理内存地址在不同进程中的虚拟地址是不同的。

2、Futex系统调用

其原型和系统调用号为

#include <linux/futex.h>#include <sys/time.h>int futex (int *uaddr, int op, int val, const struct timespec *timeout,int *uaddr2, int val3);#define __NR_futex              240/*虽然参数有点长，其实常用的就是前面三个，后面的timeout大家都能理解，其他的也常被ignore。uaddr就是用户态下共享内存的地址，里面存放的是一个对齐的整型计数器。op存放着操作类型。定义的有5中，这里我简单的介绍一下两种，剩下的感兴趣的自己去man futexFUTEX_WAIT: 原子性的检查uaddr中计数器的值是否为val,如果是则让进程休眠，直到FUTEX_WAKE或者超时(time-out)。也就是把进程挂到uaddr相对应的等待队列上去。FUTEX_WAKE: 最多唤醒val个等待在uaddr上进程。/*

3、Futex机制

所有的futex同步操作都应该从用户空间开始，首先创建一个futex同步变量，也就是位于共享内存的一个整型计数器。当进程尝试持有锁或者要进入互斥区的时候，对futex执行down操作，即原子性的给futex同步变量减1。如果同步变量变为0，则没有竞争发生，进程照常执行。如果同步变量是个负数，则意味着有竞争发生，需要调用futex系统调用的futex_wait操作休眠当前进程。
当进程释放锁或者要离开互斥区的时候，对futex进行up操作，即原子性的给futex同步变量加1。如果同步变量由0变成1，则没有竞争发生，进程照常执行。如果加之前同步变量是负数，则意味着有竞争发生，需要调用futex系统调用的futex_wake操作唤醒一个或者多个等待进程。

这里的原子性加减通常是用CAS(Compare and Swap)完成的，与平台相关。CAS的基本形式是：CAS(addr,old,new),当addr中存放的值等于old时，用new对其替换。在x86平台上有专门的一条指令来完成它: cmpxchg。

可见: futex是从用户态开始，由用户态和核心态协调完成的。

进程或者线程都可以利用futex来进行同步。
对于线程，情况比较简单，因为线程共享虚拟内存空间，虚拟地址就可以唯一的标识出futex变量，即线程用同样的虚拟地址来访问futex变量。
对于进程，情况相对复杂，因为进程有独立的虚拟内存空间，只有通过mmap()让它们共享一段地址空间来使用futex变量。每个进程用来访问futex的虚拟地址可以是不一样的，只要系统知道所有的这些虚拟地址都映射到同一个物理内存地址，并用物理内存地址来唯一标识futex变量。

4、具体案例分析

4.1 在Bionic中的实现

源码:http://www.androidos.net.cn/android/5.0.1_r1/xref/bionic/libc/bionic/pthread_mutex.cpp

4.2 C语言实现

futex 的逻辑可以用如下C语言表示

int val = 0;
void lock()
{int cif ((c = cmpxchg(val, 0, 1)) != 0) {if (c != 2)c = xchg(val, 2);while (c != 0) {futex_wait((&val, 2);c = xchg(val, 2);}}
}   void unlock()
{   if (atomic_dec(val) != 1){val = 0;    futex_wake(&val, 1);}
}
/*
val 0: unlock
val 1: lock, no waiters
val 2: lock , one or more waitersLinux提供了很多操作原子变量的API。以arch/arm/include/asm/atomic.h为例。
#define atomic_xchg(v, new) (xchg(&((v)->counter), new))-----------把new赋值给原子变量v，返回原子变量v的旧值。Linux内核中的cmpxchg函数
在Linux内核中，提供了比较并交换的函数cmpxchg，代码在include/asm-i386/cmpxchg.h中，函数的原型是：
cmpxchg(void *ptr, unsigned long old, unsigned long new);
函数完成的功能是：将old和ptr指向的内容比较，如果相等，则将new写入到ptr中，返回old，如果不相等，则返回ptr指向的内容。
*/

5、参考及扩展阅读

：https://github.com/farmerjohngit/myblog/issues/6
：深入解析Android 5.0系统