TiKV 源碼解析系列文章（十一）Storage - 事務控

背景知識

TiKV 是一個強一致的支持事務的分布式 KV 存儲。TiKV 通過 raft 來保證多副本之間的強一致，事務這塊 TiKV 參考了 Google 的?Percolator 事務模型，并進行了一些優化。

當 TiKV 的 Service 層收到請求之后，會根據請求的類型把這些請求轉發到不同的模塊進行處理。對于從 TiDB 下推的讀請求，比如 sum，avg 操作，會轉發到 Coprocessor 模塊進行處理，對于 KV 請求會直接轉發到 Storage 進行處理。

KV 操作根據功能可以被劃分為 Raw KV 操作以及 Txn KV 操作兩大類。Raw KV 操作包括 raw put、raw get、raw delete、raw batch get、raw batch put、raw batch delete、raw scan 等普通 KV 操作。 Txn KV 操作是為了實現事務機制而設計的一系列操作，如 prewrite 和 commit 分別對應于 2PC 中的 prepare 和 commit 階段的操作。

本文將為大家介紹 TiKV 源碼中的 Storage 模塊，它位于 Service 與底層 KV 存儲引擎之間，主要負責事務的并發控制。TiKV 端事務相關的實現都在 Storage 模塊中。

源碼解析

接下來我們將從 Engine、Latches、Scheduler 和 MVCC 等幾個方面來講解 Storage 相關的源碼。

1. Engine trait

TiKV 把底層 KV 存儲引擎抽象成一個 Engine trait（trait 類似其他語言的 interface），定義見?storage/kv/mod.rs。Engint trait 主要提供了讀和寫兩個接口，分別為?async_snapshot?和?async_write。調用者把要寫的內容交給?async_write，async_write?通過回調的方式告訴調用者寫操作成功完成了或者遇到錯誤了。同樣的，async_snapshot?通過回調的方式把數據庫的快照返回給調用者，供調用者讀，或者把遇到的錯誤返回給調用者。

pub?trait?Engine:?Send?+?Clone?+?'static?{
????type?Snap:?Snapshot;
????fn?async_write(&self,?ctx:?&Contect,?batch:?Vec<Modify>,?callback:?Callback<()>)?->?Result<()>;
????fn?async_snapshot(&self,?ctx:?&Context,?callback:?Callback<Self::Snap>)?->?Result<()>;
}

只要實現了以上兩個接口，都可以作為 TiKV 的底層 KV 存儲引擎。在 3.0 版本中，TiKV 支持了三種不同的 KV 存儲引擎，包括單機 RocksDB 引擎、內存 B 樹引擎和 RaftKV 引擎，分別位于?storage/kv?文件夾下面的?rocksdb_engine.rs、btree_engine.rs?和?raftkv.rs。其中單機 RocksDB 引擎和內存紅黑樹引擎主要用于單元測試和分層 benchmark，TiKV 真正使用的是 RaftKV 引擎。當調用 RaftKV 的?async_write?進行寫入操作時，如果?async_write?通過回調方式成功返回了，說明寫入操作已經通過 raft 復制給了大多數副本，并且在 leader 節點（調用者所在 TiKV）完成寫入了，后續 leader 節點上的讀就能夠看到之前寫入的內容。

2. Raw KV 執行流程

Raw KV 系列接口是繞過事務直接操縱底層數據的接口，沒有事務控制，比較簡單，所以在介紹更復雜的事務 KV 的執行流程前，我們先介紹 Raw KV 的執行流程。

Raw put

raw put 操作不需要 Storage 模塊做額外的工作，直接把要寫的內容通過 engine 的?async_write?接口發送給底層的 KV 存儲引擎就好了。調用堆棧為?service/kv.rs: raw_put?->?storage/mod.rs: async_raw_put。

impl<E:?Engine>?Storage<E>?{
????pub?fn?async_raw_put(
????????&self,????????ctx:?Context,????????cf:?String,????????key:?Vec<u8>,????????value:?Vec<u8>,????????callback:?Callback<()>,
????)?->?Result<()>?{????????//?Omit?some?limit?checks?about?key?and?value?here...????????self.engine.async_write(
????????????&ctx,
????????????vec![Modify::Put(
????????????????Self::rawkv_cf(&cf),
????????????????Key::from_encoded(key),
????????????????value,
????????????)],
????????????Box::new(|(_,?res)|?callback(res.map_err(Error::from))),
????????)?;
????????Ok(())
????}
}

Raw get

同樣的，raw get 只需要調用 engine 的?async_snapshot?拿到數據庫快照，然后直接讀取就可以了。當然對于 RaftKV 引擎，async_snapshot?在返回數據庫快照之前會做一些檢查工作，比如會檢查當前訪問的副本是否是 leader（3.0.0 版本只支持從 leader 進行讀操作，follower read 目前仍然在開發中），另外也會檢查請求中攜帶的 region 版本信息是否足夠新。

