Goddess of Justice DB, the database used for storage on IzaroDFS
Você não pode selecionar mais de 25 tópicos Os tópicos devem começar com uma letra ou um número, podem incluir traços ('-') e podem ter até 35 caracteres.
 
 
 

221 linhas
5.1 KiB

#pragma once
#include "fsized_map.h"
#include <chrono>
struct metadata_t{
bitops::regulated<uint64_t> record_cnt;
bitops::regulated<uint64_t> page_cnt;
bitops::regulated<uint64_t> delete_cnt;
bitops::regulated<uint64_t> last_page;
bitops::regulated<uint64_t> last_delete;
};
class database {
database(){}
public:
mmap_array<std::pair<bitops::regulated<uint64_t>, record>> records;
mmap_array<db_page> pages;
mmap_array<metadata_t> metadata;
mmap_array<size_t> delete_table;
database(
const std::string& records,
const std::string& pages,
const std::string& deletions,
const std::string& meta,
const size_t& record_cnt = 4096,
const size_t& page_cnt = 4096,
const size_t& delete_cnt = 512
)
: records{record_cnt, records}
, pages{page_cnt, pages}
, metadata{(size_t)1, meta}
, delete_table{delete_cnt, deletions}
{}
static database&& create(const std::string dir, size_t page_nb) {
database ret{
dir+"records",
dir+"pages",
dir+"deleted",
dir+"meta",
page_nb+page_nb/2,
page_nb,
page_nb/8
};
for(auto& n : ret.records)
{
n.second.timestamp = 0;
n.second.offset = 0;
}
for(auto& n : ret.delete_table)
{
n = std::numeric_limits<size_t>::max();
}
(*ret.metadata).last_page = 0;
(*ret.metadata).last_delete = 0;
(*ret.metadata).record_cnt = page_nb+page_nb/2;
(*ret.metadata).page_cnt = page_nb;
(*ret.metadata).delete_cnt = page_nb/8;
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wreturn-local-addr"
return std::move(ret);
#pragma GCC diagnostic pop
}
static database&& open(const std::string dir) {
mmap_array<metadata_t> tmp{(size_t)1, dir+"meta"};
database ret{
dir+"records",
dir+"pages",
dir+"deleted",
dir+"meta",
(*tmp).record_cnt,
(*tmp).page_cnt,
(*tmp).delete_cnt
};
tmp.clear();
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wreturn-local-addr"
return std::move(ret);
#pragma GCC diagnostic pop
}
void write(const record_identifier& target, const db_page& value){
uint64_t page = std::numeric_limits<uint64_t>::max();;
size_t off = std::numeric_limits<size_t>::max();
if(metadata[0].last_delete>0)
{
off = (*metadata).last_delete;
page = delete_table[off-1];
} else {
page = (*metadata).last_page;
if(page>=pages.size()) {
throw std::runtime_error("PAGE STARVATION! MUST EXIT NOW");
}
}
if(page == std::numeric_limits<uint64_t>::max())
{
throw std::runtime_error("PAGE ERROR! MUST EXIT NOW");
}
pages[page] = value;
uint64_t hashed = std::hash<record_identifier>{}(target);
uint64_t hashed_roll = hashed;
bool succeed = false;
uint64_t ts = std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::system_clock::now().time_since_epoch()).count();
std::pair<bitops::regulated<uint64_t>,record> tmp{0, record{}};
tmp.first = hashed;
tmp.second.record_head.split = target;
tmp.second.timestamp = ts;
tmp.second.offset = page;
do{
uint64_t pos = hashed_roll % records.size();
switch (static_cast<uint64_t>(records[pos].second.timestamp)) {
case 0:
[[fallthrough]];
case std::numeric_limits<uint64_t>::max():
records[pos] = tmp;
succeed = true;
break;
default:
break;
}
hashed_roll++;
}while(!succeed);
if(off != std::numeric_limits<size_t>::max())
{
(*metadata).last_delete += -1;
delete_table[off] = std::numeric_limits<size_t>::max();
} else {
(*metadata).last_page += (size_t)1;
}
}
std::pair<uint64_t, db_page> read(const record_identifier& target) {
std::pair<uint64_t, db_page> ret;
ret.first = 0;
ret.second.fill(0);
uint64_t hashed = std::hash<record_identifier>{}(target);
uint64_t hashed_roll = hashed;
do{
uint64_t pos = hashed_roll % records.size();
auto& value = records[pos].second;
switch (static_cast<uint64_t>(value.timestamp)) {
case 0:
return ret;
case std::numeric_limits<uint64_t>::max():
break;
default:
if(records[pos].first == hashed)
if(std::hash<record_identifier>{}(value.record_head.split) == hashed)
{
if(ret.first<value.timestamp)
{
ret.first = value.timestamp;
ret.second = pages[value.offset];
}
break;
}
}
hashed_roll++;
}while(true);
return ret;
}
void remove(const record_identifier& target) {
uint64_t hashed = std::hash<record_identifier>{}(target);
uint64_t hashed_roll = hashed;
do{
uint64_t pos = hashed_roll % records.size();
auto& value = records[pos].second;
switch (static_cast<uint64_t>(value.timestamp)) {
case 0:
return;
case std::numeric_limits<uint64_t>::max():
break;
default:
if(records[pos].first == hashed)
if(std::hash<record_identifier>{}(value.record_head.split) == hashed)
{
value.timestamp = std::numeric_limits<uint64_t>::max();
(*metadata).last_delete+=1;
delete_table[(*metadata).last_delete-1] = value.offset;
value.offset = 0;
break;
}
}
hashed_roll++;
}while(true); // return only happens on hitting a case 0
}
void rollback(const record_identifier&) {
}
};