futex同步机制分析之一应用

futex同步机制分析之一应用
一、多线程（进程）的同步机制
c++编程中最难的部分有哪些，估计绝大多数人都会首先提出来是多线程（进程）编程。为什么多线程编程难呢？一个主要的原因就是多线程的同步。在多线程同步中，（Linux平台Posix NPTL）中主要有三个同步手段：Semaphore（信号灯）、Mutex（互斥体）和Condition Variables（条件变量）。在c++11的标准中，同样也有包含有类似的相关数据类型(std::mutex,std::condition_variable)。
由于CPU的基于流水加上指令顺序的重排（乱序）导致多线程的运行的机制在应用根本无法控制。但在一些具体的业务场景，又要求必须按顺序一步步来。这时候，就必须使用多线程的同步强制线程按要求来工作。但是在使用这些同步机制时，需要编程者清楚操作系统的进程线程调用机制和相关的数据在多线程间的传递进行控制的方式。而这些配合有一个不能准确到位，就可能会引起程序的崩溃。但是需要注意的是，崩溃并不是最可怕的，可怕是程序跑得很好，但是达不到设计的目的。还有更可怕的，程序跑得很好，正常的情况下也符合预期，但是在某个关键点上，程序就出错了。这样的例子，并不少见，一个比一个难以解决。于是，这也成了多线程编程令一般程序员谈之变色的一个主要原因。
这次聚集于Linux平台下的多线程同步的底层同步实现机制futex.从这个方向上为解决同步问题做一点努力。

二、Linux中的futex

1、多线程介绍

Futex(Fast Userspace Mutexes)做为一种快速同步机制，从linux 2.5.7开始支持，而c++编程中使用的glibc使用NPTL（Native POSIX Thread Library）做为自己的线程库。而NPTL基本实现了POSIX。
在Linux早期系统中，并没有线程这个概念，线程的概念在Windows平台上应用居多。而Linux主要是使用Clone的方式来产生进程编程，这也是目前好多人仍然把线程叫做轻量级进程的一个缘由。
最初，Linux中使用LinuxThreads利用进程来模拟用户空间的线程，但是比较不好的，它不符合Posix的标准，都可以理解的是，CPU在同一进程间的线程之间切换的速度要比多个进程间切换的速度要快。当然，它还有其它很多的缺点。这些缺点，导致开发人员对其并不是非常满意，所以后来IBM和REDHAT团队开始了两个新的实现即：NGPT（Next-Generation POSIX Threads）和NPTL，后来由于某些原因，前者不再维护，硕果仅存，就只有NPTL了。
LinuxThreads目前很少再维护了，NPTL仍然在发展，不过，NPTL目前还有一些小的问题，比如在SMP架构的CPU上有时会有问题。

2、早期同步机制

1) 实现方式

在早期的Linux系统中，多线程的同步分为两种机制，即内核空间机制和用户空间机制。在用户空间进行同步，首先需要使用原子指令将代码锁住，如果利用atomic的原子机制编写过代码，可能这些就很好理解了。
这里举一个比喻，就好像一个高速的循环，不断的去查询一个标志（可以理解成一个整数：0表示没被锁上，1表示被锁住），所以这个又被形象的称为自旋锁。这时候，CPU是被独占的，如果长时间的CPU被独占，可想而知结果会是如何。所以使用这种机制是有一些限定要求的：
临界区代码尽量控制在100行以内。
不要调用系统函数（read等），调用其它函数也应尽量短小。
不要有大循环，不要有类似sleep动作。
大的内存复制时，memcpy等函数不建议使用。
而在内核空间中，实现同步机制和用户空间中没有本质的区别，同样也是类似的使用原子指令操作，正是基于此，内核实现进程的等待和睡眠等功能。

2）缺点

不知道大家对于条件变量的使用有没有踩过坑，在网上陈硕曾经总结过这些，形式化一下类似下面这种：

void lock(int con) {
while (!trylock(con)) {
wait();
}
}

上面刚刚说过，多线程的无序性可能会导致trylock和wait间有一个空档，在这个空档期内如果有人释放了锁，这个进程基本就睡不醒了。有没有解决办法呢，有，但是比较复杂。可以利用sigsuspend来间接实现，这不是重点，不展开。有兴趣可以在网上查找相关资料。
另外在早期的系统中，经常会使用内核对象来实现同步，这样就会导致一个问题，无论用户空间的进程是否实际需要同步，都得进内核空间中进行操作一番，而内核空间与用户空间的交互，代价还是比较大的。

3、futex的特点

