Linux进程同步机制-Futex

2023-11-01

引子

在编译2.6内核的时候，你会在编译选项中看到[*] Enable futex support这一项，上网查，有的资料会告诉你"不选这个内核不一定能正确的运行使用glibc的程序"，那futex是什么？和glibc又有什么关系呢？

futex诞生之前

在futex诞生之前，linux下的同步机制可以归为两类：用户态的同步机制和内核同步机制。用户态的同步机制基本上就是利用原子指令实现的spinlock。最简单的实现就是使用一个整型数，0表示未上锁，1表示已上锁。trylock操作就利用原子指令尝试将0改为1：

bool trylock(int lockval) {
    int old;
    atomic { old = lockval; lockval = 1; }  // 如：x86下的xchg指令
    return old == 0;
}

spinlock的lock操作则是一个死循环，不断尝试trylock，直到成功。
对于一些很小的临界区，使用spinlock是很高效的。因为trylock失败时，可以预期持有锁的线程（进程）会很快退出临界区（释放锁）。所以死循环的忙等待很可能要比进程挂起等待更高效。

但是spinlock的应用场景有限，对于大的临界区，忙等待则是件很恐怖的事情，特别是当同步机制运用于等待某一事件时（比如服务器工作线程等待客户端发起请求）。所以很多情况下进程挂起等待是很有必要的。

内核提供的同步机制，诸如semaphore等，其实骨子里也是利用原子指令实现的spinlock，内核在此基础上实现了进程的睡眠与唤醒。
使用这样的锁，能很好的支持进程挂起等待。但是最大的缺点是每次lock与unlock都是一次系统调用，即使没有锁冲突，也必须要通过系统调用进入内核之后才能识别。

理想的同步机制应该是在没有锁冲突的情况下在用户态利用原子指令就解决问题，而需要挂起等待时再使用内核提供的系统调用进行睡眠与唤醒。 换句话说，用户态的spinlock在trylock失败时，能不能让进程挂起，并且由持有锁的线程在unlock时将其唤醒？

参考文档：Linux Futex浅析

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。

futex设计思想

futex的解决思路是：在无竞争的情况下操作完全在user space进行，不需要系统调用，仅在发生竞争的时候进入内核去完成相应的处理(wait 或者 wake up)。

所以说，futex是一种user mode和kernel mode混合的同步机制，需要两种模式合作才能完成，futex变量必须位于user space，而不是内核对象，futex的代码也分为user mode和kernel mode两部分，无竞争的情况下在user mode，发生竞争时则通过sys_futex系统调用进入kernel mode进行处理

Futex系统调用

Futex是一种用户态和内核态混合机制，所以需要两个部分合作完成，linux上提供了sys_futex系统调用，对进程竞争情况下的同步处理提供支持。

 #include <linux/futex.h>
 #include <sys/time.h>
 
 #define __NR_futex    240
    
int futex (int *uaddr, int op, int val, 
           const struct timespec *timeout,
           int *uaddr2, int val3);

uaddr就是用户态下共享内存的地址，里面存放的是一个对齐的整型计数器。
op存放着操作类型。定义的有5中，这里我简单的介绍一下两种，剩下的感兴趣的自己去man futex .

FUTEX_WAIT: 原子性的检查uaddr中计数器的值是否为val,如果是则让进程休眠，直到FUTEX_WAKE或者超时(time-out)。也就是把进程挂到uaddr相对应的等待队列上去。
FUTEX_WAKE: 最多唤醒val个等待在uaddr上进程。

可见FUTEX_WAIT和FUTEX_WAKE只是用来挂起或者唤醒进程，而无法实现同步。 当然这部分工作也只能在内核态下完成。 应该仔细区分futex系统调用和futex同步机制。

Futex同步机制

Futex机制就是将futex变量和futex(2)结合使用的同步机制。当执行一个需要阻塞一个线程的futex操作时，内核只有在futex值和futex()指定的值(val)相等时才会阻塞。加载futex的值，将该值和futex()的参数进行比较，阻塞以原子的方式发生，然后当前线程会和其他阻塞在这个futex字上的其他并发线程一同排队等候。因此，futex字是将用户空间的同步和内核阻塞的实现联系起来。

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

用更加浅显的话解释：

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

futex机制的一个作用是实现锁，futex对象即锁对象，通过对用户空间的futex对象的访问，来判断是否需要执行futex(2)进行阻塞。如果无竞争，则将futex对象减一(通过原子操作[__sync_bool_compare_and_swap()]来判断并加减)；如果有竞争，则调用futex(2)阻塞。

当释放时，调用带FUTEX_WAKE参数的futex(2)，唤醒等待的线程，val值指示唤醒的线程数。需要注意的是，当调用FUTEX_WAKE操作唤醒某个等待在futex上的线程后，此时虽然有线程占用了futex，但是futex不会自动减一，所以需要手动的将futex减一。

由此可以引出一个问题，即如果在线程占有了futex后有新线程申请futex，应该如何处理，[参考资料3]的例程给出了解答，本文例程代码也给了解答。ps: 参考资料3中的例程的futex函数似乎有错误，uaddr2写成了uaddr。

参考文档：初识futex

进/线程利用futex同步

进程或者线程都可以利用futex来进行同步。

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

