v79.01 鴻蒙內核源碼分析(用戶態鎖篇) | 如何使用快鎖Futex(上) | 百篇部落格分析OpenHarmony源碼

• 2022 年 2 月 18 日
• 筆記
• OpenHarmony, 鴻蒙內核源碼分析, 鴻蒙研究站, 鴻蒙內核源碼

百篇部落格分析｜本篇為：(用戶態鎖篇) | 如何使用快鎖Futex(上)

進程通訊相關篇為:

• v26.08 鴻蒙內核源碼分析(自旋鎖) | 當立貞節牌坊的好同志
• v27.05 鴻蒙內核源碼分析(互斥鎖) | 同樣是鎖它卻更豐滿
• v28.04 鴻蒙內核源碼分析(進程通訊) | 九種進程間通訊方式速攬
• v29.05 鴻蒙內核源碼分析(訊號量) | 誰在解決任務間的同步
• v30.07 鴻蒙內核源碼分析(事件控制) | 多對多任務如何同步
• v33.03 鴻蒙內核源碼分析(消息隊列) | 進程間如何非同步傳遞大數據
• v76.01 鴻蒙內核源碼分析(共享記憶體) | 進程間最快通訊方式
• v77.02 鴻蒙內核源碼分析(消息封裝) | 剖析LiteIpc(上)進程通訊內容
• v78.01 鴻蒙內核源碼分析(消息映射) | 剖析LiteIpc(下)進程通訊機制
• v79.01 鴻蒙內核源碼分析(用戶態鎖) | 如何使用快鎖Futex(上)
• v80.01 鴻蒙內核源碼分析(內核態鎖) | 如何實現快鎖Futex(下)

快鎖上下篇

鴻蒙內核實現了Futex，系列篇將用兩篇來介紹快鎖，主要兩個原因:

• 網上介紹Futex的文章很少，全面深入內核介紹的就更少，所以來一次詳細整理和挖透。
• 涉及用戶態和內核態打配合，共同作用，既要說用戶態的使用又要說清楚內核態的實現。
  本篇為上篇，用戶態下如何使用Futex，並藉助一個demo來說清楚整個過程。

基本概念

Futex(Fast userspace mutex，用戶態快速互斥鎖)，系列篇簡稱 快鎖 ，是一個在Linux上實現鎖定和構建高級抽象鎖如訊號量和POSIX互斥的基本工具，它第一次出現在linux內核開發的2.5.7版；其語義在2.5.40固定下來，然後在2.6.x系列穩定版內核中出現，是內核提供的一種系統調用能力。通常作為基礎組件與用戶態的相關鎖邏輯結合組成用戶態鎖，是一種用戶態與內核態共同作用的鎖，其用戶態部分負責鎖邏輯，內核態部分負責鎖調度。

當用戶態執行緒請求鎖時，先在用戶態進行鎖狀態的判斷維護，若此時不產生鎖的競爭，則直接在用戶態進行上鎖返回；反之，則需要進行執行緒的掛起操作，通過Futex系統調用請求內核介入來掛起執行緒，並維護阻塞隊列。

當用戶態執行緒釋放鎖時，先在用戶態進行鎖狀態的判斷維護，若此時沒有其他執行緒被該鎖阻塞，則直接在用戶態進行解鎖返回；反之，則需要進行阻塞執行緒的喚醒操作，通過Futex系統調用請求內核介入來喚醒阻塞隊列中的執行緒。

存在意義

