Linux 核心：匠心獨運之無鎖環形隊列
| Kernel version | Linux 2.6.12 |
|---|---|
| Author | Toney |
| [email protected] | |
| Date | 2020.11.8 |
目錄
Linux 核心：匠心獨運之無鎖環形隊列
1. 前言
2. Kfifo簡介
3. Kfifo初始化
3.1 判斷一個數是否為2的幂次方
3.2 求不小于某個數2的整數次幂
3.3 為什麼要求2的幂次方呢？
4. Kfifo入隊和出隊
4.1 Kfifo右側入隊
4.2 Kfifo右側+左側入隊
4.3 無符号整數溢出回繞
4. 體會
1. 前言
金庸老爺子在《神雕俠侶》中說獨孤求敗的玄鐵重劍時，說道“重劍無鋒，大巧不工”。他說的是如果個人修養達到一定的階段，“花石草木皆可為劍”，而不需要更多技巧。在Linux核心中從來不缺少簡潔、優美、高效的實作代碼，缺少的是發現這些美的眼睛和毅力。在Linux核心中，代碼的簡潔高效并不意味采用了失傳很久的武林絕技，恰恰相反，它們往往通過最基本的知識和資料結構來實作完美的代碼，而kfifo可以說就是其中的一個典範。
這裡用“大巧不工”來形容Linux中的無鎖環形隊列顯然不合适，原因在于：無鎖環形隊列屬于精雕細琢，大道至簡、匠心獨運，簡潔而不簡單。它使用最基本的技術知識實作了重要的功能。下面我們便一睹其芳容。
2. Kfifo簡介
| 本文分析的原代碼版本 | 2.6.12 |
|---|---|
| kfifo的頭檔案 | linux-2.6.12\include\linux\kfifo.h |
| kfifo的源檔案 | linux-2.6.12\kernel\kfifo.c |
kfifo是一種"First In First Out “資料結構，它采用了前面提到的環形緩沖區來實作，提供一個無邊界的位元組流服務。采用環形緩沖區的好處為，當一個資料元素被用掉後，其餘資料元素不需要移動其存儲位置，進而減少拷貝提高效率。更重要的是，kfifo采用了并行無鎖技術，kfifo實作的單生産/單消費模式的共享隊列是不需要加鎖同步的。
并行無鎖技術的由來：
目前高性能的伺服器軟體(例如HTTP加速器)大多都運作在多核伺服器上，目前的硬體可以支援32、 64甚至更多的CPU，在這種高并發的環境下，鎖競争機制有時候比資料拷貝、上下文切換等更傷害系統的性能，是以在多核環境下，需要把重要的資料結構從鎖的保護下移到無鎖環境中，以此來提高軟體的性能。
是以，現在無鎖機制越來越流行，在不同的環境中使用不同的無鎖隊列可以節省開銷，提高程式效率。
[1] 摘自《深入淺出DPDK》第四章同步互斥機制：4.4.1 Linux核心無鎖環形緩沖
下面我們說一下kfifo的結構
struct kfifo {
unsigned char *buffer; /* the buffer holding the data */
unsigned int size; /* the size of the allocated buffer */
unsigned int in; /* data is added at offset (in % size) */
unsigned int out; /* data is extracted from off. (out % size) */
spinlock_t *lock; /* protects concurrent modifications */
};
kfifo結構中個字段的含義：
| buffer | 用于存放資料的緩存 |
|---|---|
| size | 緩沖區空間的大小，要求為2的幂次方 |
| in | 指向buffer中隊頭 |
| out | 指向buffer中的隊尾 |
| lock | 用來同步多個生産者、多個消費者的情形 |
Kfifo無鎖隊列的應用注意事項：
- 單生産者/單消費者無需使用鎖進行同步
- 未使用kfifo_reset()
- 隻有在消費者端使用了kfifo_reset_out()
以上三種條件都滿足的情況下可以使用kfifo無鎖隊列。相反，如果存在多個生産者或者多個消費者，則可以通過鎖來進行同步：
- 多個生産者一個消費者模式，生産者端加鎖同步
- 單個生産者多個消費者模式。消費者端加鎖同步
Kfifo作為一個基本FIFO結構，包括入隊函數
___kfifo_put
、出隊函數
__kfifo_get()
等基本操作。下面來一一說明。
3. Kfifo初始化
Kfifo的初始化是指為kfifo配置設定空間、初始化kfifo中的各項參數等操作。
/**
* kfifo_alloc - allocates a new FIFO and its internal buffer
* @size: the size of the internal buffer to be allocated.
* @gfp_mask: get_free_pages mask, passed to kmalloc()
* @lock: the lock to be used to protect the fifo buffer
*
* The size will be rounded-up to a power of 2.
*/
struct kfifo *kfifo_alloc(unsigned int size, unsigned int __nocast gfp_mask, spinlock_t *lock)
{
unsigned char *buffer;
struct kfifo *ret;
/*
* round up to the next power of 2, since our 'let the indices
* wrap' tachnique works only in this case.
*/
if (size & (size - 1)) {/*如果不是2的幂次方，則向上取到2的幂次方*/
BUG_ON(size > 0x80000000);
size = roundup_pow_of_two(size);
}
buffer = kmalloc(size, gfp_mask);
if (!buffer)
return ERR_PTR(-ENOMEM);
ret = kfifo_init(buffer, size, gfp_mask, lock);
if (IS_ERR(ret))
kfree(buffer);
return ret;
}
3.1 判斷一個數是否為2的幂次方
在這個
kfifo_alloc()
函數中，要求size需要為2的幂次方，如何實作高效的判斷呢？
在二進制中，2的幂次方很容易表示：一個數隻有一個bit上是1，其餘全為0，例如：
| 十進制數表示 | 二進制表示 | 是否為2的幂次方 |
|---|---|---|
| 8 | 0000 1000 | 是 |
| 30 | 0001 1110 | 否 |
| 666 | 001010011010 | 否 |
| 1024 | 0100 0000 0000 | 是 |
| 2000 | 0111 1101 0000 | 否 |
| 4096 | 0001 0000 0000 0000 | 是 |
也就是說，如果我們可以判斷：一個數的二進制上隻有一個bit位為1，那麼這個數肯定為2的幂次方。問題發生了等價轉換，那麼我們如何判斷 一個數的二進制中包含幾個1呢？？？。【這是面試中的一個常見問題和技巧】。方法就是：x & (x -1)==0, 則這個數二進制中隻有一個1，否則包含多個1。通常使用這個方法來計算一個數中包含幾個1。
[2] 《劍指offer》面試題15：二進制中1的個數
/*求一個數的二進制中1的個數*/
int numberof1(int n)
{
int count = 0;
while(n){
count++;
n = n & (n-1);
}
return count;
}
簡單的說：x & (n -1)會将x二進制中最低位上的1置為0(最後一個1置為0)。是以如果n&(n-1)==0，那個說明這個數二進制中隻有一個bit位為1，是以肯定是2的幂次方。
3.2 求不小于某個數2的整數次幂
我看還是直接看核心實作吧：
static __inline__ int generic_fls(int x)
{
int r = 32;
if (!x)
return 0;
if (!(x & 0xffff0000u)) {
x <<= 16;
r -= 16;
}
if (!(x & 0xff000000u)) {
x <<= 8;
r -= 8;
}
if (!(x & 0xf0000000u)) {
x <<= 4;
r -= 4;
}
if (!(x & 0xc0000000u)) {
x <<= 2;
r -= 2;
}
if (!(x & 0x80000000u)) {
x <<= 1;
r -= 1;1
}
return r;
}
static inline unsigned long __attribute_const__ roundup_pow_of_two(unsigned long x)
{
return (1UL << generic_fls(x - 1));
}
這個效率嘛？ 由于全是位運算，肯定為求模、取餘等四則運算效率要高, 不能放過任何一點可以優化的地方。至于這樣做的原理，自己品品吧，也是相當經典的存在。
[email protected]:/home/toney# ./a.out
12 --- output=4
16 --- output=5
24 --- output=5
32 --- output=6
128 --- output=8
1024 --- output=11
1400 --- output=11
2040 --- output=11
3.3 為什麼要求2的幂次方呢？
為了使用位運算，快， 快，不擇手段的快
4. Kfifo入隊和出隊
__kfifo_put
是Kfifo的入隊函數，源碼實作如下：
unsigned int __kfifo_put(struct kfifo *fifo,
unsigned char *buffer, unsigned int len)
{
unsigned int l;
len = min(len, fifo->size - fifo->in + fifo->out);
/* first put the data starting from fifo->in to buffer end */
l = min(len, fifo->size - (fifo->in & (fifo->size - 1)));
memcpy(fifo->buffer + (fifo->in & (fifo->size - 1)), buffer, l);
/* then put the rest (if any) at the beginning of the buffer */
memcpy(fifo->buffer, buffer + l, len - l);
fifo->in += len;
return len;
}
需要說明的是Linux 2.6.12版本的核心實作中并沒有使用記憶體屏障，而在後續版本中添加了記憶體屏障，它是實作無鎖隊列的核心和關鍵。這裡我們就按照Linux2.6.12版本實作來說明簡單原理，關于記憶體屏障，可以參考我的另一篇博文《什麼是記憶體屏障？ Why Memory Barriers ?》
| 第6行 | (in - out)表示使用的空間，size - (in - out)則表示剩餘的空間。通過min()來防止寫越界。 |
|---|---|
| 第9行 | in & (size - 1)表示in落在size空間指針的位置，作用相當于in%size, 但位運算效率更高。通過min擷取fifo右側剩餘空間大小，防止越界 |
| 第10行 | 将資料拷貝到fifo右側剩餘空間 |
| 第13行 | len-l表示右側空間不足時，左側需要填充的資料長度 |
| 第15行 | 移動in指針位置 |
__kfifo_put
( )是Kfifo的出隊函數，源碼實作如下：
unsigned int __kfifo_get(struct kfifo *fifo,
unsigned char *buffer, unsigned int len)
{
unsigned int l;
len = min(len, fifo->in - fifo->out);
/* first get the data from fifo->out until the end of the buffer */
l = min(len, fifo->size - (fifo->out & (fifo->size - 1)));
memcpy(buffer, fifo->buffer + (fifo->out & (fifo->size - 1)), l);
/* then get the rest (if any) from the beginning of the buffer */
memcpy(buffer + l, fifo->buffer, len - l);
fifo->out += len;
return len;
}
連個if都不想用，真是太摳門了，哎。你多少if-else判斷下in,out,len的關系，能讓我舒服點呀！！！
4.1 Kfifo右側入隊
當fifo右側剩餘的空間充足時，即size - in%size > len時，直接将資料填充到右側即可，位置為[in, in+len]。
l = min(len, fifo->size - (fifo->in & (fifo->size - 1)));
memcpy(fifo->buffer + (fifo->in & (fifo->size - 1)), buffer, l);
in % size
如何高效表示呢？ 對，就是
in & (size - 1)
。這裡有一個前提：那就是需要size是2的幂次方。Why ?
首先，
in % size
的範圍為[0, size-1];
in & (size -1)
的範圍為[0, size-1]。
其次，它的原理是：size為2的幂次方，size -1則表示【0，size-1】每一個bit位都是1，可以得到該範圍的所有值，這也是要求size為2的幂次方的原因。
最後，兩者在本質上是等價的，但是
in & (size -1)
隻進行位操作，效率高很多。
4.2 Kfifo右側+左側入隊
當右側長度不夠入隊長度時，需要在kfifo左側入隊，此時kfifo左右的範圍為【0，len-l】，左側的範圍為【in，in+l】。
/* first put the data starting from fifo->in to buffer end */
l = min(len, fifo->size - (fifo->in & (fifo->size - 1)));
memcpy(fifo->buffer + (fifo->in & (fifo->size - 1)), buffer, l);
/* then put the rest (if any) at the beginning of the buffer */
memcpy(fifo->buffer, buffer + l, len - l);
4.3 無符号整數溢出回繞
首先看一個例子：
void main()
{
unsigned int a = 0xfffffffa;
unsigned int b = a + 10;
unsigned int c = 4;
printf("a = %u\n",a);
printf("b = %u\n",b);
printf("b - a =%d\n",b-a);
printf("c - a =%d\n",c-a);
}
結果如下：
[email protected]:/home/toney# gcc kfifo.c
[email protected]:/home/toney# ./a.out
a = 4294967290
b = 4
b - a =10
c - a =10
[email protected]:/home/toney#
解釋如下：
a = 4294967290;
b = 4; //a + 10溢出4，--> 0x1 00 00 00 04
但是unsigned int為4位元組共計32位，是以最高位無法擷取，b隻能擷取後32bit,即0x00 00 00 04
b - a = -4294967285;即 0x1 FF FF FF F6
6 : 0110 --> 反碼 1001 = 9
-4294967285在記憶體中的存儲方式為：補碼=反碼+1，即0x1 00 00 00 09 +1 = 0x1 00 00 00 0a
是以b - a = 10;
是以，無論何時，即使發生整數回繞，kfifo中的變量都有如下關系：
妙不可言呀！
可惜我體會還是沒有那麼深刻。
4. 體會
看完kfifo的實作，最大的感覺就是？ 不不，文明人說文明話，妙，是真的妙不可言。如果說這代碼是我或者同僚寫的，我會覺得裡面會不會有很多bug,但是如果為核心大佬寫的，我覺得沒有，就是沒有，真的沒有呀！！！