From 56fee562a67b9bef9c8d4ac92d640338fc5529ed Mon Sep 17 00:00:00 2001 From: cyq <1056374449@qq.com> Date: Mon, 21 Oct 2024 04:47:41 +0800 Subject: [PATCH 1/6] =?UTF-8?q?=E5=BF=BD=E7=95=A5ttl=E7=9A=84=E5=AD=98?= =?UTF-8?q?=E5=9C=A8=EF=BC=8C=E4=BD=BF=E5=BE=97=E7=A8=8B=E5=BA=8F=E8=83=BD?= =?UTF-8?q?=E5=A4=9F=E9=A1=BA=E5=88=A9=E8=BF=90=E8=A1=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- db/db_impl.cc | 13 +++++++++++++ db/db_impl.h | 2 ++ db/db_test.cc | 3 +++ port/port.h | 3 +++ test/db_test2.cc | 1 + test/ttl_test.cc | 4 ++-- 6 files changed, 24 insertions(+), 2 deletions(-) diff --git a/db/db_impl.cc b/db/db_impl.cc index f96d245..3d26371 100644 --- a/db/db_impl.cc +++ b/db/db_impl.cc @@ -1198,6 +1198,11 @@ Status DBImpl::Put(const WriteOptions& o, const Slice& key, const Slice& val) { return DB::Put(o, key, val); } +Status DBImpl::Put(const WriteOptions& options, const Slice& key, + const Slice& value, uint64_t ttl) { + return DB::Put(options,key,value,ttl); +} + Status DBImpl::Delete(const WriteOptions& options, const Slice& key) { return DB::Delete(options, key); } @@ -1491,6 +1496,14 @@ Status DB::Put(const WriteOptions& opt, const Slice& key, const Slice& value) { return Write(opt, &batch); } +//为了通过编译，忽略ttl +Status DB::Put(const WriteOptions& options, const Slice& key, + const Slice& value, uint64_t ttl) { + WriteBatch batch; + batch.Put(key, value); + return Write(options, &batch); +} + Status DB::Delete(const WriteOptions& opt, const Slice& key) { WriteBatch batch; batch.Delete(key); diff --git a/db/db_impl.h b/db/db_impl.h index c7b0172..993c63e 100644 --- a/db/db_impl.h +++ b/db/db_impl.h @@ -38,6 +38,8 @@ class DBImpl : public DB { // Implementations of the DB interface Status Put(const WriteOptions&, const Slice& key, const Slice& value) override; + Status Put(const WriteOptions& options, const Slice& key, + const Slice& value, uint64_t ttl) override; Status Delete(const WriteOptions&, const Slice& key) override; Status Write(const WriteOptions& options, WriteBatch* updates) override; Status Get(const ReadOptions& options, const Slice& key, diff --git a/db/db_test.cc b/db/db_test.cc index a4a84cd..7baec2a 100644 --- a/db/db_test.cc +++ b/db/db_test.cc @@ -2117,6 +2117,9 @@ class ModelDB : public DB { Status Put(const WriteOptions& o, const Slice& k, const Slice& v) override { return DB::Put(o, k, v); } + Status Put(const WriteOptions& o, const Slice& k,const Slice& v,uint64_t ttl) { + return DB::Put(o,k,v); + } Status Delete(const WriteOptions& o, const Slice& key) override { return DB::Delete(o, key); } diff --git a/port/port.h b/port/port.h index 4b247f7..c2073f3 100644 --- a/port/port.h +++ b/port/port.h @@ -7,6 +7,9 @@ #include +#ifndef LEVELDB_PLATFORM_POSIX + #define LEVELDB_PLATFORM_POSIX +#endif // Include the appropriate platform specific file below. If you are // porting to a new platform, see "port_example.h" for documentation // of what the new port_.h file must provide. diff --git a/test/db_test2.cc b/test/db_test2.cc index 49e26ab..4b2b33c 100644 --- a/test/db_test2.cc +++ b/test/db_test2.cc @@ -68,6 +68,7 @@ int main() { GetData(db); delete db; } + DestroyDB("testdb",Options()); return 0; } diff --git a/test/ttl_test.cc b/test/ttl_test.cc index 06f4cda..0cc64f3 100644 --- a/test/ttl_test.cc +++ b/test/ttl_test.cc @@ -99,9 +99,9 @@ TEST(TestTTL, CompactionTTL) { db->CompactRange(nullptr, nullptr); - leveldb::Range ranges[1]; + // leveldb::Range ranges[1]; ranges[0] = leveldb::Range("-", "A"); - uint64_t sizes[1]; + // uint64_t sizes[1]; db->GetApproximateSizes(ranges, 1, sizes); ASSERT_EQ(sizes[0], 0); } From b2574f0b940bd181b727293ce5f568bd8cf36193 Mon Sep 17 00:00:00 2001 From: cyq <1056374449@qq.com> Date: Tue, 29 Oct 2024 06:19:36 +0800 Subject: [PATCH 2/6] =?UTF-8?q?=E5=AF=B9filemetadata=E5=A2=9E=E5=8A=A0dead?= =?UTF-8?q?time=E8=8C=83=E5=9B=B4=EF=BC=8C=E5=B9=B6=E5=9C=A8=E7=9B=B8?= =?UTF-8?q?=E5=85=B3=E5=9C=B0=E6=96=B9=E8=BF=9B=E8=A1=8C=E8=BE=93=E5=85=A5?= =?UTF-8?q?=E7=9A=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- db/builder.cc | 9 +++++++++ db/db_impl.cc | 44 +++++++++++++++++++++++++++++--------------- db/dbformat.cc | 5 ++++- db/memtable.cc | 2 +- db/repair.cc | 2 +- db/version_edit.cc | 10 +++++++++- db/version_edit.h | 11 +++++++++-- db/version_set.cc | 3 ++- 8 files changed, 64 insertions(+), 22 deletions(-) diff --git a/db/builder.cc b/db/builder.cc index e6329e0..5a0e07f 100644 --- a/db/builder.cc +++ b/db/builder.cc @@ -31,9 +31,18 @@ Status BuildTable(const std::string& dbname, Env* env, const Options& options, TableBuilder* builder = new TableBuilder(options, file); meta->smallest.DecodeFrom(iter->key()); Slice key; + ParsedInternalKey parsed; + meta->smallest_deadtime = UINT64_MAX; + meta->largest_deadtime = 0; for (; iter->Valid(); iter->Next()) { key = iter->key(); builder->Add(key, iter->value()); + //在构建sstable文件的时候，记录当前文件的生存期最大和最小值, + //这里要注意internalkey和metadata中对于没有生存期的表示的转换 + ParseInternalKey(key,&parsed); + if(parsed.deadTime == 0) parsed.deadTime = UINT64_MAX; + meta->smallest_deadtime = std::min(meta->smallest_deadtime,parsed.deadTime); + meta->largest_deadtime = std::max(meta->largest_deadtime,parsed.deadTime); } if (!key.empty()) { meta->largest.DecodeFrom(key); diff --git a/db/db_impl.cc b/db/db_impl.cc index 4617e18..6513f89 100644 --- a/db/db_impl.cc +++ b/db/db_impl.cc @@ -52,11 +52,12 @@ struct DBImpl::Writer { }; struct DBImpl::CompactionState { - // Files produced by compaction + // Files produced by compaction 这里的改动和filemetadata对应 struct Output { uint64_t number; uint64_t file_size; InternalKey smallest, largest; + uint64_t smallest_deadtime,largest_deadtime; }; Output* current_output() { return &outputs[outputs.size() - 1]; } @@ -536,7 +537,7 @@ Status DBImpl::WriteLevel0Table(MemTable* mem, VersionEdit* edit, level = base->PickLevelForMemTableOutput(min_user_key, max_user_key); } edit->AddFile(level, meta.number, meta.file_size, meta.smallest, - meta.largest); + meta.largest,meta.smallest_deadtime,meta.largest_deadtime); } CompactionStats stats; @@ -740,7 +741,7 @@ void DBImpl::BackgroundCompaction() { FileMetaData* f = c->input(0, 0); c->edit()->RemoveFile(c->level(), f->number); c->edit()->AddFile(c->level() + 1, f->number, f->file_size, f->smallest, - f->largest); + f->largest,f->smallest_deadtime,f->largest_deadtime); status = versions_->LogAndApply(c->edit(), &mutex_); if (!status.ok()) { RecordBackgroundError(status); @@ -814,6 +815,8 @@ Status DBImpl::OpenCompactionOutputFile(CompactionState* compact) { out.number = file_number; out.smallest.Clear(); out.largest.Clear(); + out.smallest_deadtime = UINT64_MAX; + out.largest_deadtime = 0; compact->outputs.push_back(out); mutex_.Unlock(); } @@ -889,7 +892,7 @@ Status DBImpl::InstallCompactionResults(CompactionState* compact) { for (size_t i = 0; i < compact->outputs.size(); i++) { const CompactionState::Output& out = compact->outputs[i]; compact->compaction->edit()->AddFile(level + 1, out.number, out.file_size, - out.smallest, out.largest); + out.smallest, out.largest,out.smallest_deadtime,out.largest_deadtime); } return versions_->LogAndApply(compact->compaction->edit(), &mutex_); } @@ -1003,6 +1006,17 @@ Status DBImpl::DoCompactionWork(CompactionState* compact) { compact->current_output()->smallest.DecodeFrom(key); } compact->current_output()->largest.DecodeFrom(key); + + ParsedInternalKey parsed; + ParseInternalKey(key,&parsed); + uint64_t &smallest_deadtime = compact->current_output()->smallest_deadtime; + uint64_t &largest_deadtime = compact->current_output()->largest_deadtime; + if(parsed.deadTime == 0) { + smallest_deadtime = UINT64_MAX; + } + smallest_deadtime = std::min(smallest_deadtime,parsed.deadTime); + largest_deadtime = std::max(largest_deadtime,parsed.deadTime); + compact->builder->Add(key, input->value()); // Close output file if it is big enough @@ -1206,10 +1220,10 @@ Status DBImpl::Put(const WriteOptions& o, const Slice& key, return DB::Put(o, key, val, ttl); } -Status DBImpl::Put(const WriteOptions& options, const Slice& key, - const Slice& value, uint64_t ttl) { - return DB::Put(options,key,value,ttl); -} +// Status DBImpl::Put(const WriteOptions& options, const Slice& key, +// const Slice& value, uint64_t ttl) { +// return DB::Put(options,key,value,ttl); +// } Status DBImpl::Delete(const WriteOptions& options, const Slice& key) { return DB::Delete(options, key); @@ -1505,13 +1519,13 @@ Status DB::Put(const WriteOptions& opt, const Slice& key, return Write(opt, &batch); } -//为了通过编译，忽略ttl -Status DB::Put(const WriteOptions& options, const Slice& key, - const Slice& value, uint64_t ttl) { - WriteBatch batch; - batch.Put(key, value); - return Write(options, &batch); -} +// //为了通过编译，忽略ttl +// Status DB::Put(const WriteOptions& options, const Slice& key, +// const Slice& value, uint64_t ttl) { +// WriteBatch batch; +// batch.Put(key, value); +// return Write(options, &batch); +// } Status DB::Delete(const WriteOptions& opt, const Slice& key) { WriteBatch batch; diff --git a/db/dbformat.cc b/db/dbformat.cc index 686d04f..6f11a1d 100644 --- a/db/dbformat.cc +++ b/db/dbformat.cc @@ -75,11 +75,13 @@ int InternalKeyComparator::Compare(const Slice& akey, const Slice& bkey) const { r = -1; return r; } - + const uint64_t atime = DecodeFixed64(akey.data() + akey.size() - 16); + const uint64_t btime = DecodeFixed64(bkey.data() + bkey.size() - 16); //原本应该找到了，新加判断 if((btag & 0b100) && (atag & 0b10)){ //一个是查询键，另一个有ttl const uint64_t atime = DecodeFixed64(akey.data() + akey.size() - 16); const uint64_t btime = DecodeFixed64(bkey.data() + bkey.size() - 16); + std::cout<<"atime:"<& files = current_->files_[level]; for (size_t i = 0; i < files.size(); i++) { const FileMetaData* f = files[i]; - edit.AddFile(level, f->number, f->file_size, f->smallest, f->largest); + edit.AddFile(level, f->number, f->file_size, f->smallest, f->largest, + f->smallest_deadtime,f->largest_deadtime); } } From f5e8b29d5cd9d0b6c4c53fcc1fb32fab620059d3 Mon Sep 17 00:00:00 2001 From: cyq <1056374449@qq.com> Date: Tue, 29 Oct 2024 22:00:04 +0800 Subject: [PATCH 3/6] pass the test1 --- CMakeLists.txt | 1 + db/db_test.cc | 6 +++--- db/dbformat.cc | 23 ++++++++++---------- db/dbformat.h | 6 +++++- db/memtable.cc | 42 ++++++++++++++++++++++++++++++++++-- db/skiplist.h | 2 ++ db/version_set.cc | 3 +++ table/block.cc | 11 ++++++++++ test/ttl_mmtable_test.cc | 8 +++---- test/ttl_test.cc | 55 +++++++++++++++++++++++++++--------------------- 10 files changed, 112 insertions(+), 45 deletions(-) diff --git a/CMakeLists.txt b/CMakeLists.txt index 705abce..a9b1c5e 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -107,6 +107,7 @@ configure_file( include_directories( "${PROJECT_BINARY_DIR}/include" "." + "./third_party/googletest/googletest/include/" ) if(BUILD_SHARED_LIBS) diff --git a/db/db_test.cc b/db/db_test.cc index ee3a72a..79478b0 100644 --- a/db/db_test.cc +++ b/db/db_test.cc @@ -2117,9 +2117,9 @@ class ModelDB : public DB { Status Put(const WriteOptions& o, const Slice& k, const Slice& v, uint64_t ttl = 0) override { return DB::Put(o, k, v, ttl); } - Status Put(const WriteOptions& o, const Slice& k,const Slice& v,uint64_t ttl) { - return DB::Put(o,k,v); - } + // Status Put(const WriteOptions& o, const Slice& k,const Slice& v,uint64_t ttl) { + // return DB::Put(o,k,v); + // } Status Delete(const WriteOptions& o, const Slice& key) override { return DB::Delete(o, key); } diff --git a/db/dbformat.cc b/db/dbformat.cc index 6f11a1d..2ef75e2 100644 --- a/db/dbformat.cc +++ b/db/dbformat.cc @@ -78,15 +78,15 @@ int InternalKeyComparator::Compare(const Slice& akey, const Slice& bkey) const { const uint64_t atime = DecodeFixed64(akey.data() + akey.size() - 16); const uint64_t btime = DecodeFixed64(bkey.data() + bkey.size() - 16); //原本应该找到了，新加判断 - if((btag & 0b100) && (atag & 0b10)){ //一个是查询键，另一个有ttl - const uint64_t atime = DecodeFixed64(akey.data() + akey.size() - 16); - const uint64_t btime = DecodeFixed64(bkey.data() + bkey.size() - 16); - std::cout<<"atime:"<= 8); uint64_t num = DecodeFixed64(internal_key.data() + internal_key.size() - 8); uint8_t havettl = (num & 0b10) >> 1; @@ -195,7 +198,8 @@ inline bool ParseInternalKey(const Slice& internal_key, result->deadTime = 0; result->user_key = Slice(internal_key.data(), n - 8); } - return (c <= static_cast(kTypeValue)); + // return c <= 0b111; + return ((c & 0b1) <= static_cast(kTypeValue)); } // A helper class useful for DBImpl::Get() diff --git a/db/memtable.cc b/db/memtable.cc index 8131005..f27dca3 100644 --- a/db/memtable.cc +++ b/db/memtable.cc @@ -55,7 +55,23 @@ class MemTableIterator : public Iterator { ~MemTableIterator() override = default; bool Valid() const override { return iter_.Valid(); } - void Seek(const Slice& k) override { iter_.Seek(EncodeKey(&tmp_, k)); } + void Seek(const Slice& k) override { + iter_.Seek(EncodeKey(&tmp_, k)); + MemTable::KeyComparator comp_ = iter_.get_comparator(); + while(Valid()) { + Slice now = key(); + ParsedInternalKey parsed_k,parsed_now; + ParseInternalKey(k,&parsed_k); + ParseInternalKey(now,&parsed_now); + uint64_t deadtime_k = parsed_k.deadTime; + uint64_t deadtime_now = parsed_now.deadTime; + if(deadtime_k == 0) deadtime_k = UINT64_MAX; + if(deadtime_now == 0) deadtime_now = UINT64_MAX; + if(deadtime_k > deadtime_now) {Next();continue;}; + if(comp_.comparator.Compare(k,now) <= 0) return; + Next(); + } + } void SeekToFirst() override { iter_.SeekToFirst(); } void SeekToLast() override { iter_.SeekToLast(); } void Next() override { iter_.Next(); } @@ -115,13 +131,35 @@ void MemTable::Add(SequenceNumber s, ValueType type, const Slice& key, std::memcpy(p, value.data(), val_size); assert(p + val_size == buf + encoded_len); table_.Insert(buf); - std::cout << "insert:" << key.ToString() <<" deadTime: " << deadTime << std::endl; + static int count = 0; + if(count++ % 1000 == 0) + std::cout<<"count: "< deadtime_now) { + iter.Next(); + continue; + } + std::cout<<"size : "<compare_;} + private: const SkipList* list_; Node* node_; diff --git a/db/version_set.cc b/db/version_set.cc index 764a158..1bf9ddd 100644 --- a/db/version_set.cc +++ b/db/version_set.cc @@ -263,8 +263,10 @@ static void SaveValue(void* arg, const Slice& ikey, const Slice& v) { Saver* s = reinterpret_cast(arg); ParsedInternalKey parsed_key; if (!ParseInternalKey(ikey, &parsed_key)) { + // std::cout<<"corrupt get"<state = kCorrupt; } else { + std::cout<<"tar&found"<user_key.ToString()<ucmp->Compare(parsed_key.user_key, s->user_key) == 0) { s->state = (parsed_key.type == kTypeValue) ? kFound : kDeleted; if (s->state == kFound) { @@ -272,6 +274,7 @@ static void SaveValue(void* arg, const Slice& ikey, const Slice& v) { } } } + std::cout<<"state : "<state<= target + //处理deadtime：从当前位置向后找到最新的未死亡的key while (true) { if (!ParseNextKey()) { return; } + ParsedInternalKey parsed_target,parsed_key_; + ParseInternalKey(target,&parsed_target); + ParseInternalKey(key_,&parsed_key_); + uint64_t deadtime_tar = parsed_target.deadTime; + uint64_t deadtime_key_ = parsed_key_.deadTime; + if(deadtime_tar == 0) deadtime_tar = UINT64_MAX; + if(deadtime_key_ == 0) deadtime_key_ = UINT64_MAX; + std::cout<<"key :"< deadtime_key_) continue; if (Compare(key_, target) >= 0) { return; } diff --git a/test/ttl_mmtable_test.cc b/test/ttl_mmtable_test.cc index dccf9ed..5b44033 100644 --- a/test/ttl_mmtable_test.cc +++ b/test/ttl_mmtable_test.cc @@ -8,7 +8,7 @@ using namespace leveldb; constexpr int value_size = 2048; -constexpr int data_size = 2048 << 2; +constexpr int data_size = 2048 << 15; Status OpenDB(std::string dbName, DB **db) { Options options; @@ -92,7 +92,7 @@ DB *db; uint64_t ttl1 = 3; uint64_t ttl2 = 5; - InsertData(db, ttl2); + // InsertData(db, ttl2); InsertData(db, ttl1, 2); //都没过期先找到后插的 @@ -101,8 +101,8 @@ DB *db; //再找到前一次 Env::Default()->SleepForMicroseconds(3 * 1000000); - GetData(db, false); - + GetData(db, true); + DestroyDB("testdb",Options()); delete(db); printf("-----closing-----\n"); printf("success!\n"); diff --git a/test/ttl_test.cc b/test/ttl_test.cc index 12cd583..fb91b86 100644 --- a/test/ttl_test.cc +++ b/test/ttl_test.cc @@ -26,7 +26,8 @@ void InsertData(DB *db, uint64_t ttl/* second */) { int key_ = rand() % key_num+1; std::string key = std::to_string(key_); std::string value(value_size, 'a'); - db->Put(writeOptions, key, value, ttl); + db->Put(writeOptions, std::to_string(i+1), value, ttl); + // db->Put(writeOptions, key, value, ttl); } } @@ -45,13 +46,14 @@ void GetData(DB *db, int size = (1 << 30)) { } TEST(TestTTL, ReadTTL) { + DestroyDB("testdb",Options()); DB *db; if(OpenDB("testdb", &db).ok() == false) { std::cerr << "open db failed" << std::endl; abort(); } - uint64_t ttl = 20; + uint64_t ttl = 15; InsertData(db, ttl); @@ -64,47 +66,52 @@ TEST(TestTTL, ReadTTL) { std::string key = std::to_string(key_); std::string value; status = db->Get(readOptions, key, &value); + std::cout<SleepForMicroseconds(ttl * 1000000); + Env::Default()->SleepForMicroseconds((ttl+1) * 1000000); for (int i = 0; i < 100; i++) { int key_ = rand() % key_num+1; std::string key = std::to_string(key_); std::string value; status = db->Get(readOptions, key, &value); + std::cout<GetApproximateSizes(ranges, 1, sizes); -// ASSERT_GT(sizes[0], 0); + leveldb::Range ranges[1]; + ranges[0] = leveldb::Range("-", "A"); + uint64_t sizes[1]; + db->GetApproximateSizes(ranges, 1, sizes); + ASSERT_GT(sizes[0], 0); -// Env::Default()->SleepForMicroseconds(ttl * 1000000); + Env::Default()->SleepForMicroseconds(ttl * 1000000); -// db->CompactRange(nullptr, nullptr); + db->CompactRange(nullptr, nullptr); -// leveldb::Range ranges[1]; -// ranges[0] = leveldb::Range("-", "A"); -// uint64_t sizes[1]; -// db->GetApproximateSizes(ranges, 1, sizes); -// ASSERT_EQ(sizes[0], 0); -// } + // leveldb::Range ranges[1]; + ranges[0] = leveldb::Range("-", "A"); + // uint64_t sizes[1]; + db->GetApproximateSizes(ranges, 1, sizes); + ASSERT_EQ(sizes[0], 0); + delete db; +} int main(int argc, char** argv) { From 3c058fedd083574f57b88e79a28ac4bd247619f2 Mon Sep 17 00:00:00 2001 From: cyq <1056374449@qq.com> Date: Tue, 29 Oct 2024 23:11:45 +0800 Subject: [PATCH 4/6] pass all the test --- db/builder.cc | 11 +++++++++++ db/db_impl.cc | 12 ++++++++++-- test/ttl_test.cc | 5 +++-- 3 files changed, 24 insertions(+), 4 deletions(-) diff --git a/db/builder.cc b/db/builder.cc index 5a0e07f..3d1ca0e 100644 --- a/db/builder.cc +++ b/db/builder.cc @@ -28,6 +28,10 @@ Status BuildTable(const std::string& dbname, Env* env, const Options& options, return s; } + time_t nowTime; + time(&nowTime); + assert(nowTime > 0); + TableBuilder* builder = new TableBuilder(options, file); meta->smallest.DecodeFrom(iter->key()); Slice key; @@ -41,6 +45,13 @@ Status BuildTable(const std::string& dbname, Env* env, const Options& options, //这里要注意internalkey和metadata中对于没有生存期的表示的转换 ParseInternalKey(key,&parsed); if(parsed.deadTime == 0) parsed.deadTime = UINT64_MAX; + if(parsed.deadTime < nowTime) { + static int count = 0; + if(count % 1000 == 0) { + std::cout<<"count "<smallest_deadtime = std::min(meta->smallest_deadtime,parsed.deadTime); meta->largest_deadtime = std::max(meta->largest_deadtime,parsed.deadTime); } diff --git a/db/db_impl.cc b/db/db_impl.cc index 6513f89..28ef225 100644 --- a/db/db_impl.cc +++ b/db/db_impl.cc @@ -591,6 +591,7 @@ void DBImpl::CompactRange(const Slice* begin, const Slice* end) { } } } + max_level_with_files = config::kNumLevels - 1; //TODO:强制合并所有level中的sst,但是这么做不是很优雅 TEST_CompactMemTable(); // TODO(sanjay): Skip if memtable does not overlap for (int level = 0; level < max_level_with_files; level++) { TEST_CompactRange(level, begin, end); @@ -951,11 +952,20 @@ Status DBImpl::DoCompactionWork(CompactionState* compact) { // Handle key/value, add to state, etc. bool drop = false; + time_t nowTime; + time(&nowTime); if (!ParseInternalKey(key, &ikey)) { // Do not hide error keys current_user_key.clear(); has_current_user_key = false; last_sequence_for_key = kMaxSequenceNumber; + } else if(ikey.deadTime != 0 && ikey.deadTime < nowTime){ + static int count = 0; + if(count % 1000 == 0) { + std::cout<<"count "<Compare(ikey.user_key, Slice(current_user_key)) != @@ -981,7 +991,6 @@ Status DBImpl::DoCompactionWork(CompactionState* compact) { // Therefore this deletion marker is obsolete and can be dropped. drop = true; } - last_sequence_for_key = ikey.sequence; } #if 0 @@ -993,7 +1002,6 @@ Status DBImpl::DoCompactionWork(CompactionState* compact) { compact->compaction->IsBaseLevelForKey(ikey.user_key), (int)last_sequence_for_key, (int)compact->smallest_snapshot); #endif - if (!drop) { // Open output file if necessary if (compact->builder == nullptr) { diff --git a/test/ttl_test.cc b/test/ttl_test.cc index fb91b86..40d38d4 100644 --- a/test/ttl_test.cc +++ b/test/ttl_test.cc @@ -92,7 +92,7 @@ TEST(TestTTL, CompactionTTL) { abort(); } - uint64_t ttl = 20; + uint64_t ttl = 10; InsertData(db, ttl); leveldb::Range ranges[1]; @@ -101,7 +101,8 @@ TEST(TestTTL, CompactionTTL) { db->GetApproximateSizes(ranges, 1, sizes); ASSERT_GT(sizes[0], 0); - Env::Default()->SleepForMicroseconds(ttl * 1000000); + Env::Default()->SleepForMicroseconds((ttl+1) * 1000000); + // Env::Default()->SleepForMicroseconds(ttl * 1000000); db->CompactRange(nullptr, nullptr); From 086354f9908af2bd6c1e2ee48ec0e019f79df9db Mon Sep 17 00:00:00 2001 From: cyq <1056374449@qq.com> Date: Wed, 30 Oct 2024 01:22:37 +0800 Subject: [PATCH 5/6] =?UTF-8?q?=E6=96=87=E4=BB=B6=E5=85=83=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E5=A2=9E=E5=8A=A0deadtime=E7=9A=84=E8=8C=83=E5=9B=B4?= =?UTF-8?q?=EF=BC=8C=E5=B9=B6=E6=A0=B9=E6=8D=AE=E8=8C=83=E5=9B=B4=E5=AF=B9?= =?UTF-8?q?=E6=96=87=E4=BB=B6=E5=9C=A8=E5=90=88=E5=B9=B6=E6=97=B6=E5=80=99?= =?UTF-8?q?=E8=BF=9B=E8=A1=8C=E6=B8=85=E9=99=A4=EF=BC=8C=E5=8F=AF=E4=BB=A5?= =?UTF-8?q?=E8=A7=A3=E5=86=B3=E6=89=8B=E5=8A=A8=E5=90=88=E5=B9=B6=E4=B8=8D?= =?UTF-8?q?=E5=AE=8C=E5=85=A8=E7=9A=84=E6=83=85=E5=86=B5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- db/db_impl.cc | 31 ++++++++++++++++++++++++++++++- db/db_impl.h | 1 + db/version_set.h | 2 ++ 3 files changed, 33 insertions(+), 1 deletion(-) diff --git a/db/db_impl.cc b/db/db_impl.cc index 28ef225..5d2d9e7 100644 --- a/db/db_impl.cc +++ b/db/db_impl.cc @@ -591,7 +591,7 @@ void DBImpl::CompactRange(const Slice* begin, const Slice* end) { } } } - max_level_with_files = config::kNumLevels - 1; //TODO:强制合并所有level中的sst,但是这么做不是很优雅 + // max_level_with_files = config::kNumLevels - 1; //TODO:强制合并所有level中的sst,但是这么做不是很优雅 TEST_CompactMemTable(); // TODO(sanjay): Skip if memtable does not overlap for (int level = 0; level < max_level_with_files; level++) { TEST_CompactRange(level, begin, end); @@ -706,6 +706,31 @@ void DBImpl::BackgroundCall() { background_work_finished_signal_.SignalAll(); } +bool DBImpl::RemoveExpireTable() { + bool remove = false; + VersionEdit edit; + time_t nowTime; + time(&nowTime); + Version *base = versions_->current(); + base->Ref(); + for(int level = 0; level < config::kNumLevels; level ++) { + const std::vector &files = versions_->current()->Files(level); + for(auto meta:files) { + if(meta->largest_deadtime < nowTime) { + remove = true; + edit.RemoveFile(level,meta->number); + std::cout<<"remove file : "<number<<" from level : "<LogAndApply(&edit,&mutex_); + RemoveObsoleteFiles(); + } + base->Unref(); + return remove; +} + void DBImpl::BackgroundCompaction() { mutex_.AssertHeld(); @@ -714,6 +739,10 @@ void DBImpl::BackgroundCompaction() { return; } + if(RemoveExpireTable()) { + return; + } + Compaction* c; bool is_manual = (manual_compaction_ != nullptr); InternalKey manual_end; diff --git a/db/db_impl.h b/db/db_impl.h index ef81411..92ea869 100644 --- a/db/db_impl.h +++ b/db/db_impl.h @@ -145,6 +145,7 @@ class DBImpl : public DB { EXCLUSIVE_LOCKS_REQUIRED(mutex_); Status DoCompactionWork(CompactionState* compact) EXCLUSIVE_LOCKS_REQUIRED(mutex_); + bool RemoveExpireTable(); //if remove some table thne return true otherwise return false Status OpenCompactionOutputFile(CompactionState* compact); Status FinishCompactionOutputFile(CompactionState* compact, Iterator* input); diff --git a/db/version_set.h b/db/version_set.h index ea0c925..6b804f5 100644 --- a/db/version_set.h +++ b/db/version_set.h @@ -111,6 +111,8 @@ class Version { int NumFiles(int level) const { return files_[level].size(); } + const std::vector& Files(int level) const {return files_[level]; } + // Return a human readable string that describes this version's contents. std::string DebugString() const; From 7575417f53d028dc0b50853bbc530ad28533d7dd Mon Sep 17 00:00:00 2001 From: cyq <1056374449@qq.com> Date: Wed, 30 Oct 2024 10:53:24 +0800 Subject: [PATCH 6/6] =?UTF-8?q?=E5=B0=8F=E6=94=B9=E5=8A=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- db/db_impl.cc | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/db/db_impl.cc b/db/db_impl.cc index 5d2d9e7..5a56c16 100644 --- a/db/db_impl.cc +++ b/db/db_impl.cc @@ -719,13 +719,13 @@ bool DBImpl::RemoveExpireTable() { if(meta->largest_deadtime < nowTime) { remove = true; edit.RemoveFile(level,meta->number); - std::cout<<"remove file : "<number<<" from level : "<number<<" from level : "<largest_deadtime<LogAndApply(&edit,&mutex_); - RemoveObsoleteFiles(); + // RemoveObsoleteFiles(); } base->Unref(); return remove;