基于上面的种种不便，“物不平则鸣”，牛人们（Hubertus Franke, Matthew Kirkwood, Ingo Molnar and Rusty Russell，无一不是大神啊）创建了futex。这玩意是贯通内核空间和用户空间的一把利器。看看它的中文意义“快速用户空间互斥体”。意味着，快。
Futex是怎么突出这个快的呢？在同步的线程间，开辟一段共享内存（mmap）, futex的变量保存在这段共享内存中，当线程进入（退出）同步区域时，先查询共享内存的中futex变量，如果竞争没有出现，就只修改futex状态，不再进入内核空间调用。反之，还得进入内核控制wait(wake).这样，通过一小段共享内存在用户空间的检查，就极大的避免了不应该进入内核的陷阱。自然，在低频率竞争锁时，效率被大大提高了。
通过上面的分析可以看出，Futex其实是可以看做内核空间和用户空间两种之间的一种过渡状态，或者类似于物理上的亚稳定状态，根据不同的情况，进行不两只的处理。

三、Futex的使用

1、接口调用

#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

uaddr:用户态下共享内存的地址，里面存放的是一个对齐的整型计数器。
op:操作类型。定义的有五种：
FUTEX_WAIT（Linux 2.6.0）: 检查uaddr中计数器的值是否为val（原子操作）,如果是则让线程（进程，其下二者等同）休眠，直到FUTEX_WAKE或者超时(time-out)。线程将挂到uaddr相对应的等待队列上去。反之，返回失败及错误值error EAGAIN。
FUTEX_WAKE（Linux 2.6.0）: 最多唤醒val个等待在uaddr上线程。
FUTEX_FD:用于文件句柄的同步，已经在linux2.6.26被移除，忽略。
FUTEX_REQUEUE (Linux 2.6.0)：和 FUTEX_CMP_REQUEUE 完成相同的功能，但是不检查val3即参数val3被忽略。
FUTEX_CMP_REQUEUE (Linux 2.6.7)：检查uaddr是否等于val3，如果不等，返回error EAGAIN。反之，将唤醒最多val个等待者。如果等待线程数量大于val，则其它线程从uaddr的等待队列中删除并添加到uaddr2的等待队列中。val2参数指定uaddr2的等待线程的最大数量。
val的典型值为0或1，指定为INT_MAX无意义，导致的结果是FUTEX_CMP_REQUEUE 操作和 FUTEX_WAKE 语义相同；val2的值一般为1或INT_MAX，指定为0是没有用的，这样导致 FUTEX_CMP_REQUEUE秋 FUTEX_WAIT语义相同。
FUTEX_CMP_REQUEUE 较之 FUTEX_REQUEUE更为安全，由于前者会检查uaddr的值，从而防止特定条件下的竞态条件。FUTEX_CMP_REQUEUE 和 FUTEX_REQUEUE 都可以用来避免 FUTEX_WAKE 可能产生的“惊群”现象。
值得提起注意的是，要区分开Futex的同步机制和Futex的系统调用的不同，前者既有用户空间层面的操作又有内核空间层面的操作。如果希望通过使用futex系统调用来实现性能的提升可能和直接使用自旋锁一样，没啥效果，搞不好还掉坑里去。
futex的原子操作是使用用CAS(Compare and Swap)完成的，与平台相关。CAS是完成无锁队列的一个重要手段，如果写过无锁队列的应该会清楚。在x86平台利用 cmpxchg指令来完成。

2、例子分析

这里用mutex来实现一个例子并用strace来查看futex的调用过程。

1 #include <stdio.h>
2 #include <stdlib.h>
3 #include <time.h>
4 #include <semaphore.h>
5 #include <pthread.h>
6 #include <unistd.h>
7
8 sem_t sem_a;
9 pthread_mutex_t lock;
10 int num = 0;
11 void *thread_process(void *);
12 void *thread_mutex(void*);
13
14 int main(void)
15 {
16 int ret = 0;
17 pthread_t pHandle_;
18 pthread_t pHMutex_;
19 sem_init(&sem_a,0,1);
20
21 pthread_mutex_init(&lock,NULL);
22
23 ret = pthread_create(&pHandle_,NULL,thread_process,NULL);
24 ret = pthread_create(&pHMutex_,NULL,thread_mutex,NULL);
25 pthread_join(pHandle_,NULL);
26 pthread_join(pHMutex_,NULL);
27 }
28
29 void *thread_mutex(void *params)
30 {
31 int total =0;
32 pthread_mutex_lock(&lock);
33 num++;
34 pthread_mutex_unlock(&lock);
35
36 }
37 void *thread_process(void *params)
38 {
39 int total = 0;
40 sem_wait(&sem_a);
41 sleep(5);
42
43 sem_getvalue(&sem_a,&total);
44 printf("sem value is:%d\n",total);
45 sem_post(&sem_a);
46 }

