对percolator的理解与tidb实现

论文

论文地址
https://research.google/pubs/pub36726/

概述

设计percolator的目的
为大数据集群进行增量处理更新的系统，主要用于google网页搜索索引服务。使用基于Percolator的增量处理系统代替原有的批处理索引系统后，Google在处理同样数据量的文档时，将文档的平均搜索延时降低了50%。

基于bigtable单行事务实现跨行事务
Percolator 在 Bigtable 之上实现的，以client library 的方式实现。
Percolator 利用 Bigtable 的单行事务能力，依靠client的协议和一个全局的授时服务器 TSO 以及两阶段提交协议来实现了跨机器的多行事务。

MVCC与snapshot isolation
Percolator依据全局时间戳和MVCC实现 Snapshot Isolation 隔离级别的并发控制协议。

percolator特点
- 事务: 跨行、跨表的、基于快照隔离的ACID事务 - 观察者(observers)：一种类似触发器的通知机制

设计

隔离等级

通过MVCC来实现SI(Snapshop isolation)隔离等级。

Percolator 使用Bigtable的时间戳记维度实现了数据的多版本化。优点如下： - 读操作：可以读取任何指定时间戳版本的记录 - 写操作，能很好的应对写写冲突：若两个事务操作同一记录，只有一个会提交成功

I存在write skew(写偏斜)

锁机制

Because it is built as a client library accessing Bigtable, rather than controlling access to storage itself, Percolator faces a different set of challenges implementing distributed transactions than traditional PDBMSs. Other parallel databases integrate locking into the system component that manages access to the disk: since each node already mediates access to data on the disk it can grant locks on requests and deny accesses that violate locking requirements.

Percolator锁的管理必须满足以下条件： - 能应对机器故障：若一个锁在两阶段提交时消失，系统可能将两个有冲突的事务都提交 - 高吞吐量：上千台机器会同时请求获取锁 - 低延时

锁服务要实现： - 多副本： survive failure - distributed and balanced ： handle load - 写入持久化存储系统

时间戳

Timestamp Oracle(不是Oracle数据库)： TSO通过统一中心授权可以保证按照递增的方式分配逻辑时钟,任何事件申请的时钟都不会重复，能够保证事务版本号的单调递增，确保分布式事务的时序。

所以TSO是一个分配严格的单调递增时间戳的服务器。

优化
因为每个事务都需要调用oracle两次，所以这个服务必须有很好的可伸缩性。
Oracle会定期分配一个范围的时间戳，然后将范围中的最大值写入持久化，Oracle在内存中原子递增来快速分配时间戳，查询时不涉及磁盘I/O。如果oracle重启，将以存储中的最大值作为开始值。 worker会维持一个长连接RPC到oracle，低频率的、批量的获取时间戳。

性能
Oracle中单台机器每秒向外分配接近两百万的时间戳。

关于批量获取时间戳

批量获取时间戳并不会造成乱序问题，因为就算事务A先获取时间戳T1，事务B后获取时间戳T2，T1<T2，那么分布式系统中，也无法保证事务A先执行，事务B后执行。
如果事务B先执行，那么事务A势必能发现写冲突从而rollback。

单点

为了保证单调递增的特性，所以很多TSO的开源实现都存在单点问题。如tidb的TSO。
而且，一般TSO也存在跨数据中心高延迟的问题。

其他时序方案

Logic Clock: dynamoDB
True Time : spanner
Hybrid Logic Clock : cockroachDB，没有单点问题，但是为了解决时钟误差而无法避免的时延问题。

数据存储

percolator定义了5个列

Column	Use
c:lock	An uncommitted transaction is writing this cell; contains the location of primary lock
c:write	Committed data present; stores the Bigtable timestamp of the data
c:data	Stores the data itself
c:notify Hint:	observers may need to run
c:ack	O Observer “O” has run ; stores start timestamp of successful last run

Lock

事务的锁，key value映射

1	(key,start_ts) ==> (primary_key,lock_type)

key：数据的key
start_ts：事务开始时间
primary：该锁的primary的引用。事务从待修改的keys中选择一个作为primary,其余的则作为secondary，secondary的primary_key指向primary的key，事务的上锁和解锁都由primary key决定。

