Release query-engine memory back to the OS

src/agents/evolution/QueryEvolutionProcessor.cc

@@ -1,5 +1,9 @@
 #include "QueryEvolutionProcessor.h"
 
+#if defined(__GLIBC__)
+#include <malloc.h>
+#endif
+
 #include "AttentionBrokerClient.h"
 #include "Hasher.h"
 #include "LinkSchema.h"
@@ -70,8 +74,20 @@ void QueryEvolutionProcessor::thread_process_one_query(shared_ptr<StoppableThrea
     } catch (const std::exception& exception) {
         proxy->raise_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<mutex> 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<PatternMatchingQueryProxy> QueryEvolutionProcessor::issue_sampling_query(

src/agents/evolution/QueryEvolutionProcessor.h

@@ -2,6 +2,7 @@
 
 #include <map>
 #include <memory>
+#include <set>
 #include <thread>
 
 #include "BusCommandProcessor.h"

src/agents/query_engine/PatternMatchingQueryProcessor.cc

@@ -3,6 +3,10 @@
 
 #include "Logger.h"
 
+#if defined(__GLIBC__)
+#include <malloc.h>
+#endif
+
 #include <map>
 
 #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<mutex> 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) {

src/agents/query_engine/PatternMatchingQueryProcessor.h

@@ -2,6 +2,7 @@
 
 #include <map>
 #include <memory>
+#include <set>
 #include <stack>
 #include <thread>
 

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<mutex> 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<mutex> 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() {

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.
      *

src/commons/ThreadSafeHeap.h

@@ -78,7 +78,17 @@ class ThreadSafeHeap {
     }
 
     void snapshot(vector<T>& output) {
-        priority_queue<HeapElement, vector<HeapElement>> 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<HeapElement, vector<HeapElement>> copy;
+        {
+            lock_guard<mutex> semaphore(this->api_mutex);
+            copy = this->queue;
+        }
+        output.reserve(output.size() + copy.size());
         while (!copy.empty()) {
             output.push_back(copy.top().element);
             copy.pop();

src/main/bus_node.cc

@@ -1,5 +1,9 @@
 #include <signal.h>
 
+#if defined(__GLIBC__)
+#include <malloc.h>
+#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);
 

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",

src/tests/cpp/stoppable_thread_test.cc

@@ -0,0 +1,116 @@
+#include <atomic>
+#include <map>
+#include <mutex>
+#include <string>
+#include <thread>
+
+#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<string, shared_ptr<StoppableThread>> threads;
+    mutex threads_mutex;
+    atomic<int> completed{0};
+
+    const int burst = 16;
+    for (int i = 0; i < burst; i++) {
+        string id = "worker_" + std::to_string(i);
+        lock_guard<mutex> guard(threads_mutex);
+        auto stoppable_thread = make_shared<StoppableThread>(id);
+        stoppable_thread->attach(new thread(
+            [&threads, &threads_mutex, &completed](shared_ptr<StoppableThread> 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<mutex> 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<mutex> guard(threads_mutex);
+            if (threads.empty()) {
+                break;
+            }
+        }
+        Utils::sleep(10);
+    }
+
+    lock_guard<mutex> 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<StoppableThread>("detach_then_stop");
+    atomic<bool> ran{false};
+    stoppable_thread->attach(new thread(
+        [&ran](shared_ptr<StoppableThread> 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<StoppableThread>("request_then_reap");
+    atomic<bool> ran{false};
+    atomic<bool> exited{false};
+    stoppable_thread->attach(new thread(
+        [&ran, &exited](shared_ptr<StoppableThread> 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
+}