CMU14-445 Lab

  size_t capacity;      //容量
  size_t size;          //当前大小
  std::list<frame_id_t> lists;           //存储帧的栈,用list实现
  std::unordered_map<frame_id_t , int> maps;      //哈希映射,映射帧和是否在LRU Replacer里面.
  std::mutex locks;

LRUReplacer::LRUReplacer(size_t num_pages) {
    capacity = num_pages;
    maps.clear();
    lists.clear();
    size = lists.size();
}

bool LRUReplacer::Victim(frame_id_t *frame_id) { 
    if(lists.empty()) return false; 
    locks.lock();
    frame_id_t lasts = lists.back();
    maps.erase(lasts);
    lists.pop_back();
    *frame_id = lasts; 
    locks.unlock();
    return true;
}

void LRUReplacer::Pin(frame_id_t frame_id) {
    if(maps.count(frame_id) == 0) return;
    locks.lock();
    lists.remove(frame_id);
    maps.erase(frame_id);
    locks.unlock();
}

void LRUReplacer::Unpin(frame_id_t frame_id) {
    if(maps.count(frame_id) != 0) return;
    locks.lock();
    //如果已经满了,则需要淘汰一个然后放入 LRUReplacer 中
    if (lists.size() == capacity) {
        frame_id_t last_frame = lists.back();
        maps.erase(last_frame);
        lists.pop_back();
    }
    lists.push_front(frame_id);
    maps[frame_id] = 1;
    locks.unlock();
}

  /** Array of buffer pool pages. */
  Page *pages_;
  /** Pointer to the disk manager. */
  DiskManager *disk_manager_ __attribute__((__unused__));
  /** Pointer to the log manager. */
  LogManager *log_manager_ __attribute__((__unused__));
  /** Page table for keeping track of buffer pool pages. */
  std::unordered_map<page_id_t, frame_id_t> page_table_;
  /** Replacer to find unpinned pages for replacement. */
  Replacer *replacer_;
  /** List of free pages. */
  std::list<frame_id_t> free_list_;
  /** This latch protects shared data structures. We recommend updating this comment to describe what it protects. */
  std::mutex latch_;

Page *BufferPoolManagerInstance::FetchPgImp(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.
  frame_id_t fid;
  Page* p = nullptr;
  // step 1.1.
  if(page_table_.find(page_id) != page_table_.end()){
    fid = page_table_[page_id];
    pages_[fid].pin_count_++;
    // pin it.
    replacer_->Pin(fid);
    return &pages_[fid];
  }
  latch_.lock();
  // step 1.2.
  if (free_list_.empty()) {  // pick from free list first.
    //can't find a page that can be flushed.
    if (!replacer_->Victim(&fid)) {
      return nullptr;
    } else {
      // the page placed in the lrureplacer need to be flushed.
      // send it to the disk
      p = &pages_[fid];
      if (p->is_dirty_) {
        disk_manager_->WritePage(p->page_id_, p->data_);
        p->is_dirty_ = false;
      }
      // erase the pagetable_.
      page_table_.erase(p->page_id_);
    }
  } else {
    //found:
    fid = free_list_.front();
    free_list_.pop_front();
    p = &pages_[fid];
  }
  // Step 3.
  page_table_[page_id] = fid;
  replacer_->Pin(fid);
  // Step 4.
  p->page_id_ = page_id;
  p->pin_count_ = 1;
  disk_manager_->ReadPage(page_id, p->data_);
  latch_.unlock();
  return p;
}

