tursom/memgraph
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

introduce tracking per thread, add working test

Browse Source
This commit is contained in:
antoniofilipovic
2023-10-11 13:55:42 +02:00
parent 11ee19516a
commit 94d5e22dcd
10 changed files with 149 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
include/mg_procedure.h
View File

@@ -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

40
src/memory/memory_control.cpp
View File

@@ -11,7 +11,6 @@
#include "memory_control.hpp"
#include <cstdint>
#include <iostream>
#include <sstream>
#include <thread>
#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<int64_t>(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<int64_t>(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);

10
src/memory/memory_control.hpp
View File

@@ -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<std::string, uint64_t> thread_id_to_transaction_id;
inline std::unordered_map<int, std::atomic<int>> 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<unsigned, std::atomic<int>> arena_tracking;
inline std::unordered_map<uint64_t, utils::MemoryTracker> transaction_id_tracker;
} // namespace memgraph::memory

4
src/query/db_accessor.cpp
View File

@@ -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);
}

10
src/query/db_accessor.hpp
View File

@@ -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<uint64_t> 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);

16
src/query/interpreter.cpp
View File

@@ -1268,25 +1268,24 @@ std::optional<plan::ProfilingStatsWithTotalTime> PullPlan::Pull(AnyStream *strea
std::map<std::string, TypedValue> *summary) {
std::optional<uint64_t> 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<int64_t>(*memory_limit_));
memory_tracker.SetHardLimit(static_cast<int64_t>(*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<std::function<void()>> 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<plan::ProfilingStatsWithTotalTime> 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_); };

12
src/query/procedure/mg_procedure_impl.cpp
View File

@@ -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);
});
}

3
tests/e2e/memory/procedures/CMakeLists.txt
View File

@@ -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)

89
tests/e2e/memory/procedures/query_memory_limit_proc_multi_thread.cpp
View File

@@ -10,6 +10,7 @@
// licenses/APL.txt.
#include <atomic>
#include <cassert>
#include <exception>
#include <functional>
#include <mgp.hpp>
@@ -19,60 +20,44 @@
#include <thread>
#include <utility>
#include <vector>
#include "mg_procedure.h"
#include "utils/on_scope_exit.hpp"
template <typename Callable>
class [[nodiscard]] OnScopeExit {
public:
explicit OnScopeExit(Callable &&function) : function_{std::forward<Callable>(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<void()> function_;
bool doCall_;
};
template <typename Callable>
OnScopeExit(Callable &&) -> OnScopeExit<Callable>;
// TODO(af) fix this hack
// change communication between threads with feature and promise
std::atomic<int> num_allocations{0};
std::vector<void *> 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<std::function<void(void)>> 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<std::thread> 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<std::function<void(void)>> 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;
}

7
tests/e2e/memory/query_memory_limit.cpp
View File

@@ -11,6 +11,8 @@
#include <gflags/gflags.h>
#include <algorithm>
#include <exception>
#include <ios>
#include <iostream>
#include <mgclient.hpp>
@@ -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;
}