事務(wù)是存儲系統(tǒng)中一種重要的機制，一個事務(wù)可以包含一個或多個操作，一個事務(wù)在邏輯上看，是一個不可分割的執(zhí)行單元，組成事務(wù)的操作，必須全部成功，事務(wù)才能提交，一旦其中存在操作失敗，整個事務(wù)中的全部操作都需要回滾。

舉個常見的例子，銀行轉(zhuǎn)賬問題，現(xiàn)在要把從A賬戶，轉(zhuǎn)賬1000到B賬戶。包含兩個操作：A -= 1000；B += 1000。我們要保證，這兩個操作同時成功或者失敗。這是事務(wù)的典型用法。

提起事務(wù)，你可能就會想到數(shù)據(jù)庫的事務(wù)，想到ACID。但這里說的是更寬泛的事務(wù)，主要保證A+I的實現(xiàn)。個人理解，C+D更像是事務(wù)的結(jié)果，而不是約束事務(wù)的前提。

下面會從兩個方面探討事務(wù)原子性的實現(xiàn)：

1. all-or-nothing： 事務(wù)中的多個操作，要么全部執(zhí)行，要么全部不執(zhí)行（當某個操作失敗時，回滾到事務(wù)執(zhí)行之前的狀態(tài)）
2. before-or-after：當有多個事務(wù)并行執(zhí)行時，多個事務(wù)的執(zhí)行結(jié)果看起來像串行執(zhí)行的（意思是說，邏輯上，事務(wù)是不可分割的，都應(yīng)該是串行執(zhí)行的，但實際上，為了性能，會允許一些并行執(zhí)行的發(fā)生）

另外，本文所探討的事務(wù)，主要是單機事務(wù)。分布式事務(wù)，可參見兩階段提交算法(2PC)，三階段提交算法（3PC）等。

All-or-Nothing

事務(wù)中的多個操作，要么全部執(zhí)行，要么全部不執(zhí)行。

要實現(xiàn)這個屬性，一般有兩種方法：1.為事務(wù)中的操作提供回滾機制，當部分操作失敗時，回滾事務(wù)中所有操作，比如數(shù)據(jù)庫中的undo log；2.在事務(wù)提交之前，不去修改持久化的數(shù)據(jù)，而是將數(shù)據(jù)存入臨時的buffer，在提交時，原子的修改所有數(shù)據(jù)，這樣，當部分操作失敗時，直接drop掉buffer里的數(shù)據(jù)即可。

下面介紹實現(xiàn)All-or-Nothing的一些方法。

Multi-Version

多版本，不是一種具體的實現(xiàn)，而是一種思路，簡單來說，就是當數(shù)據(jù)發(fā)生更新操作時，不會修改原數(shù)據(jù)，而是創(chuàng)建一個新的版本。多版本可以很容易的實現(xiàn)All-or-Nothing，當事務(wù)失敗時，只需要回滾到舊版本的數(shù)據(jù)即可，基本不需要額外的操作。

多版本也是Before-or-After實現(xiàn)的一種思路，在后文會詳細討論。

Write-Ahead-Log and Checkpoints

預(yù)寫式日志是比較常見的手段，在執(zhí)行操作之前，先將操作內(nèi)容寫到持久化存儲的日志中。根據(jù)WAL日志內(nèi)容和用途的不同，又可以劃分為幾種不同的實現(xiàn)。

要注意，無論是什么實現(xiàn)，WAL的寫入一般是direct寫到硬盤。

Journal

atomicity_fig1.png

Journal的思想是這樣的，在存儲引擎的數(shù)據(jù)層Data之外，額外增加一個Journal層（一般為單獨的、性能更好的數(shù)據(jù)盤，比如ssd），在執(zhí)行事務(wù)時，不會直接修改Data層，而是將事務(wù)的每個操作日志同步寫入Journal層，寫入完成，即表示事務(wù)已提交。另外存在一個后臺進程，定時將Journal層的日志，異步apply到Disk中，用Checkpoint記錄當前已經(jīng)apply的日志位置，Checkpoint之前的Journal日志，就可以刪除了。Ceph中，F(xiàn)ilestore的原子操作就是使用了Journal這種方式。

