Linux kernal 鬼斧神工，博大精深，讓人歎為觀止，拍手叫絕。然匠心獨運的設計並非撲朔迷離、盤根錯節，真正的匠心獨運乃辭簡理博、化繁為簡，在簡潔中昭顯優雅和智慧，kfifo就是這樣一種數據結構，它就是這樣簡約高效，匠心獨運，妙不可言，下面就跟大家一起探討學習。
kfifo是一種"First In First Out “數據結構，它采用了前面提到的環形緩沖區來實現，提供一個無邊界的字節流服務。采用環形緩沖區的好處為，當一個數據元素被用掉後，其余數據元素不需要移動其存儲位置，從而減少拷貝提高效率。更重要的是，kfifo采用了並行無鎖技術，kfifo實現的單生產/單消費模式的共享隊列是不需要加鎖同步的。
1: struct kfifo {
2: unsigned char *buffer; /* the buffer holding the data */
3: unsigned int size; /* the size of the allocated buffer */
4: unsigned int in; /* data is added at offset (in % size) */
5: unsigned int out; /* data is extracted from off. (out % size) */
6: spinlock_t *lock; /* protects concurrent modifications */
7: };buffer 用於存放數據的緩存 size 緩沖區空間的大小，在初化時，將它向上圓整成2的冪 in 指向buffer中隊頭 out 指向buffer中的隊尾 lock 如果使用不能保證任何時間最多只有一個讀線程和寫線程，必須使用該lock實施同步。
它的結構如圖：
這看起來與普通的環形緩沖區沒有什麼差別，但是讓人歎為觀止的地方就是它巧妙的用 in 和 out 的關系和特性，處理各種操作，下面我們來詳細分析。
首先，看一個很有趣的函數，判斷一個數是否為2的次冪，按照一般的思路，求一個數n是否為2的次冪的方法為看 n % 2 是否等於0， 我們知道“取模運算”的效率並沒有 “位運算” 的效率高，有興趣的同學可以自己做下實驗。下面再驗證一下這樣取2的模的正確性，若n為2的次冪，則n和n-1的二進制各個位肯定不同 (如8(1000)和7(0111))，&出來的結果肯定是0；如果n不為2的次冪，則各個位肯定有相同的 (如7(0111) 和6(0110))，&出來結果肯定為0。是不是很巧妙？
1: bool is_power_of_2(unsigned long n)
2: {
3: return (n != 0 && ((n & (n - 1)) == 0));
4: }
再看下kfifo內存分配和初始化的代碼，前面提到kfifo總是對size進行2次冪的圓整，這樣的好處不言而喻，可以將kfifo->size取模運算可以轉化為與運算，如下：
kfifo->in % kfifo->size 可以轉化為 kfifo->in & (kfifo->size – 1)
“取模運算”的效率並沒有 “位運算” 的效率高還記得不，不放過任何一點可以提高效率的地方。
1: struct kfifo *kfifo_alloc(unsigned int size, gfp_t gfp_mask, spinlock_t *lock)
2: {
3: unsigned char *buffer;
4: struct kfifo *ret;
5:
6: /*
7: * round up to the next power of 2, since our 'let the indices
8: * wrap' technique works only in this case.
9: */
10: if (!is_power_of_2(size)) {
11: BUG_ON(size > 0x80000000);
12: size = roundup_pow_of_two(size);
13: }
14:
15: buffer = kmalloc(size, gfp_mask);
16: if (!buffer)
17: return ERR_PTR(-ENOMEM);
18:
19: ret = kfifo_init(buffer, size, gfp_mask, lock);
20:
21: if (IS_ERR(ret))
22: kfree(buffer);
23:
24: return ret;
25: }
為什麼kfifo實現的單生產/單消費模式的共享隊列是不需要加鎖同步的呢？天底下沒有免費的午餐的道理人人都懂，下面我們就來看看kfifo實現並發無鎖的奧秘。
我們知道 編譯器編譯源代碼時，會將源代碼進行優化，將源代碼的指令進行重排序，以適合於CPU的並行執行。然而，內核同步必須避免指令重新排序，優化屏障（Optimization barrier）避免編譯器的重排序優化操作，保證編譯程序時在優化屏障之前的指令不會在優化屏障之後執行。
舉個例子，如果多核CPU執行以下程序：
1: a = 1;
2: b = a + 1;
3: assert(b == 2);
假設初始時a和b的值都是0，a處於CPU1-cache中，b處於CPU0-cache中。如果按照下面流程執行這段代碼：
1 CPU0執行a=1;
2 因為a在CPU1-cache中，所以CPU0發送一個read invalidate消息來占有數據
3 CPU0將a存入store buffer
4 CPU1接收到read invalidate消息，於是它傳遞cache-line，並從自己的cache中移出該cache-line
5 CPU0開始執行b=a+1;
6 CPU0接收到了CPU1傳遞來的cache-line，即“a=0”
7 CPU0從cache中讀取a的值，即“0”
8 CPU0更新cache-line，將store buffer中的數據寫入，即“a=1”
9 CPU0使用讀取到的a的值“0”，執行加1操作，並將結果“1”寫入b（b在CPU0-cache中，所以直接進行）
10 CPU0執行assert(b == 2); 失敗
軟件可通過讀寫屏障強制內存訪問次序。讀寫屏障像一堵牆，所有在設置讀寫屏障之前發起的內存訪問，必須先於在設置屏障之後發起的內存訪問之前完成，確保內存訪問按程序的順序完成。Linux內核提供的內存屏障API函數說明如下表。內存屏障可用於多處理器和單處理器系統，如果僅用於多處理器系統，就使用smp_xxx函數，在單處理器系統上，它們什麼都不要。
smp_rmb適用於多處理器的讀內存屏障。
smp_wmb適用於多處理器的寫內存屏障。
smp_mb適用於多處理器的內存屏障。
如果對上述代碼加上內存屏障,就能保證在CPU0取a時，一定已經設置好了a = 1:
1: void foo(void)
2: {
3: a = 1;
4: smp_wmb();
5: b = a + 1;
6: }
這裡只是簡單介紹了內存屏障的概念，如果想對內存屏障有進一步理解,請參考我的譯文《為什麼需要內存屏障》。
__kfifo_put是入隊操作，它先將數據放入buffer中，然後移動in的位置，其源代碼如下：
1: unsigned int __kfifo_put(struct kfifo *fifo,
2: const unsigned char *buffer, unsigned int len)
3: {
4: unsigned int l;
5:
6: len = min(len, fifo->size - fifo->in + fifo->out);
7:
8: /*
9: * Ensure that we sample the fifo->out index -before- we
10: * start putting bytes into the kfifo.
11: */
12:
13: smp_mb();
14:
15: /* first put the data starting from fifo->in to buffer end */
16: l = min(len, fifo->size - (fifo->in & (fifo->size - 1)));
17: memcpy(fifo->buffer + (fifo->in & (fifo->size - 1)), buffer, l);
18:
19: /* then put the rest (if any) at the beginning of the buffer */
20: memcpy(fifo->buffer, buffer + l, len - l);
21:
22: /*
23: * Ensure that we add the bytes to the kfifo -before-
24: * we update the fifo->in index.
25: */
26:
27: smp_wmb();
28:
29: fifo->in += len;
30:
31: return len;
32: }6行，環形緩沖區的剩余容量為fifo->size - fifo->in + fifo->out，讓寫入的長度取len和剩余容量中較小的，避免寫越界； 13行，加內存屏障，保證在開始放入數據之前，fifo->out取到正確的值（另一個CPU可能正在改寫out值） 16行，前面講到fifo->size已經2的次冪圓整，而且kfifo->in % kfifo->size 可以轉化為 kfifo->in & (kfifo->size – 1)，所以fifo->size - (fifo->in & (fifo->size - 1)) 即位 fifo->in 到 buffer末尾所剩余的長度，l取len和剩余長度的最小值，即為需要拷貝l 字節到fifo->buffer + fifo->in的位置上。 17行，拷貝l 字節到fifo->buffer + fifo->in的位置上，如果l = len，則已拷貝完成，第20行len – l 為0，將不執行，如果l = fifo->size - (fifo->in & (fifo->size - 1)) ，則第20行還需要把剩下的 len – l 長度拷貝到buffer的頭部。 27行，加寫內存屏障，保證in 加之前，memcpy的字節已經全部寫入buffer，如果不加內存屏障，可能數據還沒寫完，另一個CPU就來讀數據，讀到的緩沖區內的數據不完全，因為讀數據是通過 in – out 來判斷的。 29行，注意這裡 只是用了 fifo->in += len而未取模，這就是kfifo的設計精妙之處，這裡用到了unsigned int的溢出性質，當in 持續增加到溢出時又會被置為0，這樣就節省了每次in向前增加都要取模的性能，锱铢必較，精益求精，讓人不得不佩服。 __kfifo_get是出隊操作，它從buffer中取出數據，然後移動out的位置，其源代碼如下：
1: unsigned int __kfifo_get(struct kfifo *fifo,
2: unsigned char *buffer, unsigned int len)
3: {
4: unsigned int l;
5:
6: len = min(len, fifo->in - fifo->out);
7:
8: /*
9: * Ensure that we sample the fifo->in index -before- we
10: * start removing bytes from the kfifo.
11: */
12:
13: smp_rmb();
14:
15: /* first get the data from fifo->out until the end of the buffer */
16: l = min(len, fifo->size - (fifo->out & (fifo->size - 1)));
17: memcpy(buffer, fifo->buffer + (fifo->out & (fifo->size - 1)), l);
18:
19: /* then get the rest (if any) from the beginning of the buffer */
20: memcpy(buffer + l, fifo->buffer, len - l);
21:
22: /*
23: * Ensure that we remove the bytes from the kfifo -before-
24: * we update the fifo->out index.
25: */
26:
27: smp_mb();
28:
29: fifo->out += len;
30:
31: return len;
32: }
6行，可去讀的長度為fifo->in – fifo->out，讓讀的長度取len和剩余容量中較小的，避免讀越界；
13行，加讀內存屏障，保證在開始取數據之前，fifo->in取到正確的值（另一個CPU可能正在改寫in值）
16行，前面講到fifo->size已經2的次冪圓整，而且kfifo->out % kfifo->size 可以轉化為 kfifo->out & (kfifo->size – 1)，所以fifo->size - (fifo->out & (fifo->size - 1)) 即位 fifo->out 到 buffer末尾所剩余的長度，l取len和剩余長度的最小值，即為從fifo->buffer + fifo->in到末尾所要去讀的長度。
17行，從fifo->buffer + fifo->out的位置開始讀取l長度，如果l = len，則已讀取完成，第20行len – l 為0，將不執行，如果l =fifo->size - (fifo->out & (fifo->size - 1)) ，則第20行還需從buffer頭部讀取 len – l 長。 27行，加內存屏障，保證在修改out前，已經從buffer中取走了數據，如果不加屏障，可能先執行了增加out的操作，數據還沒取完，令一個CPU可能已經往buffer寫數據，將數據破壞，因為寫數據是通過fifo->size - (fifo->in & (fifo->size - 1))來判斷的 。 29行，注意這裡 只是用了 fifo->out += len 也未取模，同樣unsigned int的溢出性質，當out 持續增加到溢出時又會被置為0，如果in先溢出，出現 in < out 的情況，那麼 in – out 為負數（又將溢出），in – out 的值還是為buffer中數據的長度。 這裡圖解一下 in 先溢出的情況，size = 64， 寫入前 in = 4294967291， out = 4294967279 ，數據 in – out = 12; 寫入 數據16個字節，則 in + 16 = 4294967307，溢出為 11，此時 in – out = –4294967268，溢出為28，數據長度仍然正確，由此可見，在這種特殊情況下，這種計算仍然正確，是不是讓人歎為觀止，妙不可言？kfifo設計精巧，妙不可言，但主要為內核提供服務，內存屏障函數也主要為內核提供服務，並未開放出來，但是我們學習到了這種設計巧妙之處，就可以依葫蘆畫瓢，寫出自己的並發無鎖環形緩沖區，這將在下篇文章中給出，至於內存屏障函數的問題，好在gcc 4.2以上的版本都內置提供__sync_synchronize()這類的函數，效果相差不多。《並發無鎖環形隊列的實現》給出自己的並發無鎖的實現，有興趣的朋友可以參考一下。