incremental overflows should not show zeros values (#4538)

* incremental overflows do not show zeros; fixes #4533

* use the max per metric per session for detecting counter size

6 files changed, 272 insertions, 83 deletions

diff --git a/daemon/unit_test.c b/daemon/unit_test.c
index 9978647b43..a92a50a111 100644
--- a/daemon/unit_test.c
+++ b/daemon/unit_test.c
@@ -130,13 +130,17 @@ int check_storage_number(calculated_number n, int debug) {
             p, pdiff, pcdiff
         );
         if(len != strlen(buffer)) fprintf(stderr, "ERROR: printed number %s is reported to have length %zu but it has %zu\n", buffer, len, strlen(buffer));
-        if(dcdiff > ACCURACY_LOSS) fprintf(stderr, "WARNING: packing number " CALCULATED_NUMBER_FORMAT " has accuracy loss " CALCULATED_NUMBER_FORMAT " %%\n", n, dcdiff);
-        if(pcdiff > ACCURACY_LOSS) fprintf(stderr, "WARNING: re-parsing the packed, unpacked and printed number " CALCULATED_NUMBER_FORMAT " has accuracy loss " CALCULATED_NUMBER_FORMAT " %%\n", n, pcdiff);
+
+        if(dcdiff > ACCURACY_LOSS_ACCEPTED_PERCENT)
+            fprintf(stderr, "WARNING: packing number " CALCULATED_NUMBER_FORMAT " has accuracy loss " CALCULATED_NUMBER_FORMAT " %%\n", n, dcdiff);
+
+        if(pcdiff > ACCURACY_LOSS_ACCEPTED_PERCENT)
+            fprintf(stderr, "WARNING: re-parsing the packed, unpacked and printed number " CALCULATED_NUMBER_FORMAT " has accuracy loss " CALCULATED_NUMBER_FORMAT " %%\n", n, pcdiff);
     }
 
     if(len != strlen(buffer)) return 1;
-    if(dcdiff > ACCURACY_LOSS) return 3;
-    if(pcdiff > ACCURACY_LOSS) return 4;
+    if(dcdiff > ACCURACY_LOSS_ACCEPTED_PERCENT) return 3;
+    if(pcdiff > ACCURACY_LOSS_ACCEPTED_PERCENT) return 4;
     return 0;
 }
 