現(xiàn)在來看下，這種實現(xiàn)如何保證all-or-nothing。在執(zhí)行事務(wù)時，如果在寫入Journal的過程中，發(fā)生錯誤，那么此時，事務(wù)尚未提交，直接drop掉當前事務(wù)已經(jīng)寫入Journal的日志項，即可達到nothing的狀態(tài)。而Journal到Data層的apply，一般情況是不會失敗的，因為Data層往往為具體的硬盤或者文件，apply日志，即是修改或刪除硬盤某個block處的數(shù)據(jù)，如果發(fā)生錯誤，表示硬盤出現(xiàn)故障，可以直接讓存儲引擎crash掉。等待錯誤排除后，重啟存儲引擎，從checkpoint開始，重新apply數(shù)據(jù)即可。

Undo Log

Undo log的思想在于，事務(wù)操作執(zhí)行時，會直接修改存儲引擎的數(shù)據(jù)層Data，但在修改之前，會把修改位置原有的數(shù)據(jù)記錄在日志中，這樣，如果操作執(zhí)行失敗，那么則可以通過Undo log回滾事務(wù)已經(jīng)執(zhí)行的操作，實現(xiàn)nothing狀態(tài)。

Undo log的內(nèi)容如下，其中操作類型一般為put、delete，原數(shù)據(jù)存儲修改位置原來的數(shù)據(jù)，新數(shù)據(jù)存儲修改之后的數(shù)據(jù)。為什么要包含新數(shù)據(jù)呢，Undo log并不僅僅用于回滾，其本身一般也會承擔Redo log的功能。

----------------------
|操作類型|原數(shù)據(jù)|新數(shù)據(jù)|
----------------------

在Undo log模式下，事務(wù)操作會直接修改Data層，并且因為WAL一般為direct寫，事務(wù)在操作Data層時，往往為了提升性能，僅寫到磁盤/操作系統(tǒng)的緩存中。那么當存儲引擎在執(zhí)行事務(wù)時意外崩潰時，磁盤/操作系統(tǒng)緩存中的內(nèi)容會全部丟失，此時，Data層的狀態(tài)是未知的，只能根據(jù)日志來重放。因為我們不清楚緩存中到底包含了多少沒有持久化到磁盤的操作，所以只能從日志開頭全部重放，其效率無疑是低下的。因此，引入Checkpoint的概念，每當存儲引擎將緩存中內(nèi)容刷入硬盤時，記錄一個checkpoint在日志中，保證checkpoint之前的日志都已經(jīng)apply到硬盤。這樣，在崩潰后的恢復(fù)過程中，只需要從日志的最后一個checkpoint開始重放，大大提升了恢復(fù)的效率。

Before-or-After

當有多個事務(wù)并行執(zhí)行時，邏輯上看，多個事務(wù)的執(zhí)行結(jié)果需要是串行的。

下面介紹實現(xiàn)Before-or-After的一些方法。

Multi-Version

想必你一定聽說過多版本并發(fā)控制（MVCC），不錯，這里的多版本就是指MVCC。MVCC不是一種具體的實現(xiàn)，而是一種思路：通過保存數(shù)據(jù)的多個版本，以減少或消除不同事務(wù)間讀寫操作間的互斥性，提高并發(fā)性能。MVCC具體的實現(xiàn)方式有很多，有些是樂觀并發(fā)控制，有些是悲觀的，下面給出幾種實現(xiàn)。

為了方便下面的討論，定義事務(wù)的狀態(tài)分為：

1. pending：事務(wù)創(chuàng)建時的狀態(tài)，表示等待執(zhí)行或正在執(zhí)行
2. commited：事務(wù)執(zhí)行成功且已提交
3. aborted：事務(wù)執(zhí)行失敗且已回滾

定義存儲類型：下面討論的存儲為一個key-value存儲，通過key，可以定位到該key對應(yīng)的所有value版本，形如key -> value list，value list中存儲了該key的所有版本。

定義事務(wù)：每個事務(wù)都會有一個transaction_id，是一個全局唯一且遞增的整數(shù)。存在一個全局事務(wù)map，可通過事務(wù)id索引到具體的事務(wù)信息，包括事務(wù)的狀態(tài)。

Mark-Point

前面提到，多版本的思想是使讀操作和寫操作讀取不同的數(shù)據(jù)，以避免互斥，來提高性能。但是有一個問題，如果多個事務(wù)并發(fā)執(zhí)行時，事務(wù)id較小的事務(wù)要修改某個key，但是事務(wù)id較大的事務(wù)，卻先于它讀取了這個key的數(shù)據(jù)，這就會造成數(shù)據(jù)不一致。

atomicity_fig2.png

