trueog/KeyDB
2
0 Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor towards new design

Browse Source
This commit is contained in:
Malavan Sotheeswaran committed 2023-05-23 20:47:15 -07:00
1 parent 90fc43b849
commit edae100767
4 files changed
+57 -61
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/storage/rocksdb.cpp
+29 -45
View File
@@ -33,8 +33,8 @@ std::string prefixKey(const char *key, size_t cchKey)
return FInternalKey(key, cchKey) ? std::string(key, cchKey) : getPrefix(keyHashSlot(key, cchKey)) + std::string(key, cchKey);
}
RocksDBStorageProvider::RocksDBStorageProvider(RocksDBStorageFactory *pfactory, std::shared_ptr<rocksdb::DB> &spdb, std::shared_ptr<rocksdb::ColumnFamilyHandle> &spcolfam, const rocksdb::Snapshot *psnapshot, size_t count)
: m_pfactory(pfactory), m_spdb(spdb), m_psnapshot(psnapshot), m_spcolfamily(spcolfam), m_count(count)
RocksDBStorageProvider::RocksDBStorageProvider(RocksDBStorageFactory *pfactory, std::shared_ptr<rocksdb::DB> &spdb, std::shared_ptr<rocksdb::ColumnFamilyHandle> &spcolfam, std::shared_ptr<rocksdb::ColumnFamilyHandle> &spexpirecolfam, const rocksdb::Snapshot *psnapshot, size_t count)
: m_pfactory(pfactory), m_spdb(spdb), m_psnapshot(psnapshot), m_spcolfamily(spcolfam), m_spexpirecolfamily(spexpirecolfam), m_count(count)
{
m_readOptionsTemplate = rocksdb::ReadOptions();
m_readOptionsTemplate.verify_checksums = false;
@@ -211,56 +211,40 @@ bool RocksDBStorageProvider::enumerate_hashslot(callback fn, unsigned int hashsl
return full_iter;
}
std::vector<std::string> RocksDBStorageProvider::getExpirationCandidates() override
std::vector<std::string> RocksDBStorageProvider::getExpirationCandidates()
{
std::vector<std::string> result;
std::string slice;
auto status = m_spdb->Get(ReadOptions(), m_spcolfamily.get(), rocksdb::Slice(last_expire_key, sizeof(last_expire_key)), &slice);
long long lastTime = std::stoll(slice);
long long curTime = ustime() << TIME_BUCKET_SHIFT;
if (status.ok()) {
std::string bucketString;
int i;
for (i = lastTime; i < curTime; i++) {
std::string bucketString = INTERNAL_KEY_PREFIX + std::to_string(i);
status = m_spdb->Get(ReadOptions(), m_spcolfamily.get(), rocksdb::Slice(bucketString), &slice);
if (status.ok()) {
result += deserialize(slice);
if (m_spbatch != nullptr)
{
status = m_spbatch->Delete(m_spcolfamily.get(), rocksdb::Slice(bucketString));
}
else
{
status = m_spdb->Delete(WriteOptions(), m_spcolfamily.get(), rocksdb::Slice(bucketString));
}
}
}
status = m_spdb->Put(WriteOptions(), m_spcolfamily.get(), rocksdb::Slice(last_expire_key, sizeof(last_expire_key)), rocksdb::Slice(std::to_string(i)));
if (!status.ok()) {
assert("failed to update flash expire last time");
}
std::unique_ptr<rocksdb::Iterator> it = std::unique_ptr<rocksdb::Iterator>(m_spdb->NewIterator(ReadOptions(), m_spexpirecolfamily.get()));
for (it->SeekToFirst(); it->Valid() && result.size() < 16; it->Next()) {
if (FInternalKey(it->key().data(), it->key().size()))
continue;
result.emplace(it->value().data(), it->value().size());
if (m_spbatch != nullptr)
status = m_spbatch->Delete(m_spexpirecolfamily.get(), it->key());
else
status = m_spdb->Delete(WriteOptions(), m_spexpirecolfamily.get(), it->key());
if (!status.ok())
throw "Failed to remove expired key from rocksdb"
}
return result;
}
std::vector<std::string> RocksDBStorageProvider::getEvictionCandidates() override
std::vector<std::string> RocksDBStorageProvider::getEvictionCandidates()
{
std::vector<std::string> result;
std::string slice;
auto status = m_spdb->Get(ReadOptions(), m_spcolfamily.get(), rocksdb::Slice(last_expire_key, sizeof(last_expire_key)), &slice);
long long lastTime = std::stoll(slice);
long long curTime = ustime() << TIME_BUCKET_SHIFT;
if (status.ok()) {
std::string bucketString;
int i = lastTime;
while (result.empty()) {
std::string bucketString = INTERNAL_KEY_PREFIX + std::to_string(i);
status = m_spdb->Get(ReadOptions(), m_spcolfamily.get(), rocksdb::Slice(bucketString), &slice);
if (status.ok()) {
result += deserialize(slice);
}
i++;
if (g_pserver->maxmemory_policy & MAXMEMORY_FLAG_ALLKEYS) {
std::unique_ptr<rocksdb::Iterator> it = std::unique_ptr<rocksdb::Iterator>(m_spdb->NewIterator(ReadOptions(), m_spcolfamily.get()));
for (it->Seek(randomHashSlot()); it->Valid() && result.size() < 16; it->Next()) {
if (FInternalKey(it->key().data(), it->key().size()))
continue;
result.emplace(it->value().data(), it->value().size());
}
} else {
std::unique_ptr<rocksdb::Iterator> it = std::unique_ptr<rocksdb::Iterator>(m_spdb->NewIterator(ReadOptions(), m_spexpirecolfamily.get()));
for (it->SeekToFirst(); it->Valid() && result.size() < 16; it->Next()) {
if (FInternalKey(it->key().data(), it->key().size()))
continue;
result.emplace(it->value().data(), it->value().size());
}
}
return result;
@@ -270,7 +254,7 @@ const IStorage *RocksDBStorageProvider::clone() const
{
std::unique_lock<fastlock> l(m_lock);
const rocksdb::Snapshot *psnapshot = const_cast<RocksDBStorageProvider*>(this)->m_spdb->GetSnapshot();
return new RocksDBStorageProvider(m_pfactory, const_cast<RocksDBStorageProvider*>(this)->m_spdb, const_cast<RocksDBStorageProvider*>(this)->m_spcolfamily, psnapshot, m_count);
return new RocksDBStorageProvider(m_pfactory, const_cast<RocksDBStorageProvider*>(this)->m_spdb, const_cast<RocksDBStorageProvider*>(this)->m_spcolfamily, const_cast<RocksDBStorageProvider*>(this)->m_spexpirecolfamily, psnapshot, m_count);
}
RocksDBStorageProvider::~RocksDBStorageProvider()
src/storage/rocksdb.h
+3 -2
View File
@@ -10,7 +10,7 @@
static const char count_key[] = INTERNAL_KEY_PREFIX "__keydb__count\1";
static const char version_key[] = INTERNAL_KEY_PREFIX "__keydb__version\1";
static const char meta_key[] = INTERNAL_KEY_PREFIX "__keydb__metadata\1";
static const char last_expire_key[] = INTERNAL_KEY_PREFIX "__keydb__last_expire_time"
static const char last_expire_key[] = INTERNAL_KEY_PREFIX "__keydb__last_expire_time";
class RocksDBStorageFactory;
class RocksDBStorageProvider : public IStorage
@@ -20,12 +20,13 @@ class RocksDBStorageProvider : public IStorage
std::unique_ptr<rocksdb::WriteBatchWithIndex> m_spbatch;
const rocksdb::Snapshot *m_psnapshot = nullptr;
std::shared_ptr<rocksdb::ColumnFamilyHandle> m_spcolfamily;
std::shared_ptr<rocksdb::ColumnFamilyHandle> m_spexpirecolfamily;
rocksdb::ReadOptions m_readOptionsTemplate;
size_t m_count = 0;
mutable fastlock m_lock {"RocksDBStorageProvider"};
public:
RocksDBStorageProvider(RocksDBStorageFactory *pfactory, std::shared_ptr<rocksdb::DB> &spdb, std::shared_ptr<rocksdb::ColumnFamilyHandle> &spcolfam, const rocksdb::Snapshot *psnapshot, size_t count);
RocksDBStorageProvider(RocksDBStorageFactory *pfactory, std::shared_ptr<rocksdb::DB> &spdb, std::shared_ptr<rocksdb::ColumnFamilyHandle> &spcolfam, std::shared_ptr<rocksdb::ColumnFamilyHandle> &spexpirecolfam, const rocksdb::Snapshot *psnapshot, size_t count);
~RocksDBStorageProvider();
virtual void insert(const char *key, size_t cchKey, void *data, size_t cb, bool fOverwrite) override;
src/storage/rocksdbfactor_internal.h
+1
View File
@@ -5,6 +5,7 @@ class RocksDBStorageFactory : public IStorageFactory
{
std::shared_ptr<rocksdb::DB> m_spdb; // Note: This must be first so it is deleted last
std::vector<std::unique_ptr<rocksdb::ColumnFamilyHandle>> m_vecspcols;
std::vector<std::unique_ptr<rocksdb::ColumnFamilyHandle>> m_vecspexpirecols;
std::shared_ptr<rocksdb::SstFileManager> m_pfilemanager;
std::string m_path;
bool m_fCreatedTempFolder = false;
src/storage/rocksdbfactory.cpp
+24 -14
View File
@@ -78,6 +78,7 @@ RocksDBStorageFactory::RocksDBStorageFactory(const char *dbfile, int dbnum, cons
rocksdb::ColumnFamilyOptions cf_options(options);
cf_options.level_compaction_dynamic_level_bytes = true;
veccoldesc.push_back(rocksdb::ColumnFamilyDescriptor(std::to_string(idb), cf_options));
veccoldesc.push_back(rocksdb::ColumnFamilyDescriptor(std::to_string(idb) + "_expires", cf_options));
}
if (rgchConfig != nullptr)
@@ -97,24 +98,32 @@ RocksDBStorageFactory::RocksDBStorageFactory(const char *dbfile, int dbnum, cons
throw status.ToString();
m_spdb = std::shared_ptr<rocksdb::DB>(db);
rocksdb::ColumnFamilyDescriptor *desc = nullptr;
for (auto handle : handles)
{
std::string strVersion;
auto status = m_spdb->Get(rocksdb::ReadOptions(), handle, rocksdb::Slice(version_key, sizeof(version_key)), &strVersion);
if (!status.ok())
{
setVersion(handle);
if (!handle->GetDescriptor(desc).ok()) {
throw "Error loading rocksdb handles";
}
else
{
SymVer ver = parseVersion(strVersion.c_str());
auto cmp = compareVersion(&ver);
if (cmp == NewerVersion)
throw "Cannot load FLASH database created by newer version of KeyDB";
if (cmp == OlderVersion)
if (!strncmp(desc->name.substr(desc->name.size() - 7).c_str(), "expires", 7)) {
m_vecspexpirecols.emplace_back(handle);
} else {
std::string strVersion;
auto status = m_spdb->Get(rocksdb::ReadOptions(), handle, rocksdb::Slice(version_key, sizeof(version_key)), &strVersion);
if (!status.ok())
{
setVersion(handle);
}
else
{
SymVer ver = parseVersion(strVersion.c_str());
auto cmp = compareVersion(&ver);
if (cmp == NewerVersion)
throw "Cannot load FLASH database created by newer version of KeyDB";
if (cmp == OlderVersion)
setVersion(handle);
}
m_vecspcols.emplace_back(handle);
}
m_vecspcols.emplace_back(handle);
}
}
@@ -155,6 +164,7 @@ IStorage *RocksDBStorageFactory::create(int db, key_load_iterator iter, void *pr
{
++db; // skip default col family
std::shared_ptr<rocksdb::ColumnFamilyHandle> spcolfamily(m_vecspcols[db].release());
std::shared_ptr<rocksdb::ColumnFamilyHandle> spexpirecolfamily(m_vecspexpirecols[db].release());
size_t count = 0;
bool fUnclean = false;
@@ -191,7 +201,7 @@ IStorage *RocksDBStorageFactory::create(int db, key_load_iterator iter, void *pr
++count;
}
}
return new RocksDBStorageProvider(this, m_spdb, spcolfamily, nullptr, count);
return new RocksDBStorageProvider(this, m_spdb, spcolfamily, spexpirecolfamily, nullptr, count);
}
const char *RocksDBStorageFactory::name() const