#gStore-weekly | gStore源码解析（十）事务机制

1.1 简介

本章主要介绍gStore事务的MVCC实现和事务管理两部分，前者介绍了在事务模式下数据的存储方法，后者介绍了事务的处理流程。请注意，本章的代码省略了磁盘IO的相关操作。

1.2 MVCC实现

gStore的事务实现选择了MVCC(多版本并发控制)，保证了读写间的有限并发性，实现了READ_COMMITTED
(读已提交)、SNAPSHOT
(快照隔离)、SERIALIZABLE
(可串行化)三种隔离机制，三种隔离级别下分别简写为RC、SI、SR。

1.2.1 事务实现总述

gStore的底层存储是Key-Value模式的，Key的类型是由上层编号的unsigned，而Value的类型是IVEntry
类。完成Key-Value映射的Hash表是IVArray
类。下文提及的版本链就是IVEntry
层实现的。

根据MVCC的思想，gStore中的数据(IVEntry
)在事务模式下具有版本链vList
，每个数据的每个版本有一组时间戳对(begin_ts, end_ts)
，表示该版本适用的时间戳范围。

事务由Transaction
类实现，最重要的属性就是时间戳。Transaction
类中有TID
和CID
两类时间戳，TID
是事务开始时的时间戳，CID
是事务提交时的时间戳。版本时间戳对(begin_ts, end_ts)
中，begin_ts
是创建该版本的事务CID
，而end_ts
是创建下一个版本的CID-1
。而判断某版本是否对某事务可见的条件是begin_ts < TID < end_ts
。

1.2.2 事务写操作

执行写操作的时候，同一个数据只能有一个写操作执行，因此需要先申请加上写锁。SR级别下是读写互斥的，因此SR级别额外拥有一个读计数+写标记的读写锁，保证读读不互斥的前提下读写互斥。

由于事务尚未提交，不确定该修改是否会落盘，因此在该数据版本链上新增一个私人版本。该版本时间戳对为(-1, TID)
，其中-1
表示该版本为私人版本，TID
指示所属事务编号。这种特殊的时间戳对可以看作是实现写写阻塞的一种锁。

注意到只有SR级别需要加写锁或将读锁升级为写锁（由于读写互斥，因此读锁升级要求只有当前事务执行读操作），这是因为只有SR级别是读写互相阻塞的，需要额外记录下来。获取写锁的代码如下：

    auto TID = txn->GetTID();
 shared_ptr<Version> new_version = make_shared<Version>(INVALID_TS, TID); //[-1, TID]
 rwLatch.lockExclusive();
 if(txn->GetIsolationLevelType() == IsolationLevelType::READ_COMMITTED)
 {
  //check the timestamp(lock)
  if(!checkVersion(TID, true)){
   rwLatch.unlock();
   return 0;
  }
 }
 else if(txn->GetIsolationLevelType() == IsolationLevelType::SNAPSHOT)
 {
  //check the timestamp(lock)
  if(!checkVersion(TID, false)){
   rwLatch.unlock();
   return 0;
  }
 }
 else if(txn->GetIsolationLevelType() == IsolationLevelType::SERIALIZABLE)
 {
  if(!checkVersion(TID, false)){
   rwLatch.unlock();
   return 0;
  }
  if(has_read){
   if(glatch.tryupgradelatch(TID) == false){
    rwLatch.unlock();
    return 0;
   }
  }
  else{
   if(glatch.tryexclusivelatch(TID, true) == false){
    rwLatch.unlock();
    return 0;
   }
  }
 }
 ······
 vList.push_back(new_version);
 setVersionFlag();
 rwLatch.unlock();
 return 1;

检查版本信息(即上方代码中的checkVersion函数)的代码如下，请注意写锁不合法的情况有两种：1. 最新版本是其他事务的私有版本。2. 非RC隔离级别下，最新版本虽然是公共版本但起始时间戳starttime
已经大于当前版本。

 int n = vList.size();
 auto latest = vList[n-1];
 if(latest->get_begin_ts() == INVALID_TS ){
  if(latest->get_end_ts() == TID) return true; //owned lock
  else return false; // locked
 } 
 if(!IS_RC && TID < latest->get_begin_ts()) return false; //old write
 return true;

成功获取写锁后就可以根据insert
或remove
操作修改私人版本了，其中AddSet
和DelSet
分别表示该版本的增量和减量。代码如下:

    rwLatch.lockShared();
 for(auto it: AddSet)
  vList.back()->add(it);
 for(auto it: DelSet)
  vList.back()->remove(it);
 rwLatch.unlock();
 return 1;

1.2.3 事务读操作

事务读时需要根据设定的隔离等级判定可以获取的数据。

RC级别下可以读取脏数据，因此允许读取除了私人版本的所有数据(由getLatestVersion
函数实现)；

SI级别下读写可以并发执行，但只允许获取符合时间戳的数据(由getProperVersion
函数实现)；

SR级别下有三种情况：

1. 数据拥有其他事务的私人版本，那么读取失败。
1. 数据的最新版本begin_ts
  小于等于TID
  ，说明该事务读取的是最新数据。根据读写阻塞需要在读计数上+1（当然这不影响读计数为1时的读锁升级操作）。
