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

Fix bugs on stored procedures and threaded GC

Browse Source
This commit is contained in:
bill
2022-12-02 21:09:59 +08:00
parent 957e53eb70
commit 853b129343
10 changed files with 72 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
build.py
+1 -1
View File
@@ -184,7 +184,7 @@ class build_manager:
def __init__(self) -> None:
self.method = 'make'
self.cxx = ''
self.OptimizationLv = '4' # [O0, O1, O2, O3, Ofast]
self.OptimizationLv = '0' # [O0, O1, O2, O3, Ofast]
self.Platform = 'amd64'
self.PCH = os.environ['PCH'] if 'PCH' in os.environ else 1
self.StaticLib = 1
header.cxx
+12
View File
@@ -0,0 +1,12 @@
#include "./server/libaquery.h"
#ifndef __AQ_USE_THREADEDGC__
#include "./server/gc.h"
__AQEXPORT__(void) __AQ_Init_GC__(Context* cxt) {
GC::gc_handle = static_cast<GC*>(cxt->gc);
}
#else // __AQ_USE_THREADEDGC__
#define __AQ_Init_GC__(x)
#endif // __AQ_USE_THREADEDGC__
msc-plugin/libaquery.vcxproj
+1
View File
@@ -228,6 +228,7 @@
<ClInclude Include="..\server\priority_vector.hpp" />
<ClInclude Include="..\server\table.h" />
<ClInclude Include="..\server\types.h" />
<ClInclude Include="..\server\unordered_dense.h" />
<ClInclude Include="..\server\utils.h" />
<ClInclude Include="..\server\vector_type.hpp" />
<ClInclude Include="..\server\winhelper.h" />
msvs-py/msvs-py.pyproj
+13 -8
View File
@@ -11,6 +11,9 @@
<Name>msvs-py</Name>
<RootNamespace>msvs-py</RootNamespace>
<IsWindowsApplication>False</IsWindowsApplication>
<InterpreterId>Global|PythonCore|3.9</InterpreterId>
<LaunchProvider>Standard Python launcher</LaunchProvider>
<EnableNativeCodeDebugging>True</EnableNativeCodeDebugging>
</PropertyGroup>
<PropertyGroup Condition=" '$(Configuration)' == 'Debug' ">
<DebugSymbols>true</DebugSymbols>
@@ -41,16 +44,18 @@
<Compile Include="..\engine\projection.py" />
<Compile Include="..\engine\scan.py" />
<Compile Include="..\engine\types.py" />
<Compile Include="..\reconstruct\storage.py" />
<Compile Include="..\reconstruct\expr.py" />
<Compile Include="..\reconstruct\__init__.py" />
<Compile Include="..\reconstruct\ast.py" />
<Compile Include="..\reconstruct\storage.py" />
<Compile Include="..\reconstruct\expr.py" />
<Compile Include="..\reconstruct\__init__.py" />
<Compile Include="..\reconstruct\ast.py" />
<Compile Include="..\engine\__init__.py" />
<Compile Include="..\prompt.py" />
<Compile Include="..\build.py" />
<Compile Include="..\aquery_config.py" />
<Compile Include="..\prompt.py" />
<Compile Include="..\build.py" />
<Compile Include="..\aquery_config.py" />
</ItemGroup>
<ItemGroup>
<InterpreterReference Include="Global|PythonCore|3.9" />
</ItemGroup>
<Import Project="$(MSBuildExtensionsPath32)\Microsoft\VisualStudio\v$(VisualStudioVersion)\Python Tools\Microsoft.PythonTools.targets" />
<!-- Uncomment the CoreCompile target to enable the Build command in
Visual Studio and specify your pre- and post-build commands in
reconstruct/storage.py
+7 -6
View File
@@ -144,10 +144,9 @@ class TableInfo:
class Context:
def new(self):
self.headers = set(['\"./server/libaquery.h\"',
'\"./server/monetdb_conn.h\"'])
self.headers = set(['\"./server/monetdb_conn.h\"'])
self.ccode = ''
self.sql = ''
self.finalized = False
self.udf = None
@@ -175,7 +174,8 @@ class Context:
self.have_hge = False
self.Error = lambda *args: print(*args)
self.Info = lambda *_: None
# self.new() called everytime new query batch is started
def get_scan_var(self):
it_var = 'i' + base62uuid(2)
scan_vars = set(s.it_var for s in self.scans)
@@ -277,7 +277,8 @@ class Context:
headers = ''
# if build_driver == 'MSBuild':
# headers ='#include \"./server/pch.hpp\"\n'
with open('header.cxx', 'r') as header:
headers += header.read()
for h in self.headers:
if h[0] != '"':
headers += '#include <' + h + '>\n'
@@ -287,6 +288,6 @@ class Context:
headers += '#undef max\n'
headers += '#undef min\n'
self.ccode = headers + '\n'.join(self.procs)
self.ccode += headers + '\n'.join(self.procs)
self.headers = set()
return self.ccode
server/gc.h
+1
View File
@@ -1,3 +1,4 @@
#pragma once
#ifndef __AQ_USE_THREADEDGC__
#include <atomic>
class GC {
server/libaquery.cpp
+2 -2
View File
@@ -456,14 +456,14 @@ void GC::reg(void* v, uint32_t sz, void(*f)(void*)) { //~ 40ns expected v. free
++alive_cnt;
current_size += sz;
auto _slot = (slot_pos += 1);
_q[_slot] = {v, f};
_q[_slot-1] = {v, f};
--alive_cnt;
running = true;
}
#endif
GC* GC::gc_handle = nullptr;
inline GC* GC::gc_handle = nullptr;
#include "dragonbox/dragonbox_to_chars.hpp"
server/libaquery.h
-1
View File
@@ -4,7 +4,6 @@
#ifdef __INTELLISENSE__
#define __AQUERY_ITC_USE_SEMPH__
#define THREADING
#define __AQ_THREADED_GC__
#endif
#include <unordered_map>
server/server.cpp
+33 -15
View File
@@ -285,10 +285,10 @@ 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;
std::string pwd = (char*)(std::filesystem::current_path().string().c_str());
char sep = std::filesystem::path::preferred_separator;
pwd += sep;
std::string procedure_root = pwd + "procedures" + sep;
std::string procedure_root = (pwd + "procedures") + sep;
std::string procedure_name = "";
StoredProcedure current_procedure;
vector_type<char *> recorded_queries;
@@ -296,6 +296,18 @@ int dll_main(int argc, char** argv, Context* cxt){
bool procedure_recording = false,
procedure_replaying = false;
uint32_t procedure_module_cursor = 0;
try {
if (!std::filesystem::is_directory(procedure_root)) {
if (std::filesystem::exists(procedure_root))
std::filesystem::remove_all(procedure_root);
}
if (!std::filesystem::exists(procedure_root)) {
std::filesystem::create_directory(procedure_root);
}
}
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*>();
@@ -351,6 +363,12 @@ start:
recorded_queries.emplace_back(copy_lpstr("N"));
}
handle = dlopen(proc_name, RTLD_NOW);
#ifndef __AQ_USE_THREADEDGC__
{
typedef void (*aq_gc_init_t) (Context*);
((aq_gc_init_t)dlsym(handle, "__AQ_Init_GC__"))(cxt);
}
#endif
if (procedure_recording) {
recorded_libraries.emplace_back(handle);
}
@@ -380,6 +398,7 @@ start:
recorded_queries.emplace_back(copy_lpstr(n_recvd[i]));
}
code_snippet c = reinterpret_cast<code_snippet>(dlsym(handle, n_recvd[i]+1));
printf("%p", dlsym(handle, n_recvd[i] + 1));
timer.reset();
c(cxt);
cfg->stats.postproc_time += timer.elapsed();
@@ -543,6 +562,7 @@ start:
n_recv = current_procedure.cnt;
n_recvd = current_procedure.queries;
load_modules(current_procedure);
procedure_replaying = true;
goto start; // yes, I know, refactor later!!
}
}
@@ -568,12 +588,7 @@ start:
break;
}
}
if(handle &&
!procedure_replaying && !procedure_recording) {
printf("Destroy %p\n", handle);
dlclose(handle);
handle = nullptr;
}
printf("%lld, %lld", cfg->stats.monet_time, cfg->stats.postproc_time);
cxt->end_session();
n_recv = 0;
@@ -591,7 +606,13 @@ start:
code_snippet c = reinterpret_cast<code_snippet>(dlsym(handle, "dllmain"));
c(cxt);
}
if (handle) dlclose(handle);
if (handle &&
!procedure_replaying && !procedure_recording) {
printf("Destroy %p\n", handle);
dlclose(handle);
handle = nullptr;
}
procedure_replaying = false;
cfg->new_query = 0;
}
//puts(cfg->running? "true": "false");
@@ -624,7 +645,7 @@ int launcher(int argc, char** argv){
str = std::string("cd ") + pwd + std::string("&& python3 ./prompt.py ") + str;
return system(str.c_str());
}
#if !( defined(_MSC_VER) && defined(_DEBUG) )
#if true || !( defined(_MSC_VER) && defined(_DEBUG) )
extern "C" int __DLLEXPORT__ main(int argc, char** argv) {
#ifdef __AQ_BUILD_LAUNCHER__
return launcher(argc, argv);
@@ -639,9 +660,6 @@ extern "C" int __DLLEXPORT__ main(int argc, char** argv) {
cxt->thread_pool = tp;
#endif
#ifdef __AQ_THREADED_GC__
cxt->gc_thread = new std::thread(gc_thread, cxt);
#endif
const char* shmname;
if (argc < 0)
return dll_main(argc, argv, cxt);
@@ -678,4 +696,4 @@ extern "C" int __DLLEXPORT__ main(int argc, char** argv) {
return 0;
}
#endif // MSCDBG
#endif // MSCDBG
server/vector_type.hpp
+2 -2
View File
@@ -15,7 +15,7 @@
#include <iostream>
#include "hasher.h"
#include "types.h"
#include "gc.h"
#pragma pack(push, 1)
template<class T>
struct vector_base {};
@@ -325,7 +325,7 @@ public:
inline vector_type<_Ty> subvec_deep(uint32_t start = 0) const { return subvec_deep(start, size); }
vector_type<_Ty> getRef() { return vector_type<_Ty>(container, size); }
~vector_type() {
if (capacity > 0) free(container);
if (capacity > 0) GC::gc_handle->reg(container, sizeof(_Ty) * capacity);//free(container);
container = 0; size = capacity = 0;
}
#define Compare(_op) \