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

Interval based triggers

Browse Source 
Monetdb get col/sz
Updated RF/dataproc
Bug fixes and improvements
This commit is contained in:
bill
2023-02-08 01:28:28 +08:00
parent cf8185c5f0
commit 906daf577b
29 changed files with 1218 additions and 659 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/libaquery.cpp
+7 -4
View File
@@ -31,12 +31,14 @@ void print<__int128_t>(const __int128_t& v, const char* delimiter){
s[40] = 0;
std::cout<< get_int128str(v, s+40)<< delimiter;
}
template <>
void print<__uint128_t>(const __uint128_t&v, const char* delimiter){
char s[41];
s[40] = 0;
std::cout<< get_uint128str(v, s+40) << delimiter;
}
std::ostream& operator<<(std::ostream& os, __int128 & v)
{
print(v);
@@ -76,6 +78,7 @@ char* intToString(T val, char* buf){
return buf;
}
void skip(const char*& buf){
while(*buf && (*buf >'9' || *buf < '0')) buf++;
}
@@ -264,8 +267,8 @@ std::string base62uuid(int l) {
static uniform_int_distribution<uint64_t> u(0x10000, 0xfffff);
uint64_t uuid = (u(engine) << 32ull) +
(std::chrono::system_clock::now().time_since_epoch().count() & 0xffffffff);
//printf("%llu\n", uuid);
string ret;
string ret;
while (uuid && l-- >= 0) {
ret = string("") + base62alp[uuid % 62] + ret;
uuid /= 62;
@@ -525,8 +528,8 @@ void ScratchSpace::cleanup(){
static_cast<vector_type<void*>*>(temp_memory_fractions);
if (vec_tmpmem_fractions->size) {
for(auto& mem : *vec_tmpmem_fractions){
free(mem);
//GC::gc_handle->reg(mem);
//free(mem);
GC::gc_handle->reg(mem);
}
vec_tmpmem_fractions->clear();
}
server/libaquery.h
+86 -4
View File
@@ -66,23 +66,24 @@ struct Session{
void* memory_map;
};
struct StoredProcedure{
struct StoredProcedure {
uint32_t cnt, postproc_modules;
char **queries;
const char* name;
void **__rt_loaded_modules;
};
struct Context{
struct Context {
typedef int (*printf_type) (const char *format, ...);
void* module_function_maps = 0;
void* module_function_maps = nullptr;
Config* cfg;
int n_buffers, *sz_bufs;
void **buffers;
void* alt_server = 0;
void* alt_server = nullptr;
Log_level log_level = LOG_INFO;
Session current;
@@ -115,6 +116,12 @@ struct Context{
};
struct StoredProcedurePayload {
StoredProcedure *p;
Context* cxt;
};
int execTriggerPayload(void*);
#ifdef _WIN32
#define __DLLEXPORT__ __declspec(dllexport) __stdcall
@@ -169,4 +176,79 @@ inline void AQ_ZeroMemory(_This_Struct& __val) {
memset(&__val, 0, sizeof(_This_Struct));
}
#ifdef __USE_STD_SEMAPHORE__
#include <semaphore>
class A_Semaphore {
private:
std::binary_semaphore native_handle;
public:
A_Semaphore(bool v = false) {
native_handle = std::binary_semaphore(v);
}
void acquire() {
native_handle.acquire();
}
void release() {
native_handle.release();
}
~A_Semaphore() { }
};
#else
#ifdef _WIN32
class A_Semaphore {
private:
void* native_handle;
public:
A_Semaphore(bool);
void acquire();
void release();
~A_Semaphore();
};
#else
#ifdef __APPLE__
#include <dispatch/dispatch.h>
class A_Semaphore {
private:
dispatch_semaphore_t native_handle;
public:
explicit A_Semaphore(bool v = false) {
native_handle = dispatch_semaphore_create(v);
}
void acquire() {
// puts("acquire");
dispatch_semaphore_wait(native_handle, DISPATCH_TIME_FOREVER);
}
void release() {
// puts("release");
dispatch_semaphore_signal(native_handle);
}
~A_Semaphore() {
}
};
#else
#include <semaphore.h>
class A_Semaphore {
private:
sem_t native_handle;
public:
A_Semaphore(bool v = false) {
sem_init(&native_handle, v, 1);
}
void acquire() {
sem_wait(&native_handle);
}
void release() {
sem_post(&native_handle);
}
~A_Semaphore() {
sem_destroy(&native_handle);
}
};
#endif // __APPLE__
#endif // _WIN32
#endif //__USE_STD_SEMAPHORE__
void print_monetdb_results(void* _srv, const char* sep, const char* end, uint32_t limit);
#endif
server/monetdb_conn.cpp
+53
View File
@@ -5,9 +5,19 @@
#include <string>
#include "monetdb_conn.h"
#include "monetdbe.h"
#include "table.h"
#include <thread>
#ifdef _WIN32
#include "winhelper.h"
#else
#include <dlfcn.h>
#include <fcntl.h>
#include <sys/mman.h>
#include <atomic>
#endif // _WIN32
#undef ERROR
#undef static_assert
@@ -215,3 +225,46 @@ bool Server::havehge() {
return false;
#endif
}
void ExecuteStoredProcedureEx(const StoredProcedure *p, Context* cxt){
auto server = static_cast<Server*>(cxt->alt_server);
void* handle = nullptr;
uint32_t procedure_module_cursor = 0;
for(uint32_t i = 0; i < p->cnt; ++i) {
switch(p->queries[i][0]){
case 'Q': {
server->exec(p->queries[i]);
}
break;
case 'P': {
auto c = code_snippet(dlsym(handle, p->queries[i]+1));
c(cxt);
}
break;
case 'N': {
if(procedure_module_cursor < p->postproc_modules)
handle = p->__rt_loaded_modules[procedure_module_cursor++];
}
break;
case 'O': {
uint32_t limit;
memcpy(&limit, p->queries[i] + 1, sizeof(uint32_t));
if (limit == 0)
continue;
print_monetdb_results(server, " ", "\n", limit);
}
break;
default:
printf("Warning Q%u: unrecognized command %c.\n",
i, p->queries[i][0]);
}
}
}
int execTriggerPayload(void* args) {
auto spp = (StoredProcedurePayload*)(args);
ExecuteStoredProcedureEx(spp->p, spp->cxt);
delete spp;
return 0;
}
server/monetdb_conn.h
+1 -1
View File
@@ -24,7 +24,7 @@ struct Server{
static bool havehge();
void test(const char*);
void print_results(const char* sep = " ", const char* end = "\n");
friend void print_monetdb_results(Server* srv, const char* sep, const char* end, int limit);
friend void print_monetdb_results(void* _srv, const char* sep, const char* end, int limit);
~Server();
};
server/monetdb_ext.c
+93
View File
@@ -0,0 +1,93 @@
#include "monetdbe.h"
#include <stdint.h>
#include "mal_client.h"
#include "sql_mvc.h"
#include "sql_semantic.h"
#include "mal_exception.h"
typedef struct column_storage {
int refcnt;
int bid;
int ebid; /* extra bid */
int uibid; /* bat with positions of updates */
int uvbid; /* bat with values of updates */
storage_type st; /* ST_DEFAULT, ST_DICT, ST_FOR */
bool cleared;
bool merged; /* only merge changes once */
size_t ucnt; /* number of updates */
ulng ts; /* version timestamp */
} column_storage;
typedef struct segment {
BUN start;
BUN end;
bool deleted; /* we need to keep a dense segment set, 0 - end of last segemnt,
some segments maybe deleted */
ulng ts; /* timestamp on this segment, ie tid of some active transaction or commit time of append/delete or
rollback time, ie ready for reuse */
ulng oldts; /* keep previous ts, for rollbacks */
struct segment *next; /* usualy one should be enough */
struct segment *prev; /* used in destruction list */
} segment;
/* container structure to allow sharing this structure */
typedef struct segments {
sql_ref r;
struct segment *h;
struct segment *t;
} segments;
typedef struct storage {
column_storage cs; /* storage on disk */
segments *segs; /* local used segements */
struct storage *next;
} storage;
typedef struct {
char language; /* 'S' or 's' or 'X' */
char depth; /* depth >= 1 means no output for trans/schema statements */
int remote; /* counter to make remote function names unique */
mvc *mvc;
char others[];
} backend;
typedef struct {
Client c;
char *msg;
monetdbe_data_blob blob_null;
monetdbe_data_date date_null;
monetdbe_data_time time_null;
monetdbe_data_timestamp timestamp_null;
str mid;
} monetdbe_database_internal;
size_t
monetdbe_get_size(monetdbe_database dbhdl, const char* schema_name, const char *table_name)
{
monetdbe_database_internal* hdl = (monetdbe_database_internal*)dbhdl;
backend* be = ((backend *)(((monetdbe_database_internal*)dbhdl)->c->sqlcontext));
mvc *m = be->mvc;
mvc_trans(m);
sql_table *t = find_table_or_view_on_scope(m, NULL, schema_name, table_name, "CATALOG", false);
sql_column *col = ol_first_node(t->columns)->data;
sqlstore* store = m->store;
size_t sz = store->storage_api.count_col(m->session->tr, col, QUICK);
mvc_cancel_session(m);
return sz;
}
void*
monetdbe_get_col(monetdbe_database dbhdl, const char* schema_name, const char *table_name, uint32_t col_id) {
monetdbe_database_internal* hdl = (monetdbe_database_internal*)dbhdl;
backend* be = ((backend *)(((monetdbe_database_internal*)dbhdl)->c->sqlcontext));
mvc *m = be->mvc;
mvc_trans(m);
sql_table *t = find_table_or_view_on_scope(m, NULL, schema_name, table_name, "CATALOG", false);
sql_column *col = ol_fetch(t->columns, col_id);
sqlstore* store = m->store;
BAT *b = store->storage_api.bind_col(m->session->tr, col, RDONLY);
BATiter iter = bat_iterator(b);
mvc_cancel_session(m);
return iter.base;
}
server/server.cpp
+8 -73
View File
@@ -20,10 +20,6 @@
#include <sys/mman.h>
#include <atomic>
// fast numeric to string conversion
#include "jeaiii_to_text.h"
#include "dragonbox/dragonbox_to_chars.h"
struct SharedMemory
{
std::atomic<bool> a;
@@ -41,69 +37,7 @@ struct SharedMemory
}
};
#ifndef __USE_STD_SEMAPHORE__
#ifdef __APPLE__
#include <dispatch/dispatch.h>
class A_Semaphore {
private:
dispatch_semaphore_t native_handle;
public:
A_Semaphore(bool v = false) {
native_handle = dispatch_semaphore_create(v);
}
void acquire() {
// puts("acquire");
dispatch_semaphore_wait(native_handle, DISPATCH_TIME_FOREVER);
}
void release() {
// puts("release");
dispatch_semaphore_signal(native_handle);
}
~A_Semaphore() {
}
};
#else
#include <semaphore.h>
class A_Semaphore {
private:
sem_t native_handle;
public:
A_Semaphore(bool v = false) {
sem_init(&native_handle, v, 1);
}
void acquire() {
sem_wait(&native_handle);
}
void release() {
sem_post(&native_handle);
}
~A_Semaphore() {
sem_destroy(&native_handle);
}
};
#endif
#endif
#endif
#ifdef __USE_STD_SEMAPHORE__
#define __AQUERY_ITC_USE_SEMPH__
#include <semaphore>
class A_Semaphore {
private:
std::binary_semaphore native_handle;
public:
A_Semaphore(bool v = false) {
native_handle = std::binary_semaphore(v);
}
void acquire() {
native_handle.acquire();
}
void release() {
native_handle.release();
}
~A_Semaphore() { }
};
#endif
#endif // _WIN32
#ifdef __AQUERY_ITC_USE_SEMPH__
A_Semaphore prompt{ true }, engine{ false };
@@ -225,9 +159,10 @@ inline constexpr static unsigned char monetdbe_type_szs[] = {
1
};
constexpr uint32_t output_buffer_size = 65536;
void print_monetdb_results(Server* srv, const char* sep = " ", const char* end = "\n",
void print_monetdb_results(void* _srv, const char* sep = " ", const char* end = "\n",
uint32_t limit = std::numeric_limits<uint32_t>::max()) {
if (!srv->haserror() && srv->cnt && limit){
auto srv = static_cast<Server *>(_srv);
if (!srv->haserror() && srv->cnt && limit) {
char buffer[output_buffer_size];
auto _res = static_cast<monetdbe_result*> (srv->res);
const auto ncols = _res->ncols;
@@ -255,7 +190,7 @@ void print_monetdb_results(Server* srv, const char* sep = " ", const char* end =
puts("Error: separator or end string too long");
goto cleanup;
}
if (header_string.size() - l_sep - 1>= 0)
if (header_string.size() >= l_sep + 1)
header_string.resize(header_string.size() - l_sep - 1);
header_string += end + std::string(header_string.size(), '=') + end;
fputs(header_string.c_str(), stdout);
@@ -324,7 +259,7 @@ int dll_main(int argc, char** argv, Context* cxt){
catch (std::filesystem::filesystem_error& e) {
printf("Failed to create directory %s: %s\n", procedure_root.c_str(), e.what());
}
if (cxt->module_function_maps == nullptr)
cxt->module_function_maps = new std::unordered_map<std::string, void*>();
auto module_fn_map =
@@ -590,9 +525,9 @@ start:
break;
case 'D': // delete procedure
break;
case 'S': //save procedure
case 'S': // save procedure
break;
case 'L': //load procedure
case 'L': // load procedure
if (!load_proc_fromfile(current_procedure)) {
cxt->stored_proc.insert_or_assign(proc_name, current_procedure);
}
server/tests/thread_pool.hpp
+26 -16
View File
@@ -4,21 +4,26 @@
#include <thread>
#include <cstdio>
#include <cstdlib>
using namespace std;
FILE *fp;
long long testing_throughput(uint32_t n_jobs, bool prompt = true){
using namespace std::chrono_literals;
printf("Threadpool througput test with %u jobs. Press any key to start.\n", n_jobs);
auto tp = ThreadPool(thread::hardware_concurrency());
auto tp = ThreadPool(std::thread::hardware_concurrency());
getchar();
auto i = 0u;
fp = fopen("tmp.tmp", "wb");
auto time = chrono::high_resolution_clock::now();
while(i++ < n_jobs) tp.enqueue_task({ [](void* f) {fprintf(fp, "%d ", *(int*)f); free(f); }, new int(i) });
auto time = std::chrono::high_resolution_clock::now();
while(i++ < n_jobs) {
payload_t payload;
payload.f = [](void* f) {fprintf(fp, "%d ", *(int*)f); free(f); return 0; };
payload.args = new int(i);
tp.enqueue_task(payload);
}
puts("done dispatching.");
while (tp.busy()) this_thread::sleep_for(1s);
auto t = (chrono::high_resolution_clock::now() - time).count();
while (tp.busy()) std::this_thread::sleep_for(1s);
auto t = (std::chrono::high_resolution_clock::now() - time).count();
printf("\nTr: %u, Ti: %lld \nThroughput: %lf transactions/ns\n", i, t, i/(double)(t));
//this_thread::sleep_for(2s);
fclose(fp);
@@ -27,26 +32,31 @@ long long testing_throughput(uint32_t n_jobs, bool prompt = true){
long long testing_transaction(uint32_t n_burst, uint32_t n_batch,
uint32_t base_time, uint32_t var_time, bool prompt = true, FILE* _fp = stdout){
using namespace std::chrono_literals;
printf("Threadpool transaction test: burst: %u, batch: %u, time: [%u, %u].\n"
, n_burst, n_batch, base_time, var_time + base_time);
if (prompt) {
puts("Press any key to start.");
getchar();
}
auto tp = ThreadPool(thread::hardware_concurrency());
auto tp = ThreadPool(std::thread::hardware_concurrency());
fp = _fp;
auto i = 0u, j = 0u;
auto time = chrono::high_resolution_clock::now();
auto time = std::chrono::high_resolution_clock::now();
while(j++ < n_batch){
i = 0u;
while(i++ < n_burst)
tp.enqueue_task({ [](void* f) { fprintf(fp, "%d ", *(int*)f); free(f); }, new int(j) });
while(i++ < n_burst) {
payload_t payload;
payload.f = [](void* f) { fprintf(fp, "%d ", *(int*)f); free(f); return 0; };
payload.args = new int(j);
tp.enqueue_task(payload);
}
fflush(stdout);
this_thread::sleep_for(chrono::microseconds(rand()%var_time + base_time));
std::this_thread::sleep_for(std::chrono::microseconds(rand()%var_time + base_time));
}
puts("done dispatching.");
while (tp.busy()) this_thread::sleep_for(1s);
auto t = (chrono::high_resolution_clock::now() - time).count();
while (tp.busy()) std::this_thread::sleep_for(1s);
auto t = (std::chrono::high_resolution_clock::now() - time).count();
printf("\nTr: %u, Ti: %lld \nThroughput: %lf transactions/ns\n", j*i, t, j*i/(double)(t));
return t;
@@ -58,17 +68,17 @@ long long testing_destruction(bool prompt = true){
puts("Press any key to start.");
getchar();
}
auto time = chrono::high_resolution_clock::now();
auto time = std::chrono::high_resolution_clock::now();
for(int i = 0; i < 8; ++i)
testing_transaction(0xfff, 0xff, 400, 100, false, fp);
for(int i = 0; i < 64; ++i)
testing_transaction(0xff, 0xf, 60, 20, false, fp);
for(int i = 0; i < 1024; ++i) {
auto tp = new ThreadPool(256);
auto tp = new ThreadPool(255);
delete tp;
}
return 0;
auto t = (chrono::high_resolution_clock::now() - time).count();
auto t = (std::chrono::high_resolution_clock::now() - time).count();
fclose(fp);
return t;
}
server/threading.cpp
+45 -14
View File
@@ -1,4 +1,5 @@
#include "threading.h"
#include "libaquery.h"
#include <thread>
#include <atomic>
#include <mutex>
@@ -105,7 +106,8 @@ ThreadPool::ThreadPool(uint32_t n_threads)
current_payload = new payload_t[n_threads];
for (uint32_t i = 0; i < n_threads; ++i){
atomic_init(tf + i, static_cast<unsigned char>(0b10));
// atomic_init(tf + i, static_cast<unsigned char>(0b10));
tf[i] = static_cast<unsigned char>(0b10);
th[i] = thread(&ThreadPool::daemon_proc, this, i);
}
@@ -152,21 +154,50 @@ bool ThreadPool::busy(){
return true;
}
Trigger::Trigger(ThreadPool* tp){
IntervalBasedTriggerHost::IntervalBasedTriggerHost(ThreadPool* tp){
this->tp = tp;
this->triggers = new vector_type<IntervalBasedTrigger>;
adding_trigger = new mutex();
this->now = std::chrono::high_resolution_clock::now().time_since_epoch().count();
}
void IntervalBasedTrigger::timer::reset(){
void IntervalBasedTriggerHost::add_trigger(StoredProcedure *p, uint32_t interval) {
auto tr = IntervalBasedTrigger{.interval = interval, .time_remaining = 0, .sp = p};
auto vt_triggers = static_cast<vector_type<IntervalBasedTrigger> *>(this->triggers);
adding_trigger->lock();
vt_triggers->emplace_back(tr);
adding_trigger->unlock();
}
void IntervalBasedTriggerHost::tick() {
const auto current_time = std::chrono::high_resolution_clock::now().time_since_epoch().count();
const auto delta_t = static_cast<uint32_t>((current_time - now) / 1000000); // miliseconds precision
now = current_time;
auto vt_triggers = static_cast<vector_type<IntervalBasedTrigger> *>(this->triggers);
adding_trigger->lock();
for(auto& t : *vt_triggers) {
if(t.tick(delta_t)) {
payload_t payload;
payload.f = execTriggerPayload;
payload.args = static_cast<void*>(new StoredProcedurePayload {t.sp, cxt});
tp->enqueue_task(payload);
}
}
adding_trigger->unlock();
}
void IntervalBasedTrigger::reset() {
time_remaining = interval;
}
bool IntervalBasedTrigger::timer::tick(uint32_t t){
if (time_remaining > t) {
time_remaining -= t;
return false;
}
else{
time_remaining = interval - t%interval;
return true;
}
}
bool IntervalBasedTrigger::tick(uint32_t delta_t) {
bool ret = false;
if (time_remaining <= delta_t)
ret = true;
if (auto curr_dt = delta_t % interval; time_remaining <= curr_dt)
time_remaining = interval + time_remaining - curr_dt;
else
time_remaining = time_remaining - curr_dt;
return ret;
}
server/threading.h
+31 -16
View File
@@ -7,7 +7,7 @@ typedef int(*payload_fn_t)(void*);
struct payload_t{
payload_fn_t f;
void* args;
constexpr payload_t(payload_fn_t f, void* args) noexcept
constexpr payload_t(payload_fn_t f, void* args) noexcept
: f(f), args(args) {}
constexpr payload_t() noexcept
: f(nullptr), args(nullptr) {};
@@ -19,7 +19,7 @@ struct payload_t{
class ThreadPool{
public:
ThreadPool(uint32_t n_threads = 0);
explicit ThreadPool(uint32_t n_threads = 0);
void enqueue_task(const payload_t& payload);
bool busy();
virtual ~ThreadPool();
@@ -39,29 +39,44 @@ private:
};
class Trigger{
private:
void* triggers; //min-heap by t-rem
virtual void tick() = 0;
#include <thread>
#include <mutex>
class A_Semphore;
class TriggerHost {
protected:
void* triggers;
std::thread* handle;
ThreadPool *tp;
Context* cxt;
std::mutex* adding_trigger;
virtual void tick() = 0;
public:
Trigger(ThreadPool* tp);
TriggerHost() = default;
virtual ~TriggerHost() = default;
};
class IntervalBasedTrigger : public Trigger{
struct StoredProcedure;
struct IntervalBasedTrigger {
uint32_t interval; // in milliseconds
uint32_t time_remaining;
StoredProcedure* sp;
void reset();
bool tick(uint32_t t);
};
class IntervalBasedTriggerHost : public TriggerHost {
public:
struct timer{
uint32_t interval; // in milliseconds
uint32_t time_remaining;
void reset();
bool tick(uint32_t t);
};
void add_trigger();
explicit IntervalBasedTriggerHost(ThreadPool *tp);
void add_trigger(StoredProcedure* stored_procedure, uint32_t interval);
private:
unsigned long long now;
void tick() override;
};
class CallbackBasedTrigger : public Trigger{
class CallbackBasedTriggerHost : public TriggerHost {
public:
void add_trigger();
private:
server/winhelper.h
-12
View File
@@ -16,18 +16,6 @@ struct SharedMemory
void FreeMemoryMap();
};
#ifndef __USE_STD_SEMAPHORE__
class A_Semaphore {
private:
void* native_handle;
public:
A_Semaphore(bool);
void acquire();
void release();
~A_Semaphore();
};
#endif
#endif // WIN32
#endif // WINHELPER