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

3 files changed, 28 insertions, 3 deletions

diff --git a/database/rrd.h b/database/rrd.h
index b548aa7169..dda3e5f9a2 100644
--- a/database/rrd.h
+++ b/database/rrd.h
@@ -1265,8 +1265,10 @@ void rrdset_isnot_obsolete(RRDSET *st);
 #define rrdset_is_archived(st) (rrdset_flag_check(st, RRDSET_FLAG_ARCHIVED) && rrdset_number_of_dimensions(st))
 
 time_t rrddim_first_entry_t(RRDDIM *rd);
+time_t rrddim_first_entry_t_of_tier(RRDDIM *rd, size_t tier);
 time_t rrddim_last_entry_t(RRDDIM *rd);
 time_t rrdset_last_entry_t(RRDSET *st);
+time_t rrdset_first_entry_t_of_tier(RRDSET *st, size_t tier);
 time_t rrdset_first_entry_t(RRDSET *st);
 time_t rrdhost_last_entry_t(RRDHOST *h);
 
diff --git a/database/rrddim.c b/database/rrddim.c
index 06bfef7d72..1b3d9952c9 100644
--- a/database/rrddim.c
+++ b/database/rrddim.c
@@ -435,13 +435,18 @@ time_t rrddim_last_entry_t(RRDDIM *rd) {
     return latest;
 }
 
+time_t rrddim_first_entry_t_of_tier(RRDDIM *rd, size_t tier) {
+    if(unlikely(tier > storage_tiers || !rd->tiers[tier]))
+        return 0;
+
+    return rd->tiers[tier]->query_ops->oldest_time(rd->tiers[tier]->db_metric_handle);
+}
+
 time_t rrddim_first_entry_t(RRDDIM *rd) {
     time_t oldest = 0;
 
     for(size_t tier = 0; tier < storage_tiers ;tier++) {
-        if(unlikely(!rd->tiers[tier])) continue;
-
-        time_t t = rd->tiers[tier]->query_ops->oldest_time(rd->tiers[tier]->db_metric_handle);
+        time_t t = rrddim_first_entry_t_of_tier(rd, tier);
         if(t != 0 && (oldest == 0 || t < oldest))
             oldest = t;
     }
diff --git a/database/rrdset.c b/database/rrdset.c
index 0e64ff7635..d69c830bb1 100644
--- a/database/rrdset.c
+++ b/database/rrdset.c
@@ -549,6 +549,24 @@ time_t rrdset_first_entry_t(RRDSET *st) {
     return first_entry_t;
 }
 
+time_t rrdset_first_entry_t_of_tier(RRDSET *st, size_t tier) {
+    if(unlikely(tier > storage_tiers))
+        return 0;
+
+    RRDDIM *rd;
+    time_t first_entry_t = LONG_MAX;
+
+    rrddim_foreach_read(rd, st) {
+        time_t t = rrddim_first_entry_t_of_tier(rd, tier);
+        if(t && t < first_entry_t)
+            first_entry_t = t;
+    }
+    rrddim_foreach_done(rd);
+
+    if (unlikely(LONG_MAX == first_entry_t)) return 0;
+    return first_entry_t;
+}
+
 inline void rrdset_is_obsolete(RRDSET *st) {
     if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_ARCHIVED))) {
         info("Cannot obsolete already archived chart %s", rrdset_name(st));