cmu 15445: proj1

2021-02-09

Specification

正式进入proj1, 需要实现使用lru替换策略的并发安全的bufferpool manager;

The first programming project is to implement a buffer pool in your storage manager.

Your implementation will need to be thread-safe.

task1 LRU replacement policy

Victim(T*) : Remove the object that was accessed the least recently compared to all the elements being tracked by the Replacer, store its contents in the output parameter and return True. If the Replacer is empty return False.
Pin(T) : This method should be called after a page is pinned to a frame in the BufferPoolManager. It should remove the frame containing the pinned page from the LRUReplacer.
Unpin(T) : This method should be called when the pin_count of a page becomes 0. This method should add the frame containing the unpinned page to the LRUReplacer.
Size() : This method returns the number of frames that are currently in the LRUReplacer.

阅读相关信息:

首先**LRUReplacer extends the abstract Replacer class**,

因此先阅读`Replacer.h`这个类:

这是一个抽象类, 提供接口
virtual bool Victim(frame_id_t *frame_id) = 0;
- 定义了buffer 满的时候该换出哪一页, 结果通过传入的参数frame_id指针返回, 其中类型frame_id_t是int32_t;
- 如果没有victim(比如所有page都碰巧被pin住了)就给frame_id赋为nullptr, 并且返回false;
- 如果找到了victim, 就让这个int32_t指针指向frame的id, 不是page的id!!!并且返回true;
virtual void Pin(frame_id_t frame_id) = 0;
- 当一个frame上的page(frame中的内容就是disk上的page内容的copy)正在被访问的时候, 我们应该把这个frame pin住, 也就是说我正在访问, 你别把我这页换出了, 不然就出错了;
virtual void Unpin(frame_id_t frame_id) = 0;
- 当这个frame没有在被使用了, 那么如果你buffer pool满了就可以把我这个frame当作换出的一个candidate;
virtual size_t Size() = 0;
- 返回replacer中可以被victim的candiate的数量, 所以一开始replacer的size应该是0, 随着使用并且出现unpin, 这时候这个frame的id就会被加入replacer, 作为可以被victim的candidate;

继续阅读LruReplacer发现他引入了/common/config.h头文件

阅读config.h

可见主要定义了一些类型:

Invalid的id都是-1;
一页的大小是4096B也就是4KB和sqlite一样;
BUFFER_POOL_SIZE为10, 应该是buffer pool有10个frame的空间;
log buffer没弄懂, 应该和这一个proj无关
bucket_size是extendible hashtable的桶的数目
然后定义了frame, page, txn, log seq number的类型都是int32_t;
slot_offset_t是指的tuple的起始位置;
然后最顶上三个是一些功能性的定义, 包括环检测周期, 是否开启log, log刷盘的周期;

阅读完毕, 开始实现

头文件定义一些member variables:

size_t lruCapacity;  // maxSize lruReplacer need to store
std::mutex latch;   // thread_safe 
std::unordered_map<frame_id_t, std::list<frame_id_t>::iterator> id2ListIter;  // mapping id to list iterator 
std::list<frame_id_t> dblList;  // double linked-list simulate timestamp

具体实现:


namespace bustub {

LRUReplacer::LRUReplacer(size_t num_pages) : lruCapacity(num_pages) {}

LRUReplacer::~LRUReplacer() = default;

bool LRUReplacer::Victim(frame_id_t *frame_id) { 
    const std::lock_guard<std::mutex> lock(latch);
    if (dblList.empty()) {
        // frame_id = nullptr;  // invalid op
        return false;
    }
    // erase head, assume head stores oldest timestamp
    *frame_id = dblList.front();
    id2ListIter.erase(dblList.front());  // delete entry from map
    dblList.erase(dblList.begin());      // delete node from list
    return true;
 }

void LRUReplacer::Pin(frame_id_t frame_id) {
    // when a page is pinned, remove it from lruReplacer
    const std::lock_guard<std::mutex> lock(latch);
    auto iter = id2ListIter.find(frame_id);
    if (iter != id2ListIter.end()) { // this frame do exsit in lrureplacer, remove it
        dblList.erase((*iter).second);
        id2ListIter.erase(frame_id);  
    }
}

void LRUReplacer::Unpin(frame_id_t frame_id) {
    // when a page is unpinned, add it to lrureplacer
    const std::lock_guard<std::mutex> lock(latch);
    if (id2ListIter.count(frame_id)) {
        return;
    }
    // we should add this frame to the tail of list since we victim the head
    // besides, add a new mapping relation to map
    while (dblList.size() >= lruCapacity) {
        id2ListIter.erase(dblList.front());
        dblList.erase(dblList.begin());
    }
    id2ListIter[frame_id] = dblList.insert(dblList.end(), frame_id);
}

size_t LRUReplacer::Size() { 
    const std::lock_guard<std::mutex> lock(latch);
    return dblList.size();
}

}  // namespace bustub