• 互斥鎖(mutex)是必須進入內核態才知道鎖可不可用，沒人跟你爭就拿走鎖回到用戶態，有人爭就得乾等 (包括 有限時間等和無限等待兩種，都需讓出CPU執行權) 或者放棄本次申請回到用戶態繼續執行。那為何互斥鎖一定要陷入內核態檢查呢? 互斥鎖(mutex) 本質是競爭內核空間的某個全局變數(LosMux結構體)。應用程式也有全局變數，但其作用域只在自己的用戶空間中有效，屬於內部資源，有競爭也是應用程式自己內部解決。而應用之間的資源競爭(即內核資源)就需要內核程式來解決，內核空間只有一個，內核的全局變數當然要由內核來管理。應用程式想用內核資源就必須經過系統調用陷入內核態，由內核程式接管CPU，所謂接管本質是要改變程式狀態暫存器，CPU將從用戶態棧切換至內核態棧運行，執行完成後又要切回用戶態棧中繼續執行，如此一來棧間上下文的切換就存在系統性能的損耗。沒看明白的請前往系列篇 (互斥鎖篇) 翻看。

• 快鎖 解決思路是能否在用戶態下就知道鎖可不可用，因為競爭並不是時刻出現，跑到內核態一看其實往往沒人給你爭，白跑一趟來回太浪費性能。那問題來了，用戶態下如何知道鎖可不可用呢？ 因為不陷入內核態就訪問不到內核的全局變數。而自己私有空間的變數對別的進程又失效不能用。越深入研究內核越有一種這樣的感覺，內核的實現可以像數學一樣推導出來，非常有意思。數學其實是基於幾個常識公理推導出了整個數學體系，因為不如此邏輯就無法自洽。如果對內核有一定程度的了解，這裡自然能推導出可以藉助 共享記憶體 來實現！

使用過程

看個linux futex官方demo詳細說明下用戶態下使用Futex的整個過程，程式碼不多，但涉及內核的知識點很多，通過它可以檢驗出內核基本功紮實程度。

//futex_demo.c
#define _GNU_SOURCE
#include <stdio.h>
#include <errno.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/wait.h>
#include <sys/mman.h>
#include <sys/syscall.h>
#include <linux/futex.h>
#include <sys/time.h>
#define errExit(msg)    do { perror(msg); exit(EXIT_FAILURE); \
                        } while (0)
static uint32_t *futex1, *futex2, *iaddr;
/// 快速系統調用
static int futex(uint32_t *uaddr, int futex_op, uint32_t val,
        const struct timespec *timeout, uint32_t *uaddr2, uint32_t val3)
{
    return syscall(SYS_futex, uaddr, futex_op, val,
                    timeout, uaddr2, val3);
}
/// 申請快鎖
static void fwait(uint32_t *futexp)
{
    long s;
    while (1) {
        const uint32_t one = 1;
        if (atomic_compare_exchange_strong(futexp, &one, 0))
            break; //申請快鎖成功
        //申請快鎖失敗,需等待
        s = futex(futexp, FUTEX_WAIT, 0, NULL, NULL, 0);
        if (s == -1 && errno != EAGAIN)
            errExit("futex-FUTEX_WAIT");
    }
}
/// 釋放快鎖
static void fpost(uint32_t *futexp)
{
    long s;
    const uint32_t zero = 0;
    if (atomic_compare_exchange_strong(futexp, &zero, 1)) {//釋放快鎖成功
        s = futex(futexp, FUTEX_WAKE, 1, NULL, NULL, 0);//喚醒等鎖 進程/執行緒
        if (s  == -1)
            errExit("futex-FUTEX_WAKE");
    }
}
/// 父子進程競爭快鎖
int main(int argc, char *argv[])
{
    pid_t childPid;
    int nloops;
    setbuf(stdout, NULL);
    nloops = (argc > 1) ? atoi(argv[1]) : 3;
    iaddr = mmap(NULL, sizeof(*iaddr) * 2, PROT_READ | PROT_WRITE,
                MAP_ANONYMOUS | MAP_SHARED, -1, 0);//創建可讀可寫匿名共享記憶體
    if (iaddr == MAP_FAILED)
        errExit("mmap");
    futex1 = &iaddr[0]; //綁定鎖一地址
    futex2 = &iaddr[1]; //綁定鎖二地址
    *futex1 = 0; // 鎖一不可申請 
    *futex2 = 1; // 鎖二可申請
    childPid = fork();
    if (childPid == -1)
        errExit("fork");
    if (childPid == 0) {//子進程返回
        for (int j = 0; j < nloops; j++) {
            fwait(futex1);//申請鎖一
            printf("子進程  (%jd) %d\n", (intmax_t) getpid(), j);
            fpost(futex2);//釋放鎖二
        }
        exit(EXIT_SUCCESS);
    }
    // 父進程返回執行
    for (int j = 0; j < nloops; j++) {
        fwait(futex2);//申請鎖二
        printf("父進程 (%jd) %d\n", (intmax_t) getpid(), j);
        fpost(futex1);//釋放鎖一
    }
    wait(NULL);
    exit(EXIT_SUCCESS);
}

