Kaynağa Gözat

实现memtable的TTL功能

pull/2/head
jiyeoniya 3 hafta önce
ebeveyn
işleme
14e9383694
5 değiştirilmiş dosya ile 137 ekleme ve 49 silme
  1. +9
    -9
      db/db_impl.cc
  2. +99
    -31
      db/memtable.cc
  3. +1
    -0
      db/memtable.h
  4. +19
    -1
      db/write_batch.cc
  5. +9
    -8
      test/ttl_test.cc

+ 9
- 9
db/db_impl.cc Dosyayı Görüntüle

@ -1122,17 +1122,17 @@ Status DBImpl::Get(const ReadOptions& options, const Slice& key,
Status s;
MutexLock l(&mutex_);
SequenceNumber snapshot;
if (options.snapshot != nullptr) {
if (options.snapshot != nullptr) {//判断读取选项是否包含用户提供的快照。如果有,则用快照中的序列号。燕
snapshot =
static_cast<const SnapshotImpl*>(options.snapshot)->sequence_number();
} else {
snapshot = versions_->LastSequence();
snapshot = versions_->LastSequence(); //否则,使用当前最新的序列号。燕
}
MemTable* mem = mem_;
MemTable* imm = imm_;
Version* current = versions_->current();
mem->Ref();
mem->Ref(); //增加引用计数以确保这些对象在方法执行过程中不会被删除 燕
if (imm != nullptr) imm->Ref();
current->Ref();
@ -1144,24 +1144,24 @@ 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)) { //尝试从当前 memtable 中获取键的值。如果找到,则直接返回!! 燕
// Done
} else if (imm != nullptr && imm->Get(lkey, value, &s)) {
} else if (imm != nullptr && imm->Get(lkey, value, &s)) { //memtable没有,去imm找!! 燕
// Done
} else {
s = current->Get(options, lkey, value, &stats);
s = current->Get(options, lkey, value, &stats); //最后去sstable找!!燕
have_stat_update = true;
}
mutex_.Lock();
}
if (have_stat_update && current->UpdateStats(stats)) {
MaybeScheduleCompaction();
MaybeScheduleCompaction(); //如果有统计更新,则检查是否需要触发合并,燕
}
mem->Unref();
if (imm != nullptr) imm->Unref();
current->Unref();
return s;
current->Unref(); //减少引用计数,允许释放内存
return s; //返回状态对象 s,告知调用方是否成功找到值
}
Iterator* DBImpl::NewIterator(const ReadOptions& options) {

+ 99
- 31
db/memtable.cc Dosyayı Görüntüle

@ -8,6 +8,7 @@
#include "leveldb/env.h"
#include "leveldb/iterator.h"
#include "util/coding.h"
#include "include/leveldb/write_batch.h"
namespace leveldb {
@ -99,40 +100,107 @@ void MemTable::Add(SequenceNumber s, ValueType type, const Slice& key,
table_.Insert(buf);
}
// bool MemTable::Get(const LookupKey& key, std::string* value, Status* s) { //燕改
// Slice memkey = key.memtable_key();
// Table::Iterator iter(&table_);
// iter.Seek(memkey.data());
// if (iter.Valid()) {// 获取跳表项的内容
// // entry format is:
// // klength varint32
// // userkey char[klength]
// // tag uint64
// // vlength varint32
// // value char[vlength]
// // Check that it belongs to same user key. We do not check the
// // sequence number since the Seek() call above should have skipped
// // all entries with overly large sequence numbers.
// const char* entry = iter.key();
// uint32_t key_length;
// const char* key_ptr = GetVarint32Ptr(entry, entry + 5, &key_length);
// if (comparator_.comparator.user_comparator()->Compare(// 比较键值
// Slice(key_ptr, key_length - 8), key.user_key()) == 0) {
// // Correct user key
// const uint64_t tag = DecodeFixed64(key_ptr + key_length - 8);
// switch (static_cast<ValueType>(tag & 0xff)) {
// case kTypeValue: {
// // Slice v = GetLengthPrefixedSlice(key_ptr + key_length);
// // value->assign(v.data(), v.size());
// // return true;
// // 获取过期时间戳,燕改
// Slice v = GetLengthPrefixedSlice(key_ptr + key_length);
// uint64_t expire_time = DecodeFixed64(v.data() + v.size() - sizeof(uint64_t));
// // 检查是否已过期
// uint64_t current_time = Env::Default()->NowMicros() / 1000000; // 当前时间(秒)
// if (expire_time > 0 && expire_time < current_time) {
// *s = Status::NotFound("Key has expired"); // 已过期
// return false;
// }
// // 未过期,返回值
// value->assign(v.data() + 8, v.size() - 8); // 去除前8字节的时间戳
// return true;
// }
// case kTypeDeletion:
// *s = Status::NotFound(Slice());
// return true;
// }
// }
// }
// return false;
// }
bool MemTable::Get(const LookupKey& key, std::string* value, Status* s) {
Slice memkey = key.memtable_key();
Table::Iterator iter(&table_);
iter.Seek(memkey.data());
if (iter.Valid()) {
// entry format is:
// klength varint32
// userkey char[klength]
// tag uint64
// vlength varint32
// value char[vlength]
// Check that it belongs to same user key. We do not check the
// sequence number since the Seek() call above should have skipped
// all entries with overly large sequence numbers.
const char* entry = iter.key();
uint32_t key_length;
const char* key_ptr = GetVarint32Ptr(entry, entry + 5, &key_length);
if (comparator_.comparator.user_comparator()->Compare(
Slice(key_ptr, key_length - 8), key.user_key()) == 0) {
// Correct user key
const uint64_t tag = DecodeFixed64(key_ptr + key_length - 8);
switch (static_cast<ValueType>(tag & 0xff)) {
case kTypeValue: {
Slice v = GetLengthPrefixedSlice(key_ptr + key_length);
value->assign(v.data(), v.size());
return true;
Slice memkey = key.memtable_key();
Table::Iterator iter(&table_);
iter.Seek(memkey.data());
if (iter.Valid()) { // 获取跳表项的内容
const char* entry = iter.key();
uint32_t key_length;
const char* key_ptr = GetVarint32Ptr(entry, entry + 5, &key_length);
if (comparator_.comparator.user_comparator()->Compare(
Slice(key_ptr, key_length - 8), key.user_key()) == 0) {
const uint64_t tag = DecodeFixed64(key_ptr + key_length - 8);
switch (static_cast<ValueType>(tag & 0xff)) {
case kTypeValue: {
// 获取存储的值和时间戳
Slice v = GetLengthPrefixedSlice(key_ptr + key_length);
std::string combined_str(v.data(), v.size());
// 根据存储格式分离原始值和时间戳
std::string actual_value = combined_str.substr(0, combined_str.size() - 20);
std::string time_str = combined_str.substr(combined_str.size() - 19, 19);
// 获取当前时间(字符串格式)
auto now = std::chrono::system_clock::now();
auto now_time_t = std::chrono::system_clock::to_time_t(now);
std::tm* now_tm = std::localtime(&now_time_t);
char buffer[20];
std::strftime(buffer, sizeof(buffer), "%Y-%m-%d %H:%M:%S", now_tm);
std::string current_time_str(buffer);
// 检查过期
if (time_str <= current_time_str) {
*s = Status::NotFound("Key has expired"); // 已过期
return true;
}
// 未过期,返回实际值
value->assign(actual_value);
return true;
}
case kTypeDeletion:
*s = Status::NotFound(Slice());
return true;
}
}
case kTypeDeletion:
*s = Status::NotFound(Slice());
return true;
}
}
}
return false;
return false;
}
} // namespace leveldb

+ 1
- 0
db/memtable.h Dosyayı Görüntüle

@ -60,6 +60,7 @@ class MemTable {
// If memtable contains a deletion for key, store a NotFound() error
// in *status and return true.
// Else, return false.
std::chrono::system_clock::time_point insert_time; //
bool Get(const LookupKey& key, std::string* value, Status* s);
private:

+ 19
- 1
db/write_batch.cc Dosyayı Görüntüle

@ -23,6 +23,7 @@
#include <sstream> // For std::ostringstream 心
#include <cstdint> // 引入uint64_t类型,燕
#include "leveldb/env.h" //燕
namespace leveldb {
// WriteBatch header has an 8-byte sequence number followed by a 4-byte count.
@ -104,17 +105,34 @@ void WriteBatch::Put(const Slice& key, const Slice& value) {
PutLengthPrefixedSlice(&rep_, value);
}
// void WriteBatch::Put(const Slice& key, const Slice& value, std::uint64_t ttl) {
// WriteBatchInternal::SetCount(this, WriteBatchInternal::Count(this) + 1);
// rep_.push_back(static_cast<char>(kTypeValue));
// PutLengthPrefixedSlice(&rep_, key);
// // 获取当前时间并计算过期时间
// uint64_t current_time = Env::Default()->NowMicros() / 1000000; // 当前时间(秒)
// uint64_t expire_time = current_time + ttl; // 计算未来的过期时间
// // 将过期时间编码为64位整数
// rep_.append(reinterpret_cast<const char*>(&expire_time), sizeof(expire_time)); // 添加过期时间戳
// // 添加原始值
// PutLengthPrefixedSlice(&rep_, value);
// }
void WriteBatch::Put(const Slice& key, const Slice& value, std::uint64_t ttl) {
WriteBatchInternal::SetCount(this, WriteBatchInternal::Count(this) + 1);
rep_.push_back(static_cast<char>(kTypeValue));
PutLengthPrefixedSlice(&rep_, key);
// 获取当前时间
auto now = std::chrono::system_clock::now();
// 加上ttl
auto future_time = now + std::chrono::seconds(ttl);
// 转换为 time_t
std::time_t future_time_t = std::chrono::system_clock::to_time_t(future_time);

+ 9
- 8
test/ttl_test.cc Dosyayı Görüntüle

@ -20,11 +20,11 @@ Status OpenDB(std::string dbName, DB **db) {
void InsertData(DB *db, uint64_t ttl/* second */) {
WriteOptions writeOptions;
int key_num = data_size / value_size;
srand(static_cast<unsigned int>(time(0)));
srand(0);
for (int i = 0; i < key_num; i++) {
//int key_ = rand() % key_num+1;
int key_ = i % key_num+1;
int key_ = rand() % key_num+1;
// int key_ = i % key_num+1;
std::string key = std::to_string(key_);
std::string value(value_size, 'a');
db->Put(writeOptions, key, value, ttl);
@ -36,7 +36,8 @@ void GetData(DB *db, int size = (1 << 30)) {
int key_num = data_size / value_size;
// 使用随机种子生成随机键进行点查(单次查询),燕
srand(static_cast<unsigned int>(time(0)));
// 点查
srand(0);
for (int i = 0; i < 100; i++) {
int key_ = rand() % key_num+1;
//int key_ = i % key_num+1;
@ -53,17 +54,17 @@ TEST(TestTTL, ReadTTL) {
abort();
}
uint64_t ttl = 20;
uint64_t ttl = 200;
InsertData(db, ttl);
ReadOptions readOptions;
Status status;
int key_num = data_size / value_size;
srand(static_cast<unsigned int>(time(0)));
srand(0);
for (int i = 0; i < 100; i++) {
//int key_ = rand() % key_num+1;
int key_ = i % key_num+1;
int key_ = rand() % key_num+1;
// int key_ = i % key_num+1;
std::string key = std::to_string(key_);
std::string value;
status = db->Get(readOptions, key, &value);

Yükleniyor…
İptal
Kaydet