分布式解决方案 Percolator--详解

Percolator简介

Google在2012年将Percolator的架构设计作为论文发表，其目的是构建于BigTalbe的基础上，主要用于网页搜索索引等服务。由于BigTable只支持单行级别的事务，不支持多行事务等更复杂的事务，因此Percolator的诞生就是为了解决这个问题。Percolator支持ACID语义，通过多版本时间戳排序实现了Snapshot Isolation隔离级别，所以可以将其看作是一种通用的分布式事务解决方案。基于google的Bigtable来实现，其本质上是一个二阶段提交协议，利用了Bigtable的行事务。

架构

1 Percolator 包含三个组件：
Client：Client 是整个协议的控制中心，是两阶段提交的协调者（Coordinator）；
TSO：一个全局的授时服务，提供全局唯一且递增的时间戳（timetamp）；
Bigtable：实际持久化数据的分布式存储；

2.1. Client
二阶段提交算法中有两种角色，协调者和参入者。在Percolator中，Client充当协调者的角色，负责发起和提交事务。

2.2. Timestamp Oracle (TSO)
Percolator依赖于TSO提供一个全局唯一且递增的时间戳，来实现Snapshot Isolation。在事务的开始和提交的时候，Client都需要从TSO拿到一个时间戳。

2.3 Bigtable
Bigtable从数据模型上可以理解为一个multi-demensional有序Map，键值对形式如下

(row:string, column:string,timestamp:int64)->string

key由三元组 (row, column, timestamp) 组成，value可以是认为byte数组。

在Bigtable中，一行 (row) 可以包含多个 (column)，Bigtable提供了单行的跨多列的事务能力，Percolator利用这个特性来保证对同一个row的多个column的操作是原子性的。Percolator的元数据存储在特殊的column中，如下：

列名	作用
lock	锁信息
write	事务提交时间戳
data	数据

Percolator事务处理(文尾附图)

第一步：为每个事务分配一个开始时间戳。与MVCC类似，开始时间戳决定了该事务所能看到的数据集。

Transaction() : start ts (oracle.GetTimestamp()) {}

第二步：将事务中所有的写操作（insert/update/delete）缓冲起来，直到提交时再一并写入。

void Set(Write w) { writes .push back(w); }

第三步：prewrite (预写阶段)，该阶段为两阶段提交的第一阶段。此阶段从所有的写操作中选出一个作为主（Primary）锁，其它的操作作为次（Secondary）锁。预写阶段需要用主锁锁住事务中写操作涉及的所有数据，具体流程如下：
1）启动一个BigTable单行事务；
2）读取写操作涉及行的write元数据信息，检查是否有其他事务在该事务开始后【即时间范围为[start_ts, +++], +++为正无穷】提交并修改了改行的数据，如果有则终止本事务；否则，执行下一步；
3）读取写操作涉及行的lock元数据信息，检查是否有其他事务持有该行的锁，如果有，则代表存在写冲突，Percolator不等待锁释放，而是直接终止事务（无等待的死锁预防策略）。
4）顺利通过冲突检查后，事务开始更新数据，以事务开始时间戳作为BigTable的时间戳，将数据写入data列中。由于采用多版本并发控制因此不会覆盖原来的数据，而是新建一行写入相应的数据。
5) 更新完数据后，获取对应行事务锁，同样以事务开始时间戳BigTable的时间戳，但以主锁的{primary.row, primary.col}作为值，写入lock列。
6）进入事务提交阶段。
伪代码：

Prewrite tries to lock cell w, returning false in case of conflict.
bool Prewrite(Write w, Write primary) {
Column c = w.col;
bigtable::Txn T = bigtable::StartRowTransaction(w.row);
Abort on writes after our start timestamp . . .
if (T.Read(w.row, c+“write”, [start ts , +++]))
return false;
// . . . or locks at any timestamp.
if (T.Read(w.row, c+“lock”, [0, ∞])) return false;
T.Write(w.row, c+“data”, start ts , w.value);
T.Write(w.row, c+“lock”, start ts ,
{primary.row, primary.col}); // The primary’s location.
return T.Commit();
}

第四步：提交事务
1）获取提交时间戳commit_ts；
2）对主锁涉及的行启动一个单行事务，接着检查事务是否还持有lock列的锁，如果检查失败则终止事务；
3）如果事务持有锁，则以提交时间戳commit_ts作为BigTble的时间戳，以开始时间戳作为write列值更新数据，使数据对其他事务可见；
4）释放事务持有的主锁；
5）主锁的写操作提交后Percolator认为整个事务已完成，进入原子提交的第二阶段。第二阶段更新所有的次锁写操作，写完后释放次锁，这一步可以异步执行。
伪代码：

bool Commit() {
Write primary = writes [0];
vector secondaries(writes .begin()+1, writes .end());
if (!Prewrite(primary, primary)) return false;
for (Write w : secondaries)
if (!Prewrite(w, primary)) return false;

int commit ts = oracle .GetTimestamp();

// Commit primary first.
Write p = primary;
bigtable::Txn T = bigtable::StartRowTransaction(p.row);
if (!T.Read(p.row, p.col+“lock”, [start ts , start ts ]))
return false; // aborted while working
T.Write(p.row, p.col+“write”, commit ts,
start ts ); // Pointer to data written at start ts .
T.Erase(p.row, p.col+“lock”, commit ts);
if (!T.Commit()) return false; // commit point

// Second phase: write out write records for secondary cells.
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;
}

