鎖開銷優化以及 CAS 簡單說明

時間 2019-11-30

標籤開銷優化以及 cas 簡單說明简体版

原文原文鏈接

鎖開銷優化以及 CAS 簡單說明

鎖

互斥鎖是用來保護一個臨界區，即保護一個訪問共用資源的程序片斷，而這些共用資源又沒法同時被多個線程訪問的特性。當有線程進入臨界區段時，其餘線程或是進程必須等待。html

在談及鎖的性能開銷，通常都會說鎖的開銷很大，那鎖的開銷有多大，主要耗在哪，怎麼提升鎖的性能。linux

鎖的開銷

如今鎖的機制通常使用 futex（fast Userspace mutexes），內核態和用戶態的混合機制。尚未futex的時候，內核是如何維護同步與互斥的呢？系統內核維護一個對象，這個對象對全部進程可見，這個對象是用來管理互斥鎖而且通知阻塞的進程。若是進程A要進入臨界區，先去內核查看這個對象，有沒有別的進程在佔用這個臨界區，出臨界區的時候，也去內核查看這個對象，有沒有別的進程在等待進入臨界區，而後根據必定的策略喚醒等待的進程。這些沒必要要的系統調用(或者說內核陷入)形成了大量的性能開銷。爲了解決這個問題，Futex就應運而生。git

Futex是一種用戶態和內核態混合的同步機制。首先，同步的進程間經過mmap共享一段內存，futex變量就位於這段共享的內存中且操做是原子的，當進程嘗試進入互斥區或者退出互斥區的時候，先去查看共享內存中的futex變量，若是沒有競爭發生，則只修改futex,而不用再執行系統調用了。當經過訪問futex變量告訴進程有競爭發生，則仍是得執行系統調用去完成相應的處理(wait 或者 wake up)。簡單的說，futex就是經過在用戶態的檢查，（motivation）若是瞭解到沒有競爭就不用陷入內核了，大大提升了low-contention時候的效率。github

mutex 是在 futex 的基礎上用的內存共享變量來實現的，若是共享變量創建在進程內，它就是一個線程鎖，若是它創建在進程間共享內存上，那麼它是一個進程鎖。pthread_mutex_t 中的 _lock 字段用於標記佔用狀況，先使用CAS判斷_lock是否佔用，若未佔用，直接返回。不然，經過__lll_lock_wait_private 調用SYS_futex 系統調用迫使線程進入沉睡。 CAS是用戶態的 CPU 指令，若無競爭，簡單修改鎖狀態即返回，很是高效，只有發現競爭，才經過系統調用陷入內核態。因此，FUTEX是一種用戶態和內核態混合的同步機制，它保證了低競爭狀況下的鎖獲取效率。redis

因此若是鎖不存在衝突，每次得到鎖和釋放鎖的處理器開銷僅僅是CAS指令的開銷。算法

肯定一件事情最好的方法是實際測試和觀測它，讓咱們寫一段代碼來測試無衝突時鎖的開銷：shell

#include <pthread.h>
#include <stdlib.h>
#include <stdio.h>
#include <time.h>

static inline long long unsigned time_ns(struct timespec* const ts) {
  if (clock_gettime(CLOCK_REALTIME, ts)) {
    exit(1);
  }
  return ((long long unsigned) ts->tv_sec) * 1000000000LLU
    + (long long unsigned) ts->tv_nsec;
}


int main()
{
    int res = -1;
    pthread_mutex_t mutex;

    //初始化互斥量，使用默認的互斥量屬性
    res = pthread_mutex_init(&mutex, NULL);
    if(res != 0)
    {
        perror("pthread_mutex_init failed\n");
        exit(EXIT_FAILURE);
    }

    long MAX = 1000000000;
    long c = 0;
    struct timespec ts;

    const long long unsigned start_ns = time_ns(&ts);

    while(c < MAX) 
    {
        pthread_mutex_lock(&mutex);
        c = c + 1;
        pthread_mutex_unlock(&mutex);
    }

    const long long unsigned delta = time_ns(&ts) - start_ns;

    printf("%f\n", delta/(double)MAX);

    return 0;
}

