P4 并发控制

Task1 锁管理器

LockManager类包含两个属性类，分别是LockRequest和LockRequestQueue，行锁请求还是表锁请求都适配LockRequest，其属性：

/** Txn_id of the txn requesting the lock */
txn_id_t txn_id_;
/** 请求锁的锁定模式 */
LockMode lock_mode_;
/** Oid of the table for a table lock; oid of the table the row belong to for a row lock */
table_oid_t oid_;  // 若是表锁则表示表的Oid；若是行锁则表示行所属表的oid
/** 为行锁时是行RID;未用于表锁 */
RID rid_;  // RID应该是唯一的
/** 锁是否被授予 */
bool granted_{false};

一个资源（行或表）可能有多个事务进行请求，所以有了LockRequestQueue来先入先出地处理来的请求，其属性：

/** 同一资源(表或行)的锁请求列表 */
std::list<std::shared_ptr<LockRequest>> request_queue_;/** 通知此rid上被阻止的事务 */
std::condition_variable cv_;  // 条件变量
/** 升级事务的TXN id(如有) */
txn_id_t upgrading_ = INVALID_TXN_ID;  // 正在此资源上尝试锁升级的事务id.
/** 配合 */
std::mutex latch_;  // 与cv_配合使用实现等待资源，即这边锁住，cv_那边等待std::vector<size_t> lock_vector_;  // 该表或该行已授予的各个锁的数量，5种锁

LockManager类的属性：

TransactionManager *txn_manager_;/** 结构，用于保存给定表oid的锁请求 */
std::unordered_map<table_oid_t, std::shared_ptr<LockRequestQueue>> table_lock_map_;  // 每个表对应的的锁队列
/** Coordination */
std::mutex table_lock_map_latch_;/** 结构，用于保存给定RID的锁请求 */
std::unordered_map<RID, std::shared_ptr<LockRequestQueue>> row_lock_map_;
/** Coordination */
std::mutex row_lock_map_latch_;std::atomic<bool> enable_cycle_detection_;
std::thread *cycle_detection_thread_;/** 等待图表示。 */
// std::unordered_map<txn_id_t, std::vector<txn_id_t>> waits_for_;
std::mutex waits_for_latch_;
// 重新设计wait_for_，使得若t1等待t2的资源，令t2指向t1，释放t2时减少所有等待t2的事务的入度
std::map<txn_id_t, std::set<txn_id_t>> waits_for_;  // 保证每次都从txn_id_t最小的事务开始扫描
std::map<txn_id_t, int> visited_;  // 表示一个节点未被访问过(0)，正在访问(1)，已经被访问完(2)
bool hascycle_;                    // 是否有环
std::unordered_map<txn_id_t, std::unordered_set<table_oid_t>> map_txn_table_; // 记录事务锁住的表
std::unordered_map<txn_id_t, std::unordered_set<RID>> map_txn_row_; // 记录事务锁住的行

LockTable

//lock_manager.h的LOCK_NOTE（部分）
/**
REPEATABLE_READ:事务需要获取所有锁。所有锁都允许处于GROWING状态在收缩状态下不允许有锁
READ_COMMITTED:事务需要获取所有锁。所有锁都允许处于GROWING状态只有IS、S锁允许处于收缩状态
READ_UNCOMMITTED:事务只需要接收IX、X锁。在GROWING状态下允许使用X、IX锁。S, IS, SIX是不允许的
**/

DealLockOfTable()要做的就是进行加锁的准备工作。先将table_lock_map_属性加锁，其表示表oid和锁请求队列间的映射，若其中无对应oid的内容则初始化oid的键值对。解锁table_lock_map_加锁表的锁请求队列，查找该表的锁队列，当前事务是否曾请求过，若未请求过就新建一个锁请求std::make_shared<LockRequest>(txn_id, lock_mode, oid);插入锁队列中，调用WaitForLock()返回。若已锁住该表，若这次请求的锁和之前的模式一样，将队列解锁，返回true。剩下的情况就是两次锁模式不同，尝试锁升级（说是升级，其实是锁模式切换，去旧换新），若有其他事务在升级该表的其他锁（），则解锁队列、设置事务状态为回滚、抛出UPGRADE_CONFLICT异常。取得队列中旧锁请求，判断锁升级是否兼容：

