cancel ml threads on shutdown and join them on host free (#14240)
* cancel ml threads on shutdown and join them on host free * mark them not running before joining them
Costa Tsaousis committed
Jan 11, 2023 at 17:45 UTC
b00b62f0e784b5eb616868c60985292cb3a7239d
4 files changed
+36
-15
ml/Host.cc
+20
-12
@@ -213,7 +213,7 @@ void Host::train() {
213
worker_register_job_name(WORKER_JOB_TRAINING_TRAIN, "train");
214
worker_register_job_name(WORKER_JOB_TRAINING_STATS, "stats");
215
216
- service_register(SERVICE_THREAD_TYPE_NETDATA, NULL, (force_quit_t )ml_stop_anomaly_detection_threads, RH, true);
216
+ service_register(SERVICE_THREAD_TYPE_NETDATA, NULL, (force_quit_t )ml_cancel_anomaly_detection_threads, RH, true);
217
218
while (service_running(SERVICE_ML_TRAINING)) {
219
auto P = TrainingQueue.pop();
@@ -293,7 +293,7 @@ void Host::detect() {
293
worker_register_job_name(WORKER_JOB_DETECTION_STATS, "stats");
294
worker_register_job_name(WORKER_JOB_DETECTION_RESOURCES, "resources");
295
296
- service_register(SERVICE_THREAD_TYPE_NETDATA, NULL, (force_quit_t )ml_stop_anomaly_detection_threads, RH, true);
296
+ service_register(SERVICE_THREAD_TYPE_NETDATA, NULL, (force_quit_t )ml_cancel_anomaly_detection_threads, RH, true);
297
298
heartbeat_t HB;
299
heartbeat_init(&HB);
@@ -332,29 +332,37 @@ void Host::startAnomalyDetectionThreads() {
332
}
333
334
ThreadsRunning = true;
335
+ ThreadsCancelled = false;
336
+ ThreadsJoined = false;
337
338
char Tag[NETDATA_THREAD_TAG_MAX + 1];
339
340
snprintfz(Tag, NETDATA_THREAD_TAG_MAX, "TRAIN[%s]", rrdhost_hostname(RH));
339
- netdata_thread_create(&TrainingThread, Tag, NETDATA_THREAD_OPTION_DEFAULT, train_main, static_cast<void *>(this));
341
+ netdata_thread_create(&TrainingThread, Tag, NETDATA_THREAD_OPTION_JOINABLE, train_main, static_cast<void *>(this));
342
343
snprintfz(Tag, NETDATA_THREAD_TAG_MAX, "DETECT[%s]", rrdhost_hostname(RH));
342
- netdata_thread_create(&DetectionThread, Tag, NETDATA_THREAD_OPTION_DEFAULT, detect_main, static_cast<void *>(this));
344
+ netdata_thread_create(&DetectionThread, Tag, NETDATA_THREAD_OPTION_JOINABLE, detect_main, static_cast<void *>(this));
345
}
346
345
-void Host::stopAnomalyDetectionThreads() {
347
+void Host::stopAnomalyDetectionThreads(bool join) {
348
if (!ThreadsRunning) {
349
error("Anomaly detections threads for host %s have already been stopped.", rrdhost_hostname(RH));
350
return;
351
}
352
351
- ThreadsRunning = false;
353
+ if(!ThreadsCancelled) {
354
+ ThreadsCancelled = true;
355
353
- // Signal the training queue to stop popping-items
354
- TrainingQueue.signal();
355
- netdata_thread_cancel(TrainingThread);
356
- // netdata_thread_join(TrainingThread, nullptr);
356
+ // Signal the training queue to stop popping-items
357
+ TrainingQueue.signal();
358
+ netdata_thread_cancel(TrainingThread);
359
+ netdata_thread_cancel(DetectionThread);
360
+ }
361
358
- netdata_thread_cancel(DetectionThread);
359
- // netdata_thread_join(DetectionThread, nullptr);
362
+ if(join && !ThreadsJoined) {
363
+ ThreadsJoined = true;
364
+ ThreadsRunning = false;
365
+ netdata_thread_join(TrainingThread, nullptr);
366
+ netdata_thread_join(DetectionThread, nullptr);
367
+ }
368
}
ml/Host.h
+7
-2
@@ -26,7 +26,10 @@ public:
26
MLS(),
27
TS(),
28
HostAnomalyRate(0.0),
29
- ThreadsRunning(false) {}
29
+ ThreadsRunning(false),
30
+ ThreadsCancelled(false),
31
+ ThreadsJoined(false)
32
+ {}
33
34
void addChart(Chart *C);
35
void removeChart(Chart *C);
@@ -36,7 +39,7 @@ public:
39
void getDetectionInfoAsJson(nlohmann::json &Json) const;
40
41
void startAnomalyDetectionThreads();
39
- void stopAnomalyDetectionThreads();
42
+ void stopAnomalyDetectionThreads(bool join);
43
44
void scheduleForTraining(TrainingRequest TR);
45
void train();
@@ -50,6 +53,8 @@ private:
53
TrainingStats TS;
54
CalculatedNumber HostAnomalyRate{0.0};
55
std::atomic<bool> ThreadsRunning;
56
+ std::atomic<bool> ThreadsCancelled;
57
+ std::atomic<bool> ThreadsJoined;
58
59
Queue<TrainingRequest> TrainingQueue;
60
ml/ml.cc
+8
-1
@@ -156,7 +156,14 @@ void ml_start_anomaly_detection_threads(RRDHOST *RH) {
156
void ml_stop_anomaly_detection_threads(RRDHOST *RH) {
157
if (RH && RH->ml_host) {
158
Host *H = reinterpret_cast<Host *>(RH->ml_host);
159
- H->stopAnomalyDetectionThreads();
159
+ H->stopAnomalyDetectionThreads(true);
160
+ }
161
+}
162
+
163
+void ml_cancel_anomaly_detection_threads(RRDHOST *RH) {
164
+ if (RH && RH->ml_host) {
165
+ Host *H = reinterpret_cast<Host *>(RH->ml_host);
166
+ H->stopAnomalyDetectionThreads(false);
167
}
168
}
169
ml/ml.h
+1
@@ -31,6 +31,7 @@ void ml_dimension_delete(RRDDIM *RD);
31
32
void ml_start_anomaly_detection_threads(RRDHOST *RH);
33
void ml_stop_anomaly_detection_threads(RRDHOST *RH);
34
+void ml_cancel_anomaly_detection_threads(RRDHOST *RH);
35
36
char *ml_get_host_info(RRDHOST *RH);
37
char *ml_get_host_runtime_info(RRDHOST *RH);