Merge remote-tracking branch 'origin/master'

# Conflicts:
#	db/db_impl.cc

db/db_impl.cc

@@ -1136,11 +1136,12 @@ int64_t DBImpl::TEST_MaxNextLevelOverlappingBytes() {
   return versions_->MaxNextLevelOverlappingBytes();
 }
 
-Status DBImpl::Get(const ReadOptions& options, const Slice& key,
-                   std::string* value) {
+Status DBImpl::get_slot_num(const ReadOptions& options, const Slice& key,
+                   size_t *slot_num) {
   Status s;
   MutexLock l(&mutex_);
   SequenceNumber snapshot;
+  std::string value;
   if (options.snapshot != nullptr) {
     snapshot =
         static_cast<const SnapshotImpl*>(options.snapshot)->sequence_number();
@@ -1163,12 +1164,12 @@ Status DBImpl::Get(const ReadOptions& options, const Slice& key,
     mutex_.Unlock();
     // First look in the memtable, then in the immutable memtable (if any).
     LookupKey lkey(key, snapshot);
-    if (mem->Get(lkey, value, &s)) {
+    if (mem->Get(lkey, &value, &s)) {
       // Done
-    } else if (imm != nullptr && imm->Get(lkey, value, &s)) {
+    } else if (imm != nullptr && imm->Get(lkey, &value, &s)) {
       // Done
     } else {
-      s = current->Get(options, lkey, value, &stats);
+      s = current->Get(options, lkey, &value, &stats);
       have_stat_update = true;
     }
     mutex_.Lock();
@@ -1180,6 +1181,9 @@ Status DBImpl::Get(const ReadOptions& options, const Slice& key,
   mem->Unref();
   if (imm != nullptr) imm->Unref();
   current->Unref();
+
+  *slot_num = *(size_t *)(&value)->c_str();
+
   return s;
 }
 
@@ -1193,7 +1197,7 @@ Status DBImpl::Get_Fields(const ReadOptions& options, const Slice& key,
   }
   // 从value中提取slot_num
   size_t slot_num;
-    slot_num = *(size_t *)value.c_str();  // 这里假设value的前几个字节存储了slot_num
+  slot_num = *(size_t *)value.c_str();  // 这里假设value的前几个字节存储了slot_num
   struct slot_content sc;
   std::string vlog_value;
   // 从slot_page中获取slot内容
@@ -1209,6 +1213,21 @@ Status DBImpl::Get_Fields(const ReadOptions& options, const Slice& key,
   // TODO(end)
 }
 
+Status DBImpl::Get(const ReadOptions& options, const Slice& key,
+                   std::string* value) {
+  size_t slot_num;
+  auto s = get_slot_num(options, key, &slot_num);
+
+  // TODO: search the slotpage and get value from vlog
+  struct slot_content sc;
+  std::string vlog_value;
+  slot_page_->get_slot(slot_num, &sc);
+  vlog_set_->get_value(sc.vlog_num, sc.value_offset, &vlog_value);
+  *value = vlog_value;
+
+  return s;
+}
+
 Iterator* DBImpl::NewIterator(const ReadOptions& options) {
   SequenceNumber latest_snapshot;
   uint32_t seed;
@@ -1257,30 +1276,31 @@ Status DBImpl::Put_Fields(const WriteOptions& opt, const Slice& key,
   return DB::Put(opt, key, slot_val);
   // TODO(end)
 }
+
+// Convenience methods
 Status DBImpl::Put(const WriteOptions& o, const Slice& key, const Slice& val) {
-  return DB::Put(o, key, val);
+  // TODO: allocate slot_num in slotpage and put value in vlog
+
+  size_t slot_num = slot_page_->alloc_slot();
+  struct slot_content sc;
+  vlog_set_->put_value(&sc.vlog_num, &sc.value_offset, val);
+  slot_page_->set_slot(slot_num, &sc);
+
+  char data[sizeof(size_t)];
+  memcpy(data, &slot_num, sizeof(size_t));
+  Slice slot_val(data, sizeof(data));
+
+  return DB::Put(o, key, slot_val);
 }
 
 Status DBImpl::Delete(const WriteOptions& options, const Slice& key) {
-//  size_t slot_num = *(size_t *)value->c_str();
-//  struct slot_content sc;
-//  std::string vlog_value;
-//  slot_page_->get_slot(slot_num, &sc);
-//  vlog_set_->get_value(sc.vlog_num, sc.value_offset, &vlog_value);
-//  *value = vlog_value;
-  // TODO(begin)
-  ReadOptions ro;
-  ro.verify_checksums = true;
-  ro.fill_cache = false;
-  ro.snapshot = nullptr;
-  std::string value;
-  Get(ro, key, &value);
-  size_t slot_num = *(size_t *)value.c_str();
+  size_t slot_num;
+  auto s = get_slot_num(ReadOptions(), key, &slot_num);
+
   struct slot_content sc;
-  std::string vlog_value;
   slot_page_->get_slot(slot_num, &sc);
   vlog_set_->del_value(sc.vlog_num, sc.value_offset);
-  // TODO(end)
+
   return DB::Delete(options, key);
 }
 

db/db_impl.h

@@ -119,6 +119,9 @@ class DBImpl : public DB {
     int64_t bytes_written;
   };
 
+  Status get_slot_num(const ReadOptions& options, const Slice& key,
+                      size_t *slot_num);
+
   Iterator* NewInternalIterator(const ReadOptions&,
                                 SequenceNumber* latest_snapshot,
                                 uint32_t* seed);

db/vlog_set.cpp

@@ -22,7 +22,12 @@ VlogSet::VlogSet(std::string dbname, VlogGC *vlog_gc) : dbname(dbname), vlog_gc(
     delete this->config_file_;
     // 重新以读写模式打开
     this->config_file_ = new std::fstream(cfname, std::ios::in | std::ios::out);
+    this->config_file_->seekp(0);
+    size_t tmp = 0;
+    this->config_file_->write(reinterpret_cast<const char*>(&tmp), sizeof(size_t));
+    this->config_file_->flush();
     this->vlog_nums_ = 0;
+    register_new_vlog();
   } else {
     // config 文件存在
     size_t _vlog_nums_;
@@ -43,7 +48,7 @@ VlogSet::VlogSet(std::string dbname, VlogGC *vlog_gc) : dbname(dbname), vlog_gc(
         curr_vlog.read(reinterpret_cast<char*>(curr_vlog_header), 2*sizeof(size_t));
         curr_vlog.close();
 
-        restore_vlog_inmaps(new vlog_info(curr_vlog_num, tmp[1], tmp[0]));
+        restore_vlog_inmaps(new vlog_info(curr_vlog_num, curr_vlog_header[1], curr_vlog_header[0]));
       }
     }
   }
@@ -162,7 +167,7 @@ size_t VlogSet::register_new_vlog() {
   size_t vn = vlog_nums_;
   std::string vlog_name = get_vlog_name(vn);
   register_inconfig_file(vn);
-  create_vlog(vlog_name);
+  create_vlog(vn);
 //  auto vlog_new = new std::fstream(vlog_name, std::ios::in | std::ios::out);
 //  if (!vlog_new->is_open()) {
 //    std::cerr << "Failed to open or create the vlog file: " << vlog_new << std::endl;
@@ -190,8 +195,12 @@ void VlogSet::remove_old_vlog(size_t old_vlog_num) {
 }
 
 bool VlogSet::vlog_need_gc(size_t vlog_num) {
+  // FIXME: vlog应该已经满了才行
   std::string vlog_name = get_vlog_name(vlog_num);
   auto vi = vlog_info_map_[vlog_name];
+  if ((double)vi->curr_size/VLOG_SIZE < VLOG_GC_THREHOLD) {
+    return false;
+  }
   bool retval = vi->vlog_valid_ && (vi->discard/vi->value_nums >= GC_THREDHOLD);
   return retval;
 }
@@ -223,13 +232,23 @@ void VlogSet::remove_from_config_file(size_t vlog_num) {
   }
 }
 
-void VlogSet::create_vlog(std::string &vlog_name) {
+void VlogSet::create_vlog(size_t vlog_num) {
+  auto vlog_name = get_vlog_name(vlog_num);
   std::fstream *vlog_new = new std::fstream(vlog_name, std::ios::out);
-  char tmp[2*sizeof(size_t)];
-  memset(tmp, 0, sizeof(tmp));
-  vlog_new->write(tmp, sizeof(tmp));
+  vlog_new->close();
+  delete vlog_new;
+  // 重新以读写模式打开
+  vlog_new = new std::fstream(vlog_name, std::ios::in | std::ios::out);
+  if (!vlog_new->is_open()) {
+    std::cerr << "Failed to open or create the vlog file: " << vlog_name << std::endl;
+    std::exit(EXIT_FAILURE);
+  }
+  size_t tmp[2] = {2*sizeof(size_t), vlog_num};
+  vlog_new->seekp(0);
+  vlog_new->write(reinterpret_cast<const char*>(tmp), sizeof(tmp));
   vlog_new->flush();
   vlog_new->close();
+  delete vlog_new;
 }
 
 inline void VlogSet::restore_vlog_inmaps(struct vlog_info *vi) {
@@ -239,8 +258,10 @@ inline void VlogSet::restore_vlog_inmaps(struct vlog_info *vi) {
 }
 
 inline void VlogSet::register_vlog_inmaps(size_t vlog_num, std::string &vlog_name) {
-  vlog_info_map_[vlog_name] = new vlog_info(vlog_num);
-  vlog_handler_map_[vlog_name] = new vlog_handler();
+  auto vinfo = new vlog_info(vlog_num);
+  auto vhandler = new vlog_handler();
+  vlog_info_map_[vlog_name] = vinfo;
+  vlog_handler_map_[vlog_name] = vhandler;
 }
 
 inline void VlogSet::remove_vlog_from_maps(std::string &vlog_name) {
@@ -277,10 +298,12 @@ void VlogSet::read_vlog_value(uint32_t vlog_num, uint32_t value_offset, std::str
   char value_buff[VALUE_BUFF_SIZE];
   handler.read(value_buff, VALUE_BUFF_SIZE);
 
+  // FIXME: remove value size
   uint16_t value_size;
   memcpy(&value_size, value_buff, sizeof(uint16_t));
   value_size &= VALUE_SIZE_MASK;
-  value->assign(&value_buff[sizeof(uint16_t)], value_size);
+  *value = std::string(value_buff);
+//  value->assign(&value_buff[sizeof(uint16_t)], value_size);
   handler.close();
 }
 
@@ -292,8 +315,6 @@ void VlogSet::write_vlog_value(uint32_t vlog_num, uint32_t value_offset, const l
   handler.write(value_buff, value.size());
 
   auto vinfo = get_vlog_info(vlog_num);
-  vinfo->value_nums ++;
-  vinfo->curr_size += value.size();
 
   handler.flush();
   handler.close();
@@ -315,7 +336,7 @@ void VlogSet::mark_del_value(uint32_t vlog_num, uint32_t value_offset) {
     return ;
   }
   assert(!(value_size & VALUE_DELE_MASK));
-  uint16_t masked_value_size = value_size & 0xffff;
+  uint16_t masked_value_size = value_size | VALUE_DELE_MASK;
   memcpy(value_buff, &masked_value_size, sizeof(uint16_t));
   handler.write(value_buff, value_size);
   handler.flush();

db/vlog_set.h

@@ -21,6 +21,7 @@ friend class VlogGC;
 
 #define CONFIG_FILE_DELE_MASK       (0x1 << (sizeof(size_t)-1))
 #define CONFIG_FILE_VLOG_NUM(v)     ((v) & ~CONFIG_FILE_DELE_MASK)
+#define VLOG_GC_THREHOLD            0.8
  public:
   VlogSet(std::string dbname, VlogGC *vlog_gc);
   ~VlogSet();
@@ -37,7 +38,7 @@ friend class VlogGC;
 
   void register_inconfig_file(size_t vlog_num);
   void remove_from_config_file(size_t vlog_num);
-  void create_vlog(std::string &vlog_name);
+  void create_vlog(size_t vlog_num);
   struct vlog_info *get_writable_vlog_info(size_t value_size);
   inline void restore_vlog_inmaps(struct vlog_info *vi);
   inline void register_vlog_inmaps(size_t vlog_num, std::string &vlog_name);

test/db_test1.cc

@@ -14,11 +14,13 @@ int main() {
   string s;
   db->Get(ReadOptions(), "001", &s);
   cout<<s<<endl;
+  cout << s.size() << endl;
 
   db->Put(WriteOptions(), "002", "world");
   string s1;
   db->Delete(WriteOptions(), "002");
   db->Get(ReadOptions(), "002", &s1);
+  cout << s1.size() << endl;
   cout<<s1<<endl;
 
   delete db;