对于读操作，第一步先检查时间戳在[0, start_ts]内是否有其他事务持有该行锁。注意，根据快照隔离性质，允许 start_ts之后的事务持有其他事务持有该行的锁，因为发生在start_ts的事务读操作只能读取到 start_ts之前的数据版本，此时间之后的数据并不关心。如果检查发现[0, start_ts]内存在冲突的锁，则读事务必须等待，直到锁被释放。
如果发现没有冲突的锁，则读取[0, start_ts]内的所有数据版本，进一步判断最新的版本，如果不存在则表明找不到可读的数据，反之最新的数据即为所读数据。
伪代码：

bool Get(Row row, Column c, string* value) {
while (true) {
bigtable::Txn T = bigtable::StartRowTransaction(row);
// Check for locks that signal concurrent writes.
if (T.Read(row, c+“lock”, [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.
latest write = T.Read(row, c+“write”, [0, start ts ]);

if (!latest write.found()) return false; // no data
int data ts = latest write.start timestamp();
*value = T.Read(row, c+“data”, [data ts, data ts]);
return true;
}

Client Crash场景

Percolator的事务协调者在Client端，而Client是可能出现crash的情况的。如果Client在提交过程中出现异常，那么事务之前写入的锁会被留下来。如果这些锁没有被及时清理，会导致后续的事务无限制阻塞在锁上。

Percolator采用 lazy 的方式来清理锁，当事务 A 遇到一个事务 B 留下来的锁时，事务 A 如果确定事务 B 已经失败了，则会将事务 B 留下来的锁给清理掉。但是事务 A 很难百分百确定判断事务 B 真的失败了，那就可能导致事务 A 正在清理事务 B 留下来的锁，而事务 B 其实还没有失败，且正在进行事务提交。

为了避免出现此异常，Percolator事务模型在每个事务写入的锁中选取一个作为Primary lock，作为清理操作和事务提交的同步点。在清理操作和事务提交时都会修改primary lock的状态，因为修改锁的操作是在Bigtable的行事务下进行的，所有清理操作和事务提交中只有一个会成功，这就避免了前面提到的并发场景下可能出现的异常。

根据primary lock的状态就可以确定事务是否已经成功commit：

1）如果Primary Lock不存在，且 write 列中已经写入了 commit_ts，那么表示事务已经成功commit；
2）如果Primary Lock还存在，那说明事务还没有进入到commit阶段，也就是事务还未成功commit。

事务 A 在提交过程中遇到事务 B 留下的锁记录时需要根据事务 B 的Primary Lock的状态来进行操作。

1 如果事务 B 的Primary Lock不存在，且 write 列中有 commit_ts 了，那么事务

A 需要将事务 B 的锁记录 roll-forward。roll-forward操作是rollback操作的反向操作，也就是将锁记录清除，并在 write 列中写入 commit_ts。
如果事务 B 的Primary Lock存在，那么事务 A 可以确定事务 B 还没有成功commit，此时事务 A 可以选择将事务 B 留下锁记录清除掉，在清除掉之前，需要将事务 B 的Primary Lock先清理掉。
如果事务 B 的Primary Lock不存在，且 write 列中也没有 commit_ts 信息，那么说明事务 B 已经被 rollback 了，此时也只需要将事务 B 留下的锁清理掉即可。

虽然上面的操作逻辑不会出现不一致的情况，但是由于事务 A 可能将存活着的事务 B 的Primary Lock清理掉，导致事务 B 被rollback，这会影响到系统的整体性能。

为了解决这个问题，Percolator使用了Chubby lockservice来存储每个正在进行事务提交的Client的存活状态，这样就可以确定Client是否真的已经挂掉了。只有在Client真的挂掉了之后，冲突事务才会真的清除掉Primary Lock以及冲突锁记录。但是还可能出现Client存活，但是其实其已经Stuck住了，没有进行事务提交的动作。这时如果不清理掉其留下的锁记录，会导致其他冲突事务无法成功提交。

为了处理这种场景，每个存活状态中还存储了一个wall time，如果判断wall time太旧之后，则进行冲突锁记录的处理。长事务则需要每隔一定的时间去更新这个wall time，保证其事务不会因此被rollback掉。

最终的事务冲突逻辑如下：

如果事务 B 的Primary Lock不存在，且 write 列中有 commit_ts 了，那么事务 A 需要将事务 B 的锁记录 roll-forward。roll-forward操作是rollback操作的反向操作，也就是将锁记录清除，并在 write 列中写入 commit_ts。
如果事务 B 的Primary Lock不存在，且 write 列中也没有 commit_ts 信息，那么说明事务 B 已经被 rollback 了，此时也只需要将事务 B 留下的锁清理掉即可。
如果事务 B 的Primary Lock存在，且TTL已经过期，那么此时事务 A 可以选择将事务 B 留下锁记录清除掉，在清除掉之前，需要将事务 B 的Primary Lock先清理掉。
如果事务 B 的Primary Lock存在，且TTL还未过期，那么此时事务 A 需要等待事务 B 的commit或者rollback后继续处理。