bool BufferPoolManagerInstance::DeletePgImp(page_id_t page_id) {
  // 0.   Make sure you call 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.
  
  if (page_table_.find(page_id) == page_table_.end()) {
    return true;
  }
  // P does exists.
  frame_id_t fid = page_table_[page_id];
  Page* deletepage = &pages_[fid];
  // P is pinned
  if (deletepage->pin_count_ != 0) {
      return false;
  }
  // if dirty rewrite:
  // flush the page before deallocate it.
  latch_.lock();
  if (deletepage->is_dirty_) {
    disk_manager_->WritePage(page_id, deletepage->data_);
    deletepage->is_dirty_ = false;
  }
  DeallocatePage(page_id);
  // remove P from the page table.
  page_table_.erase(page_id);
  // reset its metadata.
  // the page returned to freelist does not stores any page.
  deletepage->page_id_ = INVALID_PAGE_ID;
  deletepage->is_dirty_ = false;
  deletepage->pin_count_ = 0;
  // return it to free_list_.
  free_list_.push_back(fid);
  latch_.unlock();
  return true;
}

bool BufferPoolManagerInstance::FlushPgImp(page_id_t page_id) {
  // Make sure you call DiskManager::WritePage!
  std::lock_guard<std::mutex> lock(latch_);
  if (page_table_.find(page_id) == page_table_.end() || page_id == INVALID_PAGE_ID)
    return false;
  disk_manager_->WritePage(page_id, pages_[page_table_[page_id]].data_);
  return true;
}

bool BufferPoolManagerInstance::UnpinPgImp(page_id_t page_id, bool is_dirty) {
  std::lock_guard<std::mutex> lock(latch_); 
  frame_id_t fid = page_table_[page_id];
  Page* p = &pages_[fid];
  p->is_dirty_ = is_dirty;  // hold the state until victim.
  if (p->pin_count_ <= 0)
    return false;
  --p->pin_count_;
  if (p->pin_count_ == 0) {
    replacer_->Unpin(fid);
    return true;
  }
  return false;
}

Page *ParallelBufferPoolManager::NewPgImp(page_id_t *page_id) {
  // create new page. We will request page allocation in a round robin manner from the underlying
  // BufferPoolManagerInstances
  // 1.   From a starting index of the BPMIs, call NewPageImpl until either 1) success and return 2) looped around to starting index and return nullptr    
  // 2.   Bump the starting index (mod number of instances) to start search at a different BPMI each time this function is called
  for(size_t i = 0; i < num_instances_; i++){
    BufferPoolManager* one_manager_instance = manager_[next_instacnce_];
    Page *page = one_manager_instance->NewPage(page_id);
    if(page != nullptr){
      return page;
    }
    next_instacnce_ = (next_instacnce_ + 1) % num_instances_;
  }
  return nullptr;
}

BufferPoolManager *ParallelBufferPoolManager::GetBufferPoolManager(page_id_t page_id) {
  // Get BufferPoolManager responsible for handling given page id. You can use this method in your other methods.
  return manager_[page_id % num_instances_];
}

Page *ParallelBufferPoolManager::FetchPgImp(page_id_t page_id) {
  // Fetch page for page_id from responsible BufferPoolManagerInstance
  return GetBufferPoolManager(page_id)->FetchPage(page_id);
}

bool ParallelBufferPoolManager::UnpinPgImp(page_id_t page_id, bool is_dirty) {
  // Unpin page_id from responsible BufferPoolManagerInstance
  return GetBufferPoolManager(page_id)->UnpinPage(page_id,is_dirty);
}

bool ParallelBufferPoolManager::FlushPgImp(page_id_t page_id) {
  // Flush page_id from responsible BufferPoolManagerInstance
  return GetBufferPoolManager(page_id)->FlushPage(page_id);
}

  page_id_t page_id_; // 页的id
  lsn_t lsn_;
  uint32_t global_depth_{0};//全局深度,也就是说下面的数组有几个元素.2^i的关系,所有局部深度不能比全局深度大
  uint8_t local_depths_[DIRECTORY_ARRAY_SIZE];//每一个页的局部深度,这个类似三级页表.第一级的页表局部深度就是2.局部深度代表你作为目录项离叶结点(bucket有多远)
  page_id_t bucket_page_ids_[DIRECTORY_ARRAY_SIZE];//下一级的页号

uint32_t HashTableDirectoryPage::GetGlobalDepthMask() { 
  uint32_t global_depth_mask =  ((1 << GetGlobalDepth()) - 1);
  return global_depth_mask; 
}