一个非常简单的测试例程，下面用strace跟踪一下调用过程：

......
1 6936 execve("./futex", ["./futex"], 0x7ffd06e47680 /* 54 vars */) = 0 <0.000261>
......
7 6936 mmap(NULL, 91985, PROT_READ, MAP_PRIVATE, 3, 0) = 0x7fdcac494000 <0.000032>
8 6936 close(3) = 0 <0.000030>
9 6936 access("/etc/ld.so.nohwcap", F_OK) = -1 ENOENT (No such file or directory) <0.000032>
10 6936 openat(AT_FDCWD, "/lib/x86_64-linux-gnu/libpthread.so.0", O_RDONLY|O_CLOEXEC) = 3 <0.000034>
11 6936 read(3, "\177ELF\2\1\1\0\0\0\0\0\0\0\0\0\3\0>\0\1\0\0\0000b\0\0\0\0\0\0"..., 832) = 832 <0.000032>
12 6936 fstat(3, {st_mode=S_IFREG|0755, st_size=144976, ...}) = 0 <0.000031>
13 6936 mmap(NULL, 8192, PROT_READ|PROT_WRITE, MAP_PRIVATE|MAP_ANONYMOUS, -1, 0) = 0x7fdcac492000 <0.000031>
14 6936 mmap(NULL, 2221184, PROT_READ|PROT_EXEC, MAP_PRIVATE|MAP_DENYWRITE, 3, 0) = 0x7fdcac065000 <0.000032>
15 6936 mprotect(0x7fdcac07f000, 2093056, PROT_NONE) = 0 <0.000033>
16 6936 mmap(0x7fdcac27e000, 8192, PROT_READ|PROT_WRITE, MAP_PRIVATE|MAP_FIXED|MAP_DENYWRITE, 3, 0x19000) = 0x7fdcac27e000 <0.000041>
17 6936 mmap(0x7fdcac280000, 13440, PROT_READ|PROT_WRITE, MAP_PRIVATE|MAP_FIXED|MAP_ANONYMOUS, -1, 0) = 0x7fdcac280000 <0.000027>
18 6936 close(3) = 0 <0.000036>
19 6936 access("/etc/ld.so.nohwcap", F_OK) = -1 ENOENT (No such file or directory) <0.000033>
20 6936 openat(AT_FDCWD, "/lib/x86_64-linux-gnu/libc.so.6", O_RDONLY|O_CLOEXEC) = 3 <0.000043>
21 6936 read(3, "\177ELF\2\1\1\3\0\0\0\0\0\0\0\0\3\0>\0\1\0\0\0\260\34\2\0\0\0\0\0"..., 832) = 832 <0.000036>

