@cryptotaxi247 / netdata
Code Commits Issues Pulls Sessions CI/CD
master
master
c 105 lines 2.89 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "aclk_query_queue.h"
4
5 #define MAX_QUERY_ENTRIES (512)
6
7 struct {
8 aclk_query_t query_workers[MAX_QUERY_ENTRIES];
9 int free_stack[MAX_QUERY_ENTRIES];
10 int top;
11 SPINLOCK spinlock;
12 } queryPool;
13
14 // Initialize the query pool
15 __attribute__((constructor)) void init_query_pool()
16 {
17 spinlock_init(&queryPool.spinlock);
18 for (int i = 0; i < MAX_QUERY_ENTRIES; i++) {
19 queryPool.free_stack[i] = i;
20 queryPool.query_workers[i].allocated = false;
21 }
22 queryPool.top = MAX_QUERY_ENTRIES;
23 }
24
25 static aclk_query_t *get_query()
26 {
27 spinlock_lock(&queryPool.spinlock);
28 if (queryPool.top == 0) {
29 spinlock_unlock(&queryPool.spinlock);
30 aclk_query_t *query = callocz(1, sizeof(aclk_query_t));
31 query->allocated = true;
32 return query;
33 }
34 int index = queryPool.free_stack[--queryPool.top];
35 memset(&queryPool.query_workers[index], 0, sizeof(aclk_query_t));
36 spinlock_unlock(&queryPool.spinlock);
37 return &queryPool.query_workers[index];
38 }
39
40 static void return_query(aclk_query_t *query)
41 {
42 if (unlikely(query->allocated)) {
43 freez(query);
44 return;
45 }
46 spinlock_lock(&queryPool.spinlock);
47 int index = (int) (query - queryPool.query_workers);
48 if (index < 0 || index >= MAX_QUERY_ENTRIES) {
49 spinlock_unlock(&queryPool.spinlock);
50 return; // Invalid (should not happen)
51 }
52 queryPool.free_stack[queryPool.top++] = index;
53 memset(query, 0, sizeof(aclk_query_t));
54 spinlock_unlock(&queryPool.spinlock);
55 }
56
57 aclk_query_t *aclk_query_new(aclk_query_type_t type)
58 {
59 aclk_query_t *query = get_query();
60 query->type = type;
61 now_monotonic_high_precision_timeval(&query->created_tv);
62 return query;
63 }
64
65 void aclk_query_free(aclk_query_t *query)
66 {
67 struct ctxs_checkpoint *cmd;
68 switch (query->type) {
69 case HTTP_API_V2:
70 freez(query->data.http_api_v2.payload);
71 if (query->data.http_api_v2.query != query->dedup_id)
72 freez(query->data.http_api_v2.query);
73 break;
74 case ALERT_START_STREAMING:
75 freez(query->data.node_id);
76 break;
77 case ALERT_CHECKPOINT:
78 freez(query->data.node_id);
79 freez(query->claim_id);
80 break;
81 case CREATE_NODE_INSTANCE:
82 freez(query->data.node_id);
83 freez(query->machine_guid);
84 break;
85 // keep following cases together
86 case CTX_STOP_STREAMING:
87 case CTX_CHECKPOINT:
88 cmd = query->data.payload;
89 freez(cmd->claim_id);
90 freez(cmd->node_id);
91 freez(cmd);
92 break;
93
94 default:
95 break;
96 }
97
98 freez(query->dedup_id);
99 freez(query->callback_topic);
100 freez(query->msg_id);
101
102 if (query->sync_completion)
103 aclk_sync_completion_signal(query->sync_completion);
104 return_query(query);
105 }