Browse Source

Merge branch 'ld' into cyq

pull/1/head
cyq 2 months ago
parent
commit
80cf080342
15 changed files with 248 additions and 60 deletions
  1. +4
    -0
      CMakeLists.txt
  2. +1
    -1
      db/c.cc
  3. +10
    -5
      db/db_impl.cc
  4. +0
    -2
      db/db_impl.h
  5. +3
    -3
      db/db_test.cc
  6. +14
    -1
      db/dbformat.cc
  7. +27
    -10
      db/dbformat.h
  8. +1
    -1
      db/dumpfile.cc
  9. +39
    -4
      db/memtable.cc
  10. +1
    -1
      db/memtable.h
  11. +36
    -6
      db/write_batch.cc
  12. +3
    -3
      include/leveldb/db.h
  13. +3
    -2
      include/leveldb/write_batch.h
  14. +85
    -0
      test/ttl_mmtable_test.cc
  15. +21
    -21
      test/ttl_test.cc

+ 4
- 0
CMakeLists.txt View File

@ -524,6 +524,10 @@ add_executable(db_test2
)
target_link_libraries(db_test2 PRIVATE leveldb)
add_executable(ttl_mmtable_test
"${PROJECT_SOURCE_DIR}/test/ttl_mmtable_test.cc"
)
target_link_libraries(ttl_mmtable_test PRIVATE leveldb)
add_executable(ttl_test
"${PROJECT_SOURCE_DIR}/test/ttl_test.cc"

+ 1
- 1
db/c.cc View File

@ -349,7 +349,7 @@ void leveldb_writebatch_iterate(const leveldb_writebatch_t* b, void* state,
void* state_;
void (*put_)(void*, const char* k, size_t klen, const char* v, size_t vlen);
void (*deleted_)(void*, const char* k, size_t klen);
void Put(const Slice& key, const Slice& value) override {
void Put(const Slice& key, const Slice& value, uint64_t deadTime) override {
(*put_)(state_, key.data(), key.size(), value.data(), value.size());
}
void Delete(const Slice& key) override {

+ 10
- 5
db/db_impl.cc View File

@ -1145,10 +1145,13 @@ Status DBImpl::Get(const ReadOptions& options, const Slice& key,
// First look in the memtable, then in the immutable memtable (if any).
LookupKey lkey(key, snapshot);
if (mem->Get(lkey, value, &s)) {
printf("in mem\n");
// Done
} else if (imm != nullptr && imm->Get(lkey, value, &s)) {
printf("in immem\n");
// Done
} else {
printf("in current\n");
s = current->Get(options, lkey, value, &stats);
have_stat_update = true;
}
@ -1194,8 +1197,9 @@ void DBImpl::ReleaseSnapshot(const Snapshot* snapshot) {
}
// Convenience methods
Status DBImpl::Put(const WriteOptions& o, const Slice& key, const Slice& val) {
return DB::Put(o, key, val);
Status DBImpl::Put(const WriteOptions& o, const Slice& key,
const Slice& val, uint64_t ttl) {
return DB::Put(o, key, val, ttl);
}
Status DBImpl::Put(const WriteOptions& options, const Slice& key,
@ -1237,7 +1241,7 @@ Status DBImpl::Write(const WriteOptions& options, WriteBatch* updates) {
// into mem_.
{
mutex_.Unlock();
status = log_->AddRecord(WriteBatchInternal::Contents(write_batch));
status = log_->AddRecord(WriteBatchInternal::Contents(write_batch)); //这里会有影响吗
bool sync_error = false;
if (status.ok() && options.sync) {
status = logfile_->Sync();
@ -1490,9 +1494,10 @@ void DBImpl::GetApproximateSizes(const Range* range, int n, uint64_t* sizes) {
// Default implementations of convenience methods that subclasses of DB
// can call if they wish
Status DB::Put(const WriteOptions& opt, const Slice& key, const Slice& value) {
Status DB::Put(const WriteOptions& opt, const Slice& key,
const Slice& value, uint64_t ttl) {
WriteBatch batch;
batch.Put(key, value);
batch.Put(key, value, ttl);
return Write(opt, &batch);
}

+ 0
- 2
db/db_impl.h View File

@ -37,8 +37,6 @@ 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;

+ 3
- 3
db/db_test.cc View File

@ -2114,8 +2114,8 @@ class ModelDB : public DB {
explicit ModelDB(const Options& options) : options_(options) {}
~ModelDB() override = default;
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 = 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);
@ -2152,7 +2152,7 @@ class ModelDB : public DB {
class Handler : public WriteBatch::Handler {
public:
KVMap* map_;
void Put(const Slice& key, const Slice& value) override {
void Put(const Slice& key, const Slice& value, uint64_t deadTime = 0) override {
(*map_)[key.ToString()] = value.ToString();
}
void Delete(const Slice& key) override { map_->erase(key.ToString()); }

+ 14
- 1
db/dbformat.cc View File

@ -12,6 +12,12 @@
namespace leveldb {
static uint64_t PackSequenceAndTypeAndTtl(uint64_t seq, ValueType t, bool havettl) {
assert(seq <= kMaxSequenceNumber);
assert(t <= kValueTypeForSeek);
return (seq << 8) | (havettl << 1) | t;
}
static uint64_t PackSequenceAndType(uint64_t seq, ValueType t) {
assert(seq <= kMaxSequenceNumber);
assert(t <= kValueTypeForSeek);
@ -20,7 +26,10 @@ static uint64_t PackSequenceAndType(uint64_t seq, ValueType t) {
void AppendInternalKey(std::string* result, const ParsedInternalKey& key) {
result->append(key.user_key.data(), key.user_key.size());
PutFixed64(result, PackSequenceAndType(key.sequence, key.type));
if(key.deadTime != 0)
PutFixed64(result, key.deadTime);
PutFixed64(result, PackSequenceAndTypeAndTtl(
key.sequence, key.type, (key.deadTime != 0)));
}
std::string ParsedInternalKey::DebugString() const {
@ -49,6 +58,10 @@ int InternalKeyComparator::Compare(const Slice& akey, const Slice& bkey) const {
// increasing user key (according to user-supplied comparator)
// decreasing sequence number
// decreasing type (though sequence# should be enough to disambiguate)
Slice user_akey = ExtractUserKey(akey);
Slice user_bkey = ExtractUserKey(bkey);
std::string a = user_akey.ToString();
std::string b = user_bkey.ToString();
int r = user_comparator_->Compare(ExtractUserKey(akey), ExtractUserKey(bkey));
if (r == 0) {
const uint64_t anum = DecodeFixed64(akey.data() + akey.size() - 8);

+ 27
- 10
db/dbformat.h View File

@ -16,6 +16,7 @@
#include "leveldb/table_builder.h"
#include "util/coding.h"
#include "util/logging.h"
#include "iostream"
namespace leveldb {
@ -69,17 +70,19 @@ static const SequenceNumber kMaxSequenceNumber = ((0x1ull << 56) - 1);
struct ParsedInternalKey {
Slice user_key;
SequenceNumber sequence;
uint64_t deadTime;
ValueType type;
ParsedInternalKey() {} // Intentionally left uninitialized (for speed)
ParsedInternalKey(const Slice& u, const SequenceNumber& seq, ValueType t)
: user_key(u), sequence(seq), type(t) {}
ParsedInternalKey(const Slice& u, const SequenceNumber& seq,
ValueType t, uint64_t d = 0)
: user_key(u), sequence(seq), type(t), deadTime(d) {}
std::string DebugString() const;
};
// Return the length of the encoding of "key".
inline size_t InternalKeyEncodingLength(const ParsedInternalKey& key) {
return key.user_key.size() + 8;
return key.user_key.size() + 8 + (key.deadTime != 0) * 8;
}
// Append the serialization of "key" to *result.
@ -94,7 +97,12 @@ bool ParseInternalKey(const Slice& internal_key, ParsedInternalKey* result);
// Returns the user key portion of an internal key.
inline Slice ExtractUserKey(const Slice& internal_key) {
assert(internal_key.size() >= 8);
return Slice(internal_key.data(), internal_key.size() - 8);
uint64_t num = DecodeFixed64(internal_key.data() + internal_key.size() - 8);
uint8_t havettl = (num & 0xff) >> 1;
size_t klen = internal_key.size() - 8;
if(havettl) klen -= 8;
Slice user_key = Slice(internal_key.data(), klen);
return user_key;
}
// A comparator for internal keys that uses a specified comparator for
@ -137,8 +145,9 @@ class InternalKey {
public:
InternalKey() {} // Leave rep_ as empty to indicate it is invalid
InternalKey(const Slice& user_key, SequenceNumber s, ValueType t) {
AppendInternalKey(&rep_, ParsedInternalKey(user_key, s, t));
InternalKey(const Slice& user_key, SequenceNumber s,
ValueType t, uint64_t deadTime = 0) {
AppendInternalKey(&rep_, ParsedInternalKey(user_key, s, t, deadTime));
}
bool DecodeFrom(const Slice& s) {
@ -172,11 +181,19 @@ inline bool ParseInternalKey(const Slice& internal_key,
ParsedInternalKey* result) {
const size_t n = internal_key.size();
if (n < 8) return false;
uint64_t num = DecodeFixed64(internal_key.data() + n - 8);
uint8_t c = num & 0xff;
result->sequence = num >> 8;
uint64_t tag = DecodeFixed64(internal_key.data() + n - 8);
uint8_t c = tag & 0xff;
uint8_t havettl = c >> 1;
assert(havettl <= 0b1);
result->sequence = tag >> 8;
result->type = static_cast<ValueType>(c);
result->user_key = Slice(internal_key.data(), n - 8);
if(havettl){
result->deadTime = DecodeFixed64(internal_key.data() + n - 16);
result->user_key = Slice(internal_key.data(), n - 16);
} else {
result->deadTime = 0;
result->user_key = Slice(internal_key.data(), n - 8);
}
return (c <= static_cast<uint8_t>(kTypeValue));
}

+ 1
- 1
db/dumpfile.cc View File

@ -74,7 +74,7 @@ Status PrintLogContents(Env* env, const std::string& fname,
// Called on every item found in a WriteBatch.
class WriteBatchItemPrinter : public WriteBatch::Handler {
public:
void Put(const Slice& key, const Slice& value) override {
void Put(const Slice& key, const Slice& value, uint64_t deadTime) override {
std::string r = " put '";
AppendEscapedStringTo(&r, key);
r += "' '";

+ 39
- 4
db/memtable.cc View File

@ -8,6 +8,8 @@
#include "leveldb/env.h"
#include "leveldb/iterator.h"
#include "util/coding.h"
#include "ctime"
#include "iostream"
namespace leveldb {
@ -74,16 +76,24 @@ class MemTableIterator : public Iterator {
Iterator* MemTable::NewIterator() { return new MemTableIterator(&table_); }
void MemTable::Add(SequenceNumber s, ValueType type, const Slice& key,
const Slice& value) {
const Slice& value, uint64_t deadTime) {
// Format of an entry is concatenation of:
// key_size : varint32 of internal_key.size()
// key bytes : char[internal_key.size()]
// tag : uint64((sequence << 8) | type)
// value_size : varint32 of value.size()
// value bytes : char[value.size()]
//变成
// key_size : varint32 of internal_key.size()
// key bytes : char[internal_key.size()]
// (deadTime) : uint64(可能有)
// tag : uint64((sequence << 8) | havettl << 1 | type)
// value_size : varint32 of value.size()
// value bytes : char[value.size()]
size_t key_size = key.size();
size_t val_size = value.size();
size_t internal_key_size = key_size + 8;
if(deadTime != 0) internal_key_size += 8;
const size_t encoded_len = VarintLength(internal_key_size) +
internal_key_size + VarintLength(val_size) +
val_size;
@ -91,22 +101,33 @@ void MemTable::Add(SequenceNumber s, ValueType type, const Slice& key,
char* p = EncodeVarint32(buf, internal_key_size);
std::memcpy(p, key.data(), key_size);
p += key_size;
EncodeFixed64(p, (s << 8) | type);
if(deadTime == 0){
EncodeFixed64(p, (s << 8) | type);
} else {
EncodeFixed64(p, deadTime);
p += 8;
EncodeFixed64(p, (s << 8) | 0b10 | type);
}
p += 8;
p = EncodeVarint32(p, val_size);
std::memcpy(p, value.data(), val_size);
assert(p + val_size == buf + encoded_len);
table_.Insert(buf);
std::cout << "insert:" << key.ToString() << std::endl;
}
bool MemTable::Get(const LookupKey& key, std::string* value, Status* s) {
Slice memkey = key.memtable_key();
Table::Iterator iter(&table_);
iter.Seek(memkey.data());
std::cout << "search:" << key.user_key().ToString() << " valid?" << iter.Valid();
if (iter.Valid()) {
// entry format is:
// klength varint32
// userkey char[klength]
// (deadTime) uint64
// tag uint64
// vlength varint32
// value char[vlength]
@ -116,12 +137,26 @@ bool MemTable::Get(const LookupKey& key, std::string* value, Status* s) {
const char* entry = iter.key();
uint32_t key_length;
const char* key_ptr = GetVarint32Ptr(entry, entry + 5, &key_length);
std::cout << " get:" << ExtractUserKey(Slice(key_ptr, key_length)).ToString();
if (comparator_.comparator.user_comparator()->Compare(
Slice(key_ptr, key_length - 8), key.user_key()) == 0) {
ExtractUserKey(Slice(key_ptr, key_length)), key.user_key()) == 0) {
// Correct user key
const uint64_t tag = DecodeFixed64(key_ptr + key_length - 8);
switch (static_cast<ValueType>(tag & 0xff)) {
switch (static_cast<ValueType>(tag & 0x01)) {
case kTypeValue: {
uint8_t havettl = (tag & 0xff) >> 1;
if(havettl){
time_t nowTime;
time(&nowTime);
assert(nowTime > 0);
const uint64_t deadTime = DecodeFixed64(key_ptr + key_length - 16);
if(static_cast<uint64_t>(nowTime) > deadTime){ //过期了
std::cout << nowTime << "dead:" << deadTime << std::endl;
*s = Status::NotFound(Slice());
return true; //todo:之前有没过期的key
}
}
Slice v = GetLengthPrefixedSlice(key_ptr + key_length);
value->assign(v.data(), v.size());
return true;

+ 1
- 1
db/memtable.h View File

@ -54,7 +54,7 @@ class MemTable {
// specified sequence number and with the specified type.
// Typically value will be empty if type==kTypeDeletion.
void Add(SequenceNumber seq, ValueType type, const Slice& key,
const Slice& value);
const Slice& value, uint64_t deadTime = 0);
// If memtable contains a value for key, store it in *value and return true.
// If memtable contains a deletion for key, store a NotFound() error

+ 36
- 6
db/write_batch.cc View File

@ -6,13 +6,20 @@
// sequence: fixed64
// count: fixed32
// data: record[count]
// record :=
// 原来record :=
// kTypeValue varstring varstring |
// kTypeDeletion varstring
// varstring :=
// len: varint32
// data: uint8[len]
// 增加ttl后record :=
// kTypeValue havettl (deadtime) varstring varstring |
// kTypeDeletion varstring
// varstring :=
// len: varint32
// data: uint8[len]
#include "leveldb/write_batch.h"
#include "db/dbformat.h"
@ -20,6 +27,8 @@
#include "db/write_batch_internal.h"
#include "leveldb/db.h"
#include "util/coding.h"
#include "ctime"
#include <cstdint>
namespace leveldb {
@ -54,12 +63,22 @@ Status WriteBatch::Iterate(Handler* handler) const {
input.remove_prefix(1);
switch (tag) {
case kTypeValue:
{
char havettl = input[0];
input.remove_prefix(1);
uint64_t deadTime = 0;
if(havettl){
deadTime = DecodeFixed64(input.data());
input.remove_prefix(8);
}
if (GetLengthPrefixedSlice(&input, &key) &&
GetLengthPrefixedSlice(&input, &value)) {
handler->Put(key, value);
handler->Put(key, value, deadTime);
} else {
return Status::Corruption("bad WriteBatch Put");
}
}
break;
case kTypeDeletion:
if (GetLengthPrefixedSlice(&input, &key)) {
@ -95,9 +114,20 @@ void WriteBatchInternal::SetSequence(WriteBatch* b, SequenceNumber seq) {
EncodeFixed64(&b->rep_[0], seq);
}
void WriteBatch::Put(const Slice& key, const Slice& value) {
void WriteBatch::Put(const Slice& key, const Slice& value, uint64_t ttl) {
WriteBatchInternal::SetCount(this, WriteBatchInternal::Count(this) + 1);
rep_.push_back(static_cast<char>(kTypeValue));
rep_.push_back(static_cast<char>(ttl != 0)); //1:havettl
if(ttl != 0){
time_t nowTime;
time(&nowTime);
assert(nowTime > 0);
assert(ttl > 0);
uint64_t deadTime = static_cast<uint64_t>(nowTime) + ttl;
PutFixed64(&rep_, deadTime);
}
PutLengthPrefixedSlice(&rep_, key);
PutLengthPrefixedSlice(&rep_, value);
}
@ -118,12 +148,12 @@ class MemTableInserter : public WriteBatch::Handler {
SequenceNumber sequence_;
MemTable* mem_;
void Put(const Slice& key, const Slice& value) override {
mem_->Add(sequence_, kTypeValue, key, value);
void Put(const Slice& key, const Slice& value, uint64_t deadTime) override {
mem_->Add(sequence_, kTypeValue, key, value, deadTime);
sequence_++;
}
void Delete(const Slice& key) override {
mem_->Add(sequence_, kTypeDeletion, key, Slice());
mem_->Add(sequence_, kTypeDeletion, key, Slice(), 0);
sequence_++;
}
};

+ 3
- 3
include/leveldb/db.h View File

@ -63,8 +63,8 @@ class LEVELDB_EXPORT DB {
// Set the database entry for "key" to "value". Returns OK on success,
// and a non-OK status on error.
// Note: consider setting options.sync = true.
virtual Status Put(const WriteOptions& options, const Slice& key,
const Slice& value) = 0;
// virtual Status Put(const WriteOptions& options, const Slice& key,
// const Slice& value) = 0;
// Remove the database entry (if any) for "key". Returns OK on
// success, and a non-OK status on error. It is not an error if "key"
@ -149,7 +149,7 @@ class LEVELDB_EXPORT DB {
// ----------------------------For TTL-----------------------------
// key设置ttl
virtual Status Put(const WriteOptions& options, const Slice& key,
const Slice& value, uint64_t ttl) = 0;
const Slice& value, uint64_t ttl = 0) = 0;
};
// Destroy the contents of the specified database.

+ 3
- 2
include/leveldb/write_batch.h View File

@ -25,6 +25,7 @@
#include "leveldb/export.h"
#include "leveldb/status.h"
#include <cstdint>
namespace leveldb {
@ -35,7 +36,7 @@ class LEVELDB_EXPORT WriteBatch {
class LEVELDB_EXPORT Handler {
public:
virtual ~Handler();
virtual void Put(const Slice& key, const Slice& value) = 0;
virtual void Put(const Slice& key, const Slice& value, uint64_t deadTime = 0) = 0;
virtual void Delete(const Slice& key) = 0;
};
@ -48,7 +49,7 @@ class LEVELDB_EXPORT WriteBatch {
~WriteBatch();
// Store the mapping "key->value" in the database.
void Put(const Slice& key, const Slice& value);
void Put(const Slice& key, const Slice& value, uint64_t ttl = 0);
// If the database contains a mapping for "key", erase it. Else do nothing.
void Delete(const Slice& key);

+ 85
- 0
test/ttl_mmtable_test.cc View File

@ -0,0 +1,85 @@
#include "leveldb/env.h"
#include "leveldb/db.h"
#include "ctime"
#include <iostream>
#include <cstdlib>
using namespace leveldb;
constexpr int value_size = 2048;
constexpr int data_size = 4096 << 1;
Status OpenDB(std::string dbName, DB **db) {
Options options;
options.create_if_missing = true;
return DB::Open(options, dbName, db);
}
void InsertData(DB *db, uint64_t ttl/* second */) {
printf("-----inserting-----\n");
Status status;
WriteOptions writeOptions;
int key_num = data_size / value_size;
srand(static_cast<unsigned int>(time(0)));
for (int i = 0; i < key_num; i++) {
//int key_ = rand() % key_num+1;
int key_ = i+1;
std::string key = std::to_string(key_);
std::string value(value_size, 'a');
status = db->Put(writeOptions, key, value, ttl);
assert(status.ok());
}
}
void GetData(DB *db, int size = (1 << 30)) {
ReadOptions readOptions;
int key_num = data_size / value_size;
// 点查
srand(static_cast<unsigned int>(time(0)));
for (int i = 0; i < 100; i++) {
int key_ = rand() % key_num+1;
std::string key = std::to_string(key_);
std::string value;
db->Get(readOptions, key, &value);
}
}
int main(int argc, char** argv) {
DB *db;
if(OpenDB("testdb", &db).ok() == false) {
std::cerr << "open db failed" << std::endl;
abort();
}
uint64_t ttl = 3;
InsertData(db, ttl);
printf("-----seeking-----\n");
ReadOptions readOptions;
Status status;
int key_num = data_size / value_size;
srand(static_cast<unsigned int>(time(0)));
for (int i = 0; i < key_num; i++) {
//int key_ = rand() % key_num+1;
int key_ = i+1;
std::string key = std::to_string(key_);
std::string value;
status = db->Get(readOptions, key, &value);
assert(status.ok());
}
Env::Default()->SleepForMicroseconds(ttl * 1000000);
for (int i = 0; i < key_num; i++) {
int key_ = rand() % key_num+1;
std::string key = std::to_string(key_);
std::string value;
status = db->Get(readOptions, key, &value);
assert(status.IsNotFound());
}
printf("success!\n");
}

+ 21
- 21
test/ttl_test.cc View File

@ -78,33 +78,33 @@ TEST(TestTTL, ReadTTL) {
}
}
TEST(TestTTL, CompactionTTL) {
DB *db;
// TEST(TestTTL, CompactionTTL) {
// DB *db;
if(OpenDB("testdb", &db).ok() == false) {
std::cerr << "open db failed" << std::endl;
abort();
}
// if(OpenDB("testdb", &db).ok() == false) {
// std::cerr << "open db failed" << std::endl;
// abort();
// }
uint64_t ttl = 20;
InsertData(db, ttl);
// uint64_t ttl = 20;
// InsertData(db, ttl);
leveldb::Range ranges[1];
ranges[0] = leveldb::Range("-", "A");
uint64_t sizes[1];
db->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);
// }
int main(int argc, char** argv) {

Loading…
Cancel
Save