前言
这是 Fall 2019 的最后一个实验，要求我们实现预写式日志、系统恢复和存档点功能，这三个功能分别对应三个类 LogManager、LogRecovery 和 CheckpointManager，下面进入正题。
代码实现
日志管理器
为了达到原子性和持久性的目标，数据库系统会将描述事务所做修改的信息保存硬盘中。这些信息确保已提交事务中执行的所有修改都反映在数据库中，还可以确保系统崩溃并重新启动后，由中止或失败的事务所做的修改不会保留在数据库中。本次实验使用预写日志记录这些修改，预写日志也是使用的最广泛的记录方式，基本原理如下图所示。
在内存中有一块缓冲区域 WAL Buffer，用于记录任何事务中执行的操作，每当执行一个操作，就会在缓冲区中添加一条记录，记录的格式有三种：物理日志、逻辑日志和混合式日志。物理日志记录了操作前后每个数据位的修改，逻辑日志只记录了 SQL 语句，而混合日志和物理日志很像，不过将偏移量换成了槽号。逻辑日志存在一些问题，比如重新执行 Now() 时间会发生改变，而物理日志的偏移量也会有问题，如果对页进行碎片整理会导致偏移量失效，所以实际上使用的是混合式的日志。
在 WAL Buffer 中添加完一条记录后，才会修改缓冲池中的数据，当日志被提交或者 WAL Buffer 满了之后（取决于具体策略），会将日志写到硬盘上，虽然这时候缓冲池中的脏页可能还没同步到硬盘上，但是只要保存完日志，我们就能保证数据已经安全了。正是因为缓冲池未动，日志先行，所以这种策略称为预写式日志。
日志管理器 LogManager 的声明如下所示，可以看到内部有两个缓冲区： log_buffer_ 和 flush_buffer_，前者用于添加记录，当满足一定条件时（后面会说到），需要交换这两个缓冲区的内容，然后使用 flush_thread_ 将 flush_buffer_ 写到硬盘上：
class LogManager { public: explicit LogManager(DiskManager *disk_manager) : next_lsn_(0), persistent_lsn_(INVALID_LSN), disk_manager_(disk_manager) { log_buffer_ = new char[LOG_BUFFER_SIZE]; flush_buffer_ = new char[LOG_BUFFER_SIZE]; } ~LogManager() { delete[] log_buffer_; delete[] flush_buffer_; log_buffer_ = nullptr; flush_buffer_ = nullptr; } void RunFlushThread(); void StopFlushThread(); /* flush log to disk */ void Flush(); lsn_t AppendLogRecord(LogRecord *log_record); inline lsn_t GetNextLSN() { return next_lsn_; } inline lsn_t GetPersistentLSN() { return persistent_lsn_; } inline void SetPersistentLSN(lsn_t lsn) { persistent_lsn_ = lsn; } inline char *GetLogBuffer() { return log_buffer_; } private: /** The atomic counter which records the next log sequence number. */ std::atomic<lsn_t> next_lsn_; /** The log records before and including the persistent lsn have been written to disk. */ std::atomic<lsn_t> persistent_lsn_; char *log_buffer_; char *flush_buffer_; int log_buffer_offset_ = 0; int flush_buffer_offset_ = 0; std::mutex latch_; std::thread *flush_thread_; std::condition_variable cv_; std::condition_variable cv_append_; std::atomic_bool need_flush_ = false; DiskManager *disk_manager_; }; 启动日志线程
当满足下述条件之一时，我们会使用日志线程将日志写到硬盘上：
log_buffer_的剩余空间不足以插入新的记录- 距离上一次保存日志的时间超过了
log_timeout - 缓冲池换出了一个脏页
实验提示说要用到 Future 和 Promise，但是感觉条件变量就够用了，加上一个 need_flush_ 判断条件避免发生虚假唤醒：
void LogManager::RunFlushThread() { if (enable_logging) { return; } enable_logging = true; flush_thread_ = new std::thread([&] { while (enable_logging) { std::unique_lock<std::mutex> lock(latch_); // flush log to disk if log time out or log buffer is full cv_.wait_for(lock, log_timeout, [&] { return need_flush_.load(); }); if (log_buffer_offset_ > 0) { std::swap(log_buffer_, flush_buffer_); std::swap(log_buffer_offset_, flush_buffer_offset_); disk_manager_->WriteLog(flush_buffer_, flush_buffer_offset_); flush_buffer_offset_ = 0; SetPersistentLSN(next_lsn_ - 1); } need_flush_ = false; cv_append_.notify_all(); } }); } 停止日志线程
当数据库系统被关闭时，我们应该停止日志线程，同时将 log_buffer_ 中的记录全部保存到硬盘中：
void LogManager::StopFlushThread() { enable_logging = false; Flush(); flush_thread_->join(); delete flush_thread_; flush_thread_ = nullptr; } void LogManager::Flush() { if (!enable_logging) { return; } std::unique_lock<std::mutex> lock(latch_); need_flush_ = true; cv_.notify_one(); // block thread until flush finished cv_append_.wait(lock, [&] { return !need_flush_.load(); }); } 添加日志记录
根据执行操作的不同，日志记录也分为多个种类：
enum class LogRecordType { INVALID = 0, INSERT, MARKDELETE, APPLYDELETE, ROLLBACKDELETE, UPDATE, BEGIN, COMMIT, ABORT, /** Creating a new page in the table heap. */ NEWPAGE, }; 日志记录由 LogRecord 类描述，每一种记录的格式如下所示：
Header (每种类型都拥有 Header，共 20 字节) -------------------------------------------- | size | LSN | transID | prevLSN | LogType | -------------------------------------------- 插入类型日志记录 -------------------------------------------------------------- | HEADER | tuple_rid | tuple_size | tuple_data(char[] array) | -------------------------------------------------------------- 删除类型日志记录 (包括 markdelete, rollbackdelete, applydelete) -------------------------------------------------------------- | HEADER | tuple_rid | tuple_size | tuple_data(char[] array) | -------------------------------------------------------------- 更新类型日志记录 ---------------------------------------------------------------------------------- | HEADER | tuple_rid | tuple_size | old_tuple_data | tuple_size | new_tuple_data | ---------------------------------------------------------------------------------- 新页类型日志记录 ----------------------------------- | HEADER | prev_page_id | page_id | ----------------------------------- 我们需要根据不同类型日志记录的格式将日志记录序列化到 log_buffer_ 中：
lsn_t LogManager::AppendLogRecord(LogRecord *log_record) { std::unique_lock<std::mutex> lock(latch_); // flush log to disk when the log buffer is full if (log_record->size_ + log_buffer_offset_ > LOG_BUFFER_SIZE) { // wake up flush thread to write log need_flush_ = true; cv_.notify_one(); // block current thread until log buffer is emptied cv_append_.wait(lock, [&] { return log_record->size_ + log_buffer_offset_ <= LOG_BUFFER_SIZE; }); } // serialize header log_record->lsn_ = next_lsn_++; memcpy(log_buffer_ + log_buffer_offset_, log_record, LogRecord::HEADER_SIZE); int pos = log_buffer_offset_ + LogRecord::HEADER_SIZE; // serialize body switch (log_record->GetLogRecordType()) { case LogRecordType::INSERT: memcpy(log_buffer_ + pos, &log_record->insert_rid_, sizeof(RID)); pos += sizeof(RID); log_record->insert_tuple_.SerializeTo(log_buffer_ + pos); break; case LogRecordType::MARKDELETE: case LogRecordType::APPLYDELETE: case LogRecordType::ROLLBACKDELETE: memcpy(log_buffer_ + pos, &log_record->delete_rid_, sizeof(RID)); pos += sizeof(RID); log_record->delete_tuple_.SerializeTo(log_buffer_ + pos); break; case LogRecordType::UPDATE: memcpy(log_buffer_ + pos, &log_record->update_rid_, sizeof(RID)); pos += sizeof(RID); log_record->old_tuple_.SerializeTo(log_buffer_ + pos); pos += 4 + static_cast<int>(log_record->old_tuple_.GetLength()); log_record->new_tuple_.SerializeTo(log_buffer_ + pos); break; case LogRecordType::NEWPAGE: memcpy(log_buffer_ + pos, &log_record->prev_page_id_, sizeof(page_id_t)); pos += sizeof(page_id_t); memcpy(log_buffer_ + pos, &log_record->page_id_, sizeof(page_id_t)); break; default: break; } // update log buffer offset log_buffer_offset_ += log_record->size_; return log_record->lsn_; } 事务管理器和缓冲池管理器
在我们调用 TablePage::InsertTuple 等方法的时候，内部会调用 LogManager::AppendLogRecord 添加日志记录，但是事务开始、提交或者终止时也需要我们添加记录：
Transaction *TransactionManager::Begin(Transaction *txn) { // Acquire the global transaction latch in shared mode. global_txn_latch_.RLock(); if (txn == nullptr) { txn = new Transaction(next_txn_id_++); } if (enable_logging) { LogRecord log_record(txn->GetTransactionId(), txn->GetPrevLSN(), LogRecordType::BEGIN); auto lsn = log_manager_->AppendLogRecord(&log_record); txn->SetPrevLSN(lsn); } txn_map[txn->GetTransactionId()] = txn; return txn; } void TransactionManager::Commit(Transaction *txn) { txn->SetState(TransactionState::COMMITTED); // 省略部分代码 if (enable_logging) { LogRecord log_record(txn->GetTransactionId(), txn->GetPrevLSN(), LogRecordType::COMMIT); auto lsn = log_manager_->AppendLogRecord(&log_record); txn->SetPrevLSN(lsn); } // Release all the locks. ReleaseLocks(txn); // Release the global transaction latch. global_txn_latch_.RUnlock(); } void TransactionManager::Abort(Transaction *txn) { txn->SetState(TransactionState::ABORTED); // 省略部分代码 if (enable_logging) { LogRecord log_record(txn->GetTransactionId(), txn->GetPrevLSN(), LogRecordType::ABORT); auto lsn = log_manager_->AppendLogRecord(&log_record); txn->SetPrevLSN(lsn); } // Release all the locks. ReleaseLocks(txn); // Release the global transaction latch. global_txn_latch_.RUnlock(); } 如前所述，将缓冲池中的脏页换出时需要强制保存日志，在 Buffer Pool Manager 实验中我们实现了一个 GetVictimFrameId 方法，只要略作修改即可：
frame_id_t BufferPoolManager::GetVictimFrameId() { frame_id_t frame_id; if (!free_list_.empty()) { frame_id = free_list_.front(); free_list_.pop_front(); } else { if (!replacer_->Victim(&frame_id)) { return INVALID_PAGE_ID; } // flush log to disk when victim a dirty page if (enable_logging) { Page &page = pages_[frame_id]; if (page.IsDirty() && page.GetLSN() > log_manager_->GetPersistentLSN()) { log_manager_->Flush(); } } } return frame_id; } 测试
日志管理器的测试结果如下所示：
系统恢复
本次实验使用的系统恢复策略较为简单，由于没有使用 Fuzzy Checkpoint，所以少了 Analysis 阶段，直接变成在 LogRecovery::Redo 函数中分析出当前活跃事务表 ATT 并进行重放，在 LogRecovery::Undo 函数中进行回滚。
日志记录反序列化
LogRecovery 会不断调用 DiskManager::ReadLog 函数直到读取完整个日志，由于我们先前将 LogRecord 进行了序列化，此处需要进行反序列化以访问记录中的信息。由于 log_buffer_ 的大小为 LOG_BUFFER_SIZE，所以将日志文件读取到 log_buffer_ 的过程中可能截断最后一条记录，这时候需要返回 false 以表示反序列化失败：
bool LogRecovery::DeserializeLogRecord(const char *data, LogRecord *log_record) { // convert data to record and check header auto record = reinterpret_cast<const LogRecord *>(data); if (record->size_ <= 0 || data + record->size_ > log_buffer_ + LOG_BUFFER_SIZE) { return false; } // copy header memcpy(reinterpret_cast<char *>(log_record), data, LogRecord::HEADER_SIZE); // copy body int pos = LogRecord::HEADER_SIZE; switch (log_record->GetLogRecordType()) { case LogRecordType::INSERT: memcpy(&log_record->insert_rid_, data + pos, sizeof(RID)); pos += sizeof(RID); log_record->insert_tuple_.DeserializeFrom(data + pos); break; case LogRecordType::MARKDELETE: case LogRecordType::APPLYDELETE: case LogRecordType::ROLLBACKDELETE: memcpy(&log_record->delete_rid_, data + pos, sizeof(RID)); pos += sizeof(RID); log_record->delete_tuple_.DeserializeFrom(data + pos); break; case LogRecordType::UPDATE: memcpy(&log_record->update_rid_, data + pos, sizeof(RID)); pos += sizeof(RID); log_record->old_tuple_.DeserializeFrom(data + pos); pos += 4 + log_record->old_tuple_.GetLength(); log_record->new_tuple_.DeserializeFrom(data + pos); break; case LogRecordType::NEWPAGE: memcpy(&log_record->prev_page_id_, data + pos, sizeof(page_id_t)); pos += sizeof(page_id_t); memcpy(&log_record->page_id_, data + pos, sizeof(page_id_t)); break; case LogRecordType::BEGIN: case LogRecordType::COMMIT: case LogRecordType::ABORT: break; default: return false; } return true; } 重放
重放过程十分简单粗暴，遍历整个日志的记录，如果记录的日志序号大于记录操作的 tuple 保存到磁盘上的日志序列号，说明 tuple 被修改后还没保存到磁盘上就宕机了，这时候需要进行回放。在遍历的时候先无脑将记录对应的事务添加到 ATT 中，直到事务被提交或者中断才将其移出 ATT。
void LogRecovery::Redo() { while (disk_manager_->ReadLog(log_buffer_, LOG_BUFFER_SIZE, offset_)) { // offset of current log buffer size_t pos = 0; LogRecord log_record; // deserialize log entry to record while (DeserializeLogRecord(log_buffer_ + pos, &log_record)) { // update lsn mapping auto lsn = log_record.lsn_; lsn_mapping_[lsn] = offset_ + pos; // Add txn to ATT with status UNDO active_txn_[log_record.txn_id_] = lsn; pos += log_record.size_; // redo if page was not wirtten to disk when crash happened switch (log_record.log_record_type_) { case LogRecordType::INSERT: { auto page = getTablePage(log_record.insert_rid_); if (page->GetLSN() < lsn) { page->WLatch(); page->InsertTuple(log_record.insert_tuple_, &log_record.insert_rid_, nullptr, nullptr, nullptr); page->WUnlatch(); } buffer_pool_manager_->UnpinPage(page->GetPageId(), page->GetLSN() < lsn); break; } case LogRecordType::UPDATE: { auto page = getTablePage(log_record.update_rid_); if (page->GetLSN() < lsn) { page->WLatch(); page->UpdateTuple(log_record.new_tuple_, &log_record.old_tuple_, log_record.update_rid_, nullptr, nullptr, nullptr); page->WUnlatch(); } buffer_pool_manager_->UnpinPage(page->GetPageId(), page->GetLSN() < lsn); break; } case LogRecordType::MARKDELETE: case LogRecordType::APPLYDELETE: case LogRecordType::ROLLBACKDELETE: { auto page = getTablePage(log_record.delete_rid_); if (page->GetLSN() < lsn) { page->WLatch(); if (log_record.log_record_type_ == LogRecordType::MARKDELETE) { page->MarkDelete(log_record.delete_rid_, nullptr, nullptr, nullptr); } else if (log_record.log_record_type_ == LogRecordType::APPLYDELETE) { page->ApplyDelete(log_record.delete_rid_, nullptr, nullptr); } else { page->RollbackDelete(log_record.delete_rid_, nullptr, nullptr); } page->WUnlatch(); } buffer_pool_manager_->UnpinPage(page->GetPageId(), page->GetLSN() < lsn); break; } case LogRecordType::COMMIT: case LogRecordType::ABORT: active_txn_.erase(log_record.txn_id_); break; case LogRecordType::NEWPAGE: { auto page_id = log_record.page_id_; auto page = getTablePage(page_id); if (page->GetLSN() < lsn) { auto prev_page_id = log_record.prev_page_id_; page->WLatch(); page->Init(page_id, PAGE_SIZE, prev_page_id, nullptr, nullptr); page->WUnlatch(); if (prev_page_id != INVALID_PAGE_ID) { auto prev_page = getTablePage(prev_page_id); if (prev_page->GetNextPageId() != page_id) { prev_page->SetNextPageId(page_id); buffer_pool_manager_->UnpinPage(prev_page_id, true); } else { buffer_pool_manager_->UnpinPage(prev_page_id, false); } } } buffer_pool_manager_->UnpinPage(page_id, page->GetLSN() < lsn); break; } default: break; } } offset_ += pos; } } 回滚
LogRecovery::Undo 会遍历 ATT 中的每一个事务，对事务的操作进行回滚，回滚的规则如下：
- 如果日志记录类型为
LogRecordType::INSERT，使用TablePage::ApplyDelete进行回滚 - 如果日志记录类型为
LogRecordType::UPDATE，使用TablePage::UpdateTuple进行回滚 - 如果日志记录类型为
LogRecordType::APPLYDELETE，使用TablePage::InsertTuple进行回滚 - 如果日志记录类型为
LogRecordType::MARKDELETE，使用TablePage::RollbackDelete进行回滚 - 如果日志记录类型为
LogRecordType::ROLLBACKDELETE，使用TablePage::MarkDelete进行回滚
void LogRecovery::Undo() { for (auto [txn_id, lsn] : active_txn_) { while (lsn != INVALID_LSN) { // read log from dist and convert log buffer entry to log record LogRecord log_record; auto offset = lsn_mapping_[lsn]; disk_manager_->ReadLog(log_buffer_, LOG_BUFFER_SIZE, offset); DeserializeLogRecord(log_buffer_, &log_record); lsn = log_record.GetPrevLSN(); // rollback switch (log_record.GetLogRecordType()) { case LogRecordType::INSERT: { auto page = getTablePage(log_record.insert_rid_); page->WLatch(); page->ApplyDelete(log_record.insert_rid_, nullptr, nullptr); page->WUnlatch(); buffer_pool_manager_->UnpinPage(page->GetPageId(), true); break; } case LogRecordType::UPDATE: { auto page = getTablePage(log_record.update_rid_); page->WLatch(); page->UpdateTuple(log_record.old_tuple_, &log_record.new_tuple_, log_record.update_rid_, nullptr, nullptr, nullptr); page->WUnlatch(); buffer_pool_manager_->UnpinPage(page->GetPageId(), true); break; } case LogRecordType::MARKDELETE: case LogRecordType::APPLYDELETE: case LogRecordType::ROLLBACKDELETE: { auto page = getTablePage(log_record.delete_rid_); page->WLatch(); if (log_record.log_record_type_ == LogRecordType::MARKDELETE) { page->RollbackDelete(log_record.delete_rid_, nullptr, nullptr); } else if (log_record.log_record_type_ == LogRecordType::APPLYDELETE) { page->InsertTuple(log_record.delete_tuple_, &log_record.delete_rid_, nullptr, nullptr, nullptr); } else { page->MarkDelete(log_record.delete_rid_, nullptr, nullptr, nullptr); } page->WUnlatch(); buffer_pool_manager_->UnpinPage(page->GetPageId(), true); break; } default: break; } } } active_txn_.clear(); lsn_mapping_.clear(); } 存档点管理器
存档点管理器用于保存日志并刷出缓冲池中所有的脏页，同时阻塞正在进行的事务。
void CheckpointManager::BeginCheckpoint() { transaction_manager_->BlockAllTransactions(); log_manager_->Flush(); buffer_pool_manager_->FlushAllPages(); } void CheckpointManager::EndCheckpoint() { transaction_manager_->ResumeTransactions(); } 测试
日志恢复和存档点的测试结果如下：
总结
本次实验主要考察对预写式日志和数据库系统恢复的理解，代码上层面多了对多线程同步技术的要求，整个实验做下来感觉比较顺（除了被段错误坑了亿点时间外），以上~~
