Linux內核同步機制--轉發自蝸窩科技

時間 2019-11-19

標籤 linux 內核同步機制轉發科技欄目 Linux 简体版

原文原文鏈接

Linux內核同步機制之（一）：原子操做

http://www.wowotech.net/linux_kenrel/atomic.html html

1、源由 node

咱們的程序邏輯常常遇到這樣的操做序列： linux

一、讀一個位於memory中的變量的值到寄存器中 android

二、修改該變量的值（也就是修改寄存器中的值）程序員

三、將寄存器中的數值寫回memory中的變量值 es6

若是這個操做序列是串行化的操做（在一個thread中串行執行），那麼一切OK，然而，世界老是不能如你所願。在多CPU體系結構中，運行在兩個CPU上的兩個內核控制路徑同時並行執行上面操做序列，有可能發生下面的場景：編程

CPU1上的操做	CPU2上的操做
讀操做
	讀操做
修改	修改
寫操做
	寫操做

多個CPUs和memory chip是經過總線互聯的，在任意時刻，只能有一個總線master設備（例如CPU、DMA controller）訪問該Slave設備（在這個場景中，slave設備是RAM chip）。所以，來自兩個CPU上的讀memory操做被串行化執行，分別得到了一樣的舊值。完成修改後，兩個CPU都想進行寫操做，把修改的值寫回到memory。可是，硬件arbiter的限制使得CPU的寫回必須是串行化的，所以CPU1首先得到了訪問權，進行寫回動做，隨後，CPU2完成寫回動做。在這種狀況下，CPU1的對memory的修改被CPU2的操做覆蓋了，所以執行結果是錯誤的。 api

不只是多CPU，在單CPU上也會因爲有多個內核控制路徑的交錯而致使上面描述的錯誤。一個具體的例子以下：數組

系統調用的控制路徑	中斷handler控制路徑
讀操做
	讀操做
	修改
	寫操做
修改
寫操做

系統調用的控制路徑上，完成讀操做後，硬件觸發中斷，開始執行中斷handler。這種場景下，中斷handler控制路徑的寫回的操做被系統調用控制路徑上的寫回覆蓋了，結果也是錯誤的。安全

2、對策

對於那些有多個內核控制路徑進行read-modify-write的變量，內核提供了一個特殊的類型atomic_t，具體定義以下：

typedef struct {
int counter;
} atomic_t;

從上面的定義來看，atomic_t實際上就是一個int類型的counter，不過定義這樣特殊的類型atomic_t是有其思考的：內核定義了若干atomic_xxx的接口API函數，這些函數只會接收atomic_t類型的參數。這樣能夠確保atomic_xxx的接口函數只會操做atomic_t類型的數據。一樣的，若是你定義了atomic_t類型的變量（你指望用atomic_xxx的接口API函數操做它），這些變量也不會被那些普通的、非原子變量操做的API函數接受。

具體的接口API函數整理以下：

接口函數	描述
static inline void atomic_add(int i, atomic_t *v)	給一個原子變量v增長i
static inline int atomic_add_return(int i, atomic_t *v)	同上，只不過將變量v的最新值返回
static inline void atomic_sub(int i, atomic_t *v)	給一個原子變量v減去i
static inline int atomic_sub_return(int i, atomic_t *v)	同上，只不過將變量v的最新值返回
static inline int atomic_cmpxchg(atomic_t *ptr, int old, int new)	比較old和原子變量ptr中的值，若是相等，那麼就把new值賦給原子變量。返回舊的原子變量ptr中的值
atomic_read	獲取原子變量的值
atomic_set	設定原子變量的值
atomic_inc(v)	原子變量的值加一
atomic_inc_return(v)	同上，只不過將變量v的最新值返回
atomic_dec(v)	原子變量的值減去一
atomic_dec_return(v)	同上，只不過將變量v的最新值返回
atomic_sub_and_test(i, v)	給一個原子變量v減去i，並判斷變量v的最新值是否等於0
atomic_add_negative(i,v)	給一個原子變量v增長i，並判斷變量v的最新值是不是負數
static inline int atomic_add_unless(atomic_t *v, int a, int u)	只要原子變量v不等於u，那麼就執行原子變量v加a的操做。若是v不等於u，返回非0值，不然返回0值

3、ARM中的實現

咱們以atomic_add爲例，描述linux kernel中原子操做的具體代碼實現細節：

#if __LINUX_ARM_ARCH__ >= 6 －－－－－－－－－－－－－－－－－－－－－－（1）
static inline void atomic_add(int i, atomic_t *v)
{
    unsigned long tmp;
    int result;

    prefetchw(&v->counter); －－－－－－－－－－－－－－－－－－－－－－－－－（2）
    __asm__ __volatile__("@ atomic_add\n" －－－－－－－－－－－－－－－－－－（3）
"1:    ldrex    %0, [%3]\n" －－－－－－－－－－－－－－－－－－－－－－－－－－（4）
"    add    %0, %0, %4\n" －－－－－－－－－－－－－－－－－－－－－－－－－－（5）
"    strex    %1, %0, [%3]\n" －－－－－－－－－－－－－－－－－－－－－－－－－（6）
"    teq    %1, #0\n" －－－－－－－－－－－－－－－－－－－－－－－－－－－－－（7）
"    bne    1b"
    : "=&r" (result), "=&r" (tmp), "+Qo" (v->counter) －－－對應％0，％1，％2
    : "r" (&v->counter), "Ir" (i) －－－－－－－－－－－－－對應％3，％4
    : "cc");
}

#else

#ifdef CONFIG_SMP
#error SMP not supported on pre-ARMv6 CPUs
#endif

static inline int atomic_add_return(int i, atomic_t *v)
{
    unsigned long flags;
    int val;

    raw_local_irq_save(flags);
    val = v->counter;
    v->counter = val += i;
    raw_local_irq_restore(flags);

    return val;
}
#define atomic_add(i, v)    (void) atomic_add_return(i, v)

#endif

（1）ARMv6以前的CPU並不支持SMP，以後的ARM架構都是支持SMP的（例如咱們熟悉的ARMv7-A）。所以，對於ARM處理，其原子操做分紅了兩個陣營，一個是支持SMP的ARMv6以後的CPU，另一個就是ARMv6以前的，只有單核架構的CPU。對於UP，原子操做就是經過關閉CPU中斷來完成的。

（2）這裏的代碼和preloading cache相關。在strex指令以前將要操做的memory內容加載到cache中能夠顯著提升性能。

（3）爲了完整性，我仍是重複一下彙編嵌入c代碼的語法：嵌入式彙編的語法格式是：asm(code : output operand list : input operand list : clobber list)。output operand list 和 input operand list是c代碼和嵌入式彙編代碼的接口，clobber list描述了彙編代碼對寄存器的修改狀況。爲什麼要有clober list？咱們的c代碼是gcc來處理的，當遇到嵌入彙編代碼的時候，gcc會將這些嵌入式彙編的文本送給gas進行後續處理。這樣，gcc須要瞭解嵌入彙編代碼對寄存器的修改狀況，不然有可能會形成大麻煩。例如：gcc對c代碼進行處理，將某些變量值保存在寄存器中，若是嵌入彙編修改了該寄存器的值，又沒有通知gcc的話，那麼，gcc會覺得寄存器中仍然保存了以前的變量值，所以不會從新加載該變量到寄存器，而是直接使用這個被嵌入式彙編修改的寄存器，這時候，咱們惟一能作的就是靜靜的等待程序的崩潰。還好，在output operand list 和 input operand list中涉及的寄存器都不須要體如今clobber list中（gcc分配了這些寄存器，固然知道嵌入彙編代碼會修改其內容），所以，大部分的嵌入式彙編的clobber list都是空的，或者只有一個cc，通知gcc，嵌入式彙編代碼更新了condition code register。

你們對着上面的code就能夠分開各段內容了。@符號標識該行是註釋。

這裏的__volatile__主要是用來防止編譯器優化的。也就是說，在編譯該c代碼的時候，若是使用優化選項（-O）進行編譯，對於那些沒有聲明__volatile__的嵌入式彙編，編譯器有可能會對嵌入c代碼的彙編進行優化，編譯的結果可能不是原來你撰寫的彙編代碼，可是若是你的嵌入式彙編使用__asm__ __volatile__(嵌入式彙編)的語法格式，那麼也就是告訴編譯器，不要隨便動個人嵌入彙編代碼哦。

（4）咱們先看ldrex和strex這兩條彙編指令的使用方法。ldr和str這兩條指令你們都是很是的熟悉了，後綴的ex表示Exclusive，是ARMv7提供的爲了實現同步的彙編指令。

LDREX <Rt>, [<Rn>]
<Rn>是base register，保存memory的address，LDREX指令從base register中獲取memory address，而且將memory的內容加載到<Rt>(destination register)中。這些操做和ldr的操做是同樣的，那麼如何體現exclusive呢？其實，在執行這條指令的時候，還放出兩條「狗」來負責觀察特定地址的訪問（就是保存在[<Rn>]中的地址了），這兩條狗一條叫作local monitor，一條叫作global monitor。
STREX <Rd>, <Rt>, [<Rn>]
和LDREX指令相似，<Rn>是base register，保存memory的address，STREX指令從base register中獲取memory address，而且將<Rt> (source register)中的內容加載到該memory中。這裏的<Rd>保存了memeory 更新成功或者失敗的結果，0表示memory更新成功，1表示失敗。STREX指令是否能成功執行是和local monitor和global monitor的狀態相關的。對於Non-shareable memory（該memory不是多個CPU之間共享的，只會被一個CPU訪問），只須要放出該CPU的local monitor這條狗就OK了，下面的表格能夠描述這種狀況

thread 1	thread 2	local monitor的狀態
		Open Access state
LDREX		Exclusive Access state
	LDREX	Exclusive Access state
	Modify	Exclusive Access state
	STREX	Open Access state
Modify		Open Access state
STREX		在Open Access state的狀態下，執行STREX指令會致使該指令執行失敗
		保持Open Access state，直到下一個LDREX指令

開始的時候，local monitor處於Open Access state的狀態，thread 1執行LDREX 命令後，local monitor的狀態遷移到Exclusive Access state（標記本地CPU對xxx地址進行了LDREX的操做），這時候，中斷髮生了，在中斷handler中，又一次執行了LDREX ，這時候，local monitor的狀態保持不變，直到STREX指令成功執行，local monitor的狀態遷移到Open Access state的狀態（清除xxx地址上的LDREX的標記）。返回thread 1的時候，在Open Access state的狀態下，執行STREX指令會致使該指令執行失敗（沒有LDREX的標記，何來STREX），說明有其餘的內核控制路徑插入了。

對於shareable memory，須要系統中全部的local monitor和global monitor共同工做，完成exclusive access，概念相似，這裏就再也不贅述了。

大概的原理已經描述完畢，下面回到具體實現面。

"1: ldrex %0, [%3]\n"

其中％3就是input operand list中的"r" (&v->counter)，r是限制符（constraint），用來告訴編譯器gcc，你看着辦吧，你幫我選擇一個通用寄存器保存該操做數吧。％0對應output openrand list中的"=&r" (result)，=表示該操做數是write only的，&表示該操做數是一個earlyclobber operand，具體是什麼意思呢？編譯器在處理嵌入式彙編的時候，傾向使用盡量少的寄存器，若是output operand沒有&修飾的話，彙編指令中的input和output操做數會使用一樣一個寄存器。所以，&確保了％3和％0使用不一樣的寄存器。

（5）完成步驟（4）後，％0這個output操做數已經被賦值爲atomic_t變量的old value，毫無疑問，這裏的操做是要給old value加上i。這裏％4對應"Ir" (i)，這裏「I」這個限制符對應ARM平臺，表示這是一個有特定限制的當即數，該數必須是0～255之間的一個整數經過rotation的操做獲得的一個32bit的當即數。這是和ARM的data-processing instructions如何解析當即數有關的。每一個指令32個bit，其中12個bit被用來表示當即數，其中8個bit是真正的數據，4個bit用來表示如何rotation。更詳細的內容請參考ARM ARM文檔。

（6）這一步將修改後的new value保存在atomic_t變量中。是否可以正確的操做的狀態標記保存在％1操做數中，也就是"=&r" (tmp)。

（7）檢查memory update的操做是否正確完成，若是OK，皆大歡喜，若是發生了問題（有其餘的內核路徑插入），那麼須要跳轉到lable 1那裏，重新進行一次read-modify-write的操做。

Linux內核同步機制之（二）：Per-CPU變量

http://www.wowotech.net/linux_kenrel/per-cpu.html

1、源由：爲什麼引入Per-CPU變量？

一、lock bus帶來的性能問題

在ARM平臺上，ARMv6以前，SWP和SWPB指令被用來支持對shared memory的訪問：

SWP <Rt>, <Rt2>, [<Rn>]

Rn中保存了SWP指令要操做的內存地址，經過該指令能夠將Rn指定的內存數據加載到Rt寄存器，同時將Rt2寄存器中的數值保存到Rn指定的內存中去。

咱們在原子操做那篇文檔中描述的read-modify-write的問題本質上是一個保持對內存read和write訪問的原子性的問題。也就是說對內存的讀和寫的訪問不能被打斷。對該問題的解決能夠經過硬件、軟件或者軟硬件結合的方法來進行。早期的ARM CPU給出的方案就是依賴硬件：SWP這個彙編指令執行了一次讀內存操做、一次寫內存操做，可是從程序員的角度看，SWP這條指令就是原子的，讀寫之間不會被任何的異步事件打斷。具體底層的硬件是如何作的呢？這時候，硬件會提供一個lock signal，在進行memory操做的時候設定lock信號，告訴總線這是一個不可被中斷的內存訪問，直到完成了SWP須要進行的兩次內存訪問以後再clear lock信號。

lock memory bus對多核系統的性能形成嚴重的影響（系統中其餘的processor對那條被lock的memory bus的訪問就被hold住了），如何解決這個問題？最好的鎖機制就是不使用鎖，所以解決這個問題能夠使用釜底抽薪的方法，那就是不在系統中的多個processor之間共享數據，給每個CPU分配一個不就OK了嗎。

固然，隨着技術的發展，在ARMv6以後的ARM CPU已經不推薦使用SWP這樣的指令，而是提供了LDREX和STREX這樣的指令。這種方法是使用軟硬件結合的方法來解決原子操做問題，看起來代碼比較複雜，可是系統的性能能夠獲得提高。其實，從硬件角度看，LDREX和STREX這樣的指令也是採用了lock-free的作法。OK，因爲再也不lock bus，看起來Per-CPU變量存在的基礎被打破了。不過考慮cache的操做，實際上它仍是有意義的。

二、cache的影響

在The Memory Hierarchy文檔中，咱們已經瞭解了關於memory一些基礎的知識，一些基礎的內容，這裏就再也不重複了。咱們假設一個多核系統中的cache以下：

每一個CPU都有本身的L1 cache（包括data cache和instruction cache），全部的CPU共用一個L2 cache。L一、L2以及main memory的訪問速度之間的差別都是很是大，最高的性能的狀況下固然是L1 cache hit，這樣就不須要訪問下一階memory來加載cache line。

