1 files changed, 42 insertions, 12 deletions

diff --git a/database/ram/rrddim_mem.c b/database/ram/rrddim_mem.c
index b17f03ca50..03b6983a29 100644
--- a/database/ram/rrddim_mem.c
+++ b/database/ram/rrddim_mem.c
@@ -9,9 +9,9 @@ void rrddim_collect_init(RRDDIM *rd) {
     rd->values[rd->rrdset->current_entry] = SN_EMPTY_SLOT;
     rd->state->handle = calloc(1, sizeof(struct mem_collect_handle));
 }
-void rrddim_collect_store_metric(RRDDIM *rd, usec_t point_in_time, storage_number number) {
+void rrddim_collect_store_metric(RRDDIM *rd, usec_t point_in_time, calculated_number number, SN_FLAGS flags) {
     (void)point_in_time;
-    rd->values[rd->rrdset->current_entry] = number;
+    rd->values[rd->rrdset->current_entry] = pack_storage_number(number, flags);
 }
 int rrddim_collect_finalize(RRDDIM *rd) {
     free((struct mem_collect_handle*)rd->state->handle);
@@ -28,33 +28,63 @@ void rrddim_query_init(RRDDIM *rd, struct rrddim_query_handle *handle, time_t st
     struct mem_query_handle* h = calloc(1, sizeof(struct mem_query_handle));
     h->slot = rrdset_time2slot(rd->rrdset, start_time);
     h->last_slot = rrdset_time2slot(rd->rrdset, end_time);
-    h->finished = 0;
+    h->dt = rd->update_every;
+
+    h->next_timestamp = start_time;
+    h->slot_timestamp = rrdset_slot2time(rd->rrdset, h->slot);
+    h->last_timestamp = rrdset_slot2time(rd->rrdset, h->last_slot);
+
+    // info("QUERY: start %ld, end %ld, next %ld, first %ld, last %ld", start_time, end_time, h->next_timestamp, h->slot_timestamp, h->last_timestamp);
+
     handle->handle = (STORAGE_QUERY_HANDLE *)h;
 }
 
-storage_number rrddim_query_next_metric(struct rrddim_query_handle *handle, time_t *current_time) {
+// Returns the metric and sets its timestamp into current_time
+// IT IS REQUIRED TO **ALWAYS** SET ALL RETURN VALUES (current_time, end_time, flags)
+// IT IS REQUIRED TO **ALWAYS** KEEP TRACK OF TIME, EVEN OUTSIDE THE DATABASE BOUNDARIES
+calculated_number rrddim_query_next_metric(struct rrddim_query_handle *handle, time_t *start_time, time_t *end_time, SN_FLAGS *flags) {
     RRDDIM *rd = handle->rd;
     struct mem_query_handle* h = (struct mem_query_handle*)handle->handle;
-    long entries = rd->rrdset->entries;
-    long slot = h->slot;
+    size_t entries = rd->rrdset->entries;
+    size_t slot = h->slot;
 
-    (void)current_time;
-    if (unlikely(h->slot == h->last_slot))
-        h->finished = 1;
-    storage_number n = rd->values[slot++];
+    time_t this_timestamp = h->next_timestamp;
+    h->next_timestamp += h->dt;
+
+    // set this timestamp for our caller
+    *start_time = this_timestamp - h->dt;
+    *end_time = this_timestamp;
 
+    if(unlikely(this_timestamp < h->slot_timestamp)) {
+        *flags = SN_EMPTY_SLOT;
+        return NAN;
+    }
+
+    if(unlikely(this_timestamp > h->last_timestamp)) {
+        *flags = SN_EMPTY_SLOT;
+        return NAN;
+    }
+
+    storage_number n = rd->values[slot++];
     if(unlikely(slot >= entries)) slot = 0;
+
     h->slot = slot;
+    h->slot_timestamp += h->dt;
 
-    return n;
+    *flags = (n & SN_ALL_FLAGS);
+    return unpack_storage_number(n);
 }
 
 int rrddim_query_is_finished(struct rrddim_query_handle *handle) {
     struct mem_query_handle* h = (struct mem_query_handle*)handle->handle;
-    return h->finished;
+    return (h->next_timestamp > handle->end_time);
 }
 
 void rrddim_query_finalize(struct rrddim_query_handle *handle) {
+#ifdef NETDATA_INTERNAL_CHECKS
+    if(!rrddim_query_is_finished(handle))
+        error("QUERY: query for chart '%s' dimension '%s' has been stopped unfinished", handle->rd->rrdset->id, handle->rd->name);
+#endif
     freez(handle->handle);
 }