Add global version allocators for C in query modules (#162)

tests/e2e/memory/CMakeLists.txt

@@ -1,2 +1,11 @@
+add_subdirectory(procedures)
+
 add_executable(memgraph__e2e__memory__control memory_control.cpp)
 target_link_libraries(memgraph__e2e__memory__control gflags mgclient mg-utils mg-io Threads::Threads)
+
+add_executable(memgraph__e2e__memory__limit_global_alloc memory_limit_global_alloc.cpp)
+target_link_libraries(memgraph__e2e__memory__limit_global_alloc gflags mgclient mg-utils mg-io Threads::Threads)
+
+add_executable(memgraph__e2e__memory__limit_global_alloc_proc memory_limit_global_alloc_proc.cpp)
+target_link_libraries(memgraph__e2e__memory__limit_global_alloc_proc gflags mgclient mg-utils mg-io Threads::Threads)
+

tests/e2e/memory/memory_limit_global_alloc.cpp

@@ -0,0 +1,26 @@
+#include <gflags/gflags.h>
+#include <mgclient.hpp>
+
+#include "utils/logging.hpp"
+#include "utils/timer.hpp"
+
+DEFINE_uint64(bolt_port, 7687, "Bolt port");
+DEFINE_uint64(timeout, 120, "Timeout seconds");
+
+int main(int argc, char **argv) {
+  google::SetUsageMessage("Memgraph E2E Memory Limit For Global Allocators");
+  gflags::ParseCommandLineFlags(&argc, &argv, true);
+  logging::RedirectToStderr();
+
+  mg::Client::Init();
+
+  auto client =
+      mg::Client::Connect({.host = "127.0.0.1", .port = static_cast<uint16_t>(FLAGS_bolt_port), .use_ssl = false});
+  if (!client) {
+    LOG_FATAL("Failed to connect!");
+  }
+
+  bool result = client->Execute("CALL libglobal_memory_limit.procedure() YIELD *");
+  MG_ASSERT(result == false);
+  return 0;
+}

tests/e2e/memory/memory_limit_global_alloc_proc.cpp

@@ -0,0 +1,31 @@
+#include <gflags/gflags.h>
+#include <mgclient.hpp>
+#include <algorithm>
+
+#include "utils/logging.hpp"
+#include "utils/timer.hpp"
+
+DEFINE_uint64(bolt_port, 7687, "Bolt port");
+DEFINE_uint64(timeout, 120, "Timeout seconds");
+
+int main(int argc, char **argv) {
+  google::SetUsageMessage("Memgraph E2E Memory Limit For Global Allocators");
+  gflags::ParseCommandLineFlags(&argc, &argv, true);
+  logging::RedirectToStderr();
+
+  mg::Client::Init();
+
+  auto client =
+      mg::Client::Connect({.host = "127.0.0.1", .port = static_cast<uint16_t>(FLAGS_bolt_port), .use_ssl = false});
+  if (!client) {
+    LOG_FATAL("Failed to connect!");
+  }
+  bool result = client->Execute("CALL libglobal_memory_limit_proc.error() YIELD *");
+  auto result1 = client->FetchAll();
+  MG_ASSERT(result1 != std::nullopt && result1->size() == 0);
+
+  result = client->Execute("CALL libglobal_memory_limit_proc.success() YIELD *");
+  auto result2 = client->FetchAll();
+  MG_ASSERT(result2 != std::nullopt && result2->size() > 0);
+  return 0;
+}

tests/e2e/memory/procedures/CMakeLists.txt

@@ -0,0 +1,5 @@
+add_library(global_memory_limit SHARED global_memory_limit.c)
+target_include_directories(global_memory_limit PRIVATE ${CMAKE_SOURCE_DIR}/include)
+
+add_library(global_memory_limit_proc SHARED global_memory_limit_proc.c)
+target_include_directories(global_memory_limit_proc PRIVATE ${CMAKE_SOURCE_DIR}/include)

tests/e2e/memory/procedures/global_memory_limit.c

@@ -0,0 +1,36 @@
+#include "mg_procedure.h"
+
+int *gVal = NULL;
+
+void set_error(struct mgp_result *result) { mgp_result_set_error_msg(result, "Something went wrong"); }
+
+static void procedure(const struct mgp_list *args, const struct mgp_graph *graph, struct mgp_result *result,
+                      struct mgp_memory *memory) {
+  struct mgp_result_record *record = mgp_result_new_record(result);
+  if (record == NULL) return set_error(result);
+
+  struct mgp_value *result_msg = mgp_value_make_string("mgp_init_module allocation works", memory);
+  if (result_msg == NULL) return set_error(result);
+
+  int result_inserted = mgp_result_record_insert(record, "result", result_msg);
+  mgp_value_destroy(result_msg);
+  if (!result_inserted) return set_error(result);
+}
+
+int mgp_init_module(struct mgp_module *module, struct mgp_memory *memory) {
+  const size_t one_gb = 1 << 30;
+  gVal = mgp_global_alloc(one_gb);
+  if (!gVal) return 1;
+
+  struct mgp_proc *proc = mgp_module_add_read_procedure(module, "procedure", procedure);
+  if (!proc) return 1;
+
+  if (!mgp_proc_add_result(proc, "result", mgp_type_string())) return 1;
+
+  return 0;
+}
+
+int mgp_shutdown_module() {
+  if (gVal) mgp_global_free(gVal);
+  return 0;
+}

tests/e2e/memory/procedures/global_memory_limit_proc.c

@@ -0,0 +1,63 @@
+#include "mg_procedure.h"
+
+int *gVal = NULL;
+
+void set_error(struct mgp_result *result) { mgp_result_set_error_msg(result, "Something went wrong"); }
+
+void set_out_of_memory_error(struct mgp_result *result) { mgp_result_set_error_msg(result, "Out of memory"); }
+
+static void error(const struct mgp_list *args, const struct mgp_graph *graph, struct mgp_result *result,
+                  struct mgp_memory *memory) {
+  const size_t one_gb = 1 << 30;
+  if (gVal) {
+    mgp_global_free(gVal);
+    gVal = NULL;
+  }
+  if (!gVal) {
+    gVal = mgp_global_alloc(one_gb);
+    if (!gVal) return set_out_of_memory_error(result);
+  }
+  struct mgp_result_record *record = mgp_result_new_record(result);
+  if (record == NULL) return set_error(result);
+  struct mgp_value *error_value = mgp_value_make_string("ERROR", memory);
+  if (error_value == NULL) return set_error(result);
+  int result_inserted = mgp_result_record_insert(record, "error_result", error_value);
+  mgp_value_destroy(error_value);
+  if (!result_inserted) return set_error(result);
+}
+
+static void success(const struct mgp_list *args, const struct mgp_graph *graph, struct mgp_result *result,
+                    struct mgp_memory *memory) {
+  const size_t bytes = 1024;
+  if (!gVal) {
+    gVal = mgp_global_alloc(bytes);
+    if (!gVal) set_out_of_memory_error(result);
+  }
+
+  struct mgp_result_record *record = mgp_result_new_record(result);
+  if (record == NULL) return set_error(result);
+  struct mgp_value *success_value = mgp_value_make_string("sucess", memory);
+  if (success_value == NULL) return set_error(result);
+  int result_inserted = mgp_result_record_insert(record, "success_result", success_value);
+  mgp_value_destroy(success_value);
+  if (!result_inserted) return set_error(result);
+}
+
+int mgp_init_module(struct mgp_module *module, struct mgp_memory *memory) {
+  struct mgp_proc *error_proc = mgp_module_add_read_procedure(module, "error", error);
+  if (!error_proc) return 1;
+
+  if (!mgp_proc_add_result(error_proc, "error_result", mgp_type_string())) return 1;
+
+  struct mgp_proc *succ_proc = mgp_module_add_read_procedure(module, "success", success);
+  if (!succ_proc) return 1;
+
+  if (!mgp_proc_add_result(succ_proc, "success_result", mgp_type_string())) return 1;
+
+  return 0;
+}
+
+int mgp_shutdown_module() {
+  if (gVal) mgp_global_free(gVal);
+  return 0;
+}

tests/e2e/memory/workloads.yaml

@@ -13,4 +13,14 @@ workloads:
     args: ["--bolt-port", *bolt_port, "--timeout", "180"]
     <<: *template_cluster
 
+  - name: "Memory limit for modules upon loading"
+    binary: "tests/e2e/memory/memgraph__e2e__memory__limit_global_alloc"
+    args: ["--bolt-port", *bolt_port, "--timeout", "180"]
+    proc: "tests/e2e/memory/procedures/"
+    <<: *template_cluster
 
+  - name: "Memory limit for modules inside a procedure"
+    binary: "tests/e2e/memory/memgraph__e2e__memory__limit_global_alloc_proc"
+    args: ["--bolt-port", *bolt_port, "--timeout", "180"]
+    proc: "tests/e2e/memory/procedures/"
+    <<: *template_cluster

tests/e2e/runner.py

@@ -51,6 +51,10 @@ def run(args):
             mg_instances[name] = mg_instance
             log_file_path = os.path.join(BUILD_DIR, 'logs', config['log_file'])
             binary_args = config['args'] + ["--log-file", log_file_path]
+            if 'proc' in workload:
+              procdir = "--query-modules-directory=" + os.path.join(BUILD_DIR, workload['proc'])
+              binary_args.append(procdir)
+
             mg_instance.start(args=binary_args)
             for query in config['setup_queries']:
                 mg_instance.query(query)