摘要：是什么在介紹的概念之前，我們需要簡單回顧一下前面源碼閱讀系列文章六中講過的和的概念以及它們和語句的關系。的任務就是實現請求，執行所有涉及到的請求，并依次返回結果。構造出了所有之后，下一步就是執行這些了。

上篇文章 中，我們介紹了數據讀寫過程中 tikv-client 需要解決的幾個具體問題，本文將繼續介紹 tikv-client 里的兩個主要的模塊——負責處理分布式計算的 copIterator 和執行二階段提交的 twoPhaseCommitter。

在介紹 copIterator 的概念之前，我們需要簡單回顧一下前面 TiDB 源碼閱讀系列文章（六）中講過的 distsql 和 coprocessor 的概念以及它們和 SQL 語句的關系。

tikv-server 通過 coprocessor 接口，支持部分 SQL 層的計算能力，大部分只涉及單表數據的常用的算子都可以下推到 tikv-server 上計算，計算下推以后，從存儲引擎讀取的數據雖然是一樣的多，但是通過網絡返回的數據會少很多，可以大幅節省序列化和網絡傳輸的開銷。

distsql 是位于 SQL 層和 coprocessor 之間的一層抽象，它把下層的 coprocessor 請求封裝起來對上層提供一個簡單的 Select 方法。執行一個單表的計算任務。最上層的 SQL 語句可能會包含 JOIN，SUBQUERY 等復雜算子，涉及很多的表，而 distsql 只涉及到單個表的數據。一個 distsql 請求會涉及到多個 region，我們要對涉及到的每一個 region 執行一次 coprocessor 請求。

所以它們的關系是這樣的，一個 SQL 語句包含多個 distsql 請求，一個 distsql 請求包含多個 coprocessor 請求。

copIterator 的任務就是實現 distsql 請求，執行所有涉及到的 coprocessor 請求，并依次返回結果。

一個 distsql 請求需要處理的數據是一個單表上的 index scan 或 table scan，在 Request 包含了轉換好的 KeyRange list。接下來，通過 region cache 提供的 LocateKey 方法，我們可以找到有哪些 region 包含了一個 key range 范圍內的數據。

找到所有 KeyRange 包含的所有的 region 以后，我們需要按照 region 的 range 把 key range list 進行切分，讓每個 coprocessor task 里的 key range list 不會超過 region 的范圍。

構造出了所有 coprocessor task 之后，下一步就是執行這些 task 了。

為了更容易理解 copIterator 的執行模式，我們先從最簡單的實現方式開始， 逐步推導到現在的設計。

copIterator 是 kv.Response 接口的實現，需要實現對應 Next 方法，在上層調用 Next ?的時候，返回一個 coprocessor response，上層通過多次調用 Next 方法，獲取多個 coprocessor response，直到所有結果獲取完。

最簡單的實現方式，是在 Next 方法里，執行一個 coprocessor task，返回這個 task 的執行結果。

這個執行方式的一個很大的問題，大量時間耗費在等待 coprocessor 請求返回結果，我們需要改進一下。

coprocessor 請求如果是由 Next 觸發的，每次調用 Next 就必須等待一個 RPC ?round trip 的延遲。我們可以改造成請求在 Next 被調用之前觸發，這樣就能在 Next 被調用的時候，更早拿到結果返回，省掉了阻塞等待的過程。

在 copIterator 創建的時候，我們啟動一個后臺 worker goroutine 來依次執行所有的 coprocessor task，并把執行結果發送到一個 response channel，這樣前臺 Next 方法只需要從這個 channel 里 ?receive 一個 coprocessor response 就可以了。如果這個 task 已經執行完成，Next 方法可以直接獲取到結果，立即返回。

當所有 coprocessor task 被?work 執行完成的時候，worker 把這個 response channel 關閉，Next 方法在 receive channel 的時候發現 channel 已經關閉，就可以返回 nil response，表示所有結果都處理完成了。

以上的執行方案還是存在一個問題，就是 coprocessor task 只有一個 worker 在執行，沒有并行，性能還是不理想。

為了增大并行度，我們可以構造多個 worker 來執行 task，把所有的 task 發送到一個 task channel，多個 worker 從這一個 channel 讀取 task，執行完成后，把結果發到 response channel，通過設置 worker 的數量控制并發度。

這樣改造以后，就可以充分的并行執行了，但是這樣帶來一個新的問題，task 是有序的，但是由于多個 worker 并行執行，返回的 response 順序是亂序的。對于不要求結果有序的 distsql 請求，這個執行模式是可行的，我們使用這個模式來執行。對于要求結果有序的 distsql 請求，就不能滿足要求了，我們需要另一種執行模式。

當 worker 執行完一個 task 之后，當前的做法是把 response 發送到一個全局的 channel 里，如果我們給每一個 task 創建一個 channel，把 response 發送到這個 task 自己的 response channel 里，Next 的時候，就可以按照 task 的順序獲取 response，保證結果的有序。

以上就是 copIterator 最終的執行模式。

理解執行模式之后，我們從源碼的角度，分析一遍完整的執行流程。

前臺的執行的第一步是 CopClient 的 Send 方法。先根據 distsql 請求里的 KeyRanges 構造 coprocessor task，用構造好的 task 創建 copIterator，然后調用 copIterator 的 open 方法，啟動多個后臺 worker goroutine，然后啟動一個 sender 用來把 task 丟進 task channel，最后 copIterator 做為 kv.Reponse 返回。

前臺執行的第二步是多次調用 kv.Response 的 Next 方法，直到獲取所有的 response。

copIterator 在 Next 里會根據結果是否有序，選擇相應的執行模式，無序的請求會從 全局 channel 里獲取結果，有序的請求會在每一個 task 的 response channel 里獲取結果。