標記點可以解決上述問題，標記點要求當前事務(wù)在執(zhí)行之前，必須等待其前一個事務(wù)將所有要修改的key都做好標記，然后才能開始執(zhí)行。當標記完成后，當前事務(wù)的讀取操作，遇到被標記的key時，就可以得知該key被前一個事務(wù)做了修改，需要等待前事務(wù)修改完成，然后才能讀取。從而解決了上一段的問題。

下面先介紹標記點方法所需的各種操作函數(shù)，然后使用標記點解決一個經(jīng)典的問題，銀行轉(zhuǎn)賬問題，即要從A的銀行賬戶，轉(zhuǎn)賬1000元到B的銀行賬戶。

事務(wù)的開始操作，獲取一個新的事務(wù)id，在全局的transaction_map中新增該事務(wù)的信息，并初始化事務(wù)狀態(tài)為pending，然后等待前一個事務(wù)完成mark，或者事務(wù)完成，才會開始執(zhí)行事務(wù)。

func begin_transaction() {
    id = new_transaction()
    previous_id = id - 1
    wait until (transaction_map[previous_id].mark_state == marked)
        or (transaction_map[previous_id].state != pending)
    return id
}

func new_transaction() {
    acquire(transaction_map_lock) // make this before-or-after
    id = next_transaction_id() // get a new id
    transaction_map[id] = new(transaction)
    transaction_map[id].state = pending
    transaction_map[id].mark_state = nil
    release(transaction_map_lock)
    return id
}

當標記完成后，調(diào)用該函數(shù)修改事務(wù)的標記狀態(tài)，通知下一個事務(wù)開始執(zhí)行。

func mark_transaction(this_transaction_id) {
    transaction_map[this_transaction_id].mark_state = marked
}

讀取操作，讀取數(shù)據(jù)版本的事務(wù)id小于當前事務(wù)id的、最新的數(shù)據(jù)。如果要讀取的key正在被之前的某事務(wù)處理，則等待其完成再讀取。

func read(key, this_transaction_id) {
    starting at end of value list until begining {
        v = previous version of key
        if v.transaction_id >= this_transaction_id {
            skip v
        }
        wait until (transaction_map[v.transaction_id].state != pending)
        if transaction_map[v.transaction_id].state == commited {
            return v.value
        } else {
            skip v
        }
    }
    return error("this key doesn't exist")
}

為要修改的key創(chuàng)建一個新的版本，并將其事務(wù)id設(shè)為當前事務(wù)的id。在標記點的實現(xiàn)中，這個函數(shù)不會被并發(fā)調(diào)用，因為begin_transaction的阻塞。

func new_version(key, this_transaction_id) {
    if transaction_map[this_transaction_id].mark_state == marked {
        return error("this transaction has already been marked")
    }
    if this_transaction_id == latest_version(key).transaction_id {
        return error("do not call this repeatedly")
    }
    append new version v to value list
    v.value = nil
    v.transaction_id = this_transaction_id
}

寫入操作，將new_value寫入到當前事務(wù)版本的value中。

func write(key, new_value, this_transaction_id) {
    starting at end of value list until begining {
        v = previous version of key
        if v.transaction_id == this_transaaction_id {
            v.value = new_value
            return
        }
    }
    return error("this version doesn't exist")
}

使用上述的方法，解決銀行轉(zhuǎn)賬問題：

func transfer(account_a, account_b, amount) {
    id = begin_transaction()
    new_version(account_a, id)
    new_version(account_b, id)
    mark_transaction(id)

    a_money = read(account_a, id)
    a_money -= amount
    write(account_a, a_money, id)

    b_money = read(account_b, id)
    b_money += amount
    write(account_b, b_money, id)

    if a_money < 0 {
        abort(id)
    }else{
        commit(id)
    }
}

標記點是一種悲觀的并發(fā)控制方法，使用標記點，所有事務(wù)執(zhí)行前，必須要知道所有要修改的key，并為每個key創(chuàng)建一個新的版本。當后續(xù)沖突的事務(wù)讀取到這個key的版本，發(fā)現(xiàn)該版本對應(yīng)的事務(wù)狀態(tài)為pending，就知悉沖突的發(fā)生，需要等待前面事務(wù)commited/aborted后，才能繼續(xù)執(zhí)行。標記點方法，除了會讓所有沖突的事務(wù)串行執(zhí)行之外，事務(wù)為要修改keys創(chuàng)建新版本的過程也是完全串行的。在沖突較多時，這么做當然沒問題，但是當沖突的概率比較低時，這樣做無疑會大幅降低性能。

Read-Capture

