CMU15445 (Fall 2019) 之 Project#1

前言

这个实验有两个任务：时钟替换算法和缓冲池管理器，分别对应 ClockReplacer 和 BufferPoolManager 类，BufferPoolManager 会用 ClockReplacer 挑选被换出的页，并通过 DiskManager 将换出的页写到数据库文件中。下面介绍这两个类的实现过程。

代码实现

如果直接克隆 Bustub 仓库，得到的是 fall 2021 的实验代码，对于 fall 2019，可以将 commit 切换至 5972018: Fix typo in type.cpp(#66)。但是这样引入一个坑，就是需要将 build_support/gtest_CMakeLists.txt.in 的内容改为：

cmake_minimum_required(VERSION 3.8)

project(googletest-download NONE)

include(ExternalProject)
ExternalProject_Add(googletest
        GIT_REPOSITORY git@github.com:google/googletest.git
        GIT_TAG main
        SOURCE_DIR "${CMAKE_BINARY_DIR}/googletest-src"
        BINARY_DIR "${CMAKE_BINARY_DIR}/googletest-build"
        CONFIGURE_COMMAND ""
        BUILD_COMMAND ""
        INSTALL_COMMAND ""
        TEST_COMMAND ""
        )

这里主要修改了 GIT_TAG 为 main，因为 googletest 仓库似乎将 master 分支重命名为 main 了。

ClockReplacer 类

项目主页对该类的实现方式做出了一点介绍：

The size of the ClockReplacer is the same as buffer pool since it contains placeholders for all of the frames in the BufferPoolManager. However, not all the frames are considered as in the ClockReplacer. The ClockReplacer is initialized to have no frame in it. Then, only the newly unpinned ones will be considered in the ClockReplacer. Adding a frame to or removing a frame from a replacer is implemented by changing a reference bit of a frame. The clock hand initially points to the placeholder of frame 0. For each frame, you need to track two things: 1. Is this frame currently in the ClockReplacer? 2. Has this frame recently been unpinned (ref flag)?

In some scenarios, the two are the same. For example, when you unpin a page, both of the above are true. However, the frame stays in the ClockReplacer until it is pinned or victimized, but its ref flag is modified by the clock hand.

简单翻译一下，就是 ClockReplacer 类内部维护了一个 frame 的集合，集合大小和缓冲池的大小一致。由于缓冲池中的某些 frame 正在被别的线程访问，这些 frame 的 pin count （等于访问该帧的线程数量）会大于 0，此时这些 frame 不允许被换出，换个角度来说，就是这些 frame 不在 ClockReplacer 维护的集合中。对于可以被换出的 frame，它必须满足两个条件：

• pin count 为 0，即该帧在 ClockReplacer 中。一旦某个帧的 pin count 大于零，就要被移出 ClockReplacer（调用 ClockReplacer::Pin）
• reference bit 为 false，即该帧最近没被访问过。对于 pin count 刚变成 0 而被加入 ClockReplacer 的帧而言，由于它刚被访问过，所以其 reference bit 为 true（调用 ClockReplacer::Unpin）

至于时钟替换算法的过程，其实就是按顺序从 frame 集合中挑选出一个满足上述换出条件的过程。为了维护时钟指针的位置并保证线程安全，需要添加一个时钟指针成员 clock_hand_ 和一个读写锁 mutex_，帧集合 frames_ 的每个元素代表该帧是否在 ClockReplacer 中及其 reference bit：

/**
 * ClockReplacer implements the clock replacement policy, which approximates the Least Recently Used policy.
 */
class ClockReplacer : public Replacer {
 public:
  /**
   * Create a new ClockReplacer.
   * @param num_pages the maximum number of pages the ClockReplacer will be required to store
   */
  explicit ClockReplacer(size_t num_pages);

  /**
   * Destroys the ClockReplacer.
   */
  ~ClockReplacer() override;

  bool Victim(frame_id_t *frame_id) override;

  void Pin(frame_id_t frame_id) override;

  void Unpin(frame_id_t frame_id) override;

  size_t Size() override;

 private:
  frame_id_t clock_hand_ = 0;
  std::vector<std::tuple<bool, bool>> frames_;
  std::shared_mutex mutex_;
};

各个方法的定义如下，里面使用了 std::lock_guard 以保证代码是异常安全的：

ClockReplacer::ClockReplacer(size_t num_pages) {
  for (size_t i = 0; i < num_pages; ++i) {
    frames_.push_back(std::make_tuple(false, false));
  }
}

ClockReplacer::~ClockReplacer() = default;

bool ClockReplacer::Victim(frame_id_t *frame_id) {
  if (Size() == 0) {
    return false;
  }

  std::lock_guard<std::shared_mutex> lock(mutex_);
  while (true) {
    auto &[contains, ref] = frames_[clock_hand_];
    if (contains) {
      if (ref) {
        ref = false;
      } else {
        *frame_id = clock_hand_;
        contains = false;
        return true;
      }
    }
    clock_hand_ = (clock_hand_ + 1) % frames_.size();
  }
}

void ClockReplacer::Pin(frame_id_t frame_id) {
  assert(static_cast<size_t>(frame_id) < frames_.size());
  std::lock_guard<std::shared_mutex> lock(mutex_);
  auto &[contains, ref] = frames_[frame_id];
  contains = false;
  ref = false;
}

void ClockReplacer::Unpin(frame_id_t frame_id) {
  assert(static_cast<size_t>(frame_id) < frames_.size());
  std::lock_guard<std::shared_mutex> lock(mutex_);
  auto &[contains, ref] = frames_[frame_id];
  contains = true;
  ref = true;
}

size_t ClockReplacer::Size() {
  std::shared_lock<std::shared_mutex> lock(mutex_);
  size_t size = 0;
  for (auto &[contains, ref] : frames_) {
    size += contains;
  }
  return size;
}

在终端输入命令：

mkdir build
cd build
cmake ..
make clock_replacer_test
./test/clock_replacer_test

测试结果如下：

BufferPoolManager 类

这里将互斥锁换成了读写锁，用于保护 page_table_、pages_ 和 free_list_，同时引入了一个辅助函数 GetVictimFrameId()：

class BufferPoolManager {
 // 省略部分代码

 protected:

  /**
   * select a victim frame from the free list or replacer.
   * @return the frame id, INVALID_PAGE_ID if the victim could not be found
   */
  frame_id_t GetVictimFrameId();

  /** This latch protects shared data structures. We recommend updating this comment to describe what it protects. */
  std::shared_mutex latch_;
};

BufferPoolManager 类要求我们实现五个函数：

• FetchPageImpl(page_id)
• NewPageImpl(page_id)
• UnpinPageImpl(page_id, is_dirty)
• FlushPageImpl(page_id)
• DeletePageImpl(page_id)
• FlushAllPagesImpl()

下面会一个个实现上述函数。

FetchPageImpl(page_id)

该函数实现了缓冲池的主要功能：向上层提供指定的 page。缓冲池管理器首先在 page_table_ 中查找 page_id 键是否存在：

• 如果存在就根据 page_id 对应的 frame_id 从缓冲池 pages_ 取出 page
• 如果不存在就通过 GetVictimFrameId() 函数选择被换出的帧，该函数首先从 free_list_ 中查找缓冲池的空位，如果没找到空位就得靠上一节实现的 ClockReplacer 选出被换出的冤大头

具体代码如下：

Page *BufferPoolManager::FetchPageImpl(page_id_t page_id) {
  // 1. Search the page table for the requested page (P).
  std::lock_guard<std::shared_mutex> lock(latch_);
  Page *page;

  // 1.1  If P exists, pin it and return it immediately.
  auto it = page_table_.find(page_id);
  if (it != page_table_.end()) {
    page = &pages_[it->second];
    if (page->pin_count_++ == 0) {
      replacer_->Pin(it->second);
    }
    return page;
  }

  // 1.2  If P does not exist, find a replacement page (R) from either the free list or the replacer.
  //      Note that pages are always found from the free list first.
  frame_id_t frame_id = GetVictimFrameId();
  if (frame_id == INVALID_PAGE_ID) {
    return nullptr;
  }

  // 2. If R is dirty, write it back to the disk.
  page = &pages_[frame_id];
  if (page->IsDirty()) {
    disk_manager_->WritePage(page->page_id_, page->data_);
  }

  // 3. Delete R from the page table and insert P.
  page_table_.erase(page->GetPageId());
  page_table_[page_id] = frame_id;

  // 4. Update P's metadata, read in the page content from disk, and then return a pointer to P.
  disk_manager_->ReadPage(page_id, page->data_);
  page->update(page_id, 1, false);
  replacer_->Pin(frame_id);
  return page;
}

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;
    }
  }

  return frame_id;
}

