diff --git a/src/agents/evolution/QueryEvolutionProcessor.cc b/src/agents/evolution/QueryEvolutionProcessor.cc index 1597620d..3f4b79b4 100644 --- a/src/agents/evolution/QueryEvolutionProcessor.cc +++ b/src/agents/evolution/QueryEvolutionProcessor.cc @@ -1,5 +1,9 @@ #include "QueryEvolutionProcessor.h" +#if defined(__GLIBC__) +#include +#endif + #include "AttentionBrokerClient.h" #include "Hasher.h" #include "LinkSchema.h" @@ -70,8 +74,20 @@ void QueryEvolutionProcessor::thread_process_one_query(shared_ptrraise_error_on_peer(exception.what()); } + // The evolution run is over, so the population and all intermediate query answers held during + // it are freed. Return that freed heap to the OS so RSS does not stay pinned at the peak. +#if defined(__GLIBC__) + malloc_trim(0); +#endif + // Self-reap: detach this finished thread and drop it from query_threads immediately, so a + // burst that finishes while the node is idle returns to baseline without waiting for the next + // command. A thread cannot join itself, hence detach instead of join. + { + lock_guard semaphore(this->query_threads_mutex); + this->query_threads.erase(monitor->get_id()); + monitor->detach(); + } LOG_DEBUG("Command finished: <" << proxy->get_command() << ">"); - // TODO add a call to remove_query_thread(monitor->get_id()); } shared_ptr QueryEvolutionProcessor::issue_sampling_query( diff --git a/src/agents/evolution/QueryEvolutionProcessor.h b/src/agents/evolution/QueryEvolutionProcessor.h index 0b229342..65c46824 100644 --- a/src/agents/evolution/QueryEvolutionProcessor.h +++ b/src/agents/evolution/QueryEvolutionProcessor.h @@ -2,6 +2,7 @@ #include #include +#include #include #include "BusCommandProcessor.h" diff --git a/src/agents/query_engine/PatternMatchingQueryProcessor.cc b/src/agents/query_engine/PatternMatchingQueryProcessor.cc index b7c89202..679183e8 100644 --- a/src/agents/query_engine/PatternMatchingQueryProcessor.cc +++ b/src/agents/query_engine/PatternMatchingQueryProcessor.cc @@ -3,6 +3,10 @@ #include "Logger.h" +#if defined(__GLIBC__) +#include +#endif + #include #include "And.h" @@ -246,8 +250,21 @@ void PatternMatchingQueryProcessor::thread_process_one_query( } catch (const std::exception& exception) { proxy->raise_error_on_peer(exception.what()); } + // At this point the query tree (root_query_element, query_sink) declared inside the try block + // above has already been destroyed, so every QueryAnswer/HandleSet allocated for this query is + // freed. Return that freed heap to the OS so RSS does not stay pinned at the peak. +#if defined(__GLIBC__) + malloc_trim(0); +#endif + // Self-reap: detach this finished thread and drop it from query_threads immediately, so a + // burst of queries that finishes while the node is idle returns to baseline without waiting + // for the next command. A thread cannot join itself, hence detach instead of join. + { + lock_guard semaphore(this->query_threads_mutex); + this->query_threads.erase(monitor->get_id()); + monitor->detach(); + } LOG_DEBUG("Command finished: <" << proxy->get_command() << ">"); - // TODO add a call to remove_query_thread(monitor->get_id()); } void PatternMatchingQueryProcessor::remove_query_thread(const string& stoppable_thread_id) { diff --git a/src/agents/query_engine/PatternMatchingQueryProcessor.h b/src/agents/query_engine/PatternMatchingQueryProcessor.h index 493c8c14..8aba78fa 100644 --- a/src/agents/query_engine/PatternMatchingQueryProcessor.h +++ b/src/agents/query_engine/PatternMatchingQueryProcessor.h @@ -2,6 +2,7 @@ #include #include +#include #include #include diff --git a/src/commons/StoppableThread.cc b/src/commons/StoppableThread.cc index 5bff31c1..63fb6738 100644 --- a/src/commons/StoppableThread.cc +++ b/src/commons/StoppableThread.cc @@ -27,20 +27,45 @@ void StoppableThread::attach(thread* thread_object) { this->thread_object = thre void StoppableThread::stop() { this->stop(true); } void StoppableThread::stop(bool join_thread) { - if (this->thread_object == NULL) { - RAISE_ERROR("No thread attached to StoppableThread: " + this->id); - } else if (!this->stop_flag) { - LOG_DEBUG("Stopping thread: " << this->id); - this->stop_flag_mutex.lock(); + // stop_flag only means "stop requested" (it is what stopped() reports). It is intentionally + // independent of whether the underlying thread has been reaped, so requesting a stop without + // joining (stop(false)) still leaves the thread to be reaped by a later stop()/detach()/dtor. + thread* to_reap = NULL; + { + lock_guard guard(this->stop_flag_mutex); this->stop_flag = true; - this->stop_flag_mutex.unlock(); if (join_thread) { - LOG_DEBUG("Joining thread: " << this->id); - this->thread_object->join(); - LOG_DEBUG("Thread joined: " << this->id << ". destroying it."); - delete this->thread_object; + // Claim the thread exactly once: whoever swaps the pointer out reaps it; concurrent or + // later callers see NULL and skip. + to_reap = this->thread_object; + this->thread_object = NULL; } } + // Join OUTSIDE the lock so the thread being joined can still call stopped() without deadlocking. + if (to_reap != NULL) { + LOG_DEBUG("Joining thread: " << this->id); + to_reap->join(); + LOG_DEBUG("Thread joined: " << this->id << ". destroying it."); + delete to_reap; + } +} + +void StoppableThread::detach() { + thread* to_reap = NULL; + { + lock_guard guard(this->stop_flag_mutex); + this->stop_flag = true; + // Claim the thread exactly once (see stop()). + to_reap = this->thread_object; + this->thread_object = NULL; + } + if (to_reap != NULL) { + LOG_DEBUG("Detaching thread: " << this->id); + // detach() releases the OS thread; deleting the std::thread handle afterwards does not + // affect the still-running OS thread. + to_reap->detach(); + delete to_reap; + } } bool StoppableThread::stopped() { diff --git a/src/commons/StoppableThread.h b/src/commons/StoppableThread.h index 7dcdf026..0a94b4df 100644 --- a/src/commons/StoppableThread.h +++ b/src/commons/StoppableThread.h @@ -51,6 +51,15 @@ class StoppableThread : public Stoppable { */ void attach(thread* thread_object); + /** + * Detach the attached thread so it cleans itself up on completion. + * + * Used when a thread needs to "reap itself" (a thread cannot join itself). After detach(), + * the thread is no longer joinable and both stop() and the destructor become no-ops with + * respect to it. + */ + void detach(); + /** * Stop attached thread. * diff --git a/src/commons/ThreadSafeHeap.h b/src/commons/ThreadSafeHeap.h index 5eaf8f3d..f64108b2 100644 --- a/src/commons/ThreadSafeHeap.h +++ b/src/commons/ThreadSafeHeap.h @@ -78,7 +78,17 @@ class ThreadSafeHeap { } void snapshot(vector& output) { - priority_queue> copy = this->queue; + // Copy the underlying queue UNDER the lock. Without it, this copy races with a concurrent + // push()/pop() (which mutate the backing vector under api_mutex); a reallocation mid-copy + // would read freed memory and produce corrupt elements (e.g. shared_ptrs built from garbage + // control blocks), leading to later double-free/heap corruption. Drain the local copy + // outside the lock so other threads are only blocked for the duration of the copy. + priority_queue> copy; + { + lock_guard semaphore(this->api_mutex); + copy = this->queue; + } + output.reserve(output.size() + copy.size()); while (!copy.empty()) { output.push_back(copy.top().element); copy.pop(); diff --git a/src/main/bus_node.cc b/src/main/bus_node.cc index f278ff35..b6b02918 100644 --- a/src/main/bus_node.cc +++ b/src/main/bus_node.cc @@ -1,5 +1,9 @@ #include +#if defined(__GLIBC__) +#include +#endif + #include "AttentionBrokerClient.h" #include "CommandRouterHttpAPISingleton.h" #include "FitnessFunctionRegistry.h" @@ -25,6 +29,23 @@ void ctrl_c_handler(int) { int main(int argc, char* argv[]) { try { +#if defined(__GLIBC__) + // Cap the number of malloc arenas. The bus node spawns many short-lived worker threads + // (one per query plus per query-element). With the glibc default (8 * nproc arenas) the + // large transient allocations done while fetching/matching atoms get scattered across + // dozens of arenas that are never given back to the OS, so RSS stays pinned at the + // high-water mark long after the queries are done. A small, fixed arena count keeps + // fragmentation/retention under control while still allowing concurrency. Honor an + // explicit MALLOC_ARENA_MAX from the environment if the operator set one. + if (getenv("MALLOC_ARENA_MAX") == nullptr) { + mallopt(M_ARENA_MAX, 4); + } + // Return free heap pages to the OS instead of caching unbounded amounts in the arenas. + // Honor an explicit MALLOC_TRIM_THRESHOLD_ from the environment if the operator set one. + if (getenv("MALLOC_TRIM_THRESHOLD_") == nullptr) { + mallopt(M_TRIM_THRESHOLD, 128 * 1024); + } +#endif auto required_cmd_args = {Helper::SERVICE, Helper::CONFIG}; auto cmd_args = Utils::parse_command_line(argc, argv); diff --git a/src/tests/cpp/BUILD b/src/tests/cpp/BUILD index 619a1ea3..f158d23b 100644 --- a/src/tests/cpp/BUILD +++ b/src/tests/cpp/BUILD @@ -749,6 +749,21 @@ cc_test( ], ) +cc_test( + name = "stoppable_thread_test", + size = "small", + srcs = ["stoppable_thread_test.cc"], + copts = [ + "-Iexternal/gtest/googletest/include", + "-Iexternal/gtest/googletest", + ], + linkstatic = 1, + deps = [ + "//commons:commons_lib", + "@com_github_google_googletest//:gtest_main", + ], +) + cc_test( name = "redis_mongodb_test", size = "medium", diff --git a/src/tests/cpp/stoppable_thread_test.cc b/src/tests/cpp/stoppable_thread_test.cc new file mode 100644 index 00000000..5fa6994a --- /dev/null +++ b/src/tests/cpp/stoppable_thread_test.cc @@ -0,0 +1,116 @@ +#include +#include +#include +#include +#include + +#include "StoppableThread.h" +#include "Utils.h" +#include "gtest/gtest.h" + +using namespace std; +using namespace commons; + +// Reproduces the self-reaping pattern used by PatternMatchingQueryProcessor and +// QueryEvolutionProcessor: a pool of worker threads where each worker, when it finishes, detaches +// itself and removes its own entry from a shared map. This is the "burst-then-idle" case: after a +// burst of workers completes, the map must drain to empty WITHOUT any external/non-worker reaping +// trigger. +TEST(StoppableThread, self_reap_burst_then_idle) { + map> threads; + mutex threads_mutex; + atomic completed{0}; + + const int burst = 16; + for (int i = 0; i < burst; i++) { + string id = "worker_" + std::to_string(i); + lock_guard guard(threads_mutex); + auto stoppable_thread = make_shared(id); + stoppable_thread->attach(new thread( + [&threads, &threads_mutex, &completed](shared_ptr monitor) { + Utils::sleep(10); // simulate a short unit of work + completed++; + // Self-reap (mirrors the processors): a thread can't join itself, so detach and + // drop its own entry. completed++ happens before the erase, so once the map is + // empty every worker is guaranteed to have finished its work. + lock_guard guard(threads_mutex); + monitor->detach(); + threads.erase(monitor->get_id()); + }, + stoppable_thread)); + threads[id] = stoppable_thread; + } + + // Bounded wait for the burst to drain, issuing no further "commands". + for (int i = 0; i < 500; i++) { + { + lock_guard guard(threads_mutex); + if (threads.empty()) { + break; + } + } + Utils::sleep(10); + } + + lock_guard guard(threads_mutex); + EXPECT_EQ(completed.load(), burst); + EXPECT_TRUE(threads.empty()); +} + +// detach() must make the thread non-joinable so that a subsequent stop() (e.g. from the +// StoppableThread destructor) is a safe no-op rather than a join-on-detached error. +TEST(StoppableThread, stop_after_detach_is_noop) { + auto stoppable_thread = make_shared("detach_then_stop"); + atomic ran{false}; + stoppable_thread->attach(new thread( + [&ran](shared_ptr monitor) { + ran = true; + while (!monitor->stopped()) { + Utils::sleep(5); + } + }, + stoppable_thread)); + + while (!ran.load()) { + Utils::sleep(5); + } + stoppable_thread->detach(); // releases the OS thread; also flips the stop flag + EXPECT_TRUE(stoppable_thread->stopped()); + EXPECT_NO_THROW(stoppable_thread->stop()); // must not join/double-free a detached thread +} + +// stop_flag means "stop requested", not "thread reaped". stop(false) only requests the stop; a later +// stop() must still join/reap the (still joinable) thread instead of being skipped because the flag +// is already set. This covers the "request-stop-now, reap-later" ordering. +TEST(StoppableThread, stop_false_then_stop_reaps) { + auto stoppable_thread = make_shared("request_then_reap"); + atomic ran{false}; + atomic exited{false}; + stoppable_thread->attach(new thread( + [&ran, &exited](shared_ptr monitor) { + ran = true; + while (!monitor->stopped()) { + Utils::sleep(5); + } + exited = true; + }, + stoppable_thread)); + + while (!ran.load()) { + Utils::sleep(5); + } + + stoppable_thread->stop(false); // request stop only: flag is set, but nothing is reaped yet + EXPECT_TRUE(stoppable_thread->stopped()); + + // The worker observes the request and exits on its own. + for (int i = 0; i < 200 && !exited.load(); i++) { + Utils::sleep(5); + } + EXPECT_TRUE(exited.load()); + + // The thread is still joinable; the full stop() must reap it (and the destructor must stay a + // safe no-op afterwards). If stop_flag short-circuited reaping, this would leak the thread. + EXPECT_NO_THROW(stoppable_thread->stop()); + EXPECT_NO_THROW(stoppable_thread->stop()); // idempotent: second reap is a no-op +}