Optimizations part 3 (#15293)

* use madvise to speed up indexing

* collect all rrddim members into a collector structure

* use tier 0 virtual point for storing last stored value

* reorganize key fields in rrddim

* remove fgets from pluginsd and replace it with read()

* properly uncork the web server sockets

* Revert "reorganize key fields in rrddim"

This reverts commit 2d45fa3959087e05462d387ff115a260f3a04b60.

* Revert "use tier 0 virtual point for storing last stored value"

This reverts commit a576cdd377ad4778a3b8608cabbb7ea7bb19a3a8.

* fix cork names

* fix compilation warnings

7 files changed, 138 insertions, 134 deletions

diff --git a/database/contexts/metric.c b/database/contexts/metric.c
index 80756b54c2..69839021a8 100644
--- a/database/contexts/metric.c
+++ b/database/contexts/metric.c
@@ -33,7 +33,7 @@ inline NETDATA_DOUBLE rrdmetric_acquired_last_stored_value(RRDMETRIC_ACQUIRED *r
     RRDMETRIC *rm = rrdmetric_acquired_value(rma);
 
     if(rm->rrddim)
-        return rm->rrddim->last_stored_value;
+        return rm->rrddim->collector.last_stored_value;
 
     return NAN;
 }
diff --git a/database/engine/journalfile.c b/database/engine/journalfile.c
index 234505cb19..9832b5784c 100644
--- a/database/engine/journalfile.c
+++ b/database/engine/journalfile.c
@@ -216,15 +216,20 @@ static struct journal_v2_header *journalfile_v2_mounted_data_get(struct rrdengin
             madvise_dontfork(journalfile->mmap.data, journalfile->mmap.size);
             madvise_dontdump(journalfile->mmap.data, journalfile->mmap.size);
 
-            // let the kernel know that we don't want read-ahead on this file
-            madvise_random(journalfile->mmap.data, journalfile->mmap.size);
-
-//            madvise_willneed(journalfile->mmap.data, journalfile->v2.size_of_directory);
-//            madvise_dontneed(journalfile->mmap.data, journalfile->mmap.size);
-
             spinlock_lock(&journalfile->v2.spinlock);
             journalfile->v2.flags |= JOURNALFILE_FLAG_IS_AVAILABLE | JOURNALFILE_FLAG_IS_MOUNTED;
+            JOURNALFILE_FLAGS flags = journalfile->v2.flags;
             spinlock_unlock(&journalfile->v2.spinlock);
+
+            if(flags & JOURNALFILE_FLAG_MOUNTED_FOR_RETENTION) {
+                // we need the entire metrics directory into memory to process it
+                madvise_willneed(journalfile->mmap.data, journalfile->v2.size_of_directory);
+            }
+            else {
+                // let the kernel know that we don't want read-ahead on this file
+                madvise_random(journalfile->mmap.data, journalfile->mmap.size);
+                // madvise_dontneed(journalfile->mmap.data, journalfile->mmap.size);
+            }
         }
     }
 
