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

Merge branch 'master' of https://github.com/sunyinqi0508/AQuery

Browse Source
This commit is contained in:
bill
2022-08-24 06:26:29 +08:00
parent c7b32e9a20 09e5f7dbc9
commit c5a6f8c423
25 changed files with 813 additions and 455 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/libaquery.h
+14 -1
View File
@@ -21,6 +21,15 @@ struct Config{
int buffer_sizes[];
};
struct Session{
struct Statistic{
size_t total_active;
size_t cnt_object;
size_t total_alloc;
};
void* memory_map;
};
struct Context{
typedef int (*printf_type) (const char *format, ...);
std::unordered_map<const char*, void*> tables;
@@ -34,6 +43,8 @@ struct Context{
void* alt_server;
Log_level log_level = LOG_INFO;
Session current;
#ifdef THREADING
void* thread_pool;
#endif
@@ -49,6 +60,8 @@ struct Context{
if (log_level <= LOG_ERROR)
print(args...);
}
void init_session();
void end_session();
};
#ifdef _WIN32
@@ -58,4 +71,4 @@ struct Context{
#endif
#define __AQEXPORT__(_Ty) extern "C" _Ty __DLLEXPORT__
#endif
#endif
server/mariadb_conn.h
+1
View File
@@ -9,6 +9,7 @@ struct Server{
MYSQL *server = 0;
Context *cxt = 0;
bool status = 0;
bool has_error = false;
char* query = 0;
int type = 0;
server/monetdb_conn.cpp
+44 -12
View File
@@ -1,6 +1,8 @@
#include "libaquery.h"
#include <cstdio>
#include "monetdb_conn.h"
#include "monetdbe.h"
#undef static_assert
const char* monetdbe_type_str[] = {
"monetdbe_bool", "monetdbe_int8_t", "monetdbe_int16_t", "monetdbe_int32_t", "monetdbe_int64_t",
@@ -37,6 +39,7 @@ Server::Server(Context* cxt){
}
void Server::connect(Context *cxt){
auto server = static_cast<monetdbe_database*>(this->server);
if (cxt){
cxt->alt_server = this;
this->cxt = cxt;
@@ -47,14 +50,14 @@ void Server::connect(Context *cxt){
}
if (server){
printf("Error: Server %llx already connected. Restart? (Y/n). \n", server);
printf("Error: Server %p already connected. Restart? (Y/n). \n", server);
char c[50];
std::cin.getline(c, 49);
for(int i = 0; i < 50; ++i){
if (!c[i] || c[i] == 'y' || c[i] == 'Y'){
monetdbe_close(*server);
free(*server);
server = 0;
this->server = 0;
break;
}
else if(c[i]&&!(c[i] == ' ' || c[i] == '\t'))
@@ -64,36 +67,65 @@ void Server::connect(Context *cxt){
server = (monetdbe_database*)malloc(sizeof(monetdbe_database));
auto ret = monetdbe_open(server, nullptr, nullptr);
if (ret == 0){
status = true;
this->server = server;
}
else{
if(server)
free(server);
this->server = 0;
status = false;
puts(ret == -1 ? "Allocation Error." : "Internal Database Error.");
}
}
void Server::exec(const char* q){
auto qresult = monetdbe_query(*server, const_cast<char*>(q), &res, &cnt);
if (res != 0)
this->cnt = res->nrows;
auto server = static_cast<monetdbe_database*>(this->server);
auto _res = static_cast<monetdbe_result*>(this->res);
monetdbe_cnt _cnt = 0;
auto qresult = monetdbe_query(*server, const_cast<char*>(q), &_res, &_cnt);
if (_res != 0){
this->cnt = _res->nrows;
this->res = _res;
}
if (qresult != nullptr){
printf("Execution Failed. %s\n", qresult);
last_error = qresult;
}
}
bool Server::haserror(){
if (last_error){
last_error = 0;
return true;
}
else{
return false;
}
}
void Server::close(){
if(this->server){
monetdbe_close(*(this->server));
free(this->server);
auto server = static_cast<monetdbe_database*>(this->server);
monetdbe_close(*(server));
free(server);
this->server = 0;
}
}
void* Server::getCol(int col_idx){
if(res){
res->ncols;
auto err_msg = monetdbe_result_fetch(res, &ret_col, col_idx);
auto _res = static_cast<monetdbe_result*>(this->res);
auto err_msg = monetdbe_result_fetch(_res,
reinterpret_cast<monetdbe_column**>(&ret_col), col_idx);
if(err_msg == nullptr)
{
cnt = ret_col->count;
auto _ret_col = static_cast<monetdbe_column*>(this->ret_col);
cnt = _ret_col->count;
printf("Dbg: Getting col %s, type: %s\n",
ret_col->name, monetdbe_type_str[ret_col->type]);
return ret_col->data;
_ret_col->name, monetdbe_type_str[_ret_col->type]);
return _ret_col->data;
}
else{
printf("Error fetching result: %s\n", err_msg);
server/monetdb_conn.h
+5 -5
View File
@@ -1,17 +1,16 @@
#include "monetdbe.h"
struct Context;
struct Server{
monetdbe_database *server = 0;
void *server = 0;
Context *cxt = 0;
bool status = 0;
char* query = 0;
int type = 1;
monetdbe_result* res = 0;
monetdbe_column* ret_col = 0;
monetdbe_cnt cnt = 0;
void* res = 0;
void* ret_col = 0;
long long cnt = 0;
char* last_error = 0;
Server(Context* cxt = nullptr);
@@ -19,5 +18,6 @@ struct Server{
void exec(const char* q);
void *getCol(int col_idx);
void close();
bool haserror();
~Server();
};
server/server.cpp
+35 -11
View File
@@ -72,6 +72,12 @@ __AQEXPORT__(bool) have_hge(){
#endif
}
void Context::init_session(){
if (log_level == LOG_INFO){
Session::Statistic stats;
}
}
int dll_main(int argc, char** argv, Context* cxt){
Config *cfg = reinterpret_cast<Config *>(argv[0]);
@@ -99,15 +105,20 @@ int dll_main(int argc, char** argv, Context* cxt){
}
for(int i = 0; i < n_recv; ++i)
{
printf("%s, %d\n", n_recvd[i], n_recvd[i][0] == 'Q');
if (n_recvd[i][0] == 'Q'){
server->exec(n_recvd[i] + 1);
printf("Exec Q%d: %s\n", i, n_recvd[i]);
}
else if (n_recvd[i][0] == 'P' && handle) {
else if (n_recvd[i][0] == 'P' && handle && !server->haserror()) {
code_snippet c = reinterpret_cast<code_snippet>(dlsym(handle, n_recvd[i]+1));
c(cxt);
}
}
if(handle) {
dlclose(handle);
handle = 0;
}
n_recv = 0;
}
if(server->last_error == nullptr){
@@ -166,11 +177,11 @@ extern "C" int __DLLEXPORT__ main(int argc, char** argv) {
cxt->log("running: %s\n", running? "true":"false");
cxt->log("ready: %s\n", ready? "true":"false");
void* handle = dlopen("./dll.so", RTLD_LAZY);
cxt->log("handle: %lx\n", handle);
cxt->log("handle: %p\n", handle);
if (handle) {
cxt->log("inner\n");
code_snippet c = reinterpret_cast<code_snippet>(dlsym(handle, "dllmain"));
cxt->log("routine: %lx\n", c);
cxt->log("routine: %p\n", c);
if (c) {
cxt->log("inner\n");
cxt->err("return: %d\n", c(cxt));
@@ -195,11 +206,24 @@ int test_main()
cxt->alt_server = new Server(cxt);
Server* server = reinterpret_cast<Server*>(cxt->alt_server);
const char* qs[]= {
"CREATE TABLE tt(a INT, b INT, c INT, d INT);",
"COPY OFFSET 2 INTO tt FROM 'D:/gg/AQuery++/test.csv' ON SERVER USING DELIMITERS ',';",
"CREATE TABLE sale(Mont INT, sales INT);",
"COPY OFFSET 2 INTO sale FROM 'D:/gg/AQuery++/moving_avg.csv' ON SERVER USING DELIMITERS ',';",
"SELECT a FROM tt, sale WHERE a = Mont ;"
"CREATE TABLE stocks(timestamp INT, price INT);",
"INSERT INTO stocks VALUES(1, 15);;",
"INSERT INTO stocks VALUES(2,19); ",
"INSERT INTO stocks VALUES(3,16);",
"INSERT INTO stocks VALUES(4,17);",
"INSERT INTO stocks VALUES(5,15);",
"INSERT INTO stocks VALUES(6,13);",
"INSERT INTO stocks VALUES(7,5);",
"INSERT INTO stocks VALUES(8,8);",
"INSERT INTO stocks VALUES(9,7);",
"INSERT INTO stocks VALUES(10,13);",
"INSERT INTO stocks VALUES(11,11);",
"INSERT INTO stocks VALUES(12,14);",
"INSERT INTO stocks VALUES(13,10);",
"INSERT INTO stocks VALUES(14,5);",
"INSERT INTO stocks VALUES(15,2);",
"INSERT INTO stocks VALUES(16,5);",
"SELECT price, timestamp FROM stocks WHERE (((price - timestamp) > 1) AND (NOT ((price * timestamp) < 100))) ;",
};
n_recv = sizeof(qs)/(sizeof (char*));
n_recvd = const_cast<char**>(qs);
@@ -215,11 +239,11 @@ int test_main()
cxt->log_level = LOG_INFO;
puts(cpp_17 ?"true":"false");
void* handle = dlopen("./dll.so", RTLD_LAZY);
cxt->log("handle: %llx\n", handle);
cxt->log("handle: %p\n", handle);
if (handle) {
cxt->log("inner\n");
code_snippet c = reinterpret_cast<code_snippet>(dlsym(handle, "dllmain"));
cxt->log("routine: %llx\n", c);
code_snippet c = reinterpret_cast<code_snippet>(dlsym(handle, "dll_6EgnKh"));
cxt->log("routine: %p\n", c);
if (c) {
cxt->log("inner\n");
cxt->log("return: %d\n", c(cxt));
server/server.vcxproj
+1 -1
View File
@@ -196,7 +196,7 @@
<MultiProcessorCompilation>true</MultiProcessorCompilation>
<EnableParallelCodeGeneration>true</EnableParallelCodeGeneration>
<LanguageStandard>stdcpplatest</LanguageStandard>
<LanguageStandard_C>stdc17</LanguageStandard_C>
<LanguageStandard_C>stdc11</LanguageStandard_C>
<AdditionalIncludeDirectories>$(ProjectDir)\..\monetdb\msvc</AdditionalIncludeDirectories>
</ClCompile>
<Link>
server/table.cpp
+1 -1
View File
@@ -25,4 +25,4 @@ std::ostream& operator<<(std::ostream& os, __uint128_t & v)
print(v);
return os;
}
#endif
#endif
server/tests/thread_pool.hpp
+36 -9
View File
@@ -1,3 +1,5 @@
#pragma once
#include "../threading.h"
#include <thread>
#include <cstdio>
@@ -5,8 +7,8 @@
using namespace std;
FILE *fp;
int testing_throughput(uint32_t n_jobs){
printf("Threadpool througput test with %u jobs.\n", n_jobs);
long long testing_throughput(uint32_t n_jobs, bool prompt = true){
printf("Threadpool througput test with %u jobs. Press any key to start.\n", n_jobs);
auto tp = ThreadPool(thread::hardware_concurrency());
getchar();
@@ -19,22 +21,26 @@ int testing_throughput(uint32_t n_jobs){
auto t = (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);
return 0;
fclose(fp);
return t;
}
int testing_transaction(uint32_t n_burst, uint32_t n_batch,
uint32_t base_time, uint32_t var_time){
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){
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());
getchar();
fp = _fp;
auto i = 0u, j = 0u;
auto time = chrono::high_resolution_clock::now();
while(j++ < n_batch){
i = 0u;
while(i++ < n_burst)
tp.enqueue_task({ [](void* f) { printf( "%d ", *(int*)f); free(f); }, new int(i) });
tp.enqueue_task({ [](void* f) { fprintf(fp, "%d ", *(int*)f); free(f); }, new int(j) });
fflush(stdout);
this_thread::sleep_for(chrono::microseconds(rand()%var_time + base_time));
}
@@ -42,6 +48,27 @@ int testing_transaction(uint32_t n_burst, uint32_t n_batch,
while (tp.busy()) this_thread::sleep_for(1s);
auto t = (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 0;
return t;
}
long long testing_destruction(bool prompt = true){
fp = fopen("tmp.tmp", "w");
if (prompt) {
puts("Press any key to start.");
getchar();
}
auto time = 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);
delete tp;
}
return 0;
auto t = (chrono::high_resolution_clock::now() - time).count();
fclose(fp);
return t;
}
server/threading.cpp
+42 -23
View File
@@ -18,9 +18,9 @@ void ThreadPool::daemon_proc(uint32_t id){
for(; tf[id]; this_thread::sleep_for(*ticking? 0ns:100ms)) {
if (A_TP_HAVE_PAYLOAD(tf[id])) {
A_TP_SET_PAYLOAD(tf[id]);
// A_TP_SET_PAYLOAD(tf[id]);
current_payload[id]();
current_payload[id].empty();
//current_payload[id].empty();
A_TP_UNSET_PAYLOAD(tf[id]);
}
}
@@ -32,31 +32,46 @@ void ThreadPool::tick(){
auto tf = static_cast<atomic<uint8_t>*>(this->thread_flags);
auto ticking = static_cast<atomic<bool>*>(this->ticking);
auto th = static_cast<thread*>(this->thread_handles);
for(; !this->terminate; this_thread::sleep_for(50ms)){
auto n_threads = (uint8_t) this->n_threads;
for(; !this->terminate; this_thread::sleep_for(5ms)){
if(*ticking) {
bool quit = false;
for(; !quit; ){
for(uint32_t i = 0; i < n_threads; ++i){
if(!A_TP_HAVE_PAYLOAD(tf[i])){
pq_lock->lock();
payload_t& p = pq->front();
current_payload[i] = p;
A_TP_SET_PAYLOAD(tf[i]);
pq->pop_front();
quit = !pq->size();
pq_lock->unlock();
if (quit) break;
size_t sz = pq->size();
while(!pq->empty()){
while(!pq->empty() && sz < (n_threads<<10)){
for(uint8_t i = 0; i < n_threads; ++i){
if(!A_TP_HAVE_PAYLOAD(tf[i])){
pq_lock->lock();
current_payload[i] = pq->front();
A_TP_SET_PAYLOAD(tf[i]);
pq->pop_front();
sz = pq->size();
pq_lock->unlock();
if (sz>=(n_threads<<10) || sz == 0) break;
}
}
}
pq_lock->lock();
while(!pq->empty()){
for(uint8_t i = 0; i < n_threads; ++i){
if(!A_TP_HAVE_PAYLOAD(tf[i])){
current_payload[i] = pq->front();
A_TP_SET_PAYLOAD(tf[i]);
pq->pop_front();
if(pq->empty()) break;
}
}
}
pq_lock->unlock();
}
puts("done");
// puts("done");
*ticking = false;
}
}
for (uint32_t i = 0; i < n_threads; ++i)
for (uint8_t i = 0; i < n_threads; ++i)
tf[i] &= 0xfd;
for (uint32_t i = 0; i < n_threads; ++i)
for (uint8_t i = 0; i < n_threads; ++i)
th[i].join();
delete[] th;
@@ -70,7 +85,11 @@ void ThreadPool::tick(){
ThreadPool::ThreadPool(uint32_t n_threads)
: n_threads(n_threads) {
printf("Thread pool started with %u threads;", n_threads);
if (n_threads <= 0){
n_threads = thread::hardware_concurrency();
this->n_threads = n_threads;
}
printf("Thread pool started with %u threads;\n", n_threads);
fflush(stdout);
this->terminate = false;
payload_queue = new deque<payload_t>;
@@ -81,14 +100,15 @@ ThreadPool::ThreadPool(uint32_t n_threads)
thread_flags = tf;
ticking = static_cast<void*>(new atomic<bool>(false));
payload_queue_lock = new mutex();
tick_handle = new thread(&ThreadPool::tick, this);
current_payload = new payload_t[n_threads];
for (uint32_t i = 0; i < n_threads; ++i){
atomic_init(tf + i, 0b10);
th[i] = thread(&ThreadPool::daemon_proc, this, i);
}
payload_queue_lock = new mutex();
tick_handle = new thread(&ThreadPool::tick, this);
current_payload = new payload_t[n_threads];
}
@@ -131,4 +151,3 @@ bool ThreadPool::busy(){
}
return true;
}
server/threading2.cpp
+167
View File
@@ -0,0 +1,167 @@
#include "threading.h"
#include <thread>
#include <atomic>
#include <mutex>
#include <deque>
using namespace std;
using namespace chrono_literals;
#define A_TP_HAVE_PAYLOAD(x) ((x) & 0b1)
#define A_TP_SET_PAYLOAD(x) ((x) |= 0b1)
#define A_TP_UNSET_PAYLOAD(x) ((x) &= 0xfe)
#define A_TP_IS_RUNNING(x) ((x) & 0b10)
void ThreadPool::daemon_proc(uint32_t id){
decltype(auto) tfid = static_cast<atomic<uint8_t>*>(this->thread_flags)[id];
auto ticking = static_cast<atomic<uint16_t>*>(this->ticking);
auto& currentpayload = this->current_payload[id];
auto pq_lock = static_cast<mutex*>(payload_queue_lock);
auto pq = static_cast<deque<payload_t> *>(payload_queue);
bool idle = true;
uint32_t timer = 0;
for(; tfid; ) {
if (A_TP_HAVE_PAYLOAD(tfid)) {
//A_TP_SET_PAYLOAD(tfid);
currentpayload();
// currentpayload.empty();
A_TP_UNSET_PAYLOAD(tfid);
if (idle) {
idle = false;
*ticking -= 1;
timer = 1;
}
}
else if (!pq->empty()) {
pq_lock->lock();
if (!pq->empty()) {
pq->front()();
pq->pop_front();
}
pq_lock->unlock();
if (idle) {
idle = false;
*ticking -= 1;
timer = 1;
}
}
else if (!idle) {
idle = true;
*ticking += 1;
timer = 1;
}
else if (*ticking == n_threads ) {
if (timer > 1000000u)
this_thread::sleep_for(chrono::nanoseconds(timer/100u));
timer = timer > 4200000000u ? 4200000000u : timer*1.0000001 + 1;
}
}
}
void ThreadPool::tick(){
auto pq_lock = static_cast<mutex*>(payload_queue_lock);
auto pq = static_cast<deque<payload_t> *>(payload_queue);
auto tf = static_cast<atomic<uint8_t>*>(this->thread_flags);
auto ticking = static_cast<atomic<uint16_t>*>(this->ticking);
auto th = static_cast<thread*>(this->thread_handles);
//uint32_t threshold = (n_threads << 10u);
for(; !this->terminate; this_thread::sleep_for(50ms)){
// if(*ticking) {
// while(pq->size() > 0)
// {
// bool pqsize = false;
// for(; !pqsize; ){
// for(uint32_t i = 0; i < n_threads; ++i){
// if(!A_TP_HAVE_PAYLOAD(tf[i])){
// pq_lock->lock();
// current_payload[i] = pq->front();
// A_TP_SET_PAYLOAD(tf[i]);
// pq->pop_front();
// pqsize =! pq->size();
// pq_lock->unlock();
// if (pqsize) break;
// }
// }
// }
// }
// // puts("done");
// *ticking = false;
// }
}
for (uint32_t i = 0; i < n_threads; ++i)
tf[i] &= 0xfd;
for (uint32_t i = 0; i < n_threads; ++i)
th[i].join();
delete[] th;
delete[] tf;
delete pq;
delete pq_lock;
delete ticking;
auto cp = static_cast<payload_t*>(current_payload);
delete[] cp;
}
ThreadPool::ThreadPool(uint32_t n_threads)
: n_threads(n_threads) {
printf("Thread pool started with %u threads;\n", n_threads);
fflush(stdout);
this->terminate = false;
payload_queue = new deque<payload_t>;
auto th = new thread[n_threads];
auto tf = new atomic<uint8_t>[n_threads];
thread_handles = th;
thread_flags = tf;
ticking = static_cast<void*>(new atomic<uint16_t>(n_threads));
payload_queue_lock = new mutex();
tick_handle = new thread(&ThreadPool::tick, this);
current_payload = new payload_t[n_threads];
for (uint32_t i = 0; i < n_threads; ++i){
atomic_init(tf + i, 0b10);
th[i] = thread(&ThreadPool::daemon_proc, this, i);
}
}
void ThreadPool::enqueue_task(const payload_t& payload){
auto pq_lock = static_cast<mutex*>(payload_queue_lock);
auto pq = static_cast<deque<payload_t> *>(payload_queue);
auto tf = static_cast<atomic<uint8_t>*>(this->thread_flags);
auto& ticking = *static_cast<atomic<uint16_t>*>(this->ticking);
if (ticking > 0){
for (uint32_t i = 0; i < n_threads; ++i){
if(!A_TP_HAVE_PAYLOAD(tf[i])){
current_payload[i] = payload;
A_TP_SET_PAYLOAD(tf[i]);
return;
}
}
}
pq_lock->lock();
pq->push_back(payload);
pq_lock->unlock();
}
ThreadPool::~ThreadPool() {
this->terminate = true;
auto tick = static_cast<thread*> (tick_handle);
tick->join();
delete tick;
puts("Thread pool terminated.");
}
bool ThreadPool::busy(){
return !(*(atomic<int16_t>*)ticking == n_threads);
}
server/types.h
+1 -1
View File
@@ -49,7 +49,7 @@ namespace types {
#define ULL_Type __uint128_t
#define LL_Type __int128_t
#else
#define F_INT128(__F_)
#define F_INT128(__F_)
#define ULL_Type unsigned long long
#define LL_Type long long
#endif
server/utils.cpp
+1 -1
View File
@@ -8,7 +8,7 @@ string base62uuid(int l = 8) {
static mt19937_64 engine(chrono::system_clock::now().time_since_epoch().count());
static uniform_int_distribution<uint64_t> u(0x10000, 0xfffff);
uint64_t uuid = (u(engine) << 32ull) + (chrono::system_clock::now().time_since_epoch().count() & 0xffffffff);
printf("%llx\n", uuid);
printf("%p\n", uuid);
string ret;
while (uuid && l-- >= 0) {
ret = string("") + base62alp[uuid % 62] + ret;