目前这个实现num_pages完全没作用, 不知道是不是出错了, 测试也无法运行测试. 先写第二个任务看看;

task2 BUFFER POOL MANAGER

The BufferPoolManager is responsible for fetching database pages from the DiskManager and storing them in memory. The BufferPoolManager can also write dirty pages out to disk when it is either explicitly instructed to do so or when it needs to evict a page to make space for a new page.

与磁盘打交道的代码已经帮我们写好了, 不需要管;

each Page object contains a block of memory that the DiskManager will use as a location to copy the contents of a physical page that it reads from disk. The BufferPoolManager will reuse the same Page object to store data as it moves back and forth to disk. This means that the same Page object may contain a different physical page throughout the life of the system. 再次强调: 内存中的page 和磁盘中的page不同

The Page object’s identifer (page_id) keeps track of what physical page it contains, 当一个page未没有包含磁盘page的时候, page_id应该设为invalid_page_id即-1;

每个内存中的page还维护了一个pin_count来记录有多少个线程正在使用这个page, 即pin住了; bufferpool manager显然不应该free掉一个pin_count > 0的page; 此外还维护了一个脏位, 当一个page被victim了, 如果dirty_flag, 那必须要把这一页的内容写入磁盘;

会用到task1实现的lru_replacer来决定evict哪个page;

For FetchPageImpl,you should return NULL if no page is available in the free list and all other pages are currently pinned.

FlushPageImpl should flush a page regardless of its pin status.

阅读相关代码

导入了page.h

阅读`/storage/page/page.h`:

friend class BufferPoolManager;

和porj1相关的大概就是这部分, 主要需要注意的是:

page_id: unique identifier, 当page中没有实际包含data的时候(即不是某个physical page的copy) 设为INVALID_PAGE_ID
pin_count表示有多个线程正在使用这一页, 当计数将为0的时候就可以加入lru_replacer中了
is_dirty: 表示这一页是否被修改过, 初步设想是在lru决定换出这一页的时候, 查询这个flag, 如果true, 那么应该写入磁盘;
data_: 一页大小为4096字节即4KB;

导入了disk_manager.h

阅读`storage/disk/disk_manager.h:`

DiskManager takes care of the allocation and deallocation of pages within a database. It performs the reading and writing of pages to and from disk, providing a logical file layer within the context of a database management system.
使用数据库文件的文件名来构造一个disk_manager: explicit DiskManager(const std::string &db_file);
与proj1相关的函数签名:

/**
 * Write a page to the database file.
 * @param page_id id of the page
 * @param page_data raw page data
 */
void WritePage(page_id_t page_id, const char *page_data);

/**
 * Read a page from the database file.
 * @param page_id id of the page
 * @param[out] page_data output buffer
 */
void ReadPage(page_id_t page_id, char *page_data);
/**
 * Allocate a page on disk.
 * @return the id of the allocated page
 */
page_id_t AllocatePage();

/**
 * Deallocate a page on disk.
 * @param page_id id of the page to deallocate
 */
void DeallocatePage(page_id_t page_id);

需要实现的函数:

FetchPageImpl(page_id): 从buffer pool取出page_id对应的页
NewPageImpl(page_id): 创建一个新的page
UnpinPageImpl(page_id, is_dirty): unpin一个page, 注意传入了脏flag
FlushPageImpl(page_id): 将page_id对应的页写入磁盘
DeletePageImpl(page_id): 将page_id这页从buffer pool删除
FlushAllPagesImpl(): 将buffer pool中所有页写入磁盘

这个task需要先厘清关系, 否则会错误频出, 以下是我写的过程中的发现:

page table是由unordered_map实现的, 负责mapping page_id_t到frame_id_t, 这里需要注意的一点是page_id是由disk_manager->AllocatePage()生成的, 官方提供的实现是简单的把一个atomic的id从0一直往上加, 所以每个page_id都唯一得代表了一个物理磁盘上的page, 内存中的page只是把磁盘上的page读入内存中的frame中的copy!!!
当需根据page_id修改一个在内存中的page, 我们需要通过page table把page_id map到frame_id, 因为在buffer pool manager中分配连续的Page *pages_ = new Page[max_pool_size]的空间, 所以我们需要的页的指针的就是pages + page_table[page_id];
上一个lru_replacer提供的pin 和 unpin接口其实很垃圾, 不符合他的名字, 应该是向lru_replacer中remove和insert的功能;

//===----------------------------------------------------------------------===//
//
//                         BusTub
//
// buffer_pool_manager.cpp
//
// Identification: src/buffer/buffer_pool_manager.cpp
//
// Copyright (c) 2015-2019, Carnegie Mellon University Database Group
//
//===----------------------------------------------------------------------===//