Write
已提交的数据对应的时间戳。key value映射

1	(key,commit_ts) ==> (start_ts)

key：数据的key
commit_ts：事务的提交时间
start_ts：事务的开始时间（此数据在data中的时间戳版本）

Data

存储数据的列，key value映射

1	(key,start_ts) ==> (value)

key：对应的主键
start_ts：事务的开始时间
value：除主键外的数据列

Notify
notify列仅仅是一个hint值（可能是个bool值），表示是否需要触发通知。

Ack
ack列是一个简单的时间戳值，表示最近执行通知的观察者的开始时间。

案例

以银行转账为案例
Bob 向 Joe 转账7元。
事务开始时间：start timestamp =7 ，提交时间：commit timestamp=8。

Bob有10元：查询column write获取最新时间戳版本的数据(data@5),然后从column data里面获取时间戳为5的数据($10），Joe($2)

stat timestamp=7 作为当前事务的开始时间戳，将Bob选为此事务的primary key，再写入column:lock对Bob上锁，同时将column:data列更新为7:$3。

start timestamp=7作为锁定Joe账户的时间戳，更新其column:data为$9，其锁是secondary指向primary

当前时间戳commit timestamp=8作为事务提交时间戳：删除primary所在的lock，在write列中写入commit_ts：data@7

在所有secondary中写入column:write且清理column:lock，事务完成。

流程

伪代码

class Transaction {
    struct Write{ Row row; Column: col; string value;};
    vector<Write> writes_;
    int start_ts_;

    Transaction():start_ts_(oracle.GetTimestamp()) {}
    
    // 往事务中添加一行数据
    void Set(Write w) {writes_.push_back(w);}
    
    bool Get(Row row, Column c, string* value) {
        while(true) {
        
            // 开启bigtable事务
            bigtable::Txn = bigtable::StartRowTransaction(row);
            
            // Check for locks that signal concurrent writes.
            // 检查[0, start_ts]内是否有锁，有则等待再重试
            if (T.Read(row, c+"locks", [0, start_ts_])) {
                // There is a pending lock; try to clean it and wait
                BackoffAndMaybeCleanupLock(row, c);
                continue;
            }
        }

        // Find the latest write below our start_timestamp.
        // 读取column:wriet的[0, start_ts]的最新提交记录版本
        latest_write = T.Read(row, c+"write", [0, start_ts_]);
        if(!latest_write.found()) return false; // no data
        
        // 根据此时间戳版本去column:data读取数据
        int data_ts = latest_write.start_timestamp();
        *value = T.Read(row, c+"data", [data_ts, data_ts]);
        return true;
    }
    
    // prewrite : 尝试上锁+写数据，如果锁冲突则返回失败
    bool Prewrite(Write w, Write primary) {
        Column c = w.col;
        
        // bigtable 事务
        bigtable::Txn T = bigtable::StartRowTransaction(w.row);

        // 检查写写冲突：从column:write获取最新数据，如果commit_ts>=事务开始时间，则冲突
        if (T.Read(w.row, c+"write", [start_ts_, max])) return false;
        
        // 检查column:lock，如果存在已存在，则锁冲突
        if (T.Read(w.row, c+"lock", [0, max])) return false;
        
        // 往column:lock写入锁，若为secondary，则指向primary
        T.Write(w.row, c+"lock", start_ts_, {primary.row, primary.col})
        
        // 往column:data写入(key, start_ts = value)
        T.Write(w.row, c+"data", start_ts_, w.value);
        
        // 提交bigtable事务
        return T.Commit();
    }
    
    // 此commit整合 2PC的prewrite和commit，
    // 对外caller而言不需要关注是2PC提交，直接commit就好
    bool Commit() {
        // ***** Prewrite *****//
        // 2PC 第一阶段：prewrite
        // 第一行为primary
        Write primary = write_[0];
        // 其余为secondary
        vector<Write> secondaries(write_.begin() + 1, write_.end());
        
        if (!Prewrite(primary, primary)) return false;
        for (Write w : secondaries)
            if (!Prewrite(w, primary)) return false;
        
        // ***** Commit *****//
        // 2PC第二阶段：commit
        
        // 从oracle获取commit timestamp
        int commit_ts = oracle.GetTimestamp();

        // Commit primary first. 
        Write p = primary;
        
        // primary : 先开始bigtable的事务
        bigtable::Txn T = bigtable::StartRowTransaction(p.row);
        
        // primary : 如果primiary的锁不存在，则abort
        if (!T.Read(p.row, p.col+"lock", [start_ts_, start_ts_]))
            return false; // aborted while working
        
        // primary : 在column:write 列写入开始时间（Pointer to data written at start_ts_）
        T.Write(p.row, p.col+"write", commit_ts, start_ts_); 
        // primary : 移除column:lock 列
        T.Erase(p.row, p.col+"lock", commit_ts);
        
        // primary : 提交bigtable 事务
        if(!T.Commit()) return false;

        // Second phase: write our write records for secondary cells.
        // 遍历所有secondary，写入column:write数据同时删除column:lock
        for (Write w:secondaries) {
            bigtable::write(w.row, w.col+"write", commit_ts, start_ts_);
            bigtable::Erase(w.row, w.col+"lock", commit_ts);
        }
        return true;
    }
};

事务
- 第一阶段 - 获取时间戳T1, - 写入column:data和锁：时间戳都为T1 - 第二阶段 - 获取时间戳T2 - 写入column:write：key:commit_ts:start_ts (key:T2:T1) - 删除锁

读取
- 获取时间戳Tx - 从column:write 读取key[0, Tx]的最大时间戳数据(获取到事务写入的commit_ts=T2) - 从T2中提取出start_ts为T1 - 从column:data中读取 key:T1的数据

所以读取到的数据是commit时间戳的数据。

清理锁

若客户端在Commit一个事务时，出现了异常，Prepare时产生的锁会被留下。为避免将新事务挂住，Percolator必须清理这些锁。

Percolator用lazy方式来处理未处理的锁：当事务在执行时，发现其他事务造成的锁未处理掉，事务将决定其他事务是否失败，以及清理其他事务的那些锁。

当客户端在执行两阶段提交的commit阶段crash时，事务会留下一个提交点commit point(至少已经写入一条write记录)，但可能会留下一些lock未被处理掉 - 如果priarmy lock 已被write所替代：意味着该事务已被提交，事务需要roll forword，也就是对所有涉及到的、未完成提交的数据，用write记录替代标准的锁standed lock。 - 如果primary lock存在：事务将roll back(因为总是最先提交primary，所以primary未被提交时，可以安全地执行回滚)

这些都是基于bigtable的事务中的。

清理操作在primary锁上是同步的，所以清理alive客户端持有的锁是安全的；然而回滚会强迫事务取消，这会严重影响性能。所以，一个事务将不会清理一个锁除非它猜测这个锁属于一个僵死的worker。
Percolator使用简单的机制来确定另一个事务的活跃度。运行中的worker会写一个token到Chubby锁服务来指示他们属于本系统，token会被其他worker视为一个代表活跃度的信号（退出时token会被自动删除）。有些worker是活跃的，但不在运行中，为了处理这种情况，我们附加的写入一个wall time到锁中；一个锁的wall time如果太老，即使token有效也会被清理。有些操作运行很长时间才会提交，针对这种情况，在整个提交过程中worker会周期的更新wall time。

通知

用户对感兴趣的列编写观察者function注册到percolator，当列发生改变时，percolator通知percolator的worker运行用户function。

通知与写操作不是原子的
通知类似于数据库中的触发器，然而不同的是，通知在其他事务(worker)中执行，所以写操作与观察者执行不是原子的，且观察者的执行会有时效性问题。

这和传统关系型数据库的ACID的C有一定的差别，我猜: 所以这里不叫trigger而是通知的原因

通知与观察者无限循环
编写观察者时，用户要自己考虑通知与观察者进入无限循环的情况(通知->观察者->通知->观察者.....)。

通知的"丢失"
一个列的多次更改只会触发一次通知，所以通知和操作系统的中断一样会存在“丢失”的问题。

实现通知机制

为了实现通知机制，Percolator需要高效找到被观察的脏cell。 Percolator在Bigtable维护一个“notify”列(notify列为一个独立的Bigtable locality group)，表示此cell是否为脏。当事务修改被观察的cell时，则设置cell的notify。worker对notify列执行一个分布式扫描来找到脏cell。找到notify则触发观察者并且等到观察者事务提交成功后，会删除对应的notify cell。

tidb的实现

percolator定义了5个列：data, write, lock, ack, notify。
tidb定义了其中3个：data, write, lock。所以tidb没有实现notify功能(tidb不需要增量处理能力)。

tidb定义了3个rocksdb的column family： - CF_DEFAULT：对应percolator的data列 - CF_LOCK：对应percolator的lock列 - CF_WRITE：对应percolator的write列

CF_DEFAULT

1	(key, start_ts) ==> value

CF_LOCK

1	key ==> lock_info

同一时刻一个key最多只有一个锁，所以，tidb的锁没有start_ts。

CF_WRITE

1	(key, commit_ts) ==> write_info

tidb

1 client 向 tidb 发起开启事务 begin
2 tidb 向 pd 获取 tso 作为当前事务的 start_ts
3 client 向 tidb 执行以下请求：
- 读操作，从 tikv 读取版本 start_ts 对应具体数据.
- 写操作，写入 memory 中。
4 client 向 tidb 发起 commit 提交事务请求
5 tidb 开始两阶段提交。
6 tidb 按照 region 对需要写的数据进行分组。
7 tidb 开始 prewrite 操作：向所有涉及改动的 region 并发执行 prewrite 请求。若其中某个prewrite 失败，根据错误类型决定处理方式：
- KeyIsLock：尝试 Resolve Lock 后，若成功，则重试当前 region 的 prewrite[步骤7]。否则，重新获取 tso 作为 start_ts 启动 2pc 提交（步骤5）。
- WriteConfict 有其它事务在写当前 key, abort事务
- 其它错误，向 client 返回失败。
8 commit : tidb 向 pd 获取 tso 作为当前事务的 commit_ts。
9 tidb 开始 commit:tidb 向 primary 所在 region 发起 commit。若 commit primary 失败，则先执行 rollback keys,然后根据错误判断是否重试:
- LockNotExist abort事务
- 其它错误，向 client 返回失败。
10 tidb 向 tikv 异步并发向剩余 region 发起 commit。
11 tidb 向 client 返回事务提交成功信息。

所有涉及重新获取 tso 重启事务的两阶段提交的地方，会先检查当前事务是否可以满足重试条件：只有单条语句组成的事务才可以重新获取tso作为start_ts。

tikv

Prewrite

伪代码 - -> 代表rpc调用, 例如tidb->tikv.Prewrite tidb调用tikv的Prewrite接口 - . 代表进程内调用, 例如memory.Put往内存模型写数据

start_ts = tidb->pd.GetTso() // get start_ts

// tidb调用tikv prewrite接口 
tidb->tikv.Prewrite(start_ts, data_list)
{ // tikv prewrite实现
    keyIsLockedArray = []
    
    // prewrite each key with start_ts in memory，中间出现失败，则整个prewrite失败
    for key in data_list 
    {
        // check write conflict: 通过raft获取key的数据
       record = raft.Get(WriteColumn, key, start_ts)
       if record.commit_ts >= start_ts
       {
           return error(write conflict, END)
       }
       
       // check lock
       lock = raft.Get(LockColumn, key)
       if lock != null && lock.ts != start_ts // lock已存在且为其他tx的锁
       {
           keyIsLockedArray.append(key)
           continue
       }
       // 往内存中的 lock 列写入 lock(start_ts,key) 为当前key加锁,
       //   若当前key被选为 primary, 则标记为 primary,
       //   若为secondary,则标明指向primary的信息。
       memory.Put(LockColumn, key, start_ts, (primary|secondary), ttl, short_value)
       memory.Put(DataColumn, key, start_ts, long_value)
    }
    if len(keyIsLockedArray) > 0
    {
        return error(keyIsLockedArray)
    }
    // 将此事务在内存模型中写入的数据 持久化到raft中
    raft.Commit(memory_data)
    return ok
}

Commit

伪代码

// tidb调用rikv的Commit接口，进行2PC的Commit阶段
tidb->tikv.Commit(keys, start_ts, commit_ts)
{
    for key in keys // do commit
    {
        lock = raft.Get(LockColumn, key)
        // lock存在且匹配，则提交
        if lock != null && lock.ts == start_ts 
        {
            memory.Put(WriteColumn, key, commit_ts, start_ts)
            memory.Del(LockColumn, key, start_ts)    
       }
       // lock does not exist or tx dismatch
       else if lock == null || lock.ts != start_ts 
       {
           record = raft.Get(WriteColumn, key, start_ts, commit_ts)
           if record != null && record.write_type == (PUT|DELETE|Lock)
           {
               continue; // already commited
           } else if record == null || record.write_type == RollBack
           {
               return error(tx conflict, lock not exist)
           }
       }       
    }
    // commit to raft
    raft.Save(memory.data)
    return ok
}

Rollback

当事务在两阶段提交过程中失败时， tidb 会向当前事务涉及到的所有 tikv 发起回滚操作。

伪代码

// tidb调用tikv的Rollback接口
tidb->tikv.Rollback(keys)
{   // Rollback接口实现

    // 检查合法性
    for key in keys
    {
        // 检查当前key的锁
        lock=memory.GetLockColumn(start_ts, key)
        if lock != null and lock.ts = start_ts
        {   // 如果锁还存在且是之前的锁，则删除锁，写入的数据
            // 且在WriteColumn写入rollback记录防止后面commit请求的到来
            memory.Del(DataColumn, key, start_ts)
            memory.Put(WriteColumn, key, start_ts, rollback)
            meomry.Del(LockColumn, key, start_ts)
            continue
        }
       
        // 检查提交情况
        record = raft.Get(WriteColumn, key, start_ts)
        if record != null  
        {
           if record.status == (PUT|DELETE)
           {
               return error(transaction is already commited.)
           } else if record.status == RollBack
           {
               continue;  // already rollbacked
           }
        } else { // record is null
           // 提交纪录不存在，说明当前 key 尚未被 prewrite 过，
           // 为预防 prewrite 在rollback之后过来(可能网络原因)，
           // 在这里留下 (key,start_ts,rollback)记录
           memory.Put(WriteColumn, key, start_ts, rollback)
           continue
        }
        
        // persist
        raft->Save(memory.data)
    }
    return ok
}

Resolve Lock

若客户端在Commit一个事务时，出现了异常，Prepare 时产生的锁会被留下。为避免将新事务hang住，Percolator必须清理这些锁。

Percolator用lazy方式处理这些锁：当事务A在执行时，发现事务B造成的锁冲突，事务A将决定事务B是否失败，以及清理事务B的那些锁。

tidb 在执行 prewrite, get 过程中，若遇到锁，在锁超时的情况下，会向 tikv 发起清锁操作。

// tidb调用tikv的ResolveLock接口
tidb->tikv.ResolveLock(start_ts, commit_ts)
{
    // 找出所有 lock.ts==start_ts 的锁并执行清锁操作
    locks = raft.Scan(LockColumn, lock.ts = start_ts)
    for (lock in locks)
    {
        // commit_ts存在，则说明已提交
        if (commit_ts != null)
        {
            // 对已上锁的key进行提交
            memory.Commit(lock.key, commit_ts)
        } else {
            memory.Rollback(lock.key)
        }       
    }
    raft->Save(memory.data)
    
    return ok
}

Get

// tidb调用tikv的Get接口
tidb->tikv.Get(key, start_ts)
{
    // check lock: 如果锁存在且在此之前加的锁，则返回锁冲突
    lock = raft->Get(LockColumn, key)
    if (lock != null && lock.ts <= start_ts)
        return error(isLocked);
    version = start_ts - 1;
    
RAFTGET:
    // get writeColumn： 获取小于start_ts的最新提交记录
    data = raft->Get(WriteColumn, key, version)
    if (data != null)
    {
        if (data.writeType == PUT)
        {
            if (data.isShortValue)
                return data.shortValue;
            // long value
            return raft->Get(DataColumn, key, start_ts)
        } else if (data.writeType == DELETE)
        {
            // 没有出现过该值，或该值最近已被删除，返回tidb空
            return ok(None)；
        } else if (data.writeType == "LOCK | ROLLBACK")
        {
            // version=commit_ts-1, 继续查找下一个最近版本
            version=commit_ts-1
            goto RAFTGET
        }
    } else {
        return ok(None)
    }
}

GC

TiDB 的事务的实现采用了MVCC机制，当新写入的数据覆盖旧的数据时，旧的数据不会被替换掉，而是与新写入的数据同时保留，并以时间戳来区分版本。
GC 的任务便是清理不再需要的旧数据。

一个 TiDB 集群中会有一个 TiDB 实例被选举为 GC leader，GC 的运行由 GC leader 来控制。

GC 会被定期触发。每次 GC 时，首先，TiDB 会计算一个称为 safe point 的时间戳，接下来 TiDB 会在保证 safe point 之后的快照全部拥有正确数据的前提下，删除更早的过期数据。

每一轮 GC 分为以下三个步骤： - Resolve Locks：该阶段会对所有 Region 扫描 safe point 之前的锁，并清理这些锁 - Delete Ranges：该阶段快速地删除由于 DROP TABLE/DROP INDEX 等操作产生的整区间的废弃数据 - Do GC：该阶段每个 TiKV 节点将会各自扫描该节点上的数据，并对每一个 key 删除其不再需要的旧版本

// tidb 向 tikv 发起 GC操作，要求清理 safe-point 版本之前的所有无意义版本

// tidb调用tikv的GC接口
tidb->tikv.Gc(savePoint)
{
    startKey=null;
    for {
        // scan (startKey, maxKey]
        keys = raft.Scan(WriteColumn, startKey, batchSize)
        if (batch == null)
            return ok;
        
        // 对每个key进行gc操作
        for (key in keys)
        {
            bool remove_older = false;
            
            // 清理write中commit_ts <= startPoint的每个item
            // item: 一个key的所有write记录
            for (item in key)
            {
                if (remove_older is true)
                {
                    memory.Del(WriteColumn, item);
                    memory.Del(DataColumn, item);
                }
                    
                // check if is ordest version to save
                if (writeType == PUT)
                {   // 保留该提交，清理所有该提交之前的数据
                    memory.Set(memoryOrder, true);
                } else if (writeType == DELETE)
                {   //  清理所有safe-point 之前的数据
                    remove_older = true;
                    memory.Del(WriteColumn, item);
                } else if (writeType == ROLLBACK | LOCK)
                {   // 清理所有小于 safe-point 的 Rollback 和 Lock
                    memory.Del(WriteColumn, item);
                }
            }
        }
        startKey = keys[keys.length() - 1];
    }
}

优化

Parallel Prewrite

tikv分批的并发进行prewrite，不会像percolator要先prewrite primary，再去prewrite secondary。
如果事务冲突，导致rollback，在tikv的rollback实现中，其会留下rollback记录，这样就会导致事务的prewrite失败，而不会产生副作用。

Short Value

对于percolator，先读取column:write 列，提取到key的start_ts，再去column:data列读取key数据本身。
这样会造成两次读取，tidb的优化是，如果数据本身很小，那么就直接存储在colulmn:write中，只需读取一次即可。

Point Read Without Timestamp

为了减少一次RPC调用和减轻TSO压力，对于单点读，并不需要获取timestamp。

因为单点读不存在跨行一致性问题(读取多行数据时，必须是同一个版本的数据)，所以直接可以读取最新的数据即可。

Calculated Commit Timestamp

如果不通过TSO获取commit_ts，则会减少一次RPC交互从而降低事务的时延。

然而，为了实现SI的RR特性(repeatable read)，所以commit_ts需要确保其他事务多次读取的值是一样的。那么commit_ts就和其他事务的读取有相关性。

下面公式可以计算出一个commit_ts

1	max{start_ts, max_read_ts_of_written_keys} < commit_ts <= now

由于不可能记录每个key的最大的读取时间，但是可以记录每个region的最大读取时间，所以公式转换为：

1	commit_ts = max{start_ts, region_1_max_read_ts, region_2_max_read_ts, ...} + 1

region_x_max_read_ts : 事务涉及到的key的region。

Single Region 1PC

对于事务只涉及到一个Region，那么其实是没有必要走2PC流程的。直接提交事务即可。