Rrddim acquire on replay set (#13932)

* prevent RRDDIM from vanishing while replay is working with it

* set chart last access time

* set chart last access time everytime someone finds it

* do not replay dimensions that are archived

* remove the obsolete flag from dimensions that replayed; do not process archived dimensions

* cleanup db_metric_handle refcount of hidden dimensions

* more information in page alignment fatal

* do not fatal() of page alignment reset when the caller is the only writer

1 files changed, 38 insertions, 22 deletions

diff --git a/database/rrddim.c b/database/rrddim.c
index e304d1eba2..157c8ebd33 100644
--- a/database/rrddim.c
+++ b/database/rrddim.c
@@ -180,24 +180,21 @@ static void rrddim_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, v
 
     debug(D_RRD_CALLS, "rrddim_free() %s.%s", rrdset_name(st), rrddim_name(rd));
 
-    if (!rrddim_flag_check(rd, RRDDIM_FLAG_ARCHIVED)) {
-
-        size_t tiers_available = 0, tiers_said_yes = 0;
-        for(size_t tier = 0; tier < storage_tiers ;tier++) {
-            if(rd->tiers[tier]) {
-                tiers_available++;
+    size_t tiers_available = 0, tiers_said_yes = 0;
+    for(size_t tier = 0; tier < storage_tiers ;tier++) {
+        if(rd->tiers[tier] && rd->tiers[tier]->db_collection_handle) {
+            tiers_available++;
 
-                if(rd->tiers[tier]->collect_ops->finalize(rd->tiers[tier]->db_collection_handle))
-                    tiers_said_yes++;
+            if(rd->tiers[tier]->collect_ops->finalize(rd->tiers[tier]->db_collection_handle))
+                tiers_said_yes++;
 
-                rd->tiers[tier]->db_collection_handle = NULL;
-            }
+            rd->tiers[tier]->db_collection_handle = NULL;
         }
+    }
 
-        if (tiers_available == tiers_said_yes && tiers_said_yes && rd->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE) {
-            /* This metric has no data and no references */
-            metaqueue_delete_dimension_uuid(&rd->metric_uuid);
-        }
+    if (tiers_available == tiers_said_yes && tiers_said_yes && rd->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE) {
+        /* This metric has no data and no references */
+        metaqueue_delete_dimension_uuid(&rd->metric_uuid);
     }
 
     rrddimvar_delete_all(rd);
@@ -246,16 +243,14 @@ static bool rrddim_conflict_callback(const DICTIONARY_ITEM *item __maybe_unused,
     rc += rrddim_set_multiplier(st, rd, ctr->multiplier);
     rc += rrddim_set_divisor(st, rd, ctr->divisor);
 
-    if(rrddim_flag_check(rd, RRDDIM_FLAG_ARCHIVED)) {
-
-        for(size_t tier = 0; tier < storage_tiers ;tier++) {
-            if (rd->tiers[tier])
-                rd->tiers[tier]->db_collection_handle =
-                    rd->tiers[tier]->collect_ops->init(rd->tiers[tier]->db_metric_handle, st->rrdhost->db[tier].tier_grouping * st->update_every);
-        }
+    for(size_t tier = 0; tier < storage_tiers ;tier++) {
+        if (rd->tiers[tier] && !rd->tiers[tier]->db_collection_handle)
+            rd->tiers[tier]->db_collection_handle =
+                rd->tiers[tier]->collect_ops->init(rd->tiers[tier]->db_metric_handle, st->rrdhost->db[tier].tier_grouping * st->update_every);
+    }
 
+    if(rrddim_flag_check(rd, RRDDIM_FLAG_ARCHIVED)) {
         rrddim_flag_clear(rd, RRDDIM_FLAG_ARCHIVED);
-
         if(!rrdset_is_ar_chart(st)) {
             rrddim_flag_set(rd, RRDDIM_FLAG_PENDING_HEALTH_INITIALIZATION);
             rrdset_flag_set(rd->rrdset, RRDSET_FLAG_PENDING_HEALTH_INITIALIZATION);
@@ -318,6 +313,27 @@ inline RRDDIM *rrddim_find(RRDSET *st, const char *id) {
     return rrddim_index_find(st, id);
 }
 
+inline RRDDIM_ACQUIRED *rrddim_find_and_acquire(RRDSET *st, const char *id) {
+    debug(D_RRD_CALLS, "rrddim_find() for chart %s, dimension %s", rrdset_name(st), id);
+
+    return (RRDDIM_ACQUIRED *)dictionary_get_and_acquire_item(st->rrddim_root_index, id);
+}
+
+RRDDIM *rrddim_acquired_to_rrddim(RRDDIM_ACQUIRED *rda) {
+    if(unlikely(!rda))
+        return NULL;
+
+    return (RRDDIM *) dictionary_acquired_item_value((const DICTIONARY_ITEM *)rda);
+}
+
+void rrddim_acquired_release(RRDDIM_ACQUIRED *rda) {
+    if(unlikely(!rda))
+        return;
+
+    RRDDIM *rd = rrddim_acquired_to_rrddim(rda);
+    dictionary_acquired_item_release(rd->rrdset->rrddim_root_index, (const DICTIONARY_ITEM *)rda);
+}
+
 // This will not return dimensions that are archived
 RRDDIM *rrddim_find_active(RRDSET *st, const char *id) {
     RRDDIM *rd = rrddim_find(st, id);