為mongos構建一個非同步網路層

摘要： 許多人以分片叢集的方式執行MongoDB伺服器。 在這種配置下， mongos位於使用者程式和分片資料之間， 使用者連線mongos並給它傳送查詢， mongos將那些查詢路由到一個或者多個分片上來完成查詢動作。 在大多數情況下， mongos可以將某個查詢精確定位到單一的分片上， 然而，...

許多人以分片叢集的方式執行MongoDB伺服器。 在這種配置下， mongos位於使用者程式和分片資料之間， 使用者連線mongos並給它傳送查詢， mongos將那些查詢路由到一個或者多個分片上來完成查詢動作。

在大多數情況下， mongos可以將某個查詢精確定位到單一的分片上， 然而， 一些查詢需要“分散-集合”的路由， 換句話說， mongos不得不將查詢傳送到所有的分片上，等待它們的響應，並且將它們整合成一個單獨的master響應。我們可以將這些請求序列的放大到所有的分片， 但是這樣一個慢的連線就會造成整個mongos系統的阻塞。 要有效的實現， 我們需要一個併發請求的方式。

考慮MongoDB 3.0 網路相關程式碼的實現結構， 唯一的實現併發執行請求的方式是在不同的執行緒裡面執行請求。有些叢集有數百個分片–這會有大量的請求放大， 你可以想象一下在mongos處理非常多的請求時會發生什麼：執行緒爆炸！！有太多的執行緒會拖慢整個系統， 造成對硬體資源的爭奪。

在MongoDb3.2， 我們寫了另一種實現方式：對mongos的非同步網路出站。這種新的網路層去除了我們的執行緒爆炸問題， 但是這個新的實現帶來了記憶體管理的困難。 我們進行了大量的實驗，失敗，修復， 迭代， 最為重要的是強迫性測試來實現一套新的callback驅動的非同步系統。

網路請求的生命週期

建立一個非同步網路層

要實現這個目的第一件事是需要一些獨立的執行緒。我們不需要幾個mongos執行緒， 每一個連線一個執行緒， 來做排程和等待網路請求的工作， 相反， 我們可以把這個工作交給一個帶有執行緒池的執行引擎來完成。該執行引擎維護了一個work item佇列， 它使用執行緒池從任務佇列中彈出任務並且執行。 標準的ASIO庫提供了執行引擎和很多其它的原語來實現我們新的網路層。

我們需要將我們的任務包裝成小而簡單的work item， 由執行引擎來執行。這意味著沒有任務可以執行阻塞性的工作， 否則就會阻塞住整個引擎。因此， 我們想要開始一個非同步的“已開啟的連線”的任務並且立刻返回， 而不是開啟一個socket並且等待連線。

我們可以把非同步的”開啟的連線“的邏輯打包成一個任務， 排隊加入到執行引擎裡面， 那麼， 當“開啟的連線”執行完成， 它可以加入下一個“請求nonce”的非同步任務到引擎裡面。類似的， 一旦我們請求一個nonce， 我們會把任務“get nonce”加入佇列來接受遠端主機的響應。

對每一個加入的任務都排隊到執行引擎裡， 我們按照這種方式處理。 因為每個任務是一個獨立的執行單元， 執行引擎沒有繫結到任何的請求或者連線，它可以並行的處理到不同主機的很多的請求。

當執行引擎最終在一個請求週期執行任務的時候， 它會在mongos觸發callback， 這允許一個mongos執行緒獲得響應並且開始給使用者application生成響應。

事實上， 我們在兩個執行緒池之間來回的傳送請求，一個是為mongos的邏輯， 一個是為網路邏輯。在這個系統裡， 我們可以有一個固定數目或者可以配置數目的執行緒， 而不是每一個連線一個執行緒。

讓我們看一下在我們實現它的過程中我們碰到的幾個技術挑戰。

技術挑戰 #1: 消失的狀態

在我們開始深入研究狀態是如何在我們鼻子下消失之前， 讓我們看一下一些C++提供的特性來幫助我們實現callback驅動的系統。特別的， C++ lambdas是這個專案的重要部分。

Lambdas 任務包

lambda 是一個可呼叫的單元， 在C++， 它由三部分組成：捕獲外部變數列表， 引數， 函式體。捕獲外部變數列表在lambda初始化的時候給現存的變數做快照， 引數是在lambda被呼叫的時候傳遞的，在lambda被呼叫的時候， lambda 函式體被執行。