咱們首先看在多個CPU之間共享內存的狀況。這種狀況下，任何一個CPU若是修改了共享內存就會致使全部其餘CPU的L1 cache上對應的cache line變成invalid（硬件完成）。雖然對性能形成影響，可是系統必須這麼作，由於須要維持cache的同步。將一個共享memory變成Per-CPU memory本質上是一個耗費更多memory來解決performance的方法。當一個在多個CPU之間共享的變量變成每一個CPU都有屬於本身的一個私有的變量的時候，咱們就沒必要考慮來自多個CPU上的併發，僅僅考慮本CPU上的併發就OK了。固然，還有一點要注意，那就是在訪問Per-CPU變量的時候，不能調度，固然更準確的說法是該task不能調度到其餘CPU上去。目前的內核的作法是在訪問Per-CPU變量的時候disable preemptive，雖然沒有可以徹底避免使用鎖的機制（disable preemptive也是一種鎖的機制），但毫無疑問，這是一種代價比較小的鎖。

2、接口

一、靜態聲明和定義Per-CPU變量的API以下表所示：

聲明和定義Per-CPU變量的API	描述
DECLARE_PER_CPU(type, name) DEFINE_PER_CPU(type, name)	普通的、沒有特殊要求的per cpu變量定義接口函數。沒有對齊的要求
DECLARE_PER_CPU_FIRST(type, name) DEFINE_PER_CPU_FIRST(type, name)	經過該API定義的per cpu變量位於整個per cpu相關section的最前面。
DECLARE_PER_CPU_SHARED_ALIGNED(type, name) DEFINE_PER_CPU_SHARED_ALIGNED(type, name)	經過該API定義的per cpu變量在SMP的狀況下會對齊到L1 cache line ，對於UP，不須要對齊到cachine line
DECLARE_PER_CPU_ALIGNED(type, name) DEFINE_PER_CPU_ALIGNED(type, name)	不管SMP或者UP，都是須要對齊到L1 cache line
DECLARE_PER_CPU_PAGE_ALIGNED(type, name) DEFINE_PER_CPU_PAGE_ALIGNED(type, name)	爲定義page aligned per cpu變量而設定的API接口
DECLARE_PER_CPU_READ_MOSTLY(type, name) DEFINE_PER_CPU_READ_MOSTLY(type, name)	經過該API定義的per cpu變量是read mostly的

看到這樣「豐富多彩」的Per-CPU變量的API，你是否是已經醉了。這些定義使用在不一樣的場合，主要的factor包括：

－該變量在section中的位置

－該變量的對齊方式

－該變量對SMP和UP的處理不一樣

－訪問per cpu的形態

例如：若是你準備定義的per cpu變量是要求按照page對齊的，那麼在定義該per cpu變量的時候須要使用DECLARE_PER_CPU_PAGE_ALIGNED。若是隻要求在SMP的狀況下對齊到cache line，那麼使用DECLARE_PER_CPU_SHARED_ALIGNED來定義該per cpu變量。

二、訪問靜態聲明和定義Per-CPU變量的API

靜態定義的per cpu變量不能象普通變量那樣進行訪問，須要使用特定的接口函數，具體以下：

get_cpu_var(var)

put_cpu_var(var)

上面這兩個接口函數已經內嵌了鎖的機制（preempt disable），用戶能夠直接調用該接口進行本CPU上該變量副本的訪問。若是用戶確認當前的執行環境已是preempt disable（或者是更厲害的鎖，例如關閉了CPU中斷），那麼能夠使用lock-free版本的Per-CPU變量的API:__get_cpu_var。

三、動態分配Per-CPU變量的API以下表所示：

動態分配和釋放Per-CPU變量的API	描述
alloc_percpu(type)	分配類型是type的per cpu變量，返回per cpu變量的地址（注意：不是各個CPU上的副本）
void free_percpu(void __percpu *ptr)	釋放ptr指向的per cpu變量空間

四、訪問動態分配Per-CPU變量的API以下表所示：

訪問Per-CPU變量的API	描述
get_cpu_ptr	這個接口是和訪問靜態Per-CPU變量的get_cpu_var接口是相似的，固然，這個接口是for 動態分配Per-CPU變量
put_cpu_ptr	同上
per_cpu_ptr(ptr, cpu)	根據per cpu變量的地址和cpu number，返回指定CPU number上該per cpu變量的地址

3、實現

一、靜態Per-CPU變量定義

咱們以DEFINE_PER_CPU的實現爲例子，描述linux kernel中如何實現靜態Per-CPU變量定義。具體代碼以下：

#define DEFINE_PER_CPU(type, name) \
DEFINE_PER_CPU_SECTION(type, name, "")

type就是變量的類型，name是per cpu變量符號。DEFINE_PER_CPU_SECTION宏能夠把一個per cpu變量放到指定的section中，具體代碼以下：

#define DEFINE_PER_CPU_SECTION(type, name, sec)                \
    __PCPU_ATTRS(sec) PER_CPU_DEF_ATTRIBUTES            \－－－－－安排section
    __typeof__(type) name－－－－－－－－－－－－－－－－－－－－－－定義變量

在這裏具體arch specific的percpu代碼中（arch/arm/include/asm/percpu.h）能夠定義PER_CPU_DEF_ATTRIBUTES，以便控制該per cpu變量的屬性，固然，若是arch specific的percpu代碼不定義，那麼在general arch-independent的代碼中（include/asm-generic/percpu.h）會定義爲空。這裏能夠順便提一下Per-CPU變量的軟件層次：

（1）arch-independent interface。在include/linux/percpu.h文件中，定義了內核其餘模塊要使用per cpu機制使用的接口API以及相關數據結構的定義。內核其餘模塊須要使用per cpu變量接口的時候須要include該頭文件

（2）arch-general interface。在include/asm-generic/percpu.h文件中。若是全部的arch相關的定義都是同樣的，那麼就把它抽取出來，放到asm-generic目錄下。毫無疑問，這個文件定義的接口和數據結構是硬件相關的，只不過軟件抽象各個arch-specific的內容，造成一個arch general layer。通常來講，咱們不須要直接include該頭文件，include/linux/percpu.h會include該頭文件。

（3）arch-specific。這是和硬件相關的接口，在arch/arm/include/asm/percpu.h，定義了ARM平臺中，具體和per cpu相關的接口代碼。

咱們回到正題，看看__PCPU_ATTRS的定義：

#define __PCPU_ATTRS(sec)                        \
    __percpu __attribute__((section(PER_CPU_BASE_SECTION sec)))    \
    PER_CPU_ATTRIBUTES

PER_CPU_BASE_SECTION 定義了基礎的section name symbol，定義以下：

#ifndef PER_CPU_BASE_SECTION
#ifdef CONFIG_SMP
#define PER_CPU_BASE_SECTION ".data..percpu"
#else
#define PER_CPU_BASE_SECTION ".data"
#endif
#endif

雖然有各類各樣的靜態Per-CPU變量定義方法，可是都是相似的，只不過是放在不一樣的section中，屬性不一樣而已，這裏就不看其餘的實現了，直接給出section的安排：

（1）普通per cpu變量的section安排

	SMP	UP
Build-in kernel	".data..percpu" section	".data" section
defined in module	".data..percpu" section	".data" section

（2）first per cpu變量的section安排

	SMP	UP
Build-in kernel	".data..percpu..first" section	".data" section
defined in module	".data..percpu..first" section	".data" section

（3）SMP shared aligned per cpu變量的section安排

	SMP	UP
Build-in kernel	".data..percpu..shared_aligned" section	".data" section
defined in module	".data..percpu" section	".data" section

（4）aligned per cpu變量的section安排

	SMP	UP
Build-in kernel	".data..percpu..shared_aligned" section	".data..shared_aligned" section
defined in module	".data..percpu" section	".data..shared_aligned" section

（5）page aligned per cpu變量的section安排

	SMP	UP
Build-in kernel	".data..percpu..page_aligned" section	".data..page_aligned" section
defined in module	".data..percpu..page_aligned" section	".data..page_aligned" section

（6）read mostly per cpu變量的section安排

	SMP	UP
Build-in kernel	".data..percpu..readmostly" section	".data..readmostly" section
defined in module	".data..percpu..readmostly" section	".data..readmostly" section

瞭解了靜態定義Per-CPU變量的實現，可是爲什麼要引入這麼多的section呢？對於kernel中的普通變量，通過了編譯和連接後，會被放置到.data或者.bss段，系統在初始化的時候會準備好一切（例如clear bss），因爲per cpu變量的特殊性，內核將這些變量放置到了其餘的section，位於kernel address space中__per_cpu_start和__per_cpu_end之間，咱們稱之Per-CPU變量的原始變量（我也想不出什麼好詞了）。

只有Per-CPU變量的原始變量仍是不夠的，必須爲每個CPU創建一個副本，怎麼建？直接靜態定義一個NR_CPUS的數組？NR_CPUS定義了系統支持的最大的processor的個數，並非實際中系統processor的數目，這樣的定義很是浪費內存。此外，靜態定義的數據在內存中連續，對於UMA系統而言是OK的，對於NUMA系統，每一個CPU上的Per-CPU變量的副本應該位於它訪問最快的那段memory上，也就是說Per-CPU變量的各個CPU副本多是散佈在整個內存地址空間的，而這些空間之間是有空洞的。本質上，副本per cpu內存的分配歸屬於內存管理子系統，所以，分配per cpu變量副本的內存本文不會詳述，大體的思路以下：

內存管理子系統會根據當前的內存配置爲每個CPU分配一大塊memory，對於UMA，這個memory也是位於main memory，對於NUMA，有多是分配最靠近該CPU的memory（也就是說該cpu訪問這段內存最快），但不管如何，這些都是內存管理子系統須要考慮的。不管靜態仍是動態per cpu變量的分配，其機制都是同樣的，只不過，對於靜態per cpu變量，須要在系統初始化的時候，對應per cpu section，預先動態分配一個一樣size的per cpu chunk。在vmlinux.lds.h文件中，定義了percpu section的排列狀況：

#define PERCPU_INPUT(cacheline)                        \
    VMLINUX_SYMBOL(__per_cpu_start) = .;                \
    *(.data..percpu..first)                        \
    . = ALIGN(PAGE_SIZE);                        \
    *(.data..percpu..page_aligned)                    \
    . = ALIGN(cacheline);                        \
    *(.data..percpu..readmostly)                    \
    . = ALIGN(cacheline);                        \
    *(.data..percpu)                        \
    *(.data..percpu..shared_aligned)                \
    VMLINUX_SYMBOL(__per_cpu_end) = .;

對於build in內核的那些per cpu變量，必然位於__per_cpu_start和__per_cpu_end之間的per cpu section。在系統初始化的時候（setup_per_cpu_areas），分配per cpu memory chunk，並將per cpu section copy到每個chunk中。

二、訪問靜態定義的per cpu變量

代碼以下：

#define get_cpu_var(var) (*({                \
    preempt_disable();                \
    &__get_cpu_var(var); }))

再看到get_cpu_var和__get_cpu_var這兩個符號，相信廣大人民羣衆已經至關的熟悉，一個持有鎖的版本，一個lock-free的版本。爲防止當前task因爲搶佔而調度到其餘的CPU上，在訪問per cpu memory的時候都須要使用preempt_disable這樣的鎖的機制。咱們來看__get_cpu_var：

#define __get_cpu_var(var) (*this_cpu_ptr(&(var)))

#define this_cpu_ptr(ptr) __this_cpu_ptr(ptr)

對於ARM平臺，咱們沒有定義__this_cpu_ptr，所以採用asm-general版本的：

#define __this_cpu_ptr(ptr) SHIFT_PERCPU_PTR(ptr, __my_cpu_offset)

SHIFT_PERCPU_PTR這個宏定義從字面上就能夠看出它是能夠從原始的per cpu變量的地址，經過簡單的變換（SHIFT）轉成實際的per cpu變量副本的地址。實際上，per cpu內存管理模塊能夠保證原始的per cpu變量的地址和各個CPU上的per cpu變量副本的地址有簡單的線性關係（就是一個固定的offset）。__my_cpu_offset這個宏定義就是和offset相關的，若是arch specific沒有定義，那麼能夠採用asm general版本的，以下：

#define __my_cpu_offset per_cpu_offset(raw_smp_processor_id())

raw_smp_processor_id能夠獲取本CPU的ID，若是沒有arch specific沒有定義__per_cpu_offset這個宏，那麼offset保存在__per_cpu_offset的數組中（下面只是數組聲明，具體定義在mm/percpu.c文件中），以下：

#ifndef __per_cpu_offset
extern unsigned long __per_cpu_offset[NR_CPUS];

#define per_cpu_offset(x) (__per_cpu_offset[x])
#endif

對於ARMV6K和ARMv7版本，offset保存在TPIDRPRW寄存器中，這樣是爲了提高系統性能。

三、動態分配per cpu變量

這部份內容留給內存管理子系統吧。

Linux內核同步機制之（三）：memory barrier

http://www.wowotech.net/kernel_synchronization/memory-barrier.html

1、前言

我記得之前上學的時候你們常常說的一個詞彙叫作所見即所得，有些編程工具是所見即所得的，給程序員帶來極大的方便。對於一個c程序員，咱們的編寫的代碼能所見即所得嗎？咱們看到的c程序的邏輯是否就是最後CPU運行的結果呢？很遺憾，不是，咱們的「所見」和最後的執行結果隔着：

一、編譯器

二、CPU取指執行

編譯器將符合人類思考的邏輯（c代碼）翻譯成了符合CPU運算規則的彙編指令，編譯器瞭解底層CPU的思惟模式，所以，它能夠在將c翻譯成彙編的時候進行優化（例如內存訪問指令的從新排序），讓產出的彙編指令在CPU上運行的時候更快。然而，這種優化產出的結果未必符合程序員原始的邏輯，所以，做爲程序員，做爲c程序員，必須有能力瞭解編譯器的行爲，並在經過內嵌在c代碼中的memory barrier來指導編譯器的優化行爲（這種memory barrier又叫作優化屏障，Optimization barrier），讓編譯器產出即高效，又邏輯正確的代碼。

CPU的核心思想就是取指執行，對於in-order的單核CPU，而且沒有cache（這種CPU在現實世界中還存在嗎？），彙編指令的取指和執行是嚴格按照順序進行的，也就是說，彙編指令就是所見即所得的，彙編指令的邏輯被嚴格的被CPU執行。然而，隨着計算機系統愈來愈複雜（多核、cache、superscalar、out-of-order），使用匯編指令這樣貼近處理器的語言也沒法保證其被CPU執行的結果的一致性，從而須要程序員（看，人仍是最不能夠替代的）告知CPU如何保證邏輯正確。

綜上所述，memory barrier是一種保證內存訪問順序的一種方法，讓系統中的HW block（各個cpu、DMA controler、device等）對內存有一致性的視角。

2、不使用memory barrier會致使問題的場景

一、編譯器的優化

咱們先看下面的一個例子：

preempt_disable（）

臨界區

preempt_enable

有些共享資源能夠經過禁止任務搶佔來進行保護，所以臨界區代碼被preempt_disable和preempt_enable給保護起來。其實，咱們知道所謂的preempt enable和disable其實就是對當前進程的struct thread_info中的preempt_count進行加一和減一的操做。具體的代碼以下：