//lock_manager.h
/**
LOCK UPGRADE:在已经锁定的资源上调用Lock()应该具有以下行为:- 如果请求的锁模式与当前持有的锁模式相同，lock()应该返回true，因为它已经拥有该锁。- 如果请求的锁模式不同，lock()应该升级事务持有的锁。基本上应该有三个步骤来执行锁升级一般- 1。检查升级的前提条件- 2。放下当前锁，保留升级位置- 3。等待新锁授予正在升级的锁请求应该优先于同一资源上等待的其他锁请求。在升级时，只允许以下转换:IS -> [S, X, IX, SIX]S  -> [X, SIX]IX -> [X, SIX]SIX-> [X]任何其他升级都被认为是不兼容的，这样的尝试应该将TransactionState设置为ABORTED，并抛出TransactionAbortException (INCOMPATIBLE_UPGRADE)。此外，应该只允许一个事务在给定资源上升级其锁。同一资源上的多个并发锁升级应该将TransactionState设置为ABORTED，并抛出TransactionAbortException (UPGRADE_CONFLICT)。
**/

升级锁其实就是将锁请求队列中lock_vector_旧锁模式对应的计数减一，然后将reuest_queue_中的旧锁删除（erase()），调用RemoveFromSet()将事务中与旧锁模式相对应的表锁集合中的表oid删除，恢复granted_、设置upgrading_为当前事务id、更新锁模式、锁请求队列request_queue_中添加新锁请求，调用WaitForLock()等待授予锁。

WaitForLock()就是真正授予锁的函数了，前面一个while循环是为了阻塞其他线程进行加锁，之后和清除旧锁过程类似，granted_置true、调用InsertToSet()将表oid插入表锁集合中、锁队列的lock_vector_中对应锁模式的计数器加一、将upgrading_恢复。

WaitForLock()开头的while循环作用是实现线程的阻塞等待锁资源。具体来说，它使用了一个条件变量cv_和一个互斥锁std::unique_lock<std::mutex> mylock(lock_queue->latch_, std::adopt_lock);（std::adopt_lock托管了lock_queue->latch_给unique_lock）来实现线程的阻塞和唤醒。循环条件是CanGrantLock()返回false，其作用是根据upgranding_（当前请求对应的事务正在升级锁，是最高优先级）和锁请求在request_queue_中的顺序（没有事务正在升级锁，迭代器遍历找到第一个与其他所有已被授予的锁兼容CompatibleLock()但还未被授予!granted_的那个锁请求）判断当前请求现在是否适合获得锁。

只要不满足条件，进入循环，lock_queue->cv_.wait(mylock);阻塞并释放托管的lock_queue->latch直到解锁表的时候notify_all()唤醒，之后检测事务状态，若是ABORTED则直接从request_queue_中删除，因为处理死锁的线程会把死锁的事务的状态置为ABORTED

UnLockTable

分别处理：

1. table_lock_map_中没有表oid到锁请求队列的映射：ATTEMPTED_UNLOCK_BUT_NO_LOCK_HELD异常
2. HaveRowLock()通过获取事务的两种行锁集合（s_row_lock_set_和x_row_lock_set_）检测该事务在该表oid的行上有锁未释放：TABLE_UNLOCKED_BEFORE_UNLOCKING_ROWS异常
3. 迭代器遍历requeset_queue_检测事务没有表oid的锁：ATTEMPTED_UNLOCK_BUT_NO_LOCK_HELD异常

处理完异常状态后：

之后真正进入解锁代码，同加锁，lock_vector_对应锁模式加一，删除request_queue_中锁请求，RemoveFromSet()删除事务的特定锁模式对应的表锁集合中的表oid，唤醒所有线程，解锁锁请求队列。

LockRow

加行锁和加表锁类似，区别在于：