悲觀并發(fā)控制，在事務(wù)執(zhí)行前會提前檢測沖突，當檢測到?jīng)_突時，阻塞參與沖突的部分事務(wù)，僅允許一個事務(wù)執(zhí)行，等待沖突消失后，其他事務(wù)再繼續(xù)執(zhí)行。而樂觀并發(fā)控制，假定沖突發(fā)生概率很低，不會在執(zhí)行前提前檢測沖突，多個并發(fā)的事務(wù)直接向下執(zhí)行，直到事務(wù)提交或者執(zhí)行過程中，如果發(fā)現(xiàn)沖突，則回滾部分參與沖突的事務(wù)，后續(xù)重新調(diào)度其執(zhí)行。相比于悲觀并發(fā)控制，在沖突發(fā)生概率不大的場景，其性能會有所提升。

讀捕獲是一種樂觀并發(fā)控制的方法。它不會像Mark-point方法那樣提前進行標記，而是放任事務(wù)直接執(zhí)行，在事務(wù)調(diào)用read方法時，會在被讀取的key上做一個標記。這個標記告訴后續(xù)可能對這個key做寫入操作的，但是事務(wù)id比當前事務(wù)要小的事務(wù)，它不再被允許對這個key的版本做修改，它需要中止事務(wù)的執(zhí)行，進入aborted狀態(tài)，并重新申請一個新的事務(wù)id，重新進行調(diào)度執(zhí)行。

下面同樣先介紹讀捕獲方法所需的各種操作函數(shù)，然后使用讀捕獲解決銀行轉(zhuǎn)賬問題，即要從A的銀行賬戶，轉(zhuǎn)賬1000元到B的銀行賬戶。

相比于標記點，begin_transaction函數(shù)不再阻塞，僅負責(zé)創(chuàng)建新事務(wù)。

func begin_transaction() {
    id = new_transaction()
    return id
}

func new_transaction() {
    acquire(transaction_map_lock) // make this before-or-after
    defer release(transaction_map_lock)

    id = next_transaction_id() // get a new id
    transaction_map[id] = new(transaction)
    transaction_map[id].state = pending
    return id
}

讀取操作，讀取數(shù)據(jù)版本的事務(wù)id小于當前事務(wù)id的、最新的數(shù)據(jù)。如果要讀取的key正在被之前的某事務(wù)處理，則等待其完成再讀取。讀取結(jié)束前，更新該key的latest_reader。

func read(key, this_transaction_id) {
    starting at end of value list until begining {
        v = previous version of key
        if v.transaction_id >= this_transaction_id {
            skip v
        }
        wait until (transaction_map[v.transaction_id].state != pending)
        if transaction_map[v.transaction_id].state == commited {
            key.latest_reader = max(key.latest_reader, this_transaction_id)
            return v.value
        } else {
            skip v
        }
    }
    return error("this key doesn't exist")
}

為key創(chuàng)建一個新的版本，此時，因為begin_transaction不再阻塞，所以可能有多個事務(wù)并發(fā)調(diào)用該函數(shù)，所以需要加鎖來保證before-or-after。

func new_version(key, this_transaction_id) {
    acquire(key_lock)
    defer release(key_lock)
     
    if this_transaction_id < key.latest_reader
    or this_transaction_id < latest_version(key).transaction_id {
        abort(this_transaction_id)
    } else if this_transaction_id == latest_version(key).transaction_id {
        return error("do not call this repeatedly")
    }
    append new version v to value list
    v.value = nil
    v.transaction_id = this_transaction_id
}

寫入到指定版本，沒什么好說的。

func write(key, new_value, this_transaction_id) {
    starting at end of value list until begining {
        v = previous version of key
        if v.transaction_id == this_transaaction_id {
            v.value = new_value
            return
        }
    }
    return error("this version doesn't exist")
}

使用上述的方法，解決銀行轉(zhuǎn)賬問題：

func transfer(account_a, account_b, amount) {
    id = begin_transaction()
    a_money = read(account_a, id)
    a_money -= amount
    new_version(account_a, id)
    write(account_a, a_money, id)

    b_money = read(account_b, id)
    b_money += amount
    new_version(account_b, id)
    write(account_b, b_money, id)

    if a_money < 0 {
        abort(id)
    }else{
        commit(id)
    }
}

Compare-and-Set