程式碼在wsl2上編譯運行結果如下:

root@DESKTOP-5PBPDNG:/home/turing# gcc ./futex_demo.c -o futex_demo
root@DESKTOP-5PBPDNG:/home/turing# ./futex_demo
父進程 (283) 0
子進程 (284) 0
父進程 (283) 1
子進程 (284) 1
父進程 (283) 2
子進程 (284) 2

解讀

• 通過系統調用mmap 創建一個可讀可寫的共享記憶體iaddr[2]整型數組，完成兩個futex鎖的初始化。內核會在記憶體分配一個共享線性區(MAP_ANONYMOUS | MAP_SHARED)，該線性區可讀可寫( PROT_READ | PROT_WRITE)

  futex1 = &iaddr[0]; //綁定鎖一地址
  futex2 = &iaddr[1]; //綁定鎖二地址
  *futex1 = 0; // 鎖一不可申請 
  *futex2 = 1; // 鎖二可申請
  

  如此futex1和futex2有初始值並都是共享變數，想詳細了解mmap內核實現的可查看系列篇 （線性區篇） 和 （共享記憶體篇） 有詳細介紹。
• childPid = fork(); 創建了一個子進程，fork會拷貝父進程線性區的映射給子進程，導致的結果就是父進程的共享線性區到子進程這也是共享線性區，映射的都是相同的物理地址。對fork不熟悉的請前往翻看，系列篇 （fork篇）| 一次調用，兩次返回 專門說它。
• fwait(申請鎖)與fpost(釋放鎖)成對出現，單獨看下申請鎖過程

  /// 申請快鎖
  static void fwait(uint32_t *futexp)
  {
      long s;
      while (1) {
          const uint32_t one = 1;
          if (atomic_compare_exchange_strong(futexp, &one, 0))
              break; //申請快鎖成功
          //申請快鎖失敗,需等待
          s = futex(futexp, FUTEX_WAIT, 0, NULL, NULL, 0);
          if (s == -1 && errno != EAGAIN)
              errExit("futex-FUTEX_WAIT");
      }
  }
  

  死循環的break條件是 atomic_compare_exchange_strong為真，這是個原子比較操作，此處必須這麼用，至於為什麼請前往翻看系列篇 （原子操作篇）| 誰在為完整性保駕護航 ，注意它是理解Futex的關鍵所在，它的含義是

  在頭文件<stdatomic.h>中定義
  _Bool atomic_compare_exchange_strong（volatile A * obj，C * expected，C desired）;
  

  將所指向的值obj與所指向的值進行原子比較expected，如果相等，則用前者替換前者desired（執行讀取 - 修改 - 寫入操作）。否則，載入實際值所指向的obj進入*expected（進行負載操作）。
  什麼意思 ? 來個直白的解釋 :
  • 如果 futexp == 1 則 atomic_compare_exchange_strong返回真，同時將 futexp的值變成0，1代表可以持有鎖，一旦持有立即變0，別人就拿不到了。所以此處甚秒。而且這發生在用戶態。
  • 如果futexp == 0 atomic_compare_exchange_strong返回假，沒有拿到鎖，就需要陷入內核態去掛起任務等待鎖的釋放

  futex(futexp, FUTEX_WAIT, 0, NULL, NULL, 0) //執行一個等鎖的系統調用
  

  參數四為NULL代表不在內核態停留直接返回用戶態，後續將在內核態部分詳細說明。