void HashTableDirectoryPage::IncrGlobalDepth() { 
  int original_depth = 1 << global_depth_;
  int new_index = original_depth;
  for(int i = 0;i < original_depth;i++){
    local_depths_[new_index] = local_depths_[i];
    bucket_page_ids_[new_index] = bucket_page_ids_[i];
    new_index++;
  }
  global_depth_++;
}

bool HashTableDirectoryPage::CanShrink() {
  int original_depth = 1 << global_depth_;
  for(int i = 0;i < original_depth;i++){
    if(local_depths_[i] >= global_depth_) return false;
  }
  return true;
}

  //  For more on BUCKET_ARRAY_SIZE see storage/page/hash_table_page_defs.h
  char occupied_[(BUCKET_ARRAY_SIZE - 1) / 8 + 1];
  // 0 if tombstone/brand new (never occupied), 1 otherwise.
  char readable_[(BUCKET_ARRAY_SIZE - 1) / 8 + 1];
  MappingType array_[0];

template <typename KeyType, typename ValueType, typename KeyComparator>
bool HASH_TABLE_BUCKET_TYPE::IsOccupied(uint32_t bucket_idx) const {
  size_t c = occupied_[bucket_idx / 8];
  c = c & (1 << (bucket_idx % 8));
  return c != 0;
}

template <typename KeyType, typename ValueType, typename KeyComparator>
void HASH_TABLE_BUCKET_TYPE::SetOccupied(uint32_t bucket_idx) {
  size_t c = occupied_[bucket_idx / 8];
  c = c | (1 << (bucket_idx % 8));
  occupied_[bucket_idx / 8] = static_cast<char>(c);
}

template <typename KeyType, typename ValueType, typename KeyComparator>
bool HASH_TABLE_BUCKET_TYPE::IsReadable(uint32_t bucket_idx) const {
  size_t c = readable_[bucket_idx / 8];
  c = c & (1 << (bucket_idx % 8));
  return c != 0;
}

template <typename KeyType, typename ValueType, typename KeyComparator>
void HASH_TABLE_BUCKET_TYPE::SetReadable(uint32_t bucket_idx) {
  size_t c = readable_[bucket_idx / 8];
  c = c | (1 << (bucket_idx % 8));
  readable_[bucket_idx / 8] = static_cast<char>(c);
}

template <typename KeyType, typename ValueType, typename KeyComparator>
bool HASH_TABLE_BUCKET_TYPE::Remove(KeyType key, ValueType value, KeyComparator cmp) {
  for(size_t i = 0; i < BUCKET_ARRAY_SIZE ; i++){
    if(IsReadable(i)){
      if(cmp(key,array_[i].first) == 0 && value == array_[i].second){
        RemoveAt(i);
        return true;
      }
    }
  }
  return false;
}

template <typename KeyType, typename ValueType, typename KeyComparator>
void HASH_TABLE_BUCKET_TYPE::RemoveAt(uint32_t bucket_idx) {
  size_t c = readable_[bucket_idx / 8];
  c = c & (~(1 << (bucket_idx % 8)));
  readable_[bucket_idx / 8] = static_cast<char>(c);
}

template <typename KeyType, typename ValueType, typename KeyComparator>
bool HASH_TABLE_BUCKET_TYPE::IsEmpty() {
  u_int8_t mask = 0;
  size_t times = BUCKET_ARRAY_SIZE / 8;
  for (size_t i = 0; i < times; i++) {
    char c = readable_[i];
    uint8_t ic = static_cast<uint8_t>(c);
    if ((ic | mask) != mask) {
      return false;
    }
  }
//剩下的
  size_t remain = BUCKET_ARRAY_SIZE % 8;
  if (remain > 0) {
    char c = readable_[times];
    uint8_t ic = static_cast<uint8_t>(c);
    for (size_t i = 0; i < remain; i++) {
//每次取1位
      if ((ic & 1) != 1) {
        return false;
      }
//迭代
      ic = ic >> 1;
    }
  }
  return true;
}

