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

背景知識(shí)

TiKV 是一個(gè)強(qiáng)一致的支持事務(wù)的分布式 KV 存儲(chǔ)。TiKV 通過(guò) raft 來(lái)保證多副本之間的強(qiáng)一致，事務(wù)這塊 TiKV 參考了 Google 的?Percolator 事務(wù)模型，并進(jìn)行了一些優(yōu)化。

當(dāng) TiKV 的 Service 層收到請(qǐng)求之后，會(huì)根據(jù)請(qǐng)求的類型把這些請(qǐng)求轉(zhuǎn)發(fā)到不同的模塊進(jìn)行處理。對(duì)于從 TiDB 下推的讀請(qǐng)求，比如 sum，avg 操作，會(huì)轉(zhuǎn)發(fā)到 Coprocessor 模塊進(jìn)行處理，對(duì)于 KV 請(qǐng)求會(huì)直接轉(zhuǎn)發(fā)到 Storage 進(jìn)行處理。

KV 操作根據(jù)功能可以被劃分為 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 操作是為了實(shí)現(xiàn)事務(wù)機(jī)制而設(shè)計(jì)的一系列操作，如 prewrite 和 commit 分別對(duì)應(yīng)于 2PC 中的 prepare 和 commit 階段的操作。

本文將為大家介紹 TiKV 源碼中的 Storage 模塊，它位于 Service 與底層 KV 存儲(chǔ)引擎之間，主要負(fù)責(zé)事務(wù)的并發(fā)控制。TiKV 端事務(wù)相關(guān)的實(shí)現(xiàn)都在 Storage 模塊中。

源碼解析

接下來(lái)我們將從 Engine、Latches、Scheduler 和 MVCC 等幾個(gè)方面來(lái)講解 Storage 相關(guān)的源碼。

1. Engine trait

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

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<()>;
}

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

2. Raw KV 執(zhí)行流程

Raw KV 系列接口是繞過(guò)事務(wù)直接操縱底層數(shù)據(jù)的接口，沒(méi)有事務(wù)控制，比較簡(jiǎn)單，所以在介紹更復(fù)雜的事務(wù) KV 的執(zhí)行流程前，我們先介紹 Raw KV 的執(zhí)行流程。

Raw put

raw put 操作不需要 Storage 模塊做額外的工作，直接把要寫的內(nèi)容通過(guò) engine 的?async_write?接口發(fā)送給底層的 KV 存儲(chǔ)引擎就好了。調(diào)用堆棧為?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 只需要調(diào)用 engine 的?async_snapshot?拿到數(shù)據(jù)庫(kù)快照，然后直接讀取就可以了。當(dāng)然對(duì)于 RaftKV 引擎，async_snapshot?在返回?cái)?shù)據(jù)庫(kù)快照之前會(huì)做一些檢查工作，比如會(huì)檢查當(dāng)前訪問(wèn)的副本是否是 leader（3.0.0 版本只支持從 leader 進(jìn)行讀操作，follower read 目前仍然在開(kāi)發(fā)中），另外也會(huì)檢查請(qǐng)求中攜帶的 region 版本信息是否足夠新。

3. Latches

在事務(wù)模式下，為了防止多個(gè)請(qǐng)求同時(shí)對(duì)同一個(gè) key 進(jìn)行寫操作，請(qǐng)求在寫這個(gè) key 之前必須先獲取這個(gè) key 的內(nèi)存鎖。為了和事務(wù)中的鎖進(jìn)行區(qū)分，我們稱這個(gè)內(nèi)存鎖為 latch，對(duì)應(yīng)的是?storage/txn/latch.rs?文件中的 Latch 結(jié)構(gòu)體。每個(gè) Latch 內(nèi)部包含一個(gè)等待隊(duì)列，沒(méi)有拿到 latch 的請(qǐng)求按先后順序插入到等待隊(duì)列中，隊(duì)首的請(qǐng)求被認(rèn)為拿到了該 latch。

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

Latches 是一個(gè)包含多個(gè) Latch 的結(jié)構(gòu)體，內(nèi)部包含一個(gè)固定長(zhǎng)度的 Vector，Vector 的每個(gè) slot 對(duì)應(yīng)一個(gè) Latch。默認(rèn)配置下 Latches 內(nèi)部 Vector 的長(zhǎng)度為 2048000。每個(gè) TiKV 有且僅有一個(gè) Latches 實(shí)例，位于?Storage.Scheduler?中。

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