22 6936 fstat(3, {st_mode=S_IFREG|0755, st_size=2030544, ...}) = 0 <0.000067>
23 6936 mmap(NULL, 4131552, PROT_READ|PROT_EXEC, MAP_PRIVATE|MAP_DENYWRITE, 3, 0) = 0x7fdcabc74000 <0.000033>
24 6936 mprotect(0x7fdcabe5b000, 2097152, PROT_NONE) = 0 <0.000033>
25 6936 mmap(0x7fdcac05b000, 24576, PROT_READ|PROT_WRITE, MAP_PRIVATE|MAP_FIXED|MAP_DENYWRITE, 3, 0x1e7000) = 0x7fdcac05b000 <0.000035>
26 6936 mmap(0x7fdcac061000, 15072, PROT_READ|PROT_WRITE, MAP_PRIVATE|MAP_FIXED|MAP_ANONYMOUS, -1, 0) = 0x7fdcac061000 <0.000033>
27 6936 close(3) = 0 <0.000030>
28 6936 mmap(NULL, 12288, PROT_READ|PROT_WRITE, MAP_PRIVATE|MAP_ANONYMOUS, -1, 0) = 0x7fdcac48f000 <0.000031>
29 6936 arch_prctl(ARCH_SET_FS, 0x7fdcac48f740) = 0 <0.000020>
30 6936 mprotect(0x7fdcac05b000, 16384, PROT_READ) = 0 <0.000034>
31 6936 mprotect(0x7fdcac27e000, 4096, PROT_READ) = 0 <0.000032>
32 6936 mprotect(0x561b1137c000, 4096, PROT_READ) = 0 <0.000032>
33 6936 mprotect(0x7fdcac4ab000, 4096, PROT_READ) = 0 <0.000032>
34 6936 munmap(0x7fdcac494000, 91985) = 0 <0.000040>
35 6936 set_tid_address(0x7fdcac48fa10) = 6936 <0.000030>
36 6936 set_robust_list(0x7fdcac48fa20, 24) = 0 <0.000030>
......
46 6936 mmap(NULL, 8392704, PROT_NONE, MAP_PRIVATE|MAP_ANONYMOUS|MAP_STACK, -1, 0) = 0x7fdcaac72000 <0.000032>
47 6936 mprotect(0x7fdcaac73000, 8388608, PROT_READ|PROT_WRITE <unfinished ...>
48 6937 set_robust_list(0x7fdcabc739e0, 24 <unfinished ...>
49 6936 <... mprotect resumed> ) = 0 <0.000022>
50 6936 clone( <unfinished ...>
51 6937 <... set_robust_list resumed> ) = 0 <0.000068>
52 6937 nanosleep({tv_sec=5, tv_nsec=0}, <unfinished ...>
53 6936 <... clone resumed> child_stack=0x7fdcab471fb0, flags=CLONE_VM|CLONE_FS|CLONE_FILES|CLONE_SIGHAND|CLONE_THREAD|CLONE_SYSVSEM|CLONE_SETTLS|CLONE_PARENT_SETTID|CLONE_CHILD_CLEARTID, parent_tidptr= 0x7fdcab4729d0, tls=0x7fdcab472700, child_tidptr=0x7fdcab4729d0) = 6938 <0.000069>
54 6936 futex(0x7fdcabc739d0, FUTEX_WAIT, 6937, NULL <unfinished ...>
55 6938 set_robust_list(0x7fdcab4729e0, 24) = 0 <0.000068>
......
58 6938 +++ exited with 0 +++
59 6937 <... nanosleep resumed> 0x7fdcabc72e90) = 0 <5.001496>
60 6937 fstat(1, {st_mode=S_IFCHR|0620, st_rdev=makedev(136, 0), ...}) = 0 <0.000022>
61 6937 mmap(NULL, 134217728, PROT_NONE, MAP_PRIVATE|MAP_ANONYMOUS|MAP_NORESERVE, -1, 0) = 0x7fdca2c72000 <0.000025>
.....
68 6937 +++ exited with 0 +++

......
71 6936 +++ exited with 0 +++

从上面的跟踪来看，一共有三个线程6936，6937，6938，特别是后面两个子线程中，分别使用了信号量和互斥体，set_robust_list说明使用的robust规则。在新的glibc中，mmap创建共享内存，然后用保护mprotect对其进行控制。而非原来单一一个整形值。
在这里可以看出，新的库里在这种无锁竞态的情况下已经不再调用内核的唤醒查看过程，这是和老的库的区别，估计也是觉得原来的到内核看一看比较不符合正常的逻辑，所以新的版本将其改正了。在robust规则下futex状态字段有三类，是否上锁，是否锁竞争，是否持锁线程死。
是否上锁，为整型的0-29位，0代表无锁，非0代表上锁，并且上锁状态同时保存持锁线程的id号。
是否持锁线程死，为整型30位。
是否锁竞争，为整型31位。
两个配对锁操作 lll_robust_lock 和 lll_robust_unlock。看它调用的锁：

int
__lll_robust_lock (void \*ptr, int flags)
{
int \*iptr = (int \*)ptr;
int id = __getpid ();
int wait_time = 25;
unsigned int val;

/* Try to set the lock word to our PID if it's clear. Otherwise,
mark it as having waiters. \*/
while (1)
{
val = \*iptr;
if (!val && atomic_compare_and_exchange_bool_acq (iptr, id, 0) == 0)
return 0;
else if (atomic_compare_and_exchange_bool_acq (iptr,
val | LLL_WAITERS, val) == 0)
break;
}

for (id |= LLL_WAITERS ; ; )
{
val = \*iptr;
if (!val && atomic_compare_and_exchange_bool_acq (iptr, id, 0) == 0)
return 0;
else if (val && !valid_pid (val & LLL_OWNER_MASK))
{
if (atomic_compare_and_exchange_bool_acq (iptr, id, val) == 0)
return EOWNERDEAD;
}
else
{
lll_timed_wait (iptr, val, wait_time, flags);
if (wait_time < MAX_WAIT_TIME)
wait_time <<= 1;
}
}
}
void
__lll_robust_unlock (void *ptr, int flags)
{
unsigned int val = atomic_load_relaxed ((unsigned int *)ptr);
while (1)
{
if (val & LLL_WAITERS)
{
lll_set_wake (ptr, 0, flags);
break;
}
else if (atomic_compare_exchange_weak_release ((unsigned int *)ptr, &val, 0))
break;
}
}