#include "buffer/buffer_pool_manager.h"
#include "common/logger.h"
#include <list>
#include <unordered_map>


namespace bustub {

BufferPoolManager::BufferPoolManager(size_t pool_size, DiskManager *disk_manager, LogManager *log_manager)
    : pool_size_(pool_size), disk_manager_(disk_manager), log_manager_(log_manager) {
  // We allocate a consecutive memory space for the buffer pool.
  pages_ = new Page[pool_size_];
  replacer_ = new LRUReplacer(pool_size);

  // Initially, every page is in the free list.
  for (size_t i = 0; i < pool_size_; ++i) {
    free_list_.emplace_back(static_cast<int>(i));
  }
  // page_table_.size: 0
}

BufferPoolManager::~BufferPoolManager() {
  delete[] pages_;
  delete replacer_;
}

Page *BufferPoolManager::FetchPageImpl(page_id_t page_id) {
  // 1.     Search the page table for the requested page (P).
  // 1.1    If P exists, pin it and return it immediately.
  // 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.
  // 2.     If R is dirty, write it back to the disk.
  // 3.     Delete R from the page table and insert P.
  // 4.     Update P's metadata, read in the page content from disk, and then return a pointer to P.
  
  std::lock_guard<std::mutex> lock(latch_);
  // LOG_INFO("successfully acquired the latch");
  Page *res = nullptr;
  // 1 
  if (page_table_.count(page_id)) {  // 1.1
    res = pages_ + page_table_[page_id];
    if (res->pin_count_ == 0) {
      replacer_->Pin(page_table_[page_id]);
    }
    ++(res->pin_count_);
    return res;
  }
  // 1.2 
  if (!free_list_.empty() || replacer_->Size()) {
    frame_id_t frame_id = -1;
    res = GetVictim(&frame_id);
    if (!res) return nullptr;
    // 2
    if (res->is_dirty_) {
      disk_manager_->WritePage(res->page_id_, res->GetData());
      res->is_dirty_ = false;
    }
    // 3
    page_table_.erase(res->page_id_);
    page_table_[page_id] = frame_id;
    // 4
    res->pin_count_ = 1;
    res->page_id_ = page_id;
    res->ResetMemory();
    disk_manager_->ReadPage(page_id, res->GetData());
    return res; 
  }
  return nullptr;

  // // 1.2 since pages are always found from free list, 
  // //     if free list empty, deal with replacer first to avoid redundent 
  // if (free_list_.empty()) {  // victim a page
  //   auto victim_page_id = INVALID_PAGE_ID;
  //   if (replacer_->Victim(&victim_page_id)) {
  //     free_list_.insert(free_list_.end(), page_table_[victim_page_id]);
  //   }
  // }
  // if (!free_list_.empty()) {
  //   // frame is fixed at the moment we get it from list
  //   res = pages_ + free_list_.back();
  //   free_list_.pop_back();
  //   // 2
  //   if (res->is_dirty_) {  // write it back to disk
  //     FlushPageImpl(res->page_id_);
  //   }
  //   // 3 deal with page table
  //   res->ResetMemory();
  //   auto frame_id = page_table_[res->page_id_];
  //   page_table_.erase(res->page_id_);
  //   page_table_[page_id] = frame_id;
  //   // 4
  //   res->page_id_ = page_id;
  //   res->is_dirty_ = false;
  //   res->pin_count_ = 1;
  //   disk_manager_->ReadPage(page_id, res->GetData());
  //   return res;
  // }
  
}


Page *BufferPoolManager::GetVictim(frame_id_t *victimFrameId) {
  // helper function: Get a victim page's ptr;
  // this function should be called only when acquiring the latch
  // in this fucntion, I will not try to acquire the latch
  // 1  free_list is not empty, remove free_list's tail 
  //    and return a ptr to the page with corresponding page_frame_id
  // 2  if free_list is empty, get victim page's page_frame_id from lru_replacer
  if (!free_list_.empty()) {
    *victimFrameId = free_list_.back();
    free_list_.pop_back();
    return pages_ + *victimFrameId;
  }
  else {
    if (replacer_->Victim(victimFrameId)) {
      return pages_ + *victimFrameId;
    }
  }
  return nullptr;
}



bool BufferPoolManager::UnpinPageImpl(page_id_t page_id, bool is_dirty) { 
  std::lock_guard<std::mutex> lock(latch_);
  Page *page = pages_ + page_table_[page_id];

  // LOG_INFO("page_id the same ? %s", page_id == page->page_id_ ? "true":"false");
  // LOG_INFO("page: %d's pin_count: %d", page->page_id_, page->pin_count_);
  if (page_id == INVALID_PAGE_ID || page->pin_count_ <= 0) {
    return false;
  } 
  // if (is_dirty) {
  //   disk_manager_->WritePage(page_id, page->GetData());
  // }
  if (is_dirty) {
    page->is_dirty_ = is_dirty;
  }
  if (--(page->pin_count_) == 0) {
    replacer_->Unpin(page_table_[page_id]);
  }
  return true;
}

bool BufferPoolManager::FlushPageImpl(page_id_t page_id) {
  // Make sure you call DiskManager::WritePage!
  std::lock_guard<std::mutex> lock(latch_);
  if (page_id == INVALID_PAGE_ID || !page_table_.count(page_id)) {
    return false;
  }
  Page *page = pages_ + page_table_[page_id];
  if (page->is_dirty_) {
    disk_manager_->WritePage(page_id, page->GetData());
    page->is_dirty_ = false;
  }
  return true;
}

Page *BufferPoolManager::NewPageImpl(page_id_t *page_id) {
  // 0.   Make sure you call DiskManager::AllocatePage!
  // 1.   If all the pages in the buffer pool are pinned, return nullptr.
  // 2.   Pick a victim page P from either the free list or the replacer. Always pick from the free list first.
  // 3.   Update P's metadata, zero out memory and add P to the page table.
  // 4.   Set the page ID output parameter. Return a pointer to P.
  std::lock_guard<std::mutex> lock(latch_);
  *page_id = disk_manager_->AllocatePage();
  // 1
  if (free_list_.empty() && replacer_->Size() == 0) {
    return nullptr;
  }
  // 2 
  // LOG_INFO("enter NewPage, before call FetchPage");
  frame_id_t frame_id = -1;
  Page *page = GetVictim(&frame_id);
  if (page->is_dirty_) {
    disk_manager_->WritePage(page->page_id_, page->GetData());
    page->is_dirty_ = false;
  }
  page_table_.erase(page->page_id_);
  // 3
  page->page_id_ = *page_id;
  page->pin_count_ = 1;
  page_table_[*page_id] = frame_id;
  page->ResetMemory();
  return page;
}

bool BufferPoolManager::DeletePageImpl(page_id_t page_id) {
  // 0.   Make sure you call DiskManager::DeallocatePage!
  // 1.   Search the page table for the requested page (P).
  // 1.   If P does not exist, return true.
  // 2.   If P exists, but has a non-zero pin-count, return false. Someone is using the page.
  // 3.   Otherwise, P can be deleted. Remove P from the page table, reset its metadata and return it to the free list.
  std::lock_guard<std::mutex> lock(latch_);
  // 0
  disk_manager_->DeallocatePage(page_id);
  // 1
  if (page_table_.count(page_id)) { 
    Page *page = pages_ + page_table_[page_id];
    if (page->pin_count_) {
      return false;
    }
    free_list_.push_back(page_table_[page_id]);
    page_table_.erase(page_id);
    replacer_->Pin(page_table_[page_id]);
    page->is_dirty_ = false;
    page->pin_count_ = 0;
    page->ResetMemory();
    page->page_id_ = INVALID_PAGE_ID;
  }
  return true;
}

void BufferPoolManager::FlushAllPagesImpl() {
  // You can do it!

  for (size_t i = 0; i < pool_size_; ++i) {
    auto *page = pages_ + i;
    if (page->is_dirty_) {
      FlushPageImpl(page->page_id_);
    }
  }
}

}  // namespace bustub