說明：如下性能測試在騰訊雲 Intel(R) Xeon(R) CPU E5-26xx v4 1核 2399.996MHz 下進行。數據庫

運行了 10 億次，平攤到每次加鎖/解鎖操做大概是 2.2ns 每次加鎖/解鎖（扣除了循環耗時 2.7ns）編程

在鎖衝突的狀況下，開銷就沒有這麼小了。windows

首先pthread_mutex_lock會真正的調用sys_futex來進入內核來試圖加鎖，被鎖住之後線程會進入睡眠，這帶來了上下文切換和線程調度的開銷。

能夠寫兩個互相解鎖的線程來測試這個過程的開銷：

// Copyright (C) 2010 Benoit Sigoure
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.

#include <pthread.h>
#include <sched.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <sys/syscall.h>
#include <sys/wait.h>
#include <time.h>
#include <unistd.h>

#include <linux/futex.h>

static inline long long unsigned time_ns(struct timespec* const ts) {
  if (clock_gettime(CLOCK_REALTIME, ts)) {
    exit(1);
  }
  return ((long long unsigned) ts->tv_sec) * 1000000000LLU
    + (long long unsigned) ts->tv_nsec;
}

static const int iterations = 500000;

static void* thread(void* restrict ftx) {
  int* futex = (int*) ftx;
  for (int i = 0; i < iterations; i++) {
    sched_yield();
    while (syscall(SYS_futex, futex, FUTEX_WAIT, 0xA, NULL, NULL, 42)) {
      // retry
      sched_yield();
    }
    *futex = 0xB;
    while (!syscall(SYS_futex, futex, FUTEX_WAKE, 1, NULL, NULL, 42)) {
      // retry
      sched_yield();
    }
  }
  return NULL;
}

int main(void) {
  struct timespec ts;
  const int shm_id = shmget(IPC_PRIVATE, sizeof (int), IPC_CREAT | 0666);
  int* futex = shmat(shm_id, NULL, 0);
  pthread_t thd;
  if (pthread_create(&thd, NULL, thread, futex)) {
    return 1;
  }
  *futex = 0xA;

  const long long unsigned start_ns = time_ns(&ts);
  for (int i = 0; i < iterations; i++) {
    *futex = 0xA;
    while (!syscall(SYS_futex, futex, FUTEX_WAKE, 1, NULL, NULL, 42)) {
      // retry
      sched_yield();
    }
    sched_yield();
    while (syscall(SYS_futex, futex, FUTEX_WAIT, 0xB, NULL, NULL, 42)) {
      // retry
      sched_yield();
    }
  }
  const long long unsigned delta = time_ns(&ts) - start_ns;

  const int nswitches = iterations << 2;
  printf("%i thread context switches in %lluns (%.1fns/ctxsw)\n",
         nswitches, delta, (delta / (float) nswitches));
  wait(futex);
  return 0;
}

編譯使用 gcc -std=gnu99 -pthread context_switch.c。

運行的結果是 2003.4ns/ctxsw，因此鎖衝突的開銷大概是不衝突開銷的 910 倍了，相差出乎意料的大。

另一個c程序能夠用來測試「純上下文切換」的開銷，線程只是使用sched_yield來放棄處理器，並不進入睡眠。

// Copyright (C) 2010 Benoit Sigoure
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.

#include <sched.h>
#include <pthread.h>
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <string.h>
#include <errno.h>

static inline long long unsigned time_ns(struct timespec* const ts) {
  if (clock_gettime(CLOCK_REALTIME, ts)) {
    exit(1);
  }
  return ((long long unsigned) ts->tv_sec) * 1000000000LLU
    + (long long unsigned) ts->tv_nsec;
}

