（一）起因 （二）混合自旋鎖 （三）q3.h 與 RingBuffer
（四）RingQueue(上) 自旋鎖 （五）RingQueue(中) 休眠的藝術
（六）RingQueue(中) 休眠的藝術 [續]
這是第五篇的後續，這部分的內容同時會更新和添加在 第五篇：RingQueue(中) 休眠的藝術 一文的末尾。
緊接上一篇的末尾，我們把 Windows 和 Linux 下的休眠策略歸納總結一下，如下圖：
sched_yield() 雖然包括了 Windows 下的 Sleep(0) 和 SwitchToThread() 的部分功能（圖中藍色框和虛線框所標注的部分），但缺少了上圖中兩個灰色文字的功能，即 ① SwitchToOtherCoresLowerThreads() 和 ② SwitchToLocalCoreLowerThreads()（其實這個函數應該包括相同優先級的情況，但由於名字太長，故省略了Equal）。而 Windows 下，則缺失了 ① SwitchToOtherCoresLowerThreads() 這個功能。也就是說，Linux 下面沒有切換到低優先級線程的功能，而 Windows 雖然提供了 SwitchToThread()，可以切換到本核心上的低優先級線程，卻也依然缺少了切換到別的核心上低優先級線程的能力，即 ① SwitchToOtherCoresLowerThreads() 這個功能。
也就是說，Windows 和 Linux 的策略即使合起來用，依然是不完整的。可是我仔細的想了一下，上面的說法其實是不太正確的。sched_yield() 其實並不是不能切換到低優先級的線程，根據上篇提到 Linux 上的三種不同調度策略，sched_yiled() 是有可能切換到比自己優先級低的線程上的，比如 SCHED_FIFO（先進先出策略）或者 SCHED_RR（輪叫策略剛好輪到低優先級的線程）。只不過，不能特別指定切換到當前線程所在的核心上的其他線程，也就是類似 Windows 上 SwitchToThread() 的功能。
而 Windows 上，雖然看起來好像更完整，卻也真正的缺少了切換到別的核心上的低優先級線程的功能，即 ① SwitchToOtherCoresLowerThreads()。
首先來看 Linux 上，缺少了只切換到當前核心的等待線程上的功能，如果切換的線程原來是在別的核心上的，那麼有可能會導致切換到的線程所使用的緩存失效，被迫重新加載，導致影響性能。（這個不知道 Linux 的策略是如何選擇的，可能要看下源碼才能明白，但是有一點是可以肯定的，就算 Linux 在某些調度策略裡（比如 SCHED_OTHER，這是普通線程所采用的策略），會優先選擇本來就在當前核心上等待的線程，但是當當前核心上沒有適合的等待線程的時候，Windows 的 SwitchToThread() 會立刻返回，而 Linux 應該還是會選擇別的核心上的其他等待線程來執行；而如果是另外兩種調度策略的話，則基本上可能不會優先選擇當前核心上的等待線程，尤其是 SCHED_FIFO 策略）。
但從總體上來看，Linux 使用了三種不同的調度策略，是要比 Windows 單一的輪叫循環策略要好一點，至少你有可能通過不同的策略的組合來實現更有效的調度方案。
而 Windows 上雖然在一定程度上有避免這種線程遷移到別的 CPU 核心上運行的可能，卻由於缺少切換到別的核心上低優先級線程的功能，而導致策略上的不完整，雖然 Windows 可能可以根據線程饑餓程度來決定要不要切換到這些本不能切換過去的低優先級線程，不過這個過程肯定是比較緩慢的，不太可控的，從而在某些特殊情況下可能會造成某種程度上的“死鎖”。
所以 Linux 和 Windows 的調度策略各有千秋，總體上來看，Linux 好像稍微好那麼一點，因為畢竟選擇要多一些，只要設計得當，情況可能會好一點。但總體上，兩者都有不足和缺陷，不夠完整。
那麼怎樣才能更完整呢？答案可能你也能猜到，通過前面的分析，我們想提供更完整的細節和可控的參數，讓調度更完整和隨心所欲。也就是根據我們的想法，指哪打哪，想切去哪就切去哪，無孔不入。我們需要一個更強大的接口，這個接口應該要考慮幾乎所有可能的情況，功能強大而不失靈活性，甚至有些時候還會有一定的侵略性。我們只是提供一個接口，具體怎麼玩，玩成什麼樣，我們不管，玩崩了那是程序員自己的事情（崩倒是應該不會，但是可能會混亂）。
我們把這個接口暫時定義為 scheduler_switch()（本來想叫 scheduler_switch_thread()，還是短一點好），函數模型大致為：
int scheduler_switch(pthread_array_t *threads, pthread_priority_t priority_threshold, int priority_type, cpuset_t cpu_mask, int force_now, int slice_count);
threads： 表示一組線程，我們從這一組線程裡，通過後面的幾個參數一起來決定最終選擇切換到哪一個線程，該線程必需是處於准備就緒狀態的，即會把已經在運行的線程排除掉。該值為 NULL 時表示從系統所有的等候線程裡挑選，即跟 sched_yield() 默認的行為一致。
priority_threshold：表示線程優先級的閥值，由後面的 priority_type 來決定是高於這個閥值、低於這個閥值、等於這個閥值或者大於等於這個閥值，等等。該值為 -1 時表示使用當前線程的優先級。
priority_type： 決定 priority_threshold 的比較類型，可分為：>，<，==，>=，<= 等類型。
cpu_mask： 允許切換到的 CPU 核心的 mask 值，該值為 0 時表示只切換到線程當前所在的核心，不為 0 時，每一個 bit 位表示一個 CPU 核心，該值類似 CPU 親緣性的 cpuset_t 。
force_now： 為 1 時，表示指定的切換線程立刻獲得時間片，而不受系統優先級和調度策略的限制，強制運行的時間片個數由 slice_count 參數決定；為 0 時，表示指定切換的線程會等待系統來決定是否立刻得到時間片運行，可能會被安排到一個很短的等候隊列裡。
slice_count： 表示切換過去後運行的時間片個數，如果為 0 時，則由系統決定具體運行多少個時間片。
返回值： 如果成功切換返回 0，如果切換失敗返回 -1（即沒有可切換的線程）。
我所說的侵略性是，你可以決定切換以後會持續運行多少個時間片而不會被中斷，而且如果你通過 threads 指定的線程組如果沒有立刻得到時間片的權力的時候，可以通過 force_now 參數來強制獲得時間片，即讓系統本來下一個會得到該時間片的線程排在我們指定的線程運行完相應的時間片後再把時間片讓給它。如果你指定的 slice_count 值過大，可能會使得其他線程得不到時間片運行，也許這個值應該設置一個上限，比如 100 個或 256 個時間片之類的。
你可以看到，這個接口函數幾乎囊括了 Windows 和 Linux 已存在的所有調度函數的功能，並且進行了一定程度的增強和擴展，並且有一定的靈活性，如果你覺得還有不夠完整的地方，也可以告訴我。至於如何實現它，我們並不關心，我們只要知道理論上是否可以實現即可，而具體的實現方法可以通過研究 Linux 內核源碼來辦到。也許並不一定很容易實現，但從理論上，我們還是有可能做得到的。Windows 上由於不開源，我們沒什麼辦法，也許研究 ReactOS 是一種選擇，但是 ReactOS 是基於 WindowsNT 內核的，技術可能有些陳舊（注：ReactOS 是一個模仿 Windows NT 和 Windows 2000 的開源操作系統項目，請參考 [wikipedia：ReactOS] ）。
下載 Linux Kernel 源碼：
https://www.kernel.org/
（建議下載 2.6.32.65 和 最新的穩定版 3.18.4，Android 是在 2.6 的基礎上修改的，有興趣也可以直接拿 Android 的內核來改。）
Ubuntu 14.04 LTS 的內核版本是 3.13 （參考：Ubuntu發行版列表）
可參考的文章：
Linux內核調度算法（1）-- 快速找到最高優先級進程
Linux內核--進程調度（一）
IBM：Linux 2.6 調度系統分析
注：通過閱讀 Linux 內核的源碼，sched_yield() 真正的執行過程並不是我前面描述的那樣的，正確的過程應該是：先在當前線程所在的核心上遍歷合適的任務線程，如果有的話，就切換到該線程；如果沒有的話，目前我還不確定會不會從別的核心上的 RunQueue 遷移任務線程到當前核心來運行，按道理講應該是這樣的。如果是這樣，那麼 sched_yield() 跟 Windows 的 Sleep(0) 的唯一區別就是：Sleep(0) 不會切換到比自己的優先級別低的線程，而 sched_yield() 可以，Sleep(0) 的策略也應該跟 sched_yield() 類似，先在被核心上找，沒有再去別的核心上找。但是，這裡我就不修正前面的描述了，請自行注意一下。
因為 3.18.4 的 Linux 改動比較大，也變得復雜了很多，比較難讀懂，找 schedule() 函數的位置都花了很大的力氣……，因此這裡以 2.6 版本的為例。
通過閱讀 2.6.32.65 版本的 Linux 內核源碼，我們知道實現系統調度的函數是：schedule(void)，這跟我們這個接口比較類似，實現的功能也是大致一樣。schedule(void) 的功能是從當前核心上的 RunQueue 裡找到下一個可運行的任務線程，並切換MMU、寄存器狀態以及堆棧值，即通常說的上下文切換（Context Switch）。
通過 /include/linux/smp.h 和 /arch/x86/include/asm/smp.h 或 /arch/arm/include/asm/smp.h，我們可以得知，smp_processor_id() 返回的是 CPU 核心的一個編號，x86 上是通過 percpu_read(cpu_number)，arm 上是通過 current_thread_info()->cpu 獲得。這樣 rq = cpu_rq(cpu); 取得的 RunQueue 就是每個 CPU 核心上獨立的運行（任務線程）隊列。
/*
* schedule() is the main scheduler function.
*/
asmlinkage void __sched schedule(void)
{
struct task_struct *prev, *next;
unsigned long *switch_count;
struct rq *rq;
int cpu;
need_resched:
preempt_disable();
cpu = smp_processor_id(); /* 獲取當前 CPU 核心編號 */
rq = cpu_rq(cpu);
rcu_sched_qs(cpu);
prev = rq->curr;
switch_count = &prev->nivcsw;
release_kernel_lock(prev);
need_resched_nonpreemptible:
schedule_debug(prev);
if (sched_feat(HRTICK))
hrtick_clear(rq);
spin_lock_irq(&rq->lock);
update_rq_clock(rq);
clear_tsk_need_resched(prev);
if (prev->state && !(preempt_count() & PREEMPT_ACTIVE)) {
if (unlikely(signal_pending_state(prev->state, prev)))
prev->state = TASK_RUNNING;
else
deactivate_task(rq, prev, 1);
switch_count = &prev->nvcsw;
}
pre_schedule(rq, prev); /* 准備調度? */
if (unlikely(!rq->nr_running))
idle_balance(cpu, rq);
put_prev_task(rq, prev); /* 記錄之前任務線程的運行時間, 更新平均運行時間和平均overlap時間. */
next = pick_next_task(rq); /* 從最高優先級的線程類別開始遍歷, 直到找到下一個可運行的任務線程. */
if (likely(prev != next)) {
sched_info_switch(prev, next);
perf_event_task_sched_out(prev, next, cpu);
rq->nr_switches++;
rq->curr = next;
++*switch_count;
/* 上下文切換, 同時釋放 runqueue 的自旋鎖. */
context_switch(rq, prev, next); /* unlocks the rq */
/*
* the context switch might have flipped the stack from under
* us, hence refresh the local variables.
*/
cpu = smp_processor_id();
rq = cpu_rq(cpu);
} else
spin_unlock_irq(&rq->lock);
post_schedule(rq);
if (unlikely(reacquire_kernel_lock(current) < 0))
goto need_resched_nonpreemptible;
preempt_enable_no_resched();
/* 是否需要重新調度 */
if (need_resched())
goto need_resched;
}
EXPORT_SYMBOL(schedule);
其中有一個很關鍵的函數是 pick_next_task(rq)，作用是尋找下一個最高優先級的可運行任務。有一點可以確定的是，是優先在本核心上的 RunQueue 上遍歷的，但看不太出來是否會在本核心上沒有適合的任務線程時，會從別的核心上的 RunQueue 遷移任務過來，這個部分可能要看 struct sched_class 的 pick_next_task() 的實現，這個應該是一個函數指針。pick_next_task(rq) 的源碼如下：↓↓↓
/*
* Pick up the highest-prio task:
*/
static inline struct task_struct *
pick_next_task(struct rq *rq)
{
const struct sched_class *class;
struct task_struct *p;
/*
* Optimization: we know that if all tasks are in
* the fair class we can call that function directly:
*/
/* fair_sched_class 是分時調度策略, 包括 SCHED_NORMAL, SCHED_BATCH, SCHED_IDLE 三種策略, */
/* 從 __setscheduler() 函數得知. */
if (likely(rq->nr_running == rq->cfs.nr_running)) {
p = fair_sched_class.pick_next_task(rq);
if (likely(p))
return p;
}
/* sched_class_highest = &rt_sched_class; 即實時調度策略, */
/* 包括 SCHED_FIFO 和 SCHED_RR, 從 __setscheduler() 函數得知. */
class = sched_class_highest;
for ( ; ; ) {
p = class->pick_next_task(rq);
if (p)
return p;
/*
* Will never be NULL as the idle class always
* returns a non-NULL p:
*/
class = class->next;
}
}
這個問題我們先打住，以後再研究，或者交給有興趣的同學去研究，有什麼成果的話麻煩告訴我一下。
我們回顧一下 scheduler_switch() 的原型，如下：
int scheduler_switch(pthread_array_t *threads, pthread_priority_t priority_threshold, int priority_type, cpuset_t cpu_mask, int force_now, int slice_count);
這裡的 pthread_array_t *threads，表示的一組線程，之所以想設置這個參數，是因為 q3.h 在 push() 和 pop() 的時候，確認提交成功與否的過程是需要序列化的，如果這個過程中的線程調度能夠按我們想要的順序調度，並且加上適當的休眠機制的話，可能可以減少競爭和提高效率。
請注意下面代碼裡第 22 和 23 行：
1 static inline int
2 push(struct queue *q, void *m)
3 {
4 uint32_t head, tail, mask, next;
5 int ok;
6
7 mask = q->head.mask;
8
9 do {
10 head = q->head.first;
11 tail = q->tail.second;
12 if ((head - tail) > mask)
13 return -1;
14 next = head + 1;
15 ok = __sync_bool_compare_and_swap(&q->head.first, head, next);
16 } while (!ok);
17
18 q->msgs[head & mask] = m;
19 asm volatile ("":::"memory");
20
21 /* 這個地方是一個阻塞的鎖, 對提交的過程進行序列化, 即從序號小的到大的一個個依次放行. */
22 while (unlikely((q->head.second != head)))
23 _mm_pause();
24
25 q->head.second = next;
26
27 return 0;
28 }
因為這裡第 22, 23 行並沒有休眠策略，直接是在原地自旋，在競爭比較激烈，且間隔時間很短的情況下（評價一個競爭的狀況有兩個參數，一個是多少人在競爭，另一個參數是產生競爭的頻率，即間隔多久會產生一次競爭。競爭者越多，而且競爭間隔也很短的話，那麼代表競爭是非常激烈的，我們這裡的情況就是這樣），沒有休眠策略很可能導致互相激烈的爭搶資源，而且也會讓 q3.h 在 push() 和 pop() 總線程數大於 CPU 核心總數的時候，產生介於 ”活鎖” 和 “死鎖” 之間的這麼一種異常狀態的問題。此時，隊列推進非常慢，而且有多慢似乎有點看臉，有時候可能要一分鐘或幾分鐘不等。
如果這個時候，能夠把還沒輪到的線程先休眠，然後按次序依次喚醒（是按照我們自己的 sequence 序號 head 來喚醒，而不是線程自己進入休眠狀態的順序，因為從第 16 行結束到第 22 行，這裡的執行順序是不確定的，有可能序號大的線程先執行到第 22 行，而序號小的因為線程被掛起了，因而晚一點才執行到第 22 行，所以喚醒的時候是依據我們自己的序號 head 從小到大依次喚醒）。但是這樣也會出現一個問題，就是當某個核心正在運行的線程（我們設這個CPU 核心為 X，線程為A）執行到第 22 行時，由於序號 head 大於 q->head.second 很多，那麼需要切換到別的線程或者進入休眠狀態。如果此時選擇切換，此時 head = q->head.second 的線程（我們設這個線程為B）會被選取且被喚醒，可是問題來了，這個有時間片的 CPU 核心 X 很可能不是 head = q->head.second 這個的線程 B 原來所在的核心（我們設這個核心為 Y），如果要喚醒的話則需要從原來的核心 Y 遷移到當前的核心 X 來，這當然不是我們想看到的，如果可以在核心 Y 上執行線程 B，這是最理想的選擇了。如果可以的話，我們會中斷核心 Y 上正在運行的線程 C，然後把時間片交給線程 B 來運行。依此類推，那麼有可能下一個被喚醒的線程也不在當前擁有時間片的 CPU 核心上，那麼又將導致類似的中斷。如果這種中斷太多，那麼效率自然也會受到一定影響。所以，如果有一個更好的統籌規劃方法就再好不過了，可是，似乎這樣的規劃策略不太好實現。
要麼我們允許線程頻繁的遷移，或者我們可以按某個比例允許兩種情況都出現，例如：0.4的概率允許被喚醒的線程遷移到當前的核心，0.6的概率不遷移，采用打斷要喚醒的線程所在的核心的方式（總的概率為1.0）。或者我們允許一次喚醒兩個相鄰的線程（指序號 head 鄰近），我們設這兩個線程分別為線程 A 和 B，那麼當 A 通過之後，緊接著就是該喚醒 B 了，我們讓線程 B 早一點喚醒，然後讓其自旋並等待通過（這其實跟現在的 q3.h 很像，不同的是我們加了休眠策略，是可以應付任意線程數的 push() 和 pop() 的）。而且如果兩個線程原來所在的核心剛好交叉的話，即線程 A 原來是在核心 X 上的，線程 B 原來是在核心 Y 上的，現在核心 Y 的線程在請求 sched_yield()，現在要喚醒的線程是 A 和 B，那麼我們讓 B 在核心 Y 上運行，並且打斷核心 X 上的線程，讓其運行線程 A。當然，我們會先執行打斷的過程，讓線程 A 在核心 X 上運行，然後再在核心 Y 上切換到線程 B 運行，如果可以這麼安排先後的話，如果不能這樣做，兩者同時進行也是可以的。如果不是交叉的話，也增加了兩個線程中的其中一個可能在它原來的核心上的機率。
我們為什麼沒有在 pthread_array_t *threads 這一組線程上使用先進先出的隊列呢，一是前面說過的，像 q3.h 這個問題，線程進入休眠的順序不一定是我們想要的順序，第二個原因是，像我們這個問題，本身就是一個 FIFO 隊列問題，裡面再套一個 FIFO 隊列似乎也不合理。但是，的確有些時候還是需要這種 FIFO 的 threads 的，也不矛盾，我們讓先進入的序號值比後進入的序號值小即可。為了簡化邏輯，我們可以用固定數組存儲元素（事先必需先指定隊列大小），用兩個單向鏈表來實現插入和遍歷的方法，一個是active_list，另一個是free_list，遍歷的時候使用序號做為選擇的依據，從而變成一個按序號大小順序出列的隊列。
警告：其實這個部分感覺有點畫蛇添足，不過我寫了就不准備刪了，有一部分也是我曾經思考過的東西，只是寫出來好像沒有想象中的理想，但可以做為一個思考的方向。
其實在 寫完上一篇 到 准備寫這篇文章之前 的某一天，我看到了這麼一篇文章，《條件變量的陷阱與思考》，也可以說是及時雨，感覺好像跟我的文章能沾上一點邊，我因此而特意查看了 glibc 源碼裡關於 pthread_cond_xxxx() 部分的相關代碼，對條件變量有了進一步的認識。由於沒深入研究，只是覺得條件變量的實現並不是想象中的那麼簡單，至少它整個機制跟我前面提到的喚醒進制是不太一樣，我們總想實現一個完美的喚醒機制，事實上可能很難實現我們心中那最完美的方式，因為有些東西在多線程編程中是不可調和的。同時因為 POSIX 規范為了簡化實現，使用的時候跟 Windows 是有很些區別的。看了這篇文章後，才發現我原來對 pthread 的條件變量理解並不到位，原本以為跟 Windows 的 Event (事件) 很類似，雖然我知道一些兩者的區別，主要集中在 CreateEvent() 的手動和自動模式, PulseEvent() 和 pthread_cond_broadcast() 之間的區別。由於我從沒真正的使用過它，所以真正的區別是看了這篇文章才知道的，最大的區別主要來自於實現的邏輯，POSIX 規范為了簡化實現，條件變量必須跟一個 pthread_mutex_t 配合使用才能正確發生作用，這跟 Windows 的 WaitForSingleObject() 和 ResetEvent() 是有些不同的，Windows 上的 Event 相關函數內部已經包含了這個互斥操作了，相當來說比較直觀和易用一些，但 pthread_cond_t 更靈活一些。
也許你會說，既然我們可能在內核裡用 scheduler_switch()，為什麼參數的定義卻用了 pthread 的類型，其實如果原理實現了，改成內核的 kthread，或者內核和 pthread 分別實現兩套函數，也是可以的，這不是大問題。
設計 scheduler_switch() 這個東西是“啟發式”的，我們的目的是想讓休眠策略能夠更完整的為我們服務，如果能改進則更好，實現不了或不好實現也不打緊，也算是拋磚引玉，希望對你有所啟發。
感謝你的閱讀，如果你覺得寫得不錯的話，麻煩點一下推薦，或者評論一下，這樣下次你找這個文章就可以從首頁的 “我贊” 和 “我評” 裡面找到了。
RingQueue 的GitHub地址是：https://github.com/shines77/RingQueue，也可以下載UTF-8編碼版：https://github.com/shines77/RingQueue-utf8。 我敢說是一個不錯的混合自旋鎖，你可以自己去下載回來看看，支持Makefile，支持CodeBlocks， 支持Visual Studio 2008, 2010, 2013等，還支持CMake，支持Windows, MinGW, cygwin, Linux, Mac OSX等等。
（一）起因 （二）混合自旋鎖 （三）q3.h 與 RingBuffer
（四）RingQueue(上) 自旋鎖 （五）RingQueue(中) 休眠的藝術
（六）RingQueue(中) 休眠的藝術 [續]
上一篇：一個無鎖消息隊列引發的血案（五）——RingQueue(中) 休眠的藝術
.
