bill/AQuery
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Optimized hashtable performance; Stored procedures

Browse Source
This commit is contained in:
bill
2022-11-28 04:32:08 +08:00
parent eb0fe8857a
commit d6e3e4878e
11 changed files with 1778 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/aggregations.h
+4 -4
View File
@@ -31,7 +31,7 @@ double avg(const VT<T>& v) {
template<class T, template<typename ...> class VT>
VT<double> sqrt(const VT<T>& v) {
VT<double> ret{ v.size };
VT<double> ret(v.size);
for (uint32_t i = 0; i < v.size; ++i) {
ret[i] = sqrt(v[i]);
}
@@ -52,7 +52,7 @@ VT<T> truncate(const VT<T>& v, const uint32_t precision) {
return v.subvec_memcpy();
auto multiplier = pow(10, precision);
auto max_truncate = std::numeric_limits<T>::max()/multiplier;
VT<T> ret{ v.size };
VT<T> ret(v.size);
for (uint32_t i = 0; i < v.size; ++i) { // round or trunc??
ret[i] = v[i] < max_truncate ? round(v[i] * multiplier)/multiplier : v[i];
}
@@ -102,7 +102,7 @@ decayed_t<VT, T> maxs(const VT<T>& arr) {
template<class T, template<typename ...> class VT>
decayed_t<VT, T> minw(uint32_t w, const VT<T>& arr) {
const uint32_t& len = arr.size;
decayed_t<VT, T> ret{ len };
decayed_t<VT, T> ret(len);
std::deque<std::pair<T, uint32_t>> cache;
for (int i = 0; i < len; ++i) {
if (!cache.empty() && cache.front().second == i - w) cache.pop_front();
@@ -194,7 +194,7 @@ decayed_t<VT, types::GetFPType<types::GetLongType<T>>> avgw(uint32_t w, const VT
uint32_t i = 0;
types::GetLongType<T> s{};
w = w > len ? len : w;
if (len) s = ret[i++] = arr[0];
if (len) s = ret[i++] = arr[0];
for (; i < w; ++i)
ret[i] = (s += arr[i]) / (FPType)(i + 1);
for (; i < len; ++i)
server/hasher.h
+24 -15
View File
@@ -3,7 +3,10 @@
#include <type_traits>
#include <tuple>
#include <functional>
#include <string_view>
#include "types.h"
// #include "robin_hood.h"
#include "unordered_dense.h"
// only works for 64 bit systems
constexpr size_t _FNV_offset_basis = 14695981039346656037ULL;
constexpr size_t _FNV_prime = 1099511628211ULL;
@@ -14,7 +17,7 @@ inline size_t append_bytes(const unsigned char* _First) noexcept {
_Val ^= static_cast<size_t>(*_First);
_Val *= _FNV_prime;
}
return _Val;
}
@@ -65,37 +68,44 @@ struct hasher {
#else
#define _current_type current_type
#endif
return std::hash<_current_type>()(std::get<i>(record)) ^ hashi<i + 1>(record);
return ankerl::unordered_dense::hash<_current_type>()(std::get<i>(record)) ^ hashi<i + 1>(record);
}
size_t operator()(const std::tuple<Types...>& record) const {
return hashi(record);
}
};
template <class T>
struct hasher<T>{
size_t operator()(const std::tuple<T>& record) const {
return ankerl::unordered_dense::hash<T>()(std::get<0>(record));
}
};
namespace std{
namespace ankerl::unordered_dense{
template<>
struct hash<astring_view> {
size_t operator()(const astring_view& _Keyval) const noexcept {
return append_bytes(_Keyval.str);
return ankerl::unordered_dense::hash<std::string_view>()(_Keyval.rstr);
//return append_bytes(_Keyval.str);
}
};
template<>
struct hash<types::date_t> {
size_t operator() (const types::date_t& _Keyval) const noexcept {
return std::hash<unsigned int>()(*(unsigned int*)(&_Keyval));
return ankerl::unordered_dense::hash<unsigned int>()(*(unsigned int*)(&_Keyval));
}
};
template<>
struct hash<types::time_t> {
size_t operator() (const types::time_t& _Keyval) const noexcept {
return std::hash<unsigned int>()(_Keyval.ms) ^
std::hash<unsigned char>()(_Keyval.seconds) ^
std::hash<unsigned char>()(_Keyval.minutes) ^
std::hash<unsigned char>()(_Keyval.hours)
return ankerl::unordered_dense::hash<unsigned int>()(_Keyval.ms) ^
ankerl::unordered_dense::hash<unsigned char>()(_Keyval.seconds) ^
ankerl::unordered_dense::hash<unsigned char>()(_Keyval.minutes) ^
ankerl::unordered_dense::hash<unsigned char>()(_Keyval.hours)
;
}
};
@@ -103,8 +113,8 @@ namespace std{
template<>
struct hash<types::timestamp_t>{
size_t operator() (const types::timestamp_t& _Keyval) const noexcept {
return std::hash<types::date_t>()(_Keyval.date) ^
std::hash<types::time_t>()(_Keyval.time);
return ankerl::unordered_dense::hash<types::date_t>()(_Keyval.date) ^
ankerl::unordered_dense::hash<types::time_t>()(_Keyval.time);
}
};
#ifdef __SIZEOF_INT128__
@@ -112,12 +122,11 @@ namespace std{
template<>
struct hash<int128_struct>{
size_t operator() (const int128_struct& _Keyval) const noexcept {
return std::hash<uint64_t>()(_Keyval.__struct.low) ^ std::hash<uint64_t>()(_Keyval.__struct.high);
return ankerl::unordered_dense::hash<uint64_t>()(_Keyval.__struct.low) ^ ankerl::unordered_dense::hash<uint64_t>()(_Keyval.__struct.high);
}
};
#endif
template <class ...Types>
struct hash<std::tuple<Types...>> : public hasher<Types...>{ };
}
server/libaquery.h
+3 -4
View File
@@ -109,10 +109,9 @@ struct Context{
void init_session();
void end_session();
void* get_module_function(const char*);
std::unordered_map<const char*, void*> tables;
std::unordered_map<const char*, uColRef *> cols;
std::unordered_map<const char*, void*> loaded_modules;
std::unordered_map<const char*, StoredProcedure> stored_proc;
std::unordered_map<std::string, void*> tables;
std::unordered_map<std::string, uColRef *> cols;
std::unordered_map<std::string, StoredProcedure> stored_proc;
};
server/server.cpp
+163 -16
View File
@@ -163,6 +163,20 @@ __AQEXPORT__(bool) have_hge(){
using prt_fn_t = char* (*)(void*, char*);
// This function contains heap allocations, free after use
template<class String_T>
char* to_lpstr(const String_T& str){
auto ret = static_cast<char*>(malloc(str.size() + 1));
memcpy(ret, str.c_str(), str.size());
ret[str.size()] = '\0';
return ret;
}
char* copy_lpstr(const char* str){
auto len = strlen(str);
auto ret = static_cast<char*>(malloc(len + 1));
memcpy(ret, str, len + 1);
return ret;
}
constexpr prt_fn_t monetdbe_prtfns[] = {
aq_to_chars<bool>, aq_to_chars<int8_t>, aq_to_chars<int16_t>, aq_to_chars<int32_t>,
@@ -270,7 +284,18 @@ int dll_main(int argc, char** argv, Context* cxt){
aq_timer timer;
Config *cfg = reinterpret_cast<Config *>(argv[0]);
std::unordered_map<std::string, void*> user_module_map;
std::string pwd = std::filesystem::current_path().c_str();
auto sep = std::filesystem::path::preferred_separator;
pwd += sep;
std::string procedure_root = pwd + "procedures" + sep;
std::string procedure_name = "";
StoredProcedure current_procedure;
vector_type<char *> recorded_queries;
vector_type<void *> recorded_libraries;
bool procedure_recording = false,
procedure_replaying = false;
uint32_t procedure_module_cursor = 0;
if (cxt->module_function_maps == nullptr)
cxt->module_function_maps = new std::unordered_map<std::string, void*>();
auto module_fn_map =
@@ -291,12 +316,12 @@ int dll_main(int argc, char** argv, Context* cxt){
puts(*(const char**)(alt_server->getCol(0)));
cxt->alt_server = alt_server;
}
bool rec = false;
while(cfg->running){
ENGINE_ACQUIRE();
if (cfg->new_query) {
cfg->stats.postproc_time = 0;
cfg->stats.monet_time = 0;
start:
void *handle = nullptr;
void *user_module_handle = nullptr;
@@ -306,7 +331,28 @@ int dll_main(int argc, char** argv, Context* cxt){
Server* server = reinterpret_cast<Server*>(cxt->alt_server);
if(n_recv > 0){
if (cfg->backend_type == BACKEND_AQuery || cfg->has_dll) {
handle = dlopen("./dll.so", RTLD_NOW);
const char* proc_name = "./dll.so";
std::string dll_path;
if (procedure_recording) {
dll_path = procedure_root +
procedure_name + std::to_string(recorded_libraries.size) + ".so";
try{
if (std::filesystem::exists(dll_path))
std::filesystem::remove(dll_path);
std::filesystem::copy_file(proc_name, dll_path);
} catch(std::filesystem::filesystem_error& e){
puts(e.what());
dll_path = proc_name;
}
proc_name = dll_path.c_str();
if(recorded_libraries.size)
recorded_queries.emplace_back(copy_lpstr("N"));
}
handle = dlopen(proc_name, RTLD_NOW);
if (procedure_recording) {
recorded_libraries.emplace_back(handle);
}
}
for (const auto& module : user_module_map){
initialize_module(module.first.c_str(), module.second, cxt);
@@ -314,18 +360,24 @@ int dll_main(int argc, char** argv, Context* cxt){
cxt->init_session();
for(int i = 0; i < n_recv; ++i)
{
//printf("%s, %d\n", n_recvd[i], n_recvd[i][0] == 'Q');
printf("%s, %d\n", n_recvd[i], n_recvd[i][0] == 'Q');
switch(n_recvd[i][0]){
case 'Q': // SQL query for monetdbe
{
if(procedure_recording){
recorded_queries.emplace_back(copy_lpstr(n_recvd[i]));
}
timer.reset();
server->exec(n_recvd[i] + 1);
cfg->stats.monet_time += timer.elapsed();
// printf("Exec Q%d: %s", i, n_recvd[i]);
printf("Exec Q%d: %s", i, n_recvd[i]);
}
break;
case 'P': // Postprocessing procedure
if(handle && !server->haserror()) {
if (procedure_recording) {
recorded_queries.emplace_back(copy_lpstr(n_recvd[i]));
}
code_snippet c = reinterpret_cast<code_snippet>(dlsym(handle, n_recvd[i]+1));
timer.reset();
c(cxt);
@@ -359,6 +411,12 @@ int dll_main(int argc, char** argv, Context* cxt){
case 'O':
{
if(!server->haserror()){
if (procedure_recording){
char* buf = (char*) malloc (sizeof(char) * 6);
memcpy(buf, n_recvd[i], 5);
buf[5] = '\0';
recorded_queries.emplace_back(buf);
}
uint32_t limit;
memcpy(&limit, n_recvd[i] + 1, sizeof(uint32_t));
if (limit == 0)
@@ -379,36 +437,115 @@ int dll_main(int argc, char** argv, Context* cxt){
user_module_map.erase(it);
}
break;
case 'N':
{
if(procedure_module_cursor < current_procedure.postproc_modules)
handle = current_procedure.__rt_loaded_modules[procedure_module_cursor++];
}
break;
case 'R': //recorded procedure
{
auto proc_name = n_recvd[i] + 1;
auto proc_name = n_recvd[i] + 2;
proc_name = *proc_name?proc_name : proc_name + 1;
const auto& load_modules = [](StoredProcedure &p){
puts(proc_name);
const auto& load_modules = [&](StoredProcedure &p) {
if (!p.__rt_loaded_modules){
p.__rt_loaded_modules = static_cast<void**>(
malloc(sizeof(void*) * p.postproc_modules));
for(uint32_t j = 0; j < p.postproc_modules; ++j){
p.__rt_loaded_modules[j] = dlopen(p.name, RTLD_NOW);
auto pj = dlopen(p.name, RTLD_NOW);
if (pj == nullptr){
printf("Error: failed to load module %s\n", p.name);
return true;
}
p.__rt_loaded_modules[j] = pj;
}
}
return false;
};
const auto& save_proc_tofile = [&](const StoredProcedure& p) {
auto config_name = procedure_root + procedure_name + ".aqp";
auto fp = fopen(config_name.c_str(), "wb");
if (fp == nullptr){
printf("Error: failed to open file %s\n", config_name.c_str());
return true;
}
fwrite(&p.cnt, sizeof(p.cnt), 1, fp);
fwrite(&p.postproc_modules, sizeof(p.postproc_modules), 1, fp);
for(uint32_t j = 0; j < p.cnt; ++j){
auto current_query = p.queries[j];
auto len_query = strlen(current_query);
fwrite(current_query, len_query + 1, 1, fp);
}
fclose(fp);
return false;
};
const auto& load_proc_fromfile = [&](StoredProcedure& p) {
auto config_name = procedure_root + p.name + ".aqp";
auto fp = fopen(config_name.c_str(), "rb");
if(fp == nullptr){
puts("ERROR: Procedure not found on disk.");
return false;
}
fread(&p.cnt, sizeof(p.cnt), 1, fp);
fread(&p.postproc_modules, sizeof(p.postproc_modules), 1, fp);
auto offset_now = ftell(fp);
fseek(fp, 0, SEEK_END);
auto queries_size = ftell(fp) - offset_now;
fseek(fp, offset_now, SEEK_SET);
p.queries = static_cast<char**>(malloc(sizeof(char*) * p.cnt));
p.queries[0] = static_cast<char*>(malloc(sizeof(char) * queries_size));
fread(&p.queries[0], queries_size, 1, fp);
for(uint32_t j = 1; j < p.cnt; ++j){
p.queries[j] = p.queries[j-1];
while(*p.queries[j] != '\0')
++p.queries[j];
}
fclose(fp);
return load_modules(p);
};
switch(n_recvd[i][1]){
case '\0':
current_procedure.name = copy_lpstr(proc_name);
current_procedure.cnt = 0;
current_procedure.queries = nullptr;
current_procedure.postproc_modules = 0;
current_procedure.__rt_loaded_modules = nullptr;
procedure_recording = true;
procedure_name = proc_name;
break;
case 'T':
current_procedure.queries = recorded_queries.container;
current_procedure.cnt = recorded_queries.size;
current_procedure.name = copy_lpstr(proc_name);
current_procedure.postproc_modules = recorded_libraries.size;
current_procedure.__rt_loaded_modules = recorded_libraries.container;
recorded_queries.size = recorded_queries.capacity = 0;
recorded_queries.container = nullptr;
recorded_libraries.size = recorded_libraries.capacity = 0;
recorded_libraries.container = nullptr;
procedure_recording = false;
save_proc_tofile(current_procedure);
cxt->stored_proc.insert_or_assign(procedure_name, current_procedure);
procedure_name = "";
break;
case 'E': // execute procedure
{
auto _proc = cxt->stored_proc.find(procedure_name.c_str());
if (_proc == cxt->stored_proc.end())
printf("Procedure %s not found.\n", procedure_name.c_str());
auto _proc = cxt->stored_proc.find(proc_name);
if (_proc == cxt->stored_proc.end()){
printf("Procedure %s not found. Trying load from disk.\n", proc_name);
if (load_proc_fromfile(current_procedure)){
cxt->stored_proc.insert_or_assign(proc_name, current_procedure);
}
}
else{
StoredProcedure &p = _proc->second;
n_recv = p.cnt;
n_recvd = p.queries;
load_modules(p);
current_procedure = _proc->second;
n_recv = current_procedure.cnt;
n_recvd = current_procedure.queries;
load_modules(current_procedure);
goto start; // yes, I know, refactor later!!
}
}
break;
@@ -418,12 +555,22 @@ int dll_main(int argc, char** argv, Context* cxt){
break;
case 'L': //load procedure
break;
case 'd': // display all procedures
for(const auto& p : cxt->stored_proc){
printf("Procedure: %s, %d queries, %d modules:\n", p.first.c_str(),
p.second.cnt, p.second.postproc_modules);
for(uint32_t j = 0; j < p.second.cnt; ++j){
printf("\tQuery %d: %s\n", j, p.second.queries[j]);
}
puts("");
}
break;
}
}
break;
}
}
if(handle) {
if(handle && procedure_replaying) {
dlclose(handle);
handle = nullptr;
}
@@ -486,7 +633,7 @@ extern "C" int __DLLEXPORT__ main(int argc, char** argv) {
#endif
// puts("running");
Context* cxt = new Context();
cxt->aquery_root_path = std::filesystem::current_path().c_str();
cxt->aquery_root_path = to_lpstr(std::filesystem::current_path().string());
// cxt->log("%d %s\n", argc, argv[1]);
#ifdef THREADING
server/table.h
+14 -3
View File
@@ -145,9 +145,19 @@ public:
ColRef<_Ty>& operator =(ColRef<_Ty>&& vt) {
vector_type<_Ty>::operator=(std::move(vt));
return *this;
}
ColView<_Ty> operator [](const vector_type<uint32_t>& idxs) const {
return ColView<_Ty>(*this, idxs);
// ColView<_Ty> operator [](vector_type<uint32_t>& idxs) const {
// return ColView<_Ty>(*this, std::move(idxs));
// }
// ColView<_Ty> operator [](const vector_type<uint32_t>& idxs) const {
// return ColView<_Ty>(*this, idxs);
// }
vector_type<_Ty> operator[](vector_type<uint32_t>& idxs) const {
vector_type<_Ty> ret(idxs.size);
for (uint32_t i = 0; i < idxs.size; ++i)
ret.container[i] = this->container[idxs[i]];
return ret;
}
vector_type<_Ty> operator [](const std::vector<bool>& idxs) const {
vector_type<_Ty> ret (this->size);
@@ -226,7 +236,7 @@ class ColView : public vector_base<_Ty> {
public:
typedef ColRef<_Ty> Decayed_t;
const uint32_t size;
const ColRef<_Ty> orig;
const ColRef<_Ty>& orig;
vector_type<uint32_t> idxs;
ColView(const ColRef<_Ty>& orig, vector_type<uint32_t>&& idxs) : orig(orig), size(idxs.size), idxs(std::move(idxs)) {}
ColView(const ColRef<_Ty>& orig, const vector_type<uint32_t>& idxs) : orig(orig), idxs(idxs), size(idxs.size) {}
@@ -274,6 +284,7 @@ public:
ret[i] = orig[idxs[i]];
return ret;
}
ColView<_Ty> subvec(uint32_t start, uint32_t end) const {
uint32_t len = end - start;
return ColView<_Ty>(orig, idxs.subvec(start, end));
server/unordered_dense.h
+1516
View File
File diff suppressed because it is too large Load Diff
server/vector_type.hpp
+24 -8
View File
@@ -36,7 +36,7 @@ public:
this->size = vt.size;
this->capacity = vt.capacity;
if (capacity) {
// puts("copy");
//puts("copy");
this->container = (_Ty*)malloc(size * sizeof(_Ty));
memcpy(container, vt.container, sizeof(_Ty) * size);
}
@@ -152,18 +152,34 @@ public:
else
return distinct_copy();
}
inline void grow() {
if (size >= capacity) { // geometric growth
uint32_t new_capacity = size + 1 + (size >> 1);
_Ty* n_container = (_Ty*)malloc(new_capacity * sizeof(_Ty));
memcpy(n_container, container, sizeof(_Ty) * size);
// TODO: think of situations where this is a temp!! (copy on write!!!)
template <bool _grow = true>
inline void grow(uint32_t sz = 0) {
if constexpr (_grow)
sz = this->size;
if (sz >= capacity) { // geometric growth
uint32_t new_capacity;
if constexpr (_grow)
new_capacity = size + 1 + (size >> 1);
else
new_capacity = sz;
_Ty* n_container = (_Ty*)realloc(container, new_capacity * sizeof(_Ty));
// memcpy(n_container, container, sizeof(_Ty) * size);
memset(n_container + size, 0, sizeof(_Ty) * (new_capacity - size));
if (capacity)
free(container);
// if (capacity)
// free(container);
container = n_container;
capacity = new_capacity;
}
}
inline void resize(const uint32_t sz){
size = sz;
grow<false>(sz);
}
inline void reserve(const uint32_t sz){
grow<false>(sz);
}
void emplace_back(const _Ty& _val) {
grow();
container[size++] = _val;