上述代码中还用了一个 Page::update 辅助函数，用于更新 page 的元数据：

/**
* update the meta data of page
* @param page_id the page id
* @param pin_count the pin count
* @param is_dirty is page dirty
* @param reset_memory whether to reset the memory of page
*/
void update(page_id_t page_id, int pin_count, bool is_dirty, bool reset_memory = false) {
  page_id_ = page_id;
  pin_count_ = pin_count;
  is_dirty_ = is_dirty;
  if (reset_memory) {
    ResetMemory();
  }
}

NewPageImpl(page_id)

该函数在缓冲池中插入一个新页，如果缓冲池中的所有页面都正在被线程访问，插入失败，否则靠 GetVictimFrameId() 计算插入位置：

Page *BufferPoolManager::NewPageImpl(page_id_t *page_id) {
  // 0. Make sure you call DiskManager::AllocatePage!
  std::lock_guard<std::shared_mutex> lock(latch_);

  // 1. If all the pages in the buffer pool are pinned, return nullptr.
  if (free_list_.empty() && replacer_->Size() == 0) {
    *page_id = INVALID_PAGE_ID;
    return nullptr;
  }

  // 2. Pick a victim page P from either the free list or the replacer. Always pick from the free list first.
  frame_id_t frame_id = GetVictimFrameId();
  if (frame_id == INVALID_PAGE_ID) {
    *page_id = INVALID_PAGE_ID;
    return nullptr;
  }

  // 3. Update P's metadata, zero out memory and add P to the page table.
  Page *page = &pages_[frame_id];
  if (page->IsDirty()) {
    disk_manager_->WritePage(page->page_id_, page->data_);
  }

  *page_id = disk_manager_->AllocatePage();
  page_table_.erase(page->GetPageId());
  page_table_[*page_id] = frame_id;
  // 需要把 dirty bit 设置为 false 才能通过 IsDirty 测试用例
  page->update(*page_id, 1, true, true);

  // 4. Set the page ID output parameter. Return a pointer to P.
  return page;
}

