微優化之八 Pipeline

來自專欄 Zero Switch

前面的章節多是理論，而這一章則更偏重方法論。

首先， 對比三個概念：

• run-to-complete + 線程池
• SEDA
• CPU instruction pipeline

其中，前兩者常常用於服務端開發。後兩者都是將一個完整的任務分成若干個階段，並且，每個階段使用獨立的「硬體」（虛擬的也算，比如線程）。CPU instruction pipeline大家或多或少都知道，通過將一個指令的各個階段流水線化，它提供了更高的吞吐，雖然每條指令本身的執行時間沒有發生變化（可以還稍微變長了一些）。這看起來很神奇，畢竟，我們的OPS提高了！

問題來了，CPU instruction pipeline與SEDA的區別在哪裡？兩者是子集關係嗎？思考一下可知，SEDA是一種通用並發處理模式，它用於處理並發任務，在這一點上，它與run-to-complete+線程池的用途相同。而CPU instruction pipeline，它用於處理單個執行流，而不是並發！

考慮run-to-complete+線程池與SEDA的區別。假設我們有一堆同質的並發任務，每個任務分成三個階段，每個階段用時10ms。假設我們總共有三個線程。

我們使用run-to-complete+線程池模式時，三個線程可以提供的OPS是1000/30 * 3= 100。

我們使用SEDA，假設三個線程組成鏈，每個線程負責一個階段，三個線程可以提供的OPS是1000 / (30 / 3) = 100。

這意味著什麼？意味著在處理並發任務時，線程池與SEDA（並發形式的pipeline）沒有本質上的區別。

那麼，為什麼CPU instruction pipeline可以提升OPS，而SEDA不能呢？畢竟，兩者看起來都是pipeline啊？

其本質在於，CPU instruction pipeline在處理不可並發的任務時，使用了額外的硬體！也即，指令處理的五個階段中，每個階段有專門的硬體負責，且這些硬體之間可以獨立工作。換成應用層的術語，在處理一個不可並發的任務時，我們將其分成數個階段，每個階段用單獨的線程來處理。由於任務不可並發，因此，線程池模式無法適用，但是，我們通過另外一種方法增加了線程數，實現了：任務處理延遲不變，但是OPS提升了！

下圖是將不可並發的任務Pipeline化的示意圖。

每個階段是一個獨立的線程來執行，每個階段之間通過消息隊列來移交任務。

問題又來了，一旦涉及到多實體通信，必然要考慮流控的問題。如果三個階段處理時間相等，則不需要考慮擁塞問題，整個任務處理的OPS可以提供三倍。而如果有某一階段處理過慢，其他階段只能stall。不妨設置三個階段用時分別為t1/t2/t3，於是，整個系統的OPS提升可以形式化為： 。

分析了t1/t2/t3的絕對大小的影響，我們再來考慮下其相對大小的影響。當三者差別不大的，簡單的自旋就可以滿足需要的，還可以減少切換帶來的開銷。但是，如果三者速度差別較大，則使用阻塞隊列是一個更好的選擇。

來看一段代碼：

#include <iostream>#include <thread>#include <chrono>#include <boost/lockfree/spsc_queue.hpp>constexpr int queue_size = 1024;using namespace std;using namespace std::literals::chrono_literals;auto now(){ return std::chrono::high_resolution_clock::now();}template <class Q>void pipeline2(Q* q, const int b){ int x {}; auto start = now(); for (;;) { if (q->pop(x)) { if (x == -1) { break; } this_thread::sleep_for(chrono::milliseconds(b)); } } auto end = now(); std::cout << "pipeline-2:" << (end - start).count() << std::endl;}template <class Q>void pipeline1(Q* q, const int a){ auto start = now(); for (int i = 0; i < queue_size; ++i) { this_thread::sleep_for(chrono::milliseconds(a)); while (!q->push(i)) { } } auto end = now(); q->push(-1); std::cout << "pipeline-1:" << (end - start).count() << std::endl;}void bench(const int x, const int y){ std::cout << "benchmark " << x << "ms vs " << y << "ms
"; using namespace boost::lockfree; spsc_queue<int, capacity<queue_size>> q; thread a([&]{ pipeline2(&q, y); }); pipeline1(&q, x); a.join(); auto start = now(); for (int i = 0; i < queue_size; ++i) { this_thread::sleep_for(chrono::milliseconds(x)); this_thread::sleep_for(chrono::milliseconds(y)); } auto end = now(); std::cout << "non-pipe: " << (end - start).count() << std::endl; std::cout << std::endl;}int main(){ bench(1, 10); bench(5, 10); bench(10, 10); bench(10, 5); bench(10, 1); bench(1, 100); bench(100, 1);}

這段代碼將一個不可並發的任務分成了兩個階段，兩個階段的用時分別為x/y。我們測試了pipeline形式下與非pipeline形式下的處理時間。Pipeline模式下，兩個線程使用spsc通信，出現擁塞是就自旋。

結果如下：

benchmark 1ms vs 10mspipeline-1:1164588517pipeline-2:10382503731non-pipe: 11534735580benchmark 5ms vs 10mspipeline-1:5265016746pipeline-2:10386036324non-pipe: 15644003038benchmark 10ms vs 10mspipeline-1:10388877083pipeline-2:10399001125non-pipe: 20756820644benchmark 10ms vs 5mspipeline-1:10313984625pipeline-2:10319078419non-pipe: 15639583040benchmark 10ms vs 1mspipeline-1:10313558976pipeline-2:10314662356non-pipe: 11529398090benchmark 1ms vs 100mspipeline-1:1144638848pipeline-2:102534712471non-pipe: 103702172237benchmark 100ms vs 1mspipeline-1:102470880332pipeline-2:102471848358non-pipe: 103694614282

我們來分析幾個典型結果：

• 1ms vs 10ms：此時，對於第二個階段，其在10382503731內處理了1024個任務，說明有1024個任務被系統處理完畢，從而，系統OPS為10382503731/1024。對比非pipeline形式，性能提升不顯著。
• 5ms vs 10ms：此時，系統OPS由第二階段決定，由於x、y相差比較小，根據公式，性能提升強於1vs10的情況。
• 10ms vs 10ms：此時，兩個階段速度匹配，性能有一倍的提升，符合公式。
• 10ms vs 1ms：根據公式，其性能提升與1ms vs 10ms相同，但是，每一階段stall的時間較長，從而消耗了不少CPU。這種情況下，實際上可以採用阻塞隊列，而不是自旋。

Pipeline原理其實比較簡單，但是，如果有應用於實踐，一定要benchmark，確保你的優化真的是個優化，而不是劣化！