replication fixes No 8 (#14061)
* replication requests with start_streaming=true are executed immediately upon reception, instead of being placed in the queue * disable thread cancelability while workers cleanup * remove obsolete worker from replication * multi-threaded replication with netdata.conf option to set number of replication threads * revert spinlock to mutex * separate worker and main thread worker jobs * restart the queue every 10 seconds only * use atomic for sender buffer percentage * reset the queue position after sleeping * use sender resets to sleep properly * fix condition * cleanup sender members related to replication
Costa Tsaousis committed
Nov 29, 2022 at 16:03 UTC
462988dac901e95e765cd6be2dc24a5c33595526
7 files changed
+489
-286
daemon/global_statistics.c
+4
-1
@@ -2474,6 +2474,8 @@ static int read_thread_cpu_time_from_proc_stat(pid_t pid __maybe_unused, kernel_
2474
static Pvoid_t workers_by_pid_JudyL_array = NULL;
2475
2476
static void workers_threads_cleanup(struct worker_utilization *wu) {
2477
+ netdata_thread_disable_cancelability();
2478
+
2479
struct worker_thread *t = wu->threads;
2480
while(t) {
2481
struct worker_thread *next = t->next;
@@ -2483,9 +2485,10 @@ static void workers_threads_cleanup(struct worker_utilization *wu) {
2485
DOUBLE_LINKED_LIST_REMOVE_UNSAFE(wu->threads, t, prev, next);
2486
freez(t);
2487
}
2486
-
2488
t = next;
2489
}
2490
+
2491
+ netdata_thread_enable_cancelability();
2492
}
2493
2494
static struct worker_thread *worker_thread_find(struct worker_utilization *wu __maybe_unused, pid_t pid) {
daemon/static_threads.c
+2
-2
@@ -134,7 +134,7 @@ const struct netdata_static_thread static_threads_common[] = {
134
#endif
135
136
{
137
- .name = "rrdcontext",
137
+ .name = "RRDCONTEXT",
138
.config_section = NULL,
139
.config_name = NULL,
140
.enabled = 1,
@@ -144,7 +144,7 @@ const struct netdata_static_thread static_threads_common[] = {
144
},
145
146
{
147
- .name = "replication",
147
+ .name = "REPLICATION",
148
.config_section = NULL,
149
.config_name = NULL,
150
.enabled = 1,
libnetdata/locks/locks.c
+4
@@ -287,6 +287,8 @@ void netdata_spinlock_init(SPINLOCK *spinlock) {
287
}
288
289
void netdata_spinlock_lock(SPINLOCK *spinlock) {
290
+ netdata_thread_disable_cancelability();
291
+
292
static const struct timespec ns = { .tv_sec = 0, .tv_nsec = 1 };
293
bool expected = false, desired = true;
294
@@ -306,6 +308,8 @@ void netdata_spinlock_lock(SPINLOCK *spinlock) {
308
309
void netdata_spinlock_unlock(SPINLOCK *spinlock) {
310
__atomic_store_n(&spinlock->locked, false, __ATOMIC_RELEASE);
311
+
312
+ netdata_thread_enable_cancelability();
313
}
314
315
#ifdef NETDATA_TRACE_RWLOCKS
streaming/replication.c
+450
-264
@@ -3,10 +3,34 @@
3
#include "replication.h"
4
#include "Judy.h"
5
6
+#define STREAMING_START_MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED 50
7
#define MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED 20
8
#define MIN_SENDER_BUFFER_PERCENTAGE_ALLOWED 10
9
10
+#define WORKER_JOB_FIND_NEXT 1
11
+#define WORKER_JOB_QUERYING 2
12
+#define WORKER_JOB_DELETE_ENTRY 3
13
+#define WORKER_JOB_FIND_CHART 4
14
+#define WORKER_JOB_CHECK_CONSISTENCY 5
15
+#define WORKER_JOB_BUFFER_COMMIT 6
16
+#define WORKER_JOB_CLEANUP 7
17
+
18
+// master thread worker jobs
19
+#define WORKER_JOB_STATISTICS 8
20
+#define WORKER_JOB_CUSTOM_METRIC_PENDING_REQUESTS 9
21
+#define WORKER_JOB_CUSTOM_METRIC_COMPLETION 10
22
+#define WORKER_JOB_CUSTOM_METRIC_ADDED 11
23
+#define WORKER_JOB_CUSTOM_METRIC_DONE 12
24
+#define WORKER_JOB_CUSTOM_METRIC_SKIPPED_NOT_CONNECTED 13
25
+#define WORKER_JOB_CUSTOM_METRIC_SKIPPED_NO_ROOM 14
26
+#define WORKER_JOB_CUSTOM_METRIC_WAITS 15
27
+#define WORKER_JOB_CUSTOM_METRIC_SENDER_RESETS 16
28
+
29
+#define ITERATIONS_IDLE_WITHOUT_PENDING_TO_RUN_SENDER_VERIFICATION 30
30
+#define SECONDS_TO_RESET_POINT_IN_TIME 10
31
+
32
static struct replication_query_statistics replication_queries = {
33
+ .spinlock = NETDATA_SPINLOCK_INITIALIZER,
34
.queries_started = 0,
35
.queries_finished = 0,
36
.points_read = 0,
@@ -14,7 +38,10 @@ static struct replication_query_statistics replication_queries = {
38
};
39
40
struct replication_query_statistics replication_get_query_statistics(void) {
17
- return replication_queries;
41
+ netdata_spinlock_lock(&replication_queries.spinlock);
42
+ struct replication_query_statistics ret = replication_queries;
43
+ netdata_spinlock_unlock(&replication_queries.spinlock);
44
+ return ret;
45
}
46
47
// ----------------------------------------------------------------------------
@@ -177,6 +204,7 @@ static time_t replicate_chart_timeframe(BUFFER *wb, RRDSET *st, time_t after, ti
204
205
// release all the dictionary items acquired
206
// finalize the queries
207
+ size_t queries = 0;
208
for(size_t i = 0; i < dimensions ;i++) {
209
struct replication_dimension *d = &data[i];
210
if(unlikely(!d->enabled)) continue;
@@ -186,12 +214,15 @@ static time_t replicate_chart_timeframe(BUFFER *wb, RRDSET *st, time_t after, ti
214
dictionary_acquired_item_release(d->dict, d->rda);
215
216
// update global statistics
189
- replication_queries.queries_started++;
190
- replication_queries.queries_finished++;
217
+ queries++;
218
}
219
220
+ netdata_spinlock_lock(&replication_queries.spinlock);
221
+ replication_queries.queries_started += queries;
222
+ replication_queries.queries_finished += queries;
223
replication_queries.points_read += points_read;
224
replication_queries.points_generated += points_generated;
225
+ netdata_spinlock_unlock(&replication_queries.spinlock);
226
227
return before;
228
}
@@ -323,7 +354,9 @@ bool replicate_chart_response(RRDHOST *host, RRDSET *st, bool start_streaming, t
354
, (unsigned long long)world_clock_time
355
);
356
357
+ worker_is_busy(WORKER_JOB_BUFFER_COMMIT);
358
sender_commit(host->sender, wb);
359
+ worker_is_busy(WORKER_JOB_CLEANUP);
360
361
return enable_streaming;
362
}
@@ -349,7 +382,7 @@ struct replication_request_details {
382
struct {
383
time_t first_entry_t; // the first entry time we have
384
time_t last_entry_t; // the last entry time we have
352
- bool last_entry_t_adjusted_to_now; // true, if the last entry time was in the future and we fixed
385
+ bool last_entry_t_adjusted_to_now; // true, if the last entry time was in the future, and we fixed
386
time_t now; // the current local world clock time
387
} local_db;
388
@@ -484,7 +517,7 @@ bool replicate_chart_request(send_command callback, void *callback_data, RRDHOST
517
else {
518
// we had sent a request - let's continue at the point we left it
519
// for this we don't take into account the actual data in our db
487
- // because the child may also have gaps and we need to get over it
520
+ // because the child may also have gaps, and we need to get over it
521
r.gap.from = r.last_request.before;
522
}
523
@@ -534,7 +567,7 @@ bool replicate_chart_request(send_command callback, void *callback_data, RRDHOST
567
if(r.wanted.after > r.wanted.before)
568
r.wanted.after = r.wanted.before;
569
537
- // the child should start streaming immediately if the wanted duration is small or we reached the last entry of the child
570
+ // the child should start streaming immediately if the wanted duration is small, or we reached the last entry of the child
571
r.wanted.start_streaming = (r.local_db.now - r.wanted.after <= host->rrdpush_replication_step || r.wanted.before == r.child_db.last_entry_t);
572
573
// the wanted timeframe is now r.wanted.after -> r.wanted.before
@@ -568,87 +601,137 @@ struct replication_sort_entry {
601
size_t unique_id; // used as a key to identify the sort entry - we never access its contents
602
};
603
604
+#define MAX_REPLICATION_THREADS 20 // + 1 for the main thread
605
+
606
// the global variables for the replication thread
607
static struct replication_thread {
608
netdata_mutex_t mutex;
609
575
- size_t pending;
576
- size_t added;
577
- size_t executed;
578
- size_t removed;
579
- size_t last_executed;
580
- time_t first_time_t;
581
- Word_t next_unique_id;
582
- struct replication_request *requests;
610
+ struct {
611
+ size_t pending; // number of requests pending in the queue
612
+ Word_t unique_id; // the last unique id we gave to a request (auto-increment, starting from 1)
613
+
614
+ // statistics
615
+ size_t added; // number of requests added to the queue
616
+ size_t removed; // number of requests removed from the queue
617
+ size_t skipped_not_connected; // number of requests skipped, because the sender is not connected to a parent
618
+ size_t skipped_no_room; // number of requests skipped, because the sender has no room for responses
619
+// size_t skipped_no_room_since_last_reset;
620
+ size_t sender_resets; // number of times a sender reset our last position in the queue
621
+ time_t first_time_t; // the minimum 'after' we encountered
622
+
623
+ struct {
624
+ Word_t after;
625
+ Word_t unique_id;
626
+ Pvoid_t JudyL_array;
627
+ } queue;
628
+
629
+ } unsafe; // protected from replication_recursive_lock()
630
584
- Word_t last_after;
585
- Word_t last_unique_id;
631
+ struct {
632
+ size_t executed; // the number of replication requests executed
633
+ size_t latest_first_time; // the 'after' timestamp of the last request we executed
634
+ } atomic; // access should be with atomic operations
635
587
- size_t skipped_not_connected;
588
- size_t skipped_no_room;
589
- size_t sender_resets;
590
- size_t waits;
636
+ struct {
637
+ size_t waits;
638
+ size_t last_executed; // caching of the atomic.executed to report number of requests executed since last time
639
592
- size_t skipped_no_room_last_run;
640
+ netdata_thread_t **threads_ptrs;
641
+ size_t threads;
642
+ } main_thread; // access is allowed only by the main thread
643
594
- Pvoid_t JudyL_array;
644
} replication_globals = {
645
.mutex = NETDATA_MUTEX_INITIALIZER,
597
- .pending = 0,
598
- .added = 0,
599
- .executed = 0,
600
- .last_executed = 0,
601
- .first_time_t = 0,
602
- .next_unique_id = 1,
603
- .skipped_no_room = 0,
604
- .skipped_no_room_last_run = 0,
605
- .skipped_not_connected = 0,
606
- .sender_resets = 0,
607
- .waits = 0,
608
- .requests = NULL,
609
- .JudyL_array = NULL,
646
+ .unsafe = {
647
+ .pending = 0,
648
+ .unique_id = 0,
649
+
650
+ .added = 0,
651
+ .removed = 0,
652
+ .skipped_not_connected = 0,
653
+ .skipped_no_room = 0,
654
+// .skipped_no_room_since_last_reset = 0,
655
+ .sender_resets = 0,
656
+
657
+ .first_time_t = 0,
658
+
659
+ .queue = {
660
+ .after = 0,
661
+ .unique_id = 0,
662
+ .JudyL_array = NULL,
663
+ },
664
+ },
665
+ .atomic = {
666
+ .executed = 0,
667
+ .latest_first_time = 0,
668
+ },
669
+ .main_thread = {
670
+ .waits = 0,
671
+ .last_executed = 0,
672
+ .threads = 0,
673
+ .threads_ptrs = NULL,
674
+ },
675
};
676
612
-static __thread int replication_recursive_mutex_recursions = 0;
677
+#define replication_set_latest_first_time(t) __atomic_store_n(&replication_globals.atomic.latest_first_time, t, __ATOMIC_RELAXED)
678
+#define replication_get_latest_first_time() __atomic_load_n(&replication_globals.atomic.latest_first_time, __ATOMIC_RELAXED)
679
614
-static void replication_recursive_lock() {
615
- if(++replication_recursive_mutex_recursions == 1)
616
- netdata_mutex_lock(&replication_globals.mutex);
680
+static inline bool replication_recursive_lock_mode(char mode) {
681
+ static __thread int recursions = 0;
682
618
-#ifdef NETDATA_INTERNAL_CHECKS
619
- if(replication_recursive_mutex_recursions < 0 || replication_recursive_mutex_recursions > 2)
620
- fatal("REPLICATION: recursions is %d", replication_recursive_mutex_recursions);
621
-#endif
622
-}
623
-
624
-static void replication_recursive_unlock() {
625
- if(--replication_recursive_mutex_recursions == 0)
626
- netdata_mutex_unlock(&replication_globals.mutex);
683
+ if(mode == 'L') { // (L)ock
684
+ if(++recursions == 1)
685
+ netdata_mutex_lock(&replication_globals.mutex);
686
+ }
687
+ else if(mode == 'U') { // (U)nlock
688
+ if(--recursions == 0)
689
+ netdata_mutex_unlock(&replication_globals.mutex);
690
+ }
691
+ else if(mode == 'C') { // (C)heck
692
+ if(recursions > 0)
693
+ return true;
694
+ else
695
+ return false;
696
+ }
697
+ else
698
+ fatal("REPLICATION: unknown lock mode '%c'", mode);
699
700
#ifdef NETDATA_INTERNAL_CHECKS
629
- if(replication_recursive_mutex_recursions < 0 || replication_recursive_mutex_recursions > 2)
630
- fatal("REPLICATION: recursions is %d", replication_recursive_mutex_recursions);
701
+ if(recursions < 0)
702
+ fatal("REPLICATION: recursions is %d", recursions);
703
#endif
704
+
705
+ return true;
706
}
707
708
+#define replication_recursive_lock() replication_recursive_lock_mode('L')
709
+#define replication_recursive_unlock() replication_recursive_lock_mode('U')
710
+#define fatal_when_replication_is_not_locked_for_me() do { \
711
+ if(!replication_recursive_lock_mode('C')) \
712
+ fatal("REPLICATION: reached %s, but replication is not locked by this thread.", __FUNCTION__); \
713
+} while(0)
714
+
715
void replication_set_next_point_in_time(time_t after, size_t unique_id) {
716
replication_recursive_lock();
636
- replication_globals.last_after = after;
637
- replication_globals.last_unique_id = unique_id;
717
+ replication_globals.unsafe.queue.after = after;
718
+ replication_globals.unsafe.queue.unique_id = unique_id;
719
replication_recursive_unlock();
720
}
721
722
// ----------------------------------------------------------------------------
723
// replication sort entry management
724
644
-static struct replication_sort_entry *replication_sort_entry_create(struct replication_request *rq) {
725
+static struct replication_sort_entry *replication_sort_entry_create_unsafe(struct replication_request *rq) {
726
+ fatal_when_replication_is_not_locked_for_me();
727
+
728
struct replication_sort_entry *rse = mallocz(sizeof(struct replication_sort_entry));
729
730
rrdpush_sender_pending_replication_requests_plus_one(rq->sender);
731
732
// copy the request
733
rse->rq = rq;
651
- rse->unique_id = replication_globals.next_unique_id++;
734
+ rse->unique_id = ++replication_globals.unsafe.unique_id;
735
736
// save the unique id into the request, to be able to delete it later
737
rq->unique_id = rse->unique_id;
@@ -663,30 +746,33 @@ static void replication_sort_entry_destroy(struct replication_sort_entry *rse) {
746
static struct replication_sort_entry *replication_sort_entry_add(struct replication_request *rq) {
747
replication_recursive_lock();
748
666
- struct replication_sort_entry *rse = replication_sort_entry_create(rq);
749
+ struct replication_sort_entry *rse = replication_sort_entry_create_unsafe(rq);
750
668
- if(rq->after < (time_t)replication_globals.last_after && rq->sender->buffer_used_percentage <= MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED && !replication_globals.skipped_no_room_last_run) {
669
- // make it find this request first
670
- replication_set_next_point_in_time(rq->after, rq->unique_id);
671
- }
751
+// if(rq->after < (time_t)replication_globals.protected.queue.after &&
752
+// rq->sender->buffer_used_percentage <= MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED &&
753
+// !replication_globals.protected.skipped_no_room_since_last_reset) {
754
+//
755
+// // make it find this request first
756
+// replication_set_next_point_in_time(rq->after, rq->unique_id);
757
+// }
758
673
- replication_globals.added++;
674
- replication_globals.pending++;
759
+ replication_globals.unsafe.added++;
760
+ replication_globals.unsafe.pending++;
761
762
Pvoid_t *inner_judy_ptr;
763
764
// find the outer judy entry, using after as key
679
- inner_judy_ptr = JudyLGet(replication_globals.JudyL_array, (Word_t) rq->after, PJE0);
765
+ inner_judy_ptr = JudyLGet(replication_globals.unsafe.queue.JudyL_array, (Word_t) rq->after, PJE0);
766
if(!inner_judy_ptr)
681
- inner_judy_ptr = JudyLIns(&replication_globals.JudyL_array, (Word_t) rq->after, PJE0);
767
+ inner_judy_ptr = JudyLIns(&replication_globals.unsafe.queue.JudyL_array, (Word_t) rq->after, PJE0);
768
769
// add it to the inner judy, using unique_id as key
770
Pvoid_t *item = JudyLIns(inner_judy_ptr, rq->unique_id, PJE0);
771
*item = rse;
772
rq->indexed_in_judy = true;
773
688
- if(!replication_globals.first_time_t || rq->after < replication_globals.first_time_t)
689
- replication_globals.first_time_t = rq->after;
774
+ if(!replication_globals.unsafe.first_time_t || rq->after < replication_globals.unsafe.first_time_t)
775
+ replication_globals.unsafe.first_time_t = rq->after;
776
777
replication_recursive_unlock();
778
@@ -694,10 +780,12 @@ static struct replication_sort_entry *replication_sort_entry_add(struct replicat
780
}
781
782
static bool replication_sort_entry_unlink_and_free_unsafe(struct replication_sort_entry *rse, Pvoid_t **inner_judy_ppptr) {
783
+ fatal_when_replication_is_not_locked_for_me();
784
+
785
bool inner_judy_deleted = false;
786
699
- replication_globals.removed++;
700
- replication_globals.pending--;
787
+ replication_globals.unsafe.removed++;
788
+ replication_globals.unsafe.pending--;
789
790
rrdpush_sender_pending_replication_requests_minus_one(rse->rq->sender);
791
@@ -708,7 +796,7 @@ static bool replication_sort_entry_unlink_and_free_unsafe(struct replication_sor
796
797
// if no items left, delete it from the outer judy
798
if(**inner_judy_ppptr == NULL) {
711
- JudyLDel(&replication_globals.JudyL_array, rse->rq->after, PJE0);
799
+ JudyLDel(&replication_globals.unsafe.queue.JudyL_array, rse->rq->after, PJE0);
800
inner_judy_deleted = true;
801
}
802
@@ -725,7 +813,7 @@ static void replication_sort_entry_del(struct replication_request *rq) {
813
replication_recursive_lock();
814
if(rq->indexed_in_judy) {
815
728
- inner_judy_pptr = JudyLGet(replication_globals.JudyL_array, rq->after, PJE0);
816
+ inner_judy_pptr = JudyLGet(replication_globals.unsafe.queue.JudyL_array, rq->after, PJE0);
817
if (inner_judy_pptr) {
818
Pvoid_t *our_item_pptr = JudyLGet(*inner_judy_pptr, rq->unique_id, PJE0);
819
if (our_item_pptr) {
@@ -754,68 +842,87 @@ static struct replication_request replication_request_get_first_available() {
842
Pvoid_t *inner_judy_pptr;
843
844
replication_recursive_lock();
757
- replication_globals.skipped_no_room_last_run = 0;
845
846
struct replication_request rq_to_return = (struct replication_request){ .found = false };
847
761
-
762
- if(unlikely(!replication_globals.last_after || !replication_globals.last_unique_id)) {
763
- replication_globals.last_after = 0;
764
- replication_globals.last_unique_id = 0;
848
+ if(unlikely(!replication_globals.unsafe.queue.after || !replication_globals.unsafe.queue.unique_id)) {
849
+ replication_globals.unsafe.queue.after = 0;
850
+ replication_globals.unsafe.queue.unique_id = 0;
851
}
852
767
- bool find_same_after = true;
768
- while(!rq_to_return.found && (inner_judy_pptr = JudyLFirstOrNext(replication_globals.JudyL_array, &replication_globals.last_after, find_same_after))) {
769
- Pvoid_t *our_item_pptr;
853
+ Word_t started_after = replication_globals.unsafe.queue.after;
854
+
855
+ size_t round = 0;
856
+ while(!rq_to_return.found) {
857
+ round++;
858
771
- while(!rq_to_return.found && (our_item_pptr = JudyLNext(*inner_judy_pptr, &replication_globals.last_unique_id, PJE0))) {
772
- struct replication_sort_entry *rse = *our_item_pptr;
773
- struct replication_request *rq = rse->rq;
774
- struct sender_state *s = rq->sender;
859
+ if(round > 2)
860
+ break;
861
776
- if(likely(s->buffer_used_percentage <= MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED)) {
777
- // there is room for this request in the sender buffer
862
+ if(round == 2) {
863
+ if(started_after == 0)
864
+ break;
865
779
- bool sender_is_connected =
780
- rrdhost_flag_check(s->host, RRDHOST_FLAG_RRDPUSH_SENDER_CONNECTED);
866
+ replication_globals.unsafe.queue.after = 0;
867
+ replication_globals.unsafe.queue.unique_id = 0;
868
+ }
869
782
- bool sender_has_been_flushed_since_this_request =
783
- rq->sender_last_flush_ut != rrdpush_sender_get_flush_time(s);
870
+ bool find_same_after = true;
871
+ while (!rq_to_return.found && (inner_judy_pptr = JudyLFirstOrNext(replication_globals.unsafe.queue.JudyL_array, &replication_globals.unsafe.queue.after, find_same_after))) {
872
+ Pvoid_t *our_item_pptr;
873
785
- if (unlikely(!sender_is_connected || sender_has_been_flushed_since_this_request)) {
786
- // skip this request, the sender is not connected or it has reconnected
874
+ if(unlikely(round == 2 && replication_globals.unsafe.queue.after > started_after))
875
+ break;
876
788
- replication_globals.skipped_not_connected++;
789
- if (replication_sort_entry_unlink_and_free_unsafe(rse, &inner_judy_pptr))
790
- // we removed the item from the outer JudyL
791
- break;
792
- }
793
- else {
794
- // this request is good to execute
877
+ while (!rq_to_return.found && (our_item_pptr = JudyLNext(*inner_judy_pptr, &replication_globals.unsafe.queue.unique_id, PJE0))) {
878
+ struct replication_sort_entry *rse = *our_item_pptr;
879
+ struct replication_request *rq = rse->rq;
880
+ struct sender_state *s = rq->sender;
881
796
- // copy the request to return it
797
- rq_to_return = *rq;
798
- rq_to_return.chart_id = string_dup(rq_to_return.chart_id);
882
+ if (likely(rrdpush_sender_get_buffer_used_percent(s) <= MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED)) {
883
+ // there is room for this request in the sender buffer
884
800
- // set the return result to found
801
- rq_to_return.found = true;
885
+ bool sender_is_connected =
886
+ rrdhost_flag_check(s->host, RRDHOST_FLAG_RRDPUSH_SENDER_CONNECTED);
887
803
- if (replication_sort_entry_unlink_and_free_unsafe(rse, &inner_judy_pptr))
804
- // we removed the item from the outer JudyL
805
- break;
888
+ bool sender_has_been_flushed_since_this_request =
889
+ rq->sender_last_flush_ut != rrdpush_sender_get_flush_time(s);
890
+
891
+ if (unlikely(!sender_is_connected || sender_has_been_flushed_since_this_request)) {
892
+ // skip this request, the sender is not connected, or it has reconnected
893
+
894
+ replication_globals.unsafe.skipped_not_connected++;
895
+ if (replication_sort_entry_unlink_and_free_unsafe(rse, &inner_judy_pptr))
896
+ // we removed the item from the outer JudyL
897
+ break;
898
+ }
899
+ else {
900
+ // this request is good to execute
901
+
902
+ // copy the request to return it
903
+ rq_to_return = *rq;
904
+ rq_to_return.chart_id = string_dup(rq_to_return.chart_id);
905
+
906
+ // set the return result to found
907
+ rq_to_return.found = true;
908
+
909
+ if (replication_sort_entry_unlink_and_free_unsafe(rse, &inner_judy_pptr))
910
+ // we removed the item from the outer JudyL
911
+ break;
912
+ }
913
+ }
914
+ else {
915
+ replication_globals.unsafe.skipped_no_room++;
916
+// replication_globals.protected.skipped_no_room_since_last_reset++;
917
}
918
}
808
- else {
809
- replication_globals.skipped_no_room++;
810
- replication_globals.skipped_no_room_last_run++;
811
- }
812
- }
919
814
- // call JudyLNext from now on
815
- find_same_after = false;
920
+ // call JudyLNext from now on
921
+ find_same_after = false;
922
817
- // prepare for the next iteration on the outer loop
818
- replication_globals.last_unique_id = 0;
923
+ // prepare for the next iteration on the outer loop
924
+ replication_globals.unsafe.queue.unique_id = 0;
925
+ }
926
}
927
928
replication_recursive_unlock();
@@ -889,6 +996,58 @@ static void replication_request_delete_callback(const DICTIONARY_ITEM *item __ma
996
string_freez(rq->chart_id);
997
}
998
999
+static bool replication_execute_request(struct replication_request *rq, bool workers) {
1000
+ bool ret = false;
1001
+
1002
+ if(likely(workers))
1003
+ worker_is_busy(WORKER_JOB_FIND_CHART);
1004
+
1005
+ RRDSET *st = rrdset_find(rq->sender->host, string2str(rq->chart_id));
1006
+ if(!st) {
1007
+ internal_error(true, "REPLAY ERROR: 'host:%s/chart:%s' not found",
1008
+ rrdhost_hostname(rq->sender->host), string2str(rq->chart_id));
1009
+
1010
+ goto cleanup;
1011
+ }
1012
+
1013
+ if(likely(workers))
1014
+ worker_is_busy(WORKER_JOB_QUERYING);
1015
+
1016
+ netdata_thread_disable_cancelability();
1017
+
1018
+ // send the replication data
1019
+ bool start_streaming = replicate_chart_response(
1020
+ st->rrdhost, st, rq->start_streaming, rq->after, rq->before);
1021
+
1022
+ netdata_thread_enable_cancelability();
1023
+
1024
+ if(start_streaming && rq->sender_last_flush_ut == rrdpush_sender_get_flush_time(rq->sender)) {
1025
+ // enable normal streaming if we have to
1026
+ // but only if the sender buffer has not been flushed since we started
1027
+
1028
+ if(rrdset_flag_check(st, RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS)) {
1029
+ rrdset_flag_clear(st, RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS);
1030
+ rrdset_flag_set(st, RRDSET_FLAG_SENDER_REPLICATION_FINISHED);
1031
+ rrdhost_sender_replicating_charts_minus_one(st->rrdhost);
1032
+
1033
+#ifdef NETDATA_LOG_REPLICATION_REQUESTS
1034
+ internal_error(true, "STREAM_SENDER REPLAY: 'host:%s/chart:%s' streaming starts",
1035
+ rrdhost_hostname(st->rrdhost), rrdset_id(st));
1036
+#endif
1037
+ }
1038
+ else
1039
+ internal_error(true, "REPLAY ERROR: 'host:%s/chart:%s' received start streaming command, but the chart is not in progress replicating",
1040
+ rrdhost_hostname(st->rrdhost), string2str(rq->chart_id));
1041
+ }
1042
+
1043
+ __atomic_add_fetch(&replication_globals.atomic.executed, 1, __ATOMIC_RELAXED);
1044
+
1045
+ ret = true;
1046
+
1047
+cleanup:
1048
+ string_freez(rq->chart_id);
1049
+ return ret;
1050
+}
1051
1052
// ----------------------------------------------------------------------------
1053
// public API
@@ -903,27 +1062,31 @@ void replication_add_request(struct sender_state *sender, const char *chart_id,
1062
.sender_last_flush_ut = rrdpush_sender_get_flush_time(sender),
1063
};
1064
906
- dictionary_set(sender->replication_requests, chart_id, &rq, sizeof(struct replication_request));
1065
+ if(start_streaming && rrdpush_sender_get_buffer_used_percent(sender) <= STREAMING_START_MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED)
1066
+ replication_execute_request(&rq, false);
1067
+
1068
+ else
1069
+ dictionary_set(sender->replication.requests, chart_id, &rq, sizeof(struct replication_request));
1070
}
1071
1072
void replication_sender_delete_pending_requests(struct sender_state *sender) {
1073
// allow the dictionary destructor to go faster on locks
1074
replication_recursive_lock();
912
- dictionary_flush(sender->replication_requests);
1075
+ dictionary_flush(sender->replication.requests);
1076
replication_recursive_unlock();
1077
}
1078
1079
void replication_init_sender(struct sender_state *sender) {
917
- sender->replication_requests = dictionary_create(DICT_OPTION_DONT_OVERWRITE_VALUE);
918
- dictionary_register_react_callback(sender->replication_requests, replication_request_react_callback, sender);
919
- dictionary_register_conflict_callback(sender->replication_requests, replication_request_conflict_callback, sender);
920
- dictionary_register_delete_callback(sender->replication_requests, replication_request_delete_callback, sender);
1080
+ sender->replication.requests = dictionary_create(DICT_OPTION_DONT_OVERWRITE_VALUE);
1081
+ dictionary_register_react_callback(sender->replication.requests, replication_request_react_callback, sender);
1082
+ dictionary_register_conflict_callback(sender->replication.requests, replication_request_conflict_callback, sender);
1083
+ dictionary_register_delete_callback(sender->replication.requests, replication_request_delete_callback, sender);
1084
}
1085
1086
void replication_cleanup_sender(struct sender_state *sender) {
1087
// allow the dictionary destructor to go faster on locks
1088
replication_recursive_lock();
926
- dictionary_destroy(sender->replication_requests);
1089
+ dictionary_destroy(sender->replication.requests);
1090
replication_recursive_unlock();
1091
}
1092
@@ -932,61 +1095,32 @@ void replication_recalculate_buffer_used_ratio_unsafe(struct sender_state *s) {
1095
size_t percentage = (s->buffer->max_size - available) * 100 / s->buffer->max_size;
1096
1097
if(percentage > MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED)
935
- s->replication_reached_max = true;
1098
+ s->replication.unsafe.reached_max = true;
1099
937
- if(s->replication_reached_max &&
1100
+ if(s->replication.unsafe.reached_max &&
1101
percentage <= MIN_SENDER_BUFFER_PERCENTAGE_ALLOWED) {
939
- s->replication_reached_max = false;
1102
+ s->replication.unsafe.reached_max = false;
1103
replication_recursive_lock();
941
- replication_set_next_point_in_time(0, 0);
942
- replication_globals.sender_resets++;
1104
+// replication_set_next_point_in_time(0, 0);
1105
+ replication_globals.unsafe.sender_resets++;
1106
replication_recursive_unlock();
1107
}
1108
946
- s->buffer_used_percentage = percentage;
1109
+ rrdpush_sender_set_buffer_used_percent(s, percentage);
1110
}
1111
1112
// ----------------------------------------------------------------------------
1113
// replication thread
1114
952
-static void replication_main_cleanup(void *ptr) {
953
- struct netdata_static_thread *static_thread = (struct netdata_static_thread *)ptr;
954
- static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;
955
-
956
- // custom code
957
- worker_unregister();
958
-
959
- static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;
960
-}
961
-
962
-#define WORKER_JOB_FIND_NEXT 1
963
-#define WORKER_JOB_QUERYING 2
964
-#define WORKER_JOB_DELETE_ENTRY 3
965
-#define WORKER_JOB_FIND_CHART 4
966
-#define WORKER_JOB_STATISTICS 5
967
-#define WORKER_JOB_ACTIVATE_ENABLE_STREAMING 6
968
-#define WORKER_JOB_CUSTOM_METRIC_PENDING_REQUESTS 7
969
-#define WORKER_JOB_CUSTOM_METRIC_COMPLETION 8
970
-#define WORKER_JOB_CUSTOM_METRIC_ADDED 9
971
-#define WORKER_JOB_CUSTOM_METRIC_DONE 10
972
-#define WORKER_JOB_CUSTOM_METRIC_SKIPPED_NOT_CONNECTED 11
973
-#define WORKER_JOB_CUSTOM_METRIC_SKIPPED_NO_ROOM 12
974
-#define WORKER_JOB_CUSTOM_METRIC_SENDER_RESETS 13
975
-#define WORKER_JOB_CUSTOM_METRIC_WAITS 14
976
-#define WORKER_JOB_CHECK_CONSISTENCY 15
977
-
978
-#define ITERATIONS_IDLE_WITHOUT_PENDING_TO_RUN_SENDER_VERIFICATION 10
979
-#define SECONDS_TO_RESET_POINT_IN_TIME 10
980
-
1115
static size_t verify_host_charts_are_streaming_now(RRDHOST *host) {
1116
internal_error(
1117
host->sender &&
984
- !host->sender->replication_pending_requests &&
985
- dictionary_entries(host->sender->replication_requests) != 0,
1118
+ !rrdpush_sender_pending_replication_requests(host->sender) &&
1119
+ dictionary_entries(host->sender->replication.requests) != 0,
1120
"REPLICATION SUMMARY: 'host:%s' reports %zu pending replication requests, but its chart replication index says there are %zu charts pending replication",
1121
rrdhost_hostname(host),
988
- host->sender->replication_pending_requests,
989
- dictionary_entries(host->sender->replication_requests)
1122
+ rrdpush_sender_pending_replication_requests(host->sender),
1123
+ dictionary_entries(host->sender->replication.requests)
1124
);
1125
1126
size_t ok = 0;
@@ -1039,42 +1173,131 @@ static void verify_all_hosts_charts_are_streaming_now(void) {
1173
errors += verify_host_charts_are_streaming_now(host);
1174
dfe_done(host);
1175
1042
- size_t executed = replication_globals.executed;
1043
- info("REPLICATION SUMMARY: finished, executed %zu replication requests, %zu charts pending replication", executed - replication_globals.last_executed, errors);
1044
- replication_globals.last_executed = executed;
1176
+ size_t executed = __atomic_load_n(&replication_globals.atomic.executed, __ATOMIC_RELAXED);
1177
+ info("REPLICATION SUMMARY: finished, executed %zu replication requests, %zu charts pending replication",
1178
+ executed - replication_globals.main_thread.last_executed, errors);
1179
+ replication_globals.main_thread.last_executed = executed;
1180
}
1181
1047
-void *replication_thread_main(void *ptr __maybe_unused) {
1048
- netdata_thread_cleanup_push(replication_main_cleanup, ptr);
1049
-
1182
+static void replication_initialize_workers(bool master) {
1183
worker_register("REPLICATION");
1051
-
1184
worker_register_job_name(WORKER_JOB_FIND_NEXT, "find next");
1185
worker_register_job_name(WORKER_JOB_QUERYING, "querying");
1186
worker_register_job_name(WORKER_JOB_DELETE_ENTRY, "dict delete");
1187
worker_register_job_name(WORKER_JOB_FIND_CHART, "find chart");
1056
- worker_register_job_name(WORKER_JOB_ACTIVATE_ENABLE_STREAMING, "enable streaming");
1188
worker_register_job_name(WORKER_JOB_CHECK_CONSISTENCY, "check consistency");
1058
- worker_register_job_name(WORKER_JOB_STATISTICS, "statistics");
1189
+ worker_register_job_name(WORKER_JOB_BUFFER_COMMIT, "commit");
1190
+ worker_register_job_name(WORKER_JOB_CLEANUP, "cleanup");
1191
+
1192
+ if(master) {
1193
+ worker_register_job_name(WORKER_JOB_STATISTICS, "statistics");
1194
+ worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_PENDING_REQUESTS, "pending requests", "requests", WORKER_METRIC_ABSOLUTE);
1195
+ worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_COMPLETION, "completion", "%", WORKER_METRIC_ABSOLUTE);
1196
+ worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_ADDED, "added requests", "requests/s", WORKER_METRIC_INCREMENTAL_TOTAL);
1197
+ worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_DONE, "finished requests", "requests/s", WORKER_METRIC_INCREMENTAL_TOTAL);
1198
+ worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_SKIPPED_NOT_CONNECTED, "not connected requests", "requests/s", WORKER_METRIC_INCREMENTAL_TOTAL);
1199
+ worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_SKIPPED_NO_ROOM, "no room requests", "requests/s", WORKER_METRIC_INCREMENTAL_TOTAL);
1200
+ worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_SENDER_RESETS, "sender resets", "resets/s", WORKER_METRIC_INCREMENTAL_TOTAL);
1201
+ worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_WAITS, "waits", "waits/s", WORKER_METRIC_INCREMENTAL_TOTAL);
1202
+ }
1203
+}
1204
+
1205
+#define REQUEST_OK (0)
1206
+#define REQUEST_QUEUE_EMPTY (-1)
1207
+#define REQUEST_CHART_NOT_FOUND (-2)
1208
+
1209
+static int replication_execute_next_pending_request(void) {
1210
+ worker_is_busy(WORKER_JOB_FIND_NEXT);
1211
+ struct replication_request rq = replication_request_get_first_available();
1212
+
1213
+ if(unlikely(!rq.found))
1214
+ return REQUEST_QUEUE_EMPTY;
1215
+
1216
+ // delete the request from the dictionary
1217
+ worker_is_busy(WORKER_JOB_DELETE_ENTRY);
1218
+ if(!dictionary_del(rq.sender->replication.requests, string2str(rq.chart_id)))
1219
+ error("REPLAY ERROR: 'host:%s/chart:%s' failed to be deleted from sender pending charts index",
1220
+ rrdhost_hostname(rq.sender->host), string2str(rq.chart_id));
1221
+
1222
+ replication_set_latest_first_time(rq.after);
1223
+
1224
+ if(unlikely(!replication_execute_request(&rq, true)))
1225
+ return REQUEST_CHART_NOT_FOUND;
1226
+
1227
+ return REQUEST_OK;
1228
+}
1229
+
1230
+static void replication_worker_cleanup(void *ptr __maybe_unused) {
1231
+ worker_unregister();
1232
+}
1233
+
1234
+static void *replication_worker_thread(void *ptr) {
1235
+ replication_initialize_workers(false);
1236
+
1237
+ netdata_thread_cleanup_push(replication_worker_cleanup, ptr);
1238
+
1239
+ while(!netdata_exit) {
1240
+ if(unlikely(replication_execute_next_pending_request() == REQUEST_QUEUE_EMPTY)) {
1241
+ worker_is_idle();
1242
+ sleep_usec(1 * USEC_PER_SEC);
1243
+ }
1244
+ }
1245
+
1246
+ netdata_thread_cleanup_pop(1);
1247
+ return NULL;
1248
+}
1249
+
1250
+static void replication_main_cleanup(void *ptr) {
1251
+ struct netdata_static_thread *static_thread = (struct netdata_static_thread *)ptr;
1252
+ static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;
1253
1060
- worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_PENDING_REQUESTS, "pending requests", "requests", WORKER_METRIC_ABSOLUTE);
1061
- worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_COMPLETION, "completion", "%", WORKER_METRIC_ABSOLUTE);
1062
- worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_ADDED, "added requests", "requests/s", WORKER_METRIC_INCREMENTAL_TOTAL);
1063
- worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_DONE, "finished requests", "requests/s", WORKER_METRIC_INCREMENTAL_TOTAL);
1064
- worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_SKIPPED_NOT_CONNECTED, "not connected requests", "requests/s", WORKER_METRIC_INCREMENTAL_TOTAL);
1065
- worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_SKIPPED_NO_ROOM, "no room requests", "requests/s", WORKER_METRIC_INCREMENTAL_TOTAL);
1066
- worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_SENDER_RESETS, "sender resets", "resets/s", WORKER_METRIC_INCREMENTAL_TOTAL);
1067
- worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_WAITS, "waits", "waits/s", WORKER_METRIC_INCREMENTAL_TOTAL);
1254
+ int threads = (int)replication_globals.main_thread.threads;
1255
+ for(int i = 0; i < threads ;i++) {
1256
+ netdata_thread_join(*replication_globals.main_thread.threads_ptrs[i], NULL);
1257
+ freez(replication_globals.main_thread.threads_ptrs[i]);
1258
+ }
1259
+ freez(replication_globals.main_thread.threads_ptrs);
1260
+ replication_globals.main_thread.threads_ptrs = NULL;
1261
+
1262
+ // custom code
1263
+ worker_unregister();
1264
+
1265
+ static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;
1266
+}
1267
+
1268
+void *replication_thread_main(void *ptr __maybe_unused) {
1269
+ replication_initialize_workers(true);
1270
+
1271
+ int threads = config_get_number(CONFIG_SECTION_DB, "replication threads", 1);
1272
+ if(threads < 1 || threads > MAX_REPLICATION_THREADS) {
1273
+ error("replication threads given %d is invalid, resetting to 1", threads);
1274
+ threads = 1;
1275
+ }
1276
+
1277
+ if(--threads) {
1278
+ replication_globals.main_thread.threads = threads;
1279
+ replication_globals.main_thread.threads_ptrs = mallocz(threads * sizeof(netdata_thread_t *));
1280
+
1281
+ for(int i = 0; i < threads ;i++) {
1282
+ replication_globals.main_thread.threads_ptrs[i] = mallocz(sizeof(netdata_thread_t));
1283
+ netdata_thread_create(replication_globals.main_thread.threads_ptrs[i], "REPLICATION",
1284
+ NETDATA_THREAD_OPTION_JOINABLE, replication_worker_thread, NULL);
1285
+ }
1286
+ }
1287
+
1288
+ netdata_thread_cleanup_push(replication_main_cleanup, ptr);
1289
1290
// start from 100% completed
1291
worker_set_metric(WORKER_JOB_CUSTOM_METRIC_COMPLETION, 100.0);
1292
1072
- time_t latest_first_time_t = 0;
1293
long run_verification_countdown = LONG_MAX; // LONG_MAX to prevent an initial verification when no replication ever took place
1294
bool slow = true; // control the time we sleep - it has to start with true!
1295
usec_t last_now_mono_ut = now_monotonic_usec();
1296
time_t replication_reset_next_point_in_time_countdown = SECONDS_TO_RESET_POINT_IN_TIME; // restart from the beginning every 10 seconds
1297
1298
+ size_t last_executed = 0;
1299
+ size_t last_sender_resets = 0;
1300
+
1301
while(!netdata_exit) {
1302
1303
// statistics
@@ -1082,16 +1305,26 @@ void *replication_thread_main(void *ptr __maybe_unused) {
1305
if(unlikely(now_mono_ut - last_now_mono_ut > default_rrd_update_every * USEC_PER_SEC)) {
1306
last_now_mono_ut = now_mono_ut;
1307
1308
+ replication_recursive_lock();
1309
+
1310
+ size_t current_executed = __atomic_load_n(&replication_globals.atomic.executed, __ATOMIC_RELAXED);
1311
+ if(last_executed != current_executed) {
1312
+ run_verification_countdown = ITERATIONS_IDLE_WITHOUT_PENDING_TO_RUN_SENDER_VERIFICATION;
1313
+ last_executed = current_executed;
1314
+ slow = false;
1315
+ }
1316
+
1317
if(replication_reset_next_point_in_time_countdown-- == 0) {
1318
// once per second, make it scan all the pending requests next time
1319
replication_set_next_point_in_time(0, 0);
1320
+// replication_globals.protected.skipped_no_room_since_last_reset = 0;
1321
replication_reset_next_point_in_time_countdown = SECONDS_TO_RESET_POINT_IN_TIME;
1322
}
1323
1091
- if(!replication_globals.pending && run_verification_countdown-- == 0) {
1324
+ if(!replication_globals.unsafe.pending && --run_verification_countdown == 0) {
1325
// reset the statistics about completion percentage
1093
- replication_globals.first_time_t = 0;
1094
- latest_first_time_t = 0;
1326
+ replication_globals.unsafe.first_time_t = 0;
1327
+ replication_set_latest_first_time(0);
1328
1329
verify_all_hosts_charts_are_streaming_now();
1330
@@ -1101,33 +1334,31 @@ void *replication_thread_main(void *ptr __maybe_unused) {
1334
1335
worker_is_busy(WORKER_JOB_STATISTICS);
1336
1104
- if(latest_first_time_t && replication_globals.pending) {
1337
+ time_t latest_first_time_t = replication_get_latest_first_time();
1338
+ if(latest_first_time_t && replication_globals.unsafe.pending) {
1339
// completion percentage statistics
1340
time_t now = now_realtime_sec();
1107
- time_t total = now - replication_globals.first_time_t;
1108
- time_t done = latest_first_time_t - replication_globals.first_time_t;
1341
+ time_t total = now - replication_globals.unsafe.first_time_t;
1342
+ time_t done = latest_first_time_t - replication_globals.unsafe.first_time_t;
1343
worker_set_metric(WORKER_JOB_CUSTOM_METRIC_COMPLETION,
1344
(NETDATA_DOUBLE) done * 100.0 / (NETDATA_DOUBLE) total);
1345
}
1346
else
1347
worker_set_metric(WORKER_JOB_CUSTOM_METRIC_COMPLETION, 100.0);
1348
1115
- worker_set_metric(WORKER_JOB_CUSTOM_METRIC_PENDING_REQUESTS, (NETDATA_DOUBLE)replication_globals.pending);
1116
- worker_set_metric(WORKER_JOB_CUSTOM_METRIC_ADDED, (NETDATA_DOUBLE)replication_globals.added);
1117
- worker_set_metric(WORKER_JOB_CUSTOM_METRIC_DONE, (NETDATA_DOUBLE)replication_globals.executed);
1118
- worker_set_metric(WORKER_JOB_CUSTOM_METRIC_SKIPPED_NOT_CONNECTED, (NETDATA_DOUBLE)replication_globals.skipped_not_connected);
1119
- worker_set_metric(WORKER_JOB_CUSTOM_METRIC_SKIPPED_NO_ROOM, (NETDATA_DOUBLE)replication_globals.skipped_no_room);
1120
- worker_set_metric(WORKER_JOB_CUSTOM_METRIC_SENDER_RESETS, (NETDATA_DOUBLE)replication_globals.sender_resets);
1121
- worker_set_metric(WORKER_JOB_CUSTOM_METRIC_WAITS, (NETDATA_DOUBLE)replication_globals.waits);
1122
- }
1349
+ worker_set_metric(WORKER_JOB_CUSTOM_METRIC_PENDING_REQUESTS, (NETDATA_DOUBLE)replication_globals.unsafe.pending);
1350
+ worker_set_metric(WORKER_JOB_CUSTOM_METRIC_ADDED, (NETDATA_DOUBLE)replication_globals.unsafe.added);
1351
+ worker_set_metric(WORKER_JOB_CUSTOM_METRIC_DONE, (NETDATA_DOUBLE)__atomic_load_n(&replication_globals.atomic.executed, __ATOMIC_RELAXED));
1352
+ worker_set_metric(WORKER_JOB_CUSTOM_METRIC_SKIPPED_NOT_CONNECTED, (NETDATA_DOUBLE)replication_globals.unsafe.skipped_not_connected);
1353
+ worker_set_metric(WORKER_JOB_CUSTOM_METRIC_SKIPPED_NO_ROOM, (NETDATA_DOUBLE)replication_globals.unsafe.skipped_no_room);
1354
+ worker_set_metric(WORKER_JOB_CUSTOM_METRIC_SENDER_RESETS, (NETDATA_DOUBLE)replication_globals.unsafe.sender_resets);
1355
+ worker_set_metric(WORKER_JOB_CUSTOM_METRIC_WAITS, (NETDATA_DOUBLE)replication_globals.main_thread.waits);
1356
1124
- worker_is_busy(WORKER_JOB_FIND_NEXT);
1125
- struct replication_request rq = replication_request_get_first_available();
1357
+ replication_recursive_unlock();
1358
+ }
1359
1127
- if(unlikely(!rq.found)) {
1128
- // make it scan all the pending requests next time
1129
- replication_set_next_point_in_time(0, 0);
1130
- replication_reset_next_point_in_time_countdown = SECONDS_TO_RESET_POINT_IN_TIME;
1360
+ if(unlikely(replication_execute_next_pending_request() == REQUEST_QUEUE_EMPTY)) {
1361
+ replication_recursive_lock();
1362
1363
// the timeout also defines now frequently we will traverse all the pending requests
1364
// when the outbound buffers of all senders is full
@@ -1136,84 +1367,39 @@ void *replication_thread_main(void *ptr __maybe_unused) {
1367
// no work to be done, wait for a request to come in
1368
timeout = 1000 * USEC_PER_MS;
1369
1139
- else if(replication_globals.pending > 0)
1140
- // there are pending requests waiting to be executed,
1141
- // but none could be executed at this time.
1142
- // try again after this time.
1143
- timeout = 100 * USEC_PER_MS;
1370
+ else if(replication_globals.unsafe.pending > 0) {
1371
+ if(replication_globals.unsafe.sender_resets == last_sender_resets) {
1372
+ timeout = 1000 * USEC_PER_MS;
1373
+ }
1374
+ else {
1375
+ // there are pending requests waiting to be executed,
1376
+ // but none could be executed at this time.
1377
+ // try again after this time.
1378
+ timeout = 100 * USEC_PER_MS;
1379
+ }
1380
1145
- else
1146
- // no pending requests, but there were requests recently (run_verification_countdown)
1381
+ last_sender_resets = replication_globals.unsafe.sender_resets;
1382
+ }
1383
+ else {
1384
+ // no requests pending, but there were requests recently (run_verification_countdown)
1385
// so, try in a short time.
1386
// if this is big, one chart replicating will be slow to finish (ping - pong just one chart)
1387
timeout = 10 * USEC_PER_MS;
1388
+ last_sender_resets = replication_globals.unsafe.sender_resets;
1389
+ }
1390
+
1391
+ replication_globals.main_thread.waits++;
1392
+ replication_recursive_unlock();
1393
1151
- replication_globals.waits++;
1394
worker_is_idle();
1395
sleep_usec(timeout);
1154
- continue;
1155
- }
1156
-
1157
- run_verification_countdown = ITERATIONS_IDLE_WITHOUT_PENDING_TO_RUN_SENDER_VERIFICATION;
1158
- slow = false;
1159
-
1160
- // delete the request from the dictionary
1161
- worker_is_busy(WORKER_JOB_DELETE_ENTRY);
1162
- if(!dictionary_del(rq.sender->replication_requests, string2str(rq.chart_id)))
1163
- error("REPLAY ERROR: 'host:%s/chart:%s' failed to be deleted from sender pending charts index",
1164
- rrdhost_hostname(rq.sender->host), string2str(rq.chart_id));
1396
1166
- worker_is_busy(WORKER_JOB_FIND_CHART);
1167
- RRDSET *st = rrdset_find(rq.sender->host, string2str(rq.chart_id));
1168
- if(!st) {
1169
- internal_error(true, "REPLAY ERROR: 'host:%s/chart:%s' not found",
1170
- rrdhost_hostname(rq.sender->host), string2str(rq.chart_id));
1397
+ // make it scan all the pending requests next time
1398
+ replication_set_next_point_in_time(0, 0);
1399
+ replication_reset_next_point_in_time_countdown = SECONDS_TO_RESET_POINT_IN_TIME;
1400
1401
continue;
1402
}
1174
-
1175
- worker_is_busy(WORKER_JOB_QUERYING);
1176
-
1177
- latest_first_time_t = rq.after;
1178
-
1179
- if(rq.after < rq.sender->replication_first_time || !rq.sender->replication_first_time)
1180
- rq.sender->replication_first_time = rq.after;
1181
-
1182
- if(rq.before < rq.sender->replication_current_time || !rq.sender->replication_current_time)
1183
- rq.sender->replication_current_time = rq.before;
1184
-
1185
- netdata_thread_disable_cancelability();
1186
-
1187
- // send the replication data
1188
- bool start_streaming = replicate_chart_response(
1189
- st->rrdhost, st, rq.start_streaming, rq.after, rq.before);
1190
-
1191
- netdata_thread_enable_cancelability();
1192
-
1193
- replication_globals.executed++;
1194
-
1195
- if(start_streaming && rq.sender_last_flush_ut == rrdpush_sender_get_flush_time(rq.sender)) {
1196
- worker_is_busy(WORKER_JOB_ACTIVATE_ENABLE_STREAMING);
1197
-
1198
- // enable normal streaming if we have to
1199
- // but only if the sender buffer has not been flushed since we started
1200
-
1201
- if(rrdset_flag_check(st, RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS)) {
1202
- rrdset_flag_clear(st, RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS);
1203
- rrdset_flag_set(st, RRDSET_FLAG_SENDER_REPLICATION_FINISHED);
1204
- rrdhost_sender_replicating_charts_minus_one(st->rrdhost);
1205
-
1206
-#ifdef NETDATA_LOG_REPLICATION_REQUESTS
1207
- internal_error(true, "STREAM_SENDER REPLAY: 'host:%s/chart:%s' streaming starts",
1208
- rrdhost_hostname(st->rrdhost), rrdset_id(st));
1209
-#endif
1210
- }
1211
- else
1212
- internal_error(true, "REPLAY ERROR: 'host:%s/chart:%s' received start streaming command, but the chart is not in progress replicating",
1213
- rrdhost_hostname(st->rrdhost), string2str(rq.chart_id));
1214
- }
1215
-
1216
- string_freez(rq.chart_id);
1403
}
1404
1405
netdata_thread_cleanup_pop(1);
streaming/replication.h
+1
@@ -6,6 +6,7 @@
6
#include "daemon/common.h"
7
8
struct replication_query_statistics {
9
+ SPINLOCK spinlock;
10
size_t queries_started;
11
size_t queries_finished;
12
size_t points_read;
streaming/rrdpush.h
+27
-18
@@ -162,32 +162,41 @@ struct sender_state {
162
struct netdata_ssl ssl; // structure used to encrypt the connection
163
#endif
164
165
- DICTIONARY *replication_requests; // de-duplication of replication requests, per chart
165
+ struct {
166
+ DICTIONARY *requests; // de-duplication of replication requests, per chart
167
167
- size_t replication_pending_requests; // the currently outstanding replication requests
168
- size_t replication_charts_replicating; // the number of unique charts having pending replication requests (on every request one is added and is removed when we finish it - it does not track completion of the replication for this chart)
168
+ struct {
169
+ size_t pending_requests; // the currently outstanding replication requests
170
+ size_t charts_replicating; // the number of unique charts having pending replication requests (on every request one is added and is removed when we finish it - it does not track completion of the replication for this chart)
171
+ } atomic;
172
170
- time_t replication_first_time; // the oldest time that has been requested to be replicated
171
- time_t replication_current_time; // the minimum(before) of the executed replication requests
173
+ struct {
174
+ bool reached_max; // used to avoid resetting the replication thread too frequently
175
+ } unsafe; // protected by sender mutex
176
173
- bool replication_reached_max; // used to avoid resetting the replication thread too frequently
177
+ } replication;
178
175
- size_t buffer_used_percentage; // the current utilization of the sending buffer
176
- usec_t last_flush_time_ut; // the last time the sender flushed the sending buffer in USEC
179
+ struct {
180
+ size_t buffer_used_percentage; // the current utilization of the sending buffer
181
+ usec_t last_flush_time_ut; // the last time the sender flushed the sending buffer in USEC
182
+ } atomic;
183
};
184
179
-#define rrdpush_sender_set_flush_time(sender) __atomic_store_n(&((sender)->last_flush_time_ut), now_realtime_usec(), __ATOMIC_RELAXED);
180
-#define rrdpush_sender_get_flush_time(sender) __atomic_load_n(&((sender)->last_flush_time_ut), __ATOMIC_RELAXED)
185
+#define rrdpush_sender_set_buffer_used_percent(sender, value) __atomic_store_n(&((sender)->atomic.buffer_used_percentage), value, __ATOMIC_RELAXED);
186
+#define rrdpush_sender_get_buffer_used_percent(sender) __atomic_load_n(&((sender)->atomic.buffer_used_percentage), __ATOMIC_RELAXED)
187
182
-#define rrdpush_sender_replicating_charts(sender) __atomic_load_n(&((sender)->replication_charts_replicating), __ATOMIC_RELAXED)
183
-#define rrdpush_sender_replicating_charts_plus_one(sender) __atomic_add_fetch(&((sender)->replication_charts_replicating), 1, __ATOMIC_RELAXED)
184
-#define rrdpush_sender_replicating_charts_minus_one(sender) __atomic_sub_fetch(&((sender)->replication_charts_replicating), 1, __ATOMIC_RELAXED)
185
-#define rrdpush_sender_replicating_charts_zero(sender) __atomic_store_n(&((sender)->replication_charts_replicating), 0, __ATOMIC_RELAXED)
188
+#define rrdpush_sender_set_flush_time(sender) __atomic_store_n(&((sender)->atomic.last_flush_time_ut), now_realtime_usec(), __ATOMIC_RELAXED);
189
+#define rrdpush_sender_get_flush_time(sender) __atomic_load_n(&((sender)->atomic.last_flush_time_ut), __ATOMIC_RELAXED)
190
187
-#define rrdpush_sender_pending_replication_requests(sender) __atomic_load_n(&((sender)->replication_pending_requests), __ATOMIC_RELAXED)
188
-#define rrdpush_sender_pending_replication_requests_plus_one(sender) __atomic_add_fetch(&((sender)->replication_pending_requests), 1, __ATOMIC_RELAXED)
189
-#define rrdpush_sender_pending_replication_requests_minus_one(sender) __atomic_sub_fetch(&((sender)->replication_pending_requests), 1, __ATOMIC_RELAXED)
190
-#define rrdpush_sender_pending_replication_requests_zero(sender) __atomic_store_n(&((sender)->replication_pending_requests), 0, __ATOMIC_RELAXED)
191
+#define rrdpush_sender_replicating_charts(sender) __atomic_load_n(&((sender)->replication.atomic.charts_replicating), __ATOMIC_RELAXED)
192
+#define rrdpush_sender_replicating_charts_plus_one(sender) __atomic_add_fetch(&((sender)->replication.atomic.charts_replicating), 1, __ATOMIC_RELAXED)
193
+#define rrdpush_sender_replicating_charts_minus_one(sender) __atomic_sub_fetch(&((sender)->replication.atomic.charts_replicating), 1, __ATOMIC_RELAXED)
194
+#define rrdpush_sender_replicating_charts_zero(sender) __atomic_store_n(&((sender)->replication.atomic.charts_replicating), 0, __ATOMIC_RELAXED)
195
+
196
+#define rrdpush_sender_pending_replication_requests(sender) __atomic_load_n(&((sender)->replication.atomic.pending_requests), __ATOMIC_RELAXED)
197
+#define rrdpush_sender_pending_replication_requests_plus_one(sender) __atomic_add_fetch(&((sender)->replication.atomic.pending_requests), 1, __ATOMIC_RELAXED)
198
+#define rrdpush_sender_pending_replication_requests_minus_one(sender) __atomic_sub_fetch(&((sender)->replication.atomic.pending_requests), 1, __ATOMIC_RELAXED)
199
+#define rrdpush_sender_pending_replication_requests_zero(sender) __atomic_store_n(&((sender)->replication.atomic.pending_requests), 0, __ATOMIC_RELAXED)
200
201
struct receiver_state {
202
RRDHOST *host;
streaming/sender.c
+1
-1
@@ -1345,7 +1345,7 @@ void *rrdpush_sender_thread(void *ptr) {
1345
rrdpush_sender_thread_close_socket(s->host);
1346
}
1347
1348
- worker_set_metric(WORKER_SENDER_JOB_REPLAY_DICT_SIZE, (NETDATA_DOUBLE) dictionary_entries(s->replication_requests));
1348
+ worker_set_metric(WORKER_SENDER_JOB_REPLAY_DICT_SIZE, (NETDATA_DOUBLE) dictionary_entries(s->replication.requests));
1349
}
1350
1351
netdata_thread_cleanup_pop(1);