1. 行锁不支持意愿锁
2. 收缩阶段时，加X锁直接LOCK_ON_SHRINKING异常
3. 收缩阶段，可重复读和读未提交都不允许加锁，LOCK_ON_SHRINKING和LOCK_SHARED_ON_READ_UNCOMMITTED异常
4. 收缩阶段，读已提交只允许加S锁

DealLockOfRow()与DealLockOfTable()类似，区别在于：

1. 行锁开头需要检查本事务是否有相应表锁CheckTableLock()（取事务的行锁集合查找对应表的oid）没有就直接抛异常TABLE_LOCK_NOT_PRESENT
2. 升级锁时，因为只有两种锁，所以判断兼容就只需要判断不能是X锁转S锁即可

WaitForRowLock()与WaitForLock()极其相似，逻辑相同。

UnLockRow

解锁与解表锁同。

Task2 死锁检测

锁管理器应该在后台线程中运行死锁检测，定期构建等待图，并根据需要中止事务以消除死锁。

您可能需要从成员变量 txn_manager_ 访问事务的状态。如果’ txn_manager_ ‘被设置为’ nullptr '， ’ StartDeadlockDetection '将不会被调用，并且您不需要检测死锁。

/** 等待图表示。 */
std::mutex waits_for_latch_;
// 重新设计wait_for_，使得若t1等待t2的资源，令t2指向t1，释放t2时减少所有等待t2的事务的入度
std::map<txn_id_t, std::set<txn_id_t>> waits_for_;  // 保证每次都从txn_id_t最小的事务开始扫描
std::map<txn_id_t, int> visited_;  // 表示一个节点未被访问过(0)，正在访问(1)，已经被访问完(2)
//visited_的下标和waits_for_中的key和value集合里的值是匹配的
bool hascycle_;                    // 是否有环
std::unordered_map<txn_id_t, std::unordered_set<table_oid_t>> map_txn_table_;  // 记录事务锁住的表
std::unordered_map<txn_id_t, std::unordered_set<RID>> map_txn_row_; // 记录事务锁住的行

等待图一般都是等待的事务指向被等待的事务，但这里反过来了，因为之后需要当死锁时删除等待的事务，这样更快。

AddEdge()将等待的事务插入insert()进被等待的事务key对应的集合中。

RemoveEdge()删除erase()事务。

HasCycle()判断等待图中是否有环，先将访问列表visited_中所有结点初始化为0，之后遍历waits_for_，只要key事务未访问过就DFS()，使用的是深度优先搜索：

// 从txn_id开始深度优先遍历
void LockManager::DFS(txn_id_t txn_id) {visited_[txn_id] = 1;  // 该节点正在被访问for (const txn_id_t id : waits_for_[txn_id]) {//遍历等待该事务的事务集合if (visited_[id] == 0) {DFS(id);if (hascycle_) {//当递归找到一个环，直接返回return;}} else if (visited_[id] == 1) {hascycle_ = true;  // id被访问过, 现在又将被访问, 有环return;}}visited_[txn_id] = 2;  // 所有与txn_id相关的节点均已被访问，改为2, 无环
}

DFS()返回后若找到一个环就遍历visited_，如果是无环的事务结点会被改置为2，还为1的只会是return返回的有环结点，将该结点赋给入参，返回true。

GetEdgeList()嵌套循环waits_for_将边记录下返回。

RunCycleDetection()死锁检验线程的运行函数：

std::thread *cycle_detection_thread_;
std::atomic<bool> enable_cycle_detection_;void StartDeadlockDetection() {BUSTUB_ENSURE(txn_manager_ != nullptr, "txn_manager_ is not set.")enable_cycle_detection_ = true;cycle_detection_thread_ = new std::thread(&LockManager::RunCycleDetection, this);
}

