Stop training thread from processing training requests once cancelled. (#14423)

2 files changed, 21 insertions, 7 deletions

diff --git a/ml/Host.cc b/ml/Host.cc
index 662b470776..a5f276a807 100644
--- a/ml/Host.cc
+++ b/ml/Host.cc
@@ -223,6 +223,11 @@ void Host::train() {
         TrainingRequest TrainingReq = P.first;
         size_t Size = P.second;
 
+        if (ThreadsCancelled) {
+            info("Stopping training thread because it was cancelled.");
+            break;
+        }
+
         usec_t AllottedUT = (Cfg.TrainEvery * RH->rrd_update_every * USEC_PER_SEC) / Size;
         if (AllottedUT > USEC_PER_SEC)
             AllottedUT = USEC_PER_SEC;
@@ -340,11 +345,13 @@ void Host::startAnomalyDetectionThreads() {
 
     char Tag[NETDATA_THREAD_TAG_MAX + 1];
 
+// #define ML_DISABLE_JOINING
+
     snprintfz(Tag, NETDATA_THREAD_TAG_MAX, "MLTR[%s]", rrdhost_hostname(RH));
-    netdata_thread_create(&TrainingThread, Tag, NETDATA_THREAD_OPTION_DEFAULT, train_main, static_cast<void *>(this));
+    netdata_thread_create(&TrainingThread, Tag, NETDATA_THREAD_OPTION_JOINABLE, train_main, static_cast<void *>(this));
 
     snprintfz(Tag, NETDATA_THREAD_TAG_MAX, "MLDT[%s]", rrdhost_hostname(RH));
-    netdata_thread_create(&DetectionThread, Tag, NETDATA_THREAD_OPTION_DEFAULT, detect_main, static_cast<void *>(this));
+    netdata_thread_create(&DetectionThread, Tag, NETDATA_THREAD_OPTION_JOINABLE, detect_main, static_cast<void *>(this));
 }
 
 void Host::stopAnomalyDetectionThreads(bool join) {
@@ -362,7 +369,7 @@ void Host::stopAnomalyDetectionThreads(bool join) {
         netdata_thread_cancel(DetectionThread);
     }
 
-    if(join && !ThreadsJoined) {
+    if (join && !ThreadsJoined) {
         ThreadsJoined = true;
         ThreadsRunning = false;
 
@@ -374,7 +381,7 @@ void Host::stopAnomalyDetectionThreads(bool join) {
         // to enable again:
         // NETDATA_THREAD_OPTION_DEFAULT needs to become NETDATA_THREAD_OPTION_JOINABLE
 
-        //netdata_thread_join(TrainingThread, nullptr);
-        //netdata_thread_join(DetectionThread, nullptr);
+        netdata_thread_join(TrainingThread, nullptr);
+        netdata_thread_join(DetectionThread, nullptr);
     }
 }
diff --git a/ml/Queue.h b/ml/Queue.h
index b78979565d..37a74bd077 100644
--- a/ml/Queue.h
+++ b/ml/Queue.h
@@ -32,8 +32,15 @@ public:
         while (Q.empty()) {
             pthread_cond_wait(&CV, M.inner());
 
-            if (Exit)
-                pthread_exit(nullptr);
+            if (Exit) {
+                // This should happen only when we are destroying a host.
+                // Callers should use a flag dedicated to checking if we
+                // are about to delete the host or exit the agent. The original
+                // implementation would call pthread_exit which would cause
+                // the queue's mutex to be destroyed twice (and fail on the
+                // 2nd time)
+                return { T(), 0 };
+            }
         }
 
         T V = Q.front();