Latches 的?gen_lock?接口用于計(jì)算寫入請(qǐng)求執(zhí)行前所需要獲取的所有 latch。gen_lock?通過(guò)計(jì)算所有 key 的 hash，然后用這些 hash 對(duì) Vector 的長(zhǎng)度進(jìn)行取模得到多個(gè) slots，對(duì)這些 slots 經(jīng)過(guò)排序去重得到該命令需要的所有 latch。這個(gè)過(guò)程中的排序是為了保證獲取 latch 的順序性防止出現(xiàn)死鎖情況。

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 和事務(wù)調(diào)度器 Scheduler

Storage

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

engine：代表的是底層的 KV 存儲(chǔ)引擎。

sched：事務(wù)調(diào)度器，負(fù)責(zé)并發(fā)事務(wù)請(qǐng)求的調(diào)度工作。

read_pool：讀取線程池，所有只讀 KV 請(qǐng)求，包括事務(wù)的非事務(wù)的，如 raw get、txn kv get 等最終都會(huì)在這個(gè)線程池內(nèi)執(zhí)行。由于只讀請(qǐng)求不需要獲取 latches，所以為其分配一個(gè)獨(dú)立的線程池直接執(zhí)行，而不是與非只讀事務(wù)共用事務(wù)調(diào)度器。

gc_worker：從 3.0 版本開(kāi)始，TiKV 支持分布式 GC，每個(gè) TiKV 有一個(gè)?gc_worker?線程負(fù)責(zé)定期從 PD 更新 safepoint，然后進(jìn)行 GC 工作。

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

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

對(duì)于只讀請(qǐng)求，包括 txn get 和 txn scan，Storage 調(diào)用 engine 的?async_snapshot?獲取數(shù)據(jù)庫(kù)快照之后交給?read_pool?線程池進(jìn)行處理。寫入請(qǐng)求，包括 prewrite、commit、rollback 等，直接交給 Scheduler 進(jìn)行處理。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...}

接下來(lái)簡(jiǎn)單介紹下 Scheduler 幾個(gè)重要的成員：

id_alloc：到達(dá) Scheduler 的請(qǐng)求都會(huì)被分配一個(gè)唯一的 command id。

latches：寫請(qǐng)求到達(dá) Scheduler 之后會(huì)嘗試獲取所需要的 latch，如果暫時(shí)獲取不到所需要的 latch，其對(duì)應(yīng)的 command id 會(huì)被插入到 latch 的 waiting list 里，當(dāng)前面的請(qǐng)求執(zhí)行結(jié)束后會(huì)喚醒 waiting list 里的請(qǐng)求繼續(xù)執(zhí)行，這部分邏輯我們將會(huì)在下一節(jié) prewrite 請(qǐng)求在 scheduler 中的執(zhí)行流程中介紹。

task_contexts：用于存儲(chǔ) Scheduler 中所有請(qǐng)求的上下文，比如暫時(shí)未能獲取所需 latch 的請(qǐng)求都會(huì)被暫存在?task_contexts中。

sched_pending_write_threshold：用于統(tǒng)計(jì) Scheduler 內(nèi)所有寫入請(qǐng)求的寫入流量，可以通過(guò)該指標(biāo)對(duì) Scheduler 的寫入操作進(jìn)行流控。

worker_pool，high_priority_pool：兩個(gè)線程池，寫請(qǐng)求在調(diào)用 engine 的 async_write 之前需要進(jìn)行事務(wù)約束的檢驗(yàn)工作，這些工作都是在這個(gè)兩個(gè)線程池中執(zhí)行的。

prewrite 請(qǐng)求在 Scheduler 中的執(zhí)行流程

下面我們以 prewrite 請(qǐng)求為例子來(lái)講解下寫請(qǐng)求在 Scheduler 中是如何處理的：

1）Scheduler 收到 prewrite 請(qǐng)求的時(shí)候首先會(huì)進(jìn)行流控判斷，如果 Scheduler 里的請(qǐng)求過(guò)多，會(huì)直接返回?SchedTooBusy?錯(cuò)誤，提示等一會(huì)再發(fā)送，否則進(jìn)入下一步。