1. 数据的最新版本begin_ts
  大于TID
  ，说明该事务读取的老数据。该情况况下不可能成功写入(因为SR要求可串行化，读老数据写新数据是不允许的)因此不需要加读锁。

    if(txn->GetIsolationLevelType() == IsolationLevelType::READ_COMMITTED)
 {
  getLatestVersion(txn->GetTID(), AddSet, DelSet); //get latest committed version or owned uncommitted version
 }
 else if (txn->GetIsolationLevelType() == IsolationLevelType::SNAPSHOT)
 {
  rwLatch.lockShared();
  getProperVersion(txn->GetTID(), AddSet, DelSet); //get version according to timestamp
  rwLatch.unlock();
 }
 else if (txn->GetIsolationLevelType() == IsolationLevelType::SERIALIZABLE)
 {
  rwLatch.lockShared();
  int ret = checkheadVersion(txn->GetTID());
  if(ret == -1){
   rwLatch.unlock();
   assert(latched == false);
   return false;
  }
  else if(ret == 1 && first_read){
   latched = glatch.trysharedlatch(txn->GetTID());
   if(!latched){
    rwLatch.unlock();
    assert(latched == false);
    return false;
   }
  }
  getProperVersion(txn->GetTID(), AddSet, DelSet);
  rwLatch.unlock();
 }
 else //not defined
 {
  assert(false);
 }
 return true;

1.3 事务管理

事务管理由Txn_manager
类完成。Txn_manager
类分配各事务的编号(TID
)，管理各事务的状态，并在需要的时候进行MVCC的垃圾回收。垃圾回收Checkpoint
执行时要求没有事务正在运行，用一个锁checkpoint_lock
实现。

下图是一个事务的完整生命周期。事务初始为WAITING
状态。调用Begin
函数后转为RUNNING
状态，表示可以执行Query
。根据执行的情况，事务可能会被Rollback
进入ABORTED
状态，此时需要重新执行该事务，且该事务执行到一半的影响不会被记录到数据库中，保证数据库的一致性；也可能成功执行后通过Commit
函数最终提交，保证自己修改的数据能够被其他时间戳合法的事务看见。

		   	Begin()            Commit()
	WAITING--------->RUNNING------------>COMMITTED
			 Rollback()             |
						|
						v
					 ABORTED

1.3.1 Begin

Begin
函数将事务编号分配给一个客户端开启事务的请求，并修改该事务的状态为RUNNING

 checkpoint_lock.lockShared();
 txn_id_t TID = this->ArrangeTID();
    ······
 shared_ptr<Transaction> txn = make_shared<Transaction>(this->db_name, Util::get_cur_time(), TID, isolationlevel);
 txn->SetCommitID(TID);
 add_transaction(TID, txn);
 txn->SetState(TransactionState::RUNNING);
 return TID;

1.3.2 Rollback

由于执行过程中的各种情况可能导致回滚，Rollback
将事务的状态修改为ABORTED
。代码中的TransactionRollback
是用来清除该事务残留的各类锁的。

 shared_ptr<Transaction> txn = get_transaction(TID);
 if (txn == nullptr) {
  cerr << "wrong transaction id!" << endl;
  return -1;
 }
 if(db != nullptr)
  db->TransactionRollback(txn);
    ······
 txn->SetState(TransactionState::ABORTED);
 txn->SetEndTime(Util::get_cur_time());
 checkpoint_lock.unlock();
 return 0;

1.3.3 Commit

事务提交时调用Commit
函数。首先获取CID
，然后调用TransactionCommit
函数将该事务执行写操作时创建的私人版本时间戳对由(-1, TID)
修改为(CID, INF)
，表示私人版本成为最新的正式版本。最后给这些脏数据打上标记，以便执行垃圾回收。

另外，Commit结束后会根据Commit的次数执行Checkpoint回收版本，确保版本链不至于过长。

 shared_ptr<Transaction> txn = get_transaction(TID);
 ······
 txn_id_t CID = this->ArrangeCommitID();
 txn->SetCommitID(CID);
 if(db != nullptr)
  db->TransactionCommit(txn);
 ······
 txn->SetState(TransactionState::COMMITTED);
 txn->SetEndTime(Util::get_cur_time());
 add_dirty_keys(txn);
 checkpoint_lock.unlock();
 committed_num++;
 int cycle = 50000;
 if(committed_num.compare_exchange_strong(cycle, 0)){
  Checkpoint();
  cerr << "checkpoint done!" << endl;
 }
 return 0;

1.3.4 Checkpoint

Checkpoint
函数选出所有脏数据，将这些数据的所有版本链的信息整合在一起并最终合并到主数据中，以此清理所有的版本链。Checkpoint
函数结束后，内存将不再拥有任何版本。

 checkpoint_lock.lockExclusive();
 vector<unsigned> sub_ids , obj_ids, obj_literal_ids, pre_ids;
 sub_ids.insert(sub_ids.begin(), DirtyKeys[0].begin(), DirtyKeys[0].end());
 pre_ids.insert(pre_ids.begin(), DirtyKeys[1].begin(), DirtyKeys[1].end());
 for(auto key: DirtyKeys[2])
 {
  if(Util::is_entity_ele(key)){
   obj_ids.push_back(key);
  }
  else{
   obj_literal_ids.push_back(key);
  }
 }
  if(db != nullptr)
  db->VersionClean(sub_ids, obj_ids, obj_literal_ids, pre_ids);
 checkpoint_lock.unlock();