From 1f13f9efd9f04d1c34dac4485d6f4a4e2daaf3a9 Mon Sep 17 00:00:00 2001 From: CCG Date: Tue, 30 Jun 2026 17:18:58 -0300 Subject: [PATCH 1/7] Return memory back to OS --- .../evolution/QueryEvolutionProcessor.cc | 29 +++++++++++++++++- .../evolution/QueryEvolutionProcessor.h | 11 +++++++ .../PatternMatchingQueryProcessor.cc | 30 ++++++++++++++++++- .../PatternMatchingQueryProcessor.h | 11 +++++++ src/main/bus_node.cc | 18 +++++++++++ src/tests/scripts/run_toy.sh | 30 ++++++++++--------- 6 files changed, 113 insertions(+), 16 deletions(-) diff --git a/src/agents/evolution/QueryEvolutionProcessor.cc b/src/agents/evolution/QueryEvolutionProcessor.cc index 1597620d..f3fe0ef3 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" @@ -34,6 +38,7 @@ shared_ptr QueryEvolutionProcessor::factory_empty_proxy() { void QueryEvolutionProcessor::run_command(shared_ptr proxy) { lock_guard semaphore(this->query_threads_mutex); + reap_finished_threads(); auto query_proxy = dynamic_pointer_cast(proxy); string thread_id = "thread<" + proxy->my_id() + "_" + std::to_string(proxy->get_serial()) + ">"; LOG_DEBUG("Starting new thread: " << thread_id << " to run command: <" << proxy->get_command() @@ -70,8 +75,30 @@ 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 + // This thread cannot join itself, so just flag it as finished. The next run_command will join + // and erase it from query_threads. + { + lock_guard semaphore(this->query_threads_mutex); + this->finished_query_threads.insert(monitor->get_id()); + } LOG_DEBUG("Command finished: <" << proxy->get_command() << ">"); - // TODO add a call to remove_query_thread(monitor->get_id()); +} + +void QueryEvolutionProcessor::reap_finished_threads() { + // Must be called while holding query_threads_mutex. + for (const auto& thread_id : this->finished_query_threads) { + auto it = this->query_threads.find(thread_id); + if (it != this->query_threads.end()) { + it->second->stop(); // joins the (already finished) thread and releases its stack + this->query_threads.erase(it); + } + } + this->finished_query_threads.clear(); } shared_ptr QueryEvolutionProcessor::issue_sampling_query( diff --git a/src/agents/evolution/QueryEvolutionProcessor.h b/src/agents/evolution/QueryEvolutionProcessor.h index 0b229342..73dc6975 100644 --- a/src/agents/evolution/QueryEvolutionProcessor.h +++ b/src/agents/evolution/QueryEvolutionProcessor.h @@ -2,6 +2,7 @@ #include #include +#include #include #include "BusCommandProcessor.h" @@ -72,11 +73,21 @@ class QueryEvolutionProcessor : public BusCommandProcessor { vector, float>>& selected); void thread_process_one_query(shared_ptr, shared_ptr proxy); void remove_query_thread(const string& stoppable_thread_id); + + /** + * Joins and removes every query thread that already finished running. + * + * A running query thread cannot join itself, so it only marks its id as finished when it is + * about to return. The actual join + erase (which releases the thread stack and lets the + * StoppableThread memory be reclaimed) is performed here, from a different thread. + */ + void reap_finished_threads(); string answer_to_string_1(shared_ptr answer, shared_ptr db); string answer_to_string_2(shared_ptr answer, shared_ptr db); string answer_to_string(shared_ptr answer); map> query_threads; + set finished_query_threads; mutex query_threads_mutex; shared_ptr proxy; set visited_individuals; diff --git a/src/agents/query_engine/PatternMatchingQueryProcessor.cc b/src/agents/query_engine/PatternMatchingQueryProcessor.cc index b7c89202..3375fe86 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" @@ -57,6 +61,7 @@ shared_ptr PatternMatchingQueryProcessor::factory_empty_proxy() void PatternMatchingQueryProcessor::run_command(shared_ptr proxy) { lock_guard semaphore(this->query_threads_mutex); + reap_finished_threads(); auto query_proxy = dynamic_pointer_cast(proxy); string thread_id = "thread<" + proxy->my_id() + "_" + std::to_string(proxy->get_serial()) + ">"; LOG_DEBUG("Starting new thread: " << thread_id << " to run command: <" << proxy->get_command() @@ -246,8 +251,31 @@ 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 + // This thread cannot join itself, so just flag it as finished. The next run_command (or the + // graceful shutdown path) will join and erase it from query_threads. + { + lock_guard semaphore(this->query_threads_mutex); + this->finished_query_threads.insert(monitor->get_id()); + } LOG_DEBUG("Command finished: <" << proxy->get_command() << ">"); - // TODO add a call to remove_query_thread(monitor->get_id()); +} + +void PatternMatchingQueryProcessor::reap_finished_threads() { + // Must be called while holding query_threads_mutex. + for (const auto& thread_id : this->finished_query_threads) { + auto it = this->query_threads.find(thread_id); + if (it != this->query_threads.end()) { + it->second->stop(); // joins the (already finished) thread and releases its stack + this->query_threads.erase(it); + } + } + this->finished_query_threads.clear(); } 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..a5ed5c7f 100644 --- a/src/agents/query_engine/PatternMatchingQueryProcessor.h +++ b/src/agents/query_engine/PatternMatchingQueryProcessor.h @@ -2,6 +2,7 @@ #include #include +#include #include #include @@ -87,7 +88,17 @@ class PatternMatchingQueryProcessor : public BusCommandProcessor { void remove_query_thread(const string& stoppable_thread_id); + /** + * Joins and removes every query thread that already finished running. + * + * A running query thread cannot join itself, so it only marks its id as finished when it is + * about to return. The actual join + erase (which releases the thread stack and lets the + * StoppableThread/query tree memory be reclaimed) is performed here, from a different thread. + */ + void reap_finished_threads(); + map> query_threads; + set finished_query_threads; mutex query_threads_mutex; shared_ptr proxy; shared_ptr atomdb; diff --git a/src/main/bus_node.cc b/src/main/bus_node.cc index f278ff35..6fb8c122 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,20 @@ 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. + 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/scripts/run_toy.sh b/src/tests/scripts/run_toy.sh index 69645d24..b627615d 100755 --- a/src/tests/scripts/run_toy.sh +++ b/src/tests/scripts/run_toy.sh @@ -16,28 +16,30 @@ POPULATION_SIZE=100 NUM_GENERATIONS=5 NUM_ITERATIONS=5 -echo "--------------------------------------------------" -echo "Cleaning docker containers" -echo -das-cli db stop --prune -docker stop `docker ps | tail -n +2 | cut -d" " -f1` ; docker container prune -f ; docker ps -a +# echo "--------------------------------------------------" +# echo "Cleaning docker containers" +# echo +# das-cli db stop --prune +# docker stop `docker ps | tail -n +2 | cut -d" " -f1` ; docker container prune -f ; docker ps -a -echo -echo "--------------------------------------------------" -echo "Loading knowledge base" -echo -das-cli db start -make run-db-loader OPTIONS="--config=/opt/das/config/das.json --file=$KB --threads=4 --chunk=5000" +# echo +# echo "--------------------------------------------------" +# echo "Loading knowledge base" +# echo +# das-cli db start +# make run-db-loader OPTIONS="--config=/opt/das/config/das.json --file=$KB --threads=4 --chunk=5000" + +rm -f /tmp/ab.log /tmp/qa.log /tmp/ev.log echo echo "--------------------------------------------------" echo "Stating agents" echo -make run-attention-broker &>> /tmp/ab.log & +make run-attention-broker 2>&1 >> /tmp/ab.log & sleep 2 -make run-busnode OPTIONS="--service=query-engine --config=/opt/das/config/das.json" &>> /tmp/qa.log & +make run-busnode OPTIONS="--service=query-engine --config=/opt/das/config/das.json" 2>&1 >> /tmp/qa.log & sleep 2 -make run-busnode OPTIONS="--service=evolution-agent --bus-endpoint=localhost:40002 --config=/opt/das/config/das.json" &>> /tmp/ev.log & +make run-busnode OPTIONS="--service=evolution-agent --bus-endpoint=localhost:40002 --config=/opt/das/config/das.json" 2>&1 >> /tmp/ev.log & sleep 2 docker update -m $AVAIABLE_RAM --memory-swap $AVAIABLE_RAM $(docker ps -q) echo "Done. Logs are in /tmp/ab.log /tmp/qa.log /tmp/ev.log" From 20b48b71b912003f509fe5cfb824bf156eab4442 Mon Sep 17 00:00:00 2001 From: CCG Date: Tue, 30 Jun 2026 19:32:31 -0300 Subject: [PATCH 2/7] Revert run_toy.sh --- src/tests/scripts/run_toy.sh | 30 ++++++++++++++---------------- 1 file changed, 14 insertions(+), 16 deletions(-) diff --git a/src/tests/scripts/run_toy.sh b/src/tests/scripts/run_toy.sh index b627615d..69645d24 100755 --- a/src/tests/scripts/run_toy.sh +++ b/src/tests/scripts/run_toy.sh @@ -16,30 +16,28 @@ POPULATION_SIZE=100 NUM_GENERATIONS=5 NUM_ITERATIONS=5 -# echo "--------------------------------------------------" -# echo "Cleaning docker containers" -# echo -# das-cli db stop --prune -# docker stop `docker ps | tail -n +2 | cut -d" " -f1` ; docker container prune -f ; docker ps -a - -# echo -# echo "--------------------------------------------------" -# echo "Loading knowledge base" -# echo -# das-cli db start -# make run-db-loader OPTIONS="--config=/opt/das/config/das.json --file=$KB --threads=4 --chunk=5000" +echo "--------------------------------------------------" +echo "Cleaning docker containers" +echo +das-cli db stop --prune +docker stop `docker ps | tail -n +2 | cut -d" " -f1` ; docker container prune -f ; docker ps -a -rm -f /tmp/ab.log /tmp/qa.log /tmp/ev.log +echo +echo "--------------------------------------------------" +echo "Loading knowledge base" +echo +das-cli db start +make run-db-loader OPTIONS="--config=/opt/das/config/das.json --file=$KB --threads=4 --chunk=5000" echo echo "--------------------------------------------------" echo "Stating agents" echo -make run-attention-broker 2>&1 >> /tmp/ab.log & +make run-attention-broker &>> /tmp/ab.log & sleep 2 -make run-busnode OPTIONS="--service=query-engine --config=/opt/das/config/das.json" 2>&1 >> /tmp/qa.log & +make run-busnode OPTIONS="--service=query-engine --config=/opt/das/config/das.json" &>> /tmp/qa.log & sleep 2 -make run-busnode OPTIONS="--service=evolution-agent --bus-endpoint=localhost:40002 --config=/opt/das/config/das.json" 2>&1 >> /tmp/ev.log & +make run-busnode OPTIONS="--service=evolution-agent --bus-endpoint=localhost:40002 --config=/opt/das/config/das.json" &>> /tmp/ev.log & sleep 2 docker update -m $AVAIABLE_RAM --memory-swap $AVAIABLE_RAM $(docker ps -q) echo "Done. Logs are in /tmp/ab.log /tmp/qa.log /tmp/ev.log" From 57b1b4c3463341b7d582854458807d446239b012 Mon Sep 17 00:00:00 2001 From: CCG Date: Tue, 30 Jun 2026 20:23:08 -0300 Subject: [PATCH 3/7] CodeRabbit changes --- .../evolution/QueryEvolutionProcessor.cc | 21 ++--- .../evolution/QueryEvolutionProcessor.h | 10 --- .../PatternMatchingQueryProcessor.cc | 21 ++--- .../PatternMatchingQueryProcessor.h | 10 --- src/commons/StoppableThread.cc | 45 ++++++++--- src/commons/StoppableThread.h | 9 +++ src/main/bus_node.cc | 5 +- src/tests/cpp/BUILD | 15 ++++ src/tests/cpp/stoppable_thread_test.cc | 80 +++++++++++++++++++ 9 files changed, 152 insertions(+), 64 deletions(-) create mode 100644 src/tests/cpp/stoppable_thread_test.cc diff --git a/src/agents/evolution/QueryEvolutionProcessor.cc b/src/agents/evolution/QueryEvolutionProcessor.cc index f3fe0ef3..6f261497 100644 --- a/src/agents/evolution/QueryEvolutionProcessor.cc +++ b/src/agents/evolution/QueryEvolutionProcessor.cc @@ -38,7 +38,6 @@ shared_ptr QueryEvolutionProcessor::factory_empty_proxy() { void QueryEvolutionProcessor::run_command(shared_ptr proxy) { lock_guard semaphore(this->query_threads_mutex); - reap_finished_threads(); auto query_proxy = dynamic_pointer_cast(proxy); string thread_id = "thread<" + proxy->my_id() + "_" + std::to_string(proxy->get_serial()) + ">"; LOG_DEBUG("Starting new thread: " << thread_id << " to run command: <" << proxy->get_command() @@ -80,27 +79,17 @@ void QueryEvolutionProcessor::thread_process_one_query(shared_ptr semaphore(this->query_threads_mutex); - this->finished_query_threads.insert(monitor->get_id()); + monitor->detach(); + this->query_threads.erase(monitor->get_id()); } LOG_DEBUG("Command finished: <" << proxy->get_command() << ">"); } -void QueryEvolutionProcessor::reap_finished_threads() { - // Must be called while holding query_threads_mutex. - for (const auto& thread_id : this->finished_query_threads) { - auto it = this->query_threads.find(thread_id); - if (it != this->query_threads.end()) { - it->second->stop(); // joins the (already finished) thread and releases its stack - this->query_threads.erase(it); - } - } - this->finished_query_threads.clear(); -} - shared_ptr QueryEvolutionProcessor::issue_sampling_query( shared_ptr proxy) { auto pm_proxy = diff --git a/src/agents/evolution/QueryEvolutionProcessor.h b/src/agents/evolution/QueryEvolutionProcessor.h index 73dc6975..65c46824 100644 --- a/src/agents/evolution/QueryEvolutionProcessor.h +++ b/src/agents/evolution/QueryEvolutionProcessor.h @@ -73,21 +73,11 @@ class QueryEvolutionProcessor : public BusCommandProcessor { vector, float>>& selected); void thread_process_one_query(shared_ptr, shared_ptr proxy); void remove_query_thread(const string& stoppable_thread_id); - - /** - * Joins and removes every query thread that already finished running. - * - * A running query thread cannot join itself, so it only marks its id as finished when it is - * about to return. The actual join + erase (which releases the thread stack and lets the - * StoppableThread memory be reclaimed) is performed here, from a different thread. - */ - void reap_finished_threads(); string answer_to_string_1(shared_ptr answer, shared_ptr db); string answer_to_string_2(shared_ptr answer, shared_ptr db); string answer_to_string(shared_ptr answer); map> query_threads; - set finished_query_threads; mutex query_threads_mutex; shared_ptr proxy; set visited_individuals; diff --git a/src/agents/query_engine/PatternMatchingQueryProcessor.cc b/src/agents/query_engine/PatternMatchingQueryProcessor.cc index 3375fe86..a2de3952 100644 --- a/src/agents/query_engine/PatternMatchingQueryProcessor.cc +++ b/src/agents/query_engine/PatternMatchingQueryProcessor.cc @@ -61,7 +61,6 @@ shared_ptr PatternMatchingQueryProcessor::factory_empty_proxy() void PatternMatchingQueryProcessor::run_command(shared_ptr proxy) { lock_guard semaphore(this->query_threads_mutex); - reap_finished_threads(); auto query_proxy = dynamic_pointer_cast(proxy); string thread_id = "thread<" + proxy->my_id() + "_" + std::to_string(proxy->get_serial()) + ">"; LOG_DEBUG("Starting new thread: " << thread_id << " to run command: <" << proxy->get_command() @@ -257,27 +256,17 @@ void PatternMatchingQueryProcessor::thread_process_one_query( #if defined(__GLIBC__) malloc_trim(0); #endif - // This thread cannot join itself, so just flag it as finished. The next run_command (or the - // graceful shutdown path) will join and erase it from query_threads. + // 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->finished_query_threads.insert(monitor->get_id()); + monitor->detach(); + this->query_threads.erase(monitor->get_id()); } LOG_DEBUG("Command finished: <" << proxy->get_command() << ">"); } -void PatternMatchingQueryProcessor::reap_finished_threads() { - // Must be called while holding query_threads_mutex. - for (const auto& thread_id : this->finished_query_threads) { - auto it = this->query_threads.find(thread_id); - if (it != this->query_threads.end()) { - it->second->stop(); // joins the (already finished) thread and releases its stack - this->query_threads.erase(it); - } - } - this->finished_query_threads.clear(); -} - void PatternMatchingQueryProcessor::remove_query_thread(const string& stoppable_thread_id) { lock_guard semaphore(this->query_threads_mutex); this->query_threads.erase(this->query_threads.find(stoppable_thread_id)); diff --git a/src/agents/query_engine/PatternMatchingQueryProcessor.h b/src/agents/query_engine/PatternMatchingQueryProcessor.h index a5ed5c7f..8aba78fa 100644 --- a/src/agents/query_engine/PatternMatchingQueryProcessor.h +++ b/src/agents/query_engine/PatternMatchingQueryProcessor.h @@ -88,17 +88,7 @@ class PatternMatchingQueryProcessor : public BusCommandProcessor { void remove_query_thread(const string& stoppable_thread_id); - /** - * Joins and removes every query thread that already finished running. - * - * A running query thread cannot join itself, so it only marks its id as finished when it is - * about to return. The actual join + erase (which releases the thread stack and lets the - * StoppableThread/query tree memory be reclaimed) is performed here, from a different thread. - */ - void reap_finished_threads(); - map> query_threads; - set finished_query_threads; mutex query_threads_mutex; shared_ptr proxy; shared_ptr atomdb; diff --git a/src/commons/StoppableThread.cc b/src/commons/StoppableThread.cc index 5bff31c1..1be2fc52 100644 --- a/src/commons/StoppableThread.cc +++ b/src/commons/StoppableThread.cc @@ -27,19 +27,42 @@ 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(); + { + // Set the flag under the lock, but join OUTSIDE the lock so the thread being joined can + // still call stopped() without deadlocking. + lock_guard guard(this->stop_flag_mutex); + if (this->stop_flag) { + return; // already stopped or detached + } 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; + } + if (this->thread_object == NULL) { + return; // nothing attached + } + 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; + this->thread_object = NULL; + } +} + +void StoppableThread::detach() { + { + lock_guard guard(this->stop_flag_mutex); + if (this->stop_flag) { + return; // already stopped or detached } + this->stop_flag = true; + } + if (this->thread_object != 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. + this->thread_object->detach(); + delete this->thread_object; + this->thread_object = NULL; } } 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/main/bus_node.cc b/src/main/bus_node.cc index 6fb8c122..b6b02918 100644 --- a/src/main/bus_node.cc +++ b/src/main/bus_node.cc @@ -41,7 +41,10 @@ int main(int argc, char* argv[]) { mallopt(M_ARENA_MAX, 4); } // Return free heap pages to the OS instead of caching unbounded amounts in the arenas. - mallopt(M_TRIM_THRESHOLD, 128 * 1024); + // 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 c1b0e9c1..9ab5aa19 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..b1ff3e4d --- /dev/null +++ b/src/tests/cpp/stoppable_thread_test.cc @@ -0,0 +1,80 @@ +#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 +} From 667c39dac046b9c54ad133f4170e3fbe07b9a401 Mon Sep 17 00:00:00 2001 From: CCG Date: Tue, 30 Jun 2026 21:45:33 -0300 Subject: [PATCH 4/7] CodeRabbit changes --- src/commons/StoppableThread.cc | 40 ++++++++++++++------------ src/tests/cpp/stoppable_thread_test.cc | 36 +++++++++++++++++++++++ 2 files changed, 57 insertions(+), 19 deletions(-) diff --git a/src/commons/StoppableThread.cc b/src/commons/StoppableThread.cc index 1be2fc52..63fb6738 100644 --- a/src/commons/StoppableThread.cc +++ b/src/commons/StoppableThread.cc @@ -27,42 +27,44 @@ void StoppableThread::attach(thread* thread_object) { this->thread_object = thre void StoppableThread::stop() { this->stop(true); } void StoppableThread::stop(bool join_thread) { + // 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; { - // Set the flag under the lock, but join OUTSIDE the lock so the thread being joined can - // still call stopped() without deadlocking. lock_guard guard(this->stop_flag_mutex); - if (this->stop_flag) { - return; // already stopped or detached - } this->stop_flag = true; + if (join_thread) { + // 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; + } } - if (this->thread_object == NULL) { - return; // nothing attached - } - if (join_thread) { + // 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); - this->thread_object->join(); + to_reap->join(); LOG_DEBUG("Thread joined: " << this->id << ". destroying it."); - delete this->thread_object; - this->thread_object = NULL; + delete to_reap; } } void StoppableThread::detach() { + thread* to_reap = NULL; { lock_guard guard(this->stop_flag_mutex); - if (this->stop_flag) { - return; // already stopped or detached - } this->stop_flag = true; + // Claim the thread exactly once (see stop()). + to_reap = this->thread_object; + this->thread_object = NULL; } - if (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. - this->thread_object->detach(); - delete this->thread_object; - this->thread_object = NULL; + to_reap->detach(); + delete to_reap; } } diff --git a/src/tests/cpp/stoppable_thread_test.cc b/src/tests/cpp/stoppable_thread_test.cc index b1ff3e4d..5fa6994a 100644 --- a/src/tests/cpp/stoppable_thread_test.cc +++ b/src/tests/cpp/stoppable_thread_test.cc @@ -78,3 +78,39 @@ TEST(StoppableThread, stop_after_detach_is_noop) { 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 +} From c9b8ac4f499e811a59a3f36cde281bd71b987ed6 Mon Sep 17 00:00:00 2001 From: CCG Date: Wed, 1 Jul 2026 11:18:08 -0300 Subject: [PATCH 5/7] Add lock_guard to ThreadSafeHeap::snapshot() --- src/commons/ThreadSafeHeap.h | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/src/commons/ThreadSafeHeap.h b/src/commons/ThreadSafeHeap.h index 5eaf8f3d..c73a1021 100644 --- a/src/commons/ThreadSafeHeap.h +++ b/src/commons/ThreadSafeHeap.h @@ -78,7 +78,16 @@ 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; + } while (!copy.empty()) { output.push_back(copy.top().element); copy.pop(); From f64566ed873d028f1d90b63487bc6554a69c77ee Mon Sep 17 00:00:00 2001 From: ccgsnet Date: Wed, 1 Jul 2026 11:27:00 -0300 Subject: [PATCH 6/7] CodeRabbit changes --- src/commons/ThreadSafeHeap.h | 1 + 1 file changed, 1 insertion(+) diff --git a/src/commons/ThreadSafeHeap.h b/src/commons/ThreadSafeHeap.h index c73a1021..f64108b2 100644 --- a/src/commons/ThreadSafeHeap.h +++ b/src/commons/ThreadSafeHeap.h @@ -88,6 +88,7 @@ class ThreadSafeHeap { 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(); From a02a4653d6c9a6ce92fc0cae0f57f98dd3601c43 Mon Sep 17 00:00:00 2001 From: CCG Date: Thu, 2 Jul 2026 11:44:19 -0300 Subject: [PATCH 7/7] get_id() before detach() --- src/agents/evolution/QueryEvolutionProcessor.cc | 2 +- src/agents/query_engine/PatternMatchingQueryProcessor.cc | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/agents/evolution/QueryEvolutionProcessor.cc b/src/agents/evolution/QueryEvolutionProcessor.cc index 6f261497..3f4b79b4 100644 --- a/src/agents/evolution/QueryEvolutionProcessor.cc +++ b/src/agents/evolution/QueryEvolutionProcessor.cc @@ -84,8 +84,8 @@ void QueryEvolutionProcessor::thread_process_one_query(shared_ptr semaphore(this->query_threads_mutex); - monitor->detach(); this->query_threads.erase(monitor->get_id()); + monitor->detach(); } LOG_DEBUG("Command finished: <" << proxy->get_command() << ">"); } diff --git a/src/agents/query_engine/PatternMatchingQueryProcessor.cc b/src/agents/query_engine/PatternMatchingQueryProcessor.cc index a2de3952..679183e8 100644 --- a/src/agents/query_engine/PatternMatchingQueryProcessor.cc +++ b/src/agents/query_engine/PatternMatchingQueryProcessor.cc @@ -261,8 +261,8 @@ void PatternMatchingQueryProcessor::thread_process_one_query( // for the next command. A thread cannot join itself, hence detach instead of join. { lock_guard semaphore(this->query_threads_mutex); - monitor->detach(); this->query_threads.erase(monitor->get_id()); + monitor->detach(); } LOG_DEBUG("Command finished: <" << proxy->get_command() << ">"); }