@cryptotaxi247 / netdata
Code Commits Issues Pulls Sessions CI/CD
master
master
c 103 lines 3.7 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "stream-control.h"
4 #include "stream.h"
5 #include "stream-replication-sender.h"
6
7 static struct {
8 PAD64(uint32_t) backfill_runners;
9 PAD64(uint32_t) replication_runners;
10 PAD64(uint32_t) user_data_queries_runners;
11 PAD64(uint32_t) user_weights_queries_runners;
12 } sc;
13
14 // --------------------------------------------------------------------------------------------------------------------
15 // backfilling
16
17 ALWAYS_INLINE static uint32_t backfill_runners(void) {
18 return __atomic_load_n(&sc.backfill_runners, __ATOMIC_RELAXED);
19 }
20
21 ALWAYS_INLINE void stream_control_backfill_query_started(void) {
22 __atomic_add_fetch(&sc.backfill_runners, 1, __ATOMIC_RELAXED);
23 }
24
25 ALWAYS_INLINE void stream_control_backfill_query_finished(void) {
26 __atomic_sub_fetch(&sc.backfill_runners, 1, __ATOMIC_RELAXED);
27 }
28
29 // --------------------------------------------------------------------------------------------------------------------
30 // replication
31
32 ALWAYS_INLINE static uint32_t replication_runners(void) {
33 return __atomic_load_n(&sc.replication_runners, __ATOMIC_RELAXED);
34 }
35
36 ALWAYS_INLINE void stream_control_replication_query_started(void) {
37 __atomic_add_fetch(&sc.replication_runners, 1, __ATOMIC_RELAXED);
38 }
39
40 ALWAYS_INLINE void stream_control_replication_query_finished(void) {
41 __atomic_sub_fetch(&sc.replication_runners, 1, __ATOMIC_RELAXED);
42 }
43
44 // --------------------------------------------------------------------------------------------------------------------
45 // user data queries
46
47 ALWAYS_INLINE static uint32_t user_data_query_runners(void) {
48 return __atomic_load_n(&sc.user_data_queries_runners, __ATOMIC_RELAXED);
49 }
50
51 ALWAYS_INLINE void stream_control_user_data_query_started(void) {
52 __atomic_add_fetch(&sc.user_data_queries_runners, 1, __ATOMIC_RELAXED);
53 }
54
55 ALWAYS_INLINE void stream_control_user_data_query_finished(void) {
56 __atomic_sub_fetch(&sc.user_data_queries_runners, 1, __ATOMIC_RELAXED);
57 }
58
59 // --------------------------------------------------------------------------------------------------------------------
60 // user weights queries
61
62 ALWAYS_INLINE static uint32_t user_weights_query_runners(void) {
63 return __atomic_load_n(&sc.user_weights_queries_runners, __ATOMIC_RELAXED);
64 }
65
66 ALWAYS_INLINE void stream_control_user_weights_query_started(void) {
67 __atomic_add_fetch(&sc.user_weights_queries_runners, 1, __ATOMIC_RELAXED);
68 }
69
70 ALWAYS_INLINE void stream_control_user_weights_query_finished(void) {
71 __atomic_sub_fetch(&sc.user_weights_queries_runners, 1, __ATOMIC_RELAXED);
72 }
73
74 // --------------------------------------------------------------------------------------------------------------------
75 // consumer API
76
77 ALWAYS_INLINE bool stream_control_ml_should_be_running(void) {
78 return backfill_runners() == 0 &&
79 replication_runners() == 0 &&
80 user_data_query_runners() == 0 &&
81 user_weights_query_runners() == 0;
82 }
83
84 ALWAYS_INLINE bool stream_control_children_should_be_accepted(void) {
85 // we should not check for replication here.
86 // replication benefits from multiple nodes (merges the extents)
87 // and also the nodes should be close in time in the db
88 // - checking for replication leaves the last few nodes locked-out (since all the others are replicating)
89
90 return backfill_runners() == 0;
91 }
92
93 ALWAYS_INLINE bool stream_control_replication_should_be_running(void) {
94 return backfill_runners() == 0 &&
95 user_data_query_runners() == 0 &&
96 user_weights_query_runners() == 0;
97 }
98
99 ALWAYS_INLINE bool stream_control_health_should_be_running(void) {
100 return backfill_runners() == 0 &&
101 // replication_runners() == 0 &&
102 (user_data_query_runners() + user_weights_query_runners()) <= 1;
103 }