@@ -413,7 +418,7 @@ void journalfile_v2_data_set(struct rrdengine_journalfile *journalfile, int fd,
     struct journal_v2_header *j2_header = journalfile->mmap.data;
     journalfile->v2.first_time_s = (time_t)(j2_header->start_time_ut / USEC_PER_SEC);
     journalfile->v2.last_time_s = (time_t)(j2_header->end_time_ut / USEC_PER_SEC);
-    // journalfile->v2.size_of_directory = j2_header->metric_offset + j2_header->metric_count * sizeof(struct journal_metric_list);
+    journalfile->v2.size_of_directory = j2_header->metric_offset + j2_header->metric_count * sizeof(struct journal_metric_list);
 
     journalfile_v2_mounted_data_unmount(journalfile, true, true);
 
diff --git a/database/engine/journalfile.h b/database/engine/journalfile.h
index 62e8f55ee7..5cdf72b9d0 100644
--- a/database/engine/journalfile.h
+++ b/database/engine/journalfile.h
@@ -40,7 +40,7 @@ struct rrdengine_journalfile {
         time_t first_time_s;
         time_t last_time_s;
         time_t not_needed_since_s;
-        // uint32_t size_of_directory;
+        uint32_t size_of_directory;
     } v2;
 
     struct {
diff --git a/database/rrd.h b/database/rrd.h
index c41f45bbe2..b6fde67e77 100644
--- a/database/rrd.h
+++ b/database/rrd.h
@@ -241,9 +241,9 @@ typedef enum __attribute__ ((__packed__)) rrddim_options {
     // this is 8-bit
 } RRDDIM_OPTIONS;
 
-#define rrddim_option_check(rd, option) ((rd)->options & (option))
-#define rrddim_option_set(rd, option)   (rd)->options |= (option)
-#define rrddim_option_clear(rd, option) (rd)->options &= ~(option)
+#define rrddim_option_check(rd, option) ((rd)->collector.options & (option))
+#define rrddim_option_set(rd, option)   (rd)->collector.options |= (option)
+#define rrddim_option_clear(rd, option) (rd)->collector.options &= ~(option)
 
 // flags are runtime changing status flags (atomics are required to alter/access them)
 typedef enum __attribute__ ((__packed__)) rrddim_flags {
@@ -352,42 +352,19 @@ struct rrddim {
     STRING *name;                                   // the name of this dimension (as presented to user)
 
     RRD_ALGORITHM algorithm;                        // the algorithm that is applied to add new collected values
-    RRDDIM_OPTIONS options;                         // permanent configuration options
     RRD_MEMORY_MODE rrd_memory_mode;                // the memory mode for this dimension
     RRDDIM_FLAGS flags;                             // run time changing status flags
 
     int32_t multiplier;                             // the multiplier of the collected values
     int32_t divisor;                                // the divider of the collected values
 
-    uint32_t collections_counter;                   // the number of times we added values to this rrddim
-
     // ------------------------------------------------------------------------
     // operational state members
 
-    rrd_ml_dimension_t *ml_dimension;                   // machine learning data about this dimension
-
-    // ------------------------------------------------------------------------
-    // linking to siblings and parents
-
     struct rrdset *rrdset;
+    rrd_ml_dimension_t *ml_dimension;                   // machine learning data about this dimension
     RRDMETRIC_ACQUIRED *rrdmetric;                  // the rrdmetric of this dimension
 
-    // ------------------------------------------------------------------------
-    // data collection members
-
-    struct timeval last_collected_time;             // when was this dimension last updated
-                                                    // this is actual date time we updated the last_collected_value
-                                                    // THIS IS DIFFERENT FROM THE SAME MEMBER OF RRDSET
-
-    collected_number collected_value_max;           // the absolute maximum of the collected value
-
-    NETDATA_DOUBLE calculated_value;                // the current calculated value, after applying the algorithm - resets to zero after being used
-    NETDATA_DOUBLE last_calculated_value;           // the last calculated value processed
-    NETDATA_DOUBLE last_stored_value;               // the last value as stored in the database (after interpolation)
-
-    collected_number collected_value;               // the current value, as collected - resets to 0 after being used
-    collected_number last_collected_value;          // the last value that was collected, after being processed
-
 #ifdef NETDATA_LOG_COLLECTION_ERRORS
     usec_t rrddim_store_metric_last_ut;             // the timestamp we last called rrddim_store_metric()
     size_t rrddim_store_metric_count;               // the rrddim_store_metric() counter
@@ -406,6 +383,28 @@ struct rrddim {
     } db;
 
     // ------------------------------------------------------------------------
+    // data collection members
+
+    struct {
+        RRDDIM_OPTIONS options;                         // permanent configuration options
+
+        uint32_t counter;                               // the number of times we added values to this rrddim
+
+        collected_number collected_value;               // the current value, as collected - resets to 0 after being used
+        collected_number collected_value_max;           // the absolute maximum of the collected value
+        collected_number last_collected_value;          // the last value that was collected, after being processed
+
+        struct timeval last_collected_time;             // when was this dimension last updated
+                                                        // this is actual date time we updated the last_collected_value
+                                                        // THIS IS DIFFERENT FROM THE SAME MEMBER OF RRDSET
+
+        NETDATA_DOUBLE calculated_value;                // the current calculated value, after applying the algorithm - resets to zero after being used
+        NETDATA_DOUBLE last_calculated_value;           // the last calculated value processed
+
+        NETDATA_DOUBLE last_stored_value;               // the last value as stored in the database (after interpolation)
+    } collector;
+
+    // ------------------------------------------------------------------------
 
     struct rrddim_tier tiers[];                     // our tiers of databases
 };
@@ -415,13 +414,13 @@ size_t rrddim_size(void);
 #define rrddim_id(rd) string2str((rd)->id)
 #define rrddim_name(rd) string2str((rd) ->name)
 
-#define rrddim_check_updated(rd) ((rd)->options & RRDDIM_OPTION_UPDATED)
-#define rrddim_set_updated(rd) (rd)->options |= RRDDIM_OPTION_UPDATED
-#define rrddim_clear_updated(rd) (rd)->options &= ~RRDDIM_OPTION_UPDATED
+#define rrddim_check_updated(rd) ((rd)->collector.options & RRDDIM_OPTION_UPDATED)
+#define rrddim_set_updated(rd) (rd)->collector.options |= RRDDIM_OPTION_UPDATED
+#define rrddim_clear_updated(rd) (rd)->collector.options &= ~RRDDIM_OPTION_UPDATED
 
-#define rrddim_check_exposed(rd) ((rd)->options & RRDDIM_OPTION_EXPOSED)
-#define rrddim_set_exposed(rd) (rd)->options |= RRDDIM_OPTION_EXPOSED
-#define rrddim_clear_exposed(rd) (rd)->options &= ~RRDDIM_OPTION_EXPOSED
+#define rrddim_check_exposed(rd) ((rd)->collector.options & RRDDIM_OPTION_EXPOSED)
+#define rrddim_set_exposed(rd) (rd)->collector.options |= RRDDIM_OPTION_EXPOSED
+#define rrddim_clear_exposed(rd) (rd)->collector.options &= ~RRDDIM_OPTION_EXPOSED
 
 // returns the RRDDIM cache filename, or NULL if it does not exist
 const char *rrddim_cache_filename(RRDDIM *rd);
diff --git a/database/rrddim.c b/database/rrddim.c
index 73a5c8daa6..96560b1c1a 100644
--- a/database/rrddim.c
+++ b/database/rrddim.c
@@ -49,7 +49,7 @@ static void rrddim_insert_callback(const DICTIONARY_ITEM *item __maybe_unused, v
     rd->rrdset = st;
 
     if(rrdset_flag_check(st, RRDSET_FLAG_STORE_FIRST))
-        rd->collections_counter = 1;
+        rd->collector.counter = 1;
 
     if(ctr->memory_mode == RRD_MEMORY_MODE_MAP || ctr->memory_mode == RRD_MEMORY_MODE_SAVE) {
         if(!rrddim_memory_load_or_create_map_save(st, rd, ctr->memory_mode)) {
@@ -565,16 +565,16 @@ inline collected_number rrddim_set_by_pointer(RRDSET *st, RRDDIM *rd, collected_
 collected_number rrddim_timed_set_by_pointer(RRDSET *st __maybe_unused, RRDDIM *rd, struct timeval collected_time, collected_number value) {
     debug(D_RRD_CALLS, "rrddim_set_by_pointer() for chart %s, dimension %s, value " COLLECTED_NUMBER_FORMAT, rrdset_name(st), rrddim_name(rd), value);
 
-    rd->last_collected_time = collected_time;
-    rd->collected_value = value;
+    rd->collector.last_collected_time = collected_time;
+    rd->collector.collected_value = value;
     rrddim_set_updated(rd);
-    rd->collections_counter++;
+    rd->collector.counter++;
 
     collected_number v = (value >= 0) ? value : -value;
-    if (unlikely(v > rd->collected_value_max))
-        rd->collected_value_max = v;
+    if (unlikely(v > rd->collector.collected_value_max))
+        rd->collector.collected_value_max = v;
 
-    return rd->last_collected_value;
+    return rd->collector.last_collected_value;
 }
 
 
@@ -644,9 +644,9 @@ void rrddim_memory_file_update(RRDDIM *rd) {
     if(!rd || !rd->db.rd_on_file) return;
     struct rrddim_map_save_v019 *rd_on_file = rd->db.rd_on_file;
 
-    rd_on_file->last_collected_time.tv_sec = rd->last_collected_time.tv_sec;
-    rd_on_file->last_collected_time.tv_usec = rd->last_collected_time.tv_usec;
-    rd_on_file->last_collected_value = rd->last_collected_value;
+    rd_on_file->last_collected_time.tv_sec = rd->collector.last_collected_time.tv_sec;
+    rd_on_file->last_collected_time.tv_usec = rd->collector.last_collected_time.tv_usec;
+    rd_on_file->last_collected_value = rd->collector.last_collected_value;
 }
 
 void rrddim_memory_file_free(RRDDIM *rd) {
@@ -727,7 +727,7 @@ bool rrddim_memory_load_or_create_map_save(RRDSET *st, RRDDIM *rd, RRD_MEMORY_MO
     }
 
     if(!reset) {
-        rd->last_collected_value = rd_on_file->last_collected_value;
+        rd->collector.last_collected_value = rd_on_file->last_collected_value;
 
         if(rd_on_file->algorithm != rd->algorithm)
             netdata_log_info("File %s does not have the expected algorithm (expected %u '%s', found %u '%s'). Previous values may be wrong.",
diff --git a/database/rrdset.c b/database/rrdset.c
index f4de6e2847..3e1a1249a5 100644
--- a/database/rrdset.c
+++ b/database/rrdset.c
@@ -759,9 +759,9 @@ void rrdset_reset(RRDSET *st) {
 
     RRDDIM *rd;
     rrddim_foreach_read(rd, st) {
-        rd->last_collected_time.tv_sec = 0;
-        rd->last_collected_time.tv_usec = 0;
-        rd->collections_counter = 0;
+        rd->collector.last_collected_time.tv_sec = 0;
+        rd->collector.last_collected_time.tv_usec = 0;
+        rd->collector.counter = 0;
 
         if(!rrddim_flag_check(rd, RRDDIM_FLAG_ARCHIVED)) {
             for(size_t tier = 0; tier < storage_tiers ;tier++)
@@ -1361,7 +1361,7 @@ static inline size_t rrdset_done_interpolate(
             switch(rd->algorithm) {
                 case RRD_ALGORITHM_INCREMENTAL:
                     new_value = (NETDATA_DOUBLE)
-                            (      rd->calculated_value
+                            (      rd->collector.calculated_value
                                    * (NETDATA_DOUBLE)(next_store_ut - last_collect_ut)
                                    / (NETDATA_DOUBLE)(now_collect_ut - last_collect_ut)
                             );
@@ -1372,14 +1372,14 @@ static inline size_t rrdset_done_interpolate(
                                 " / (%llu - %llu)"
                               , rrddim_name(rd)
                               , new_value
-                              , rd->calculated_value
+                              , rd->collector.calculated_value
                               , next_store_ut, last_collect_ut
                               , now_collect_ut, last_collect_ut
                     );
 
-                    rd->calculated_value -= new_value;
-                    new_value += rd->last_calculated_value;
-                    rd->last_calculated_value = 0;
+                    rd->collector.calculated_value -= new_value;
+                    new_value += rd->collector.last_calculated_value;
+                    rd->collector.last_calculated_value = 0;
                     new_value /= (NETDATA_DOUBLE)st->update_every;
 
                     if(unlikely(next_store_ut - last_stored_ut < update_every_ut)) {
@@ -1402,18 +1402,18 @@ static inline size_t rrdset_done_interpolate(
                         // do not interpolate
                         // just show the calculated value
 
-                        new_value = rd->calculated_value;
+                        new_value = rd->collector.calculated_value;
                     }
                     else {
                         // we have missed an update
                         // interpolate in the middle values
 
                         new_value = (NETDATA_DOUBLE)
-                                (   (     (rd->calculated_value - rd->last_calculated_value)
+                                (   (     (rd->collector.calculated_value - rd->collector.last_calculated_value)
                                           * (NETDATA_DOUBLE)(next_store_ut - last_collect_ut)
                                           / (NETDATA_DOUBLE)(now_collect_ut - last_collect_ut)
                                     )
-                                    +  rd->last_calculated_value
+                                    +  rd->collector.last_calculated_value
                                 );
 
                         rrdset_debug(st, "%s: CALC2 DEF " NETDATA_DOUBLE_FORMAT " = ((("
@@ -1421,9 +1421,9 @@ static inline size_t rrdset_done_interpolate(
                                             " * %llu"
                                             " / %llu) + " NETDATA_DOUBLE_FORMAT, rrddim_name(rd)
                                   , new_value
-                                  , rd->calculated_value, rd->last_calculated_value
+                                  , rd->collector.calculated_value, rd->collector.last_calculated_value
                                   , (next_store_ut - first_ut)
-                                  , (now_collect_ut - first_ut), rd->last_calculated_value
+                                  , (now_collect_ut - first_ut), rd->collector.last_calculated_value
                         );
                     }
                     break;
@@ -1441,7 +1441,7 @@ static inline size_t rrdset_done_interpolate(
                 continue;
             }
 
-            if(likely(rrddim_check_updated(rd) && rd->collections_counter > 1 && iterations < gap_when_lost_iterations_above)) {
+            if(likely(rrddim_check_updated(rd) && rd->collector.counter > 1 && iterations < gap_when_lost_iterations_above)) {
                 uint32_t dim_storage_flags = storage_flags;
 
                 if (ml_dimension_is_anomalous(rd, current_time_s, new_value, true)) {
@@ -1453,7 +1453,7 @@ static inline size_t rrdset_done_interpolate(
                     rrddim_push_metrics_v2(rsb, rd, next_store_ut, new_value, dim_storage_flags);
 
                 rrddim_store_metric(rd, next_store_ut, new_value, dim_storage_flags);
-                rd->last_stored_value = new_value;
+                rd->collector.last_stored_value = new_value;
             }
             else {
                 (void) ml_dimension_is_anomalous(rd, current_time_s, 0, false);
@@ -1464,7 +1464,7 @@ static inline size_t rrdset_done_interpolate(
                     rrddim_push_metrics_v2(rsb, rd, next_store_ut, NAN, SN_FLAG_NONE);
 
                 rrddim_store_metric(rd, next_store_ut, NAN, SN_FLAG_NONE);
-                rd->last_stored_value = NAN;
+                rd->collector.last_stored_value = NAN;
             }
 
             stored_entries++;
@@ -1667,22 +1667,22 @@ void rrdset_timed_done(RRDSET *st, struct timeval now, bool pending_rrdset_next)
         if(likely(rrddim_check_updated(rd))) {
             // if the new is smaller than the old (an overflow, or reset), set the old equal to the new
             // to reset the calculation (it will give zero as the calculation for this second)
-            if(unlikely(rd->algorithm == RRD_ALGORITHM_PCENT_OVER_DIFF_TOTAL && rd->last_collected_value > rd->collected_value)) {
+            if(unlikely(rd->algorithm == RRD_ALGORITHM_PCENT_OVER_DIFF_TOTAL && rd->collector.last_collected_value > rd->collector.collected_value)) {
                 debug(D_RRD_STATS, "'%s' / '%s': RESET or OVERFLOW. Last collected value = " COLLECTED_NUMBER_FORMAT ", current = " COLLECTED_NUMBER_FORMAT
                 , rrdset_id(st)
                 , rrddim_name(rd)
-                , rd->last_collected_value
-                , rd->collected_value
+                , rd->collector.last_collected_value
+                , rd->collector.collected_value
                 );
 
                 if(!(rrddim_option_check(rd, RRDDIM_OPTION_DONT_DETECT_RESETS_OR_OVERFLOWS)))
                     has_reset_value = 1;
 
-                rd->last_collected_value = rd->collected_value;
+                rd->collector.last_collected_value = rd->collector.collected_value;
             }
 
-            last_collected_total += rd->last_collected_value;
-            collected_total += rd->collected_value;
+            last_collected_total += rd->collector.last_collected_value;
+            collected_total += rd->collector.collected_value;
 
             if(unlikely(rrddim_flag_check(rd, RRDDIM_FLAG_OBSOLETE))) {
                 error("Dimension %s in chart '%s' has the OBSOLETE flag set, but it is collected.", rrddim_name(rd), rrdset_id(st));
@@ -1706,7 +1706,7 @@ void rrdset_timed_done(RRDSET *st, struct timeval now, bool pending_rrdset_next)
         if(unlikely(!rd)) continue;
 
         if(unlikely(!rrddim_check_updated(rd))) {
-            rd->calculated_value = 0;
+            rd->collector.calculated_value = 0;
             continue;
         }
 
@@ -1716,25 +1716,25 @@ void rrdset_timed_done(RRDSET *st, struct timeval now, bool pending_rrdset_next)
                 " last_calculated_value = " NETDATA_DOUBLE_FORMAT
                 " calculated_value = " NETDATA_DOUBLE_FORMAT
                      , rrddim_name(rd)
-                     , rd->last_collected_value
-                     , rd->collected_value
-                     , rd->last_calculated_value
-                     , rd->calculated_value
+                     , rd->collector.last_collected_value
+                     , rd->collector.collected_value
+                     , rd->collector.last_calculated_value
+                     , rd->collector.calculated_value
         );
 
         switch(rd->algorithm) {
             case RRD_ALGORITHM_ABSOLUTE:
-                rd->calculated_value = (NETDATA_DOUBLE)rd->collected_value
-                                       * (NETDATA_DOUBLE)rd->multiplier
-                                       / (NETDATA_DOUBLE)rd->divisor;
+                rd->collector.calculated_value = (NETDATA_DOUBLE)rd->collector.collected_value
+                                                 * (NETDATA_DOUBLE)rd->multiplier
+                                                 / (NETDATA_DOUBLE)rd->divisor;
 
                 rrdset_debug(st, "%s: CALC ABS/ABS-NO-IN " NETDATA_DOUBLE_FORMAT " = "
                             COLLECTED_NUMBER_FORMAT
                             " * " NETDATA_DOUBLE_FORMAT
                             " / " NETDATA_DOUBLE_FORMAT
                           , rrddim_name(rd)
-                          , rd->calculated_value
-                          , rd->collected_value
+                          , rd->collector.calculated_value
+                          , rd->collector.collected_value
                           , (NETDATA_DOUBLE)rd->multiplier
                           , (NETDATA_DOUBLE)rd->divisor
                 );
@@ -1742,28 +1742,28 @@ void rrdset_timed_done(RRDSET *st, struct timeval now, bool pending_rrdset_next)
 
             case RRD_ALGORITHM_PCENT_OVER_ROW_TOTAL:
                 if(unlikely(!collected_total))
-                    rd->calculated_value = 0;
+                    rd->collector.calculated_value = 0;
                 else
                     // the percentage of the current value
                     // over the total of all dimensions
-                    rd->calculated_value =
+                    rd->collector.calculated_value =
                             (NETDATA_DOUBLE)100
-                            * (NETDATA_DOUBLE)rd->collected_value
+                            * (NETDATA_DOUBLE)rd->collector.collected_value
                             / (NETDATA_DOUBLE)collected_total;
 
                 rrdset_debug(st, "%s: CALC PCENT-ROW " NETDATA_DOUBLE_FORMAT " = 100"
                             " * " COLLECTED_NUMBER_FORMAT
                             " / " COLLECTED_NUMBER_FORMAT
                           , rrddim_name(rd)
-                          , rd->calculated_value
-                          , rd->collected_value
+                          , rd->collector.calculated_value
+                          , rd->collector.collected_value
                           , collected_total
                 );
                 break;
 
             case RRD_ALGORITHM_INCREMENTAL:
-                if(unlikely(rd->collections_counter <= 1)) {
-                    rd->calculated_value = 0;
+                if(unlikely(rd->collector.counter <= 1)) {
+                    rd->collector.calculated_value = 0;
                     continue;
                 }
 
@@ -1771,19 +1771,19 @@ void rrdset_timed_done(RRDSET *st, struct timeval now, bool pending_rrdset_next)
                 // to reset the calculation (it will give zero as the calculation for this second).
                 // It is imperative to set the comparison to uint64_t since type collected_number is signed and
                 // produces wrong results as far as incremental counters are concerned.
-                if(unlikely((uint64_t)rd->last_collected_value > (uint64_t)rd->collected_value)) {
+                if(unlikely((uint64_t)rd->collector.last_collected_value > (uint64_t)rd->collector.collected_value)) {
                     debug(D_RRD_STATS, "'%s' / '%s': RESET or OVERFLOW. Last collected value = " COLLECTED_NUMBER_FORMAT ", current = " COLLECTED_NUMBER_FORMAT
                           , rrdset_id(st)
                           , rrddim_name(rd)
-                          , rd->last_collected_value
-                          , rd->collected_value);
+                          , rd->collector.last_collected_value
+                          , rd->collector.collected_value);
 
                     if(!(rrddim_option_check(rd, RRDDIM_OPTION_DONT_DETECT_RESETS_OR_OVERFLOWS)))
                         has_reset_value = 1;
 
-                    uint64_t last = (uint64_t)rd->last_collected_value;
-                    uint64_t new = (uint64_t)rd->collected_value;
-                    uint64_t max = (uint64_t)rd->collected_value_max;
+                    uint64_t last = (uint64_t)rd->collector.last_collected_value;
+                    uint64_t new = (uint64_t)rd->collector.collected_value;
+                    uint64_t max = (uint64_t)rd->collector.collected_value_max;
                     uint64_t cap = 0;
 
                     // Signed values are handled by exploiting two's complement which will produce positive deltas
@@ -1800,19 +1800,19 @@ void rrdset_timed_done(RRDSET *st, struct timeval now, bool pending_rrdset_next)
                     // overflow.
                     // TODO: remember recent history of rates and compare with current rate to reduce this chance.
                     if (delta < max_acceptable_rate) {
-                        rd->calculated_value +=
+                        rd->collector.calculated_value +=
                                 (NETDATA_DOUBLE) delta
                                 * (NETDATA_DOUBLE) rd->multiplier
                                 / (NETDATA_DOUBLE) rd->divisor;
                     } else {
                         // This is a reset. Any overflow with a rate greater than MAX_INCREMENTAL_PERCENT_RATE will also
                         // be detected as a reset instead.
-                        rd->calculated_value += (NETDATA_DOUBLE)0;
+                        rd->collector.calculated_value += (NETDATA_DOUBLE)0;
                     }
                 }
                 else {
-                    rd->calculated_value +=
-                            (NETDATA_DOUBLE) (rd->collected_value - rd->last_collected_value)
+                    rd->collector.calculated_value +=
+                            (NETDATA_DOUBLE) (rd->collector.collected_value - rd->collector.last_collected_value)
                             * (NETDATA_DOUBLE) rd->multiplier
                             / (NETDATA_DOUBLE) rd->divisor;
                 }
@@ -1823,35 +1823,35 @@ void rrdset_timed_done(RRDSET *st, struct timeval now, bool pending_rrdset_next)
                                     " * " NETDATA_DOUBLE_FORMAT
                             " / " NETDATA_DOUBLE_FORMAT
                           , rrddim_name(rd)
-                          , rd->calculated_value
-                          , rd->collected_value, rd->last_collected_value
+                          , rd->collector.calculated_value
+                          , rd->collector.collected_value, rd->collector.last_collected_value
                           , (NETDATA_DOUBLE)rd->multiplier
                           , (NETDATA_DOUBLE)rd->divisor
                 );
                 break;
 
             case RRD_ALGORITHM_PCENT_OVER_DIFF_TOTAL:
-                if(unlikely(rd->collections_counter <= 1)) {
-                    rd->calculated_value = 0;
+                if(unlikely(rd->collector.counter <= 1)) {
+                    rd->collector.calculated_value = 0;
                     continue;
                 }
 
                 // the percentage of the current increment
                 // over the increment of all dimensions together
                 if(unlikely(collected_total == last_collected_total))
-                    rd->calculated_value = 0;
+                    rd->collector.calculated_value = 0;
                 else
-                    rd->calculated_value =
+                    rd->collector.calculated_value =
                             (NETDATA_DOUBLE)100
-                            * (NETDATA_DOUBLE)(rd->collected_value - rd->last_collected_value)
+                            * (NETDATA_DOUBLE)(rd->collector.collected_value - rd->collector.last_collected_value)
                             / (NETDATA_DOUBLE)(collected_total - last_collected_total);
 
                 rrdset_debug(st, "%s: CALC PCENT-DIFF " NETDATA_DOUBLE_FORMAT " = 100"
                             " * (" COLLECTED_NUMBER_FORMAT " - " COLLECTED_NUMBER_FORMAT ")"
                             " / (" COLLECTED_NUMBER_FORMAT " - " COLLECTED_NUMBER_FORMAT ")"
                           , rrddim_name(rd)
-                          , rd->calculated_value
-                          , rd->collected_value, rd->last_collected_value
+                          , rd->collector.calculated_value
+                          , rd->collector.collected_value, rd->collector.last_collected_value
                           , collected_total, last_collected_total
                 );
                 break;
@@ -1859,11 +1859,11 @@ void rrdset_timed_done(RRDSET *st, struct timeval now, bool pending_rrdset_next)
             default:
                 // make the default zero, to make sure
                 // it gets noticed when we add new types
-                rd->calculated_value = 0;
+                rd->collector.calculated_value = 0;
 
                 rrdset_debug(st, "%s: CALC " NETDATA_DOUBLE_FORMAT " = 0"
                           , rrddim_name(rd)
-                          , rd->calculated_value
+                          , rd->collector.calculated_value
                 );
                 break;
         }
@@ -1874,10 +1874,10 @@ void rrdset_timed_done(RRDSET *st, struct timeval now, bool pending_rrdset_next)
                     " last_calculated_value = " NETDATA_DOUBLE_FORMAT
                     " calculated_value = " NETDATA_DOUBLE_FORMAT
                     , rrddim_name(rd)
-                    , rd->last_collected_value
-                    , rd->collected_value
-                    , rd->last_calculated_value
-                    , rd->calculated_value
+                    , rd->collector.last_collected_value
+                    , rd->collector.collected_value
+                    , rd->collector.last_calculated_value
+                    , rd->collector.calculated_value
         );
     }
 
@@ -1913,9 +1913,9 @@ void rrdset_timed_done(RRDSET *st, struct timeval now, bool pending_rrdset_next)
         if(unlikely(!rrddim_check_updated(rd)))
             continue;
 
-        rrdset_debug(st, "%s: setting last_collected_value (old: " COLLECTED_NUMBER_FORMAT ") to last_collected_value (new: " COLLECTED_NUMBER_FORMAT ")", rrddim_name(rd), rd->last_collected_value, rd->collected_value);
+        rrdset_debug(st, "%s: setting last_collected_value (old: " COLLECTED_NUMBER_FORMAT ") to last_collected_value (new: " COLLECTED_NUMBER_FORMAT ")", rrddim_name(rd), rd->collector.last_collected_value, rd->collector.collected_value);
 
-        rd->last_collected_value = rd->collected_value;
+        rd->collector.last_collected_value = rd->collector.collected_value;
 
         switch(rd->algorithm) {
             case RRD_ALGORITHM_INCREMENTAL:
@@ -1923,10 +1923,10 @@ void rrdset_timed_done(RRDSET *st, struct timeval now, bool pending_rrdset_next)
                     rrdset_debug(st, "%s: setting last_calculated_value (old: " NETDATA_DOUBLE_FORMAT ") to "
                                      "last_calculated_value (new: " NETDATA_DOUBLE_FORMAT ")"
                         , rrddim_name(rd)
-                        , rd->last_calculated_value + rd->calculated_value
-                        , rd->calculated_value);
+                        , rd->collector.last_calculated_value + rd->collector.calculated_value
+                        , rd->collector.calculated_value);
 
-                    rd->last_calculated_value += rd->calculated_value;
+                    rd->collector.last_calculated_value += rd->collector.calculated_value;
                 }
                 else {
                     rrdset_debug(st, "THIS IS THE FIRST POINT");
@@ -1939,15 +1939,15 @@ void rrdset_timed_done(RRDSET *st, struct timeval now, bool pending_rrdset_next)
                 rrdset_debug(st, "%s: setting last_calculated_value (old: " NETDATA_DOUBLE_FORMAT ") to "
                                  "last_calculated_value (new: " NETDATA_DOUBLE_FORMAT ")"
                     , rrddim_name(rd)
-                    , rd->last_calculated_value
-                    , rd->calculated_value);
+                    , rd->collector.last_calculated_value
+                    , rd->collector.calculated_value);
 
-                rd->last_calculated_value = rd->calculated_value;
+                rd->collector.last_calculated_value = rd->collector.calculated_value;
                 break;
         }
 
-        rd->calculated_value = 0;
-        rd->collected_value = 0;
+        rd->collector.calculated_value = 0;
+        rd->collector.collected_value = 0;
         rrddim_clear_updated(rd);
 
         rrdset_debug(st, "%s: END "
@@ -1956,10 +1956,10 @@ void rrdset_timed_done(RRDSET *st, struct timeval now, bool pending_rrdset_next)
                     " last_calculated_value = " NETDATA_DOUBLE_FORMAT
                     " calculated_value = " NETDATA_DOUBLE_FORMAT
                     , rrddim_name(rd)
-                    , rd->last_collected_value
-                    , rd->collected_value
-                    , rd->last_calculated_value
-                    , rd->calculated_value
+                    , rd->collector.last_collected_value
+                    , rd->collector.collected_value
+                    , rd->collector.last_calculated_value
+                    , rd->collector.calculated_value
         );
     }
 
diff --git a/database/rrdvar.c b/database/rrdvar.c
index 914a5d6edb..b03efc9589 100644
--- a/database/rrdvar.c
+++ b/database/rrdvar.c
@@ -272,9 +272,9 @@ void rrdvar_store_for_chart(RRDHOST *host, RRDSET *st) {
 
     RRDDIM *rd;
     rrddim_foreach_read(rd, st) {
-        rrddimvar_add_and_leave_released(rd, RRDVAR_TYPE_CALCULATED, NULL, NULL, &rd->last_stored_value, RRDVAR_FLAG_NONE);
-        rrddimvar_add_and_leave_released(rd, RRDVAR_TYPE_COLLECTED, NULL, "_raw", &rd->last_collected_value, RRDVAR_FLAG_NONE);
-        rrddimvar_add_and_leave_released(rd, RRDVAR_TYPE_TIME_T, NULL, "_last_collected_t", &rd->last_collected_time.tv_sec, RRDVAR_FLAG_NONE);
+        rrddimvar_add_and_leave_released(rd, RRDVAR_TYPE_CALCULATED, NULL, NULL, &rd->collector.last_stored_value, RRDVAR_FLAG_NONE);
+        rrddimvar_add_and_leave_released(rd, RRDVAR_TYPE_COLLECTED, NULL, "_raw", &rd->collector.last_collected_value, RRDVAR_FLAG_NONE);
+        rrddimvar_add_and_leave_released(rd, RRDVAR_TYPE_TIME_T, NULL, "_last_collected_t", &rd->collector.last_collected_time.tv_sec, RRDVAR_FLAG_NONE);
     }
     rrddim_foreach_done(rd);
 }