static const int iterations = 500000;

static void* thread(void*ctx) {
  (void)ctx;
  for (int i = 0; i < iterations; i++)
      sched_yield();
  return NULL;
}

int main(void) {
  struct sched_param param;
  param.sched_priority = 1;
  if (sched_setscheduler(getpid(), SCHED_FIFO, &param))
    fprintf(stderr, "sched_setscheduler(): %s\n", strerror(errno));

  struct timespec ts;
  pthread_t thd;
  if (pthread_create(&thd, NULL, thread, NULL)) {
    return 1;
  }

  long long unsigned start_ns = time_ns(&ts);
  for (int i = 0; i < iterations; i++)
      sched_yield();
  long long unsigned delta = time_ns(&ts) - start_ns;

  const int nswitches = iterations << 2;
  printf("%i thread context switches in %lluns (%.1fns/ctxsw)\n",
         nswitches, delta, (delta / (float) nswitches));
  return 0;
}

「純上下文切換」消耗了大概381.2ns/ctxsw。

這樣咱們大體能夠把鎖衝突的開銷分紅三部分，「純上下文切換」開銷，大概是 381.2ns，調度器開銷（把線程從睡眠變成就緒或者反過來）大概是1622.2ns，在多核系統上，還存在跨處理器調度的開銷，那部分開銷很大。在真實的應用場景裏，還要考慮上下文切換帶來的cache不命中和TLB不命中的開銷，開銷只會進一步加大。

鎖的優化

從上面能夠知道，真正消耗時間的不是上鎖的次數，而是鎖衝突的次數。減小鎖衝突的次數纔是提高性能的關鍵。使用更細粒度的鎖，能夠減小鎖衝突。這裏說的粒度包括時間和空間，好比哈希表包含一系列哈希桶，爲每一個桶設置一把鎖，空間粒度就會小不少－－哈希值相互不衝突的訪問不會致使鎖衝突，這比爲整個哈希表維護一把鎖的衝突機率低不少。減小時間粒度也很容易理解，加鎖的範圍只包含必要的代碼段，儘可能縮短得到鎖到釋放鎖之間的時間，最重要的是，絕對不要在鎖中進行任何可能會阻塞的操做。使用讀寫鎖也是一個很好的減小衝突的方式，讀操做之間不互斥，大大減小了衝突。

假設單向鏈表中的插入/刪除操做不多，主要操做是搜索，那麼基於單一鎖的方法性能會不好。在這種狀況下，應該考慮使用讀寫鎖，即 pthread_rwlock_t，這麼作就容許多個線程同時搜索鏈表。插入和刪除操做仍然會鎖住整個鏈表。假設執行的插入和搜索操做數量差很少相同，可是刪除操做不多，那麼在插入期間鎖住整個鏈表是不合適的，在這種狀況下，最好容許在鏈表中的分離點（disjoint point）上執行併發插入，一樣使用基於讀寫鎖的方式。在兩個級別上執行鎖定，鏈表有一個讀寫鎖，各個節點包含一個互斥鎖，在插入期間，寫線程在鏈表上創建讀鎖，而後繼續處理。在插入數據以前，鎖住要在其後添加新數據的節點，插入以後釋放此節點，而後釋放讀寫鎖。刪除操做在鏈表上創建寫鎖。不須要得到與節點相關的鎖；互斥鎖只創建在某一個操做節點之上，大大減小鎖衝突的次數。