3. Latches

在事務模式下，為了防止多個請求同時對同一個 key 進行寫操作，請求在寫這個 key 之前必須先獲取這個 key 的內存鎖。為了和事務中的鎖進行區分，我們稱這個內存鎖為 latch，對應的是?storage/txn/latch.rs?文件中的 Latch 結構體。每個 Latch 內部包含一個等待隊列，沒有拿到 latch 的請求按先后順序插入到等待隊列中，隊首的請求被認為拿到了該 latch。

#[derive(Clone)]struct?Latch?{????pub?waiting:?VecDeque<u64>,
}

Latches 是一個包含多個 Latch 的結構體，內部包含一個固定長度的 Vector，Vector 的每個 slot 對應一個 Latch。默認配置下 Latches 內部 Vector 的長度為 2048000。每個 TiKV 有且僅有一個 Latches 實例，位于?Storage.Scheduler?中。

pub?struct?Latches?{????slots:?Vec<Mutex<Latch>>,
????size:?usize,
}

Latches 的?gen_lock?接口用于計算寫入請求執行前所需要獲取的所有 latch。gen_lock?通過計算所有 key 的 hash，然后用這些 hash 對 Vector 的長度進行取模得到多個 slots，對這些 slots 經過排序去重得到該命令需要的所有 latch。這個過程中的排序是為了保證獲取 latch 的順序性防止出現死鎖情況。

impl?Latches?{
????pub?fn?gen_lock<H:?Hash>(&self,?keys:?&[H])?->?Lock?{????????//?prevent?from?deadlock,?so?we?sort?and?deduplicate?the?index.
????????let?mut?slots:?Vec<usize>?=?keys.iter().map(|x|
????????self.calc_slot(x)).collect();
????????slots.sort();
????????slots.dedup();
????????Lock::new(slots)
????}
}

4. Storage 和事務調度器 Scheduler

Storage

Storage 定義在?storage/mod.rs?文件中，下面我們介紹下 Storage 幾個重要的成員：

engine：代表的是底層的 KV 存儲引擎。

sched：事務調度器，負責并發事務請求的調度工作。

read_pool：讀取線程池，所有只讀 KV 請求，包括事務的非事務的，如 raw get、txn kv get 等最終都會在這個線程池內執行。由于只讀請求不需要獲取 latches，所以為其分配一個獨立的線程池直接執行，而不是與非只讀事務共用事務調度器。

gc_worker：從 3.0 版本開始，TiKV 支持分布式 GC，每個 TiKV 有一個?gc_worker?線程負責定期從 PD 更新 safepoint，然后進行 GC 工作。

pessimistic_txn_enabled： 另外 3.0 版本也支持悲觀事務，pessimistic_txn_enabled?為 true 表示 TiKV 以支持悲觀事務的模式啟動，關于悲觀事務后續會有一篇源碼閱讀文章專門介紹，這里我們先跳過。

