Faster queries (#12988)

* faster rrdeng_load_metric_next()

* no need to check validity for number - already done at the query side

* solve discrepancy between query create and free

* inline unpack_storage_number

database/engine/rrdengine.h

@@ -49,10 +49,12 @@ struct rrdeng_query_handle {
     time_t next_page_time;
     time_t now;
     unsigned position;
+    unsigned entries;
     storage_number *page;
     usec_t page_end_time;
     uint32_t page_length;
     usec_t dt;
+    time_t dt_sec;
 };
 
 typedef enum {

database/engine/rrdengineapi.c

@@ -554,7 +554,7 @@ void rrdeng_load_metric_init(RRDDIM *rd, struct rrddim_query_handle *rrdimm_hand
         handle->next_page_time = INVALID_TIME;
 }
 
-static int rrdeng_load_page_next(struct rrddim_query_handle *rrdimm_handle, unsigned *position_ptr) {
+static int rrdeng_load_page_next(struct rrddim_query_handle *rrdimm_handle) {
     struct rrdeng_query_handle *handle = (struct rrdeng_query_handle *)rrdimm_handle->handle;
 
     struct rrdengine_instance *ctx = handle->ctx;
@@ -576,13 +576,13 @@ static int rrdeng_load_page_next(struct rrddim_query_handle *rrdimm_handle, unsi
         handle->next_page_time = (handle->page_end_time / USEC_PER_SEC) + 1;
 
         if (unlikely(handle->next_page_time > rrdimm_handle->end_time))
-            goto no_more_metrics;
+            return 1;
     }
 
     usec_t next_page_time = handle->next_page_time * USEC_PER_SEC;
     descr = pg_cache_lookup_next(ctx, handle->page_index, &handle->page_index->id, next_page_time, rrdimm_handle->end_time * USEC_PER_SEC);
     if (NULL == descr)
-        goto no_more_metrics;
+        return 1;
 
 #ifdef NETDATA_INTERNAL_CHECKS
     rrd_stat_atomic_add(&ctx->stats.metric_API_consumers, 1);
@@ -591,7 +591,7 @@ static int rrdeng_load_page_next(struct rrddim_query_handle *rrdimm_handle, unsi
     handle->descr = descr;
     pg_cache_atomic_get_pg_info(descr, &page_end_time, &page_length);
     if (unlikely(INVALID_TIME == descr->start_time || INVALID_TIME == page_end_time))
-        goto no_more_metrics;
+        return 1;
 
     if (unlikely(descr->start_time != page_end_time && next_page_time > descr->start_time)) {
         // we're in the middle of the page somewhere
@@ -605,17 +605,16 @@ static int rrdeng_load_page_next(struct rrddim_query_handle *rrdimm_handle, unsi
     handle->page_end_time = page_end_time;
     handle->page_length = page_length;
     handle->page = descr->pg_cache_descr->page;
-    usec_t entries = page_length / sizeof(storage_number);
+    usec_t entries = handle->entries = page_length / sizeof(storage_number);
     if (likely(entries > 1))
         handle->dt = (page_end_time - descr->start_time) / (entries - 1);
     else
         handle->dt = 0;
 
-    *position_ptr = position;
-    return 0;
+    handle->dt_sec = handle->dt / USEC_PER_SEC;
+    handle->position = position;
 
-no_more_metrics:
-    return 1;
+    return 0;
 }
 
 /* Returns the metric and sets its timestamp into current_time */
@@ -626,22 +625,25 @@ storage_number rrdeng_load_metric_next(struct rrddim_query_handle *rrdimm_handle
         return SN_EMPTY_SLOT;
 
     struct rrdeng_page_descr *descr = handle->descr;
-
-    storage_number *page = handle->page;
     unsigned position = handle->position + 1;
+    time_t now = handle->now + handle->dt_sec;
 
-    if (unlikely(!descr || position >= (handle->page_length / sizeof(storage_number)))) {
+    if (unlikely(!descr || position >= handle->entries)) {
         // We need to get a new page
-        if(rrdeng_load_page_next(rrdimm_handle, &position))
-            goto no_more_metrics;
+        if(rrdeng_load_page_next(rrdimm_handle)) {
+            // next calls will not load any more metrics
+            handle->next_page_time = INVALID_TIME;
+            return SN_EMPTY_SLOT;
+        }
 
         descr = handle->descr;
-        page = handle->page;
+        position = handle->position;
+        now = (descr->start_time + position * handle->dt) / USEC_PER_SEC;
     }
 
-    storage_number ret = page[position];
+    storage_number ret = handle->page[position];
     handle->position = position;
-    time_t now = handle->now = (descr->start_time + position * handle->dt) / USEC_PER_SEC;
+    handle->now = now;
 
     if (unlikely(now >= rrdimm_handle->end_time)) {
         // next calls will not load any more metrics
@@ -650,10 +652,6 @@ storage_number rrdeng_load_metric_next(struct rrddim_query_handle *rrdimm_handle
 
     *current_time = now;
     return ret;
-
-no_more_metrics:
-    handle->next_page_time = INVALID_TIME;
-    return SN_EMPTY_SLOT;
 }
 
 int rrdeng_load_metric_is_finished(struct rrddim_query_handle *rrdimm_handle)

libnetdata/storage_number/storage_number.c

@@ -92,56 +92,22 @@ RET_SN:
 }
 
 // Lookup table to make storage number unpacking efficient.
-static calculated_number lut10x[4 * 8];
+calculated_number unpack_storage_number_lut10x[4 * 8];
 
 __attribute__((constructor)) void initialize_lut(void) {
     // The lookup table is partitioned in 4 subtables based on the
     // values of the factor and exp bits.
     for (int i = 0; i < 8; i++) {
         // factor = 0
-        lut10x[0 * 8 + i] = 1 / pow(10, i);    // exp = 0
-        lut10x[1 * 8 + i] = pow(10, i);        // exp = 1
+        unpack_storage_number_lut10x[0 * 8 + i] = 1 / pow(10, i);    // exp = 0
+        unpack_storage_number_lut10x[1 * 8 + i] = pow(10, i);        // exp = 1
 
         // factor = 1
-        lut10x[2 * 8 + i] = 1 / pow(100, i);   // exp = 0
-        lut10x[3 * 8 + i] = pow(100, i);       // exp = 1
+        unpack_storage_number_lut10x[2 * 8 + i] = 1 / pow(100, i);   // exp = 0
+        unpack_storage_number_lut10x[3 * 8 + i] = pow(100, i);       // exp = 1
     }
 }
 
-calculated_number unpack_storage_number(storage_number value) {
-    if(!value) return 0;
-
-    int sign = 1, exp = 0;
-    int factor = 0;
-
-    // bit 32 = 0:positive, 1:negative
-    if(unlikely(value & (1 << 31)))
-        sign = -1;
-
-    // bit 31 = 0:divide, 1:multiply
-    if(unlikely(value & (1 << 30)))
-        exp = 1;
-
-    // bit 27 SN_EXISTS_100
-    if(unlikely(value & (1 << 26)))
-        factor = 1;
-
-    // bit 26 SN_EXISTS_RESET
-    // bit 25 SN_ANOMALY_BIT
-
-    // 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, factor = %d, n = " CALCULATED_NUMBER_FORMAT "\n", value, sign, exp, mul, factor, n);
-
-    return sign * lut10x[(factor * 16) + (exp * 8) + mul] * n;
-}
-
 /*
 int print_calculated_number(char *str, calculated_number value)
 {

libnetdata/storage_number/storage_number.h

@@ -80,7 +80,7 @@ typedef uint32_t storage_number;
 #define did_storage_number_reset(value)  ((((storage_number) (value)) & SN_EXISTS_RESET) != 0)
 
 storage_number pack_storage_number(calculated_number value, uint32_t flags);
-calculated_number unpack_storage_number(storage_number value);
+static inline calculated_number unpack_storage_number(storage_number value) __attribute__((const));
 
 int print_calculated_number(char *str, calculated_number value);
 
@@ -98,4 +98,41 @@ int print_calculated_number(char *str, calculated_number value);
 // period of at least every other 10 samples.
 #define MAX_INCREMENTAL_PERCENT_RATE 10
 
+
+static inline calculated_number unpack_storage_number(storage_number value) {
+    extern calculated_number unpack_storage_number_lut10x[4 * 8];
+
+    if(!value) return 0;
+
+    int sign = 1, exp = 0;
+    int factor = 0;
+
+    // bit 32 = 0:positive, 1:negative
+    if(unlikely(value & (1 << 31)))
+        sign = -1;
+
+    // bit 31 = 0:divide, 1:multiply
+    if(unlikely(value & (1 << 30)))
+        exp = 1;
+
+    // bit 27 SN_EXISTS_100
+    if(unlikely(value & (1 << 26)))
+        factor = 1;
+
+    // bit 26 SN_EXISTS_RESET
+    // bit 25 SN_ANOMALY_BIT
+
+    // 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, factor = %d, n = " CALCULATED_NUMBER_FORMAT "\n", value, sign, exp, mul, factor, n);
+
+    return sign * unpack_storage_number_lut10x[(factor * 16) + (exp * 8) + mul] * n;
+}
+
 #endif /* NETDATA_STORAGE_NUMBER_H */

web/api/queries/average/average.c

@@ -10,9 +10,8 @@ struct grouping_average {
     size_t count;
 };
 
-void *grouping_create_average(RRDR *r) {
-    (void)r;
-    return callocz(1, sizeof(struct grouping_average));
+void grouping_create_average(RRDR *r) {
+    r->internal.grouping_data = callocz(1, sizeof(struct grouping_average));
 }
 
 // resets when switches dimensions
@@ -29,11 +28,9 @@ void grouping_free_average(RRDR *r) {
 }
 
 void grouping_add_average(RRDR *r, calculated_number value) {
-    if(likely(!isnan(value))) {
-        struct grouping_average *g = (struct grouping_average *)r->internal.grouping_data;
-        g->sum += value;
-        g->count++;
-    }
+    struct grouping_average *g = (struct grouping_average *)r->internal.grouping_data;
+    g->sum += value;
+    g->count++;
 }
 
 calculated_number grouping_flush_average(RRDR *r,  RRDR_VALUE_FLAGS *rrdr_value_options_ptr) {

web/api/queries/average/average.h

@@ -6,7 +6,7 @@
 #include "../query.h"
 #include "../rrdr.h"
 
-extern void *grouping_create_average(RRDR *r);
+extern void grouping_create_average(RRDR *r);
 extern void grouping_reset_average(RRDR *r);
 extern void grouping_free_average(RRDR *r);
 extern void grouping_add_average(RRDR *r, calculated_number value);

web/api/queries/des/des.c

@@ -69,14 +69,14 @@ static inline void set_beta(RRDR *r, struct grouping_des *g) {
     //info("beta for chart '%s' is " CALCULATED_NUMBER_FORMAT, r->st->name, g->beta);
 }
 
-void *grouping_create_des(RRDR *r) {
+void grouping_create_des(RRDR *r) {
     struct grouping_des *g = (struct grouping_des *)mallocz(sizeof(struct grouping_des));
     set_alpha(r, g);
     set_beta(r, g);
     g->level = 0.0;
     g->trend = 0.0;
     g->count = 0;
-    return g;
+    r->internal.grouping_data = g;
 }
 
 // resets when switches dimensions
@@ -99,28 +99,26 @@ void grouping_free_des(RRDR *r) {
 void grouping_add_des(RRDR *r, calculated_number value) {
     struct grouping_des *g = (struct grouping_des *)r->internal.grouping_data;
 
-    if(calculated_number_isnumber(value)) {
-        if(likely(g->count > 0)) {
-            // we have at least a number so far
+    if(likely(g->count > 0)) {
+        // we have at least a number so far
 
-            if(unlikely(g->count == 1)) {
-                // the second value we got
-                g->trend = value - g->trend;
-                g->level = value;
-            }
-
-            // for the values, except the first
-            calculated_number last_level = g->level;
-            g->level = (g->alpha * value) + (g->alpha_other * (g->level + g->trend));
-            g->trend = (g->beta * (g->level - last_level)) + (g->beta_other * g->trend);
-        }
-        else {
-            // the first value we got
-            g->level = g->trend = value;
+        if(unlikely(g->count == 1)) {
+            // the second value we got
+            g->trend = value - g->trend;
+            g->level = value;
         }
 
-        g->count++;
+        // for the values, except the first
+        calculated_number last_level = g->level;
+        g->level = (g->alpha * value) + (g->alpha_other * (g->level + g->trend));
+        g->trend = (g->beta * (g->level - last_level)) + (g->beta_other * g->trend);
     }
+    else {
+        // the first value we got
+        g->level = g->trend = value;
+    }
+
+    g->count++;
 
     //fprintf(stderr, "value: " CALCULATED_NUMBER_FORMAT ", level: " CALCULATED_NUMBER_FORMAT ", trend: " CALCULATED_NUMBER_FORMAT "\n", value, g->level, g->trend);
 }

web/api/queries/des/des.h

@@ -8,7 +8,7 @@
 
 extern void grouping_init_des(void);
 
-extern void *grouping_create_des(RRDR *r);
+extern void grouping_create_des(RRDR *r);
 extern void grouping_reset_des(RRDR *r);
 extern void grouping_free_des(RRDR *r);
 extern void grouping_add_des(RRDR *r, calculated_number value);

web/api/queries/incremental_sum/incremental_sum.c

@@ -11,9 +11,8 @@ struct grouping_incremental_sum {
     size_t count;
 };
 
-void *grouping_create_incremental_sum(RRDR *r) {
-    (void)r;
-    return callocz(1, sizeof(struct grouping_incremental_sum));
+void grouping_create_incremental_sum(RRDR *r) {
+    r->internal.grouping_data = callocz(1, sizeof(struct grouping_incremental_sum));
 }
 
 // resets when switches dimensions
@@ -31,17 +30,15 @@ void grouping_free_incremental_sum(RRDR *r) {
 }
 
 void grouping_add_incremental_sum(RRDR *r, calculated_number value) {
-    if(!isnan(value)) {
-        struct grouping_incremental_sum *g = (struct grouping_incremental_sum *)r->internal.grouping_data;
+    struct grouping_incremental_sum *g = (struct grouping_incremental_sum *)r->internal.grouping_data;
 
-        if(unlikely(!g->count)) {
-            g->first = value;
-            g->count++;
-        }
-        else {
-            g->last = value;
-            g->count++;
-        }
+    if(unlikely(!g->count)) {
+        g->first = value;
+        g->count++;
+    }
+    else {
+        g->last = value;
+        g->count++;
     }
 }
 

web/api/queries/incremental_sum/incremental_sum.h

@@ -6,7 +6,7 @@
 #include "../query.h"
 #include "../rrdr.h"
 
-extern void *grouping_create_incremental_sum(RRDR *r);
+extern void grouping_create_incremental_sum(RRDR *r);
 extern void grouping_reset_incremental_sum(RRDR *r);
 extern void grouping_free_incremental_sum(RRDR *r);
 extern void grouping_add_incremental_sum(RRDR *r, calculated_number value);