properly initialize stream receiving thread

[netdata.git] / src / rrdset.c

diff --git a/src/rrdset.c b/src/rrdset.c
index 3432788b941c80200c18f69f9d03eb17e71a0dfb..f88ce81804b1ceec2536ec36febb6419323fb6d0 100644 (file)
--- a/src/rrdset.c
+++ b/src/rrdset.c
@@ -196,7 +196,7 @@ void rrdset_reset(RRDSET *st) {
 // RRDSET - helpers for rrdset_create()
 
 inline long align_entries_to_pagesize(long entries) {
-    if(central_netdata_to_push_data)
+    if(rrdpush_exclusive)
         return entries;
 
     if(entries < 5) entries = 5;
@@ -529,26 +529,40 @@ inline void rrdset_next_usec(RRDSET *st, usec_t microseconds) {
 //        usec_t now_usec = timeval_usec(&now);
 //        usec_t last_usec = timeval_usec(&st->last_collected_time);
 //#endif
-        usec_t since_last_usec = dt_usec(&now, &st->last_collected_time);
+        susec_t since_last_usec = dt_usec_signed(&now, &st->last_collected_time);
+
+        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;
+
+            memcpy(&st->last_updated, &now, sizeof(struct timeval));
+            timeval_align(&st->last_updated, st->update_every);
+            st->last_updated.tv_sec -= st->update_every;
+
+            microseconds = st->update_every * USEC_PER_SEC;
+            since_last_usec = st->update_every * USEC_PER_SEC;
+        }
 
         // verify the microseconds given is good
-        if(unlikely(microseconds > since_last_usec)) {
-            debug(D_RRD_CALLS, "dt %llu usec given is too big - it leads %llu usec to the future, for chart '%s' (%s).", microseconds, microseconds - since_last_usec, st->name, st->id);
+        if(unlikely(microseconds > (usec_t)since_last_usec)) {
+            debug(D_RRD_CALLS, "dt %llu usec given is too big - it leads %llu usec to the future, for chart '%s' (%s).", microseconds, microseconds - (usec_t)since_last_usec, st->name, st->id);
 
 //#ifdef NETDATA_INTERNAL_CHECKS
 //            if(unlikely(last_usec + microseconds > now_usec + 1000))
-//                error("dt %llu usec given is too big - it leads %llu usec to the future, for chart '%s' (%s).", microseconds, microseconds - since_last_usec, st->name, st->id);
+//                error("dt %llu usec given is too big - it leads %llu usec to the future, for chart '%s' (%s).", microseconds, microseconds - (usec_t)since_last_usec, st->name, st->id);
 //#endif
 
-            microseconds = since_last_usec;
+            microseconds = (usec_t)since_last_usec;
         }
-        else if(unlikely(microseconds < since_last_usec * 0.8)) {
-            debug(D_RRD_CALLS, "dt %llu usec given is too small - expected %llu usec up to -20%%, for chart '%s' (%s).", microseconds, since_last_usec, st->name, st->id);
+        else if(unlikely(microseconds < (usec_t)since_last_usec * 0.8)) {
+            debug(D_RRD_CALLS, "dt %llu usec given is too small - expected %llu usec up to -20%%, for chart '%s' (%s).", microseconds, (usec_t)since_last_usec, st->name, st->id);
 
 //#ifdef NETDATA_INTERNAL_CHECKS
-//            error("dt %llu usec given is too small - expected %llu usec up to -20%%, for chart '%s' (%s).", microseconds, since_last_usec, st->name, st->id);
+//            error("dt %llu usec given is too small - expected %llu usec up to -20%%, for chart '%s' (%s).", microseconds, (usec_t)since_last_usec, st->name, st->id);
 //#endif
-            microseconds = since_last_usec;
+            microseconds = (usec_t)since_last_usec;
         }
     }
     debug(D_RRD_CALLS, "rrdset_next_usec() for chart %s with microseconds %llu", st->name, microseconds);
@@ -595,13 +609,16 @@ static inline void rrdset_done_push_int(RRDSET *st) {
         rrdset_update_last_collected_time(st);
     }
 
+    st->counter++;
+    st->counter_done++;
+
     rrdset_done_push(st);
 }
 
 void rrdset_done(RRDSET *st) {
     if(unlikely(netdata_exit)) return;
 
-    if(unlikely(central_netdata_to_push_data)) {
+    if(unlikely(rrdpush_exclusive)) {
         rrdset_done_push_int(st);
         return;
     }
@@ -1190,5 +1207,8 @@ 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);
 }