break active-active loop from replicating non-existing child to each other (#13968)
break active-active loop from replicating non-existing child to each-other
Costa Tsaousis committed
Nov 9, 2022 at 20:43 UTC
96ba2dadb48439e667c62b721a6d8c41a92a5187
4 files changed
+12
-9
database/rrd.h
+2
@@ -731,6 +731,8 @@ typedef enum rrdhost_flags {
731
RRDHOST_FLAG_ACLK_STREAM_CONTEXTS = (1 << 24), // when set, we should send ACLK stream context updates
732
// Metadata
733
RRDHOST_FLAG_METADATA_UPDATE = (1 << 25), // metadata needs to be stored in the database
734
+
735
+ RRDHOST_FLAG_RRDPUSH_RECEIVER_DISCONNECTED = ( 1 << 26), // set when the receiver part is disconnected
736
} RRDHOST_FLAGS;
737
738
#define rrdhost_flag_check(host, flag) (__atomic_load_n(&((host)->flags), __ATOMIC_SEQ_CST) & (flag))
streaming/receiver.c
+5
@@ -726,6 +726,9 @@ static int rrdpush_receive(struct receiver_state *rpt)
726
727
rrdcontext_host_child_connected(rpt->host);
728
729
+
730
+ rrdhost_flag_clear(rpt->host, RRDHOST_FLAG_RRDPUSH_RECEIVER_DISCONNECTED);
731
+
732
size_t count = streaming_parser(rpt, &cd, fp_in, fp_out,
733
#ifdef ENABLE_HTTPS
734
(rpt->ssl.conn) ? &rpt->ssl : NULL
@@ -734,6 +737,8 @@ static int rrdpush_receive(struct receiver_state *rpt)
737
#endif
738
);
739
740
+ rrdhost_flag_set(rpt->host, RRDHOST_FLAG_RRDPUSH_RECEIVER_DISCONNECTED);
741
+
742
log_stream_connection(rpt->client_ip, rpt->client_port,
743
rpt->key, rpt->host->machine_guid, rpt->hostname,
744
"DISCONNECTED");
streaming/rrdpush.c
+2
-6
@@ -378,16 +378,12 @@ void rrdset_done_push(RRDSET *st) {
378
RRDHOST *host = st->rrdhost;
379
380
// fetch the flags we need to check with one atomic operation
381
- RRDHOST_FLAGS host_flags = rrdhost_flag_check(host,
382
- RRDHOST_FLAG_RRDPUSH_SENDER_READY_4_METRICS
383
- | RRDHOST_FLAG_RRDPUSH_SENDER_LOGGED_STATUS
384
- | RRDHOST_FLAG_RRDPUSH_SENDER_SPAWN
385
- );
381
+ RRDHOST_FLAGS host_flags = __atomic_load_n(&host->flags, __ATOMIC_SEQ_CST);
382
383
// check if we are not connected
384
if(unlikely(!(host_flags & RRDHOST_FLAG_RRDPUSH_SENDER_READY_4_METRICS))) {
385
390
- if(unlikely(!(host_flags & RRDHOST_FLAG_RRDPUSH_SENDER_SPAWN)))
386
+ if(unlikely(!(host_flags & (RRDHOST_FLAG_RRDPUSH_SENDER_SPAWN | RRDHOST_FLAG_RRDPUSH_RECEIVER_DISCONNECTED))))
387
rrdpush_sender_thread_spawn(host);
388
389
if(unlikely(!(host_flags & RRDHOST_FLAG_RRDPUSH_SENDER_LOGGED_STATUS))) {
streaming/sender.c
+3
-3
@@ -165,13 +165,13 @@ void sender_commit(struct sender_state *s, BUFFER *wb) {
165
166
167
static inline void rrdpush_sender_thread_close_socket(RRDHOST *host) {
168
- rrdhost_flag_clear(host, RRDHOST_FLAG_RRDPUSH_SENDER_READY_4_METRICS);
169
- rrdhost_flag_clear(host, RRDHOST_FLAG_RRDPUSH_SENDER_CONNECTED);
170
-
168
if(host->sender->rrdpush_sender_socket != -1) {
169
close(host->sender->rrdpush_sender_socket);
170
host->sender->rrdpush_sender_socket = -1;
171
}
172
+
173
+ rrdhost_flag_clear(host, RRDHOST_FLAG_RRDPUSH_SENDER_READY_4_METRICS);
174
+ rrdhost_flag_clear(host, RRDHOST_FLAG_RRDPUSH_SENDER_CONNECTED);
175
}
176
177
static inline void rrdpush_sender_add_host_variable_to_buffer(BUFFER *wb, const RRDVAR_ACQUIRED *rva) {