pub?struct?Storage<E:?Engine>?{
????engine:?E,
????sched:?Scheduler<E>,
????read_pool:?ReadPool,
????gc_worker:?GCWorker<E>,
????pessimistic_txn_enabled:?bool,????//?Other?fields...}

對于只讀請求，包括 txn get 和 txn scan，Storage 調用 engine 的?async_snapshot?獲取數據庫快照之后交給?read_pool?線程池進行處理。寫入請求，包括 prewrite、commit、rollback 等，直接交給 Scheduler 進行處理。Scheduler 的定義在?storage/txn/scheduler.rs?中。

Scheduler

pub?struct?Scheduler<E:?Engine>?{
????engine:?Option<E>,
????inner:?Arc<SchedulerInner>,
}struct?SchedulerInner?{
????id_alloc,?AtomicU64,
????task_contexts:?Vec<Mutex<HashMap<u64,?TaskContext>>>,
????lathes:?Latches,
????sched_pending_write_threshold:?usize,
????worker_pool:?SchedPool,
????high_priority_pool:?SchedPool,????//?Some?other?fields...}

接下來簡單介紹下 Scheduler 幾個重要的成員：

id_alloc：到達 Scheduler 的請求都會被分配一個唯一的 command id。

latches：寫請求到達 Scheduler 之后會嘗試獲取所需要的 latch，如果暫時獲取不到所需要的 latch，其對應的 command id 會被插入到 latch 的 waiting list 里，當前面的請求執行結束后會喚醒 waiting list 里的請求繼續執行，這部分邏輯我們將會在下一節 prewrite 請求在 scheduler 中的執行流程中介紹。

task_contexts：用于存儲 Scheduler 中所有請求的上下文，比如暫時未能獲取所需 latch 的請求都會被暫存在?task_contexts中。

sched_pending_write_threshold：用于統計 Scheduler 內所有寫入請求的寫入流量，可以通過該指標對 Scheduler 的寫入操作進行流控。

worker_pool，high_priority_pool：兩個線程池，寫請求在調用 engine 的 async_write 之前需要進行事務約束的檢驗工作，這些工作都是在這個兩個線程池中執行的。

prewrite 請求在 Scheduler 中的執行流程

下面我們以 prewrite 請求為例子來講解下寫請求在 Scheduler 中是如何處理的：

1）Scheduler 收到 prewrite 請求的時候首先會進行流控判斷，如果 Scheduler 里的請求過多，會直接返回?SchedTooBusy?錯誤，提示等一會再發送，否則進入下一步。

2）接著會嘗試獲取所需要的 latch，如果獲取 latch 成功那么直接進入下一步。如果獲取 latch 失敗，說明有其他請求占住了 latch，這種情況說明其他請求可能也正在對相同的 key 進行操作，那么當前 prewrite 請求會被暫時掛起來，請求的上下文會暫存在 Scheduler 的?task_contexts?里面。當前面的請求執行結束之后會將該 prewrite 請求重新喚醒繼續執行。

impl<E:?Engine>?Scheduler<E>?{
????fn?try_to_wake_up(&self,?cid:?u64)?{????????if?self.inner.acquire_lock(cid)?{????????????self.get_snapshot(cid);
????????}
????}
????fn?release_lock(&self,?lock:?&Lock,?cid:?u64)?{
????????let?wakeup_list?=?self.inner.latches.release(lock,?cid);????????for?wcid?in?wakeup_list?{????????????self.try_to_wake_up(wcid);
????????}
????}
}

3）獲取 latch 成功之后會調用 Scheduler 的?get_snapshot?接口從 engine 獲取數據庫的快照。get_snapshot?內部實際上就是調用 engine 的?async_snapshot?接口。然后把 prewrite 請求以及剛剛獲取到的數據庫快照交給?worker_pool?進行處理。如果該 prewrite 請求優先級字段是?high?就會被分發到?high_priority_pool?進行處理。high_priority_pool?是為了那些高優先級請求而設計的，比如 TiDB 系統內部的一些請求要求 TiKV 快速返回，不能由于?worker_pool?繁忙而被卡住。需要注意的是，目前?high_priority_pool?與?worker_pool?僅僅是語義上不同的兩個線程池，它們內部具有相同的操作系統調度優先級。鄭州專業不孕不育醫院：http://yyk.39.net/zz3/zonghe/1d427.html

4）worker_pool?收到 prewrite 請求之后，主要工作是從拿到的數據庫快照里確認當前 prewrite 請求是否能夠執行，比如是否已經有更大 ts 的事務已經對數據進行了修改，具體的細節可以參考?Percolator 論文，或者參考我們的官方博客?《TiKV 事務模型概覽》。當判斷 prewrite 是可以執行的，會調用 engine 的?async_write?接口執行真正的寫入操作。這部分的具體的代碼見?storage/txn/process.rs?中的?process_write_impl?函數。

5）當?async_write?執行成功或失敗之后，會調用 Scheduler 的?release_lock?函數來釋放 latch 并且喚醒等待在這些 latch 上的請求繼續執行。

5. MVCC

TiKV MVCC 相關的代碼位于?storage/mvcc?文件夾下，強烈建議大家在閱讀這部分代碼之前先閱讀?Percolator 論文，或者我們的官方博客?《TiKV 事務模型概覽》。

MVCC 下面有兩個比較關鍵的結構體，分別為?MvccReader?和?MvccTxn。MvccReader?位于?storage/mvcc/reader/reader.rs?文件中，它主要提供讀功能，將多版本的處理細節隱藏在內部。比如?MvccReader?的?get?接口，傳入需要讀的 key 以及 ts，返回這個 ts 可以看到的版本或者返回?key is lock?錯誤等。

impl<S:?Snapshot>?MvccReader<S>?{
????pub?fn?get(&mut?self,?key:?&Key,?mut?ts:?u64)?->?Result<Option<Value>>;
}

MvccTxn?位于?storage/mvcc/txn.rs?文件中，它主要提供寫之前的事務約束檢驗功能，上一節 prewrite 請求的處理流程中第四步就是通過調用?MvccTxn?的 prewrite 接口來進行的事務約束檢驗。焦作國醫胃腸醫院口碑怎么樣：http://jz.lieju.com/zhuankeyiyuan/37756433.htm

小結

TiKV 端事務相關的實現都位于 Storage 模塊中，該文帶大家簡單概覽了下這部分幾個關鍵的點，想了解更多細節的讀者可以自行閱讀這部分的源碼（code talks XD）。另外從 3.0 版本開始，TiDB 和 TiKV 支持悲觀事務，TiKV 端對應的代碼主要位于?storage/lock_manager?以及上面提到的 MVCC 模塊中。