fix numa plugin to work only on the expected values; fixes #1868

[netdata.git] / src / rrdset.c

diff --git a/src/rrdset.c b/src/rrdset.c
index d16bb259a69a29f137e72212f643ffe3cc77efce..13d6014e8abf5af92c54a6dd479c29551162ec1a 100644 (file)
--- a/src/rrdset.c
+++ b/src/rrdset.c
@@ -187,7 +187,7 @@ void rrdset_reset(RRDSET *st) {
     rrddim_foreach_read(rd, st) {
         rd->last_collected_time.tv_sec = 0;
         rd->last_collected_time.tv_usec = 0;
-        rd->counter = 0;
+        rd->collections_counter = 0;
         memset(rd->values, 0, rd->entries * sizeof(storage_number));
     }
 }
@@ -195,16 +195,16 @@ void rrdset_reset(RRDSET *st) {
 // ----------------------------------------------------------------------------
 // RRDSET - helpers for rrdset_create()
 
-inline long align_entries_to_pagesize(long entries) {
-    if(rrdpush_exclusive)
-        return entries;
+inline long align_entries_to_pagesize(RRD_MEMORY_MODE mode, long entries) {
+    if(unlikely(entries < 5)) entries = 5;
+    if(unlikely(entries > RRD_HISTORY_ENTRIES_MAX)) entries = RRD_HISTORY_ENTRIES_MAX;
 
-    if(entries < 5) entries = 5;
-    if(entries > RRD_HISTORY_ENTRIES_MAX) entries = RRD_HISTORY_ENTRIES_MAX;
+    if(unlikely(mode == RRD_MEMORY_MODE_NONE || mode == RRD_MEMORY_MODE_RAM))
+        return entries;
 
     long page = (size_t)sysconf(_SC_PAGESIZE);
     long size = sizeof(RRDDIM) + entries * sizeof(storage_number);
-    if(size % page) {
+    if(unlikely(size % page)) {
         size -= (size % page);
         size += page;
 
@@ -215,11 +215,16 @@ inline long align_entries_to_pagesize(long entries) {
     return entries;
 }
 
-static inline void timeval_align(struct timeval *tv, int update_every) {
+static inline void last_collected_time_align(struct timeval *tv, int update_every) {
     tv->tv_sec -= tv->tv_sec % update_every;
     tv->tv_usec = 500000;
 }
 
+static inline void last_updated_time_align(struct timeval *tv, int update_every) {
+    tv->tv_sec -= tv->tv_sec % update_every;
+    tv->tv_usec = 0;
+}
+
 // ----------------------------------------------------------------------------
 // RRDSET - free a chart
 
@@ -322,7 +327,7 @@ RRDSET *rrdset_create(RRDHOST *host, const char *type, const char *id, const cha
     // get the options from the config, we need to create it
 
     long rentries = config_get_number(config_section, "history", host->rrd_history_entries);
-    long entries = align_entries_to_pagesize(rentries);
+    long entries = align_entries_to_pagesize(host->rrd_memory_mode, rentries);
     if(entries != rentries) entries = config_set_number(config_section, "history", entries);
 
     if(host->rrd_memory_mode == RRD_MEMORY_MODE_NONE && entries != rentries)
@@ -398,7 +403,8 @@ RRDSET *rrdset_create(RRDHOST *host, const char *type, const char *id, const cha
 
             // make sure the database is aligned
             if(st->last_updated.tv_sec)
-                timeval_align(&st->last_updated, update_every);
+                last_updated_time_align(&st->last_updated, update_every);
+
 
             // make sure we have the right memory mode
             // even if we cleared the memory
@@ -543,15 +549,17 @@ inline void rrdset_next_usec(RRDSET *st, usec_t microseconds) {
 
         if(unlikely(since_last_usec < 0)) {
             // oops! the database is in the future
-            error("Database for chart '%s' on host '%s' is %lld microseconds in the future.", st->id, st->rrdhost->hostname, -since_last_usec);
-            memcpy(&st->last_collected_time, &now, sizeof(struct timeval));
-            st->last_collected_time.tv_sec -= st->update_every;
+            error("Database for chart '%s' on host '%s' is %lld microseconds in the future. Adjusting it to current time.", st->id, st->rrdhost->hostname, -since_last_usec);
+
+            st->last_collected_time.tv_sec  = now.tv_sec - st->update_every;
+            st->last_collected_time.tv_usec = now.tv_usec;
+            last_collected_time_align(&st->last_collected_time, st->update_every);
 
-            memcpy(&st->last_updated, &now, sizeof(struct timeval));
-            timeval_align(&st->last_updated, st->update_every);
-            st->last_updated.tv_sec -= st->update_every;
+            st->last_updated.tv_sec  = now.tv_sec - st->update_every;
+            st->last_updated.tv_usec = now.tv_usec;
+            last_updated_time_align(&st->last_updated, st->update_every);
 
-            microseconds = st->update_every * USEC_PER_SEC;
+            microseconds    = st->update_every * USEC_PER_SEC;
             since_last_usec = st->update_every * USEC_PER_SEC;
         }
 
@@ -589,7 +597,7 @@ inline void rrdset_next_usec(RRDSET *st, usec_t microseconds) {
 
 static inline void rrdset_init_last_collected_time(RRDSET *st) {
     now_realtime_timeval(&st->last_collected_time);
-    timeval_align(&st->last_collected_time, st->update_every);
+    last_collected_time_align(&st->last_collected_time, st->update_every);
 }
 
 static inline usec_t rrdset_update_last_collected_time(RRDSET *st) {
@@ -602,12 +610,12 @@ static inline usec_t rrdset_update_last_collected_time(RRDSET *st) {
 
 static inline void rrdset_init_last_updated_time(RRDSET *st) {
     // copy the last collected time to last updated time
-    memcpy(&st->last_updated, &st->last_collected_time, sizeof(struct timeval));
-    timeval_align(&st->last_updated, st->update_every);
-    st->last_updated.tv_usec = 0;
+    st->last_updated.tv_sec  = st->last_collected_time.tv_sec;
+    st->last_updated.tv_usec = st->last_collected_time.tv_usec;
+    last_updated_time_align(&st->last_updated, st->update_every);
 }
 
-static inline void rrdset_done_push_int(RRDSET *st) {
+static inline void rrdset_done_push_exclusive(RRDSET *st) {
     if(unlikely(!st->last_collected_time.tv_sec)) {
         // it is the first entry
         // set the last_collected_time to now
@@ -619,17 +627,20 @@ static inline void rrdset_done_push_int(RRDSET *st) {
         rrdset_update_last_collected_time(st);
     }
 
-    st->counter++;
     st->counter_done++;
 
+    rrdset_rdlock(st);
     rrdset_done_push(st);
+    rrdset_unlock(st);
 }
 
 void rrdset_done(RRDSET *st) {
     if(unlikely(netdata_exit)) return;
 
-    if(unlikely(rrdpush_exclusive)) {
-        rrdset_done_push_int(st);
+    if(unlikely(st->rrd_memory_mode == RRD_MEMORY_MODE_NONE)) {
+        if(unlikely(st->rrdhost->rrdpush_enabled))
+            rrdset_done_push_exclusive(st);
+
         return;
     }
 
@@ -748,12 +759,15 @@ void rrdset_done(RRDSET *st) {
     }
     st->counter_done++;
 
+    if(unlikely(st->rrdhost->rrdpush_enabled))
+        rrdset_done_push(st);
+
     // calculate totals and count the dimensions
     int dimensions = 0;
     st->collected_total = 0;
     rrddim_foreach_read(rd, st) {
         dimensions++;
-        if(likely(rrddim_flag_check(rd, RRDDIM_FLAG_UPDATED)))
+        if(likely(rd->updated))
             st->collected_total += rd->collected_value;
     }
 
@@ -764,7 +778,7 @@ void rrdset_done(RRDSET *st) {
     // at this stage we do not interpolate anything
     rrddim_foreach_read(rd, st) {
 
-        if(unlikely(!rrddim_flag_check(rd, RRDDIM_FLAG_UPDATED))) {
+        if(unlikely(!rd->updated)) {
             rd->calculated_value = 0;
             continue;
         }
@@ -826,7 +840,7 @@ void rrdset_done(RRDSET *st) {
                 break;
 
             case RRD_ALGORITHM_INCREMENTAL:
-                if(unlikely(rd->counter <= 1)) {
+                if(unlikely(rd->collections_counter <= 1)) {
                     rd->calculated_value = 0;
                     continue;
                 }
@@ -866,7 +880,7 @@ void rrdset_done(RRDSET *st) {
                 break;
 
             case RRD_ALGORITHM_PCENT_OVER_DIFF_TOTAL:
-                if(unlikely(rd->counter <= 1)) {
+                if(unlikely(rd->collections_counter <= 1)) {
                     rd->calculated_value = 0;
                     continue;
                 }
@@ -1050,7 +1064,7 @@ void rrdset_done(RRDSET *st) {
                 continue;
             }
 
-            if(likely(rrddim_flag_check(rd, RRDDIM_FLAG_UPDATED) && rd->counter > 1 && iterations < st->gap_when_lost_iterations_above)) {
+            if(likely(rd->updated && rd->collections_counter > 1 && iterations < st->gap_when_lost_iterations_above)) {
                 rd->values[st->current_entry] = pack_storage_number(new_value, storage_flags );
                 rd->last_stored_value = new_value;
 
@@ -1113,7 +1127,7 @@ void rrdset_done(RRDSET *st) {
     st->last_collected_total  = st->collected_total;
 
     rrddim_foreach_read(rd, st) {
-        if(unlikely(!rrddim_flag_check(rd, RRDDIM_FLAG_UPDATED)))
+        if(unlikely(!rd->updated))
             continue;
 
         if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
@@ -1145,7 +1159,7 @@ void rrdset_done(RRDSET *st) {
 
         rd->calculated_value = 0;
         rd->collected_value = 0;
-        rrddim_flag_clear(rd, RRDDIM_FLAG_UPDATED);
+        rd->updated = 0;
 
         if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
             debug(D_RRD_STATS, "%s/%s: END "
@@ -1217,8 +1231,4 @@ void rrdset_done(RRDSET *st) {
 
     if(unlikely(pthread_setcancelstate(pthreadoldcancelstate, NULL) != 0))
         error("Cannot set pthread cancel state to RESTORE (%d).", pthreadoldcancelstate);
-
-    if(unlikely(rrdpush_enabled))
-        rrdset_done_push_int(st);
 }
-