#define preempt_disable() \
do { \
preempt_count_inc(); \
barrier(); \
} while (0)

linux kernel中的定義和咱們的想像同樣，除了barrier這個優化屏障。barrier就象是c代碼中的一個柵欄，將代碼邏輯分紅兩段，barrier以前的代碼和barrier以後的代碼在通過編譯器編譯後順序不能亂掉。也就是說，barrier以後的c代碼對應的彙編，不能跑到barrier以前去，反之亦然。之因此這麼作是由於在咱們這個場景中，若是編譯爲了榨取CPU的performace而對彙編指令進行重排，那麼臨界區的代碼就有可能位於preempt_count_inc以外，從而起不到保護做用。

如今，咱們知道了增長barrier的做用，問題來了，barrier是否夠呢？對於multi-core的系統，只有當該task被調度到該CPU上執行的時候，該CPU纔會訪問該task的preempt count，所以對於preempt enable和disable而言，不存在多個CPU同時訪問的場景。可是，即使這樣，若是CPU是亂序執行（out-of-order excution）的呢？其實，咱們也不用擔憂，正如前面敘述的，preempt count這個memory其實是不存在多個cpu同時訪問的狀況，所以，它實際上會本cpu的進程上下文和中斷上下文訪問。能終止當前thread執行preempt_disable的只有中斷。爲了方便描述，咱們給代碼編址，以下：

地址	該地址的彙編指令	CPU的執行順序
a	preempt_disable（）	臨界區指令1
a+4	臨界區指令1	preempt_disable（）
a+8	臨界區指令2	臨界區指令2
a+12	preempt_enable	preempt_enable

當發生中斷的時候，硬件會獲取當前PC值，並精確的獲得了發生指令的地址。有兩種狀況：

（1）在地址a發生中斷。對於out-of-order的CPU，臨界區指令1已經執行完畢，preempt_disable正在pipeline中等待執行。因爲是在a地址發生中斷，也就是preempt_disable地址上發生中斷，對於硬件而言，它會保證a地址以前（包括a地址）的指令都被執行完畢，而且a地址以後的指令都沒有執行。所以，在這種狀況下，臨界區指令1的執行結果被拋棄掉，所以，實際臨界區指令不會先於preempt_disable執行

（2）在地址a＋4發生中斷。這時候，雖然發生中斷的那一刻的地址上的指令（臨界區指令1）已經執行完畢了，可是硬件會保證地址a＋4以前的全部的指令都執行完畢，所以，實際上CPU會執行完preempt_disable，而後跳轉的中斷異常向量執行。

上面描述的是優化屏障在內存中的變量的應用，下面咱們看看硬件寄存器的場景。通常而言，串口的驅動都會包括控制檯部分的代碼，例如：

static struct console xx_serial_console = {
……
.write = xx_serial_console_write,
……
};

若是系統enable了串口控制檯，那麼當你的驅動調用printk的時候，實際上最終是經過console的write函數輸出到了串口控制檯。而這個console write的函數可能會包含下面的代碼：

do {
獲取TX FIFO狀態寄存器
barrier();
} while (TX FIFO沒有ready);
寫TX FIFO寄存器;

對於某些CPU archtecture而言（至少ARM是這樣的），外設硬件的IO地址也被映射到了一段內存地址空間，對編譯器而言，它並不知道這些地址空間是屬於外設的。所以，對於上面的代碼，若是沒有barrier的話，獲取TX FIFO狀態寄存器的指令可能和寫TX FIFO寄存器指令進行從新排序，在這種狀況下，程序邏輯就不對了，由於咱們必需要保證TX FIFO ready的狀況下才能寫TX FIFO寄存器。

對於multi core的狀況，上面的代碼邏輯也是OK的，由於在調用console write函數的時候，要獲取一個console semaphore，確保了只有一個thread進入，所以，console write的代碼不會在多個CPU上併發。和preempt count的例子同樣，咱們能夠問一樣的問題，若是CPU是亂序執行（out-of-order excution）的呢？barrier只是保證compiler輸出的彙編指令的順序是OK的，不能確保CPU執行時候的亂序。對這個問題的回答來自ARM architecture的內存訪問模型：對於program order是A1-->A2的狀況（A1和A2都是對Device或是Strongly-ordered的memory進行訪問的指令），ARM保證A1也是先於A2執行的。所以，在這樣的場景下，使用barrier足夠了。對於X86也是相似的，雖然它沒有對IO space採樣memory mapping的方式，可是，X86的全部操做IO端口的指令都是被順執行的，不須要考慮memory access order。

二、cpu architecture和cache的組織

注：本章節的內容來自對Paul E. McKenney的Why memory barriers文檔理解，更細緻的內容能夠參考該文檔。這個章節有些晦澀，須要一些耐心。做爲一個c程序員，你可能會抱怨，爲什麼設計CPU的硬件工程師不能屏蔽掉memory barrier的內容，讓c程序員關注在本身須要關注的程序邏輯上呢？本章能夠展開敘述，或許能解決一些疑問。

（1）基本概念

在The Memory Hierarchy文檔中，咱們已經瞭解了關於cache一些基礎的知識，一些基礎的內容，這裏就再也不重複了。咱們假設一個多核系統中的cache以下：

咱們先了解一下各個cpu cache line狀態的遷移過程：

（a）咱們假設在有一個memory中的變量爲多個CPU共享，那麼剛開始的時候，全部的CPU的本地cache中都沒有該變量的副本，全部的cacheline都是invalid狀態。

（b）所以當cpu 0 讀取該變量的時候發生cache miss（更具體的說叫作cold miss或者warmup miss）。當該值從memory中加載到chache 0中的cache line以後，該cache line的狀態被設定爲shared，而其餘的cache都是Invalid。

（c）當cpu 1 讀取該變量的時候，chache 1中的對應的cache line也變成shared狀態。其實shared狀態就是表示共享變量在一個或者多個cpu的cache中有副本存在。既然是被多個cache所共享，那麼其中一個CPU就不能武斷修改本身的cache而不通知其餘CPU的cache，不然會有一致性問題。

（d）老是read多沒勁，咱們讓CPU n對共享變量來一個load and store的操做。這時候，CPU n發送一個read invalidate命令，加載了Cache n的cache line，並將狀態設定爲exclusive，同時將全部其餘CPU的cache對應的該共享變量的cacheline設定爲invalid狀態。正由於如此，CPU n其實是獨佔了變量對應的cacheline（其餘CPU的cacheline都是invalid了，系統中就這麼一個副本），就算是寫該變量，也不須要通知其餘的CPU。CPU隨後的寫操做將cacheline設定爲modified狀態，表示cache中的數據已經dirty，和memory中的不一致了。modified狀態和exclusive狀態都是獨佔該cacheline，可是modified狀態下，cacheline的數據是dirty的，而exclusive狀態下，cacheline中的數據和memory中的數據是一致的。當該cacheline被替換出cache的時候，modified狀態的cacheline須要write back到memory中，而exclusive狀態不須要。

（e）在cacheline沒有被替換出CPU n的cache以前，CPU 0再次讀該共享變量，這時候會怎麼樣呢？固然是cache miss了（由於以前因爲CPU n寫的動做而致使其餘cpu的cache line變成了invalid，這種cache miss叫作communiction miss）。此外，因爲CPU n的cache line是modified狀態，它必須響應這個讀得操做（memory中是dirty的）。所以，CPU 0的cacheline變成share狀態（在此以前，CPU n的cache line應該會發生write back動做，從而致使其cacheline也是shared狀態）。固然，也多是CPU n的cache line不發生write back動做而是變成invalid狀態，CPU 0的cacheline變成modified狀態，這和具體的硬件設計相關。

（2）Store buffer

咱們考慮另一個場景：在上一節中step e中的操做變成CPU 0對共享變量進行寫的操做。這時候，寫的性能變得很是的差，由於CPU 0必需要等到CPU n上的cacheline 數據傳遞到其cacheline以後，才能進行寫的操做（CPU n上的cacheline 變成invalid狀態，CPU 0則切換成exclusive狀態，爲後續的寫動做作準備）。而從一個CPU的cacheline傳遞數據到另一個CPU的cacheline是很是消耗時間的，而這時候，CPU 0的寫的動做只是hold住，直到cacheline的數據完成傳遞。而實際上，這樣的等待是沒有意義的，所以，這時候cacheline的數據仍然會被覆蓋掉。爲了解決這個問題，多核系統中的cache修改以下：

這樣，問題解決了，寫操做沒必要等到cacheline被加載，而是直接寫到store buffer中而後歡快的去幹其餘的活。在CPU n的cacheline把數據傳遞到其cache 0的cacheline以後，硬件將store buffer中的內容寫入cacheline。

雖然性能問題解決了，可是邏輯錯誤也隨之引入，咱們能夠看下面的例子：

咱們假設a和b是共享變量，初始值都是0，能夠被cpu0和cpu1訪問。cpu 0的cache中保存了b的值（exclusive狀態），沒有a的值，而cpu 1的cache中保存了a的值，沒有b的值，cpu 0執行的彙編代碼是（用的是ARM彙編，沒有辦法，其餘的都不是那麼熟悉）：