小结：

Futex变量的特征：1)位于共享的用户空间中 2)是一个32位的整型 3)对它的操作是原子的
Futex在程序low-contention的时候能获得比传统同步机制更好的性能。
不要直接使用Futex系统调用。
Futex同步机制可以用于进程间同步，也可以用于线程间同步。

示例代码：

/*
 * This sample show how to use futex betwen two process, and use system v 
 * shared memory to store data
 */

#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/ipc.h>
#include <sys/mman.h>
#include <sys/types.h>
#include <sys/syscall.h>
#include <sys/wait.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <errno.h>

#if __GLIBC_PREREQ(2, 3)    
#if defined FUTEX_WAIT || defined FUTEX_WAKE 
#include <linux/futex.h>
#else
#define FUTEX_WAIT      0
#define FUTEX_WAKE      1
#endif

#ifndef __NR_futex
#define __NR_futex     202
#endif
#endif

#define FILE_MODE (S_IRUSR | S_IWUSR)

const char shmfile[] = "/tmp";
const int size = 100;

struct namelist 
{
	int  id; 
	char name[20];
};

int main(void)
{
	int fd, pid, status;	
	int *ptr;
	struct stat stat;
		
	// create a Posix shared memory
	int flags = O_RDWR | O_CREAT;
	fd = shm_open(shmfile, flags, FILE_MODE);
    if (fd < 0)
    {
        printf("shm_open failed, errormsg=%s errno=%d", strerror(errno), errno);
        return 0;
    }
	ftruncate(fd, size);
	ptr = (int *)mmap(NULL, size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);	

	pid = fork();
	if (pid == 0) { // child process
        sleep(5);
		printf("Child %d: start/n", getpid());
		
		fd = shm_open(shmfile, flags, FILE_MODE);
		fstat(fd, &stat);		
		ptr = (int *)mmap(NULL, size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);	
		close(fd);
		struct namelist tmp;

		// store total num in ptr[0];
		*ptr = 3;
		
		namelist *cur = (namelist *)(ptr+1);

		// store items
		tmp.id = 1;
		strcpy(tmp.name, "Nellson");
		*cur++ = tmp;
		tmp.id = 2;
		strcpy(tmp.name, "Daisy");
		*cur++ = tmp;
		tmp.id = 3;
		strcpy(tmp.name, "Robbie");
		*cur++ = tmp;

        printf("wake up parent/n");
        syscall(__NR_futex ,ptr, FUTEX_WAKE, 1, NULL );

		exit(0);
	} else{ // parent process
        printf("parent start waiting/n");
        syscall(__NR_futex , ptr, FUTEX_WAIT, *(int *)ptr, NULL );
        printf("parent end waiting/n");

		struct namelist tmp;

		int total = *ptr;
		printf("/nThere is %d item in the shm/n", total);	
		
		ptr++;
		namelist *cur = (namelist *)ptr;

		for (int i = 0; i< total; i++) {
			tmp = *cur;
			printf("%d: %s/n", tmp.id, tmp.name);
			cur++;
		}

		printf("/n");
		waitpid(pid, &status, 0);
	}

	// remvoe a Posix shared memory from system
	printf("Parent %d get child status:%d/n", getpid(), status);
	return 0;
}

结果：

root:/home/ftpuser/ipc# g++ -o  futex -lrt futex.cc     
root:/home/ftpuser/ipc# ./futex
parent start waiting
Child 2825: start
wake up parent
parent end waiting

There is 3 item in the shm
1: Nellson
2: Daisy
3: Robbie

参考文档：Linux进程同步机制-Futex - nellson

总结：

大部分的glibc的同步方式，mutex或者semaphore，大多基于futex的方式，首先进行用户态检查，未果的话进行futex系统调用。这是我疑惑为什么futex这么常用却在代码层面上看不到它，原因是我们使用的都是基于futex的机制
pthread库中的pthread_join也是基于futex的哦，当父进程执行pthread_join它的某一个子线程时，如果子线程已经执行完毕，则父进程不会调用futex系统调用，如果子线程仍然执行中，那么父进程调用futex系统调用进行FUTEX_WAIT休眠，等待子线程的唤醒

互斥锁pthread_mutex_t的实现原理

// pthread_mutex_lock:
atomic_dec(pthread_mutex_t.value);
if (pthread_mutex_t.value!=0)
  futex(WAIT)
else
  success

// pthread_mutex_unlock:
atomic_inc(pthread_mutex_t.value);
if(pthread_mutex_t.value!=1)
futex(WAKEUP)
else
success

信号量sem_t的实现原理

sem_wait(sem_t *sem)
{
for (;;) {
   if (atomic_decrement_if_positive(sem->count))
       break;
   futex_wait(&sem->count, 0)
   }
}

sem_post(sem_t *sem)
{
   n = atomic_increment(sem->count);
   // Pass the new value of sem->count
   futex_wake(&sem->count, n + 1);
}

参考文档: Futex设计与实现

更多参考资料：

[1] https://zh.wikipedia.org/zh-hans/Futex
[2] https://blog.csdn.net/nellson/article/details/5400360 [Linux进程同步机制-Futex]
[3] http://man7.org/linux/man-pages/man2/futex.2.html

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)

Linux C

CC