1.3 具体实现细节
1. 我们需要弄清楚各种锁
2. 任务一实现上的一些说明
1. 对于S锁
只需要考虑下面这些情况
简单附上一些代码
if (mode == LockMode::SHARED) {
auto shared_wait_for = [&]() { return !lock_queue.upgrading_ && !lock_queue.hasExclusiveLock(txn); };
while (!shared_wait_for()) {
lock_queue.cv_.wait(_lock);
}
txn->GetSharedLockSet()->emplace(rid);
}
lock_queue.hasExclusiveLock(txn)函数
inline bool hasExclusiveLock(Transaction *txn) {
std::list<LockRequest>::iterator curr = request_queue_.begin();
for (; curr != request_queue_.end(); curr++) {
if (curr->lock_mode_ == LockMode::EXCLUSIVE) {
return true;
}
}
return false;
}
2. 对于X锁
同样附上一些简单的代码
if (mode == LockMode::EXCLUSIVE) {
auto exclusive_wait_for = [&]() { return !lock_queue.upgrading_ && lock_queue.request_queue_.size() == 1; };
while (!exclusive_wait_for()) {
lock_queue.cv_.wait(_lock);
}
txn->GetExclusiveLockSet()->emplace(rid);
}
和共享锁的实现基本类似
3. 对于U锁
附上带注释的核心逻辑代码。基本可以说明白这个地方
// 标记位更新。表明现在正处于等待update锁阶段
queue.upgrading_ = true;
// 假如说当前的request_queue中只有当前update lock这一个请求。则可以加U锁，否则应该wait
while (lock_table_[rid].request_queue_.size() != 1) {
queue.cv_.wait(unique_lock);
}
// 加X锁。并把标记位重制
queue.request_queue_.back() = LockRequest(txn->GetTransactionId(), LockMode::EXCLUSIVE);
queue.upgrading_ = false;
Task2 DEADLOCK DETECTION
这个任务要求你的LM能够进行死锁检测。死锁检测算法就是最常见的资源分配图算法
下面用cmu课上ppt的例子来看一下这个算法。
这样形成了一个环就发生了死锁，这个时候就需要abort
2.1 一些来自TA和官网的建议
2.2 具体实现
1. remove和add的操作比较简单
就直接附上代码不说废话
void LockManager::AddEdge(txn_id_t t1, txn_id_t t2) {
for (const auto &txn_id : waits_for_[t1]) {
if (txn_id == t2) {
return;
}
}
waits_for_[t1].push_back(t2);
}
void LockManager::RemoveEdge(txn_id_t t1, txn_id_t t2) {
LOG_DEBUG("we can remove edge");
auto &vec = waits_for_[t1];
for (auto iter = vec.begin(); iter != vec.end(); ++iter) {
if (*iter == t2) {
vec.erase(iter);
return;
}
}
}
2. hasCycle函数
这个函数的实现。就是对于我们上面算法的实现。由于依赖图是一个有向图。因此我们需要知道如何判断一个有向图是否有环。
用简单的dfs就可以实现这一功能。当然除了一个visited数组来判断这个元素是否被遍历过之外。我们还需要另外一个数组recStack用来 keep track of vertices in the recursion stack.
具体的过程就和下图一样
下面附上上面那个算法的代码实习。但是关于本任务需要的代码没有给出
// This function is a variation of DFSUtil() in https://www.geeksforgeeks.org/archives/18212
bool Graph::isCyclicUtil(int v, bool visited[], bool *recStack)
{
if (visited[v] == false)
{
// Mark the current node as visited and part of recursion stack
visited[v] = true;
recStack[v] = true;
// Recur for all the vertices adjacent to this vertex
list<int>::iterator i;
for(i = adj[v].begin(); i != adj[v].end(); ++i)
{
if( !visited[*i] && isCyclicUtil(*i, visited, recStack) )
return true;
else if (recStack[*i])
return true;
}
}
recStack[v] = false; // remove the vertex from recursion stack
return false;
}
3. 对于构建图的补充
因为构建图之前我们要获取所有已经加锁和等待锁的事物id。
这里用两个函数来实现这两个步骤
这里附上找到所有等待锁事物的函数。另一个不再给出。
std::unordered_set<txn_id_t> getWaitingSet() {
std::list<LockRequest>::iterator wait_start;
std::unordered_set<txn_id_t> blocking;
// 遍历找到wait_start的位置
std::list<LockRequest>::iterator curr = request_queue_.begin();
for (; curr != request_queue_.end() && (curr->lock_mode_ == LockMode::SHARED || curr ->lock_mode_ == LockMode::EXCLUSIVE); curr++) {
}
wait_start = curr;
for (; wait_start != request_queue_.end(); wait_start++) {
if (GetTransaction(wait_start->txn_id_)->GetState() != TransactionState::ABORTED) {
blocking.insert(wait_start->txn_id_);
}
}
return blocking;
}
};
Task3
3.1 隔离级别
Read Uncommitted（读取未提交内容）
在该隔离级别，所有事务都可以看到其他未提交事务的执行结果。本隔离级别很少用于实际应用，因为它的性能也不比其他级别好多少。读取未提交的数据，也被称之为脏读（Dirty Read）。
就好比还没确定的消息，你却先知道了发布出去，最后又变更了，这样就发生了错误
Read Committed（读取提交内容）
这是大多数数据库系统的默认隔离级别（但不是MySQL默认的）。它满足了隔离的简单定义：一个事务只能看见已经提交事务所做的改变。这种隔离级别 也支持所谓的不可重复读（Nonrepeatable Read），因为同一事务的其他实例在该实例处理其间可能会有新的commit，所以同一select可能返回不同结果。
Repeatable Read（可重读）
这是MySQL的默认事务隔离级别，它确保同一事务的多个实例在并发读取数据时，会看到同样的数据行。不过理论上，这会导致另一个棘手的问题：幻读 （Phantom Read）。简单的说，幻读指当用户读取某一范围的数据行时，另一个事务又在该范围内插入了新行，当用户再读取该范围的数据行时，会发现有新的“幻影” 行。
3.2 对于seqScan的并发控制
由于对于整个表的遍历。也就是读操作，在并发的情况下可能发生错误。所以必须加以控制
这里附上一些加锁的代码进行解释
首先是对于隔离级别的区分
Transaction *txn = exec_ctx_->GetTransaction();
LockManager *lock_mgr = exec_ctx_->GetLockManager();
if (lock_mgr != nullptr) {
switch (txn->GetIsolationLevel()) {
case IsolationLevel::READ_UNCOMMITTED:
break;
case IsolationLevel::READ_COMMITTED:
case IsolationLevel::REPEATABLE_READ:
RID r = iter->GetRid();
if (txn->GetSharedLockSet()->empty() && txn->GetExclusiveLockSet()->empty()) {
lock_mgr->LockShared(txn, r);
txn->GetSharedLockSet()->insert(r);
}
break;
}
}
3.3 对于update的并发控制
这个更新比较简单。
主要有下面两个原则
if (lock_mgr != nullptr) {
if (txn->IsSharedLocked(*rid)) {
lock_mgr->LockUpgrade(txn, *rid);
txn->GetSharedLockSet()->erase(*rid);
txn->GetExclusiveLockSet()->insert(*rid);
} else if (txn->GetExclusiveLockSet()->empty()) {
lock_mgr->LockExclusive(txn, *rid);
}
}
对于delete的并发控制和update的完全一样。只需要加入上面的原则即可
总结
总算磕磕绊绊把四个lab都写完了。感谢在知乎和github以及qq群里面得到的各种帮助。后面准备配合这门课的要求把对应章节的书的内容读一下。同时把之前没有写完的上课笔记写完。顺带有时间把前面lab的博客改一下。因为实现方面有了变化。后面就准备开搞下一个lab了。不知道大家是推荐824分布式还是推荐os那。