ldr     r2, [pc, #28]   -------------------------- 取變量a的地址
ldr     r4, [pc, #20]   -------------------------- 取變量b的地址
mov     r3, #1
str     r3, [r2]           --------------------------a=1
str     r3, [r4]           --------------------------b=1

CPU 1執行的代碼是：

             ldr     r2, [pc, #28]   -------------------------- 取變量a的地址

             ldr     r3, [pc, #20] -------------------------- 取變量b的地址
start:     ldr     r3, [r3]          -------------------------- 取變量b的值
            cmp     r3, #0          ------------------------ b的值是否等於0？
            beq     start            ------------------------ 等於0的話跳轉到start

            ldr     r2, [r2]          -------------------------- 取變量a的值

當cpu 1執行到--取變量a的值--這條指令的時候，b已是被cpu0修改成1了，這也就是說a＝1這個代碼已經執行了，所以，從彙編代碼的邏輯來看，這時候a值應該是肯定的1。然而並不是如此，cpu 0和cpu 1執行的指令和動做描述以下：

cpu 0執行的指令	cpu 0動做描述	cpu 1執行的指令	cpu 1動做描述
str r3, [r2] （a=1）	一、發生cache miss 二、將1保存在store buffer中三、發送read invalidate命令，試圖從cpu 1的cacheline中獲取數據，並invalidate其cache line 注：這裏無需等待response，馬上執行下一條指令	ldr r3, [r3] （獲取b的值）	一、發生cache miss 二、發送read命令，試圖加載b對應的cacheline 注：這裏cpu必須等待read response，下面的指令依賴於這個讀取的結果
str r3, [r4] （b=1）	一、cache hit 二、cacheline中的值被修改成1，狀態變成modified
	響應cpu 1的read命令，發送read response（b＝1）給CPU 0。write back，將狀態設定爲shared
		cmp r3, #0	一、cpu 1收到來自cpu 0的read response，加載b對應的cacheline，狀態爲shared 二、b等於1，所以沒必要跳轉到start執行
		ldr r2, [r2] （獲取a的值）	一、cache hit 二、獲取了a的舊值，也就是0
			響應CPU 0的read invalid命令，將a對應的cacheline設爲invalid狀態，發送read response和invalidate ack。可是已經釀成大錯了。
	收到來自cpu 1的響應，將store buffer中的1寫入cache line。

對於硬件，CPU不清楚具體的代碼邏輯，它不可能直接幫助軟件工程師，只是提供一些memory barrier的指令，讓軟件工程師告訴CPU他想要的內存訪問邏輯順序。這時候，cpu 0的代碼修改以下：

ldr     r2, [pc, #28]   -------------------------- 取變量a的地址
ldr     r4, [pc, #20]   -------------------------- 取變量b的地址
mov     r3, #1
str     r3, [r2]           --------------------------a=1

確保清空store buffer的memory barrier instruction
str     r3, [r4]           --------------------------b=1

這種狀況下，cpu 0和cpu 1執行的指令和動做描述以下：

cpu 0執行的指令	cpu 0動做描述	cpu 1執行的指令	cpu 1動做描述
str r3, [r2] （a=1）	一、發生cache miss 二、將1保存在store buffer中三、發送read invalidate命令，試圖從cpu 1的cacheline中獲取數據，並invalidate其cache line 注：這裏無需等待response，馬上執行下一條指令	ldr r3, [r3] （獲取b的值）	一、發生cache miss 二、發送read命令，試圖加載b對應的cacheline 注：這裏cpu必須等待read response，下面的指令依賴於這個讀取的結果
memory barrier instruction	CPU收到memory barrier指令，知道軟件要控制訪問順序，所以不會執行下一條str指令，要等到收到read response和invalidate ack後，將store buffer中全部數據寫到cacheline以後纔會執行後續的store指令
		cmp r3, #0 beq start	一、cpu 1收到來自cpu 0的read response，加載b對應的cacheline，狀態爲shared 二、b等於0，跳轉到start執行
			響應CPU 0的read invalid命令，將a對應的cacheline設爲invalid狀態，發送read response和invalidate ack。
	收到來自cpu 1的響應，將store buffer中的1寫入cache line。
str r3, [r4] （b=1）	一、cache hit，可是cacheline狀態是shared，須要發送invalidate到cpu 1 二、將1保存在store buffer中注：這裏無需等待invalidate ack，馬上執行下一條指令
…	…	…	…

因爲增長了memory barrier，保證了a、b這兩個變量的訪問順序，從而保證了程序邏輯。

（3）Invalidate Queue

咱們先回憶一下爲什麼出現了stroe buffer：爲了加快cache miss狀態下寫的性能，硬件提供了store buffer，以便讓CPU先寫入，從而沒必要等待invalidate ack（這些交互是爲了保證各個cpu的cache的一致性）。然而，store buffer的size比較小，不須要特別多的store命令（假設每次都是cache miss）就能夠將store buffer填滿，這時候，沒有空間寫了，所以CPU也只能是等待invalidate ack了，這個狀態和memory barrier指令的效果是同樣的。

怎麼解決這個問題？CPU設計的硬件工程師對性能的追求是不會停歇的。咱們首先看看invalidate ack爲什麼如此之慢呢？這主要是由於cpu在收到invalidate命令後，要對cacheline執行invalidate命令，確保該cacheline的確是invalid狀態後，纔會發送ack。若是cache正忙於其餘工做，固然不能馬上執行invalidate命令，也就沒法會ack。

怎麼破？CPU設計的硬件工程師提供了下面的方法：

Invalidate Queue這個HW block從名字就能夠看出來是保存invalidate請求的隊列。其餘CPU發送到本CPU的invalidate命令會保存於此，這時候，並不須要等到實際對cacheline的invalidate操做完成，CPU就能夠回invalidate ack了。

同store buffer同樣，雖然性能問題解決了，可是對memory的訪問順序致使的邏輯錯誤也隨之引入，咱們能夠看下面的例子（和store buffer中的例子相似）：

咱們假設a和b是共享變量，初始值都是0，能夠被cpu0和cpu1訪問。cpu 0的cache中保存了b的值（exclusive狀態），而CPU 1和CPU 0的cache中都保存了a的值，狀態是shared。cpu 0執行的彙編代碼是：

ldr     r2, [pc, #28]   -------------------------- 取變量a的地址
ldr     r4, [pc, #20]   -------------------------- 取變量b的地址
mov     r3, #1
str     r3, [r2]           --------------------------a=1

確保清空store buffer的memory barrier instruction
str     r3, [r4]           --------------------------b=1

CPU 1執行的代碼是：

             ldr     r2, [pc, #28]   -------------------------- 取變量a的地址

             ldr     r3, [pc, #20] -------------------------- 取變量b的地址
start:     ldr     r3, [r3]          -------------------------- 取變量b的值
            cmp     r3, #0          ------------------------ b的值是否等於0？
            beq     start            ------------------------ 等於0的話跳轉到start

            ldr     r2, [r2]          -------------------------- 取變量a的值

這種狀況下，cpu 0和cpu 1執行的指令和動做描述以下：

cpu 0執行的指令	cpu 0動做描述	cpu 1執行的指令	cpu 1動做描述
str r3, [r2] （a=1）	一、a值在CPU 0的cache中狀態是shared，是read only的，所以，須要通知其餘的CPU 二、將1保存在store buffer中三、發送invalidate命令，試圖invalidate CPU 1中a對應的cache line 注：這裏無需等待response，馬上執行下一條指令	ldr r3, [r3] （獲取b的值）	一、發生cache miss 二、發送read命令，試圖加載b對應的cacheline 注：這裏cpu必須等待read response，下面的指令依賴於這個讀取的結果
			收到來自CPU 0的invalidate命令，放入invalidate queue，馬上回ack。
memory barrier instruction	CPU收到memory barrier指令，知道軟件要控制訪問順序，所以不會執行下一條str指令，要等到收到invalidate ack後，將store buffer中全部數據寫到cacheline以後纔會執行後續的store指令
	收到invalidate ack後，將store buffer中的1寫入cache line。OK，能夠繼續執行下一條指令了
str r3, [r4] （b=1）	一、cache hit 二、cacheline中的值被修改成1，狀態變成modified
	收到CPU 1發送來的read命令，將b值（等於1）放入read response中，回送給CPU 1，write back並將狀態修改成shared。
			收到response（b＝1），並加載cacheline，狀態是shared
		cmp r3, #0	b等於1，不會執行beq指令，而是執行下一條指令
		ldr r2, [r2] （獲取a的值）	一、cache hit （尚未執行invalidate動做，命令還在invalidate queue中呢）二、獲取了a的舊值，也就是0
			對a對應的cacheline執行invalidate 命令，可是，已經晚了

可怕的memory misorder問題又來了，都是因爲引入了invalidate queue引發，看來咱們還須要一個memory barrier的指令，咱們將程序修改以下：

             ldr     r2, [pc, #28]   -------------------------- 取變量a的地址

             ldr     r3, [pc, #20] -------------------------- 取變量b的地址
start:     ldr     r3, [r3]          -------------------------- 取變量b的值
            cmp     r3, #0          ------------------------ b的值是否等於0？
            beq     start            ------------------------ 等於0的話跳轉到start

確保清空invalidate queue的memory barrier instruction

            ldr     r2, [r2]          -------------------------- 取變量a的值

這種狀況下，cpu 0和cpu 1執行的指令和動做描述以下：

cpu 0執行的指令	cpu 0動做描述	cpu 1執行的指令	cpu 1動做描述
str r3, [r2] （a=1）	一、a值在CPU 0的cache中狀態是shared，是read only的，所以，須要通知其餘的CPU 二、將1保存在store buffer中三、發送invalidate命令，試圖invalidate CPU 1中a對應的cache line 注：這裏無需等待response，馬上執行下一條指令	ldr r3, [r3] （獲取b的值）	一、發生cache miss 二、發送read命令，試圖加載b對應的cacheline 注：這裏cpu必須等待read response，下面的指令依賴於這個讀取的結果
			收到來自CPU 0的invalidate命令，放入invalidate queue，馬上回ack。
memory barrier instruction	CPU收到memory barrier指令，知道軟件要控制訪問順序，所以不會執行下一條str指令，要等到收到invalidate ack後，將store buffer中全部數據寫到cacheline以後纔會執行後續的store指令
	收到invalidate ack後，將store buffer中的1寫入cache line。OK，能夠繼續執行下一條指令了
str r3, [r4] （b=1）	一、cache hit 二、cacheline中的值被修改成1，狀態變成modified
	收到CPU 1發送來的read命令，將b值（等於1）放入read response中，回送給CPU 1，write back並將狀態修改成shared。
			收到response（b＝1），並加載cacheline，狀態是shared
		cmp r3, #0	b等於1，不會執行beq指令，而是執行下一條指令
		memory barrier instruction	CPU收到memory barrier指令，知道軟件要控制訪問順序，所以不會執行下一條ldr指令，要等到執行完invalidate queue中的全部的invalidate命令以後纔會執行下一個ldr指令
		ldr r2, [r2] （獲取a的值）	一、cache miss 二、發送read命令，從CPU 0那裏加載新的a值

因爲增長了memory barrier，保證了a、b這兩個變量的訪問順序，從而保證了程序邏輯。

3、linux kernel的API

linux kernel的memory barrier相關的API列表以下：

接口名稱	做用
barrier()	優化屏障，阻止編譯器爲了進行性能優化而進行的memory access reorder
mb()	內存屏障（包括讀和寫），用於SMP和UP
rmb()	讀內存屏障，用於SMP和UP
wmb()	寫內存屏障，用於SMP和UP
smp_mb()	用於SMP場合的內存屏障，對於UP不存在memory order的問題（對彙編指令），所以，在UP上就是一個優化屏障，確保彙編和c代碼的memory order是一致的
smp_rmb()	用於SMP場合的讀內存屏障
smp_wmb()	用於SMP場合的寫內存屏障

barrier()這個接口和編譯器有關，對於gcc而言，其代碼以下：

#define barrier() __asm__ __volatile__("": : :"memory")

這裏的__volatile__主要是用來防止編譯器優化的。而這裏的優化是針對代碼塊而言的，使用嵌入式彙編的代碼分紅三塊：

一、嵌入式彙編以前的c代碼塊

二、嵌入式彙編代碼塊

三、嵌入式彙編以後的c代碼塊

這裏__volatile__就是告訴編譯器：不要由於性能優化而將這些代碼重排，我須要清清爽爽的保持這三塊代碼塊的順序（代碼塊內部是否重排不是這裏的__volatile__管轄範圍了）。

barrier中的嵌入式彙編中的clobber list沒有描述彙編代碼對寄存器的修改狀況，只是有一個memory的標記。咱們知道，clober list是gcc和gas的接口，用於gas通知gcc它對寄存器和memory的修改狀況。所以，這裏的memory就是告知gcc，在彙編代碼中，我修改了memory中的內容，嵌入式彙編以前的c代碼塊和嵌入式彙編以後的c代碼塊看到的memory是不同的，對memory的訪問不能依賴於嵌入式彙編以前的c代碼塊中寄存器的內容，須要從新加載。

優化屏障是和編譯器相關的，而內存屏障是和CPU architecture相關的，固然，咱們選擇ARM爲例來描述內存屏障。

Linux內核同步機制之（四）：spin lock

http://www.wowotech.net/kernel_synchronization/spinlock.html

1、前言

在linux kernel的實現中，常常會遇到這樣的場景：共享數據被中斷上下文和進程上下文訪問，該如何保護呢？若是隻有進程上下文的訪問，那麼能夠考慮使用semaphore或者mutex的鎖機制，可是如今中斷上下文也參和進來，那些能夠致使睡眠的lock就不能使用了，這時候，能夠考慮使用spin lock。本文主要介紹了linux kernel中的spin lock的原理以及代碼實現。因爲spin lock是architecture dependent代碼，所以，咱們在第四章討論了ARM32和ARM64上的實現細節。

注：本文須要進程和中斷處理的基本知識做爲支撐。

2、工做原理

一、spin lock的特色

咱們能夠總結spin lock的特色以下：

（1）spin lock是一種死等的鎖機制。當發生訪問資源衝突的時候，能夠有兩個選擇：一個是死等，一個是掛起當前進程，調度其餘進程執行。spin lock是一種死等的機制，當前的執行thread會不斷的從新嘗試直到獲取鎖進入臨界區。

（2）只容許一個thread進入。semaphore能夠容許多個thread進入，spin lock不行，一次只能有一個thread獲取鎖並進入臨界區，其餘的thread都是在門口不斷的嘗試。

（3）執行時間短。因爲spin lock死等這種特性，所以它使用在那些代碼不是很是複雜的臨界區（固然也不能太簡單，不然使用原子操做或者其餘適用簡單場景的同步機制就OK了），若是臨界區執行時間太長，那麼不斷在臨界區門口「死等」的那些thread是多麼的浪費CPU啊（固然，現代CPU的設計都會考慮同步原語的實現，例如ARM提供了WFE和SEV這樣的相似指令，避免CPU進入busy loop的悲慘境地）

（4）能夠在中斷上下文執行。因爲不睡眠，所以spin lock能夠在中斷上下文中適用。

二、場景分析

對於spin lock，其保護的資源可能來自多個CPU CORE上的進程上下文和中斷上下文的中的訪問，其中，進程上下文包括：用戶進程經過系統調用訪問，內核線程直接訪問，來自workqueue中work function的訪問（本質上也是內核線程）。中斷上下文包括：HW interrupt context（中斷handler）、軟中斷上下文（soft irq，固然因爲各類緣由，該softirq被推遲到softirqd的內核線程中執行的時候就不屬於這個場景了，屬於進程上下文那個分類了）、timer的callback函數（本質上也是softirq）、tasklet（本質上也是softirq）。

先看最簡單的單CPU上的進程上下文的訪問。若是一個全局的資源被多個進程上下文訪問，這時候，內核如何交錯執行呢？對於那些沒有打開preemptive選項的內核，全部的系統調用都是串行化執行的，所以不存在資源爭搶的問題。若是內核線程也訪問這個全局資源呢？本質上內核線程也是進程，相似普通進程，只不過普通進程時而在用戶態運行、時而經過系統調用陷入內核執行，而內核線程永遠都是在內核態運行，可是，結果是同樣的，對於non-preemptive的linux kernel，只要在內核態，就不會發生進程調度，所以，這種場景下，共享數據根本不須要保護（沒有併發，談何保護呢）。若是時間停留在這裏該多麼好，單純而美好，在繼續前進以前，讓咱們先享受這一刻。

當打開premptive選項後，事情變得複雜了，咱們考慮下面的場景：

（1）進程A在某個系統調用過程當中訪問了共享資源R

（2）進程B在某個系統調用過程當中也訪問了共享資源R

會不會形成衝突呢？假設在A訪問共享資源R的過程當中發生了中斷，中斷喚醒了沉睡中的，優先級更高的B，在中斷返回現場的時候，發生進程切換，B啓動執行，並經過系統調用訪問了R，若是沒有鎖保護，則會出現兩個thread進入臨界區，致使程序執行不正確。OK，咱們加上spin lock看看如何：A在進入臨界區以前獲取了spin lock，一樣的，在A訪問共享資源R的過程當中發生了中斷，中斷喚醒了沉睡中的，優先級更高的B，B在訪問臨界區以前仍然會試圖獲取spin lock，這時候因爲A進程持有spin lock而致使B進程進入了永久的spin……怎麼破？linux的kernel很簡單，在A進程獲取spin lock的時候，禁止本CPU上的搶佔（上面的永久spin的場合僅僅在本CPU的進程搶佔本CPU的當前進程這樣的場景中發生）。若是A和B運行在不一樣的CPU上，那麼狀況會簡單一些：A進程雖然持有spin lock而致使B進程進入spin狀態，不過因爲運行在不一樣的CPU上，A進程會持續執行並會很快釋放spin lock，解除B進程的spin狀態。

多CPU core的場景和單核CPU打開preemptive選項的效果是同樣的，這裏再也不贅述。

咱們繼續向前分析，如今要加入中斷上下文這個因素。訪問共享資源的thread包括：

（1）運行在CPU0上的進程A在某個系統調用過程當中訪問了共享資源R

（2）運行在CPU1上的進程B在某個系統調用過程當中也訪問了共享資源R

（3）外設P的中斷handler中也會訪問共享資源R

在這樣的場景下，使用spin lock能夠保護訪問共享資源R的臨界區嗎？咱們假設CPU0上的進程A持有spin lock進入臨界區，這時候，外設P發生了中斷事件，而且調度到了CPU1上執行，看起來沒有什麼問題，執行在CPU1上的handler會稍微等待一會CPU0上的進程A，等它馬上臨界區就會釋放spin lock的，可是，若是外設P的中斷事件被調度到了CPU0上執行會怎麼樣？CPU0上的進程A在持有spin lock的狀態下被中斷上下文搶佔，而搶佔它的CPU0上的handler在進入臨界區以前仍然會試圖獲取spin lock，悲劇發生了，CPU0上的P外設的中斷handler永遠的進入spin狀態，這時候，CPU1上的進程B也不可避免在試圖持有spin lock的時候失敗而致使進入spin狀態。爲了解決這樣的問題，linux kernel採用了這樣的辦法：若是涉及到中斷上下文的訪問，spin lock須要和禁止本CPU上的中斷聯合使用。

linux kernel中提供了豐富的bottom half的機制，雖然同屬中斷上下文，不過仍是稍有不一樣。咱們能夠把上面的場景簡單修改一下：外設P不是中斷handler中訪問共享資源R，而是在的bottom half中訪問。使用spin lock+禁止本地中斷固然是能夠達到保護共享資源的效果，可是使用牛刀來殺雞彷佛有點小題大作，這時候disable bottom half就OK了。

最後，咱們討論一下中斷上下文之間的競爭。同一種中斷handler之間在uni core和multi core上都不會並行執行，這是linux kernel的特性。若是不一樣中斷handler須要使用spin lock保護共享資源，對於新的內核（不區分fast handler和slow handler），全部handler都是關閉中斷的，所以使用spin lock不須要關閉中斷的配合。bottom half又分紅softirq和tasklet，同一種softirq會在不一樣的CPU上併發執行，所以若是某個驅動中的sofirq的handler中會訪問某個全局變量，對該全局變量是須要使用spin lock保護的，不用配合disable CPU中斷或者bottom half。tasklet更簡單，由於同一種tasklet不會多個CPU上併發，具體我就不分析了，你們自行思考吧。

3、通用代碼實現

一、文件整理

和體系結構無關的代碼以下：

（1）include/linux/spinlock_types.h。這個頭文件定義了通用spin lock的基本的數據結構（例如spinlock_t）和如何初始化的接口（DEFINE_SPINLOCK）。這裏的「通用」是指不論SMP仍是UP都通用的那些定義。

（2）include/linux/spinlock_types_up.h。這個頭文件不該該直接include，在include/linux/spinlock_types.h文件會根據系統的配置（是否SMP）include相關的頭文件，若是UP則會include該頭文件。這個頭文定義UP系統中和spin lock的基本的數據結構和如何初始化的接口。固然，對於non-debug版本而言，大部分struct都是empty的。

（3）include/linux/spinlock.h。這個頭文件定義了通用spin lock的接口函數聲明，例如spin_lock、spin_unlock等，使用spin lock模塊接口API的驅動模塊或者其餘內核模塊都須要include這個頭文件。

（4）include/linux/spinlock_up.h。這個頭文件不該該直接include，在include/linux/spinlock.h文件會根據系統的配置（是否SMP）include相關的頭文件。這個頭文件是debug版本的spin lock須要的。

（5）include/linux/spinlock_api_up.h。同上，只不過這個頭文件是non-debug版本的spin lock須要的

（6）linux/spinlock_api_smp.h。SMP上的spin lock模塊的接口聲明

（7）kernel/locking/spinlock.c。SMP上的spin lock實現。

頭文件有些凌亂，咱們對UP和SMP上spin lock頭文件進行整理：

UP須要的頭文件	SMP須要的頭文件
linux/spinlock_type_up.h: linux/spinlock_types.h: linux/spinlock_up.h: linux/spinlock_api_up.h: linux/spinlock.h	asm/spinlock_types.h linux/spinlock_types.h: asm/spinlock.h linux/spinlock_api_smp.h: linux/spinlock.h

二、數據結構

根據第二章的分析，咱們能夠基本能夠推斷出spin lock的實現。首先定義一個spinlock_t的數據類型，其本質上是一個整數值（對該數值的操做須要保證原子性），該數值表示spin lock是否可用。初始化的時候被設定爲1。當thread想要持有鎖的時候調用spin_lock函數，該函數將spin lock那個整數值減去1，而後進行判斷，若是等於0，表示能夠獲取spin lock，若是是負數，則說明其餘thread的持有該鎖，本thread須要spin。

內核中的spinlock_t的數據類型定義以下：

typedef struct spinlock {
struct raw_spinlock rlock;
} spinlock_t;

typedef struct raw_spinlock {
arch_spinlock_t raw_lock;
} raw_spinlock_t;

因爲各類緣由（各類鎖的debug、鎖的validate機制，多平臺支持什麼的），spinlock_t的定義沒有那麼直觀，爲了讓事情簡單一些，咱們去掉那些繁瑣的成員。struct spinlock中定義了一個struct raw_spinlock的成員，爲什麼會如此呢？好吧，咱們又須要回到kernel歷史課本中去了。在舊的內核中（好比我熟悉的linux 2.6.23內核），spin lock的命令規則是這樣：

通用（適用於各類arch）的spin lock使用spinlock_t這樣的type name，各類arch定義本身的struct raw_spinlock。聽起來不錯的主意和命名方式，直到linux realtime tree（PREEMPT_RT）提出對spinlock的挑戰。real time linux是一個試圖將linux kernel增長硬實時性能的一個分支（你知道的，linux kernel mainline只是支持soft realtime），多年來，不少來自realtime branch的特性被merge到了mainline上，例如：高精度timer、中斷線程化等等。realtime tree但願能夠對現存的spinlock進行分類：一種是在realtime kernel中能夠睡眠的spinlock，另一種就是在任何狀況下都不能夠睡眠的spinlock。分類很清楚可是如何起名字？起名字絕對是個技術活，起得好了事半功倍，可維護性好，什麼文檔啊、註釋啊都素那浮雲，閱讀代碼就是享受，如沐春風。起得很差，註定被後人唾棄，或者拖出來吊打（這讓我想起給我兒子起名字的那段不堪回首的歲月……）。最終，spin lock的命名規範定義以下：

（1）spinlock，在rt linux（配置了PREEMPT_RT）的時候可能會被搶佔（實際底層多是使用支持PI（優先級翻轉）的mutext）。

（2）raw_spinlock，即使是配置了PREEMPT_RT也要頑強的spin

（3）arch_spinlock，spin lock是和architecture相關的，arch_spinlock是architecture相關的實現

對於UP平臺，全部的arch_spinlock_t都是同樣的，定義以下：

typedef struct { } arch_spinlock_t;

什麼都沒有，一切都是空啊。固然，這也符合前面的分析，對於UP，即使是打開的preempt選項，所謂的spin lock也不過就是disable preempt而已，不需定義什麼spin lock的變量。

對於SMP平臺，這和arch相關，咱們在下一節描述。

三、spin lock接口API

咱們整理spin lock相關的接口API以下：

spinlock中的定義	raw_spinlock的定義	接口API的類型
DEFINE_SPINLOCK	DEFINE_RAW_SPINLOCK	定義spin lock並初始化
spin_lock_init	raw_spin_lock_init	動態初始化spin lock
spin_lock	raw_spin_lock	獲取指定的spin lock
spin_lock_irq	raw_spin_lock_irq	獲取指定的spin lock同時disable本CPU中斷
spin_lock_irqsave	raw_spin_lock_irqsave	保存本CPU當前的irq狀態，disable本CPU中斷並獲取指定的spin lock
spin_lock_bh	raw_spin_lock_bh	獲取指定的spin lock同時disable本CPU的bottom half
spin_unlock	raw_spin_unlock	釋放指定的spin lock
spin_unlock_irq	raw_spin_unock_irq	釋放指定的spin lock同時enable本CPU中斷
spin_unlock_irqstore	raw_spin_unlock_irqstore	釋放指定的spin lock同時恢復本CPU的中斷狀態
spin_unlock_bh	raw_spin_unlock_bh	獲取指定的spin lock同時enable本CPU的bottom half
spin_trylock	raw_spin_trylock	嘗試去獲取spin lock，若是失敗，不會spin，而是返回非零值
spin_is_locked	raw_spin_is_locked	判斷spin lock是不是locked，若是其餘的thread已經獲取了該lock，那麼返回非零值，不然返回0

在具體的實現面，咱們不可能把每個接口函數的代碼都呈現出來，咱們選擇最基礎的spin_lock爲例子，其餘的讀者能夠本身閱讀代碼來理解。

spin_lock的代碼以下：

static inline void spin_lock(spinlock_t *lock)
{
raw_spin_lock(&lock->rlock);
}

固然，在linux mainline代碼中，spin_lock和raw_spin_lock是同樣的，在realtime linux patch中，spin_lock應該被換成能夠sleep的版本，固然具體如何實現我沒有去看（也許直接使用了Mutex，畢竟它提供了優先級繼承特性來解決了優先級翻轉的問題），有興趣的讀者能夠自行閱讀，咱們這裏重點看看（本文也主要focus這個主題）真正的，不睡眠的spin lock，也就是是raw_spin_lock，代碼以下：

#define raw_spin_lock(lock)    _raw_spin_lock(lock)

UP中的實現：

#define _raw_spin_lock(lock)            __LOCK(lock)

#define __LOCK(lock) \
do { preempt_disable(); ___LOCK(lock); } while (0)

SMP的實現：

void __lockfunc _raw_spin_lock(raw_spinlock_t *lock)
{
    __raw_spin_lock(lock);
}

static inline void __raw_spin_lock(raw_spinlock_t *lock)
{
    preempt_disable();
    spin_acquire(&lock->dep_map, 0, 0, _RET_IP_);
    LOCK_CONTENDED(lock, do_raw_spin_trylock, do_raw_spin_lock);
}

UP中很簡單，本質上就是一個preempt_disable而已，和咱們在第二章中分析的一致。SMP中稍顯複雜，preempt_disable固然也是必須的，spin_acquire能夠略過，這是和運行時檢查鎖的有效性有關的，若是沒有定義CONFIG_LOCKDEP其實就是空函數。若是沒有定義CONFIG_LOCK_STAT（和鎖的統計信息相關），LOCK_CONTENDED就是調用do_raw_spin_lock而已，若是沒有定義CONFIG_DEBUG_SPINLOCK，它的代碼以下：

static inline void do_raw_spin_lock(raw_spinlock_t *lock) __acquires(lock)
{
__acquire(lock);
arch_spin_lock(&lock->raw_lock);
}

__acquire和靜態代碼檢查相關，忽略之，最終實際的獲取spin lock仍是要靠arch相關的代碼實現。

4、ARM平臺的細節

代碼位於arch/arm/include/asm/spinlock.h和spinlock_type.h，和通用代碼相似，spinlock_type.h定義ARM相關的spin lock定義以及初始化相關的宏；spinlock.h中包括了各類具體的實現。

一、回憶過去

在分析新的spin lock代碼以前，讓咱們先回到2.6.23版本的內核中，看看ARM平臺如何實現spin lock的。和arm平臺相關spin lock數據結構的定義以下（那時候仍是使用raw_spinlock_t而不是arch_spinlock_t）：

typedef struct {
volatile unsigned int lock;
} raw_spinlock_t;

一個整數就OK了，0表示unlocked，1表示locked。配套的API包括__raw_spin_lock和__raw_spin_unlock。__raw_spin_lock會持續判斷lock的值是否等於0，若是不等於0（locked）那麼其餘thread已經持有該鎖，本thread就不斷的spin，判斷lock的數值，一直等到該值等於0爲止，一旦探測到lock等於0，那麼就設定該值爲1，表示本thread持有該鎖了，固然，這些操做要保證原子性，細節和exclusive版本的ldr和str（即ldrex和strexeq）相關，這裏略過。馬上臨界區後，持鎖thread會調用__raw_spin_unlock函數是否spin lock，其實就是把0這個數值賦給lock。

這個版本的spin lock的實現固然能夠實現功能，並且在沒有衝突的時候表現出不錯的性能，不過存在一個問題：不公平。也就是全部的thread都是在無序的爭搶spin lock，誰先搶到誰先得，無論thread等了好久仍是剛剛開始spin。在衝突比較少的狀況下，不公平不會體現的特別明顯，然而，隨着硬件的發展，多核處理器的數目愈來愈多，多核之間的衝突愈來愈劇烈，無序競爭的spinlock帶來的performance issue終於浮現出來，根據Nick Piggin的描述：

On an 8 core (2 socket) Opteron, spinlock unfairness is extremely noticable, with a userspace test having a difference of up to 2x runtime per thread, and some threads are starved or "unfairly" granted the lock up to 1 000 000 (!) times.

多麼的不公平，有些可憐的thread須要飢餓的等待1000000次。本質上無序競爭從機率論的角度看應該是均勻分佈的，不過因爲硬件特性致使這麼嚴重的不公平，咱們來看一看硬件block：

lock本質上是保存在main memory中的，因爲cache的存在，固然不須要每次都有訪問main memory。在多核架構下，每一個CPU都有本身的L1 cache，保存了lock的數據。假設CPU0獲取了spin lock，那麼執行完臨界區，在釋放鎖的時候會調用smp_mb invalide其餘忙等待的CPU的L1 cache，這樣後果就是釋放spin lock的那個cpu能夠更快的訪問L1cache，操做lock數據，從而大大增長的下一次獲取該spin lock的機會。

二、回到如今：arch_spinlock_t

ARM平臺中的arch_spinlock_t定義以下（little endian）：

typedef struct {
    union {
        u32 slock;
        struct __raw_tickets {
            u16 owner;
            u16 next;
        } tickets;
    };
} arch_spinlock_t;

原本覺得一個簡單的整數類型的變量就搞定的spin lock看起來沒有那麼簡單，要理解這個數據結構，須要瞭解一些ticket-based spin lock的概念。若是你有機會去九毛九去排隊吃飯（聲明：不是九毛九的飯託，僅僅是喜歡麪食而常去吃而已）就會理解ticket-based spin lock。大概是由於便宜，每次去九毛九老是沒法長驅直入，門口的笑容可掬的靚女會給一個ticket，上面寫着15號，同時會告訴你，當前狀態是10號已經入席，11號在等待。

回到arch_spinlock_t，這裏的owner就是當前已經入席的那個號碼，next記錄的是下一個要分發的號碼。下面的描述使用普通的計算機語言和在九毛九就餐（假設九毛九隻有一張餐桌）的例子來進行描述，估計可讓吃貨更有興趣閱讀下去。最開始的時候，slock被賦值爲0，也就是說owner和next都是0，owner和next相等，表示unlocked。當第一個個thread調用spin_lock來申請lock（第一我的就餐）的時候，owner和next相等，表示unlocked，這時候該thread持有該spin lock（能夠擁有九毛九的惟一的那個餐桌），而且執行next++，也就是將next設定爲1（再來人就分配1這個號碼讓他等待就餐）。也許該thread執行很快（吃飯吃的快），沒有其餘thread來競爭就調用spin_unlock了（無人等待就餐，生意慘淡啊），這時候執行owner++，也就是將owner設定爲1（表示當前持有1這個號碼牌的人能夠就餐）。姍姍來遲的1號得到了直接就餐的機會，next++以後等於2。1號這個傢伙吃飯巨慢，這是不文明現象（thread不能持有spin lock過久），可是存在。又來一我的就餐，分配當前next值的號碼2，固然也會執行next++，以便下一我的或者3的號碼牌。持續來人就會分配三、四、五、6這些號碼牌，next值不斷的增長，可是owner巋然不動，直到欠扁的1號吃飯完畢（調用spin_unlock），釋放飯桌這個惟一資源，owner++以後等於2，表示持有2那個號碼牌的人能夠進入就餐了。

三、接口實現

一樣的，這裏也只是選擇一個典型的API來分析，其餘的你們能夠自行學習。咱們選擇的是arch_spin_lock，其ARM32的代碼以下：

static inline void arch_spin_lock(arch_spinlock_t *lock)
{
    unsigned long tmp;
    u32 newval;
    arch_spinlock_t lockval;

    prefetchw(&lock->slock);－－－－－－－－－－－－－－－－－－－－－－－－（1）
    __asm__ __volatile__(
"1:    ldrex    %0, [%3]\n"－－－－－－－－－－－－－－－－－－－－－－－－－（2）
"    add    %1, %0, %4\n"
"    strex    %2, %1, [%3]\n"－－－－－－－－－－－－－－－－－－－－－－－－（3）
"    teq    %2, #0\n"－－－－－－－－－－－－－－－－－－－－－－－－－－－－（4）
"    bne    1b"
    : "=&r" (lockval), "=&r" (newval), "=&r" (tmp)
    : "r" (&lock->slock), "I" (1 << TICKET_SHIFT)
    : "cc");

    while (lockval.tickets.next != lockval.tickets.owner) {－－－－－－－－－－－－（5）
        wfe();－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－（6）
        lockval.tickets.owner = ACCESS_ONCE(lock->tickets.owner);－－－－－－（7）
    }

    smp_mb();－－－－－－－－－－－－－－－－－－－－－－－－－－－－－－（8）
}

（1）和preloading cache相關的操做，主要是爲了性能考慮

（2）將slock的值保存在lockval這個臨時變量中

（3）將spin lock中的next加一

（4）判斷是否有其餘的thread插入。更具體的細節參考<Linux內核同步機制之（一）：原子操做>中的描述

（5）判斷當前spin lock的狀態，若是是unlocked，那麼直接獲取到該鎖

（6）若是當前spin lock的狀態是locked，那麼調用wfe進入等待狀態。更具體的細節請參考ARM WFI和WFE指令中的描述。

（7）其餘的CPU喚醒了本cpu的執行，說明owner發生了變化，該新的own賦給lockval，而後繼續判斷spin lock的狀態，也就是回到step 5。

（8）memory barrier的操做，具體能夠參考<memory barrier>中的描述。

arch_spin_lock函數ARM64的代碼（來自4.1.10內核）以下：

static inline void arch_spin_lock(arch_spinlock_t *lock)
{
    unsigned int tmp;
    arch_spinlock_t lockval, newval;

    asm volatile(
    /* Atomically increment the next ticket. */
"    prfm    pstl1strm, %3\n"
"1:    ldaxr    %w0, %3\n"－－－－－（A）－－－－－－－－－－－lockval = lock
"    add    %w1, %w0, %w5\n"－－－－－－－－－－－－－newval ＝ lockval + (1 << 16)，至關於next++
"    stxr    %w2, %w1, %3\n"－－－－－－－－－－－－－－lock ＝ newval
"    cbnz    %w2, 1b\n"－－－－－－－－－－－－－－是否有其餘PE的執行流插入？有的話，重來。
    /* Did we get the lock? */
"    eor    %w1, %w0, %w0, ror #16\n"－－lockval中的next域就是本身的號碼牌，判斷是否等於owner
"    cbz    %w1, 3f\n"－－－－－－－－－－－－－－－－若是等於，持鎖進入臨界區
    /*
     * No: spin on the owner. Send a local event to avoid missing an
     * unlock before the exclusive load.
     */
"    sevl\n"
"2:    wfe\n"－－－－－－－－－－－－－－－－－－－－不然進入spin
"    ldaxrh    %w2, %4\n"－－－－（A）－－－－－－－－－其餘cpu喚醒本cpu，獲取當前owner值
"    eor    %w1, %w2, %w0, lsr #16\n"－－－－－－－－－本身的號碼牌是否等於owner？
"    cbnz    %w1, 2b\n"－－－－－－－－－－若是等於，持鎖進入臨界區，否者回到2，即繼續spin
    /* We got the lock. Critical section starts here. */
"3:"
    : "=&r" (lockval), "=&r" (newval), "=&r" (tmp), "+Q" (*lock)
    : "Q" (lock->owner), "I" (1 << TICKET_SHIFT)
    : "memory");
}

基本的代碼邏輯的描述都已經嵌入代碼中，這裏須要特別說明的有兩個知識點：

（1）Load-Acquire/Store-Release指令的應用。Load-Acquire/Store-Release指令是ARMv8的特性，在執行load和store操做的時候順便執行了memory barrier相關的操做，在spinlock這個場景，使用Load-Acquire/Store-Release指令代替dmb指令能夠節省一條指令。上面代碼中的（A）就標識了使用Load-Acquire指令的位置。Store-Release指令在哪裏呢？在arch_spin_unlock中，這裏就不貼代碼了。Load-Acquire/Store-Release指令的做用以下：

－Load-Acquire能夠確保系統中全部的observer看到的都是該指令先執行，而後是該指令以後的指令（program order）再執行

－Store-Release指令能夠確保系統中全部的observer看到的都是該指令以前的指令（program order）先執行，Store-Release指令隨後執行

（2）第二個知識點是關於在arch_spin_unlock代碼中爲什麼沒有SEV指令？關於這個問題能夠參考ARM ARM文檔中的Figure B2-5，這個圖是PE（n）的global monitor的狀態遷移圖。當PE（n）對x地址發起了exclusive操做的時候，PE（n）的global monitor從open access遷移到exclusive access狀態，來自其餘PE上針對x（該地址已經被mark for PE（n））的store操做會致使PE（n）的global monitor從exclusive access遷移到open access狀態，這時候，PE（n）的Event register會被寫入event，就好象生成一個event，將該PE喚醒，從而能夠省略一個SEV的指令。

注：

（1）+表示在嵌入的彙編指令中，該操做數會被指令讀取（也就是說是輸入參數）也會被彙編指令寫入（也就是說是輸出參數）。
（2）=表示在嵌入的彙編指令中，該操做數會是write only的，也就是說只作輸出參數。
（3）I表示操做數是當即數

Linux內核同步機制之（五）：Read/Write spin lock

http://www.wowotech.net/kernel_synchronization/rw-spinlock.html

1、爲什麼會有rw spin lock？

在有了強大的spin lock以後，爲什麼還會有rw spin lock呢？無他，僅僅是爲了增長內核的併發，從而增長性能而已。spin lock嚴格的限制只有一個thread能夠進入臨界區，可是實際中，有些對共享資源的訪問能夠嚴格區分讀和寫的，這時候，其實多個讀的thread進入臨界區是OK的，使用spin lock則限制一個讀thread進入，從而致使性能的降低。

本文主要描述RW spin lock的工做原理及其實現。須要說明的是Linux內核同步機制之（四）：spin lock是本文的基礎，請先閱讀該文檔以便保證閱讀的暢順。

2、工做原理

一、應用舉例

咱們來看一個rw spinlock在文件系統中的例子：

static struct file_system_type *file_systems;
static DEFINE_RWLOCK(file_systems_lock);

linux內核支持多種文件系統類型，例如EXT4，YAFFS2等，每種文件系統都用struct file_system_type來表示。內核中全部支持的文件系統用一個鏈表來管理，file_systems指向這個鏈表的第一個node。訪問這個鏈表的時候，須要用file_systems_lock來保護，場景包括：

（1）register_filesystem和unregister_filesystem分別用來向系統註冊和註銷一個文件系統。

（2）fs_index或者fs_name等函數會遍歷該鏈表，找到對應的struct file_system_type的名字或者index。

這些操做能夠分紅兩類，第一類就是須要對鏈表進行更新的動做，例如向鏈表中增長一個file system type（註冊）或者減小一個（註銷）。另一類就是僅僅對鏈表進行遍歷的操做，並不修改鏈表的內容。在不修改鏈表的內容的前提下，多個thread進入這個臨界區是OK的，都能返回正確的結果。可是對於第一類操做則否則，這樣的更新鏈表的操做是排他的，只能是同時有一個thread在臨界區中。

二、基本的策略

使用普通的spin lock能夠完成上一節中描述的臨界區的保護，可是，因爲spin lock的特定就是隻容許一個thread進入，所以這時候就禁止了多個讀thread進入臨界區，而實際上多個read thread能夠同時進入的，但如今也只能是不停的spin，cpu強大的運算能力沒法發揮出來，若是使用不斷retry檢查spin lock的狀態的話（而不是使用相似ARM上的WFE這樣的指令），對系統的功耗也是影響很大的。所以，必須有新的策略來應對：

咱們首先看看加鎖的邏輯：

（1）假設臨界區內沒有任何的thread，這時候任何read thread或者write thread能夠進入，可是隻能是其一。

（2）假設臨界區內有一個read thread，這時候新來的read thread能夠任意進入，可是write thread不能夠進入

（3）假設臨界區內有一個write thread，這時候任何的read thread或者write thread都不能夠進入

（4）假設臨界區內有一個或者多個read thread，write thread固然不能夠進入臨界區，可是該write thread也沒法阻止後續read thread的進入，他要一直等到臨界區一個read thread也沒有的時候，才能夠進入，多麼可憐的write thread。

unlock的邏輯以下：

（1）在write thread離開臨界區的時候，因爲write thread是排他的，所以臨界區有且只有一個write thread，這時候，若是write thread執行unlock操做，釋放掉鎖，那些處於spin的各個thread（read或者write）能夠競爭上崗。

（2）在read thread離開臨界區的時候，須要根據狀況來決定是否讓其餘處於spin的write thread們參與競爭。若是臨界區仍然有read thread，那麼write thread仍是須要spin（注意：這時候read thread能夠進入臨界區，聽起來也是不公平的）直到全部的read thread釋放鎖（離開臨界區），這時候write thread們能夠參與到臨界區的競爭中，若是獲取到鎖，那麼該write thread能夠進入。

3、實現

一、通用代碼文件的整理

rw spin lock的頭文件的結構和spin lock是同樣的。include/linux/rwlock_types.h文件中定義了通用rw spin lock的基本的數據結構（例如rwlock_t）和如何初始化的接口（DEFINE_RWLOCK）。include/linux/rwlock.h。這個頭文件定義了通用rw spin lock的接口函數聲明，例如read_lock、write_lock、read_unlock、write_unlock等。include/linux/rwlock_api_smp.h文件定義了SMP上的rw spin lock模塊的接口聲明。

須要特別說明的是：用戶不須要include上面的頭文件，基本上普通spinlock和rw spinlock使用統一的頭文件接口，用戶只須要include一個include/linux/spinlock.h文件就OK了。

二、數據結構。rwlock_t數據結構定義以下：

typedef struct {
arch_rwlock_t raw_lock;
} rwlock_t;

rwlock_t依賴arch對rw spinlock相關的定義。

三、API

咱們整理RW spinlock的接口API以下表：

rw spinlock API	接口API描述
DEFINE_RWLOCK	定義rw spin lock並初始化
rwlock_init	動態初始化rw spin lock
read_lock write_lock	獲取指定的rw spin lock
read_lock_irq write_lock_irq	獲取指定的rw spin lock同時disable本CPU中斷
read_lock_irqsave write_lock_irqsave	保存本CPU當前的irq狀態，disable本CPU中斷並獲取指定的rw spin lock
read_lock_bh write_lock_bh	獲取指定的rw spin lock同時disable本CPU的bottom half
read_unlock write_unlock	釋放指定的spin lock
read_unlock_irq write_unlock_irq	釋放指定的rw spin lock同時enable本CPU中斷
read_unlock_irqrestore write_unlock_irqrestore	釋放指定的rw spin lock同時恢復本CPU的中斷狀態
read_unlock_bh write_unlock_bh	獲取指定的rw spin lock同時enable本CPU的bottom half
read_trylock write_trylock	嘗試去獲取rw spin lock，若是失敗，不會spin，而是返回非零值

在具體的實現面，如何將archtecture independent的代碼轉到具體平臺的代碼的思路是和spin lock同樣的，這裏再也不贅述。

二、ARM上的實現

對於arm平臺，rw spin lock的代碼位於arch/arm/include/asm/spinlock.h和spinlock_type.h（其實普通spin lock的代碼也是在這兩個文件中），和通用代碼相似，spinlock_type.h定義ARM相關的rw spin lock定義以及初始化相關的宏；spinlock.h中包括了各類具體的實現。咱們先看arch_rwlock_t的定義：

typedef struct {
u32 lock;
} arch_rwlock_t;

毫無壓力，就是一個32-bit的整數。從定義就能夠看出rw spinlock不是ticket-based spin lock。咱們再看看arch_write_lock的實現：

static inline void arch_write_lock(arch_rwlock_t *rw)
{
    unsigned long tmp;

    prefetchw(&rw->lock); －－－－－－－知道後面須要訪問這個內存，先通知hw進行preloading cache
    __asm__ __volatile__(
"1:    ldrex    %0, [%1]\n" －－－－－獲取lock的值並保存在tmp中
"    teq    %0, #0\n" －－－－－－－－判斷是否等於0
    WFE("ne") －－－－－－－－－－若是tmp不等於0，那麼說明有read 或者write的thread持有鎖，那麼仍是靜靜的等待吧。其餘thread會在unlock的時候Send Event來喚醒該CPU的
"    strexeq    %0, %2, [%1]\n" －－－－若是tmp等於0，將0x80000000這個值賦給lock
"    teq    %0, #0\n" －－－－－－－－是否str成功，若是有其餘thread在上面的過程插入進來就會失敗
"    bne    1b" －－－－－－－－－若是不成功，那麼須要從新來過，不然持有鎖，進入臨界區
    : "=&r" (tmp) －－－－％0
    : "r" (&rw->lock), "r" (0x80000000)－－－－－－－％1和％2
    : "cc");

    smp_mb(); －－－－－－－memory barrier的操做
}

對於write lock，只要臨界區有一個thread進行讀或者寫的操做（具體判斷是針對32bit的lock進行，覆蓋了writer和reader thread），該thread都會進入spin狀態。若是臨界區沒有任何的讀寫thread，那麼writer進入臨界區，並設定lock＝0x80000000。咱們再來看看write unlock的操做：

static inline void arch_write_unlock(arch_rwlock_t *rw)
{
    smp_mb(); －－－－－－－memory barrier的操做

    __asm__ __volatile__(
    "str    %1, [%0]\n"－－－－－－－－－－－恢復0值
    :
    : "r" (&rw->lock), "r" (0) －－－－－－－－％0和％1
    : "cc");

    dsb_sev();－－－－－－－memory barrier的操做加上send event，wakeup其餘 thread（那些cpu處於WFE狀態）
}

write unlock看起來很簡單，就是一個lock＝0x0的操做。瞭解了write相關的操做後，咱們再來看看read的操做：

static inline void arch_read_lock(arch_rwlock_t *rw)
{
    unsigned long tmp, tmp2;

    prefetchw(&rw->lock);
    __asm__ __volatile__(
"1:    ldrex    %0, [%2]\n"－－－－－－－－獲取lock的值並保存在tmp中
"    adds    %0, %0, #1\n"－－－－－－－－tmp = tmp + 1
"    strexpl    %1, %0, [%2]\n"－－－－若是tmp結果非負值，那麼就執行該指令，將tmp值存入lock
    WFE("mi")－－－－－－－－－若是tmp是負值，說明有write thread，那麼就進入wait for event狀態
"    rsbpls    %0, %1, #0\n"－－－－－判斷strexpl指令是否成功執行
"    bmi    1b"－－－－－－－－－－若是不成功，那麼須要從新來過，不然持有鎖，進入臨界區
    : "=&r" (tmp), "=&r" (tmp2)－－－－－－－－－－％0和％1
    : "r" (&rw->lock)－－－－－－－－－－－－－－－％2
    : "cc");

    smp_mb();
}

上面的代碼比較簡單，須要說明的是adds指令更新了狀態寄存器（指令中s那個字符就是這個意思），strexpl會根據adds指令的執行結果來判斷是否執行。pl的意思就是positive or zero，也就是說，若是結果是正數或者0（沒有thread在臨界區或者臨界區內有若干read thread），該指令都會執行，若是是負數（有write thread在臨界區），那麼就不執行。OK，最後咱們來看read unlock的函數：

static inline void arch_read_unlock(arch_rwlock_t *rw)
{
    unsigned long tmp, tmp2;

    smp_mb();

    prefetchw(&rw->lock);
    __asm__ __volatile__(
"1:    ldrex    %0, [%2]\n"－－－－－－－－獲取lock的值並保存在tmp中
"    sub    %0, %0, #1\n"－－－－－－－－tmp = tmp - 1
"    strex    %1, %0, [%2]\n"－－－－－－將tmp值存入lock中
"    teq    %1, #0\n"－－－－－－是否str成功，若是有其餘thread在上面的過程插入進來就會失敗
"    bne    1b"－－－－－－－若是不成功，那麼須要從新來過，不然離開臨界區
    : "=&r" (tmp), "=&r" (tmp2)－－－－－－－－－－－－％0和％1
    : "r" (&rw->lock)－－－－－－－－－－－－－－－－－％2
    : "cc");

    if (tmp == 0)
        dsb_sev();－－－－－若是read thread已經等於0，說明是最後一個離開臨界區的reader，那麼調用sev去喚醒WFE的cpu core
}

最後，總結一下：

32個bit的lock，0～30的bit用來記錄進入臨界區的read thread的數目，第31個bit用來記錄write thread的數目，因爲只容許一個write thread進入臨界區，所以1個bit就OK了。在這樣的設計下，read thread的數目最大就是2的30次冪減去1的數值，超過這個數值就溢出了，固然這個數值在目前的系統中已經足夠的大了，姑且認爲它是安全的吧。

4、後記

read/write spinlock對於read thread和write thread採用相同的優先級，read thread必須等待write thread完成離開臨界區才能夠進入，而write thread須要等到全部的read thread完成操做離開臨界區才能進入。正如咱們前面所說，這看起來對write thread有些不公平，但這就是read/write spinlock的特色。此外，在內核中，已經不鼓勵對read/write spinlock的使用了，RCU是更好的選擇。如何解決read/write spinlock優先級問題？RCU又是什麼呢？咱們下回分解。

Linux內核同步機制之（六）：Seqlock

http://www.wowotech.net/kernel_synchronization/seqlock.html

1、前言

普通的spin lock對待reader和writer是一視同仁，RW spin lock給reader賦予了更高的優先級，那麼有沒有讓writer優先的鎖的機制呢？答案就是seqlock。本文主要描述linux kernel 4.0中的seqlock的機制，首先是seqlock的工做原理，若是想淺嘗輒止，那麼瞭解了概念性的東東就OK了，也就是第二章了，固然，我仍是推薦普通的驅動工程師瞭解seqlock的API，第三章給出了一個簡單的例子，瞭解了這些，在驅動中（或者在其餘內核模塊）使用seqlock就能夠易如反掌了。細節是魔鬼，概念性的東西須要天才的思考，不是說就代碼實現的細節就無足輕重，若是想進入seqlock的心裏世界，推薦閱讀第四章seqlock的代碼實現，這一章和cpu體系結構相關的內容咱們選擇了ARM64（呵呵～～要跟上時代的步伐）。最後一章是參考資料，若是以爲本文描述不清楚，能夠參考這些經典文獻，在無數不眠之夜，她們給我心靈的慰籍，也願可以給讀者帶來快樂。

2、工做原理

一、overview

seqlock這種鎖機制是傾向writer thread，也就是說，除非有其餘的writer thread進入了臨界區，不然它會長驅直入，不管有多少的reader thread都不能阻擋writer的腳步。writer thread這麼霸道，reader腫麼辦？對於seqlock，reader這一側須要進行數據訪問的過程當中檢測是否有併發的writer thread操做，若是檢測到併發的writer，那麼從新read。經過不斷的retry，直到reader thread在臨界區的時候，沒有任何的writer thread插入便可。這樣的設計對reader而言不是很公平，特別是若是writer thread負荷比較重的時候，reader thread可能會retry屢次，從而致使reader thread這一側性能的降低。

總結一下seqlock的特色：臨界區只容許一個writer thread進入，在沒有writer thread的狀況下，reader thread能夠隨意進入，也就是說reader不會阻擋reader。在臨界區只有有reader thread的狀況下，writer thread能夠馬上執行，不會等待。

二、writer thread的操做

對於writer thread，獲取seqlock操做以下：

（1）獲取鎖（例如spin lock），該鎖確保臨界區只有一個writer進入。

（2）sequence counter加一

釋放seqlock操做以下：

（1）釋放鎖，容許其餘writer thread進入臨界區。

（2）sequence counter加一（注意：不是減一哦，sequence counter是一個不斷累加的counter）

由上面的操做可知，若是臨界區沒有任何的writer thread，那麼sequence counter是偶數（sequence counter初始化爲0），若是臨界區有一個writer thread（固然，也只能有一個），那麼sequence counter是奇數。

三、reader thread的操做以下：

（1）獲取sequence counter的值，若是是偶數，能夠進入臨界區，若是是奇數，那麼等待writer離開臨界區（sequence counter變成偶數）。進入臨界區時候的sequence counter的值咱們稱之old sequence counter。

（2）進入臨界區，讀取數據

（3）獲取sequence counter的值，若是等於old sequence counter，說明一切OK，不然回到step（1）

四、適用場景。通常而言，seqlock適用於：

（1）read操做比較頻繁

（2）write操做較少，可是性能要求高，不但願被reader thread阻擋（之因此要求write操做較少主要是考慮read side的性能）

（3）數據類型比較簡單，可是數據的訪問又沒法利用原子操做來保護。咱們舉一個簡單的例子來描述：假設須要保護的數據是一個鏈表，header--->A node--->B node--->C node--->null。reader thread遍歷鏈表的過程當中，將B node的指針賦給了臨時變量x，這時候，中斷髮生了，reader thread被preempt（注意，對於seqlock，reader並無禁止搶佔）。這樣在其餘cpu上執行的writer thread有充足的時間釋放B node的memory（注意：reader thread中的臨時變量x還指向這段內存）。當read thread恢復執行，並經過x這個指針進行內存訪問（例如試圖經過next找到C node），悲劇發生了……

3、API示例

在kernel中，jiffies_64保存了從系統啓動以來的tick數目，對該數據的訪問（以及其餘jiffies相關數據）須要持有jiffies_lock這個seq lock。

一、reader side代碼以下：

u64 get_jiffies_64(void)
{

    do {
        seq = read_seqbegin(&jiffies_lock);
        ret = jiffies_64;
    } while (read_seqretry(&jiffies_lock, seq));
}

二、writer side代碼以下：

static void tick_do_update_jiffies64(ktime_t now)
{
write_seqlock(&jiffies_lock);

臨界區會修改jiffies_64等相關變量，具體代碼略
write_sequnlock(&jiffies_lock);
}

對照上面的代碼，任何工程師均可以比着葫蘆畫瓢，使用seqlock來保護本身的臨界區。固然，seqlock的接口API很是豐富，有興趣的讀者能夠自行閱讀seqlock.h文件。

4、代碼實現

一、seq lock的定義

typedef struct {
struct seqcount seqcount;－－－－－－－－－－sequence counter
spinlock_t lock;
} seqlock_t;

seq lock實際上就是spin lock ＋ sequence counter。

二、write_seqlock/write_sequnlock

static inline void write_seqlock(seqlock_t *sl)
{
    spin_lock(&sl->lock);

    sl->sequence++;
    smp_wmb();
}

惟一須要說明的是smp_wmb這個用於SMP場合下的寫內存屏障，它確保了編譯器以及CPU都不會打亂sequence counter內存訪問以及臨界區內存訪問的順序（臨界區的保護是依賴sequence counter的值，所以不能打亂其順序）。write_sequnlock很是簡單，留給你們本身看吧。

三、read_seqbegin

static inline unsigned read_seqbegin(const seqlock_t *sl)
{
    unsigned ret;

repeat:
    ret = ACCESS_ONCE(sl->sequence); －－－進入臨界區以前，先要獲取sequenc counter的快照
    if (unlikely(ret & 1)) { －－－－－若是是奇數，說明有writer thread
        cpu_relax();
        goto repeat; －－－－若是有writer，那麼先不要進入臨界區，不斷的polling sequenc counter
    }

    smp_rmb(); －－－確保sequenc counter和臨界區的內存訪問順序
    return ret;
}

若是有writer thread，read_seqbegin函數中會有一個不斷polling sequenc counter，直到其變成偶數的過程，在這個過程當中，若是不加以控制，那麼總體系統的性能會有損失（這裏的性能指的是功耗和速度）。所以，在polling過程當中，有一個cpu_relax的調用，對於ARM64，其代碼是：

static inline void cpu_relax(void)
{
asm volatile("yield" ::: "memory");
}

yield指令用來告知硬件系統，本cpu上執行的指令是polling操做，沒有那麼急迫，若是有任何的資源衝突，本cpu可讓出控制權。

四、read_seqretry

static inline unsigned read_seqretry(const seqlock_t *sl, unsigned start)
{
smp_rmb();－－－確保sequenc counter和臨界區的內存訪問順序
return unlikely(sl->sequence != start);
}

start參數就是進入臨界區時候的sequenc counter的快照，比對當前退出臨界區的sequenc counter，若是相等，說明沒有writer進入打攪reader thread，那麼能夠愉快的離開臨界區。

還有一個比較有意思的邏輯問題：read_seqbegin爲什麼要進行奇偶判斷？把一切都推到read_seqretry中進行判斷不能夠嗎？也就是說，爲什麼read_seqbegin要等到沒有writer thread的狀況下才進入臨界區？其實有writer thread也能夠進入，反正在read_seqretry中能夠進行奇偶以及相等判斷，從而保證邏輯的正確性。固然，這樣想也是對的，不過在performance上有欠缺，reader在檢測到有writer thread在臨界區後，仍然放reader thread進入，可能會致使writer thread的一些額外的開銷（cache miss），所以，最好的方法是在read_seqbegin中攔截。

5、參考文獻

一、Understanding the Linux Kernel 3rd Edition

二、Linux Kernel Development 3rd Edition

三、Perfbook (https://www.kernel.org/pub/linux/kernel/people/paulmck/perfbook/perfbook.html)

RCU synchronize原理分析

http://www.wowotech.net/kernel_synchronization/223.html

RCU（Read-Copy Update）是Linux內核比較成熟的新型讀寫鎖，具備較高的讀寫併發性能，經常用在須要互斥的性能關鍵路徑。在kernel中，rcu有tiny rcu和tree rcu兩種實現，tiny rcu更加簡潔，一般用在小型嵌入式系統中，tree rcu則被普遍使用在了server, desktop以及android系統中。本文將以tree rcu爲分析對象。

1 如何度過寬限期

RCU的核心理念是讀者訪問的同時，寫者能夠更新訪問對象的副本，但寫者須要等待全部讀者完成訪問以後，才能刪除老對象。這個過程實現的關鍵和難點就在於如何判斷全部的讀者已經完成訪問。一般把寫者開始更新，到全部讀者完成訪問這段時間叫作寬限期（Grace Period）。內核中實現寬限期等待的函數是synchronize_rcu。

1.1 讀者鎖的標記

在普通的TREE RCU實現中，rcu_read_lock和rcu_read_unlock的實現很是簡單，分別是關閉搶佔和打開搶佔：

 
 
 
 
  
  
  
  static inline void __rcu_read_lock(void)
  
  
  
  {
  
  
  
   preempt_disable();
  
  
  
  }
  
  
  
   
  
  
  
  static inline void __rcu_read_unlock(void)
  
  
  
  {
  
  
  
   preempt_enable();
  
  
  
  }

這時是否度過寬限期的判斷就比較簡單：每一個CPU都通過一次搶佔。由於發生搶佔，就說明不在rcu_read_lock和rcu_read_unlock之間，必然已經完成訪問或者還未開始訪問。

1.2 每一個CPU度過quiescnet state

接下來咱們看每一個CPU上報完成搶佔的過程。kernel把這個完成搶佔的狀態稱爲quiescent state。每一個CPU在時鐘中斷的處理函數中，都會判斷當前CPU是否度過quiescent state。

 
 
 
 
  
  
  
  void update_process_times(int user_tick)
  
  
  
  {
  
  
  
  ......
  
  
  
   rcu_check_callbacks(cpu, user_tick);
  
  
  
  ......
  
  
  
  }
  
  
  
   
  
  
  
  void rcu_check_callbacks(int cpu, int user)
  
  
  
  {
  
  
  
  ......
  
  
  
   if (user || rcu_is_cpu_rrupt_from_idle()) {
  
  
  
   /*在用戶態上下文，或者idle上下文，說明已經發生過搶佔*/
  
  
  
   rcu_sched_qs(cpu);
  
  
  
   rcu_bh_qs(cpu);
  
  
  
   } else if (!in_softirq()) {
  
  
  
   /*僅僅針對使用rcu_read_lock_bh類型的rcu，不在softirq，
  
  
  
   *說明已經不在read_lock關鍵區域*/
  
  
  
   rcu_bh_qs(cpu);
  
  
  
   }
  
  
  
   rcu_preempt_check_callbacks(cpu);
  
  
  
   if (rcu_pending(cpu))
  
  
  
   invoke_rcu_core();
  
  
  
  ......
  
  
  
  }

這裏補充一個細節說明，Tree RCU有多個類型的RCU State，用於不一樣的RCU場景，包括rcu_sched_state、rcu_bh_state和rcu_preempt_state。不一樣的場景使用不一樣的RCU API，度過寬限期的方式就有所區別。例如上面代碼中的rcu_sched_qs和rcu_bh_qs，就是爲了標記不一樣的state度過quiescent state。普通的RCU例如內核線程、系統調用等場景，使用rcu_read_lock或者rcu_read_lock_sched，他們的實現是同樣的；軟中斷上下文則能夠使用rcu_read_lock_bh，使得寬限期更快度過。

細分這些場景是爲了提升RCU的效率。rcu_preempt_state將在下文進行說明。

1.3 彙報寬限期度過

每一個CPU度過quiescent state以後，須要向上彙報直至全部CPU完成quiescent state，從而標識寬限期的完成，這個彙報過程在軟中斷RCU_SOFTIRQ中完成。軟中斷的喚醒則是在上述的時鐘中斷中進行。

update_process_times

-> rcu_check_callbacks

-> invoke_rcu_core

RCU_SOFTIRQ軟中斷處理的彙報流程以下：

rcu_process_callbacks

-> __rcu_process_callbacks

-> rcu_check_quiescent_state

-> rcu_report_qs_rdp

-> rcu_report_qs_rnp

其中rcu_report_qs_rnp是從葉子節點向根節點的遍歷過程，同一個節點的子節點都經過quiescent state後，該節點也設置爲經過。

這個樹狀的彙報過程，也就是「Tree RCU」這個名字得來的原因。

樹結構每層的節點數量和葉子節點數量由一系列的宏定義來決定：

 
 
 
 
  
  
  
  #define MAX_RCU_LVLS 4
  
  
  
  #define RCU_FANOUT_1 (CONFIG_RCU_FANOUT_LEAF)
  
  
  
  #define RCU_FANOUT_2 (RCU_FANOUT_1 * CONFIG_RCU_FANOUT)
  
  
  
  #define RCU_FANOUT_3 (RCU_FANOUT_2 * CONFIG_RCU_FANOUT)
  
  
  
  #define RCU_FANOUT_4 (RCU_FANOUT_3 * CONFIG_RCU_FANOUT)
  
  
  
   
  
  
  
  #if NR_CPUS <= RCU_FANOUT_1
  
  
  
  # define RCU_NUM_LVLS 1
  
  
  
  # define NUM_RCU_LVL_0 1
  
  
  
  # define NUM_RCU_LVL_1 (NR_CPUS)
  
  
  
  # define NUM_RCU_LVL_2 0
  
  
  
  # define NUM_RCU_LVL_3 0
  
  
  
  # define NUM_RCU_LVL_4 0
  
  
  
  #elif NR_CPUS <= RCU_FANOUT_2
  
  
  
  # define RCU_NUM_LVLS 2
  
  
  
  # define NUM_RCU_LVL_0 1
  
  
  
  # define NUM_RCU_LVL_1 DIV_ROUND_UP(NR_CPUS, RCU_FANOUT_1)
  
  
  
  # define NUM_RCU_LVL_2 (NR_CPUS)
  
  
  
  # define NUM_RCU_LVL_3 0
  
  
  
  # define NUM_RCU_LVL_4 0
  
  
  
  #elif NR_CPUS <= RCU_FANOUT_3
  
  
  
  # define RCU_NUM_LVLS 3
  
  
  
  # define NUM_RCU_LVL_0 1
  
  
  
  # define NUM_RCU_LVL_1 DIV_ROUND_UP(NR_CPUS, RCU_FANOUT_2)
  
  
  
  # define NUM_RCU_LVL_2 DIV_ROUND_UP(NR_CPUS, RCU_FANOUT_1)
  
  
  
  # define NUM_RCU_LVL_3 (NR_CPUS)
  
  
  
  # define NUM_RCU_LVL_4 0
  
  
  
  #elif NR_CPUS <= RCU_FANOUT_4
  
  
  
  # define RCU_NUM_LVLS 4
  
  
  
  # define NUM_RCU_LVL_0 1
  
  
  
  # define NUM_RCU_LVL_1 DIV_ROUND_UP(NR_CPUS, RCU_FANOUT_3)
  
  
  
  # define NUM_RCU_LVL_2 DIV_ROUND_UP(NR_CPUS, RCU_FANOUT_2)
  
  
  
  # define NUM_RCU_LVL_3 DIV_ROUND_UP(NR_CPUS, RCU_FANOUT_1)
  
  
  
  # define NUM_RCU_LVL_4 (NR_CPUS)

1.3 寬限期的發起與完成

全部寬限期的發起和完成都是由同一個內核線程rcu_gp_kthread來完成。經過判斷rsp->gp_flags & RCU_GP_FLAG_INIT來決定是否發起一個gp；經過判斷! (rnp->qsmask) && !rcu_preempt_blocked_readers_cgp(rnp))來決定是否結束一個gp。

發起一個GP時，rsp->gpnum++；結束一個GP時，rsp->completed = rsp->gpnum。

1.4 rcu callbacks處理

rcu的callback一般是在sychronize_rcu中添加的wakeme_after_rcu，也就是喚醒synchronize_rcu的進程，它正在等待GP的結束。

callbacks的處理一樣在軟中斷RCU_SOFTIRQ中完成

rcu_process_callbacks

-> __rcu_process_callbacks

-> invoke_rcu_callbacks

-> rcu_do_batch

-> __rcu_reclaim

這裏RCU的callbacks鏈表採用了一種分段鏈表的方式，整個callback鏈表，根據具體GP結束的時間，分紅若干段：nxtlist -- *nxttail[RCU_DONE_TAIL] -- *nxttail[RCU_WAIT_TAIL] -- *nxttail[RCU_NEXT_READY_TAIL] -- *nxttail[RCU_NEXT_TAIL]。

rcu_do_batch只處理nxtlist -- *nxttail[RCU_DONE_TAIL]之間的callbacks。每一個GP結束都會從新調整callback所處的段位，每一個新的callback將會添加在末尾，也就是*nxttail[RCU_NEXT_TAIL]。

2 可搶佔的RCU

若是config文件定義了CONFIG_TREE_PREEMPT_RCU=y，那麼sychronize_rcu將默認使用rcu_preempt_state。這類rcu的特色就在於read_lock期間是容許其它進程搶佔的，所以它判斷寬限期度過的方法就不太同樣。

從rcu_read_lock和rcu_read_unlock的定義就能夠知道，TREE_PREEMPT_RCU並非以簡單的通過搶佔爲CPU渡過GP的標準，而是有個rcu_read_lock_nesting計數

 
 
 
 
  
  
  
  void __rcu_read_lock(void)
  
  
  
  {
  
  
  
   current->rcu_read_lock_nesting++;
  
  
  
   barrier(); /* critical section after entry code. */
  
  
  
  }
  
  
  
   
  
  
  
  void __rcu_read_unlock(void)
  
  
  
  {
  
  
  
   struct task_struct *t = current;
  
  
  
   
  
  
  
   if (t->rcu_read_lock_nesting != 1) {
  
  
  
   --t->rcu_read_lock_nesting;
  
  
  
   } else {
  
  
  
   barrier(); /* critical section before exit code. */
  
  
  
   t->rcu_read_lock_nesting = INT_MIN;
  
  
  
   barrier(); /* assign before ->rcu_read_unlock_special load */
  
  
  
   if (unlikely(ACCESS_ONCE(t->rcu_read_unlock_special)))
  
  
  
   rcu_read_unlock_special(t);
  
  
  
   barrier(); /* ->rcu_read_unlock_special load before assign */
  
  
  
   t->rcu_read_lock_nesting = 0;
  
  
  
   }
  
  
  
  }

當搶佔發生時，__schedule函數會調用rcu_note_context_switch來通知RCU更新狀態，若是當前CPU處於rcu_read_lock狀態，當前進程將會放入rnp->blkd_tasks阻塞隊列，並呈如今rnp->gp_tasks鏈表中。

從上文1.3節寬限期的結束處理過程咱們能夠知道，rcu_gp_kthread會判斷! (rnp->qsmask) && !rcu_preempt_blocked_readers_cgp(rnp))兩個條件來決定GP是否完成，其中!rnp->qsmask表明每一個CPU都通過一次quiescent state，quiescent state的定義與傳統RCU一致；!rcu_preempt_blocked_readers_cgp(rnp)這個條件就表明了rcu是否還有阻塞的進程。

Linux內核同步機制之（七）：RCU基礎

http://www.wowotech.net/kernel_synchronization/rcu_fundamentals.html

1、前言

關於RCU的文檔包括兩份，一份講基本的原理（也就是本文了），一份講linux kernel中的實現。第二章描述了爲什麼有RCU這種同步機制，特別是在cpu core數目不斷遞增的今天，一個性能更好的同步機制是如何解決問題的，固然，再好的工具都有其適用場景，本章也給出了RCU的一些應用限制。第三章的第一小節描述了RCU的設計概念，其實RCU的設計概念比較簡單，比較容易理解，比較困難的是產品級別的RCU實現，咱們會在下一篇文檔中描述。第三章的第二小節描述了RCU的相關操做，其實就是對應到了RCU的外部接口API上來。最後一章是參考文獻，perfbook是一本神奇的數，喜歡並行編程的同窗絕對不能錯過的一本書，強烈推薦。和perfbook比起來，本文顯得很是的醜陋（主要是有些RCU的知識仍是理解不深入，可能須要再仔細看看linux kernel中的實現才能瞭解其真正含義），除了是中文表述以外，沒有任何的優勢，英語比較好的同窗能夠直接參考該書。

2、爲什麼有RCU這種同步機制呢？

前面咱們講了spin lock，rw spin lock和seq lock，爲什麼又出現了RCU這樣的同步機制呢？這個問題相似於問：有了刀槍劍戟這樣的工具，爲什麼會出現流星錘這樣的兵器呢？每種兵器都有本身的適用場合，內核同步機制亦然。RCU在必定的應用場景下，解決了過去同步機制的問題，這也是它之因此存在的基石。本章主要包括兩部份內容：一部分是如何解決其餘內核機制的問題，另一部分是受限的場景爲什麼？

一、性能問題

咱們先回憶一下spin lcok、RW spin lcok和seq lock的基本原理。對於spin lock而言，臨界區的保護是經過next和owner這兩個共享變量進行的。線程調用spin_lock進入臨界區，這裏包括了三個動做：

（1）獲取了本身的號碼牌（也就是next值）和容許哪個號碼牌進入臨界區（owner）

（2）設定下一個進入臨界區的號碼牌（next++）

（3）判斷本身的號碼牌是不是容許進入的那個號碼牌（next == owner），若是是，進入臨界區，否者spin（不斷的獲取owner的值，判斷是否等於本身的號碼牌，對於ARM64處理器而言，能夠使用WFE來下降功耗）。

注意：（1）是取值，（2）是更新並寫回，所以（1）和（2）必須是原子操做，中間不能插入任何的操做。

線程調用spin_unlock離開臨界區，執行owner++，表示下一個線程能夠進入。

RW spin lcok和seq lock都相似spin lock，它們都是基於一個memory中的共享變量（對該變量的訪問是原子的）。咱們假設系統架構以下：

當線程在多個cpu上爭搶進入臨界區的時候，都會操做那個在多個cpu之間共享的數據lock（玫瑰色的block）。cpu 0操做了lock，爲了數據的一致性，cpu 0的操做會致使其餘cpu的L1中的lock變成無效，在隨後的來自其餘cpu對lock的訪問會致使L1 cache miss（更準確的說是communication cache miss），必須從下一個level的cache中獲取，一樣的，其餘cpu的L1 cache中的lock也被設定爲invalid，從而引發下一次其餘cpu上的communication cache miss。

RCU的read side不須要訪問這樣的「共享數據」，從而極大的提高了reader側的性能。

二、reader和writer能夠併發執行

spin lock是互斥的，任什麼時候候只有一個thread（reader or writer）進入臨界區，rw spin lock要好一些，容許多個reader併發執行，提升了性能。不過，reader和updater不能併發執行，RCU解除了這些限制，容許一個updater（不能多個updater進入臨界區，這能夠經過spinlock來保證）和多個reader併發執行。咱們能夠比較一下rw spin lock和RCU，參考下圖：

rwlock容許多個reader併發，所以，在上圖中，三個rwlock reader愉快的並行執行。當rwlock writer試圖進入的時候（紅色虛線），只能spin，直到全部的reader退出臨界區。一旦有rwlock writer在臨界區，任何的reader都不能進入，直到writer完成數據更新，馬上臨界區。綠色的reader thread們又能夠進行愉快玩耍了。rwlock的一個特色就是肯定性，白色的reader必定是讀取的是old data，而綠色的reader必定獲取的是writer更新以後的new data。RCU和傳統的鎖機制不一樣，當RCU updater進入臨界區的時候，即使是有reader在也無所謂，它能夠長驅直入，不須要spin。一樣的，即使有一個updater正在臨界區裏面工做，這並不能阻擋RCU reader的步伐。因而可知，RCU的併發性能要好於rwlock，特別若是考慮cpu的數目比較多的狀況，那些處於spin狀態的cpu在無謂的消耗，多麼惋惜，隨着cpu的數目增長，rwlock性能不斷的降低。RCU reader和updater因爲能夠併發執行，所以這時候的被保護的數據有兩份，一份是舊的，一份是新的，對於白色的RCU reader，其讀取的數據多是舊的，也多是新的，和數據訪問的timing相關，固然，當RCU update完成更新以後，新啓動的RCU reader（綠色block）讀取的必定是新的數據。

三、適用的場景

咱們前面說過，每種鎖都有本身的適用的場景：spin lock不區分reader和writer，對於那些讀寫強度不對稱的是不適合的，RW spin lcok和seq lock解決了這個問題，不過seq lock傾向writer，而RW spin lock更照顧reader。看起來一切都已經很完美了，可是，隨着計算機硬件技術的發展，CPU的運算速度愈來愈快，相比之下，存儲器件的速度發展較爲滯後。在這種背景下，獲取基於counter（須要訪問存儲器件）的鎖（例如spin lock，rwlock）的機制開銷比較大。並且，目前的趨勢是：CPU和存儲器件之間的速度差異在逐漸擴大。所以，那些基於一個multi-processor之間的共享的counter的鎖機制已經不能知足性能的需求，在這種狀況下，RCU機制應運而生（固然，更準確的說RCU一種內核同步機制，但不是一種lock，本質上它是lock-free的），它克服了其餘鎖機制的缺點，可是，甘蔗沒有兩頭甜，RCU的使用場景比較受限，主要適用於下面的場景：

（1）RCU只能保護動態分配的數據結構，而且必須是經過指針訪問該數據結構

（2）受RCU保護的臨界區內不能sleep（SRCU不是本文的內容）

（3）讀寫不對稱，對writer的性能沒有特別要求，可是reader性能要求極高。

（4）reader端對新舊數據不敏感。

3、RCU的基本思路

一、原理

RCU的基本思路能夠經過下面的圖片體現：

RCU涉及的數據有兩種，一個是指向要保護數據的指針，咱們稱之RCU protected pointer。另一個是經過指針訪問的共享數據，咱們稱之RCU protected data，固然，這個數據必須是動態分配的。對共享數據的訪問有兩種，一種是writer，即對數據要進行更新，另一種是reader。若是在有reader在臨界區內進行數據訪問，對於傳統的，基於鎖的同步機制而言，reader會阻止writer進入（例如spin lock和rw spin lock。seqlock不會這樣，所以本質上seqlock也是lock-free的），由於在有reader訪問共享數據的狀況下，write直接修改data會破壞掉共享數據。怎麼辦呢？固然是移除了reader對共享數據的訪問以後，再讓writer進入了（writer稍顯悲劇）。對於RCU而言，其原理是相似的，爲了可以讓writer進入，必須首先移除reader對共享數據的訪問，怎麼移除呢？建立一個新的copy是一個不錯的選擇。所以RCU writer的動做分紅了兩步：

（1）removal。write分配一個new version的共享數據進行數據更新，更新完畢後將RCU protected pointer指向新版本的數據。一旦把RCU protected pointer指向的新的數據，也就意味着將其推向前臺，公佈與衆（reader都是經過pointer訪問數據的）。經過這樣的操做，原來read 0、一、2對共享數據的reference被移除了（對於新版本的受RCU保護的數據而言），它們都是在舊版本的RCU protected data上進行數據訪問。

（2）reclamation。共享數據不能有兩個版本，所以必定要在適當的時機去回收舊版本的數據。固然，不能太着急，不能reader線程還訪問着old version的數據的時候就強行回收，這樣會讓reader crash的。reclamation必須發生在全部的訪問舊版本數據的那些reader離開臨界區以後再回收，而這段等待的時間被稱爲grace period。

順便說明一下，reclamation並不須要等待read3和4，由於write端的爲RCU protected pointer賦值的語句是原子的，亂入的reader線程要麼看到的是舊的數據，要麼是新的數據。對於read3和4，它們訪問的是新的共享數據，所以不會reference舊的數據，所以reclamation不須要等待read3和4離開臨界區。

二、基本RCU操做

對於reader，RCU的操做包括：

（1）rcu_read_lock，用來標識RCU read side臨界區的開始。

（2）rcu_dereference，該接口用來獲取RCU protected pointer。reader要訪問RCU保護的共享數據，固然要獲取RCU protected pointer，而後經過該指針進行dereference的操做。

（3）rcu_read_unlock，用來標識reader離開RCU read side臨界區

對於writer，RCU的操做包括：

（1）rcu_assign_pointer。該接口被writer用來進行removal的操做，在witer完成新版本數據分配和更新以後，調用這個接口可讓RCU protected pointer指向RCU protected data。

（2）synchronize_rcu。writer端的操做能夠是同步的，也就是說，完成更新操做以後，能夠調用該接口函數等待全部在舊版本數據上的reader線程離開臨界區，一旦從該函數返回，說明舊的共享數據沒有任何引用了，能夠直接進行reclaimation的操做。

（3）call_rcu。固然，某些狀況下（例如在softirq context中），writer沒法阻塞，這時候能夠調用call_rcu接口函數，該函數僅僅是註冊了callback就直接返回了，在適當的時機會調用callback函數，完成reclaimation的操做。這樣的場景實際上是分開removal和reclaimation的操做在兩個不一樣的線程中：updater和reclaimer。

4、參考文檔

一、perfbook

二、linux-4.1.10\Documentation\RCU\*