@cryptotaxi247 / netdata-1
Code Commits Issues Pulls Sessions CI/CD
master
master
cc 107 lines 2.69 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "ml/ml_queue.h"
4 #include "ml_private.h"
5
6 ml_queue_t *ml_queue_init()
7 {
8 ml_queue_t *q = new ml_queue_t();
9
10 netdata_mutex_init(&q->mutex);
11 netdata_cond_init(&q->cond_var);
12 q->exit = false;
13 return q;
14 }
15
16 void ml_queue_destroy(ml_queue_t *q)
17 {
18 netdata_mutex_destroy(&q->mutex);
19 netdata_cond_destroy(&q->cond_var);
20 delete q;
21 }
22
23 void ml_queue_push(ml_queue_t *q, const ml_queue_item_t req)
24 {
25 netdata_mutex_lock(&q->mutex);
26
27 switch (req.type) {
28 case ML_QUEUE_ITEM_TYPE_CREATE_NEW_MODEL:
29 q->create_model_queue.push(req.create_new_model);
30 q->stats.total_create_new_model_requests_pushed += 1;
31 break;
32
33 case ML_QUEUE_ITEM_TYPE_ADD_EXISTING_MODEL:
34 q->add_model_queue.push(req.add_existing_model);
35 q->stats.total_add_existing_model_requests_pushed += 1;
36 break;
37
38 case ML_QUEUE_ITEM_STOP_REQUEST:
39 // Stop requests don't need to be queued
40 break;
41 }
42
43 netdata_cond_signal(&q->cond_var);
44 netdata_mutex_unlock(&q->mutex);
45 }
46
47 ml_queue_item_t ml_queue_pop(ml_queue_t *q)
48 {
49 netdata_mutex_lock(&q->mutex);
50
51 ml_queue_item_t req;
52 req.type = ML_QUEUE_ITEM_STOP_REQUEST;
53
54 while (q->create_model_queue.empty() && q->add_model_queue.empty()) {
55 netdata_cond_wait(&q->cond_var, &q->mutex);
56
57 if (q->exit) {
58 netdata_mutex_unlock(&q->mutex);
59 return req;
60 }
61 }
62
63 // Prioritize adding model requests
64 if (!q->add_model_queue.empty()) {
65 req.type = ML_QUEUE_ITEM_TYPE_ADD_EXISTING_MODEL;
66 req.add_existing_model = q->add_model_queue.front();
67 q->add_model_queue.pop();
68 q->stats.total_add_existing_model_requests_popped += 1;
69 } else if (!q->create_model_queue.empty()) {
70 req.type = ML_QUEUE_ITEM_TYPE_CREATE_NEW_MODEL;
71 req.create_new_model = q->create_model_queue.front();
72 q->create_model_queue.pop();
73 q->stats.total_create_new_model_requests_popped += 1;
74 }
75
76 netdata_mutex_unlock(&q->mutex);
77 return req;
78 }
79
80 ml_queue_size_t ml_queue_size(ml_queue_t *q)
81 {
82 netdata_mutex_lock(&q->mutex);
83 ml_queue_size_t qs = ml_queue_size_t {
84 q->create_model_queue.size(),
85 q->add_model_queue.size(),
86 };
87 netdata_mutex_unlock(&q->mutex);
88
89 return qs;
90 }
91
92 void ml_queue_signal(ml_queue_t *q)
93 {
94 netdata_mutex_lock(&q->mutex);
95 q->exit = true;
96 netdata_cond_signal(&q->cond_var);
97 netdata_mutex_unlock(&q->mutex);
98 }
99
100 ml_queue_stats_t ml_queue_stats(ml_queue_t *q)
101 {
102 netdata_mutex_lock(&q->mutex);
103 ml_queue_stats_t stats = q->stats;
104 netdata_mutex_unlock(&q->mutex);
105
106 return stats;
107 }