2）接著會(huì)嘗試獲取所需要的 latch，如果獲取 latch 成功那么直接進(jìn)入下一步。如果獲取 latch 失敗，說(shuō)明有其他請(qǐng)求占住了 latch，這種情況說(shuō)明其他請(qǐng)求可能也正在對(duì)相同的 key 進(jìn)行操作，那么當(dāng)前 prewrite 請(qǐng)求會(huì)被暫時(shí)掛起來(lái)，請(qǐng)求的上下文會(huì)暫存在 Scheduler 的?task_contexts?里面。當(dāng)前面的請(qǐng)求執(zhí)行結(jié)束之后會(huì)將該 prewrite 請(qǐng)求重新喚醒繼續(xù)執(zhí)行。

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 成功之后會(huì)調(diào)用 Scheduler 的?get_snapshot?接口從 engine 獲取數(shù)據(jù)庫(kù)的快照。get_snapshot?內(nèi)部實(shí)際上就是調(diào)用 engine 的?async_snapshot?接口。然后把 prewrite 請(qǐng)求以及剛剛獲取到的數(shù)據(jù)庫(kù)快照交給?worker_pool?進(jìn)行處理。如果該 prewrite 請(qǐng)求優(yōu)先級(jí)字段是?high?就會(huì)被分發(fā)到?high_priority_pool?進(jìn)行處理。high_priority_pool?是為了那些高優(yōu)先級(jí)請(qǐng)求而設(shè)計(jì)的，比如 TiDB 系統(tǒng)內(nèi)部的一些請(qǐng)求要求 TiKV 快速返回，不能由于?worker_pool?繁忙而被卡住。需要注意的是，目前?high_priority_pool?與?worker_pool?僅僅是語(yǔ)義上不同的兩個(gè)線程池，它們內(nèi)部具有相同的操作系統(tǒng)調(diào)度優(yōu)先級(jí)。鄭州專業(yè)不孕不育醫(yī)院：http://yyk.39.net/zz3/zonghe/1d427.html

4）worker_pool?收到 prewrite 請(qǐng)求之后，主要工作是從拿到的數(shù)據(jù)庫(kù)快照里確認(rèn)當(dāng)前 prewrite 請(qǐng)求是否能夠執(zhí)行，比如是否已經(jīng)有更大 ts 的事務(wù)已經(jīng)對(duì)數(shù)據(jù)進(jìn)行了修改，具體的細(xì)節(jié)可以參考?Percolator 論文，或者參考我們的官方博客?《TiKV 事務(wù)模型概覽》。當(dāng)判斷 prewrite 是可以執(zhí)行的，會(huì)調(diào)用 engine 的?async_write?接口執(zhí)行真正的寫入操作。這部分的具體的代碼見(jiàn)?storage/txn/process.rs?中的?process_write_impl?函數(shù)。

5）當(dāng)?async_write?執(zhí)行成功或失敗之后，會(huì)調(diào)用 Scheduler 的?release_lock?函數(shù)來(lái)釋放 latch 并且喚醒等待在這些 latch 上的請(qǐng)求繼續(xù)執(zhí)行。

5. MVCC

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

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

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

MvccTxn?位于?storage/mvcc/txn.rs?文件中，它主要提供寫之前的事務(wù)約束檢驗(yàn)功能，上一節(jié) prewrite 請(qǐng)求的處理流程中第四步就是通過(guò)調(diào)用?MvccTxn?的 prewrite 接口來(lái)進(jìn)行的事務(wù)約束檢驗(yàn)。焦作國(guó)醫(yī)胃腸醫(yī)院口碑怎么樣：http://jz.lieju.com/zhuankeyiyuan/37756433.htm

小結(jié)

TiKV 端事務(wù)相關(guān)的實(shí)現(xiàn)都位于 Storage 模塊中，該文帶大家簡(jiǎn)單概覽了下這部分幾個(gè)關(guān)鍵的點(diǎn)，想了解更多細(xì)節(jié)的讀者可以自行閱讀這部分的源碼（code talks XD）。另外從 3.0 版本開(kāi)始，TiDB 和 TiKV 支持悲觀事務(wù)，TiKV 端對(duì)應(yīng)的代碼主要位于?storage/lock_manager?以及上面提到的 MVCC 模塊中。