從 task channel 獲取到一個 task 之后，worker 會執行 handleTask 來發送 RPC 請求，并處理請求的異常，當 region 分裂的時候，我們需要重新構造 新的 task，并重新發送。對于有序的 distsql 請求，分裂后的多個 task 的執行結果需要發送到舊的 task 的 response channel 里，所以一個 task 的 response channel 可能會返回多個 response，發送完成后需要 關閉 task 的 response channel。

2PC 是實現分布式事務的一種方式，保證跨越多個網絡節點的事務的原子性，不會出現事務只提交一半的問題。

在 TiDB，使用的 2PC 模型是 Google percolator 模型，簡單的理解，percolator 模型和傳統的 2PC 的區別主要在于消除了事務管理器的單點，把事務狀態信息保存在每個 key 上，大幅提高了分布式事務的線性 scale 能力，雖然仍然存在一個 timestamp oracle 的單點，但是因為邏輯非常簡單，而且可以 batch 執行，所以并不會成為系統的瓶頸。

關于 percolator 模型的細節，可以參考這篇文章的介紹 https://pingcap.com/blog-cn/percolator-and-txn/

當一個事務準備提交的時候，會創建一個 twoPhaseCommiter，用來執行分布式的事務。

構造的時候，需要做以下幾件事情

從 memBuffer 和 lockedKeys 里收集所有的 key 和 mutation

memBuffer 里的 key 是有序排列的，我們從頭遍歷 memBuffer 可以順序的收集到事務里需要修改的 key，value 長度為 0 的 entry 表示 DELETE 操作，value 長度大于 0 表示 PUT 操作，memBuffer 里的第一個 key 做為事務的 primary key。lockKeys 里保存的是不需要修改，但需要加讀鎖的 key，也會做為 mutation 的 LOCK 操作，寫到 TiKV 上。

計算事務的大小是否超過限制

在收集 mutation 的時候，會統計整個事務的大小，如果超過了最大事務限制，會返回報錯。

太大的事務可能會讓 TiKV 集群壓力過大，執行失敗并導致集群不可用，所以要對事務的大小做出硬性的限制。

計算事務的 TTL 時間

如果一個事務的 key 通過 prewrite 加鎖后，事務沒有執行完，tidb-server 就掛掉了，這時候集群內其他 tidb-server 是無法讀取這個 key 的，如果沒有 TTL，就會死鎖。設置了 TTL 之后，讀請求就可以在 TTL 超時之后執行清鎖，然后讀取到數據。

我們計算一個事務的超時時間需要考慮正常執行一個事務需要花費的時間，如果太短會出現大的事務無法正常執行完的問題，如果太長，會有異常退出導致某個 key 長時間無法訪問的問題。所以使用了這樣一個算法，TTL 和事務的大小的平方根成正比，并控制在一個最小值和一個最大值之間。

在 twoPhaseCommiter 創建好以后，下一步就是執行 execute 函數。

在 execute 函數里，需要在 defer 函數里執行 cleanupKeys，在事務沒有成功執行的時候，清理掉多余的鎖，如果不做這一步操作，殘留的鎖會讓讀請求阻塞，直到 TTL 過期才會被清理。第一步會執行 prewriteKeys，如果成功，會從 PD?獲取一個 commitTS 用來執行 commit 操作。取到了 commitTS 之后，還需要做以下驗證:

commitTS 比 startTS 大

schema 沒有過期

事務的執行時間沒有過長

如果沒有通過檢查，事務會失敗報錯。

通過檢查之后，執行最后一步 commitKeys，如果沒有錯誤，事務就提交完成了。

當 commitKeys 請求遇到了網絡超時，那么這個事務是否已經提交是不確定的，這時候不能執行 cleanupKeys 操作，否則就破壞了事務的一致性。我們對這種情況返回一個特殊的 undetermined error，讓上層來處理。上層會在遇到這種 error 的時候，把連接斷開，而不是返回給用一個執行失敗的錯誤。

prewriteKeys, ?commitKeys 和 cleanupKeys 有很多相同的邏輯，需要把 keys 根據 region 分成 batch，然后對每個 batch 執行一次 RPC。

當 RPC?返回 region 過期的錯誤時，我們需要把這個 region 上的 keys 重新分成 batch，發送 RPC 請求。

這部分邏輯我們把它抽出來，放在 doActionOnKeys 和 doActionOnBatches 里，并實現 prewriteSinlgeBatch，commitSingleBatch，cleanupSingleBatch 函數，用來執行單個 batch 的 RPC 請求。

雖大部分邏輯是相同的，但是不同的請求在執行順序上有一些不同，在 doActionOnKeys 里需要特殊的判斷和處理。

prewrite 分成的多個 batch 需要同步并行的執行。

commit 分成的多個 batch 需要先執行第一個 batch，成功后再異步并行執行其他的 batch。

cleanup 分成的多個 batch 需要異步并行執行。

doActionOnBatches 會開啟多個 goroutines 并行的執行多個 batch，如果遇到了 error，會把其他正在執行的 context cancel 掉，然后返回第一個遇到的 error。

執行 prewriteSingleBatch 的時候，有可能會遇到 region 分裂錯誤，這時候 batch 里的 key 就不再是一個 region 上的 key 了，我們會在這里遞歸的調用 prewriteKeys，重新走一遍拆分 batch 然后執行 doActionOnBatch 和 prewriteSingleBatch 的流程。這部分邏輯在 commitSingleBatch 和 cleanupSingleBatch 里也都有。

twoPhaseCommitter 包含的邏輯只是事務模型的一小部分，主要的邏輯在 tikv-server 端，超出了這篇文章的范圍，就不在這里詳細討論了。

作者：周昱行