auto lambda = [capture list](parameters){
// body
};

lambda(); // runs body

讓我們更加仔細的看一下捕獲變數列表，下面的程式碼展示了lambda從外部環境捕獲一個變數N， 並且在後面的程式碼列印N。

int N = 1;
auto print_number = [N](){
std::cout << N << std::endl;
};

print_number(); // prints “1”

lambdas通過傳值或者通過引用能夠捕獲變數， 預設情況下， 它們通過傳值捕獲變數並且進行復制：

int N = 1;
auto print_number = [N](){
// We have our own copy of N
std::cout << N << std::endl;
};

N = 123456;

print_number(); // still prints “1”

當lambdas通過引用捕獲變數的時候， 它們會使用原本的變數， 而不是複製：

int N = 1;
auto print_number = [&N](){
// Now, we use the original N
std::cout << N << std::endl;
};

N = 123456;

print_number(); // prints “123456”

如果我們採用引用外部變數， 我們可以避免代價昂貴的變數複製。另外， 複製一些物件是沒有意義的， 我們需要原本的變數。考慮一個Timer類， 它記錄了自從建構函式之後的時間。 要獲得一個可靠的時間， 我們需要一個指向到原本Timer物件的引用， 而不是一份複製。Timers可能甚至不允許複製它們自己， 因為複製出來的Timer物件能夠幹什麼哪？複製的Timer應該從0：00開始嗎？ 還是它應該從原本的Timer經過一段時間後開始計時哪？這兩者可能存在爭議。

回到網路連線上， 開啟一個連線是很慢的， 正如我們討論過的， 因此我們嘗試用lambda寫一個非同步的open_connection()方法：

void open_connection(Command cmd) {
tcp::socket sock(_engine);

// pass a lambda to async_connect
async_connect(sock,[cmd](error_code err) {
if (!err) {
authenticate(sock, cmd);
}
});

return;
}

這裡， 我們呼叫async_connect()函式， 它的第一個引數是socket， 第二個引數是lambda。 當它被呼叫的時候， 這個lambda函式首先檢查網路錯誤， 然後開始下一個任務， authentication().

async_connect()函式在網路操作完成的時候會呼叫lambda： 如果我們的伺服器相距很近這會很快完成， 或者它們相距很遠就沒有那麼快。我們無法知道什麼時候lambda會被呼叫， 同時， open_connection()會立刻返回， 這是好的。但是說如果我們想確切的直達async_connection()花費了多長時間？假設我們使用上面描述的Timer類， 我們不能複製它， 因此我們會通過引用獲得timer：

void open_connection(Command cmd) {
tcp::socket sock(_engine)
Timer timer; // starts timing now

async_connect(sock,[&timer, cmd](error_code err) {
std::cout << timer.secs() << “ seconds” << std::endl;

if (!err) {
authenticate(sock, cmd);
}
});

return;
}

這是有問題的， 事實上， 我們手頭上有個大麻煩。lambda函式給async_connect()傳遞一個Timer的引用，我們不知道什麼時候lambda函式會被執行， 但是我們明確的知道它不會馬上執行。但是open_connection()會立刻返回， 並且當它返回的時候它的stack會消失， 我們的Timer物件建立在stack上面！

如有一個引用變數指向一個清理掉的變數， lambda會如何哪？ 當然是發生段錯誤。

我們需要保證每一個非同步任務都打包了必要的狀態。對於不能幹淨的複製和打包的東西， 我們要確保引用狀態比任務有更長的生命週期。

我們有兩個方式來保持這樣的存活狀態。

方法A： 在一個持久化結構裡面儲存狀態

這種方法很好因為我們能夠控制Timer物件以及它們的生命週期， 它們從來不會被悄悄清理掉因為是我們負責清理它們。

缺點也是清楚的：我們需要負責清理Timer物件。 這需要我們付出我們不想要的額外的負擔， 並且我們還必須去做正確。

方法 B: 使用C++的shared_ptr來保證狀態存活

我們另外一個可選方案是使用C++ shared_ptr. shared_ptr看起來並且使用起來很像普通的指標， 除了它儲存了一個引用計數來記錄正在使用該指標的使用者個數。 shared_ptr指標指向的物件存活會一直儲存， 直到所有的使用者都釋放該指標。