在实现过程中, 遇到的第一个大问题就是死锁:

本来这些需要实现的接口是可以互相调用的, 但是这就存在一个问题:
这些接口函数都需要获取latch来实现线程安全;
一个函数比如newPage获取了latch后调用flushPage;
flushPage中第一行代码也是获取latch, 显然无法获得, 就会死锁

解决办法:

一开始网上乱查了个adopt_lock, 结果发现这个只能让程序跑起来, 错误百出;
可重用锁不考虑, 被陈硕疯狂diss;
新增了一个辅助函数, GetVictim(), 意思就是我需要一个page的空间的时候, 你必须要给我搞这样一块空间来, 不管是从free_list还是lru_replacer;
当需要写文件到磁盘的时候不调用flushPage, 直接disk_manager操作;
辅助函数中不用获取latch: 因为调用辅助函数的函数caller一定已经在第一行代码拿到latch了, 而整个class只有一个latch, 所以不会有别的函数能够调用这个辅助函数, 所以线程安全;

以上代码本地测试通过, 官方网站测试还没学会怎么用, build失败.

官方网站满分, 可以放心写下一个部分了(修改了一个bug, 就是当我要拿的页存在页表中, 这页也可能在lru中!!!所以fetchPage如果pin_count == 0, 一定要调用replacer->unpin);

;