Stop training thread from processing training requests once cancelled. (#14423)
vkalintiris committed
Feb 3, 2023 at 21:40 UTC
3c73da09c3944947541ea359fe545d9f3f58a727
2 files changed
+21
-7
ml/Host.cc
+12
-5
@@ -223,6 +223,11 @@ void Host::train() {
223
TrainingRequest TrainingReq = P.first;
224
size_t Size = P.second;
225
226
+ if (ThreadsCancelled) {
227
+ info("Stopping training thread because it was cancelled.");
228
+ break;
229
+ }
230
+
231
usec_t AllottedUT = (Cfg.TrainEvery * RH->rrd_update_every * USEC_PER_SEC) / Size;
232
if (AllottedUT > USEC_PER_SEC)
233
AllottedUT = USEC_PER_SEC;
@@ -340,11 +345,13 @@ void Host::startAnomalyDetectionThreads() {
345
346
char Tag[NETDATA_THREAD_TAG_MAX + 1];
347
348
+// #define ML_DISABLE_JOINING
349
+
350
snprintfz(Tag, NETDATA_THREAD_TAG_MAX, "MLTR[%s]", rrdhost_hostname(RH));
344
- netdata_thread_create(&TrainingThread, Tag, NETDATA_THREAD_OPTION_DEFAULT, train_main, static_cast<void *>(this));
351
+ netdata_thread_create(&TrainingThread, Tag, NETDATA_THREAD_OPTION_JOINABLE, train_main, static_cast<void *>(this));
352
353
snprintfz(Tag, NETDATA_THREAD_TAG_MAX, "MLDT[%s]", rrdhost_hostname(RH));
347
- netdata_thread_create(&DetectionThread, Tag, NETDATA_THREAD_OPTION_DEFAULT, detect_main, static_cast<void *>(this));
354
+ netdata_thread_create(&DetectionThread, Tag, NETDATA_THREAD_OPTION_JOINABLE, detect_main, static_cast<void *>(this));
355
}
356
357
void Host::stopAnomalyDetectionThreads(bool join) {
@@ -362,7 +369,7 @@ void Host::stopAnomalyDetectionThreads(bool join) {
369
netdata_thread_cancel(DetectionThread);
370
}
371
365
- if(join && !ThreadsJoined) {
372
+ if (join && !ThreadsJoined) {
373
ThreadsJoined = true;
374
ThreadsRunning = false;
375
@@ -374,7 +381,7 @@ void Host::stopAnomalyDetectionThreads(bool join) {
381
// to enable again:
382
// NETDATA_THREAD_OPTION_DEFAULT needs to become NETDATA_THREAD_OPTION_JOINABLE
383
377
- //netdata_thread_join(TrainingThread, nullptr);
378
- //netdata_thread_join(DetectionThread, nullptr);
384
+ netdata_thread_join(TrainingThread, nullptr);
385
+ netdata_thread_join(DetectionThread, nullptr);
386
}
387
}
ml/Queue.h
+9
-2
@@ -32,8 +32,15 @@ public:
32
while (Q.empty()) {
33
pthread_cond_wait(&CV, M.inner());
34
35
- if (Exit)
36
- pthread_exit(nullptr);
35
+ if (Exit) {
36
+ // This should happen only when we are destroying a host.
37
+ // Callers should use a flag dedicated to checking if we
38
+ // are about to delete the host or exit the agent. The original
39
+ // implementation would call pthread_exit which would cause
40
+ // the queue's mutex to be destroyed twice (and fail on the
41
+ // 2nd time)
42
+ return { T(), 0 };
43
+ }
44
}
45
46
T V = Q.front();