DeletePageImpl(page_id)

该函数从缓冲池和数据库文件中删除一个 page，并将其 page_id 设置为 INVALID_PAGE_ID：

bool BufferPoolManager::DeletePageImpl(page_id_t page_id) {
  // 0.   Make sure you call DiskManager::DeallocatePage!
  std::lock_guard<std::shared_mutex> lock(latch_);

  // 1. search the page table for the requested page (P).
  // If P does not exist, return true.
  auto it = page_table_.find(page_id);
  if (it == page_table_.end()) {
    return true;
  }

  // 2. If P exists, but has a non-zero pin-count, return false. Someone is using the page.
  Page &page = pages_[it->second];
  if (page.pin_count_ > 0) {
    return false;
  }

  // 3. Otherwise, P can be deleted. Remove P from the page table, reset its metadata and return it to the free list.
  disk_manager_->DeallocatePage(page_id);
  page_table_.erase(page_id);
  page.update(INVALID_PAGE_ID, 0, false, true);
  free_list_.push_back(it->second);
  return true;
}

UnpinPageImpl(page_id, is_dirty)

该函数用以减少对某个页的引用数 pin count，当 pin_count 为 0 时需要将其添加到 ClockReplacer 中：

bool BufferPoolManager::UnpinPageImpl(page_id_t page_id, bool is_dirty) {
  std::lock_guard<std::shared_mutex> lock(latch_);
  auto it = page_table_.find(page_id);
  if (it == page_table_.end()) {
    return false;
  }

  Page &page = pages_[it->second];
  if (page.pin_count_ <= 0) {
    return false;
  }

  // add page to replacer when the pin count is 0
  page.is_dirty_ |= is_dirty;
  if (--page.pin_count_ == 0) {
    replacer_->Unpin(it->second);
  }

  return true;
}

FlushPageImpl(page_id)

如果缓冲池的 page 被修改过，需要将其写入磁盘以保持同步：

bool BufferPoolManager::FlushPageImpl(page_id_t page_id) {
  // Make sure you call DiskManager::WritePage!
  std::shared_lock<std::shared_mutex> lock(latch_);
  auto it = page_table_.find(page_id);
  if (it == page_table_.end()) {
    return false;
  }

  // write page to disk if it's dirty
  Page &page = pages_[it->second];
  if (page.IsDirty()) {
    disk_manager_->WritePage(page_id, pages_[it->second].data_);
    page.is_dirty_ = false;
  }

  return true;
}

FlushAllPagesImpl()

该函数将缓冲池中的所有 page 写入磁盘：

void BufferPoolManager::FlushAllPagesImpl() {
  // You can do it!
  std::lock_guard<std::shared_mutex> lock(latch_);
  for (size_t i = 0; i < pool_size_; ++i) {
    Page &page = pages_[i];
    if (page.page_id_ != INVALID_PAGE_ID && page.IsDirty()) {
      disk_manager_->WritePage(i, page.data_);
      page.is_dirty_ = false;
    }
  }
}

测试

在终端输入指令：

cd build
make buffer_pool_manager_test
./test/buffer_pool_manager_test

测试结果如下：

总结

该实验考察了学生对并发和 STL 的掌握程度，由于注释中列出了实现步骤（最搞的是 You can do it! 注释），所以代码写起来也比较顺畅，以上~~