Futex學習筆記

1. Futex同步機制簡介

Futex是fast userspace mutex的縮寫，意思是快速用戶空間互斥體。它由Hubertus Franke，Matthew Kirkwood，Ingo Molnar和Rusty Russell設計並維護。我們這里討論Futex是因為在Android中不但線程函數中使用到了Futex，甚至一些模塊中在直接使用Futex作為進程間同步的手段，了解Futex的原理將有助於我們更深入的理解這些模塊的運行機制。

Linux從2.5.7開始支持Futex。在Unix系統中，傳統的進程間同步機制都是通過對內核對象操作來完成的，這個內核對象在需要同步的進程中都是可見的，進程間的同步是通過系統調用在內核中完成。這種同步方式因為涉及用戶態和內核態的切換，效率比較低。而且只要使用了傳統的同步機制，進入臨界區時即使沒有其他的進程競爭也必須切換到內核態來檢查內核同步對象的狀態，這種不必要的切換顯然帶來了大量的浪費。

Futex就是為了解決這個問題而誕生的。Futex是一種用戶態和內核態混合的同步機制，使用Futex同步機制，如果用於進程間同步，需要先調用mmap創建一塊共享內存，Futex變量就位於共享區。同時對Futex變量的操作必須是原子的，當進程試圖進入臨界區或者退出臨界區的時候，首先檢查共享內存中的Futex變量，如果沒有其他的進程也申請使用臨界區，則只修改Futex變量而不再執行系統調用。如果同時有其他進程在申請使用臨界區，還是需要通過系統調用去執行等待或喚醒操作。這樣通過用戶態的Futex變量的控制，減少了進程在用戶態和內核態之間切換的次數，從而減少了系統同步的開銷。

futex (fast userspace mutex) 是Linux的一個基礎組件，可以用來構建各種更高級別的同步機制，比如鎖或者信號量等等，POSIX信號量就是基於futex構建的。

傳統的SystemV IPC(inter process communication)進程間同步機制都是通過內核對象來實現的，以 semaphore 為例，當進程間要同步的時候，必須通過系統調用semop(2)進入內核進行PV操作。

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進行處理。

 

2. Futex用戶態操作

Futex的系統調用FUTEX_WAIT和FUTEX_WAKE只是用來掛起或者喚醒進程，Futex的同步機制還包括用戶態下的判斷操作。用戶態下的操作沒有固定的函數調用，只是一種檢測共享變量的方法。下面將介紹如何將Futex用於臨界區。

首先需要創建一個整型計數器作為Futex變量，如果是進程間同步，這個變量必須位於共享內存。Futex變量的初始值為0。

當進程或線程嘗試持有鎖的時候，檢查Futex變量的值是否為0。如果是，則將Futex變量的值設為1以后再繼續，如果不是，將Futex變量的值設為2以后再執行FUTEX_WAIT()的系統調用進入等待。

前面的值0表示無鎖的狀態，1表示有鎖無競爭的狀態，2表示有競爭的狀態。

當進程或線程釋放鎖的時候，如果Futex變量的值為1，說明沒有其他的線程在等待鎖，這樣把Futex變量的值設為0就結束了。如果Futex變量的值為2，說明還有線程在等待鎖，將Futex變量的值設為0，同時還需要執行FUTEX_WAKE()系統調用來喚醒等待的進程。

在對Futex變量操作時，比較和賦值操作必須是原子的。Bionic提供了用於這種原子操作的函數__bionic_cmpxchg()，這個函數的原型如下：

int __bionic_cmpxchg(int32_t old_value, int32_t new_value, volatile int32_t* ptr)

這個函數會比較ptr地址中的值和old_value的值，相等則將new_value賦予ptr指向的地址。函數如果成功則返回0，否則返回非0。

 

3. Futex系統調用

在Linux中，Futex系統調用的定義如下

#define __NR_futex 240

3.1. Futex系統調用的用法為:

int futex (int *uaddr, int op, int val, const struct timespec *timeout, int *uaddr2, int val3);

uaddr是Futex變量，一個共享的整型計數器。

op表示操作類型，有五種預定義的值，但是在bionic庫中只使用了下面兩種：

1) FUTEX_WAIT: 內核將檢查uaddr中計數器的值是否等於val，如果等於則掛起進程，直到在uaddr上到來了FUTEX_WAKE調用或者超時時間到。

