blob: c9461b2338f2731ba52a584b5517f2ccb8ea34d5 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
|
// SPDX-License-Identifier: GPL-3.0-or-later
#include "aclk_query_queue.h"
#include "aclk_query.h"
#include "aclk_stats.h"
static netdata_mutex_t aclk_query_queue_mutex = NETDATA_MUTEX_INITIALIZER;
#define ACLK_QUEUE_LOCK netdata_mutex_lock(&aclk_query_queue_mutex)
#define ACLK_QUEUE_UNLOCK netdata_mutex_unlock(&aclk_query_queue_mutex)
static struct aclk_query_queue {
aclk_query_t head;
aclk_query_t tail;
int block_push;
} aclk_query_queue = {
.head = NULL,
.tail = NULL,
.block_push = 0
};
static inline int _aclk_queue_query(aclk_query_t query)
{
query->created = now_realtime_usec();
ACLK_QUEUE_LOCK;
if (aclk_query_queue.block_push) {
ACLK_QUEUE_UNLOCK;
if(!netdata_exit)
error("Query Queue is blocked from accepting new requests. This is normally the case when ACLK prepares to shutdown.");
aclk_query_free(query);
return 1;
}
if (!aclk_query_queue.head) {
aclk_query_queue.head = query;
aclk_query_queue.tail = query;
ACLK_QUEUE_UNLOCK;
return 0;
}
// TODO deduplication
aclk_query_queue.tail->next = query;
aclk_query_queue.tail = query;
ACLK_QUEUE_UNLOCK;
return 0;
}
int aclk_queue_query(aclk_query_t query)
{
int ret = _aclk_queue_query(query);
if (!ret) {
QUERY_THREAD_WAKEUP;
if (aclk_stats_enabled) {
ACLK_STATS_LOCK;
aclk_metrics_per_sample.queries_queued++;
ACLK_STATS_UNLOCK;
}
}
return ret;
}
aclk_query_t aclk_queue_pop(void)
{
aclk_query_t ret;
ACLK_QUEUE_LOCK;
if (aclk_query_queue.block_push) {
ACLK_QUEUE_UNLOCK;
if(!netdata_exit)
error("POP Query Queue is blocked from accepting new requests. This is normally the case when ACLK prepares to shutdown.");
return NULL;
}
ret = aclk_query_queue.head;
if (!ret) {
ACLK_QUEUE_UNLOCK;
return ret;
}
aclk_query_queue.head = ret->next;
if (unlikely(!aclk_query_queue.head))
aclk_query_queue.tail = aclk_query_queue.head;
ACLK_QUEUE_UNLOCK;
ret->next = NULL;
return ret;
}
void aclk_queue_flush(void)
{
aclk_query_t query = aclk_queue_pop();
while (query) {
aclk_query_free(query);
query = aclk_queue_pop();
};
}
aclk_query_t aclk_query_new(aclk_query_type_t type)
{
aclk_query_t query = callocz(1, sizeof(struct aclk_query));
query->type = type;
return query;
}
void aclk_query_free(aclk_query_t query)
{
if (query->type == HTTP_API_V2) {
freez(query->data.http_api_v2.payload);
if (query->data.http_api_v2.query != query->dedup_id)
freez(query->data.http_api_v2.query);
}
if (query->type == CHART_NEW)
freez(query->data.chart_add_del.chart_name);
if (query->type == ALARM_STATE_UPDATE && query->data.alarm_update)
json_object_put(query->data.alarm_update);
freez(query->dedup_id);
freez(query->callback_topic);
freez(query->msg_id);
freez(query);
}
void aclk_queue_lock(void)
{
ACLK_QUEUE_LOCK;
aclk_query_queue.block_push = 1;
ACLK_QUEUE_UNLOCK;
}
|