鎖自己的行爲也存在進一步優化的可能性，sys_futex系統調用的做用在於讓被鎖住的當前線程睡眠，讓出處理器供其它線程使用，既然這個過程的消耗很高，也就是說若是被鎖定的時間不超過這個數值的話，根本沒有必要進內核加鎖——釋放的處理器時間還不夠消耗的。sys_futex的時間消耗夠跑不少次 CAS 的，也就是說，對於一個鎖衝突比較頻繁並且平均鎖定時間比較短的系統，一個值得考慮的優化方式是先循環調用 CAS 來嘗試得到鎖（這個操做也被稱做自旋鎖），在若干次失敗後再進入內核真正加鎖。固然這個優化只能在多處理器的系統裏起做用（得有另外一個處理器來解鎖，不然自旋鎖無心義）。在glibc的pthread實現裏，經過對pthread_mutex設置PTHREAD_MUTEX_ADAPTIVE_NP屬性就可使用這個機制。

CAS

鎖產生的一些問題：

等待互斥鎖會消耗寶貴的時間，鎖的開銷很大。
低優先級的線程能夠得到互斥鎖，所以阻礙須要同一互斥鎖的高優先級線程。這個問題稱爲優先級倒置（priority inversion ）
可能由於分配的時間片結束，持有互斥鎖的線程被取消調度。這對於等待同一互斥鎖的其餘線程有不利影響，由於等待時間如今會更長。這個問題稱爲鎖護送（lock convoying）

無鎖編程的好處之一是一個線程被掛起，不會影響到另外一個線程的執行，避免鎖護送；在鎖衝突頻繁且平均鎖定時間較短的系統，避免上下文切換和調度開銷。

CAS (comapre and swap 或者 check and set)，比較並替換，引用 wiki，它是一種用於線程數據同步的原子指令。

CAS 核心算法涉及到三個參數，即內存值，更新值和指望值；CAS 指令會先檢查一個內存位置是否包含預期的值；若是是這樣，就把新的值複製到這個位置，返回 true；若是不是則返回 false。
CAS 對應一條彙編指令 CMPXCHG，所以是原子性的。

bool compare_and_swap (int *accum, int *dest, int newval)
{
  if ( *accum == *dest ) {
      *dest = newval;
      return true;
  }
  return false;
}

通常，程序會在循環裏使用 CAS 不斷去完成一個事務性的操做，通常包含拷貝一個共享的變量到一個局部變量，而後再使用這個局部變量執行任務計算獲得新的值，最後再使用 CAS 比較保存再局部變量的舊值和內存值來嘗試提交你的修改，若是嘗試失敗，會從新讀取一遍內存值，再從新計算，最後再使用 CAS 嘗試提交修改，如此循環。好比：

void LockFreeQueue::push(Node* newHead)
{
    for (;;)
    {
        // 拷貝共享變量(m_Head) 到一個局部變量
        Node* oldHead = m_Head;

        // 執行任務，能夠不用關注其餘線程
        newHead->next = oldHead;

        // 下一步嘗試提交更改到共享變量
        // 若是共享變量沒有被其餘線程修改過，仍爲 oldHead，則 CAS 將 newHead 賦值給共享變量 m_Head 並返回
        // 不然繼續循環重試
        if (_InterlockedCompareExchange(&m_Head, newHead, oldHead))
            return;
    }
}

上面的數據結構設置了一個共享的頭節點 m_Head，當 push 一個新的節點時，會把新節點加在頭節點後面；不要相信程序的執行是連續的，CPU 的執行是多線程併發。在 _InterlockedCompareExchange 即 CAS 以前，線程可能由於時間片用完被調度出去，新調度進來的線程執行完了 push 操做，多個線程共享了 m_Head 變量，此時 m_Head 已被修改了，若是原來線程繼續執行，把 oldHead 覆蓋到 m_Head，就會丟失其餘線程 push 進來的節點。因此須要比較 m_Head 是否是還等於 oldHead，若是是，說明頭節點不變，可使用 newHead 覆蓋 m_Head；若是不是，說明有其餘線程 push 了新的節點，那麼須要使用最新的 m_Head 更新 oldHead 的值從新走一下循環，_InterlockedCompareExchange 會自動把 m_Head 賦值給 oldHead。

ABA 問題

