From 11ee19516aa3ac3595da7058e012de4b30eae6cc Mon Sep 17 00:00:00 2001 From: antoniofilipovic Date: Tue, 10 Oct 2023 18:51:24 +0200 Subject: [PATCH] add limits --- include/mg_procedure.h | 10 +- src/memory/memory_control.cpp | 9 +- src/memory/memory_control.hpp | 2 + src/query/db_accessor.cpp | 8 ++ src/query/db_accessor.hpp | 17 +++ src/query/procedure/mg_procedure_impl.cpp | 13 +++ tests/e2e/memory/procedures/CMakeLists.txt | 4 + .../query_memory_limit_proc_multi_thread.cpp | 110 ++++++++++++++++++ tests/e2e/memory/query_memory_limit.cpp | 65 +++++++++++ tests/e2e/memory/workloads.yaml | 101 +++++++++------- 10 files changed, 294 insertions(+), 45 deletions(-) create mode 100644 tests/e2e/memory/procedures/query_memory_limit_proc_multi_thread.cpp create mode 100644 tests/e2e/memory/query_memory_limit.cpp diff --git a/include/mg_procedure.h b/include/mg_procedure.h index cf908ddce..9bff63471 100644 --- a/include/mg_procedure.h +++ b/include/mg_procedure.h @@ -111,6 +111,13 @@ enum mgp_error mgp_global_aligned_alloc(size_t size_in_bytes, size_t alignment, /// The behavior is undefined if `ptr` is not a value returned from a prior /// mgp_global_alloc() or mgp_global_aligned_alloc(). void mgp_global_free(void *p); + +/// State of the graph database. +struct mgp_graph; + +enum mgp_error mgp_track_thread_allocations(struct mgp_graph *graph, const char *thread_id); + +enum mgp_error mgp_untrack_thread_allocations(struct mgp_graph *graph, const char *); ///@} /// @name Operations on mgp_value @@ -851,9 +858,6 @@ enum mgp_error mgp_edge_set_properties(struct mgp_edge *e, struct mgp_map *prope enum mgp_error mgp_edge_iter_properties(struct mgp_edge *e, struct mgp_memory *memory, struct mgp_properties_iterator **result); -/// State of the graph database. -struct mgp_graph; - /// Get the vertex corresponding to given ID, or NULL if no such vertex exists. /// Resulting vertex must be freed using mgp_vertex_destroy. /// Return mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE if unable to allocate the vertex. diff --git a/src/memory/memory_control.cpp b/src/memory/memory_control.cpp index ef10202fa..6b04b6636 100644 --- a/src/memory/memory_control.cpp +++ b/src/memory/memory_control.cpp @@ -267,7 +267,12 @@ void UpdateThreadToTransactionId(const std::thread::id &thread_id, uint64_t tran std::ostringstream oss; oss << thread_id; thread_id_to_transaction_id[oss.str()] = transaction_id; - std::cout << "set:" << oss.str() << std::endl; + std::cout << "set tracking for thread:" << oss.str() << ", on transaction: " << transaction_id << std::endl; +} + +void UpdateThreadToTransactionId(const char *thread_id, uint64_t transaction_id) { + thread_id_to_transaction_id[std::string(thread_id)] = transaction_id; + std::cout << "set tracking:" << std::string(thread_id) << ", on transaction: " << transaction_id << std::endl; } void ResetThreadToTransactionId(const std::thread::id &thread_id) { @@ -276,6 +281,8 @@ void ResetThreadToTransactionId(const std::thread::id &thread_id) { thread_id_to_transaction_id.erase(oss.str()); } +void ResetThreadToTransactionId(const char *thread_id) { thread_id_to_transaction_id.erase(std::string(thread_id)); } + void PurgeUnusedMemory() { #if USE_JEMALLOC mallctl("arena." STRINGIFY(MALLCTL_ARENAS_ALL) ".purge", nullptr, nullptr, nullptr, 0); diff --git a/src/memory/memory_control.hpp b/src/memory/memory_control.hpp index a30e36d0e..298ff003d 100644 --- a/src/memory/memory_control.hpp +++ b/src/memory/memory_control.hpp @@ -27,6 +27,8 @@ bool AddTrackingOnArena(unsigned); bool RemoveTrackingOnArena(unsigned); void UpdateThreadToTransactionId(const std::thread::id &, uint64_t); void ResetThreadToTransactionId(const std::thread::id &); +void UpdateThreadToTransactionId(const char *, uint64_t); +void ResetThreadToTransactionId(const char *); // TODO(AF) : Do we need arena to transaction id and transaction_id to tracker? inline std::unordered_map thread_id_to_transaction_id; diff --git a/src/query/db_accessor.cpp b/src/query/db_accessor.cpp index cef047a24..c9433291d 100644 --- a/src/query/db_accessor.cpp +++ b/src/query/db_accessor.cpp @@ -21,6 +21,14 @@ namespace memgraph::query { SubgraphDbAccessor::SubgraphDbAccessor(query::DbAccessor db_accessor, Graph *graph) : db_accessor_(db_accessor), graph_(graph) {} +void SubgraphDbAccessor::TrackThreadAllocations(const char *thread_id) { + return db_accessor_.TrackThreadAllocations(thread_id); +} + +void SubgraphDbAccessor::UntrackThreadAllocations(const char *thread_id) { + return db_accessor_.UntrackThreadAllocations(thread_id); +} + storage::PropertyId SubgraphDbAccessor::NameToProperty(const std::string_view name) { return db_accessor_.NameToProperty(name); } diff --git a/src/query/db_accessor.hpp b/src/query/db_accessor.hpp index 0028c9ae7..a9a92f492 100644 --- a/src/query/db_accessor.hpp +++ b/src/query/db_accessor.hpp @@ -17,6 +17,7 @@ #include #include +#include "memory/memory_control.hpp" #include "query/exceptions.hpp" #include "storage/v2/edge_accessor.hpp" #include "storage/v2/id_types.hpp" @@ -372,6 +373,18 @@ class DbAccessor final { void FinalizeTransaction() { accessor_->FinalizeTransaction(); } + void TrackThreadAllocations(const char *thread_id) { + memgraph::memory::UpdateThreadToTransactionId(thread_id, *accessor_->GetTransactionId()); + auto arena = memgraph::memory::GetArenaForThread(); + memgraph::memory::AddTrackingOnArena(arena); + } + + void UntrackThreadAllocations(const char *thread_id) { + memgraph::memory::ResetThreadToTransactionId(thread_id); + auto arena = memgraph::memory::GetArenaForThread(); + memgraph::memory::RemoveTrackingOnArena(arena); + } + std::optional GetTransactionId() { return accessor_->GetTransactionId(); } VerticesIterable Vertices(storage::View view) { return VerticesIterable(accessor_->Vertices(view)); } @@ -645,6 +658,10 @@ class SubgraphDbAccessor final { static SubgraphDbAccessor *MakeSubgraphDbAccessor(DbAccessor *db_accessor, Graph *graph); + void TrackThreadAllocations(const char *thread_id); + + void UntrackThreadAllocations(const char *thread_id); + storage::PropertyId NameToProperty(std::string_view name); storage::LabelId NameToLabel(std::string_view name); diff --git a/src/query/procedure/mg_procedure_impl.cpp b/src/query/procedure/mg_procedure_impl.cpp index 0294b6f73..587f8b580 100644 --- a/src/query/procedure/mg_procedure_impl.cpp +++ b/src/query/procedure/mg_procedure_impl.cpp @@ -3540,3 +3540,16 @@ mgp_error mgp_log(const mgp_log_level log_level, const char *output) { throw std::invalid_argument{fmt::format("Invalid log level: {}", log_level)}; }); } + +mgp_error mgp_track_thread_allocations(mgp_graph *graph, const char *thread_id) { + return WrapExceptions([&]() { + std::visit([thread_id](auto *db_accessor) -> void { db_accessor->TrackThreadAllocations(thread_id); }, graph->impl); + }); +} + +mgp_error mgp_untrack_thread_allocations(mgp_graph *graph, const char *thread_id) { + return WrapExceptions([&]() { + std::visit([thread_id](auto *db_accessor) -> void { db_accessor->UntrackThreadAllocations(thread_id); }, + graph->impl); + }); +} diff --git a/tests/e2e/memory/procedures/CMakeLists.txt b/tests/e2e/memory/procedures/CMakeLists.txt index 21201e59b..efb141f20 100644 --- a/tests/e2e/memory/procedures/CMakeLists.txt +++ b/tests/e2e/memory/procedures/CMakeLists.txt @@ -3,3 +3,7 @@ target_include_directories(global_memory_limit PRIVATE ${CMAKE_SOURCE_DIR}/inclu add_library(global_memory_limit_proc SHARED global_memory_limit_proc.c) target_include_directories(global_memory_limit_proc PRIVATE ${CMAKE_SOURCE_DIR}/include) + + +add_library(query_memory_limit_proc_multi_thread SHARED query_memory_limit_proc_multi_thread.cpp) +target_include_directories(query_memory_limit_proc_multi_thread PRIVATE ${CMAKE_SOURCE_DIR}/include) #todo link with utils not to use onscopeexit diff --git a/tests/e2e/memory/procedures/query_memory_limit_proc_multi_thread.cpp b/tests/e2e/memory/procedures/query_memory_limit_proc_multi_thread.cpp new file mode 100644 index 000000000..2a21d154a --- /dev/null +++ b/tests/e2e/memory/procedures/query_memory_limit_proc_multi_thread.cpp @@ -0,0 +1,110 @@ +// Copyright 2023 Memgraph Ltd. +// +// Use of this software is governed by the Business Source License +// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source +// License, and you may not use this file except in compliance with the Business Source License. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0, included in the file +// licenses/APL.txt. + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include "mg_procedure.h" + +template +class [[nodiscard]] OnScopeExit { + public: + explicit OnScopeExit(Callable &&function) : function_{std::forward(function)}, doCall_{true} {} + OnScopeExit(OnScopeExit const &) = delete; + OnScopeExit(OnScopeExit &&) = delete; + OnScopeExit &operator=(OnScopeExit const &) = delete; + OnScopeExit &operator=(OnScopeExit &&) = delete; + ~OnScopeExit() { + if (doCall_) function_(); + } + + void Disable() { doCall_ = false; } + + private: + std::function function_; + bool doCall_; +}; +template +OnScopeExit(Callable &&) -> OnScopeExit; + +// TODO(af) fix this hack +std::atomic num_allocations{0}; + +void AllocFunc(mgp_graph *graph) { + std::ostringstream oss; + oss << std::this_thread::get_id(); + std::string thread_id = oss.str(); + [[maybe_unused]] const enum mgp_error tracking_error = mgp_track_thread_allocations(graph, thread_id.c_str()); + const size_t two_sixty_eight_mb = 1 << 28; + void *ptr = nullptr; + + OnScopeExit> cleanup{[&ptr]() { + if (nullptr == ptr) { + return; + } + mgp_global_free(ptr); + }}; + + const enum mgp_error alloc_err = mgp_global_alloc(two_sixty_eight_mb, (void **)(&ptr)); + if (alloc_err != mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE) { + num_allocations.fetch_add(1, std::memory_order_relaxed); + } + + [[maybe_unused]] const enum mgp_error untracking_error = mgp_untrack_thread_allocations(graph, thread_id.c_str()); +} + +void DualThread(mgp_list *args, mgp_graph *memgraph_graph, mgp_result *result, mgp_memory *memory) { + mgp::MemoryDispatcherGuard guard{memory}; + const auto arguments = mgp::List(args); + const auto record_factory = mgp::RecordFactory(result); + + try { + std::vector threads; + + for (int i = 0; i < 2; i++) { + threads.emplace_back(AllocFunc, memgraph_graph); + } + + for (int i = 0; i < 2; i++) { + threads[i].join(); + } + + auto new_record = record_factory.NewRecord(); + + new_record.Insert("allocated_all", num_allocations.load(std::memory_order_relaxed) == 2); + // ASSERT only one exception should occur + } catch (std::exception &e) { + record_factory.SetErrorMessage(e.what()); + } +} + +extern "C" int mgp_init_module(struct mgp_module *module, struct mgp_memory *memory) { + try { + mgp::memory = memory; + + AddProcedure(DualThread, std::string("dual_thread").c_str(), mgp::ProcedureType::Read, {}, + {mgp::Return(std::string("allocated_all").c_str(), mgp::Type::Bool)}, module, memory); + + } catch (const std::exception &e) { + return 1; + } + + return 0; +} + +extern "C" int mgp_shutdown_module() { return 0; } diff --git a/tests/e2e/memory/query_memory_limit.cpp b/tests/e2e/memory/query_memory_limit.cpp new file mode 100644 index 000000000..3e7c0c6c5 --- /dev/null +++ b/tests/e2e/memory/query_memory_limit.cpp @@ -0,0 +1,65 @@ +// Copyright 2023 Memgraph Ltd. +// +// Use of this software is governed by the Business Source License +// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source +// License, and you may not use this file except in compliance with the Business Source License. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0, included in the file +// licenses/APL.txt. + +#include +#include +#include +#include + +#include "utils/logging.hpp" +#include "utils/timer.hpp" + +DEFINE_uint64(bolt_port, 7687, "Bolt port"); +DEFINE_uint64(timeout, 120, "Timeout seconds"); +DEFINE_bool(multi_db, false, "Run test in multi db environment"); + +int main(int argc, char **argv) { + google::SetUsageMessage("Memgraph E2E Query Memory Limit In Multi-Thread For Global Allocators"); + gflags::ParseCommandLineFlags(&argc, &argv, true); + memgraph::logging::RedirectToStderr(); + + mg::Client::Init(); + + auto client = + mg::Client::Connect({.host = "127.0.0.1", .port = static_cast(FLAGS_bolt_port), .use_ssl = false}); + if (!client) { + LOG_FATAL("Failed to connect!"); + } + + if (FLAGS_multi_db) { + client->Execute("CREATE DATABASE clean;"); + client->DiscardAll(); + client->Execute("USE DATABASE clean;"); + client->DiscardAll(); + client->Execute("MATCH (n) DETACH DELETE n;"); + client->DiscardAll(); + } + + MG_ASSERT( + client->Execute("CALL libquery_memory_limit_proc_multi_thread.dual_thread() YIELD allocated_all RETURN " + "allocated_all QUERY MEMORY LIMIT 500MB")); + bool error{false}; + try { + auto result_rows = client->FetchAll(); + if (result_rows) { + auto row = *result_rows->begin(); + std::cout << row[0].ValueBool() << std::endl; + } + + } catch (const mg::ClientException &e) { + std::cout << e.what() << std::endl; + error = true; + } + + MG_ASSERT(error, "Error should have happend"); + + return 0; +} diff --git a/tests/e2e/memory/workloads.yaml b/tests/e2e/memory/workloads.yaml index 460d6bc3f..3d519b364 100644 --- a/tests/e2e/memory/workloads.yaml +++ b/tests/e2e/memory/workloads.yaml @@ -23,50 +23,69 @@ disk_cluster: &disk_cluster - "STORAGE MODE ON_DISK_TRANSACTIONAL" validation_queries: [] +args_query_limit: &args_query_limit + - "--bolt-port" + - *bolt_port + - "--storage-gc-cycle-sec=180" + - "--log-level=TRACE" + +in_memory_query_limit_cluster: &in_memory_query_limit_cluster + cluster: + main: + args: *args_query_limit + log_file: "memory-e2e.log" + setup_queries: [] + validation_queries: [] workloads: - - name: "Memory control" - binary: "tests/e2e/memory/memgraph__e2e__memory__control" - args: ["--bolt-port", *bolt_port, "--timeout", "180"] - <<: *in_memory_cluster + # - name: "Memory control" + # binary: "tests/e2e/memory/memgraph__e2e__memory__control" + # args: ["--bolt-port", *bolt_port, "--timeout", "180"] + # <<: *in_memory_cluster - - name: "Memory control multi database" - binary: "tests/e2e/memory/memgraph__e2e__memory__control" - args: ["--bolt-port", *bolt_port, "--timeout", "180", "--multi-db", "true"] - <<: *in_memory_cluster + # - name: "Memory control multi database" + # binary: "tests/e2e/memory/memgraph__e2e__memory__control" + # args: ["--bolt-port", *bolt_port, "--timeout", "180", "--multi-db", "true"] + # <<: *in_memory_cluster - - name: "Memory limit for modules upon loading" - binary: "tests/e2e/memory/memgraph__e2e__memory__limit_global_alloc" + # - 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/" + # <<: *in_memory_cluster + + # - name: "Memory limit for modules upon loading multi database" + # binary: "tests/e2e/memory/memgraph__e2e__memory__limit_global_alloc" + # args: ["--bolt-port", *bolt_port, "--timeout", "180", "--multi-db", "true"] + # proc: "tests/e2e/memory/procedures/" + # <<: *in_memory_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/" + # <<: *in_memory_cluster + + # - name: "Memory limit for modules inside a procedure multi database" + # binary: "tests/e2e/memory/memgraph__e2e__memory__limit_global_alloc_proc" + # args: ["--bolt-port", *bolt_port, "--timeout", "180", "--multi-db", "true"] + # proc: "tests/e2e/memory/procedures/" + # <<: *in_memory_cluster + + # - name: "Memory limit for modules upon loading for on-disk storage" + # binary: "tests/e2e/memory/memgraph__e2e__memory__limit_global_alloc" + # args: ["--bolt-port", *bolt_port, "--timeout", "180"] + # proc: "tests/e2e/memory/procedures/" + # <<: *disk_cluster + + # - name: "Memory limit for modules inside a procedure for on-disk storage" + # binary: "tests/e2e/memory/memgraph__e2e__memory__limit_global_alloc_proc" + # args: ["--bolt-port", *bolt_port, "--timeout", "180"] + # proc: "tests/e2e/memory/procedures/" + # <<: *disk_cluster + + - name: "Memory control query limit" + binary: "tests/e2e/memory/memgraph__e2e__memory__limit_query_alloc_proc_multi_thread" args: ["--bolt-port", *bolt_port, "--timeout", "180"] proc: "tests/e2e/memory/procedures/" - <<: *in_memory_cluster - - - name: "Memory limit for modules upon loading multi database" - binary: "tests/e2e/memory/memgraph__e2e__memory__limit_global_alloc" - args: ["--bolt-port", *bolt_port, "--timeout", "180", "--multi-db", "true"] - proc: "tests/e2e/memory/procedures/" - <<: *in_memory_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/" - <<: *in_memory_cluster - - - name: "Memory limit for modules inside a procedure multi database" - binary: "tests/e2e/memory/memgraph__e2e__memory__limit_global_alloc_proc" - args: ["--bolt-port", *bolt_port, "--timeout", "180", "--multi-db", "true"] - proc: "tests/e2e/memory/procedures/" - <<: *in_memory_cluster - - - name: "Memory limit for modules upon loading for on-disk storage" - binary: "tests/e2e/memory/memgraph__e2e__memory__limit_global_alloc" - args: ["--bolt-port", *bolt_port, "--timeout", "180"] - proc: "tests/e2e/memory/procedures/" - <<: *disk_cluster - - - name: "Memory limit for modules inside a procedure for on-disk storage" - binary: "tests/e2e/memory/memgraph__e2e__memory__limit_global_alloc_proc" - args: ["--bolt-port", *bolt_port, "--timeout", "180"] - proc: "tests/e2e/memory/procedures/" - <<: *disk_cluster + <<: *in_memory_query_limit_cluster