使用shared_ptr也有它的優缺點， 一個最重要的有點事它的實現非常的簡單：不需要維護我們的Timer物件集合。

但是， 因為我們將Timer物件的控制權讓給了C++， 我們不能夠假定它們的生命週期。它們不是被我們清理的， 我們無法確認某個時間點它們是否還在。過渡使用shared_ptr會導致令人討厭的並且難以檢測的bug。可以從這篇博文獲得有趣的， 推薦的閱讀材料。在這條道路上， 我們必須小心行事。

兩種方法的故事

對於MongoDB的網路層， 沒有一個適用於所有情況的方法，在一些狀況下， 使用持久化結構更合理， 對於其它的狀況， shared_ptr是更加簡潔， 安全的方法。我們採用混合的方式使用兩種方法。

技術挑戰 #2： 消失的狀態 (又一次！)

這很好， 因為網路錯誤發生在我呼叫的primary執行路徑上， 這條路徑在上面用藍色的點線顯示。primary執行路徑是在任務的每個階段的lambda函式體裡。 在這裡我們能接受網路錯誤並且決定是否將下一個任務新增到呼叫鏈裡面。

我們第一種刪除方法是從mongos執行緒強制性的刪除操作， mongos執行緒會清理一些狀態， 把操作標記為已刪除， 就是這樣！

這是不好的， 正如你猜測的一樣。比如， 記憶體裡的狀態有可能已經被另一個操作重用了。

一條路徑適用所有情況

為了避免這些意外， 我們強制執行一條規則：只有primary執行路徑才能夠結束一個任務， 因為只有primary路徑才有操作的上下文。

我們首先通過在操作裡儲存一個簡單的cancelled 標記來實現它。

// Basic “network operation” class
class NetworkOp {
bool cancelled;
}
// Secondary path
cancel(NetworkOp *op) {
op->cancelled = true;
}
// Primary path
if (op->cancelled) {
done(op);
}

看下面的程式碼很明顯的我們在引發一個段錯誤：

// Secondary path
cancel(NetworkOp *op) {
// op could be a null pointer!
op->cancelled = true;
}

保持鎖定，保持安全

我們需要保護我們的共享狀態， 它可以通過mutex來實現。但是mutex應該放在哪裡？我們不能將它放在NetworkOp類裡面。很像上面的Timer物件， mutex必須儲存在獨立於操作的持久化結構裡面， 但是我們手上有有很多的問題：誰負責清理持久化物件？

考慮到這個問題的特性， shared_ptr是更好的解決方法， 我們設計了一個結構體， 叫做SafeOp， 它儲存了一個mutex和NetworkOp*：

// “network operation” class
class NetworkOp {
bool cancelled;
}

// "access control" object
class SafeOp {
mutex lock;
NetworkOp* op;
}

不是處理NetworkOp的裸指標， 兩種路徑都擁有指向SafeOp物件的指標， 雙方都達成一致：除非第一個鎖定SafeOp的mutex， 它們不會訪問或者修改NetworkOp。

// Primary path
done(shared_ptr safe) {
// lock before
// cleanup 
safe->lock.lock();

safe->op->done();
safe->op = NULL;

safe.unlock();

}
// Secondary path
cancel(shared_ptr safe) {
// once we lock, can't
// change under us
safe->lock.lock();

if (safe->op) {
safe->op->cancelled =
true;
}

safe->lock.unlock();
}

實現， 測試， 重複

它花費了好幾個小時的汗水、眼淚和頭疼， 並且這通常在軟體開發中常見的情況， 我們第一個嘗試通常不是最佳的實現， 我們不得不迭代， 迭代， 段錯誤， 重複迭代，並且， 測試，測試， 測試！當開發一些新的且複雜的專案， 就像這個專案， 我們需要失敗儘快返回，其通常可能產生最好的產品。 我們寫了很多層級的測試用例（單元俄式， 繼承測試， 壓力測試等等）來向上、向下和側向測試網路層。

因此我們如何詳細地測試我們新的callback驅動的， 非同步網路系統？那是其他時間點的一個話題。

原文連結：

https://engineering.mongodb.com/post/building-an-async-networking-layer-for-mongos