@@ -159,6 +163,9 @@ void benchmark_storage_number(int loop, int multiplier) {
     storage_number s;
     unsigned long long user, system, total, mine, their;
 
+    calculated_number storage_number_positive_min = unpack_storage_number(STORAGE_NUMBER_POSITIVE_MIN_RAW);
+    calculated_number storage_number_positive_max = unpack_storage_number(STORAGE_NUMBER_POSITIVE_MAX_RAW);
+
     char buffer[100];
 
     struct rusage now, last;
@@ -181,11 +188,11 @@ void benchmark_storage_number(int loop, int multiplier) {
     }
 
     fprintf(stderr, "\nNETDATA FLOATING POINT\n");
-    fprintf(stderr, "MIN POSITIVE VALUE " CALCULATED_NUMBER_FORMAT "\n", storage_number_min(1));
-    fprintf(stderr, "MAX POSITIVE VALUE " CALCULATED_NUMBER_FORMAT "\n", (calculated_number)STORAGE_NUMBER_POSITIVE_MAX);
-    fprintf(stderr, "MIN NEGATIVE VALUE " CALCULATED_NUMBER_FORMAT "\n", (calculated_number)STORAGE_NUMBER_NEGATIVE_MIN);
-    fprintf(stderr, "MAX NEGATIVE VALUE " CALCULATED_NUMBER_FORMAT "\n", -storage_number_min(1));
-    fprintf(stderr, "Maximum accuracy loss: " CALCULATED_NUMBER_FORMAT "%%\n\n\n", (calculated_number)ACCURACY_LOSS);
+    fprintf(stderr, "MIN POSITIVE VALUE " CALCULATED_NUMBER_FORMAT "\n", unpack_storage_number(STORAGE_NUMBER_POSITIVE_MIN_RAW));
+    fprintf(stderr, "MAX POSITIVE VALUE " CALCULATED_NUMBER_FORMAT "\n", unpack_storage_number(STORAGE_NUMBER_POSITIVE_MAX_RAW));
+    fprintf(stderr, "MIN NEGATIVE VALUE " CALCULATED_NUMBER_FORMAT "\n", unpack_storage_number(STORAGE_NUMBER_NEGATIVE_MIN_RAW));
+    fprintf(stderr, "MAX NEGATIVE VALUE " CALCULATED_NUMBER_FORMAT "\n", unpack_storage_number(STORAGE_NUMBER_NEGATIVE_MAX_RAW));
+    fprintf(stderr, "Maximum accuracy loss accepted: " CALCULATED_NUMBER_FORMAT "%%\n\n\n", (calculated_number)ACCURACY_LOSS_ACCEPTED_PERCENT);
 
     // ------------------------------------------------------------------------
 
@@ -194,11 +201,11 @@ void benchmark_storage_number(int loop, int multiplier) {
 
     // do the job
     for(j = 1; j < 11 ;j++) {
-        n = STORAGE_NUMBER_POSITIVE_MIN * j;
+        n = storage_number_positive_min * j;
 
         for(i = 0; i < loop ;i++) {
             n *= multiplier;
-            if(n > STORAGE_NUMBER_POSITIVE_MAX) n = STORAGE_NUMBER_POSITIVE_MIN;
+            if(n > storage_number_positive_max) n = storage_number_positive_min;
 
             print_calculated_number(buffer, n);
         }
@@ -219,11 +226,11 @@ void benchmark_storage_number(int loop, int multiplier) {
 
     // do the job
     for(j = 1; j < 11 ;j++) {
-        n = STORAGE_NUMBER_POSITIVE_MIN * j;
+        n = storage_number_positive_min * j;
 
         for(i = 0; i < loop ;i++) {
             n *= multiplier;
-            if(n > STORAGE_NUMBER_POSITIVE_MAX) n = STORAGE_NUMBER_POSITIVE_MIN;
+            if(n > storage_number_positive_max) n = storage_number_positive_min;
             snprintfz(buffer, 100, CALCULATED_NUMBER_FORMAT, n);
         }
     }
@@ -250,13 +257,13 @@ void benchmark_storage_number(int loop, int multiplier) {
 
     // do the job
     for(j = 1; j < 11 ;j++) {
-        n = STORAGE_NUMBER_POSITIVE_MIN * j;
+        n = storage_number_positive_min * j;
 
         for(i = 0; i < loop ;i++) {
             n *= multiplier;
-            if(n > STORAGE_NUMBER_POSITIVE_MAX) n = STORAGE_NUMBER_POSITIVE_MIN;
+            if(n > storage_number_positive_max) n = storage_number_positive_min;
 
-            s = pack_storage_number(n, 1);
+            s = pack_storage_number(n, SN_EXISTS);
             d = unpack_storage_number(s);
             print_calculated_number(buffer, d);
         }
@@ -282,7 +289,7 @@ void benchmark_storage_number(int loop, int multiplier) {
 }
 
 static int check_storage_number_exists() {
-    uint32_t flags = SN_EXISTS;
+    uint32_t flags;
 
 
     for(flags = 0; flags < 7 ; flags++) {
@@ -309,10 +316,12 @@ static int check_storage_number_exists() {
     return 0;
 }
 
-int unit_test_storage()
-{
+int unit_test_storage() {
     if(check_storage_number_exists()) return 0;
 
+    calculated_number storage_number_positive_min = unpack_storage_number(STORAGE_NUMBER_POSITIVE_MIN_RAW);
+    calculated_number storage_number_negative_max = unpack_storage_number(STORAGE_NUMBER_NEGATIVE_MAX_RAW);
+
     calculated_number c, a = 0;
     int i, j, g, r = 0;
 
@@ -325,14 +334,15 @@ int unit_test_storage()
             a += 0.0000001;
             c = a * g;
             for(i = 0; i < 21 ;i++, c *= 10) {
-                if(c > 0 && c < STORAGE_NUMBER_POSITIVE_MIN) continue;
-                if(c < 0 && c > STORAGE_NUMBER_NEGATIVE_MAX) continue;
+                if(c > 0 && c < storage_number_positive_min) continue;
+                if(c < 0 && c > storage_number_negative_max) continue;
 
                 if(check_storage_number(c, 1)) return 1;
             }
         }
     }
 
+    // if(check_storage_number(858993459.1234567, 1)) return 1;
     benchmark_storage_number(1000000, 2);
     return r;
 }
@@ -575,28 +585,36 @@ struct test test4 = {
 };
 
 // --------------------------------------------------------------------------------------------------------------------
-// test5
+// test5 - 32 bit overflows
 
 struct feed_values test5_feed[] = {
-        { 500000, 1000 },
-        { 1000000, 2000 },
-        { 1000000, 2000 },
-        { 1000000, 2000 },
-        { 1000000, 3000 },
-        { 1000000, 2000 },
-        { 1000000, 2000 },
-        { 1000000, 2000 },
-        { 1000000, 2000 },
-        { 1000000, 2000 },
+        { 0,       0x00000000FFFFFFFFULL / 3 * 0 },
+        { 1000000, 0x00000000FFFFFFFFULL / 3 * 1 },
+        { 1000000, 0x00000000FFFFFFFFULL / 3 * 2 },
+        { 1000000, 0x00000000FFFFFFFFULL / 3 * 0 },
+        { 1000000, 0x00000000FFFFFFFFULL / 3 * 1 },
+        { 1000000, 0x00000000FFFFFFFFULL / 3 * 2 },
+        { 1000000, 0x00000000FFFFFFFFULL / 3 * 0 },
+        { 1000000, 0x00000000FFFFFFFFULL / 3 * 1 },
+        { 1000000, 0x00000000FFFFFFFFULL / 3 * 2 },
+        { 1000000, 0x00000000FFFFFFFFULL / 3 * 0 },
 };
 
 calculated_number test5_results[] = {
-        1000, 500, 0, 500, 500, 0, 0, 0, 0
+        0x00000000FFFFFFFFULL / 3,
+        0x00000000FFFFFFFFULL / 3,
+        0x00000000FFFFFFFFULL / 3,
+        0x00000000FFFFFFFFULL / 3,
+        0x00000000FFFFFFFFULL / 3,
+        0x00000000FFFFFFFFULL / 3,
+        0x00000000FFFFFFFFULL / 3,
+        0x00000000FFFFFFFFULL / 3,
+        0x00000000FFFFFFFFULL / 3,
 };
 
 struct test test5 = {
         "test5",            // name
-        "test incremental values ups and downs",
+        "test 32-bit incremental values overflow",
         1,                  // update_every
         1,                  // multiplier
         1,                  // divisor
@@ -610,6 +628,135 @@ struct test test5 = {
 };
 
 // --------------------------------------------------------------------------------------------------------------------
+// test5b - 16 bit overflows
+
+struct feed_values test5b_feed[] = {
+        { 0,       0x000000000000FFFFULL / 3 * 0 },
+        { 1000000, 0x000000000000FFFFULL / 3 * 1 },
+        { 1000000, 0x000000000000FFFFULL / 3 * 2 },
+        { 1000000, 0x000000000000FFFFULL / 3 * 0 },
+        { 1000000, 0x000000000000FFFFULL / 3 * 1 },
+        { 1000000, 0x000000000000FFFFULL / 3 * 2 },
+        { 1000000, 0x000000000000FFFFULL / 3 * 0 },
+        { 1000000, 0x000000000000FFFFULL / 3 * 1 },
+        { 1000000, 0x000000000000FFFFULL / 3 * 2 },
+        { 1000000, 0x000000000000FFFFULL / 3 * 0 },
+};
+
+calculated_number test5b_results[] = {
+        0x000000000000FFFFULL / 3,
+        0x000000000000FFFFULL / 3,
+        0x000000000000FFFFULL / 3,
+        0x000000000000FFFFULL / 3,
+        0x000000000000FFFFULL / 3,
+        0x000000000000FFFFULL / 3,
+        0x000000000000FFFFULL / 3,
+        0x000000000000FFFFULL / 3,
+        0x000000000000FFFFULL / 3,
+};
+
+struct test test5b = {
+        "test5b",            // name
+        "test 16-bit incremental values overflow",
+        1,                  // update_every
+        1,                  // multiplier
+        1,                  // divisor
+        RRD_ALGORITHM_INCREMENTAL, // algorithm
+        10,                 // feed entries
+        9,                  // result entries
+        test5b_feed,        // feed
+        test5b_results,     // results
+        NULL,               // feed2
+        NULL                // results2
+};
+
+// --------------------------------------------------------------------------------------------------------------------
+// test5c - 8 bit overflows
+
+struct feed_values test5c_feed[] = {
+        { 0,       0x00000000000000FFULL / 3 * 0 },
+        { 1000000, 0x00000000000000FFULL / 3 * 1 },
+        { 1000000, 0x00000000000000FFULL / 3 * 2 },
+        { 1000000, 0x00000000000000FFULL / 3 * 0 },
+        { 1000000, 0x00000000000000FFULL / 3 * 1 },
+        { 1000000, 0x00000000000000FFULL / 3 * 2 },
+        { 1000000, 0x00000000000000FFULL / 3 * 0 },
+        { 1000000, 0x00000000000000FFULL / 3 * 1 },
+        { 1000000, 0x00000000000000FFULL / 3 * 2 },
+        { 1000000, 0x00000000000000FFULL / 3 * 0 },
+};
+
+calculated_number test5c_results[] = {
+        0x00000000000000FFULL / 3,
+        0x00000000000000FFULL / 3,
+        0x00000000000000FFULL / 3,
+        0x00000000000000FFULL / 3,
+        0x00000000000000FFULL / 3,
+        0x00000000000000FFULL / 3,
+        0x00000000000000FFULL / 3,
+        0x00000000000000FFULL / 3,
+        0x00000000000000FFULL / 3,
+};
+
+struct test test5c = {
+        "test5c",            // name
+        "test 8-bit incremental values overflow",
+        1,                  // update_every
+        1,                  // multiplier
+        1,                  // divisor
+        RRD_ALGORITHM_INCREMENTAL, // algorithm
+        10,                 // feed entries
+        9,                  // result entries
+        test5c_feed,        // feed
+        test5c_results,     // results
+        NULL,               // feed2
+        NULL                // results2
+};
+
+// --------------------------------------------------------------------------------------------------------------------
+// test5d - 64 bit overflows
+
+struct feed_values test5d_feed[] = {
+        { 0,       0xFFFFFFFFFFFFFFFFULL / 3 * 0 },
+        { 1000000, 0xFFFFFFFFFFFFFFFFULL / 3 * 1 },
+        { 1000000, 0xFFFFFFFFFFFFFFFFULL / 3 * 2 },
+        { 1000000, 0xFFFFFFFFFFFFFFFFULL / 3 * 0 },
+        { 1000000, 0xFFFFFFFFFFFFFFFFULL / 3 * 1 },
+        { 1000000, 0xFFFFFFFFFFFFFFFFULL / 3 * 2 },
+        { 1000000, 0xFFFFFFFFFFFFFFFFULL / 3 * 0 },
+        { 1000000, 0xFFFFFFFFFFFFFFFFULL / 3 * 1 },
+        { 1000000, 0xFFFFFFFFFFFFFFFFULL / 3 * 2 },
+        { 1000000, 0xFFFFFFFFFFFFFFFFULL / 3 * 0 },
+};
+
+calculated_number test5d_results[] = {
+        0xFFFFFFFFFFFFFFFFULL / 3,
+        0xFFFFFFFFFFFFFFFFULL / 3,
+        0xFFFFFFFFFFFFFFFFULL / 3,
+        0xFFFFFFFFFFFFFFFFULL / 3,
+        0xFFFFFFFFFFFFFFFFULL / 3,
+        0xFFFFFFFFFFFFFFFFULL / 3,
+        0xFFFFFFFFFFFFFFFFULL / 3,
+        0xFFFFFFFFFFFFFFFFULL / 3,
+        0xFFFFFFFFFFFFFFFFULL / 3,
+};
+
+struct test test5d = {
+        "test5d",            // name
+        "test 64-bit incremental values overflow",
+        1,                  // update_every
+        1,                  // multiplier
+        1,                  // divisor
+        RRD_ALGORITHM_INCREMENTAL, // algorithm
+        10,                 // feed entries
+        9,                  // result entries
+        test5d_feed,        // feed
+        test5d_results,     // results
+        NULL,               // feed2
+        NULL                // results2
+};
+
+// --------------------------------------------------------------------------------------------------------------------
 // test6
 
 struct feed_values test6_feed[] = {
@@ -1131,7 +1278,7 @@ int run_test(struct test *test)
     unsigned long max = (st->counter < test->result_entries)?st->counter:test->result_entries;
     for(c = 0 ; c < max ; c++) {
         calculated_number v = unpack_storage_number(rd->values[c]);
-        calculated_number n = test->results[c];
+        calculated_number n = unpack_storage_number(pack_storage_number(test->results[c], SN_EXISTS));
         int same = (calculated_number_round(v * 10000000.0) == calculated_number_round(n * 10000000.0))?1:0;
         fprintf(stderr, "    %s/%s: checking position %lu (at %lu secs), expecting value " CALCULATED_NUMBER_FORMAT ", found " CALCULATED_NUMBER_FORMAT ", %s\n",
             test->name, rd->name, c+1,
@@ -1267,6 +1414,15 @@ int run_all_mockup_tests(void)
     if(run_test(&test5))
         return 1;
 
+    if(run_test(&test5b))
+        return 1;
+
+    if(run_test(&test5c))
+        return 1;
+
+    if(run_test(&test5d))
+        return 1;
+
     if(run_test(&test6))
         return 1;
 
diff --git a/database/rrd.h b/database/rrd.h
index 19eb100cd4..24705ebb1c 100644
--- a/database/rrd.h
+++ b/database/rrd.h
@@ -176,7 +176,9 @@ struct rrddim {
     char *cache_filename;                           // the filename we load/save from/to this set
 
     size_t collections_counter;                     // the number of times we added values to this rrdim
-    size_t unused[10];
+    size_t unused[9];
+
+    collected_number collected_value_max;           // the absolute maximum of the collected value
 
     unsigned int updated:1;                         // 1 when the dimension has been updated since the last processing
     unsigned int exposed:1;                         // 1 when set what have sent this dimension to the central netdata
diff --git a/database/rrddim.c b/database/rrddim.c
index 95e485106c..e98f702fe0 100644
--- a/database/rrddim.c
+++ b/database/rrddim.c
@@ -239,6 +239,7 @@ RRDDIM *rrddim_add_custom(RRDSET *st, const char *id, const char *name, collecte
     rd->last_calculated_value = 0;
     rd->collected_value = 0;
     rd->last_collected_value = 0;
+    rd->collected_value_max = 0;
     rd->collected_volume = 0;
     rd->stored_volume = 0;
     rd->last_stored_value = 0;
@@ -380,6 +381,9 @@ inline collected_number rrddim_set_by_pointer(RRDSET *st, RRDDIM *rd, collected_
 
     rd->collections_counter++;
 
+    collected_number v = (value >= 0) ? value : -value;
+    if(unlikely(v > rd->collected_value_max)) rd->collected_value_max = v;
+
     // fprintf(stderr, "%s.%s %llu " COLLECTED_NUMBER_FORMAT " dt %0.6f" " rate " CALCULATED_NUMBER_FORMAT "\n", st->name, rd->name, st->usec_since_last_update, value, (float)((double)st->usec_since_last_update / (double)1000000), (calculated_number)((value - rd->last_collected_value) * (calculated_number)rd->multiplier / (calculated_number)rd->divisor * 1000000.0 / (calculated_number)st->usec_since_last_update));
 
     return rd->last_collected_value;
diff --git a/database/rrdset.c b/database/rrdset.c
index 3f5ba73b62..d74ac91abd 100644
--- a/database/rrdset.c
+++ b/database/rrdset.c
@@ -1080,7 +1080,7 @@ static inline size_t rrdset_done_interpolate(
                       , get_storage_number_flags(rd->values[current_entry])
                       , t1
                       , accuracy
-                      , (accuracy > ACCURACY_LOSS) ? " **TOO BIG** " : ""
+                      , (accuracy > ACCURACY_LOSS_ACCEPTED_PERCENT) ? " **TOO BIG** " : ""
                 );
 
                 rd->collected_volume += t1;
@@ -1093,7 +1093,7 @@ static inline size_t rrdset_done_interpolate(
                       , rd->stored_volume
                       , rd->collected_volume
                       , accuracy
-                      , (accuracy > ACCURACY_LOSS) ? " **TOO BIG** " : ""
+                      , (accuracy > ACCURACY_LOSS_ACCEPTED_PERCENT) ? " **TOO BIG** " : ""
                 );
             }
             #endif
@@ -1381,13 +1381,29 @@ void rrdset_done(RRDSET *st) {
                     if(!(rrddim_flag_check(rd, RRDDIM_FLAG_DONT_DETECT_RESETS_OR_OVERFLOWS)))
                         storage_flags = SN_EXISTS_RESET;
 
-                    rd->last_collected_value = rd->collected_value;
-                }
+                    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 cap = 0;
+
+                         if(max > 0x00000000FFFFFFFFULL) cap = 0xFFFFFFFFFFFFFFFFULL;
+                    else if(max > 0x000000000000FFFFULL) cap = 0x00000000FFFFFFFFULL;
+                    else if(max > 0x00000000000000FFULL) cap = 0x000000000000FFFFULL;
+                    else                                 cap = 0x00000000000000FFULL;
+
+                    uint64_t delta = cap - last + new;
 
-                rd->calculated_value +=
-                        (calculated_number)(rd->collected_value - rd->last_collected_value)
-                        * (calculated_number)rd->multiplier
-                        / (calculated_number)rd->divisor;
+                    rd->calculated_value +=
+                            (calculated_number) delta
+                            * (calculated_number) rd->multiplier
+                            / (calculated_number) rd->divisor;
+                }
+                else {
+                    rd->calculated_value +=
+                            (calculated_number) (rd->collected_value - rd->last_collected_value)
+                            * (calculated_number) rd->multiplier
+                            / (calculated_number) rd->divisor;
+                }
 
                 #ifdef NETDATA_INTERNAL_CHECKS
                 rrdset_debug(st, "%s: CALC INC PRE "
diff --git a/libnetdata/storage_number/storage_number.c b/libnetdata/storage_number/storage_number.c
index db4cb700b4..6825fa7d03 100644
--- a/libnetdata/storage_number/storage_number.c
+++ b/libnetdata/storage_number/storage_number.c
@@ -2,19 +2,20 @@
 
 #include "../libnetdata.h"
 
-storage_number pack_storage_number(calculated_number value, uint32_t flags)
-{
+storage_number pack_storage_number(calculated_number value, uint32_t flags) {
     // bit 32 = sign 0:positive, 1:negative
     // bit 31 = 0:divide, 1:multiply
     // bit 30, 29, 28 = (multiplier or divider) 0-7 (8 total)
-    // bit 27, 26, 25 flags
+    // bit 27 SN_EXISTS_100
+    // bit 26 SN_EXISTS_RESET
+    // bit 25 SN_EXISTS
     // bit 24 to bit 1 = the value
 
     storage_number r = get_storage_number_flags(flags);
     if(!value) return r;
 
     int m = 0;
-    calculated_number n = value;
+    calculated_number n = value, factor = 10;
 
     // if the value is negative
     // add the sign bit and make it positive
@@ -23,11 +24,16 @@ storage_number pack_storage_number(calculated_number value, uint32_t flags)
         n = -n;
     }
 
+    if(n / 10000000.0 > 0x00ffffff) {
+        factor = 100;
+        r |= SN_EXISTS_100;
+    }
+
     // make its integer part fit in 0x00ffffff
     // by dividing it by 10 up to 7 times
     // and increasing the multiplier
     while(m < 7 && n > (calculated_number)0x00ffffff) {
-        n /= 10;
+        n /= factor;
         m++;
     }
 
@@ -71,35 +77,44 @@ storage_number pack_storage_number(calculated_number value, uint32_t flags)
     return r;
 }
 
-calculated_number unpack_storage_number(storage_number value)
-{
+calculated_number unpack_storage_number(storage_number value) {
     if(!value) return 0;
 
     int sign = 0, exp = 0;
+    int factor = 10;
 
-    value ^= get_storage_number_flags(value);
-
-    if(value & (1 << 31)) {
+    // bit 32 = 0:positive, 1:negative
+    if(unlikely(value & (1 << 31)))
         sign = 1;
-        value ^= 1 << 31;
-    }
 
-    if(value & (1 << 30)) {
+    // bit 31 = 0:divide, 1:multiply
+    if(unlikely(value & (1 << 30)))
         exp = 1;
-        value ^= 1 << 30;
-    }
 
-    int mul = value >> 27;
-    value ^= mul << 27;
+    // bit 27 SN_EXISTS_100
+    if(unlikely(value & (1 << 26)))
+        factor = 100;
+
+    // bit 26 SN_EXISTS_RESET
+    // bit 25 SN_EXISTS
+
+    // bit 30, 29, 28 = (multiplier or divider) 0-7 (8 total)
+    int mul = (value & ((1<<29)|(1<<28)|(1<<27))) >> 27;
+
+    // bit 24 to bit 1 = the value, so remove all other bits
+    value ^= value & ((1<<31)|(1<<30)|(1<<29)|(1<<28)|(1<<27)|(1<<26)|(1<<25)|(1<<24));
 
     calculated_number n = value;
 
-    // fprintf(stderr, "UNPACK: %08X, sign = %d, exp = %d, mul = %d, n = " CALCULATED_NUMBER_FORMAT "\n", value, sign, exp, mul, n);
+    // fprintf(stderr, "UNPACK: %08X, sign = %d, exp = %d, mul = %d, factor = %d, n = " CALCULATED_NUMBER_FORMAT "\n", value, sign, exp, mul, factor, n);
 
-    while(mul > 0) {
-        if(exp) n *= 10;
-        else n /= 10;
-        mul--;
+    if(exp) {
+        for(; mul; mul--)
+            n *= factor;
+    }
+    else {
+        for( ; mul ; mul--)
+            n /= 10;
     }
 
     if(sign) n = -n;
diff --git a/libnetdata/storage_number/storage_number.h b/libnetdata/storage_number/storage_number.h
index 5353ab60b9..af7d29f3a3 100644
--- a/libnetdata/storage_number/storage_number.h
+++ b/libnetdata/storage_number/storage_number.h
@@ -23,7 +23,7 @@ typedef double calculated_number;
 #define LONG_DOUBLE_MODIFIER "f"
 typedef double LONG_DOUBLE;
 
-#else
+#else // NETDATA_WITHOUT_LONG_DOUBLE
 
 typedef long double calculated_number;
 #define CALCULATED_NUMBER_FORMAT "%0.7Lf"
@@ -33,7 +33,7 @@ typedef long double calculated_number;
 #define LONG_DOUBLE_MODIFIER "Lf"
 typedef long double LONG_DOUBLE;
 
-#endif
+#endif // NETDATA_WITHOUT_LONG_DOUBLE
 
 //typedef long long calculated_number;
 //#define CALCULATED_NUMBER_FORMAT "%lld"
@@ -50,6 +50,7 @@ typedef long double collected_number;
 #define calculated_number_llrint(x) llrintl(x)
 #define calculated_number_round(x) roundl(x)
 #define calculated_number_fabs(x) fabsl(x)
+#define calculated_number_pow(x, y) powl(x, y)
 #define calculated_number_epsilon (calculated_number)0.0000001
 
 #define calculated_number_equal(a, b) (calculated_number_fabs((a) - (b)) < calculated_number_epsilon)
@@ -57,18 +58,12 @@ typedef long double collected_number;
 typedef uint32_t storage_number;
 #define STORAGE_NUMBER_FORMAT "%u"
 
-#define SN_NOT_EXISTS       (0x0 << 24)
-#define SN_EXISTS           (0x1 << 24)
-#define SN_EXISTS_RESET     (0x2 << 24)
-#define SN_EXISTS_UNDEF1    (0x3 << 24)
-#define SN_EXISTS_UNDEF2    (0x4 << 24)
-#define SN_EXISTS_UNDEF3    (0x5 << 24)
-#define SN_EXISTS_UNDEF4    (0x6 << 24)
-
-#define SN_FLAGS_MASK       (~(0x6 << 24))
+#define SN_EXISTS           (1 << 24) // the value exists
+#define SN_EXISTS_RESET     (1 << 25) // the value has been overflown
+#define SN_EXISTS_100       (1 << 26) // very large value (multipler is 100 instead of 10)
 
 // extract the flags
-#define get_storage_number_flags(value) ((((storage_number)(value)) & (1 << 24)) | (((storage_number)(value)) & (2 << 24)) | (((storage_number)(value)) & (4 << 24)))
+#define get_storage_number_flags(value) ((((storage_number)(value)) & (1 << 24)) | (((storage_number)(value)) & (1 << 25)) | (((storage_number)(value)) & (1 << 26)))
 #define SN_EMPTY_SLOT 0x00000000
 
 // checks
@@ -80,13 +75,14 @@ calculated_number unpack_storage_number(storage_number value);
 
 int print_calculated_number(char *str, calculated_number value);
 
-#define STORAGE_NUMBER_POSITIVE_MAX (167772150000000.0)
-#define STORAGE_NUMBER_POSITIVE_MIN (0.0000001)
-#define STORAGE_NUMBER_NEGATIVE_MAX (-0.0000001)
-#define STORAGE_NUMBER_NEGATIVE_MIN (-167772150000000.0)
+//                                          sign       div/mul    <--- multiplier / divider --->     10/100       RESET      EXISTS     VALUE
+#define STORAGE_NUMBER_POSITIVE_MAX_RAW (storage_number)( (0 << 31) | (1 << 30) | (1 << 29) | (1 << 28) | (1<<27) | (1 << 26) | (0 << 25) | (1 << 24) | 0x00ffffff )
+#define STORAGE_NUMBER_POSITIVE_MIN_RAW (storage_number)( (0 << 31) | (0 << 30) | (1 << 29) | (1 << 28) | (1<<27) | (0 << 26) | (0 << 25) | (1 << 24) | 0x00000001 )
+#define STORAGE_NUMBER_NEGATIVE_MAX_RAW (storage_number)( (1 << 31) | (0 << 30) | (1 << 29) | (1 << 28) | (1<<27) | (0 << 26) | (0 << 25) | (1 << 24) | 0x00000001 )
+#define STORAGE_NUMBER_NEGATIVE_MIN_RAW (storage_number)( (1 << 31) | (1 << 30) | (1 << 29) | (1 << 28) | (1<<27) | (1 << 26) | (0 << 25) | (1 << 24) | 0x00ffffff )
 
 // accepted accuracy loss
-#define ACCURACY_LOSS 0.0001
+#define ACCURACY_LOSS_ACCEPTED_PERCENT 0.0001
 #define accuracy_loss(t1, t2) (((t1) == (t2) || (t1) == 0.0 || (t2) == 0.0) ? 0.0 : (100.0 - (((t1) > (t2)) ? ((t2) * 100.0 / (t1) ) : ((t1) * 100.0 / (t2)))))
 
 #endif /* NETDATA_STORAGE_NUMBER_H */