由於 CAS 須要在提交修改時檢查指望值和內存值有沒有發生變化，若是沒有則進行更新，可是若是原來一個值從 A 變成 B 又變成 A，那麼使用 CAS 檢查的時候發現值沒有發生變化，但實際上已經發生了一系列變化。

內存的回收利用會致使 CAS 出現嚴重的問題：

T* ptr1 = new T(8, 18);
T* old = ptr1; 
delete ptr1;
T* ptr2 = new T(0, 1);
 
// 咱們不能保證操做系統不會從新使用 ptr1 內存地址，通常的內存管理器都會這樣子作
if (old1 == ptr2) {
    // 這裏表示，剛剛回收的 ptr1 指向的內存被用於後面申請的 ptr2了
}

ABA問題是無鎖結構實現中常見的一種問題，可基本表述爲：

進程P1讀取了一個數值A
P1被掛起(時間片耗盡、中斷等)，進程P2開始執行
P2修改數值A爲數值B，而後又修改回A
P1被喚醒，比較後發現數值A沒有變化，程序繼續執行。

對於P1來講，數值A未發生過改變，但實際上A已經被變化過了，繼續使用可能會出現問題。在CAS操做中，因爲比較的可能是指針，這個問題將會變得更加嚴重。試想以下狀況：

有一個堆(先入後出)中有top和節點A，節點A目前位於堆頂top指針指向A。如今有一個進程P1想要pop一個節點，所以按照以下無鎖操做進行

pop()
{
  do{
    ptr = top; // ptr = top = NodeA
    next_prt = top->next; // next_ptr = NodeX
  } while(CAS(top, ptr, next_ptr) != true);
  return ptr;   
}

而進程P2在執行CAS操做以前打斷了P1，並對堆進行了一系列的pop和push操做，使堆變爲以下結構：

進程P2首先pop出NodeA，以後又Push了兩個NodeB和C，因爲內存管理機制中普遍使用的內存重用機制，致使NodeC的地址與以前的NodeA一致。

這時P1又開始繼續運行，在執行CAS操做時，因爲top依舊指向的是NodeA的地址(實際上已經變爲NodeC)，所以將top的值修改成了NodeX，這時堆結構以下：

通過CAS操做後，top指針錯誤的指向了NodeX而不是NodeB。

ABA 解決方法

Tagged state reference，增長額外的 tag bits 位，它像一個版本號；好比，其中一種算法是在內存地址的低位記錄指針的修改次數，在指針修改時，下一次 CAS 會返回失敗，即便由於內存重用機制致使地址同樣。有時咱們稱這種機制位 ABA‘，由於咱們使第二個 A 稍微有點不一樣於第一個。tag 的位數長度會影響記錄修改的次數，在現有的 CPU 下，使用 60 bit tag，在不重啓程序10年纔會產生溢出問題；在 X64 CPU，趨向於支持 128 bit 的 CAS 指令，這樣更能保證避免出現 ABA 問題。

下面參考 liblfds 庫代碼說明下 Tagged state reference 的實現過程。

咱們想要避免 ABA 問題的方法之一是使用更長的指針，這樣便須要一個支持 dword 長度的 CAS 指令。liblfds 是怎麼跨平臺實現 128 bit 指令的呢？

在 liblfds 下，CAS 指令爲 LFDS710_PAL_ATOMIC_DWCAS 宏，它的完整形式是：

LFDS710_PAL_ATOMIC_DWCAS( pointer_to_destination, pointer_to_compare, pointer_to_new_destination, cas_strength, result)

pointer_to_destination: [in, out]，指向目標的指針，是一個由兩個 64 bit 整數組成的數組；
pointer_to_compare: [in, out]，用於和目標指針比較的指針，一樣是一個由兩個 64 bit 整數組成的數組；
pointer_to_new_destination: [in]，和目標指針交換的新指針；
result: [out]，若是 128 bit 的 pointer_to_compare 與 pointer_to_destination 相等，則使用 pointer_to_new_destination 覆蓋 pointer_to_destination，result 返回 1；若是不相等，則 pointer_to_destination 不變，且 pointer_to_compare 的值變爲 pointer_to_destination。