• childPid == 0是子進程的返回。不斷地申請futex1 釋放futex2

  if (childPid == 0) {//子進程返回
      for (int j = 0; j < nloops; j++) {
          fwait(futex1);
          printf("子進程  (%jd) %d\n", (intmax_t) getpid(), j);
          fpost(futex2);
      }
      exit(EXIT_SUCCESS);
  }
  

• 最後的父進程的返回，不斷地申請futex2 釋放futex1

  // 父進程返回執行
  for (int j = 0; j < nloops; j++) {
      fwait(futex2);
      printf("父進程 (%jd) %d\n", (intmax_t) getpid(), j);
      fpost(futex1);
  }
  wait(NULL);
  exit(EXIT_SUCCESS);
  

• 兩把鎖的初值為 *futex1 = 0; *futex2 = 1; ，父進程在 fwait(futex2)所以父進程的printf將先執行，*futex2 = 0;鎖二變成不可申請，列印完成後釋放fpost(futex1)使其結果為*futex1 = 1; 表示鎖一可以申請了，而子進程在等fwait(futex1)，交替下來執行的結果為

    父進程 (283) 0
    子進程 (284) 0
    父進程 (283) 1
    子進程 (284) 1
    父進程 (283) 2
    子進程 (284) 2
  

幾個問題

以上是個簡單的例子，只發生在兩個進程搶一把鎖的情況下，如果再多幾個進程搶一把鎖時情況就變複雜多了。
例如會遇到以下情況:

• 鴻蒙內核進程池默認上限是64個，除去兩個內核進程外，剩下的都歸屬用戶進程，理論上用戶進程可以創建很多快鎖,這些快鎖可以用於進程間(共享快鎖)也可以用於執行緒間(私有快鎖)，在快鎖的生命周期中該如何保存 ?
• 無鎖時，前面已經有進程在申請鎖時，如何處理好新等鎖進程和舊等鎖進程的關係 ?
• 釋放鎖時，需要喚醒已經在等鎖的進程，喚醒的順序由什麼條件決定 ?

這些工作在用戶態下肯定沒辦法完成，需要內核處理，請查看 (內核態鎖篇) | 如何實現快鎖Futex(下)，詳細解構其實現過程。

百文說內核 | 抓住主脈絡

• 百文相當於摸出內核的肌肉和器官系統，讓人開始豐滿有立體感，因是直接從注釋源碼起步，在加註釋過程中，每每有心得處就整理,慢慢形成了以下文章。內容立足源碼，常以生活場景打比方儘可能多的將內核知識點置入某種場景，具有畫面感，容易理解記憶。說別人能聽得懂的話很重要! 百篇部落格絕不是百度教條式的在說一堆詰屈聱牙的概念，那沒什麼意思。更希望讓內核變得栩栩如生，倍感親切。
• 與程式碼需不斷debug一樣，文章內容會存在不少錯漏之處，請多包涵，但會反覆修正，持續更新，v**.xx 代表文章序號和修改的次數，精雕細琢，言簡意賅，力求打造精品內容。
• 百文在 < 鴻蒙研究站 | 開源中國 | 部落格園 | 51cto | csdn | 知乎 | 掘金 > 站點發布，公眾號回復 百文 可方便閱讀。

按功能模組:

百萬注源碼 | 處處扣細節

• 百萬漢字註解內核目的是要看清楚其毛細血管，細胞結構，等於在拿放大鏡看內核。內核並不神秘，帶著問題去源碼中找答案是很容易上癮的，你會發現很多文章對一些問題的解讀是錯誤的，或者說不深刻難以自圓其說，你會慢慢形成自己新的解讀，而新的解讀又會碰到新的問題，如此層層遞進，滾滾向前，拿著放大鏡根本不願意放手。

• < gitee | github | coding | gitcode > 四大碼倉推送 | 同步官方源碼，公眾號中回復 百萬 可方便閱讀。

  

關注不迷路 | 程式碼即人生