Skip to content

Feature:Add buff pool manager #1

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
xifyang wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Feature:Add buff pool manager #1

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
164 changes: 159 additions & 5 deletions src/buffer/buffer_pool_manager_instance.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@

#include "buffer/buffer_pool_manager_instance.h"

#include "common/logger.h"
#include "common/macros.h"

namespace bustub {
Expand Down Expand Up @@ -48,11 +49,27 @@ BufferPoolManagerInstance::~BufferPoolManagerInstance() {

bool BufferPoolManagerInstance::FlushPgImp(page_id_t page_id) {
// Make sure you call DiskManager::WritePage!
return false;
std::unique_lock<std::mutex> lock(latch_);
if (page_table_.find(page_id) == page_table_.end()) {
return false;
}
frame_id_t flush_fid = page_table_[page_id];
Page *flush_page = &pages_[flush_fid];
disk_manager_->WritePage(flush_page->GetPageId(), flush_page->GetData());
flush_page->is_dirty_ = false;
LOG_INFO("Succeed to flush Page: page_id:%d, frame_id:%d", flush_page->GetPageId(), flush_fid);
return true;
}

void BufferPoolManagerInstance::FlushAllPgsImp() {
// You can do it!
std::unique_lock<std::mutex> lock(latch_);
for (auto &entry : page_table_) {
Page *flush_page = &pages_[entry.second];
disk_manager_->WritePage(flush_page->GetPageId(), flush_page->GetData());
flush_page->is_dirty_ = false;
LOG_INFO("Succeed to flush Page:%d, frame_id:%d", flush_page->GetPageId(), entry.first);
}
}

Page *BufferPoolManagerInstance::NewPgImp(page_id_t *page_id) {
Expand All @@ -61,7 +78,55 @@ Page *BufferPoolManagerInstance::NewPgImp(page_id_t *page_id) {
// 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.
return nullptr;
page_id_t new_pid = AllocatePage();
frame_id_t victm_fid = -1;
std::unique_lock<std::mutex> lock(latch_);
LOG_INFO("Starting to allocate frame_id for page_id:%d", new_pid);
do {
// Pick a victim page P from either the free list
if (!free_list_.empty()) {
victm_fid = free_list_.front();
free_list_.pop_front();
LOG_INFO("Succeed to get frame_id:%d from free list for page_id:%d", victm_fid, new_pid);
break;
}
// If all the pages in the buffer pool are pinned, return nullptr.
bool is_all_spinned = true;
for (int i = 0; i < static_cast<int>(pool_size_); i++) {
if (pages_[i].GetPinCount() == 0) {
is_all_spinned = false;
}
}
if (is_all_spinned) {
LOG_INFO("Failed to get frame for page_id:%d ,All page have been spinned", new_pid);
break;
}
// Pick a victim page P from the replacer, erase P from the page table, write page if dirty
if (!replacer_->Victim(&victm_fid)) {
LOG_INFO("Failed to victim frame for page_id:%d", new_pid);
break;
}
Page *replace_page = &pages_[victm_fid];
if (replace_page->IsDirty()) {
disk_manager_->WritePage(replace_page->GetPageId(), replace_page->GetData());
}
page_table_.erase(replace_page->page_id_);
LOG_INFO("Succeed to victim frame_id:%d for page_id:%d", victm_fid, new_pid);
} while (false);
if (victm_fid == -1) {
return nullptr;
}
// Add P to the page table, update P's metadata, zero out memory
Page *victm_page = &pages_[victm_fid];
page_table_[new_pid] = victm_fid;
victm_page->page_id_ = new_pid;
victm_page->is_dirty_ = false;
victm_page->pin_count_++;
victm_page->ResetMemory();
replacer_->Pin(victm_fid);
*page_id = new_pid;
LOG_INFO("Succed to allocate frame_id for page_id, page_id:%d, frame_id:%d", victm_page->page_id_, victm_fid);
return victm_page;
}

Page *BufferPoolManagerInstance::FetchPgImp(page_id_t page_id) {
Expand All @@ -72,7 +137,58 @@ Page *BufferPoolManagerInstance::FetchPgImp(page_id_t page_id) {
// 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.
return nullptr;
std::unique_lock<std::mutex> lock(latch_);
Page *fetch_page = nullptr;
frame_id_t fetch_fid = -1;
LOG_INFO("Starting to fetch page_id:%d", page_id);
do {
if (page_table_.find(page_id) != page_table_.end()) {
fetch_fid = page_table_[page_id];
fetch_page = &pages_[fetch_fid];
fetch_page->pin_count_++;
replacer_->Pin(fetch_fid);
LOG_INFO("page_id:%d exits in page table, succeed to fetch ", page_id);
return fetch_page;
}
LOG_INFO("page_id:%d don't exit in page table, start to allocate page", page_id);
if (!free_list_.empty()) {
fetch_fid = free_list_.front();
free_list_.pop_front();
LOG_INFO("Suceed allocate page for page_id:%d in free list", page_id);
break;
}
bool is_all_spinned = true;
for (int i = 0; i < static_cast<int>(pool_size_); i++) {
if (pages_[i].GetPinCount() == 0) {
is_all_spinned = false;
}
}
if (is_all_spinned) {
LOG_INFO("Failed to allocate frame for page_id:%d ,All page have been spinned", page_id);
break;
}
if (!replacer_->Victim(&fetch_fid)) {
LOG_INFO("Failed to victim frame for page_id:%d", page_id);
break;
}
Page *replace_page = &pages_[fetch_fid];
if (replace_page->IsDirty()) {
disk_manager_->WritePage(replace_page->GetPageId(), replace_page->GetData());
}
page_table_.erase(replace_page->page_id_);
} while (false);
if (fetch_fid == -1) {
return nullptr;
}
fetch_page = &pages_[fetch_fid];
fetch_page->page_id_ = page_id;
fetch_page->is_dirty_ = false;
disk_manager_->ReadPage(page_id, fetch_page->GetData());
page_table_[page_id] = fetch_fid;
fetch_page->pin_count_++;
replacer_->Pin(fetch_fid);
LOG_INFO("Succeed to fetch page_id:%d", page_id);
return fetch_page;
}

bool BufferPoolManagerInstance::DeletePgImp(page_id_t page_id) {
Expand All @@ -81,10 +197,48 @@ bool BufferPoolManagerInstance::DeletePgImp(page_id_t page_id) {
// 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.
return false;
std::unique_lock<std::mutex> lock(latch_);
LOG_INFO("Starting to delete page_id:%d", page_id);
if (page_table_.find(page_id) == page_table_.end()) {
LOG_INFO("Page_id:%d don't exit", page_id);
return true;
}
frame_id_t delete_fid = page_table_[page_id];
Page *delete_page = &pages_[delete_fid];
if (delete_page->GetPinCount() != 0) {
LOG_INFO("Failed to delete page_id:%d, page has pin_count:%d", page_id, delete_page->GetPinCount());
return false;
}
if (delete_page->IsDirty()) {
LOG_INFO("Page_id:%d is dirty,need to write to disk", page_id);
disk_manager_->WritePage(delete_page->GetPageId(), delete_page->GetData());
}
delete_page->page_id_ = INVALID_PAGE_ID;
delete_page->is_dirty_ = false;
delete_page->ResetMemory();
free_list_.push_back(delete_fid);
page_table_.erase(page_id);
LOG_INFO("Succeed to delete page_id:%d", page_id);
return true;
}

bool BufferPoolManagerInstance::UnpinPgImp(page_id_t page_id, bool is_dirty) { return false; }
bool BufferPoolManagerInstance::UnpinPgImp(page_id_t page_id, bool is_dirty) {
std::unique_lock<std::mutex> lock(latch_);
if (page_table_.find(page_id) == page_table_.end()) {
LOG_INFO("Page_id:%d don't exit", page_id);
return false;
}
frame_id_t unpin_fid = page_table_[page_id];
Page *unpin_page = &pages_[unpin_fid];
unpin_page->is_dirty_ = is_dirty;
if (unpin_page->GetPinCount() <= 0) {
return false;
}
if (--unpin_page->pin_count_ == 0) {
replacer_->Unpin(unpin_fid);
}
return true;
}

page_id_t BufferPoolManagerInstance::AllocatePage() {
const page_id_t next_page_id = next_page_id_;
Expand Down
44 changes: 33 additions & 11 deletions src/buffer/lru_replacer.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,16 +14,38 @@

namespace bustub {

LRUReplacer::LRUReplacer(size_t num_pages) {}

LRUReplacer::~LRUReplacer() = default;

bool LRUReplacer::Victim(frame_id_t *frame_id) { return false; }

void LRUReplacer::Pin(frame_id_t frame_id) {}

void LRUReplacer::Unpin(frame_id_t frame_id) {}

size_t LRUReplacer::Size() { return 0; }
bool LRUReplacer::Victim(frame_id_t *frame_id) {
std::unique_lock<std::mutex> lock(mutex_);
if (lru_list_.empty() || lru_map_.empty()) {
return false;
}
frame_id_t victim_fid = lru_list_.back();
lru_list_.pop_back();
lru_map_.erase(victim_fid);
*frame_id = victim_fid;
return true;
}

void LRUReplacer::Pin(frame_id_t frame_id) {
std::unique_lock<std::mutex> lock(mutex_);
if (lru_map_.find(frame_id) == lru_map_.end()) {
// frame_id don't exit
return;
}
auto pin_iterator = lru_map_[frame_id];
lru_list_.splice(lru_list_.begin(), lru_list_, pin_iterator);
lru_list_.pop_front();
lru_map_.erase(frame_id);
}

void LRUReplacer::Unpin(frame_id_t frame_id) {
std::unique_lock<std::mutex> lock(mutex_);
if (lru_map_.find(frame_id) != lru_map_.end() || lru_map_.size() == capacity_) {
// frame_id exit or list is full
return;
}
lru_list_.push_front(frame_id);
lru_map_[frame_id] = lru_list_.begin();
}

} // namespace bustub
16 changes: 13 additions & 3 deletions src/include/buffer/lru_replacer.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@

#include <list>
#include <mutex> // NOLINT
#include <unordered_map>
#include <vector>

#include "buffer/replacer.h"
Expand All @@ -30,23 +31,32 @@ class LRUReplacer : public Replacer {
* Create a new LRUReplacer.
* @param num_pages the maximum number of pages the LRUReplacer will be required to store
*/
explicit LRUReplacer(size_t num_pages);
explicit LRUReplacer(size_t num_pages) {
capacity_ = num_pages;
}

/**
* Destroys the LRUReplacer.
*/
~LRUReplacer() override;
~LRUReplacer() override = default;

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;
size_t Size() override {
std::unique_lock<std::mutex> lock(mutex_);
return lru_map_.size();
}

private:
// TODO(student): implement me!
std::mutex mutex_;
size_t capacity_;
std::list<frame_id_t> lru_list_;
std::unordered_map<frame_id_t, std::list<frame_id_t>::iterator> lru_map_;
};

} // namespace bustub