RunCycleDetection()是一个循环函数，由enable_cycle_detection_控制，每次循环都会清空并重新创建等待图waits_for_并记录事务锁住的表和行（map_txn_table_和map_txn_row_），然后内部一个循环，由HasCycle()控制，只要有死锁就设置死锁事务状态为回滚，删除waits_for_中死锁事务相关的边（即等待关系），然后从死锁事务锁住的表和行（map_txn_table_和map_txn_row_）中获取表和行的id，从tabe_lock_map_和row_lock_map_中获得死锁锁住的表和行的锁请求队列，lock_vector_中该死锁事务的锁模式计数器减一，RemoFromSet()和RemoveFromRowSet()删除锁集合中的表oid和rid，删除锁请求队列中死锁事务的请求，lock_queue->cv_.notify_all();唤醒锁住的表和行的其他请求线程。

Task3 并发查询执行器

Isolation Level

• 一个事务应该为所有写操作持有X锁，直到它提交或终止，无论它的隔离级别如何。
• 对于 REPEATABLE_READ，一个事务应该为所有读操作持有S锁，直到提交或终止。
• 对于 READ_COMMITTED，一个事务应该为所有读操作占用S锁，但可以立即释放它们。
• 对于 READ_UNCOMMITTED，事务不需要为读操作获取任何S锁。

seq_scan_executor

Init()锁表，Next()锁表

//Init()开头
isl_ = exec_ctx_->GetTransaction()->GetIsolationLevel();
if (exec_ctx_->IsDelete()) {  // 当前操作是delete或updateTryLockTable(LockManager::LockMode::INTENTION_EXCLUSIVE, plan_->table_oid_);  // IX锁
} else {if (isl_ == IsolationLevel::REPEATABLE_READ || isl_ == IsolationLevel::READ_COMMITTED) {TryLockTable(LockManager::LockMode::INTENTION_SHARED, plan_->table_oid_);  // IS锁}
}//Next()开头
if (exec_ctx_->IsDelete()) {TryLockRow(LockManager::LockMode::EXCLUSIVE, plan_->table_oid_, it_->GetRID());  // 删除或更新操作加写锁
} else {if (isl_ == IsolationLevel::REPEATABLE_READ || isl_ == IsolationLevel::READ_COMMITTED) {TryLockRow(LockManager::LockMode::SHARED, plan_->table_oid_, it_->GetRID());  // 只读操作加读锁}
}

TryLockXXX()是将加锁函数try-catch包围，拼凑出错时的异常信息，解锁TryUnLockXXX()同理。

在Next()获取元组后，判断元组未删除（pair.first.is_deleted_）若执行器上下文属性exec_ctx_的is_delete_（标识这次查询表是否是为了删除/更改表中数据）为false，且隔离级别为READ_COMMITTED，就立即解锁行。已删除时直接解锁行。当查询到表末尾后，隔离状态是READ_COMMITTED且执行器不会删除/修改数据，则解锁表。

insert_executor

try {if (!exec_ctx_->GetLockManager()->LockTable(exec_ctx_->GetTransaction(), LockManager::LockMode::INTENTION_EXCLUSIVE,plan_->table_oid_)) {throw ExecutionException(std::string("Insert Table Ix lock fail"));}
} catch (TransactionAbortException &e) {throw ExecutionException(std::string("Insert Table Ix lock fail"));
}

delete_executor

如果你在执行器context中正确地基于’ IsDelete() ‘实现了’ SeqScanExecutor '，那么你不需要在这个执行器中使用任何锁。

transaction_manager

主要实现两个函数：TransactionManager::Abort()和TransactionManager::Commit()

Commit()先调用ReleaseLocks()释放事务持有的所有锁：先从事务的各个锁集合中收集行、表锁到新容器中，然后嵌套循环解锁行锁、循环解锁表锁。之后设置事务状态为COMMITTED

Abort()负责恢复事务的写集中的所有更改。每次对表进行插入、删除时都会在事务的table_write_set_和index_write_set_中插入一条TableWriteRecord和IndexWriteRecord，其中有操作的各种信息（如操作的表oid、对应的表堆、行rid，索引oid等）。遍历两个写集合，对里面存放的修改过的元组（主要是插入和删除的）的元信息进行回退（设置is_deleted_），然后弹出集合，设置事务状态为ABOTTED