From 853b1293437571cbe85dba04169e2109acf31fbe Mon Sep 17 00:00:00 2001
From: bill <sunyinqi0508@gmail.com>
Date: Fri, 2 Dec 2022 21:09:59 +0800
Subject: [PATCH] Fix bugs on stored procedures and threaded GC

---
 build.py                     |  2 +-
 header.cxx                   | 12 +++++++++
 msc-plugin/libaquery.vcxproj |  1 +
 msvs-py/msvs-py.pyproj       | 21 ++++++++++------
 reconstruct/storage.py       | 13 +++++-----
 server/gc.h                  |  1 +
 server/libaquery.cpp         |  4 +--
 server/libaquery.h           |  1 -
 server/server.cpp            | 48 +++++++++++++++++++++++++-----------
 server/vector_type.hpp       |  4 +--
 10 files changed, 72 insertions(+), 35 deletions(-)

diff --git a/build.py b/build.py
index 5ce74e8..ec59122 100644
--- a/build.py
+++ b/build.py
@@ -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
diff --git a/header.cxx b/header.cxx
index e69de29..612b4a9 100644
--- a/header.cxx
+++ b/header.cxx
@@ -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__
diff --git a/msc-plugin/libaquery.vcxproj b/msc-plugin/libaquery.vcxproj
index f0d3dd4..82162c9 100644
--- a/msc-plugin/libaquery.vcxproj
+++ b/msc-plugin/libaquery.vcxproj
@@ -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" />
diff --git a/msvs-py/msvs-py.pyproj b/msvs-py/msvs-py.pyproj
index 58b0511..c17a7f4 100644
--- a/msvs-py/msvs-py.pyproj
+++ b/msvs-py/msvs-py.pyproj
@@ -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
diff --git a/reconstruct/storage.py b/reconstruct/storage.py
index 4636bd5..d507342 100644
--- a/reconstruct/storage.py
+++ b/reconstruct/storage.py
@@ -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
diff --git a/server/gc.h b/server/gc.h
index 2a63b3d..1099248 100644
--- a/server/gc.h
+++ b/server/gc.h
@@ -1,3 +1,4 @@
+#pragma once
 #ifndef __AQ_USE_THREADEDGC__
 #include <atomic>
 class GC {
diff --git a/server/libaquery.cpp b/server/libaquery.cpp
index 83abb28..9521ce9 100644
--- a/server/libaquery.cpp
+++ b/server/libaquery.cpp
@@ -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" 
 
diff --git a/server/libaquery.h b/server/libaquery.h
index c6e51d9..e5b9516 100644
--- a/server/libaquery.h
+++ b/server/libaquery.h
@@ -4,7 +4,6 @@
 #ifdef __INTELLISENSE__
 	#define __AQUERY_ITC_USE_SEMPH__
 	#define THREADING
-	#define __AQ_THREADED_GC__
 #endif
 
 #include <unordered_map>
diff --git a/server/server.cpp b/server/server.cpp
index f7cf636..64f6544 100644
--- a/server/server.cpp
+++ b/server/server.cpp
@@ -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
\ No newline at end of file
+#endif // MSCDBG
diff --git a/server/vector_type.hpp b/server/vector_type.hpp
index 10ff661..d69c56c 100644
--- a/server/vector_type.hpp
+++ b/server/vector_type.hpp
@@ -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) \