use 2 levels of judy arrays to speed up replication on very busy parents (#14031)
* use 2 levels of judy arrays to speed up replication on very busy parents * delete requests from judy when executed * do not process requests when the sender is not connected; not all requests removed are executed, so count the executed accurately * flush replication data on sender disconnect * cache used buffer ratio in sender structure * cleanup replication requests when they are not valid any more * properly update inner and outer judy arrays on deletions * detailed replication stats * fix bug in dictionary where deletion and flushes lead to crashes * replication should only report retention of tier 0 * replication now has 2 buffer limits 10 -> 30 % * detailed statistics about resets and skipped requests * register worker metrics * added counter for waits * make new requests discoverable * make it continue on iterations properly
Costa Tsaousis committed
Nov 22, 2022 at 02:08 UTC
4a7048fc1ffc5d5d9a1160debbfb191a4aee99b2
9 files changed
+293
-184
database/rrd.h
+2
@@ -1265,8 +1265,10 @@ void rrdset_isnot_obsolete(RRDSET *st);
1265
#define rrdset_is_archived(st) (rrdset_flag_check(st, RRDSET_FLAG_ARCHIVED) && rrdset_number_of_dimensions(st))
1266
1267
time_t rrddim_first_entry_t(RRDDIM *rd);
1268
+time_t rrddim_first_entry_t_of_tier(RRDDIM *rd, size_t tier);
1269
time_t rrddim_last_entry_t(RRDDIM *rd);
1270
time_t rrdset_last_entry_t(RRDSET *st);
1271
+time_t rrdset_first_entry_t_of_tier(RRDSET *st, size_t tier);
1272
time_t rrdset_first_entry_t(RRDSET *st);
1273
time_t rrdhost_last_entry_t(RRDHOST *h);
1274
database/rrddim.c
+8
-3
@@ -435,13 +435,18 @@ time_t rrddim_last_entry_t(RRDDIM *rd) {
435
return latest;
436
}
437
438
+time_t rrddim_first_entry_t_of_tier(RRDDIM *rd, size_t tier) {
439
+ if(unlikely(tier > storage_tiers || !rd->tiers[tier]))
440
+ return 0;
441
+
442
+ return rd->tiers[tier]->query_ops->oldest_time(rd->tiers[tier]->db_metric_handle);
443
+}
444
+
445
time_t rrddim_first_entry_t(RRDDIM *rd) {
446
time_t oldest = 0;
447
448
for(size_t tier = 0; tier < storage_tiers ;tier++) {
442
- if(unlikely(!rd->tiers[tier])) continue;
443
-
444
- time_t t = rd->tiers[tier]->query_ops->oldest_time(rd->tiers[tier]->db_metric_handle);
449
+ time_t t = rrddim_first_entry_t_of_tier(rd, tier);
450
if(t != 0 && (oldest == 0 || t < oldest))
451
oldest = t;
452
}
database/rrdset.c
+18
@@ -549,6 +549,24 @@ time_t rrdset_first_entry_t(RRDSET *st) {
549
return first_entry_t;
550
}
551
552
+time_t rrdset_first_entry_t_of_tier(RRDSET *st, size_t tier) {
553
+ if(unlikely(tier > storage_tiers))
554
+ return 0;
555
+
556
+ RRDDIM *rd;
557
+ time_t first_entry_t = LONG_MAX;
558
+
559
+ rrddim_foreach_read(rd, st) {
560
+ time_t t = rrddim_first_entry_t_of_tier(rd, tier);
561
+ if(t && t < first_entry_t)
562
+ first_entry_t = t;
563
+ }
564
+ rrddim_foreach_done(rd);
565
+
566
+ if (unlikely(LONG_MAX == first_entry_t)) return 0;
567
+ return first_entry_t;
568
+}
569
+
570
inline void rrdset_is_obsolete(RRDSET *st) {
571
if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_ARCHIVED))) {
572
info("Cannot obsolete already archived chart %s", rrdset_name(st));
libnetdata/dictionary/dictionary.c
+14
-6
@@ -1075,6 +1075,13 @@ static inline const char *item_get_name(const DICTIONARY_ITEM *item) {
1075
return item->caller_name;
1076
}
1077
1078
+static inline size_t item_get_name_len(const DICTIONARY_ITEM *item) {
1079
+ if(item->options & ITEM_OPTION_ALLOCATED_NAME)
1080
+ return string_strlen(item->string_name);
1081
+ else
1082
+ return strlen(item->caller_name);
1083
+}
1084
+
1085
static DICTIONARY_ITEM *dict_item_create(DICTIONARY *dict __maybe_unused, size_t *allocated_bytes, DICTIONARY_ITEM *master_item) {
1086
DICTIONARY_ITEM *item;
1087
@@ -1794,10 +1801,10 @@ void dictionary_flush(DICTIONARY *dict) {
1801
if(unlikely(!dict))
1802
return;
1803
1797
- // delete the index
1798
- dictionary_index_lock_wrlock(dict);
1799
- hashtable_destroy_unsafe(dict);
1800
- dictionary_index_lock_unlock(dict);
1804
+// // delete the index
1805
+// dictionary_index_lock_wrlock(dict);
1806
+// hashtable_destroy_unsafe(dict);
1807
+// dictionary_index_lock_unlock(dict);
1808
1809
// delete all items
1810
ll_recursive_lock(dict, DICTIONARY_LOCK_WRITE); // get write lock here, to speed it up (it is recursive)
@@ -1805,8 +1812,9 @@ void dictionary_flush(DICTIONARY *dict) {
1812
for (item = dict->items.list; item; item = item_next) {
1813
item_next = item->next;
1814
1808
- if(!item_flag_check(item, ITEM_FLAG_DELETED))
1809
- dict_item_free_or_mark_deleted(dict, item);
1815
+// if(!item_flag_check(item, ITEM_FLAG_DELETED))
1816
+// dict_item_free_or_mark_deleted(dict, item);
1817
+ dict_item_del(dict, item_get_name(item), (ssize_t)item_get_name_len(item));
1818
}
1819
ll_recursive_unlock(dict, DICTIONARY_LOCK_WRITE);
1820
streaming/replication.c
+241
-174
@@ -3,6 +3,9 @@
3
#include "replication.h"
4
#include "Judy.h"
5
6
+#define MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED 30
7
+#define MIN_SENDER_BUFFER_PERCENTAGE_ALLOWED 10
8
+
9
static time_t replicate_chart_timeframe(BUFFER *wb, RRDSET *st, time_t after, time_t before, bool enable_streaming) {
10
size_t dimensions = rrdset_number_of_dimensions(st);
11
@@ -394,17 +397,6 @@ bool replicate_chart_request(send_command callback, void *callback_data, RRDHOST
397
return send_replay_chart_cmd(callback, callback_data, st, start_streaming, first_entry_wanted, last_entry_wanted);
398
}
399
397
-// ----------------------------------------------------------------------------
398
-
399
-static size_t sender_buffer_used_percent(struct sender_state *s) {
400
- netdata_mutex_lock(&s->mutex);
401
- size_t available = cbuffer_available_size_unsafe(s->host->sender->buffer);
402
- netdata_mutex_unlock(&s->mutex);
403
-
404
- return (s->host->sender->buffer->max_size - available) * 100 / s->host->sender->buffer->max_size;
405
-}
406
-
407
-
400
// ----------------------------------------------------------------------------
401
// replication thread
402
@@ -416,18 +408,18 @@ struct replication_request {
408
STRING *chart_id;
409
time_t after; // key for sorting (JudyL)
410
time_t before;
411
+ Word_t unique_id;
412
bool start_streaming;
413
bool found;
414
+ bool index_in_judy;
415
};
416
417
// replication sort entry in JudyL array
418
// used for sorting all requests, across all nodes
419
struct replication_sort_entry {
426
- struct replication_request req;
420
+ struct replication_request *rq;
421
428
- const void *unique_id; // used as a key to identify the sort entry - we never access its contents
429
- bool executed;
430
- struct replication_sort_entry *next;
422
+ size_t unique_id; // used as a key to identify the sort entry - we never access its contents
423
};
424
425
// the global variables for the replication thread
@@ -435,18 +427,33 @@ static struct replication_thread {
427
netdata_mutex_t mutex;
428
429
size_t added;
430
+ size_t executed;
431
size_t removed;
432
time_t first_time_t;
433
size_t requests_count;
434
+ Word_t next_unique_id;
435
struct replication_request *requests;
436
437
+ Word_t last_after;
438
+ Word_t last_unique_id;
439
+
440
+ size_t skipped_not_connected;
441
+ size_t skipped_no_room;
442
+ size_t sender_resets;
443
+ size_t waits;
444
+
445
Pvoid_t JudyL_array;
446
} rep = {
447
.mutex = NETDATA_MUTEX_INITIALIZER,
448
.added = 0,
447
- .removed = 0,
449
+ .executed = 0,
450
.first_time_t = 0,
451
.requests_count = 0,
452
+ .next_unique_id = 1,
453
+ .skipped_no_room = 0,
454
+ .skipped_not_connected = 0,
455
+ .sender_resets = 0,
456
+ .waits = 0,
457
.requests = NULL,
458
.JudyL_array = NULL,
459
};
@@ -476,159 +483,172 @@ static void replication_recursive_unlock() {
483
// ----------------------------------------------------------------------------
484
// replication sort entry management
485
479
-static struct replication_sort_entry *replication_sort_entry_create(struct replication_request *r, const void *unique_id) {
480
- struct replication_sort_entry *t = mallocz(sizeof(struct replication_sort_entry));
486
+static struct replication_sort_entry *replication_sort_entry_create(struct replication_request *rq) {
487
+ struct replication_sort_entry *rse = mallocz(sizeof(struct replication_sort_entry));
488
489
// copy the request
483
- t->req = *r;
484
- t->req.chart_id = string_dup(r->chart_id);
485
-
490
+ rse->rq = rq;
491
+ rse->unique_id = rep.next_unique_id++;
492
487
- t->unique_id = unique_id;
488
- t->executed = false;
489
- t->next = NULL;
490
- return t;
493
+ // save the unique id into the request, to be able to delete it later
494
+ rq->unique_id = rse->unique_id;
495
+ rq->index_in_judy = false;
496
+ return rse;
497
}
498
493
-static void replication_sort_entry_destroy(struct replication_sort_entry *t) {
494
- string_freez(t->req.chart_id);
495
- freez(t);
499
+static void replication_sort_entry_destroy(struct replication_sort_entry *rse) {
500
+ freez(rse);
501
}
502
498
-static struct replication_sort_entry *replication_sort_entry_add(struct replication_request *r, const void *unique_id) {
499
- struct replication_sort_entry *t = replication_sort_entry_create(r, unique_id);
500
-
503
+static struct replication_sort_entry *replication_sort_entry_add(struct replication_request *rq) {
504
replication_recursive_lock();
505
506
+ struct replication_sort_entry *rse = replication_sort_entry_create(rq);
507
+
508
+ if(rq->after < (time_t)rep.last_after) {
509
+ // make it find this request first
510
+ rep.last_after = rq->after;
511
+ rep.last_unique_id = rq->unique_id;
512
+ }
513
+
514
rep.added++;
515
+ rep.requests_count++;
516
505
- Pvoid_t *PValue;
517
+ Pvoid_t *inner_judy_ptr;
518
507
- PValue = JudyLGet(rep.JudyL_array, (Word_t) r->after, PJE0);
508
- if(!PValue)
509
- PValue = JudyLIns(&rep.JudyL_array, (Word_t) r->after, PJE0);
519
+ // find the outer judy entry, using after as key
520
+ inner_judy_ptr = JudyLGet(rep.JudyL_array, (Word_t) rq->after, PJE0);
521
+ if(!inner_judy_ptr)
522
+ inner_judy_ptr = JudyLIns(&rep.JudyL_array, (Word_t) rq->after, PJE0);
523
511
- t->next = *PValue;
512
- *PValue = t;
524
+ // add it to the inner judy, using unique_id as key
525
+ Pvoid_t *item = JudyLIns(inner_judy_ptr, rq->unique_id, PJE0);
526
+ *item = rse;
527
+ rq->index_in_judy = true;
528
514
- if(!rep.first_time_t || r->after < rep.first_time_t)
515
- rep.first_time_t = r->after;
529
+ if(!rep.first_time_t || rq->after < rep.first_time_t)
530
+ rep.first_time_t = rq->after;
531
532
replication_recursive_unlock();
533
519
- return t;
534
+ return rse;
535
}
536
522
-static void replication_sort_entry_del(struct sender_state *sender, STRING *chart_id, time_t after, const DICTIONARY_ITEM *item) {
523
- Pvoid_t *PValue;
524
- struct replication_sort_entry *to_delete = NULL;
525
-
526
- replication_recursive_lock();
537
+static bool replication_sort_entry_unlink_and_free_unsafe(struct replication_sort_entry *rse, Pvoid_t **inner_judy_ppptr) {
538
+ bool inner_judy_deleted = false;
539
540
rep.removed++;
541
+ rep.requests_count--;
542
530
- PValue = JudyLGet(rep.JudyL_array, after, PJE0);
531
- if(PValue) {
532
- struct replication_sort_entry *t = *PValue;
533
- t->executed = true; // make sure we don't get it again
543
+ rse->rq->index_in_judy = false;
544
535
- if(!t->next) {
536
- // we are alone here, delete the judy entry
545
+ // delete it from the inner judy
546
+ JudyLDel(*inner_judy_ppptr, rse->rq->unique_id, PJE0);
547
538
- if(t->unique_id != item)
539
- fatal("Item to delete is not matching host '%s', chart '%s', time %ld.",
540
- rrdhost_hostname(sender->host), string2str(chart_id), after);
548
+ // if no items left, delete it from the outer judy
549
+ if(**inner_judy_ppptr == NULL) {
550
+ JudyLDel(&rep.JudyL_array, rse->rq->after, PJE0);
551
+ inner_judy_deleted = true;
552
+ }
553
542
- to_delete = t;
543
- JudyLDel(&rep.JudyL_array, after, PJE0);
544
- }
545
- else {
546
- // find our entry in the linked list
554
+ // free memory
555
+ replication_sort_entry_destroy(rse);
556
548
- struct replication_sort_entry *t_old = NULL;
549
- do {
550
- if(t->unique_id == item) {
551
- to_delete = t;
557
+ return inner_judy_deleted;
558
+}
559
553
- if(t_old)
554
- t_old->next = t->next;
555
- else
556
- *PValue = t->next;
560
+static void replication_sort_entry_del(struct replication_request *rq) {
561
+ Pvoid_t *inner_judy_pptr;
562
+ struct replication_sort_entry *rse_to_delete = NULL;
563
558
- break;
559
- }
564
+ replication_recursive_lock();
565
+ if(rq->index_in_judy) {
566
+
567
+ inner_judy_pptr = JudyLGet(rep.JudyL_array, rq->after, PJE0);
568
+ if (inner_judy_pptr) {
569
+ Pvoid_t *our_item_pptr = JudyLGet(*inner_judy_pptr, rq->unique_id, PJE0);
570
+ if (our_item_pptr) {
571
+ rse_to_delete = *our_item_pptr;
572
+ replication_sort_entry_unlink_and_free_unsafe(rse_to_delete, &inner_judy_pptr);
573
+ }
574
+ }
575
561
- t_old = t;
562
- t = t->next;
576
+ if (!rse_to_delete)
577
+ fatal("Cannot find sort entry to delete for host '%s', chart '%s', time %ld.",
578
+ rrdhost_hostname(rq->sender->host), string2str(rq->chart_id), rq->after);
579
564
- } while(t);
565
- }
580
}
581
568
- if(!to_delete)
569
- fatal("Cannot find sort entry to delete for host '%s', chart '%s', time %ld.",
570
- rrdhost_hostname(sender->host), string2str(chart_id), after);
571
-
582
replication_recursive_unlock();
573
-
574
- replication_sort_entry_destroy(to_delete);
583
}
584
585
static struct replication_request replication_request_get_first_available() {
578
- struct replication_sort_entry *found = NULL;
579
- Pvoid_t *PValue;
580
- Word_t Index;
586
+ Pvoid_t *inner_judy_pptr;
587
588
replication_recursive_lock();
589
584
- rep.requests_count = JudyLCount(rep.JudyL_array, 0, 0xFFFFFFFF, PJE0);
585
- if(!rep.requests_count) {
586
- replication_recursive_unlock();
587
- return (struct replication_request){ .found = false };
590
+ struct replication_request rq = (struct replication_request){ .found = false };
591
+
592
+
593
+ if(rep.last_after && rep.last_unique_id) {
594
+ rep.last_after--;
595
+ rep.last_unique_id--;
596
+ }
597
+ else {
598
+ rep.last_after = 0;
599
+ rep.last_unique_id = 0;
600
}
601
590
- Index = 0;
591
- PValue = JudyLFirst(rep.JudyL_array, &Index, PJE0);
592
- while(!found && PValue) {
593
- struct replication_sort_entry *t;
594
-
595
- for(t = *PValue; t ;t = t->next) {
596
- if(!t->executed
597
- && sender_buffer_used_percent(t->req.sender) <= 10
598
- && t->req.sender_last_flush_ut == __atomic_load_n(&t->req.sender->last_flush_time_ut, __ATOMIC_SEQ_CST)
599
- ) {
600
- found = t;
601
- found->executed = true;
602
- break;
602
+ while(!rq.found && (inner_judy_pptr = JudyLNext(rep.JudyL_array, &rep.last_after, PJE0))) {
603
+ Pvoid_t *our_item_pptr;
604
+
605
+ while(!rq.found && (our_item_pptr = JudyLNext(*inner_judy_pptr, &rep.last_unique_id, PJE0))) {
606
+ struct replication_sort_entry *rse = *our_item_pptr;
607
+ struct sender_state *s = rse->rq->sender;
608
+
609
+ bool sender_is_connected =
610
+ rrdhost_flag_check(s->host, RRDHOST_FLAG_RRDPUSH_SENDER_CONNECTED);
611
+
612
+ bool sender_has_been_flushed_since_this_request =
613
+ rse->rq->sender_last_flush_ut != __atomic_load_n(&s->last_flush_time_ut, __ATOMIC_SEQ_CST);
614
+
615
+ bool sender_has_room_to_spare =
616
+ s->replication_sender_buffer_percent_used <= MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED;
617
+
618
+ if(unlikely(!sender_is_connected || sender_has_been_flushed_since_this_request)) {
619
+ rep.skipped_not_connected++;
620
+ if(replication_sort_entry_unlink_and_free_unsafe(rse, &inner_judy_pptr))
621
+ break;
622
}
604
- }
623
606
- if(!found)
607
- PValue = JudyLNext(rep.JudyL_array, &Index, PJE0);
608
- }
624
+ else if(sender_has_room_to_spare) {
625
+ // copy the request to return it
626
+ rq = *rse->rq;
627
+
628
+ // set the return result to found
629
+ rq.found = true;
630
610
- // copy the values we need, while we have the lock
611
- struct replication_request ret;
631
+ if(replication_sort_entry_unlink_and_free_unsafe(rse, &inner_judy_pptr))
632
+ break;
633
+ }
634
+ else
635
+ rep.skipped_no_room++;
636
+ }
637
613
- if(found) {
614
- ret = found->req;
615
- ret.chart_id = string_dup(ret.chart_id);
616
- ret.found = true;
638
+ // prepare for the next iteration on the outer loop
639
+ rep.last_unique_id = 0;
640
}
618
- else
619
- ret.found = false;
641
642
replication_recursive_unlock();
622
-
623
- return ret;
643
+ return rq;
644
}
645
646
// ----------------------------------------------------------------------------
647
// replication request management
648
629
-static void replication_request_react_callback(const DICTIONARY_ITEM *item, void *value __maybe_unused, void *sender_state __maybe_unused) {
649
+static void replication_request_react_callback(const DICTIONARY_ITEM *item __maybe_unused, void *value __maybe_unused, void *sender_state __maybe_unused) {
650
struct sender_state *s = sender_state; (void)s;
631
- struct replication_request *r = value;
651
+ struct replication_request *rq = value;
652
653
// IMPORTANT:
654
// We use the react instead of the insert callback
@@ -640,34 +660,35 @@ static void replication_request_react_callback(const DICTIONARY_ITEM *item, void
660
// may see the replication sort entry, but fail to find the dictionary item
661
// related to it.
662
643
- replication_sort_entry_add(r, item);
644
- __atomic_fetch_add(&r->sender->replication_pending_requests, 1, __ATOMIC_SEQ_CST);
663
+ replication_sort_entry_add(rq);
664
+ __atomic_fetch_add(&rq->sender->replication_pending_requests, 1, __ATOMIC_SEQ_CST);
665
}
666
667
static bool replication_request_conflict_callback(const DICTIONARY_ITEM *item __maybe_unused, void *old_value, void *new_value, void *sender_state) {
668
struct sender_state *s = sender_state; (void)s;
649
- struct replication_request *r = old_value; (void)r;
650
- struct replication_request *r_new = new_value;
669
+ struct replication_request *rq = old_value; (void)rq;
670
+ struct replication_request *rq_new = new_value;
671
672
internal_error(
673
true,
674
"STREAM %s [send to %s]: ignoring duplicate replication command received for chart '%s' (existing from %llu to %llu [%s], new from %llu to %llu [%s])",
675
rrdhost_hostname(s->host), s->connected_to, dictionary_acquired_item_name(item),
656
- (unsigned long long)r->after, (unsigned long long)r->before, r->start_streaming ? "true" : "false",
657
- (unsigned long long)r_new->after, (unsigned long long)r_new->before, r_new->start_streaming ? "true" : "false");
676
+ (unsigned long long)rq->after, (unsigned long long)rq->before, rq->start_streaming ? "true" : "false",
677
+ (unsigned long long)rq_new->after, (unsigned long long)rq_new->before, rq_new->start_streaming ? "true" : "false");
678
659
- string_freez(r_new->chart_id);
679
+ string_freez(rq_new->chart_id);
680
681
return false;
682
}
683
664
-static void replication_request_delete_callback(const DICTIONARY_ITEM *item, void *value, void *sender_state __maybe_unused) {
665
- struct replication_request *r = value;
684
+static void replication_request_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, void *value, void *sender_state __maybe_unused) {
685
+ struct replication_request *rq = value;
686
667
- replication_sort_entry_del(r->sender, r->chart_id, r->after, item);
687
+ if(rq->index_in_judy)
688
+ replication_sort_entry_del(rq);
689
669
- string_freez(r->chart_id);
670
- __atomic_fetch_sub(&r->sender->replication_pending_requests, 1, __ATOMIC_SEQ_CST);
690
+ string_freez(rq->chart_id);
691
+ __atomic_fetch_sub(&rq->sender->replication_pending_requests, 1, __ATOMIC_SEQ_CST);
692
}
693
694
@@ -675,7 +696,7 @@ static void replication_request_delete_callback(const DICTIONARY_ITEM *item, voi
696
// public API
697
698
void replication_add_request(struct sender_state *sender, const char *chart_id, time_t after, time_t before, bool start_streaming) {
678
- struct replication_request tmp = {
699
+ struct replication_request rq = {
700
.sender = sender,
701
.chart_id = string_strdupz(chart_id),
702
.after = after,
@@ -684,7 +705,7 @@ void replication_add_request(struct sender_state *sender, const char *chart_id,
705
.sender_last_flush_ut = __atomic_load_n(&sender->last_flush_time_ut, __ATOMIC_SEQ_CST),
706
};
707
687
- dictionary_set(sender->replication_requests, chart_id, &tmp, sizeof(struct replication_request));
708
+ dictionary_set(sender->replication_requests, chart_id, &rq, sizeof(struct replication_request));
709
}
710
711
void replication_flush_sender(struct sender_state *sender) {
@@ -708,6 +729,26 @@ void replication_cleanup_sender(struct sender_state *sender) {
729
replication_recursive_unlock();
730
}
731
732
+void replication_recalculate_buffer_used_ratio_unsafe(struct sender_state *s) {
733
+ size_t available = cbuffer_available_size_unsafe(s->host->sender->buffer);
734
+ size_t percentage = (s->buffer->max_size - available) * 100 / s->buffer->max_size;
735
+
736
+ if(percentage > MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED)
737
+ s->replication_reached_max = true;
738
+
739
+ if(s->replication_reached_max &&
740
+ percentage <= MIN_SENDER_BUFFER_PERCENTAGE_ALLOWED) {
741
+ s->replication_reached_max = false;
742
+ replication_recursive_lock();
743
+ rep.last_after = 0;
744
+ rep.last_unique_id = 0;
745
+ rep.sender_resets++;
746
+ replication_recursive_unlock();
747
+ }
748
+
749
+ s->replication_sender_buffer_percent_used = percentage;
750
+}
751
+
752
// ----------------------------------------------------------------------------
753
// replication thread
754
@@ -721,99 +762,125 @@ static void replication_main_cleanup(void *ptr) {
762
static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;
763
}
764
724
-#define WORKER_JOB_ITERATION 1
725
-#define WORKER_JOB_REPLAYING 2
726
-#define WORKER_JOB_CUSTOM_METRIC_PENDING_REQUESTS 3
727
-#define WORKER_JOB_CUSTOM_METRIC_COMPLETION 4
728
-#define WORKER_JOB_CUSTOM_METRIC_ADDED 5
729
-#define WORKER_JOB_CUSTOM_METRIC_DONE 6
765
+#define WORKER_JOB_FIND_NEXT 1
766
+#define WORKER_JOB_QUERYING 2
767
+#define WORKER_JOB_DELETE_ENTRY 3
768
+#define WORKER_JOB_FIND_CHART 4
769
+#define WORKER_JOB_STATISTICS 5
770
+#define WORKER_JOB_ACTIVATE_ENABLE_STREAMING 6
771
+#define WORKER_JOB_CUSTOM_METRIC_PENDING_REQUESTS 7
772
+#define WORKER_JOB_CUSTOM_METRIC_COMPLETION 8
773
+#define WORKER_JOB_CUSTOM_METRIC_ADDED 9
774
+#define WORKER_JOB_CUSTOM_METRIC_DONE 10
775
+#define WORKER_JOB_CUSTOM_METRIC_SKIPPED_NOT_CONNECTED 11
776
+#define WORKER_JOB_CUSTOM_METRIC_SKIPPED_NO_ROOM 12
777
+#define WORKER_JOB_CUSTOM_METRIC_SENDER_RESETS 13
778
+#define WORKER_JOB_CUSTOM_METRIC_WAITS 14
779
780
void *replication_thread_main(void *ptr __maybe_unused) {
781
netdata_thread_cleanup_push(replication_main_cleanup, ptr);
782
783
worker_register("REPLICATION");
784
736
- worker_register_job_name(WORKER_JOB_ITERATION, "iteration");
737
- worker_register_job_name(WORKER_JOB_REPLAYING, "replaying");
785
+ worker_register_job_name(WORKER_JOB_FIND_NEXT, "find next");
786
+ worker_register_job_name(WORKER_JOB_QUERYING, "querying");
787
+ worker_register_job_name(WORKER_JOB_DELETE_ENTRY, "dict delete");
788
+ worker_register_job_name(WORKER_JOB_FIND_CHART, "find chart");
789
+ worker_register_job_name(WORKER_JOB_ACTIVATE_ENABLE_STREAMING, "enable streaming");
790
791
worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_PENDING_REQUESTS, "pending requests", "requests", WORKER_METRIC_ABSOLUTE);
792
worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_COMPLETION, "completion", "%", WORKER_METRIC_ABSOLUTE);
793
worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_ADDED, "added requests", "requests/s", WORKER_METRIC_INCREMENTAL_TOTAL);
794
worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_DONE, "finished requests", "requests/s", WORKER_METRIC_INCREMENTAL_TOTAL);
795
+ worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_SKIPPED_NOT_CONNECTED, "not connected requests", "requests/s", WORKER_METRIC_INCREMENTAL_TOTAL);
796
+ worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_SKIPPED_NO_ROOM, "no room requests", "requests/s", WORKER_METRIC_INCREMENTAL_TOTAL);
797
+ worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_SENDER_RESETS, "sender resets", "resets/s", WORKER_METRIC_INCREMENTAL_TOTAL);
798
+ worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_WAITS, "waits", "waits/s", WORKER_METRIC_INCREMENTAL_TOTAL);
799
744
- while(!netdata_exit) {
745
- worker_is_busy(WORKER_JOB_ITERATION);
746
-
747
- // this call also updates our statistics
748
- struct replication_request r = replication_request_get_first_available();
800
+ time_t latest_first_time_t = 0;
801
750
- if(r.found) {
751
- // delete the request from the dictionary
752
- dictionary_del(r.sender->replication_requests, string2str(r.chart_id));
753
- }
802
+ while(!netdata_exit) {
803
+ worker_is_busy(WORKER_JOB_FIND_NEXT);
804
+ struct replication_request rq = replication_request_get_first_available();
805
806
+ worker_is_busy(WORKER_JOB_STATISTICS);
807
worker_set_metric(WORKER_JOB_CUSTOM_METRIC_PENDING_REQUESTS, (NETDATA_DOUBLE)rep.requests_count);
808
worker_set_metric(WORKER_JOB_CUSTOM_METRIC_ADDED, (NETDATA_DOUBLE)rep.added);
757
- worker_set_metric(WORKER_JOB_CUSTOM_METRIC_DONE, (NETDATA_DOUBLE)rep.removed);
809
+ worker_set_metric(WORKER_JOB_CUSTOM_METRIC_DONE, (NETDATA_DOUBLE)rep.executed);
810
+ worker_set_metric(WORKER_JOB_CUSTOM_METRIC_SKIPPED_NOT_CONNECTED, (NETDATA_DOUBLE)rep.skipped_not_connected);
811
+ worker_set_metric(WORKER_JOB_CUSTOM_METRIC_SKIPPED_NO_ROOM, (NETDATA_DOUBLE)rep.skipped_no_room);
812
+ worker_set_metric(WORKER_JOB_CUSTOM_METRIC_SENDER_RESETS, (NETDATA_DOUBLE)rep.sender_resets);
813
+ worker_set_metric(WORKER_JOB_CUSTOM_METRIC_WAITS, (NETDATA_DOUBLE)rep.waits);
814
759
- if(!r.found && !rep.requests_count) {
760
- worker_set_metric(WORKER_JOB_CUSTOM_METRIC_COMPLETION, 100.0);
761
- worker_is_idle();
762
- sleep_usec(1000 * USEC_PER_MS);
763
- continue;
815
+ if(latest_first_time_t) {
816
+ time_t now = now_realtime_sec();
817
+ time_t total = now - rep.first_time_t;
818
+ time_t done = latest_first_time_t - rep.first_time_t;
819
+ worker_set_metric(WORKER_JOB_CUSTOM_METRIC_COMPLETION, (NETDATA_DOUBLE)done * 100.0 / (NETDATA_DOUBLE)total);
820
}
821
766
- if(!r.found) {
822
+ if(!rq.found) {
823
worker_is_idle();
768
- sleep_usec(1 * USEC_PER_MS);
824
+
825
+ if(!rep.requests_count)
826
+ worker_set_metric(WORKER_JOB_CUSTOM_METRIC_COMPLETION, 100.0);
827
+
828
+ // make it start from the beginning
829
+ rep.last_after = 0;
830
+ rep.last_unique_id = 0;
831
+
832
+ rep.waits++;
833
+
834
+ sleep_usec(1000 * USEC_PER_MS);
835
continue;
836
}
837
+ else {
838
+ // delete the request from the dictionary
839
+ worker_is_busy(WORKER_JOB_DELETE_ENTRY);
840
+ dictionary_del(rq.sender->replication_requests, string2str(rq.chart_id));
841
+ }
842
772
- RRDSET *st = rrdset_find(r.sender->host, string2str(r.chart_id));
843
+ worker_is_busy(WORKER_JOB_FIND_CHART);
844
+ RRDSET *st = rrdset_find(rq.sender->host, string2str(rq.chart_id));
845
if(!st) {
846
internal_error(true, "REPLAY: chart '%s' not found on host '%s'",
775
- string2str(r.chart_id), rrdhost_hostname(r.sender->host));
847
+ string2str(rq.chart_id), rrdhost_hostname(rq.sender->host));
848
849
continue;
850
}
851
780
- worker_is_busy(WORKER_JOB_REPLAYING);
852
+ worker_is_busy(WORKER_JOB_QUERYING);
853
782
- time_t latest_first_time_t = r.after;
854
+ latest_first_time_t = rq.after;
855
784
- if(r.after < r.sender->replication_first_time || !r.sender->replication_first_time)
785
- r.sender->replication_first_time = r.after;
856
+ if(rq.after < rq.sender->replication_first_time || !rq.sender->replication_first_time)
857
+ rq.sender->replication_first_time = rq.after;
858
787
- if(r.before < r.sender->replication_min_time || !r.sender->replication_min_time)
788
- r.sender->replication_min_time = r.before;
859
+ if(rq.before < rq.sender->replication_min_time || !rq.sender->replication_min_time)
860
+ rq.sender->replication_min_time = rq.before;
861
862
netdata_thread_disable_cancelability();
863
864
// send the replication data
865
bool start_streaming = replicate_chart_response(st->rrdhost, st,
794
- r.start_streaming, r.after, r.before);
866
+ rq.start_streaming, rq.after, rq.before);
867
868
netdata_thread_enable_cancelability();
869
798
- if(start_streaming && r.sender_last_flush_ut == __atomic_load_n(&r.sender->last_flush_time_ut, __ATOMIC_SEQ_CST)) {
799
- __atomic_fetch_add(&r.sender->receiving_metrics, 1, __ATOMIC_SEQ_CST);
870
+ rep.executed++;
871
+
872
+ if(start_streaming && rq.sender_last_flush_ut == __atomic_load_n(&rq.sender->last_flush_time_ut, __ATOMIC_SEQ_CST)) {
873
+ worker_is_busy(WORKER_JOB_ACTIVATE_ENABLE_STREAMING);
874
+ __atomic_fetch_add(&rq.sender->receiving_metrics, 1, __ATOMIC_SEQ_CST);
875
876
// enable normal streaming if we have to
877
// but only if the sender buffer has not been flushed since we started
878
879
debug(D_REPLICATION, "Enabling metric streaming for chart %s.%s",
805
- rrdhost_hostname(r.sender->host), rrdset_id(st));
880
+ rrdhost_hostname(rq.sender->host), rrdset_id(st));
881
882
rrdset_flag_set(st, RRDSET_FLAG_SENDER_REPLICATION_FINISHED);
883
}
809
-
810
- // statistics
811
- {
812
- time_t now = now_realtime_sec();
813
- time_t total = now - rep.first_time_t;
814
- time_t done = latest_first_time_t - rep.first_time_t;
815
- worker_set_metric(WORKER_JOB_CUSTOM_METRIC_COMPLETION, (NETDATA_DOUBLE)done * 100.0 / (NETDATA_DOUBLE)total);
816
- }
884
}
885
886
netdata_thread_cleanup_pop(1);
streaming/replication.h
+1
@@ -18,5 +18,6 @@ void replication_init_sender(struct sender_state *sender);
18
void replication_cleanup_sender(struct sender_state *sender);
19
void replication_flush_sender(struct sender_state *sender);
20
void replication_add_request(struct sender_state *sender, const char *chart_id, time_t after, time_t before, bool start_streaming);
21
+void replication_recalculate_buffer_used_ratio_unsafe(struct sender_state *s);
22
23
#endif /* REPLICATION_H */
streaming/rrdpush.c
+1
-1
@@ -293,7 +293,7 @@ static inline void rrdpush_send_chart_definition(BUFFER *wb, RRDSET *st) {
293
rrdsetvar_print_to_streaming_custom_chart_variables(st, wb);
294
295
if (stream_has_capability(host->sender, STREAM_CAP_REPLICATION)) {
296
- time_t first_entry_local = rrdset_first_entry_t(st);
296
+ time_t first_entry_local = rrdset_first_entry_t_of_tier(st, 0);
297
time_t last_entry_local = st->last_updated.tv_sec;
298
299
if(!last_entry_local) {
streaming/rrdpush.h
+2
@@ -166,6 +166,8 @@ struct sender_state {
166
size_t replication_pending_requests;
167
time_t replication_first_time;
168
time_t replication_min_time;
169
+ size_t replication_sender_buffer_percent_used;
170
+ bool replication_reached_max;
171
172
usec_t last_flush_time_ut;
173
size_t receiving_metrics;
streaming/sender.c
+6
@@ -163,6 +163,8 @@ void sender_commit(struct sender_state *s, BUFFER *wb) {
163
s->flags |= SENDER_FLAG_OVERFLOW;
164
#endif
165
166
+ replication_recalculate_buffer_used_ratio_unsafe(s);
167
+
168
netdata_mutex_unlock(&s->mutex);
169
rrdpush_signal_sender_to_wake_up(s);
170
}
@@ -176,6 +178,8 @@ static inline void rrdpush_sender_thread_close_socket(RRDHOST *host) {
178
179
rrdhost_flag_clear(host, RRDHOST_FLAG_RRDPUSH_SENDER_READY_4_METRICS);
180
rrdhost_flag_clear(host, RRDHOST_FLAG_RRDPUSH_SENDER_CONNECTED);
181
+
182
+ replication_flush_sender(host->sender);
183
}
184
185
static inline void rrdpush_sender_add_host_variable_to_buffer(BUFFER *wb, const RRDVAR_ACQUIRED *rva) {
@@ -263,6 +267,7 @@ static inline void rrdpush_sender_thread_data_flush(RRDHOST *host) {
267
268
netdata_mutex_lock(&host->sender->mutex);
269
cbuffer_flush(host->sender->buffer);
270
+ replication_recalculate_buffer_used_ratio_unsafe(host->sender);
271
netdata_mutex_unlock(&host->sender->mutex);
272
273
rrdpush_sender_thread_reset_all_charts(host);
@@ -770,6 +775,7 @@ static ssize_t attempt_to_send(struct sender_state *s) {
775
else
776
debug(D_STREAM, "STREAM: send() returned 0 -> no error but no transmission");
777
778
+ replication_recalculate_buffer_used_ratio_unsafe(s);
779
netdata_mutex_unlock(&s->mutex);
780
781
return ret;