split--伪代码
 
	bucket_page = Fetch(bucket_idx);
    if(bucket为空){
        构造分裂映像
        split_page = NewPage(&split_idx);
        if(全局深度 == 局部深度){
            需要增长目录;
            产生的许多空索引需要进行填入page_id信息以及局部深度信息
            指向相同的page_id的索引之间相差了1<<(全局深度);
            修改目录的信息以及设置分裂后的桶的相关信息            
        }else{
            仅需要分裂桶;
            先增加局部深度，接着确定兄弟索引的个数;
            修改分裂后兄弟索引的指向;
        }
        分配旧桶里面的数据，一般会均分到两个桶里面
    }
	执行插入即可

template <typename KeyType, typename ValueType, typename KeyComparator>
bool HASH_TABLE_TYPE::GetValue(Transaction *transaction, const KeyType &key, std::vector<ValueType> *result) {
  table_latch_.RLock();
  HashTableDirectoryPage *dir_page = FetchDirectoryPage();
  page_id_t bucket_page_id = KeyToPageId(key,dir_page);
  Page* page = FetchBucketPage(bucket_page_id);
  page->RLatch();

  HASH_TABLE_BUCKET_TYPE *bucket_page = reinterpret_cast<HASH_TABLE_BUCKET_TYPE*>(page->GetData());
  bool res = bucket_page->GetValue(key,comparator_,result);

  page->RUnlatch();

  buffer_pool_manager_->UnpinPage(directory_page_id_,false);
  buffer_pool_manager_->UnpinPage(bucket_page_id,false);

  table_latch_.RUnlock();

  return res;
}

template <typename KeyType, typename ValueType, typename KeyComparator>
bool HASH_TABLE_TYPE::Insert(Transaction *transaction, const KeyType &key, const ValueType &value) {
  table_latch_.WLock();
  HashTableDirectoryPage* dir_page = FetchDirectoryPage();
  page_id_t bucket_page_id = KeyToPageId(key,dir_page);
  Page* page = FetchBucketPage(bucket_page_id);

  page->WLatch();

  HASH_TABLE_BUCKET_TYPE* bucket_page = reinterpret_cast<HASH_TABLE_BUCKET_TYPE*>(page->GetData());

  bool flag = bucket_page->IsFull();
  if(!flag){
    bool ret = bucket_page->Insert(key,value,comparator_);
    page->WUnlatch();

    buffer_pool_manager_->UnpinPage(bucket_page_id,true);
    buffer_pool_manager_->UnpinPage(dir_page->GetPageId(),false);

    table_latch_.WUnlock();
    return ret;
  }
  page->WUnlatch();
  buffer_pool_manager_->UnpinPage(bucket_page_id,false);
  buffer_pool_manager_->UnpinPage(dir_page->GetPageId(),false);

  table_latch_.WUnlock();
  return SplitInsert(transaction,key,value);
}

template <typename KeyType, typename ValueType, typename KeyComparator>
bool HASH_TABLE_TYPE::Remove(Transaction *transaction, const KeyType &key, const ValueType &value) {
  table_latch_.RLock();

  HashTableDirectoryPage* dir_page = FetchDirectoryPage();
  uint32_t bucket_page_id = KeyToPageId(key,dir_page);
  Page* page = FetchBucketPage(bucket_page_id);
  page->WLatch();
  HASH_TABLE_BUCKET_TYPE* bucket_page = reinterpret_cast<HASH_TABLE_BUCKET_TYPE*>(page->GetData())  ;

  bool ret = bucket_page->Remove(key,value,comparator_);

  page->WUnlatch();
  buffer_pool_manager_->UnpinPage(dir_page->GetPageId(),false);
  buffer_pool_manager_->UnpinPage(bucket_page_id,true); 
  table_latch_.RUnlock();

  if(bucket_page->IsEmpty()){
    Merge(transaction,key,value);
    return ret;
  }
  return ret;
}