從上面能夠看出，liblfds 庫使用一個由兩個元素組成的一維數組來表示 128 bit 指針。

Linux 提供了 cmpxchg16b 用於實現 128 bit 的 CAS 指令，而在 Windows，使用 _InterlockedCompareExchange128。只有 128 位指針徹底相等的狀況下，才視爲相等。

參考 liblfds/liblfds7.1.0/liblfds710/inc/liblfds710/lfds710_porting_abstraction_layer_compiler.h 下關於 CAS 的 windows 實現：

#define LFDS710_PAL_ATOMIC_DWCAS( pointer_to_destination, pointer_to_compare, pointer_to_new_destination, cas_strength, result ) \
{ \                                              
    LFDS710_PAL_BARRIER_COMPILER_FULL; \
    (result) = (char unsigned) _InterlockedCompareExchange128( (__int64 volatile *) (pointer_to_destination), (__int64) (pointer_to_new_destination[1]), (__int64) (pointer_to_new_destination[0]), (__int64 *) (pointer_to_compare) ); \
    LFDS710_PAL_BARRIER_COMPILER_FULL; \
}

再重點研究 new_top 的定義和提交修改過程。

new_top 是一個具備兩個元素的一維數組，元素是 struct lfds710_stack_element 指針，兩個元素分別使用 POINTER 0 和 COUNTER 1 標記。COUNTER 至關於前面說的 tag 標記，POINTER 保存的時真正的節點指針。在 X64 下，指針長度是 64 bit，因此這裏使用的是 64 bit tag 記錄 pointer 修改記錄。

liblfds 用原 top 的 COUNTER + 1來初始化 new top COUNTER，即便用 COUNTER 標記 ss->top 的更換次數，這樣每一次更換 top，top 裏的 COUNTER 都會變。

只有在 ss->top 和 original_top 的 POINTER 和 COUNTER 徹底相等的狀況下，new_top 纔會覆蓋到 ss->top，不然會使用 ss->top 覆蓋 original_top，下次循環用最新的 original_top 再次操做和比較。

參考 liblfds/liblfds7.1.0/liblfds710/src/lfds710_stack/lfds710_stack_push.c，無鎖堆棧的實現：

void lfds710_stack_push( struct lfds710_stack_state *ss,
                         struct lfds710_stack_element *se )
{
  char unsigned
    result;

  lfds710_pal_uint_t
    backoff_iteration = LFDS710_BACKOFF_INITIAL_VALUE;

  struct lfds710_stack_element LFDS710_PAL_ALIGN(LFDS710_PAL_ALIGN_DOUBLE_POINTER)
    *new_top[PAC_SIZE],
    *volatile original_top[PAC_SIZE];

  LFDS710_PAL_ASSERT( ss != NULL );
  LFDS710_PAL_ASSERT( se != NULL );

  new_top[POINTER] = se;

  original_top[COUNTER] = ss->top[COUNTER];
  original_top[POINTER] = ss->top[POINTER];

  do
  {
    se->next = original_top[POINTER];
    LFDS710_MISC_BARRIER_STORE;

    new_top[COUNTER] = original_top[COUNTER] + 1;
    LFDS710_PAL_ATOMIC_DWCAS( ss->top, original_top, new_top, LFDS710_MISC_CAS_STRENGTH_WEAK, result );

    if( result == 0 )
      LFDS710_BACKOFF_EXPONENTIAL_BACKOFF( ss->push_backoff, backoff_iteration );
  }
  while( result == 0 );

  LFDS710_BACKOFF_AUTOTUNE( ss->push_backoff, backoff_iteration );

  return;
}