compare_and_set(CAS)也是一種配合多版本樂觀并發(fā)控制的方法。其思路是，前期執(zhí)行時，不做任何標記，但在讀取數(shù)據(jù)時，需要記錄當前讀取的數(shù)據(jù)的版本。在事務(wù)的提交過程，需要依賴一個原子的CAS操作，CAS操作傳入當前事務(wù)之前讀取數(shù)據(jù)時記錄的每個數(shù)據(jù)的版本，還有要設(shè)置的值，它會重新讀取一次之前讀取過的key的版本，比較傳入的版本和重新讀取的版本是否相同，如果不相同，則中斷事務(wù)。如果相同，則可以調(diào)用write操作，完成寫入操作，并提交事務(wù)。

獲取事務(wù)id并創(chuàng)建新事務(wù)，不會阻塞。

func begin_transaction() {
    id = new_transaction()
    return id
}

func new_transaction() {
    acquire(transaction_map_lock) // make this before-or-after
    defer release(transaction_map_lock)

    id = next_transaction_id() // get a new id
    transaction_map[id] = new(transaction)
    transaction_map[id].state = pending
    transaction_map[id].mark_state = nil
    return id
}

在CAS模式實現(xiàn)下的MVCC中，不存在分離的new_version和write操作，兩者都被封裝到compare_and_set函數(shù)中原子的執(zhí)行，所以沒有提交的事務(wù)，其不會在數(shù)據(jù)層留下任何標記。所以不需要像前面幾種方法一樣等待pending狀態(tài)的事務(wù)結(jié)束。

func read(key, this_transaction_id) {
    starting at end of value list until begining {
        v = previous version of key
        if v.transaction_id >= this_transaction_id {
            skip v
        }
        if transaction_map[v.transaction_id].state == commited {
            return v.value and v.version_id
        } else {
            skip v
        }
    }
    return error("this key doesn't exist")
}

基于CAS實現(xiàn)MVCC，寫操作不能單獨調(diào)用，必須封裝在CAS中。注意，CAS操作必須是原子的，這一版需要依賴底層接口的支持。CAS操作傳入當前事務(wù)之前讀取數(shù)據(jù)時記錄的每個數(shù)據(jù)的版本，還有要設(shè)置的值，首先重新讀取一個之前讀取過的key的版本，比較傳入的版本和重新讀取的版本是否相同，如果不相同，則中斷事務(wù)。如果相同，則可以調(diào)用write操作，完成寫入操作，并提交事務(wù)。

再重申一遍，整個compare_and_set函數(shù)必須是原子的。

func write(key, new_value, this_transaction_id) {
    acquire(key_lock)
    defer release(key_lock)     
    if this_transaction_id < latest_version(key).transaction_id {
        abort(this_transaction_id)
    } else if this_transaction_id == latest_version(key).transaction_id {
        return error("do not call this repeatedly")
    }
   append new version v to value list
    v.version_id = next_version_id()
    v.value = new_value
    v.transaction_id = this_transaction_id
}

// CAS must be atomicity, this need to be ensured by underlying system 
func compare_and_set(key_list, version_list, value_map, this_transaction_id) {
    for i:=0; i<len(key_list); i++ {
        _, cur_version = read(key_list[i], this_transaction_id)
        if cur_version != version_list[i] {
            abort(this_transaction_id)
        }
    }
    for k, v := range value_map {
        write(k, v, this_transaction_id)
    }
    commit(this_transaction_id)
}

使用上述的方法，解決銀行轉(zhuǎn)賬問題：

func transfer(account_a, account_b, amount) {
    id = begin_transaction()
    a_money, a_version = read(account_a, id)
    a_money -= amount

    b_money, b_version = read(account_b, id)
    b_money += amount

    if a_money < 0 {
        abort(id)
    }else{
        compare_and_set([]{account_a,account_b}, []{a_version,b_version}, map[key]value{account_a: a_money, account_b: b_money})
    }
}

Lock

加鎖，是實現(xiàn)Before-or-After最常用的手段。

Simple Locking

最簡單的方式，搞一把全局大鎖，每個事務(wù)執(zhí)行時先要獲取這把鎖，提交后釋放。這種方式，將事務(wù)的實行完全串行化，性能較低。

Two-Phase Locking

降低鎖的粒度，每個數(shù)據(jù)項一把鎖，當事務(wù)執(zhí)行時，分為兩個階段，第一個階段，為所有事務(wù)執(zhí)行鎖需要讀取和寫入的數(shù)據(jù)項進行加鎖；第二個階段，對數(shù)據(jù)進行讀取和寫入，每處理完一個數(shù)據(jù)，即可釋放其對應(yīng)的鎖。

另外，加鎖時最好按照一定的順序進行，以避免死鎖的發(fā)生。