diff --git a/db/db_impl.cc b/db/db_impl.cc index 1434df4..6765cb2 100644 --- a/db/db_impl.cc +++ b/db/db_impl.cc @@ -154,7 +154,10 @@ DBImpl::DBImpl(const Options& raw_options, const std::string& dbname) background_garbage_collect_scheduled_(false), manual_compaction_(nullptr), versions_(new VersionSet(dbname_, &options_, table_cache_, - &internal_comparator_)) {} + &internal_comparator_)) { + InitializeExistingLogs(); + // std::cout<<"init map"<smallest_snapshot) { // Hidden by an newer entry for same user key drop = true; // (A) + // Parse the value based on its first character + if(ikey.type != kTypeDeletion){ + Slice value = input->value(); + char type = value[0]; + if (type == 0x00) { + // Value is less than 100 bytes, use it directly + } else { + // Value is >= 100 bytes, read from external file + uint64_t file_id, valuelog_offset; + std::string file_id_str = value.ToString().substr(1, 8); + Slice file_id_slice(file_id_str); + bool res = GetVarint64(&file_id_slice, &file_id); + if (!res) return Status::Corruption("can't decode file id"); + if(valuelog_origin[file_id] == 0){ + valuelog_origin[file_id] = valuelog_usage[file_id]; + } + valuelog_usage[file_id]--; + // std::cout << "file_id: " << file_id << " usage: " << valuelog_usage[file_id] << std::endl; + } + } } else if (ikey.type == kTypeDeletion && ikey.sequence <= compact->smallest_snapshot && compact->compaction->IsBaseLevelForKey(ikey.user_key)) { @@ -1072,7 +1095,7 @@ Status DBImpl::DoCompactionWork(CompactionState* compact) { if (compact->builder->NumEntries() == 0) { compact->current_output()->smallest.DecodeFrom(key); } - compact->current_output()->largest.DecodeFrom(key); + compact->current_output()->largest.DecodeFrom(key); compact->builder->Add(key, input->value()); // Close output file if it is big enough @@ -1084,7 +1107,6 @@ Status DBImpl::DoCompactionWork(CompactionState* compact) { } } } - input->Next(); } @@ -1122,6 +1144,9 @@ Status DBImpl::DoCompactionWork(CompactionState* compact) { } VersionSet::LevelSummaryStorage tmp; Log(options_.info_log, "compacted to: %s", versions_->LevelSummary(&tmp)); + // for(int i=0;ic_str() + 1, value->length()); Slice new_value; - int value_offset = sizeof(uint64_t) * 2; // 16 uint64_t file_id, valuelog_offset; bool res = GetVarint64(&value_log_slice, &file_id); if (!res) return Status::Corruption("can't decode file id"); res = GetVarint64(&value_log_slice, &valuelog_offset); if (!res) return Status::Corruption("can't decode valuelog offset"); - // std::cout<<"file_id: "<> DBImpl::WriteValueLog( // 如果超出fixed_size if(offset>=config::value_log_size){ + int file_capacity=ReadFileSize(valuelogfile_number_); + // std::cout<<"file_capacity: "<> DBImpl::WriteValueLog( std::vector> res; for (const auto& [key_slice, value_slice] : kv) { - - // 写入 value 的长度 uint64_t value_len = value_slice.size(); valueFile.write(reinterpret_cast(&value_len), @@ -1652,7 +1677,7 @@ std::vector> DBImpl::WriteValueLog( } // 更新文件数据大小 - file_data_size += sizeof(uint64_t) + key_len + sizeof(uint64_t) + value_len; + file_data_size ++; // 记录 file_id 和 offset res.push_back({valuelogfile_number_, offset}); // 更新偏移量 @@ -1758,7 +1783,6 @@ Status DBImpl::ReadValueLog(uint64_t file_id, uint64_t offset, // 垃圾回收实现 void DBImpl::GarbageCollect() { // 遍历数据库目录，找到所有 valuelog 文件 - std::vector filenames; Status s = env_->GetChildren(dbname_, &filenames); Log(options_.info_log, "start gc "); @@ -1767,11 +1791,24 @@ void DBImpl::GarbageCollect() { for (const auto& filename:filenames) { if (IsValueLogFile(filename)){ uint64_t cur_log_number = GetValueLogID(filename); + if (cur_log_number == valuelogfile_number_) { + continue; + } auto tmp_name = ValueLogFileName(dbname_, cur_log_number); - - if(!versions_->checkOldValueLog(tmp_name))valuelog_set.emplace(filename); + // std::cout <checkOldValueLog(tmp_name) && + valuelog_origin[cur_log_number]) { + if ((float)(valuelog_usage[cur_log_number] / + valuelog_origin[cur_log_number]) <= 0.6) { + valuelog_set.emplace(filename); + } + } } } + // std::cout << "valuelog_set size: " << valuelog_set.size() << std::endl; + Log(options_.info_log, "valuelog_set size: %d", valuelog_set.size()); + //bool tmp_judge=false;//only clean one file for (std::string valuelog_name : valuelog_set) { Log(options_.info_log, ("gc processing: "+valuelog_name).data()); @@ -2064,4 +2101,43 @@ Status DestroyDB(const std::string& dbname, const Options& options) { return result; } +// 读取所有现有日志文件的 file_data_size +void DBImpl::InitializeExistingLogs() { + std::vector filenames; + Status s = env_->GetChildren(dbname_, &filenames); + Log(options_.info_log, "start set file map "); + assert(s.ok()); + std::set valuelog_set; + for (const auto& filename : filenames) { + if (IsValueLogFile(filename)) { + uint64_t cur_log_number = GetValueLogID(filename); + uint64_t file_data_size = ReadFileSize(cur_log_number); + valuelog_usage.emplace(cur_log_number,file_data_size); + // std::cout << "cur_log_number: " << cur_log_number << " file_data_size: " << file_data_size << std::endl; + + } + } +} + +// 读取单个文件的 file_data_size +uint64_t DBImpl::ReadFileSize(uint64_t log_number) { + auto file_name = ValueLogFileName(dbname_, log_number); + std::ifstream valueFile(file_name, std::ios::in | std::ios::binary); + if (!valueFile.is_open()) { + std::cerr << "Failed to open file: " << file_name << std::endl; + return 0; + } + + uint64_t file_data_size = 0; + valueFile.read(reinterpret_cast(&file_data_size), sizeof(uint64_t)); + if (valueFile.fail() || valueFile.bad()) { + std::cerr << "Failed to read data size from file: " << file_name + << std::endl; + valueFile.close(); + return 0; + } + valueFile.close(); + return file_data_size; +} + } // namespace leveldb diff --git a/db/db_impl.h b/db/db_impl.h index ab8ec0e..9d829c6 100644 --- a/db/db_impl.h +++ b/db/db_impl.h @@ -69,7 +69,7 @@ class DBImpl : public DB { std::vector> WriteValueLog( std::vector> value) override; void addNewValueLog() override EXCLUSIVE_LOCKS_REQUIRED(mutex_); - ; + std::pair getNewValuelog(); // use for compaction // Status ReadValueLog(uint64_t file_id, uint64_t offset,Slice* // value)override; @@ -101,6 +101,11 @@ class DBImpl : public DB { // bytes. void RecordReadSample(Slice key); + void InitializeExistingLogs(); + + uint64_t ReadFileSize(uint64_t log_number); + + private: friend class DB; struct CompactionState; @@ -230,6 +235,9 @@ class DBImpl : public DB { uint64_t valuelogfile_number_; log::Writer* log_; std::map oldvaluelog_ids; + std::map valuelog_usage; + std::map valuelog_origin; + uint32_t seed_ GUARDED_BY(mutex_); // For sampling. // Queue of writers. diff --git a/test/test.cpp b/test/test.cpp index 56e79af..e8bfbf5 100644 --- a/test/test.cpp +++ b/test/test.cpp @@ -129,7 +129,7 @@ TEST(Test, Garbage_Collect_TEST) { abort(); } std::vector values; - for(int i=0;i<5000;i++){ + for(int i=0;i<50000;i++){ std::string key=std::to_string(i); std::string value; for(int j=0;j<1000;j++){ @@ -142,7 +142,7 @@ TEST(Test, Garbage_Collect_TEST) { db->TEST_GarbageCollect(); std::cout<<"finish gc"<