2) FUTEX_WAKE: 內核喚醒val個等待在uaddr上的進程。

val存放與操作op相關的值。

timeout用作操作FUTEX_WAIT中，表示等待超時時間。

uAddr2和val3很少使用。

 

3.2. 在bonic中，提供了兩個函數來包裝Futex系統調用

extern int __futex_wait(volatile void *ftx, int val, const struct timespec *timeout);
extern int __futex_wake(volatile void *ftx, int count);

bionic庫中還有兩個類似的函數，它們的原型如下：

extern int  __futex_wake_ex(volatile void *ftx, int pshared, int val);
extern int  __futex_wait_ex(volatile void *ftx, int pshared, int val, const struct timespec *timeout);

這兩個函數比前面的多了一個參數pshared。pshared的值為true表示wake和wait操作是用於進程間的掛起和喚醒；值為false表示操作用於進程內線程的掛起和喚醒。當pshared的值為false時，執行Futex系統調用的操作碼為:

FUTEX_WAIT|FUTEX_PRIVATE_FLAG，
FUTEX_WAKE|FUTEX_PRIVATE_FLAG

這樣內核如果檢測到操作有 FUTEX_PRIVATE_FLAG 標記，能以更快的速度執行掛起和喚醒操作。

__futex_wait 和 __futex_wake函數相當於pshared等於true的情況。

3.3. Android中有些模塊還會使用下面的Futex函數：

extern int __futex_syscall3(volatile void *ftx, int op, int val);
extern int __futex_syscall4(volatile void *ftx, int op, int val, const struct timespec *timeout);

__futex_syscall3()相當於__futex_wake()，而__futex_syscall4()相當於__futex_wait()。這兩個函數與前面的區別是能指定操作碼op做為參數。操作碼可以是FUTEX_WAIT，FUTEX_WAKE或者它們和FUTEX_PRIVATE_FLAG的組合。

Android中操作碼的定義如下：

#define FUTEX_WAIT    0
#define FUTEX_WAKE    1

#ifndef FUTEX_PRIVATE_FLAG
#define FUTEX_PRIVATE_FLAG  128
#endif

#ifndef FUTEX_WAIT_PRIVATE
#define FUTEX_WAIT_PRIVATE  (FUTEX_WAIT|FUTEX_PRIVATE_FLAG)
#endif

#ifndef FUTEX_WAKE_PRIVATE
#define FUTEX_WAKE_PRIVATE  (FUTEX_WAKE|FUTEX_PRIVATE_FLAG)
#endif

 

4. 使用例子

/* 
 * 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  /* syscall number */
#endif  
#endif  
  
#define FILE_MODE (S_IRUSR | S_IWUSR)  
  
const char shmfile[] = "/tmp";  //if change to "./tmp", report error!
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\n", 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(2);  
        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;  
          
        struct namelist *cur = (struct 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
        int i;
        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++;  
        struct namelist *cur = (struct namelist *)ptr;  
  
        for (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@ubuntu:/work/8.futex_test# gcc futex_test.c -o pp -lrt
root@ubuntu:/work/8.futex_test# ./pp
parent start waiting
Child 21650: start
wake up parent
parent end waiting

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

Parent 21649 get child status:0

*/

 

5. 補充

1. kernel\tools\testing\selftests\futex 有對futex的測試代碼

2. 系統調用futex實現

kernel/futex.c 中導出的系統調用：

long get_robust_list(int pid, struct robust_list_head **head_ptr, size_t *len_ptr);

long set_robust_list(struct robust_list_head *head, size_t len);

int futex(int *uaddr, int futex_op, int val, const struct timespec *timeout,/* or: uint32_t val2 */int *uaddr2, int val3);

還有 futex_time32 

3.頭文件

include\uapi\linux\futex.h //內核中futex導出給用戶空間的頭文件
include\linux\futex.h      //內核中的實現頭文件，包含上面uapi下的

 

 

參考：
A futex overview and update：https://lwn.net/Articles/360699/
http://blog.sina.com.cn/s/blog_dae890d10101f01l.html
http://blog.sina.com.cn/s/blog_dae890d10101f01i.html
http://blog.sina.com.cn/s/blog_dae890d10101f01q.html