avoid bug

bpt/include/bpt/disk_manager.h

@@ -17,7 +17,7 @@ class DiskManager {
    * for first_empty_page_id).
    */
  public:
-  explicit DiskManager(const std::string &file_path_);
+  explicit DiskManager(const std::string &file_path_, bool renew = false);
   ~DiskManager();
   char *RawDataMemory();
   size_t RawDatMemorySize();

bpt/src/disk_manager.cpp

@@ -3,13 +3,14 @@
 #include <exception>
 #include <stdexcept>
 const size_t kPageSize = 4096;
-DiskManager::DiskManager(const std::string &file_path_)
+DiskManager::DiskManager(const std::string &file_path_, bool renew)
     : file_path(file_path_),
       first_empty_page_id(0),
       current_total_page_count(0),
       current_none_empty_page_count(0),
       raw_data_memory(nullptr),
       fp(nullptr) {
+  if (renew) remove(file_path.c_str());
   fp = fopen(file_path.c_str(), "r+b");
   if (fp == nullptr) {
     // File doesn't exist, create a new one
@@ -46,7 +47,7 @@ char *DiskManager::RawDataMemory() { return raw_data_memory; }
 size_t DiskManager::RawDatMemorySize() { return kPageSize - meta_data_size; }
 
 void DiskManager::FullyFlush() {
-  if(fp==nullptr) return;
+  if (fp == nullptr) return;
   fseek(fp, 0, SEEK_SET);
   fwrite(&first_empty_page_id, sizeof(page_id_t), 1, fp);
   fwrite(&current_total_page_count, sizeof(size_t), 1, fp);

test/CMakeLists.txt

@@ -6,8 +6,10 @@ endif()
 add_executable(replacer_test replacer_test.cpp)
 target_link_libraries(replacer_test bpt GTest::gtest_main)
 add_executable(buffer_pool_manager_test buffer_pool_manager_test.cpp)
-target_link_libraries(buffer_pool_manager_test bpt GTest::gtest_main)
+target_link_libraries(buffer_pool_manager_test bpt GTest::gtest_main spdlog::spdlog)
 add_executable(page_guard_test page_guard_test.cpp)
 target_link_libraries(page_guard_test bpt GTest::gtest_main)
 add_executable(bpt_basic_test bpt_basic_test.cpp)
 target_link_libraries(bpt_basic_test bpt GTest::gtest_main)
+add_executable(buffer_pool_manager_extreme_test buffer_pool_manager_extreme_test.cpp)
+target_link_libraries(buffer_pool_manager_extreme_test bpt)

test/MemoryRiver.hpp

@@ -68,7 +68,7 @@ class MemoryRiver {
     }
     if (FN != "") file_name = FN;
     if (file_name == "") return;
-    disk_manager = new DiskManager(file_name);
+    disk_manager = new DiskManager(file_name, true);
     bpm = new BufferPoolManager(100, 5, disk_manager);
     raw_mem = bpm->RawDataMemory();
     memset(raw_mem, 0, bpm->RawDatMemorySize());
@@ -135,10 +135,10 @@ class MemoryRiver {
     guard.AsMut<Page>()->dat.elements[element_id].nxt_blank = first_blank_element_pair_id;
     first_blank_element_pair_id = index;
     guard.AsMut<Page>()->dat.elements_count--;
-    if (guard.AsMut<Page>()->dat.elements_count == 0) {
+    // if (guard.AsMut<Page>()->dat.elements_count == 0) {
-      guard.Drop();
+    //   guard.Drop();
-      bpm->DeletePage(frame_id);
+    //   bpm->DeletePage(frame_id);
-    }
+    // }
   }
 };
 

test/MemoryRiverStd.hpp

@@ -0,0 +1,226 @@
+#ifndef BPT_MemoryRiver_HPP
+#define BPT_MemoryRiver_HPP
+
+#include <algorithm>
+#include <cstring>
+#include <fstream>
+#include <queue>
+#include <stack>
+#include <string>
+#include <unordered_map>
+namespace sol {
+template <class T, const int info_len = 2, const int kBufSize = 50>
+class MemoryRiver {
+ private:
+  static const int kPageSize = 4096;
+  static const int kDataBiginOffset = ((info_len + 2) * sizeof(int) + kPageSize - 1) / kPageSize * kPageSize;
+  static const int sizeofT = sizeof(T);
+  struct DataType {
+    int next_vacant_data_index;
+    T val;
+  };
+  static const int kBlockSize = (sizeof(DataType) + kPageSize - 1) / kPageSize * kPageSize;
+  static const int kDataPerBlock = kBlockSize / sizeof(DataType);
+  struct BlockType {
+    DataType data[kDataPerBlock];
+  };
+  char rest[kBlockSize];
+  static_assert(kBlockSize % kPageSize == 0, "kBlockSize % kPageSize != 0");
+  static_assert(kBlockSize >= sizeof(BlockType), "kBlockSize < sizeof(BlockType)");
+  std::string file_name;
+  int total_block_number = 0, first_vacant_data_index = 0;
+  /**
+   * DataIndex=(BlockIndex-1)*kDataPerBlock+InnnerIndex
+   */
+  std::unordered_map<int, BlockType *> cache;
+  std::queue<int> vis_que;
+  BlockType *cache_mem, **cache_mem_ptr, **cache_mem_ptr_beg;
+  void LoadCache(int block_index) {
+    BlockType *tmp = *(--cache_mem_ptr);
+    fs.seekg(kDataBiginOffset + (block_index - 1) * kBlockSize, std::ios::beg);
+    fs.read(reinterpret_cast<char *>(tmp), sizeof(BlockType));
+    cache[block_index] = tmp;
+    vis_que.push(block_index);
+  }
+  void ReleaseOldestCache() {
+    int block_index = vis_que.front();
+    vis_que.pop();
+    fs.seekp(kDataBiginOffset + (block_index - 1) * kBlockSize, std::ios::beg);
+    fs.write(reinterpret_cast<char *>(cache[block_index]), sizeof(BlockType));
+    *(cache_mem_ptr++) = cache[block_index];
+    cache.erase(block_index);
+  }
+  BlockType *OrderBlock(int block_index) {
+    if (cache.find(block_index) != cache.end()) return cache[block_index];
+    if (cache.size() == kBufSize) ReleaseOldestCache();
+    LoadCache(block_index);
+    return cache[block_index];
+  }
+  int AppEndBlock() {
+    if (cache.size() == kBufSize) ReleaseOldestCache();
+    BlockType *tmp = *(--cache_mem_ptr);
+    fs.seekp(0, std::ios::end);
+    fs.write(rest, kBlockSize);
+    ++total_block_number;
+    cache[total_block_number] = tmp;
+    vis_que.push(total_block_number);
+    return total_block_number;
+  }
+
+ public:
+  std::fstream fs;
+  MemoryRiver() {
+    cache_mem = new BlockType[kBufSize + 5];
+    typedef BlockType *BlockTypePtr;
+    cache_mem_ptr = new BlockTypePtr[kBufSize + 5];
+    for (int i = 0; i < kBufSize + 5; i++) cache_mem_ptr[i] = &cache_mem[i];
+    cache_mem_ptr_beg = cache_mem_ptr;
+    cache_mem_ptr += kBufSize + 5;
+  }
+  MemoryRiver(const MemoryRiver &) = delete;
+  MemoryRiver &operator=(const MemoryRiver &) = delete;
+  MemoryRiver(const std::string &file_name) : file_name(file_name) {
+    cache_mem = new BlockType[kBufSize + 5];
+    typedef BlockType *BlockTypePtr;
+    cache_mem_ptr = new BlockTypePtr[kBufSize + 5];
+    for (int i = 0; i < kBufSize + 5; i++) cache_mem_ptr[i] = &cache_mem[i];
+    cache_mem_ptr_beg = cache_mem_ptr;
+    cache_mem_ptr += kBufSize + 5;
+    OpenFile(file_name);
+  }
+
+  inline bool IsOpen() const noexcept { return fs.is_open(); }
+  ~MemoryRiver() {
+    CloseFile();
+    delete[] cache_mem;
+    delete[] cache_mem_ptr_beg;
+  }
+  void CloseFile() {
+    if (!fs.is_open()) return;
+    while (cache.size() > 0) ReleaseOldestCache();
+    fs.seekp(sizeof(int) * info_len, std::ios::beg);
+    fs.write(reinterpret_cast<char *>(&first_vacant_data_index), sizeof(int));
+    fs.write(reinterpret_cast<char *>(&total_block_number), sizeof(int));
+    fs.close();
+    file_name = "";
+    first_vacant_data_index = 0;
+    total_block_number = 0;
+  }
+  void OpenFile(const std::string &__file_name) {
+    if (fs.is_open()) CloseFile();
+    file_name = __file_name;
+    fs.open(file_name, std::ios::in | std::ios::out | std::ios::binary);
+    if (!fs.is_open()) {
+      fs.open(file_name, std::ios::out | std::ios::binary);
+      fs.seekp(0, std::ios::beg);
+      int tmp = 0;
+      total_block_number = 0;
+      first_vacant_data_index = 0;
+      memset(rest, 0, kPageSize);
+      for (int i = 0; i < kDataBiginOffset / kPageSize; ++i) {
+        fs.write(reinterpret_cast<char *>(rest), kPageSize);
+      }
+      fs.close();
+      fs.open(file_name, std::ios::in | std::ios::out | std::ios::binary);
+    }
+    fs.seekg(sizeof(int) * info_len, std::ios::beg);
+    fs.read(reinterpret_cast<char *>(&first_vacant_data_index), sizeof(int));
+    fs.read(reinterpret_cast<char *>(&total_block_number), sizeof(int));
+  }
+  void initialise(std::string FN = "") {
+    if (fs.is_open()) {
+      std::string name_bak = file_name;
+      CloseFile();
+      file_name = name_bak;
+    }
+    if (FN != "") file_name = FN;
+    if (file_name == "") return;
+    fs.open(file_name, std::ios::out | std::ios::binary);
+    fs.seekp(0, std::ios::beg);
+    int tmp = 0;
+    total_block_number = 0;
+    first_vacant_data_index = 0;
+    memset(rest, 0, kPageSize);
+    for (int i = 0; i < kDataBiginOffset / kPageSize; ++i) {
+      fs.write(reinterpret_cast<char *>(rest), kPageSize);
+    }
+    fs.close();
+    fs.open(file_name, std::ios::in | std::ios::out | std::ios::binary);
+  }
+
+  void get_info(int &tmp, int n) noexcept {
+    if (n > info_len) return;
+    fs.seekg((n - 1) * sizeof(int), std::ios::beg);
+    fs.read(reinterpret_cast<char *>(&tmp), sizeof(int));
+  }
+
+  void write_info(int tmp, int n) noexcept {
+    if (n > info_len) return;
+    fs.seekp((n - 1) * sizeof(int), std::ios::beg);
+    fs.write(reinterpret_cast<char *>(&tmp), sizeof(int));
+  }
+
+  void LoadInfoTo(int *dest) {
+    fs.seekg(0, std::ios::beg);
+    fs.read(reinterpret_cast<char *>(dest), sizeof(int) * info_len);
+  }
+
+  void WriteInfoFrom(int *src) {
+    fs.seekp(0, std::ios::beg);
+    fs.write(reinterpret_cast<char *>(src), sizeof(int) * info_len);
+  }
+
+  int write(T &t) noexcept {
+    if (first_vacant_data_index == 0) {
+      int new_block_index = AppEndBlock();
+      int index = (new_block_index - 1) * kDataPerBlock + 1;
+      BlockType *blk_ptr = OrderBlock(new_block_index);
+      if (kDataPerBlock > 1)
+        first_vacant_data_index = index + 1;
+      else
+        first_vacant_data_index = 0;
+      for (int i = 1; i < kDataPerBlock - 1; i++) blk_ptr->data[i].next_vacant_data_index = index + i + 1;
+      blk_ptr->data[kDataPerBlock - 1].next_vacant_data_index = 0;
+      blk_ptr->data[0].next_vacant_data_index = 0;
+      // blk_ptr->data[0].val = t;
+      std::memmove(&blk_ptr->data[0].val, &t, sizeofT);
+      return index;
+    } else {
+      int block_index = (first_vacant_data_index - 1) / kDataPerBlock + 1;
+      int inner_index = (first_vacant_data_index - 1) % kDataPerBlock + 1;
+      BlockType *blk_ptr = OrderBlock(block_index);
+      int index = first_vacant_data_index;
+      first_vacant_data_index = blk_ptr->data[inner_index - 1].next_vacant_data_index;
+      blk_ptr->data[inner_index - 1].next_vacant_data_index = 0;
+      // blk_ptr->data[inner_index - 1].val = t;
+      std::memmove(&blk_ptr->data[inner_index - 1].val, &t, sizeofT);
+      return index;
+    }
+  }
+
+  void update(T &t, const int index) noexcept {
+    int block_index = (index - 1) / kDataPerBlock + 1;
+    int inner_index = (index - 1) % kDataPerBlock + 1;
+    BlockType *blk_ptr = OrderBlock(block_index);
+    // blk_ptr->data[inner_index - 1].val = t;
+    std::memmove(&blk_ptr->data[inner_index - 1].val, &t, sizeofT);
+  }
+
+  void read(T &t, const int index) noexcept {
+    int block_index = (index - 1) / kDataPerBlock + 1;
+    int inner_index = (index - 1) % kDataPerBlock + 1;
+    BlockType *blk_ptr = OrderBlock(block_index);
+    // t = blk_ptr->data[inner_index - 1].val;
+    std::memmove(&t, &blk_ptr->data[inner_index - 1].val, sizeofT);
+  }
+
+  void Delete(int index) noexcept {
+    int block_index = (index - 1) / kDataPerBlock + 1;
+    int inner_index = (index - 1) % kDataPerBlock + 1;
+    BlockType *blk_ptr = OrderBlock(block_index);
+    blk_ptr->data[inner_index - 1].next_vacant_data_index = first_vacant_data_index;
+    first_vacant_data_index = index;
+  }
+};
+}  // namespace sol
+#endif  // BPT_MemoryRiver_HPP

test/buffer_pool_manager_extreme_test.cpp

@@ -0,0 +1,97 @@
+#include<iostream>
+#include"./MemoryRiver.hpp"
+#include"./MemoryRiverStd.hpp"
+#include<cstdlib>
+
+using namespace std;
+
+struct T{
+    int a[1000];
+    int sum,num;
+    int f(){
+        int res=0;
+        for (size_t i = 0; i < 1000; i++)
+        {
+            res+=a[i];
+        }
+        if (res!=sum) return -1;
+        return res;
+    }
+
+    T():sum(0),num(0){
+        for (size_t i = 0; i < 1000; i++)
+        {
+            a[i]=rand()*rand()>>10;
+            sum+=a[i];
+        }
+    }
+
+    void g(){
+        for (size_t i = 0; i < 1000; i++)
+        {
+            a[i]*=2;
+        }
+        sum*=2;
+    }
+
+    bool operator!=(const T & other){
+        if (sum!=other.sum) return true;
+        if (num!=other.num) return true;
+        for (size_t i = 0; i < 1000; i++)
+        {
+            if (a[i] != other.a[i]) return true;
+        }
+        return false;
+    }
+};
+
+const int n=30000;
+sol::MemoryRiver<T,4> STD("STD.file");
+MemoryRiver<T,4> mr("save.file");
+int off[n],offSTD[n];
+T tmp,tmpSTD;
+
+int main(){
+    srand(2333);
+    mr.initialise();
+    STD.initialise();
+    for (size_t i = 0; i < n; i++)
+    {
+        tmp.num=i;
+        off[i] = mr.write(tmp);
+        offSTD[i] = STD.write(tmp);
+    }
+    int r=0,rSTD=0;
+    while (r<n)
+    {
+        r=0;
+        rSTD = -1;
+        mr.get_info(r,3);
+        STD.get_info(rSTD,3);
+        if (r != rSTD) {
+            cout<<0;
+            return 0;
+        }
+        mr.read(tmp,off[r]);
+        STD.read(tmpSTD,offSTD[rSTD]);
+        if (tmp.f() != tmpSTD.f()) {
+            cout<<0;
+            return 0;
+        }
+        r+=2;
+        rSTD+=2;
+        mr.write_info(r,3);
+        STD.write_info(rSTD,3);
+    }
+    for (size_t i = 0; i < n; i++)
+    {
+        mr.read(tmp,off[i]);
+        STD.read(tmpSTD,offSTD[i]);
+        if (tmp!=tmpSTD) {
+            cout<<0;
+            return 0;
+        }
+    }
+    cout<<1;
+    return 0;
+}

test/buffer_pool_manager_test.cpp

@@ -1,13 +1,19 @@
 #include "bpt/buffer_pool_manager.h"
 #include <gtest/gtest.h>
+#include <spdlog/async.h>
+#include <spdlog/sinks/basic_file_sink.h>
+#include <spdlog/sinks/stdout_color_sinks.h>
+#include <spdlog/spdlog.h>
 #include <cstddef>
 #include <cstdio>
 #include <cstring>
+#include <deque>
 #include <memory>
 #include <random>
 #include <string>
 #include <vector>
 #include "MemoryRiver.hpp"
+#include "MemoryRiverStd.hpp"
 #include "bpt/bpt_page.hpp"
 #include "bpt/config.h"
 #include "bpt/disk_manager.h"
@@ -193,16 +199,17 @@ TEST(StoreTest, Test1) {
 }
 
 TEST(MemoryRiver, T1) {
-  remove("/tmp/test2.db");
   typedef unsigned long long DataType;
+  std::vector<DataType> record;
+  std::vector<size_t> id_record;
+  const int test_cnt = 30000;
+  remove("/tmp/test2.db");
+  {
     MemoryRiver<DataType> river;
     river.initialise("/tmp/test2.db");
     int x = 3;
     river.write_info(x, 1);
     DataType dat1 = 0x1f2f3f4f5f6f7f8f;
-  std::vector<DataType> record;
-  std::vector<size_t> id_record;
-  int test_cnt = 3;
     for (int i = 0; i < test_cnt; i++) {
       DataType tmp = dat1 + i;
       size_t element_id = river.write(tmp);
@@ -214,4 +221,132 @@ TEST(MemoryRiver, T1) {
       river.read(tmp, id_record[i]);
       EXPECT_EQ(record[i], tmp);
     }
+    std::random_device r;
+    std::default_random_engine rng(r());
+    for (int i = 0; i < 1000; i++) {
+      int t = rng() % record.size();
+      river.Delete(id_record[t]);
+      record.erase(record.begin() + t);
+      id_record.erase(id_record.begin() + t);
+    }
+  }
+  {
+    MemoryRiver<DataType> river("/tmp/test2.db");
+    int x;
+    river.get_info(x, 1);
+    EXPECT_EQ(3, x);
+    for (int i = 0; i < test_cnt; i++) {
+      DataType tmp;
+      river.read(tmp, id_record[i]);
+      EXPECT_EQ(record[i], tmp);
+    }
+  }
+}
+
+template <size_t length>
+class FixLengthString {
+ public:
+  char data[length];
+  FixLengthString &operator=(const FixLengthString &other) {
+    memcpy(data, other.data, length);
+    return *this;
+  }
+  bool operator==(const FixLengthString &other) const { return memcmp(data, other.data, length) == 0; }
+};
+TEST(MemoryRiver, T2) {
+  spdlog::set_level(spdlog::level::debug);
+  auto logger_ptr = spdlog::stderr_color_mt("stderr_logger");
+  const static size_t string_len = 5;
+  typedef FixLengthString<string_len> DataType;
+  std::deque<size_t> index_collection;
+  std::unordered_map<size_t, std::pair<int, int>> index_track;
+  size_t interal_id_tot = 0;
+  const unsigned int RndSeed = 3794;  // testing::GTEST_FLAG(random_seed);
+  std::mt19937 rnd(RndSeed);
+  remove("/tmp/T2.std");
+  remove("/tmp/T2.dat");
+  const int kInfoLength = 100;
+  {
+    sol::MemoryRiver<DataType, kInfoLength> STD("/tmp/T2.std");
+    MemoryRiver<DataType, kInfoLength> mr("/tmp/T2.dat");
+    int total_opts = 5;
+    while (total_opts-- > 0) {
+      int opt = rnd() % 6;
+      switch (opt) {
+        case 0: {
+          // get_info
+          int idx = 1 + rnd() % kInfoLength;
+          int std_ans, mr_ans;
+          STD.get_info(std_ans, idx);
+          mr.get_info(mr_ans, idx);
+          EXPECT_EQ(std_ans, mr_ans);
+          break;
+        }
+        case 1: {
+          // write_info
+          int idx = 1 + rnd() % kInfoLength;
+          int val = rnd();
+          STD.write_info(val, idx);
+          mr.write_info(val, idx);
+          break;
+        }
+        case 2: {
+          // write
+          interal_id_tot++;
+          index_collection.push_back(interal_id_tot);
+          DataType tmp;
+          for (int i = 0; i < string_len; i++) tmp.data[i] = 'a' + rnd() % 26;
+          tmp.data[string_len - 1] = '\0';
+          index_track[interal_id_tot].first = STD.write(tmp);
+          index_track[interal_id_tot].second = mr.write(tmp);
+          logger_ptr->info("Write: {}", tmp.data);
+          logger_ptr->info("internal id: {}", interal_id_tot);
+          logger_ptr->info("index in STD: {}", index_track[interal_id_tot].first);
+          logger_ptr->info("index in MR: {}", index_track[interal_id_tot].second);
+          break;
+        }
+        case 3: {
+          // update
+          if (index_collection.empty()) goto nxt;
+          size_t selected_internal_id = index_collection[rnd() % index_collection.size()];
+          DataType tmp;
+          for (int i = 0; i < string_len; i++) tmp.data[i] = 'a' + rnd() % 26;
+          tmp.data[string_len - 1] = '\0';
+          STD.update(tmp, index_track[selected_internal_id].first);
+          mr.update(tmp, index_track[selected_internal_id].second);
+          logger_ptr->info("Update: {}", tmp.data);
+          logger_ptr->info("internal id: {}", selected_internal_id);
+          logger_ptr->info("index in STD: {}", index_track[selected_internal_id].first);
+          logger_ptr->info("index in MR: {}", index_track[selected_internal_id].second);
+          break;
+        }
+        case 4: {
+          // read
+          if (index_collection.empty()) goto nxt;
+          size_t selected_internal_id = index_collection[rnd() % index_collection.size()];
+          DataType std_ans, mr_ans;
+          STD.read(std_ans, index_track[selected_internal_id].first);
+          mr.read(mr_ans, index_track[selected_internal_id].second);
+          logger_ptr->info("Read: {}", selected_internal_id);
+          logger_ptr->info("MR: read {} from {}", mr_ans.data, index_track[selected_internal_id].second);
+          logger_ptr->info("STD: read {} from {}", std_ans.data, index_track[selected_internal_id].first);
+          EXPECT_EQ(std_ans, mr_ans);
+        }
+        case 5: {
+          // Delete
+          if (index_collection.empty()) goto nxt;
+          size_t selected_internal_id = index_collection[rnd() % index_collection.size()];
+          logger_ptr->info("Delete: {}", selected_internal_id);
+          logger_ptr->info("index in STD: {}", index_track[selected_internal_id].first);
+          logger_ptr->info("index in MR: {}", index_track[selected_internal_id].second);
+          STD.Delete(index_track[selected_internal_id].first);
+          mr.Delete(index_track[selected_internal_id].second);
+          index_collection.erase(std::find(index_collection.begin(), index_collection.end(), selected_internal_id));
+          index_track.erase(selected_internal_id);
+          break;
+        }
+      }
+    nxt:;
+    }
+  }
 }