From 94d5e22dcd8d72eef686fd866ea86e42238c3868 Mon Sep 17 00:00:00 2001 From: antoniofilipovic Date: Wed, 11 Oct 2023 13:55:42 +0200 Subject: [PATCH] introduce tracking per thread, add working test --- include/mg_procedure.h | 24 ++++- src/memory/memory_control.cpp | 40 ++++++--- src/memory/memory_control.hpp | 10 ++- src/query/db_accessor.cpp | 4 + src/query/db_accessor.hpp | 10 +++ src/query/interpreter.cpp | 16 ++-- src/query/procedure/mg_procedure_impl.cpp | 12 +++ tests/e2e/memory/procedures/CMakeLists.txt | 3 +- .../query_memory_limit_proc_multi_thread.cpp | 89 +++++++++++-------- tests/e2e/memory/query_memory_limit.cpp | 7 +- 10 files changed, 149 insertions(+), 66 deletions(-) diff --git a/include/mg_procedure.h b/include/mg_procedure.h index 9bff63471..cd17ed321 100644 --- a/include/mg_procedure.h +++ b/include/mg_procedure.h @@ -115,9 +115,31 @@ void mgp_global_free(void *p); /// State of the graph database. struct mgp_graph; +/// Allocations are tracked only for master thread. If new threads are spawned +/// inside procedure, by calling following function with thread id +/// you can start tracking allocations for that thread too. This +/// is important if you need query memory limit to work +/// for given procedure or per procedure memory limit. 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 *); +/// Once allocations are tracked for custom thread, you need to stop tracking allocations +/// for given thread, before thread finishes with execution, or is detached. +/// Otherwise it might result in slowdown of system due to unnecessary tracking of +/// allocations. +enum mgp_error mgp_untrack_thread_allocations(struct mgp_graph *graph, const char *thread_id); + +/// Allocations are tracked only for master thread. If new threads are spawned +/// inside procedure, by calling following function +/// you can start tracking allocations for current thread too. This +/// is important if you need query memory limit to work +/// for given procedure or per procedure memory limit. +enum mgp_error mgp_track_current_thread_allocations(struct mgp_graph *graph); + +/// Once allocations are tracked for current thread, you need to stop tracking allocations +/// for given thread, before thread finishes with execution, or is detached. +/// Otherwise it might result in slowdown of system due to unnecessary tracking of +/// allocations. +enum mgp_error mgp_untrack_current_thread_allocations(struct mgp_graph *graph); ///@} /// @name Operations on mgp_value diff --git a/src/memory/memory_control.cpp b/src/memory/memory_control.cpp index 6b04b6636..82014c225 100644 --- a/src/memory/memory_control.cpp +++ b/src/memory/memory_control.cpp @@ -11,7 +11,6 @@ #include "memory_control.hpp" #include -#include #include #include #include "utils/logging.hpp" @@ -28,12 +27,24 @@ namespace memgraph::memory { // NOLINTNEXTLINE(cppcoreguidelines-macro-usage) #define STRINGIFY(x) STRINGIFY_HELPER(x) -std::string get_thread_id() { +std::string get_string_thread_id(const std::thread::id &thread_id) { std::ostringstream oss; - oss << std::this_thread::get_id(); + oss << thread_id; return oss.str(); } +// TODO (af) think if following implementation would make sense +/* +std::string get_thread_id() { + static thread_local std::thread::id current_thread_id = std::this_thread::get_id(); + // figure out how to cache this part + std::ostringstream oss; + oss << current_thread_id; + return oss.str(); +} +*/ +std::string get_thread_id() { return get_string_thread_id(std::this_thread::get_id()); } + #if USE_JEMALLOC static void *my_alloc(extent_hooks_t *extent_hooks, void *new_addr, size_t size, size_t alignment, bool *zero, @@ -68,9 +79,6 @@ void *my_alloc(extent_hooks_t *extent_hooks, void *new_addr, size_t size, size_t if (*commit) [[likely]] { memgraph::utils::total_memory_tracker.Alloc(static_cast(size)); if (arena_tracking[arena_ind]) [[unlikely]] { - std::string thread_id = get_thread_id(); - std::cout << "aloc:" << thread_id << std::endl; - std::cout << "thread id to transaction id" << thread_id_to_transaction_id[get_thread_id()] << std::endl; transaction_id_tracker[thread_id_to_transaction_id[get_thread_id()]].Alloc(static_cast(size)); } } @@ -264,25 +272,29 @@ bool RemoveTrackingOnArena(unsigned arena_id) { } void UpdateThreadToTransactionId(const std::thread::id &thread_id, uint64_t transaction_id) { - std::ostringstream oss; - oss << thread_id; - thread_id_to_transaction_id[oss.str()] = transaction_id; - std::cout << "set tracking for thread:" << oss.str() << ", on transaction: " << transaction_id << std::endl; + thread_id_to_transaction_id[get_string_thread_id(thread_id)] = transaction_id; } 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) { - std::ostringstream oss; - oss << thread_id; - thread_id_to_transaction_id.erase(oss.str()); + thread_id_to_transaction_id.erase(get_string_thread_id(thread_id)); } void ResetThreadToTransactionId(const char *thread_id) { thread_id_to_transaction_id.erase(std::string(thread_id)); } +void AddTrackingsOnCurrentThread(uint64_t transaction_id) { + UpdateThreadToTransactionId(std::this_thread::get_id(), transaction_id); + AddTrackingOnArena(memgraph::memory::GetArenaForThread()); +} + +void RemoveTrackingsOnCurrentThread() { + ResetThreadToTransactionId(std::this_thread::get_id()); + RemoveTrackingOnArena(GetArenaForThread()); +} + 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 298ff003d..3f6e470ab 100644 --- a/src/memory/memory_control.hpp +++ b/src/memory/memory_control.hpp @@ -22,6 +22,9 @@ namespace memgraph::memory { void PurgeUnusedMemory(); void SetHooks(); + +// TODO(af) This should all be part of memgraph::memory::thread namespace, and moved to different file +// This should be part of class unsigned GetArenaForThread(); bool AddTrackingOnArena(unsigned); bool RemoveTrackingOnArena(unsigned); @@ -29,10 +32,13 @@ void UpdateThreadToTransactionId(const std::thread::id &, uint64_t); void ResetThreadToTransactionId(const std::thread::id &); void UpdateThreadToTransactionId(const char *, uint64_t); void ResetThreadToTransactionId(const char *); +void AddTrackingsOnCurrentThread(uint64_t); +void RemoveTrackingsOnCurrentThread(); -// TODO(AF) : Do we need arena to transaction id and transaction_id to tracker? inline std::unordered_map thread_id_to_transaction_id; -inline std::unordered_map> arena_tracking; +// TODO(af): think if we need to solve issue of tracking allocations for arena +// if user forgets to unregister tracking for that thread before it dies. +inline std::unordered_map> arena_tracking; inline std::unordered_map transaction_id_tracker; } // namespace memgraph::memory diff --git a/src/query/db_accessor.cpp b/src/query/db_accessor.cpp index c9433291d..5df81bf6c 100644 --- a/src/query/db_accessor.cpp +++ b/src/query/db_accessor.cpp @@ -25,10 +25,14 @@ void SubgraphDbAccessor::TrackThreadAllocations(const char *thread_id) { return db_accessor_.TrackThreadAllocations(thread_id); } +void SubgraphDbAccessor::TrackCurrentThreadAllocations() { return db_accessor_.TrackCurrentThreadAllocations(); } + void SubgraphDbAccessor::UntrackThreadAllocations(const char *thread_id) { return db_accessor_.UntrackThreadAllocations(thread_id); } +void SubgraphDbAccessor::UntrackCurrentThreadAllocations() { return db_accessor_.TrackCurrentThreadAllocations(); } + 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 a9a92f492..279d5a517 100644 --- a/src/query/db_accessor.hpp +++ b/src/query/db_accessor.hpp @@ -379,12 +379,18 @@ class DbAccessor final { memgraph::memory::AddTrackingOnArena(arena); } + void TrackCurrentThreadAllocations() { + memgraph::memory::AddTrackingsOnCurrentThread(*accessor_->GetTransactionId()); + } + void UntrackThreadAllocations(const char *thread_id) { memgraph::memory::ResetThreadToTransactionId(thread_id); auto arena = memgraph::memory::GetArenaForThread(); memgraph::memory::RemoveTrackingOnArena(arena); } + void UntrackCurrentThreadAllocations() { memgraph::memory::RemoveTrackingsOnCurrentThread(); } + std::optional GetTransactionId() { return accessor_->GetTransactionId(); } VerticesIterable Vertices(storage::View view) { return VerticesIterable(accessor_->Vertices(view)); } @@ -660,8 +666,12 @@ class SubgraphDbAccessor final { void TrackThreadAllocations(const char *thread_id); + void TrackCurrentThreadAllocations(); + void UntrackThreadAllocations(const char *thread_id); + void UntrackCurrentThreadAllocations(); + storage::PropertyId NameToProperty(std::string_view name); storage::LabelId NameToLabel(std::string_view name); diff --git a/src/query/interpreter.cpp b/src/query/interpreter.cpp index 20ce59cc0..1c42037d7 100644 --- a/src/query/interpreter.cpp +++ b/src/query/interpreter.cpp @@ -1268,25 +1268,24 @@ std::optional PullPlan::Pull(AnyStream *strea std::map *summary) { std::optional transaction_id = ctx_.db_accessor->GetTransactionId(); MG_ASSERT(transaction_id.has_value()); + unsigned arena_ind{0}; if (memory_limit_) { - // TODO(af) update for transaction id + // TODO (AF) think to isolate this in namespace or make a class memgraph::memory::transaction_id_tracker.emplace(std::piecewise_construct, std::forward_as_tuple(*transaction_id), std::forward_as_tuple()); auto &memory_tracker = memgraph::memory::transaction_id_tracker[*transaction_id]; memory_tracker.SetMaximumHardLimit(static_cast(*memory_limit_)); memory_tracker.SetHardLimit(static_cast(*memory_limit_)); - - auto arena_id = memgraph::memory::GetArenaForThread(); - memgraph::memory::AddTrackingOnArena(arena_id); + arena_ind = memgraph::memory::GetArenaForThread(); + memgraph::memory::AddTrackingOnArena(arena_ind); memgraph::memory::UpdateThreadToTransactionId(std::this_thread::get_id(), *transaction_id); } utils::OnScopeExit> reset_query_limit{ - [memory_limit = memory_limit_, transaction_id = *transaction_id]() { + [memory_limit = memory_limit_, transaction_id = *transaction_id, arena_ind]() { if (memory_limit) { - // TODO(af) update for transaction with id + // TODO (AF) think to isolate this in namespace or make a class memgraph::memory::transaction_id_tracker.erase(transaction_id); - auto arena_id = memgraph::memory::GetArenaForThread(); - memgraph::memory::RemoveTrackingOnArena(arena_id); + memgraph::memory::RemoveTrackingOnArena(arena_ind); memgraph::memory::ResetThreadToTransactionId(std::this_thread::get_id()); } }}; @@ -1314,7 +1313,6 @@ std::optional PullPlan::Pull(AnyStream *strea } ctx_.evaluation_context.memory = &*pool_memory; - ctx_.db_accessor->id(); // Returns true if a result was pulled. const auto pull_result = [&]() -> bool { return cursor_->Pull(frame_, ctx_); }; diff --git a/src/query/procedure/mg_procedure_impl.cpp b/src/query/procedure/mg_procedure_impl.cpp index 587f8b580..6d9d4c438 100644 --- a/src/query/procedure/mg_procedure_impl.cpp +++ b/src/query/procedure/mg_procedure_impl.cpp @@ -3547,9 +3547,21 @@ mgp_error mgp_track_thread_allocations(mgp_graph *graph, const char *thread_id) }); } +mgp_error mgp_track_current_thread_allocations(mgp_graph *graph) { + return WrapExceptions([&]() { + std::visit([](auto *db_accessor) -> void { db_accessor->TrackCurrentThreadAllocations(); }, 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); }); } + +mgp_error mgp_untrack_current_thread_allocations(mgp_graph *graph) { + return WrapExceptions([&]() { + std::visit([](auto *db_accessor) -> void { db_accessor->UntrackCurrentThreadAllocations(); }, graph->impl); + }); +} diff --git a/tests/e2e/memory/procedures/CMakeLists.txt b/tests/e2e/memory/procedures/CMakeLists.txt index efb141f20..c663ddb65 100644 --- a/tests/e2e/memory/procedures/CMakeLists.txt +++ b/tests/e2e/memory/procedures/CMakeLists.txt @@ -6,4 +6,5 @@ target_include_directories(global_memory_limit_proc PRIVATE ${CMAKE_SOURCE_DIR}/ 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 +target_include_directories(query_memory_limit_proc_multi_thread PRIVATE ${CMAKE_SOURCE_DIR}/include) +target_link_libraries(query_memory_limit_proc_multi_thread mg-utils) 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 index 2a21d154a..0177d431c 100644 --- a/tests/e2e/memory/procedures/query_memory_limit_proc_multi_thread.cpp +++ b/tests/e2e/memory/procedures/query_memory_limit_proc_multi_thread.cpp @@ -10,6 +10,7 @@ // licenses/APL.txt. #include +#include #include #include #include @@ -19,60 +20,44 @@ #include #include #include + #include "mg_procedure.h" +#include "utils/on_scope_exit.hpp" -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_(); - } +enum mgp_error Alloc(void *ptr) { + const size_t two_sixty_eight_mb = 1 << 28; - void Disable() { doCall_ = false; } + return mgp_global_alloc(two_sixty_eight_mb, (void **)(&ptr)); +} - private: - std::function function_; - bool doCall_; -}; -template -OnScopeExit(Callable &&) -> OnScopeExit; - -// TODO(af) fix this hack +// change communication between threads with feature and promise std::atomic num_allocations{0}; +std::vector ptrs_; 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; + [[maybe_unused]] const enum mgp_error tracking_error = mgp_track_current_thread_allocations(graph); + void *ptr = nullptr; - OnScopeExit> cleanup{[&ptr]() { - if (nullptr == ptr) { - return; + ptrs_.emplace_back(ptr); + try { + const enum mgp_error alloc_err = Alloc(ptr); + if (alloc_err != mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE) { + num_allocations.fetch_add(1, std::memory_order_relaxed); } - 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); + } catch (const std::exception &e) { + // not to terminate std::thread here + assert(false); } - [[maybe_unused]] const enum mgp_error untracking_error = mgp_untrack_thread_allocations(graph, thread_id.c_str()); + [[maybe_unused]] const enum mgp_error untracking_error = mgp_untrack_current_thread_allocations(graph); } 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); - + num_allocations.store(0, std::memory_order_relaxed); try { std::vector threads; @@ -83,11 +68,38 @@ void DualThread(mgp_list *args, mgp_graph *memgraph_graph, mgp_result *result, m for (int i = 0; i < 2; i++) { threads[i].join(); } + for (void *ptr : ptrs_) { + if (ptr != nullptr) { + mgp_global_free(ptr); + } + } 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()); + } +} + +void Regular(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 { + void *ptr{nullptr}; + + memgraph::utils::OnScopeExit> cleanup{[&ptr]() { + if (nullptr == ptr) { + return; + } + mgp_global_free(ptr); + }}; + + const enum mgp_error alloc_err = Alloc(ptr); + auto new_record = record_factory.NewRecord(); + new_record.Insert("allocated", alloc_err != mgp_error::MGP_ERROR_UNABLE_TO_ALLOCATE); } catch (std::exception &e) { record_factory.SetErrorMessage(e.what()); } @@ -100,6 +112,9 @@ extern "C" int mgp_init_module(struct mgp_module *module, struct mgp_memory *mem AddProcedure(DualThread, std::string("dual_thread").c_str(), mgp::ProcedureType::Read, {}, {mgp::Return(std::string("allocated_all").c_str(), mgp::Type::Bool)}, module, memory); + AddProcedure(Regular, std::string("regular").c_str(), mgp::ProcedureType::Read, {}, + {mgp::Return(std::string("allocated").c_str(), mgp::Type::Bool)}, module, memory); + } catch (const std::exception &e) { return 1; } diff --git a/tests/e2e/memory/query_memory_limit.cpp b/tests/e2e/memory/query_memory_limit.cpp index 3e7c0c6c5..e23781e8a 100644 --- a/tests/e2e/memory/query_memory_limit.cpp +++ b/tests/e2e/memory/query_memory_limit.cpp @@ -11,6 +11,8 @@ #include #include +#include +#include #include #include @@ -51,10 +53,11 @@ int main(int argc, char **argv) { auto result_rows = client->FetchAll(); if (result_rows) { auto row = *result_rows->begin(); - std::cout << row[0].ValueBool() << std::endl; + error = row[0].ValueBool() == false; + std::cout << std::boolalpha << error << ", error status" << std::endl; } - } catch (const mg::ClientException &e) { + } catch (const std::exception &e) { std::cout << e.what() << std::endl; error = true; }