src/rrdset.c

   1 #define NETDATA_RRD_INTERNALS 1
   2 #include "common.h"
   3
   4 #define RRD_DEFAULT_GAP_INTERPOLATIONS 1
   5
   6 // ----------------------------------------------------------------------------
   7 // RRDSET index
   8
   9 int rrdset_compare(void* a, void* b) {
  10     if(((RRDSET *)a)->hash < ((RRDSET *)b)->hash) return -1;
  11     else if(((RRDSET *)a)->hash > ((RRDSET *)b)->hash) return 1;
  12     else return strcmp(((RRDSET *)a)->id, ((RRDSET *)b)->id);
  13 }
  14
  15 static RRDSET *rrdset_index_find(RRDHOST *host, const char *id, uint32_t hash) {
  16     RRDSET tmp;
  17     strncpyz(tmp.id, id, RRD_ID_LENGTH_MAX);
  18     tmp.hash = (hash)?hash:simple_hash(tmp.id);
  19
  20     return (RRDSET *)avl_search_lock(&(host->rrdset_root_index), (avl *) &tmp);
  21 }
  22
  23 // ----------------------------------------------------------------------------
  24 // RRDSET name index
  25
  26 #define rrdset_from_avlname(avlname_ptr) ((RRDSET *)((avlname_ptr) - offsetof(RRDSET, avlname)))
  27
  28 int rrdset_compare_name(void* a, void* b) {
  29     RRDSET *A = rrdset_from_avlname(a);
  30     RRDSET *B = rrdset_from_avlname(b);
  31
  32     // fprintf(stderr, "COMPARING: %s with %s\n", A->name, B->name);
  33
  34     if(A->hash_name < B->hash_name) return -1;
  35     else if(A->hash_name > B->hash_name) return 1;
  36     else return strcmp(A->name, B->name);
  37 }
  38
  39 RRDSET *rrdset_index_add_name(RRDHOST *host, RRDSET *st) {
  40     void *result;
  41     // fprintf(stderr, "ADDING: %s (name: %s)\n", st->id, st->name);
  42     result = avl_insert_lock(&host->rrdset_root_index_name, (avl *) (&st->avlname));
  43     if(result) return rrdset_from_avlname(result);
  44     return NULL;
  45 }
  46
  47 RRDSET *rrdset_index_del_name(RRDHOST *host, RRDSET *st) {
  48     void *result;
  49     // fprintf(stderr, "DELETING: %s (name: %s)\n", st->id, st->name);
  50     result = (RRDSET *)avl_remove_lock(&((host)->rrdset_root_index_name), (avl *)(&st->avlname));
  51     if(result) return rrdset_from_avlname(result);
  52     return NULL;
  53 }
  54
  55
  56 // ----------------------------------------------------------------------------
  57 // RRDSET - find charts
  58
  59 static inline RRDSET *rrdset_index_find_name(RRDHOST *host, const char *name, uint32_t hash) {
  60     void *result = NULL;
  61     RRDSET tmp;
  62     tmp.name = name;
  63     tmp.hash_name = (hash)?hash:simple_hash(tmp.name);
  64
  65     // fprintf(stderr, "SEARCHING: %s\n", name);
  66     result = avl_search_lock(&host->rrdset_root_index_name, (avl *) (&(tmp.avlname)));
  67     if(result) {
  68         RRDSET *st = rrdset_from_avlname(result);
  69         if(strcmp(st->magic, RRDSET_MAGIC))
  70             error("Search for RRDSET %s returned an invalid RRDSET %s (name %s)", name, st->id, st->name);
  71
  72         // fprintf(stderr, "FOUND: %s\n", name);
  73         return rrdset_from_avlname(result);
  74     }
  75     // fprintf(stderr, "NOT FOUND: %s\n", name);
  76     return NULL;
  77 }
  78
  79 inline RRDSET *rrdset_find(RRDHOST *host, const char *id) {
  80     debug(D_RRD_CALLS, "rrdset_find() for chart '%s' in host '%s'", id, host->hostname);
  81     RRDSET *st = rrdset_index_find(host, id, 0);
  82     return(st);
  83 }
  84
  85 inline RRDSET *rrdset_find_bytype(RRDHOST *host, const char *type, const char *id) {
  86     debug(D_RRD_CALLS, "rrdset_find_bytype() for chart '%s.%s' in host '%s'", type, id, host->hostname);
  87
  88     char buf[RRD_ID_LENGTH_MAX + 1];
  89     strncpyz(buf, type, RRD_ID_LENGTH_MAX - 1);
  90     strcat(buf, ".");
  91     int len = (int) strlen(buf);
  92     strncpyz(&buf[len], id, (size_t) (RRD_ID_LENGTH_MAX - len));
  93
  94     return(rrdset_find(host, buf));
  95 }
  96
  97 inline RRDSET *rrdset_find_byname(RRDHOST *host, const char *name) {
  98     debug(D_RRD_CALLS, "rrdset_find_byname() for chart '%s' in host '%s'", name, host->hostname);
  99     RRDSET *st = rrdset_index_find_name(host, name, 0);
 100     return(st);
 101 }
 102
 103
 104 // ----------------------------------------------------------------------------
 105 // RRDSET - rename charts
 106
 107 char *rrdset_strncpyz_name(char *to, const char *from, size_t length) {
 108     char c, *p = to;
 109
 110     while (length-- && (c = *from++)) {
 111         if(c != '.' && !isalnum(c))
 112             c = '_';
 113
 114         *p++ = c;
 115     }
 116
 117     *p = '\0';
 118
 119     return to;
 120 }
 121
 122 void rrdset_set_name(RRDSET *st, const char *name) {
 123     if(unlikely(st->name && !strcmp(st->name, name)))
 124         return;
 125
 126     debug(D_RRD_CALLS, "rrdset_set_name() old: %s, new: %s", st->name, name);
 127
 128     char b[CONFIG_MAX_VALUE + 1];
 129     char n[RRD_ID_LENGTH_MAX + 1];
 130
 131     snprintfz(n, RRD_ID_LENGTH_MAX, "%s.%s", st->type, name);
 132     rrdset_strncpyz_name(b, n, CONFIG_MAX_VALUE);
 133
 134     if(st->name) {
 135         rrdset_index_del_name(st->rrdhost, st);
 136         st->name = config_set_default(st->config_section, "name", b);
 137         st->hash_name = simple_hash(st->name);
 138         rrdsetvar_rename_all(st);
 139     }
 140     else {
 141         st->name = config_get(st->config_section, "name", b);
 142         st->hash_name = simple_hash(st->name);
 143     }
 144
 145     pthread_rwlock_wrlock(&st->rwlock);
 146     RRDDIM *rd;
 147     for(rd = st->dimensions; rd ;rd = rd->next)
 148         rrddimvar_rename_all(rd);
 149     pthread_rwlock_unlock(&st->rwlock);
 150
 151     if(unlikely(rrdset_index_add_name(st->rrdhost, st) != st))
 152         error("RRDSET: INTERNAL ERROR: attempted to index duplicate chart name '%s'", st->name);
 153 }
 154
 155
 156 // ----------------------------------------------------------------------------
 157 // RRDSET - reset a chart
 158
 159 void rrdset_reset(RRDSET *st) {
 160     debug(D_RRD_CALLS, "rrdset_reset() %s", st->name);
 161
 162     st->last_collected_time.tv_sec = 0;
 163     st->last_collected_time.tv_usec = 0;
 164     st->last_updated.tv_sec = 0;
 165     st->last_updated.tv_usec = 0;
 166     st->current_entry = 0;
 167     st->counter = 0;
 168     st->counter_done = 0;
 169
 170     RRDDIM *rd;
 171     for(rd = st->dimensions; rd ; rd = rd->next) {
 172         rd->last_collected_time.tv_sec = 0;
 173         rd->last_collected_time.tv_usec = 0;
 174         rd->counter = 0;
 175         memset(rd->values, 0, rd->entries * sizeof(storage_number));
 176     }
 177 }
 178
 179 // ----------------------------------------------------------------------------
 180 // RRDSET - helpers for rrdset_create()
 181
 182 static inline long align_entries_to_pagesize(long entries) {
 183     if(entries < 5) entries = 5;
 184     if(entries > RRD_HISTORY_ENTRIES_MAX) entries = RRD_HISTORY_ENTRIES_MAX;
 185
 186 #ifdef NETDATA_LOG_ALLOCATIONS
 187     long page = (size_t)sysconf(_SC_PAGESIZE);
 188
 189     long size = sizeof(RRDDIM) + entries * sizeof(storage_number);
 190     if(size % page) {
 191         size -= (size % page);
 192         size += page;
 193
 194         long n = (size - sizeof(RRDDIM)) / sizeof(storage_number);
 195         return n;
 196     }
 197
 198     return entries;
 199 #else
 200     return entries;
 201 #endif
 202 }
 203
 204 static inline void timeval_align(struct timeval *tv, int update_every) {
 205     tv->tv_sec -= tv->tv_sec % update_every;
 206     tv->tv_usec = 500000;
 207 }
 208
 209 // ----------------------------------------------------------------------------
 210 // RRDSET - free a chart
 211
 212 void rrdset_free(RRDSET *st) {
 213     pthread_rwlock_wrlock(&st->rwlock);
 214
 215     while(st->variables)  rrdsetvar_free(st->variables);
 216     while(st->alarms)     rrdsetcalc_unlink(st->alarms);
 217     while(st->dimensions) rrddim_free(st, st->dimensions);
 218
 219     if(unlikely(rrdset_index_del(st->rrdhost, st) != st))
 220         error("RRDSET: INTERNAL ERROR: attempt to remove from index chart '%s', removed a different chart.", st->id);
 221
 222     rrdset_index_del_name(st->rrdhost, st);
 223
 224     st->rrdfamily->use_count--;
 225     if(!st->rrdfamily->use_count)
 226         rrdfamily_free(st->rrdhost, st->rrdfamily);
 227
 228     pthread_rwlock_unlock(&st->rwlock);
 229
 230     // free directly allocated memory
 231     freez(st->config_section);
 232
 233     if(st->rrd_memory_mode == RRD_MEMORY_MODE_SAVE || st->rrd_memory_mode == RRD_MEMORY_MODE_MAP) {
 234         debug(D_RRD_CALLS, "Unmapping stats '%s'.", st->name);
 235         munmap(st, st->memsize);
 236     }
 237     else
 238         freez(st);
 239 }
 240
 241 // ----------------------------------------------------------------------------
 242 // RRDSET - create a chart
 243
 244 RRDSET *rrdset_create(RRDHOST *host, const char *type, const char *id, const char *name, const char *family
 245                       , const char *context, const char *title, const char *units, long priority
 246                       , int update_every, RRDSET_TYPE chart_type) {
 247
 248     if(!type || !type[0]) {
 249         fatal("Cannot create rrd stats without a type.");
 250         return NULL;
 251     }
 252
 253     if(!id || !id[0]) {
 254         fatal("Cannot create rrd stats without an id.");
 255         return NULL;
 256     }
 257
 258     // ------------------------------------------------------------------------
 259     // check if it already exists
 260
 261     char fullid[RRD_ID_LENGTH_MAX + 1];
 262     snprintfz(fullid, RRD_ID_LENGTH_MAX, "%s.%s", type, id);
 263
 264     RRDSET *st = rrdset_find(host, fullid);
 265     if(st) {
 266         debug(D_RRD_CALLS, "RRDSET '%s', already exists.", fullid);
 267         return st;
 268     }
 269
 270     char fullfilename[FILENAME_MAX + 1];
 271
 272     // ------------------------------------------------------------------------
 273     // compose the config_section for this chart
 274
 275     char config_section[RRD_ID_LENGTH_MAX + 1];
 276     if(host == localhost)
 277         strcpy(config_section, fullid);
 278     else
 279         snprintfz(config_section, RRD_ID_LENGTH_MAX, "%s/%s", host->machine_guid, fullid);
 280
 281     // ------------------------------------------------------------------------
 282     // get the options from the config, we need to create it
 283
 284     long rentries = config_get_number(config_section, "history", host->rrd_history_entries);
 285     long entries = align_entries_to_pagesize(rentries);
 286     if(entries != rentries) entries = config_set_number(config_section, "history", entries);
 287
 288     int enabled = config_get_boolean(config_section, "enabled", 1);
 289     if(!enabled) entries = 5;
 290
 291     unsigned long size = sizeof(RRDSET);
 292     char *cache_dir = rrdset_cache_dir(host, fullid, config_section);
 293
 294     // ------------------------------------------------------------------------
 295     // load it or allocate it
 296
 297     debug(D_RRD_CALLS, "Creating RRD_STATS for '%s.%s'.", type, id);
 298
 299     snprintfz(fullfilename, FILENAME_MAX, "%s/main.db", cache_dir);
 300     if(host->rrd_memory_mode != RRD_MEMORY_MODE_RAM) {
 301         st = (RRDSET *) mymmap(fullfilename, size, ((host->rrd_memory_mode == RRD_MEMORY_MODE_MAP) ? MAP_SHARED : MAP_PRIVATE), 0);
 302         if(st) {
 303             memset(&st->avl, 0, sizeof(avl));
 304             memset(&st->avlname, 0, sizeof(avl));
 305             memset(&st->variables_root_index, 0, sizeof(avl_tree_lock));
 306             memset(&st->dimensions_index, 0, sizeof(avl_tree_lock));
 307             memset(&st->rwlock, 0, sizeof(pthread_rwlock_t));
 308
 309             st->name = NULL;
 310             st->type = NULL;
 311             st->family = NULL;
 312             st->context = NULL;
 313             st->title = NULL;
 314             st->units = NULL;
 315             st->dimensions = NULL;
 316             st->next = NULL;
 317             st->variables = NULL;
 318             st->alarms = NULL;
 319             st->flags = 0x00000000;
 320
 321             if(strcmp(st->magic, RRDSET_MAGIC) != 0) {
 322                 errno = 0;
 323                 info("Initializing file %s.", fullfilename);
 324                 memset(st, 0, size);
 325             }
 326             else if(strcmp(st->id, fullid) != 0) {
 327                 errno = 0;
 328                 error("File %s contents are not for chart %s. Clearing it.", fullfilename, fullid);
 329                 // munmap(st, size);
 330                 // st = NULL;
 331                 memset(st, 0, size);
 332             }
 333             else if(st->memsize != size || st->entries != entries) {
 334                 errno = 0;
 335                 error("File %s does not have the desired size. Clearing it.", fullfilename);
 336                 memset(st, 0, size);
 337             }
 338             else if(st->update_every != update_every) {
 339                 errno = 0;
 340                 error("File %s does not have the desired update frequency. Clearing it.", fullfilename);
 341                 memset(st, 0, size);
 342             }
 343             else if((now_realtime_sec() - st->last_updated.tv_sec) > update_every * entries) {
 344                 errno = 0;
 345                 error("File %s is too old. Clearing it.", fullfilename);
 346                 memset(st, 0, size);
 347             }
 348
 349             // make sure the database is aligned
 350             if(st->last_updated.tv_sec)
 351                 timeval_align(&st->last_updated, update_every);
 352
 353             // make sure we have the right memory mode
 354             // even if we cleared the memory
 355             st->rrd_memory_mode = host->rrd_memory_mode;
 356         }
 357     }
 358
 359     if(unlikely(!st)) {
 360         st = callocz(1, size);
 361         st->rrd_memory_mode = RRD_MEMORY_MODE_RAM;
 362     }
 363
 364     st->config_section = strdup(config_section);
 365     st->rrdhost = host;
 366     st->memsize = size;
 367     st->entries = entries;
 368     st->update_every = update_every;
 369
 370     if(st->current_entry >= st->entries) st->current_entry = 0;
 371
 372     strcpy(st->cache_filename, fullfilename);
 373     strcpy(st->magic, RRDSET_MAGIC);
 374
 375     strcpy(st->id, fullid);
 376     st->hash = simple_hash(st->id);
 377
 378     st->cache_dir = cache_dir;
 379
 380     st->chart_type = rrdset_type_id(config_get(st->config_section, "chart type", rrdset_type_name(chart_type)));
 381     st->type       = config_get(st->config_section, "type", type);
 382     st->family     = config_get(st->config_section, "family", family?family:st->type);
 383     st->units      = config_get(st->config_section, "units", units?units:"");
 384
 385     st->context    = config_get(st->config_section, "context", context?context:st->id);
 386     st->hash_context = simple_hash(st->context);
 387
 388     st->priority = config_get_number(st->config_section, "priority", priority);
 389     if(enabled)
 390         rrdset_flag_set(st, RRDSET_FLAG_ENABLED);
 391     else
 392         rrdset_flag_clear(st, RRDSET_FLAG_ENABLED);
 393
 394     rrdset_flag_clear(st, RRDSET_FLAG_DETAIL);
 395     rrdset_flag_clear(st, RRDSET_FLAG_DEBUG);
 396
 397     // if(!strcmp(st->id, "disk_util.dm-0")) {
 398     //     st->debug = 1;
 399     //     error("enabled debugging for '%s'", st->id);
 400     // }
 401     // else error("not enabled debugging for '%s'", st->id);
 402
 403     st->green = NAN;
 404     st->red = NAN;
 405
 406     st->last_collected_time.tv_sec = 0;
 407     st->last_collected_time.tv_usec = 0;
 408     st->counter_done = 0;
 409
 410     st->gap_when_lost_iterations_above = (int) (
 411             config_get_number(st->config_section, "gap when lost iterations above", RRD_DEFAULT_GAP_INTERPOLATIONS) + 2);
 412
 413     avl_init_lock(&st->dimensions_index, rrddim_compare);
 414     avl_init_lock(&st->variables_root_index, rrdvar_compare);
 415
 416     pthread_rwlock_init(&st->rwlock, NULL);
 417     rrdhost_rwlock(host);
 418
 419     if(name && *name) rrdset_set_name(st, name);
 420     else rrdset_set_name(st, id);
 421
 422     {
 423         char varvalue[CONFIG_MAX_VALUE + 1];
 424         char varvalue2[CONFIG_MAX_VALUE + 1];
 425         snprintfz(varvalue, CONFIG_MAX_VALUE, "%s (%s)", title?title:"", st->name);
 426         json_escape_string(varvalue2, varvalue, sizeof(varvalue2));
 427         st->title = config_get(st->config_section, "title", varvalue2);
 428     }
 429
 430     st->rrdfamily = rrdfamily_create(host, st->family);
 431
 432     st->next = host->rrdset_root;
 433     host->rrdset_root = st;
 434
 435     if(host->health_enabled) {
 436         rrdsetvar_create(st, "last_collected_t", RRDVAR_TYPE_TIME_T, &st->last_collected_time.tv_sec, 0);
 437         rrdsetvar_create(st, "collected_total_raw", RRDVAR_TYPE_TOTAL, &st->last_collected_total, 0);
 438         rrdsetvar_create(st, "green", RRDVAR_TYPE_CALCULATED, &st->green, 0);
 439         rrdsetvar_create(st, "red", RRDVAR_TYPE_CALCULATED, &st->red, 0);
 440         rrdsetvar_create(st, "update_every", RRDVAR_TYPE_INT, &st->update_every, 0);
 441     }
 442
 443     if(unlikely(rrdset_index_add(host, st) != st))
 444         error("RRDSET: INTERNAL ERROR: attempt to index duplicate chart '%s'", st->id);
 445
 446     rrdsetcalc_link_matching(st);
 447     rrdcalctemplate_link_matching(st);
 448
 449     rrdhost_unlock(host);
 450
 451     return(st);
 452 }
 453
 454
 455 // ----------------------------------------------------------------------------
 456 // RRDSET - data collection iteration control
 457
 458 void rrdset_next_usec_unfiltered(RRDSET *st, usec_t microseconds) {
 459     if(unlikely(!st->last_collected_time.tv_sec || !microseconds)) {
 460         // the first entry
 461         microseconds = st->update_every * USEC_PER_SEC;
 462     }
 463     st->usec_since_last_update = microseconds;
 464 }
 465
 466 void rrdset_next_usec(RRDSET *st, usec_t microseconds)
 467 {
 468     struct timeval now;
 469     now_realtime_timeval(&now);
 470
 471     if(unlikely(!st->last_collected_time.tv_sec)) {
 472         // the first entry
 473         microseconds = st->update_every * USEC_PER_SEC;
 474     }
 475     else if(unlikely(!microseconds)) {
 476         // no dt given by the plugin
 477         microseconds = dt_usec(&now, &st->last_collected_time);
 478     }
 479     else {
 480         // microseconds has the time since the last collection
 481 #ifdef NETDATA_INTERNAL_CHECKS
 482         usec_t now_usec = timeval_usec(&now);
 483         usec_t last_usec = timeval_usec(&st->last_collected_time);
 484 #endif
 485         usec_t since_last_usec = dt_usec(&now, &st->last_collected_time);
 486
 487         // verify the microseconds given is good
 488         if(unlikely(microseconds > since_last_usec)) {
 489             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);
 490
 491 #ifdef NETDATA_INTERNAL_CHECKS
 492             if(unlikely(last_usec + microseconds > now_usec + 1000))
 493                 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);
 494 #endif
 495
 496             microseconds = since_last_usec;
 497         }
 498         else if(unlikely(microseconds < since_last_usec * 0.8)) {
 499             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);
 500
 501 #ifdef NETDATA_INTERNAL_CHECKS
 502             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);
 503 #endif
 504             microseconds = since_last_usec;
 505         }
 506     }
 507     debug(D_RRD_CALLS, "rrdset_next_usec() for chart %s with microseconds %llu", st->name, microseconds);
 508
 509     if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
 510         debug(D_RRD_STATS, "%s: NEXT: %llu microseconds", st->name, microseconds);
 511
 512     st->usec_since_last_update = microseconds;
 513 }
 514
 515
 516 // ----------------------------------------------------------------------------
 517 // RRDSET - process the collected values for all dimensions of a chart
 518
 519 usec_t rrdset_done(RRDSET *st) {
 520     if(unlikely(netdata_exit)) return 0;
 521
 522     debug(D_RRD_CALLS, "rrdset_done() for chart %s", st->name);
 523
 524     RRDDIM *rd;
 525
 526     int
 527             pthreadoldcancelstate;  // store the old cancelable pthread state, to restore it at the end
 528
 529     char
 530             store_this_entry = 1,   // boolean: 1 = store this entry, 0 = don't store this entry
 531             first_entry = 0;        // boolean: 1 = this is the first entry seen for this chart, 0 = all other entries
 532
 533     unsigned int
 534             stored_entries = 0;     // the number of entries we have stored in the db, during this call to rrdset_done()
 535
 536     usec_t
 537             last_collect_ut,        // the timestamp in microseconds, of the last collected value
 538             now_collect_ut,         // the timestamp in microseconds, of this collected value (this is NOW)
 539             last_stored_ut,         // the timestamp in microseconds, of the last stored entry in the db
 540             next_store_ut,          // the timestamp in microseconds, of the next entry to store in the db
 541             update_every_ut = st->update_every * USEC_PER_SEC; // st->update_every in microseconds
 542
 543     if(unlikely(pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &pthreadoldcancelstate) != 0))
 544         error("Cannot set pthread cancel state to DISABLE.");
 545
 546     // a read lock is OK here
 547     pthread_rwlock_rdlock(&st->rwlock);
 548
 549 /*
 550     // enable the chart, if it was disabled
 551     if(unlikely(rrd_delete_unupdated_dimensions) && !st->enabled)
 552         st->enabled = 1;
 553 */
 554
 555     // check if the chart has a long time to be updated
 556     if(unlikely(st->usec_since_last_update > st->entries * update_every_ut)) {
 557         info("%s: took too long to be updated (%0.3Lf secs). Resetting it.", st->name, (long double)(st->usec_since_last_update / 1000000.0));
 558         rrdset_reset(st);
 559         st->usec_since_last_update = update_every_ut;
 560         first_entry = 1;
 561     }
 562
 563     if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
 564         debug(D_RRD_STATS, "%s: microseconds since last update: %llu", st->name, st->usec_since_last_update);
 565
 566     // set last_collected_time
 567     if(unlikely(!st->last_collected_time.tv_sec)) {
 568         // it is the first entry
 569         // set the last_collected_time to now
 570         now_realtime_timeval(&st->last_collected_time);
 571         timeval_align(&st->last_collected_time, st->update_every);
 572
 573         last_collect_ut = st->last_collected_time.tv_sec * USEC_PER_SEC + st->last_collected_time.tv_usec - update_every_ut;
 574
 575         // the first entry should not be stored
 576         store_this_entry = 0;
 577         first_entry = 1;
 578
 579         if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
 580             debug(D_RRD_STATS, "%s: has not set last_collected_time. Setting it now. Will not store the next entry.", st->name);
 581     }
 582     else {
 583         // it is not the first entry
 584         // calculate the proper last_collected_time, using usec_since_last_update
 585         last_collect_ut = st->last_collected_time.tv_sec * USEC_PER_SEC + st->last_collected_time.tv_usec;
 586         usec_t ut = last_collect_ut + st->usec_since_last_update;
 587         st->last_collected_time.tv_sec = (time_t) (ut / USEC_PER_SEC);
 588         st->last_collected_time.tv_usec = (suseconds_t) (ut % USEC_PER_SEC);
 589     }
 590
 591     // if this set has not been updated in the past
 592     // we fake the last_update time to be = now - usec_since_last_update
 593     if(unlikely(!st->last_updated.tv_sec)) {
 594         // it has never been updated before
 595         // set a fake last_updated, in the past using usec_since_last_update
 596         usec_t ut = st->last_collected_time.tv_sec * USEC_PER_SEC + st->last_collected_time.tv_usec - st->usec_since_last_update;
 597         st->last_updated.tv_sec = (time_t) (ut / USEC_PER_SEC);
 598         st->last_updated.tv_usec = (suseconds_t) (ut % USEC_PER_SEC);
 599
 600         // the first entry should not be stored
 601         store_this_entry = 0;
 602         first_entry = 1;
 603
 604         if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
 605             debug(D_RRD_STATS, "%s: initializing last_updated to now - %llu microseconds (%0.3Lf). Will not store the next entry.", st->name, st->usec_since_last_update, (long double)ut/1000000.0);
 606     }
 607
 608     // check if we will re-write the entire data set
 609     if(unlikely(dt_usec(&st->last_collected_time, &st->last_updated) > st->entries * update_every_ut)) {
 610         info("%s: too old data (last updated at %ld.%ld, last collected at %ld.%ld). Resetting it. Will not store the next entry.", st->name, st->last_updated.tv_sec, st->last_updated.tv_usec, st->last_collected_time.tv_sec, st->last_collected_time.tv_usec);
 611         rrdset_reset(st);
 612
 613         st->usec_since_last_update = update_every_ut;
 614
 615         now_realtime_timeval(&st->last_collected_time);
 616         timeval_align(&st->last_collected_time, st->update_every);
 617
 618         usec_t ut = st->last_collected_time.tv_sec * USEC_PER_SEC + st->last_collected_time.tv_usec - st->usec_since_last_update;
 619         st->last_updated.tv_sec = (time_t) (ut / USEC_PER_SEC);
 620         st->last_updated.tv_usec = (suseconds_t) (ut % USEC_PER_SEC);
 621
 622         // the first entry should not be stored
 623         store_this_entry = 0;
 624         first_entry = 1;
 625     }
 626
 627     // these are the 3 variables that will help us in interpolation
 628     // last_stored_ut = the last time we added a value to the storage
 629     // now_collect_ut = the time the current value has been collected
 630     // next_store_ut  = the time of the next interpolation point
 631     last_stored_ut = st->last_updated.tv_sec * USEC_PER_SEC + st->last_updated.tv_usec;
 632     now_collect_ut = st->last_collected_time.tv_sec * USEC_PER_SEC + st->last_collected_time.tv_usec;
 633     next_store_ut  = (st->last_updated.tv_sec + st->update_every) * USEC_PER_SEC;
 634
 635     if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG))) {
 636         debug(D_RRD_STATS, "%s: last_collect_ut = %0.3Lf (last collection time)", st->name, (long double)last_collect_ut/1000000.0);
 637         debug(D_RRD_STATS, "%s: now_collect_ut  = %0.3Lf (current collection time)", st->name, (long double)now_collect_ut/1000000.0);
 638         debug(D_RRD_STATS, "%s: last_stored_ut  = %0.3Lf (last updated time)", st->name, (long double)last_stored_ut/1000000.0);
 639         debug(D_RRD_STATS, "%s: next_store_ut   = %0.3Lf (next interpolation point)", st->name, (long double)next_store_ut/1000000.0);
 640     }
 641
 642     if(unlikely(!st->counter_done)) {
 643         store_this_entry = 0;
 644         if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
 645             debug(D_RRD_STATS, "%s: Will not store the next entry.", st->name);
 646     }
 647     st->counter_done++;
 648
 649     // calculate totals and count the dimensions
 650     int dimensions;
 651     st->collected_total = 0;
 652     for( rd = st->dimensions, dimensions = 0 ; rd ; rd = rd->next, dimensions++ )
 653         if(likely(rrddim_flag_check(rd, RRDDIM_FLAG_UPDATED)))
 654             st->collected_total += rd->collected_value;
 655
 656     uint32_t storage_flags = SN_EXISTS;
 657
 658     // process all dimensions to calculate their values
 659     // based on the collected figures only
 660     // at this stage we do not interpolate anything
 661     for( rd = st->dimensions ; rd ; rd = rd->next ) {
 662
 663         if(unlikely(!rrddim_flag_check(rd, RRDDIM_FLAG_UPDATED))) {
 664             rd->calculated_value = 0;
 665             continue;
 666         }
 667
 668         if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
 669             debug(D_RRD_STATS, "%s/%s: START "
 670                     " last_collected_value = " COLLECTED_NUMBER_FORMAT
 671                     " collected_value = " COLLECTED_NUMBER_FORMAT
 672                     " last_calculated_value = " CALCULATED_NUMBER_FORMAT
 673                     " calculated_value = " CALCULATED_NUMBER_FORMAT
 674                                       , st->id, rd->name
 675                                       , rd->last_collected_value
 676                                       , rd->collected_value
 677                                       , rd->last_calculated_value
 678                                       , rd->calculated_value
 679             );
 680
 681         switch(rd->algorithm) {
 682             case RRD_ALGORITHM_ABSOLUTE:
 683                 rd->calculated_value = (calculated_number)rd->collected_value
 684                                        * (calculated_number)rd->multiplier
 685                                        / (calculated_number)rd->divisor;
 686
 687                 if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
 688                     debug(D_RRD_STATS, "%s/%s: CALC ABS/ABS-NO-IN "
 689                             CALCULATED_NUMBER_FORMAT " = "
 690                             COLLECTED_NUMBER_FORMAT
 691                             " * " CALCULATED_NUMBER_FORMAT
 692                             " / " CALCULATED_NUMBER_FORMAT
 693                           , st->id, rd->name
 694                           , rd->calculated_value
 695                           , rd->collected_value
 696                           , (calculated_number)rd->multiplier
 697                           , (calculated_number)rd->divisor
 698                     );
 699                 break;
 700
 701             case RRD_ALGORITHM_PCENT_OVER_ROW_TOTAL:
 702                 if(unlikely(!st->collected_total))
 703                     rd->calculated_value = 0;
 704                 else
 705                     // the percentage of the current value
 706                     // over the total of all dimensions
 707                     rd->calculated_value =
 708                             (calculated_number)100
 709                             * (calculated_number)rd->collected_value
 710                             / (calculated_number)st->collected_total;
 711
 712                 if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
 713                     debug(D_RRD_STATS, "%s/%s: CALC PCENT-ROW "
 714                             CALCULATED_NUMBER_FORMAT " = 100"
 715                                     " * " COLLECTED_NUMBER_FORMAT
 716                             " / " COLLECTED_NUMBER_FORMAT
 717                           , st->id, rd->name
 718                           , rd->calculated_value
 719                           , rd->collected_value
 720                           , st->collected_total
 721                     );
 722                 break;
 723
 724             case RRD_ALGORITHM_INCREMENTAL:
 725                 if(unlikely(rd->counter <= 1)) {
 726                     rd->calculated_value = 0;
 727                     continue;
 728                 }
 729
 730                 // if the new is smaller than the old (an overflow, or reset), set the old equal to the new
 731                 // to reset the calculation (it will give zero as the calculation for this second)
 732                 if(unlikely(rd->last_collected_value > rd->collected_value)) {
 733                     debug(D_RRD_STATS, "%s.%s: RESET or OVERFLOW. Last collected value = " COLLECTED_NUMBER_FORMAT ", current = " COLLECTED_NUMBER_FORMAT
 734                           , st->name, rd->name
 735                           , rd->last_collected_value
 736                           , rd->collected_value);
 737
 738                     if(!(rrddim_flag_check(rd, RRDDIM_FLAG_DONT_DETECT_RESETS_OR_OVERFLOWS)))
 739                         storage_flags = SN_EXISTS_RESET;
 740
 741                     rd->last_collected_value = rd->collected_value;
 742                 }
 743
 744                 rd->calculated_value +=
 745                         (calculated_number)(rd->collected_value - rd->last_collected_value)
 746                         * (calculated_number)rd->multiplier
 747                         / (calculated_number)rd->divisor;
 748
 749                 if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
 750                     debug(D_RRD_STATS, "%s/%s: CALC INC PRE "
 751                             CALCULATED_NUMBER_FORMAT " = ("
 752                             COLLECTED_NUMBER_FORMAT " - " COLLECTED_NUMBER_FORMAT
 753                             ")"
 754                                     " * " CALCULATED_NUMBER_FORMAT
 755                             " / " CALCULATED_NUMBER_FORMAT
 756                           , st->id, rd->name
 757                           , rd->calculated_value
 758                           , rd->collected_value, rd->last_collected_value
 759                           , (calculated_number)rd->multiplier
 760                           , (calculated_number)rd->divisor
 761                     );
 762                 break;
 763
 764             case RRD_ALGORITHM_PCENT_OVER_DIFF_TOTAL:
 765                 if(unlikely(rd->counter <= 1)) {
 766                     rd->calculated_value = 0;
 767                     continue;
 768                 }
 769
 770                 // if the new is smaller than the old (an overflow, or reset), set the old equal to the new
 771                 // to reset the calculation (it will give zero as the calculation for this second)
 772                 if(unlikely(rd->last_collected_value > rd->collected_value)) {
 773                     debug(D_RRD_STATS, "%s.%s: RESET or OVERFLOW. Last collected value = " COLLECTED_NUMBER_FORMAT ", current = " COLLECTED_NUMBER_FORMAT
 774                           , st->name, rd->name
 775                           , rd->last_collected_value
 776                           , rd->collected_value);
 777
 778                     if(!(rrddim_flag_check(rd, RRDDIM_FLAG_DONT_DETECT_RESETS_OR_OVERFLOWS)))
 779                         storage_flags = SN_EXISTS_RESET;
 780
 781                     rd->last_collected_value = rd->collected_value;
 782                 }
 783
 784                 // the percentage of the current increment
 785                 // over the increment of all dimensions together
 786                 if(unlikely(st->collected_total == st->last_collected_total))
 787                     rd->calculated_value = 0;
 788                 else
 789                     rd->calculated_value =
 790                             (calculated_number)100
 791                             * (calculated_number)(rd->collected_value - rd->last_collected_value)
 792                             / (calculated_number)(st->collected_total - st->last_collected_total);
 793
 794                 if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
 795                     debug(D_RRD_STATS, "%s/%s: CALC PCENT-DIFF "
 796                             CALCULATED_NUMBER_FORMAT " = 100"
 797                                     " * (" COLLECTED_NUMBER_FORMAT " - " COLLECTED_NUMBER_FORMAT ")"
 798                                     " / (" COLLECTED_NUMBER_FORMAT " - " COLLECTED_NUMBER_FORMAT ")"
 799                           , st->id, rd->name
 800                           , rd->calculated_value
 801                           , rd->collected_value, rd->last_collected_value
 802                           , st->collected_total, st->last_collected_total
 803                     );
 804                 break;
 805
 806             default:
 807                 // make the default zero, to make sure
 808                 // it gets noticed when we add new types
 809                 rd->calculated_value = 0;
 810
 811                 if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
 812                     debug(D_RRD_STATS, "%s/%s: CALC "
 813                             CALCULATED_NUMBER_FORMAT " = 0"
 814                           , st->id, rd->name
 815                           , rd->calculated_value
 816                     );
 817                 break;
 818         }
 819
 820         if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
 821             debug(D_RRD_STATS, "%s/%s: PHASE2 "
 822                     " last_collected_value = " COLLECTED_NUMBER_FORMAT
 823                     " collected_value = " COLLECTED_NUMBER_FORMAT
 824                     " last_calculated_value = " CALCULATED_NUMBER_FORMAT
 825                     " calculated_value = " CALCULATED_NUMBER_FORMAT
 826                                       , st->id, rd->name
 827                                       , rd->last_collected_value
 828                                       , rd->collected_value
 829                                       , rd->last_calculated_value
 830                                       , rd->calculated_value
 831             );
 832
 833     }
 834
 835     // at this point we have all the calculated values ready
 836     // it is now time to interpolate values on a second boundary
 837
 838     if(unlikely(now_collect_ut < next_store_ut)) {
 839         // this is collected in the same interpolation point
 840
 841         if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
 842             debug(D_RRD_STATS, "%s: THIS IS IN THE SAME INTERPOLATION POINT", st->name);
 843
 844 #ifdef NETDATA_INTERNAL_CHECKS
 845         info("%s is collected in the same interpolation point: short by %llu microseconds", st->name, next_store_ut - now_collect_ut);
 846 #endif
 847     }
 848
 849     usec_t first_ut = last_stored_ut;
 850     long long iterations = (now_collect_ut - last_stored_ut) / (update_every_ut);
 851     if((now_collect_ut % (update_every_ut)) == 0) iterations++;
 852
 853     for( ; next_store_ut <= now_collect_ut ; last_collect_ut = next_store_ut, next_store_ut += update_every_ut, iterations-- ) {
 854 #ifdef NETDATA_INTERNAL_CHECKS
 855         if(iterations < 0) { error("%s: iterations calculation wrapped! first_ut = %llu, last_stored_ut = %llu, next_store_ut = %llu, now_collect_ut = %llu", st->name, first_ut, last_stored_ut, next_store_ut, now_collect_ut); }
 856 #endif
 857
 858         if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG))) {
 859             debug(D_RRD_STATS, "%s: last_stored_ut = %0.3Lf (last updated time)", st->name, (long double)last_stored_ut/1000000.0);
 860             debug(D_RRD_STATS, "%s: next_store_ut  = %0.3Lf (next interpolation point)", st->name, (long double)next_store_ut/1000000.0);
 861         }
 862
 863         st->last_updated.tv_sec = (time_t) (next_store_ut / USEC_PER_SEC);
 864         st->last_updated.tv_usec = 0;
 865
 866         for( rd = st->dimensions ; likely(rd) ; rd = rd->next ) {
 867             calculated_number new_value;
 868
 869             switch(rd->algorithm) {
 870                 case RRD_ALGORITHM_INCREMENTAL:
 871                     new_value = (calculated_number)
 872                             (      rd->calculated_value
 873                                    * (calculated_number)(next_store_ut - last_collect_ut)
 874                                    / (calculated_number)(now_collect_ut - last_collect_ut)
 875                             );
 876
 877                     if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
 878                         debug(D_RRD_STATS, "%s/%s: CALC2 INC "
 879                                 CALCULATED_NUMBER_FORMAT " = "
 880                                 CALCULATED_NUMBER_FORMAT
 881                                 " * %llu"
 882                                         " / %llu"
 883                               , st->id, rd->name
 884                               , new_value
 885                               , rd->calculated_value
 886                               , (next_store_ut - last_stored_ut)
 887                               , (now_collect_ut - last_stored_ut)
 888                         );
 889
 890                     rd->calculated_value -= new_value;
 891                     new_value += rd->last_calculated_value;
 892                     rd->last_calculated_value = 0;
 893                     new_value /= (calculated_number)st->update_every;
 894
 895                     if(unlikely(next_store_ut - last_stored_ut < update_every_ut)) {
 896                         if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
 897                             debug(D_RRD_STATS, "%s/%s: COLLECTION POINT IS SHORT " CALCULATED_NUMBER_FORMAT " - EXTRAPOLATING",
 898                                     st->id, rd->name
 899                                   , (calculated_number)(next_store_ut - last_stored_ut)
 900                             );
 901                         new_value = new_value * (calculated_number)(st->update_every * 1000000) / (calculated_number)(next_store_ut - last_stored_ut);
 902                     }
 903                     break;
 904
 905                 case RRD_ALGORITHM_ABSOLUTE:
 906                 case RRD_ALGORITHM_PCENT_OVER_ROW_TOTAL:
 907                 case RRD_ALGORITHM_PCENT_OVER_DIFF_TOTAL:
 908                 default:
 909                     if(iterations == 1) {
 910                         // this is the last iteration
 911                         // do not interpolate
 912                         // just show the calculated value
 913
 914                         new_value = rd->calculated_value;
 915                     }
 916                     else {
 917                         // we have missed an update
 918                         // interpolate in the middle values
 919
 920                         new_value = (calculated_number)
 921                                 (   (     (rd->calculated_value - rd->last_calculated_value)
 922                                           * (calculated_number)(next_store_ut - last_collect_ut)
 923                                           / (calculated_number)(now_collect_ut - last_collect_ut)
 924                                     )
 925                                     +  rd->last_calculated_value
 926                                 );
 927
 928                         if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
 929                             debug(D_RRD_STATS, "%s/%s: CALC2 DEF "
 930                                     CALCULATED_NUMBER_FORMAT " = ((("
 931                                             "(" CALCULATED_NUMBER_FORMAT " - " CALCULATED_NUMBER_FORMAT ")"
 932                                             " * %llu"
 933                                             " / %llu) + " CALCULATED_NUMBER_FORMAT
 934                                   , st->id, rd->name
 935                                   , new_value
 936                                   , rd->calculated_value, rd->last_calculated_value
 937                                   , (next_store_ut - first_ut)
 938                                   , (now_collect_ut - first_ut), rd->last_calculated_value
 939                             );
 940                     }
 941                     break;
 942             }
 943
 944             if(unlikely(!store_this_entry)) {
 945                 rd->values[st->current_entry] = pack_storage_number(0, SN_NOT_EXISTS);
 946                 continue;
 947             }
 948
 949             if(likely(rrddim_flag_check(rd, RRDDIM_FLAG_UPDATED) && rd->counter > 1 && iterations < st->gap_when_lost_iterations_above)) {
 950                 rd->values[st->current_entry] = pack_storage_number(new_value, storage_flags );
 951                 rd->last_stored_value = new_value;
 952
 953                 if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
 954                     debug(D_RRD_STATS, "%s/%s: STORE[%ld] "
 955                             CALCULATED_NUMBER_FORMAT " = " CALCULATED_NUMBER_FORMAT
 956                           , st->id, rd->name
 957                           , st->current_entry
 958                           , unpack_storage_number(rd->values[st->current_entry]), new_value
 959                     );
 960             }
 961             else {
 962                 if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
 963                     debug(D_RRD_STATS, "%s/%s: STORE[%ld] = NON EXISTING "
 964                           , st->id, rd->name
 965                           , st->current_entry
 966                     );
 967                 rd->values[st->current_entry] = pack_storage_number(0, SN_NOT_EXISTS);
 968                 rd->last_stored_value = NAN;
 969             }
 970
 971             stored_entries++;
 972
 973             if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG))) {
 974                 calculated_number t1 = new_value * (calculated_number)rd->multiplier / (calculated_number)rd->divisor;
 975                 calculated_number t2 = unpack_storage_number(rd->values[st->current_entry]);
 976                 calculated_number accuracy = accuracy_loss(t1, t2);
 977                 debug(D_RRD_STATS, "%s/%s: UNPACK[%ld] = " CALCULATED_NUMBER_FORMAT " FLAGS=0x%08x (original = " CALCULATED_NUMBER_FORMAT ", accuracy loss = " CALCULATED_NUMBER_FORMAT "%%%s)"
 978                       , st->id, rd->name
 979                       , st->current_entry
 980                       , t2
 981                       , get_storage_number_flags(rd->values[st->current_entry])
 982                       , t1
 983                       , accuracy
 984                       , (accuracy > ACCURACY_LOSS) ? " **TOO BIG** " : ""
 985                 );
 986
 987                 rd->collected_volume += t1;
 988                 rd->stored_volume += t2;
 989                 accuracy = accuracy_loss(rd->collected_volume, rd->stored_volume);
 990                 debug(D_RRD_STATS, "%s/%s: VOLUME[%ld] = " CALCULATED_NUMBER_FORMAT ", calculated  = " CALCULATED_NUMBER_FORMAT ", accuracy loss = " CALCULATED_NUMBER_FORMAT "%%%s"
 991                       , st->id, rd->name
 992                       , st->current_entry
 993                       , rd->stored_volume
 994                       , rd->collected_volume
 995                       , accuracy
 996                       , (accuracy > ACCURACY_LOSS) ? " **TOO BIG** " : ""
 997                 );
 998
 999             }
1000         }
1001         // reset the storage flags for the next point, if any;
1002         storage_flags = SN_EXISTS;
1003
1004         st->counter++;
1005         st->current_entry = ((st->current_entry + 1) >= st->entries) ? 0 : st->current_entry + 1;
1006         last_stored_ut = next_store_ut;
1007     }
1008
1009     st->last_collected_total  = st->collected_total;
1010
1011     for( rd = st->dimensions; rd ; rd = rd->next ) {
1012         if(unlikely(!rrddim_flag_check(rd, RRDDIM_FLAG_UPDATED)))
1013             continue;
1014
1015         if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
1016             debug(D_RRD_STATS, "%s/%s: setting last_collected_value (old: " COLLECTED_NUMBER_FORMAT ") to last_collected_value (new: " COLLECTED_NUMBER_FORMAT ")", st->id, rd->name, rd->last_collected_value, rd->collected_value);
1017
1018         rd->last_collected_value = rd->collected_value;
1019
1020         switch(rd->algorithm) {
1021             case RRD_ALGORITHM_INCREMENTAL:
1022                 if(unlikely(!first_entry)) {
1023                     if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
1024                         debug(D_RRD_STATS, "%s/%s: setting last_calculated_value (old: " CALCULATED_NUMBER_FORMAT ") to last_calculated_value (new: " CALCULATED_NUMBER_FORMAT ")", st->id, rd->name, rd->last_calculated_value + rd->calculated_value, rd->calculated_value);
1025                     rd->last_calculated_value += rd->calculated_value;
1026                 }
1027                 else {
1028                     if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
1029                         debug(D_RRD_STATS, "%s: THIS IS THE FIRST POINT", st->name);
1030                 }
1031                 break;
1032
1033             case RRD_ALGORITHM_ABSOLUTE:
1034             case RRD_ALGORITHM_PCENT_OVER_ROW_TOTAL:
1035             case RRD_ALGORITHM_PCENT_OVER_DIFF_TOTAL:
1036                 if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
1037                     debug(D_RRD_STATS, "%s/%s: setting last_calculated_value (old: " CALCULATED_NUMBER_FORMAT ") to last_calculated_value (new: " CALCULATED_NUMBER_FORMAT ")", st->id, rd->name, rd->last_calculated_value, rd->calculated_value);
1038                 rd->last_calculated_value = rd->calculated_value;
1039                 break;
1040         }
1041
1042         rd->calculated_value = 0;
1043         rd->collected_value = 0;
1044         rrddim_flag_clear(rd, RRDDIM_FLAG_UPDATED);
1045
1046         if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
1047             debug(D_RRD_STATS, "%s/%s: END "
1048                     " last_collected_value = " COLLECTED_NUMBER_FORMAT
1049                     " collected_value = " COLLECTED_NUMBER_FORMAT
1050                     " last_calculated_value = " CALCULATED_NUMBER_FORMAT
1051                     " calculated_value = " CALCULATED_NUMBER_FORMAT
1052                                       , st->id, rd->name
1053                                       , rd->last_collected_value
1054                                       , rd->collected_value
1055                                       , rd->last_calculated_value
1056                                       , rd->calculated_value
1057             );
1058     }
1059
1060     // ALL DONE ABOUT THE DATA UPDATE
1061     // --------------------------------------------------------------------
1062
1063 /*
1064     // find if there are any obsolete dimensions (not updated recently)
1065     if(unlikely(rrd_delete_unupdated_dimensions)) {
1066
1067         for( rd = st->dimensions; likely(rd) ; rd = rd->next )
1068             if((rd->last_collected_time.tv_sec + (rrd_delete_unupdated_dimensions * st->update_every)) < st->last_collected_time.tv_sec)
1069                 break;
1070
1071         if(unlikely(rd)) {
1072             RRDDIM *last;
1073             // there is dimension to free
1074             // upgrade our read lock to a write lock
1075             pthread_rwlock_unlock(&st->rwlock);
1076             pthread_rwlock_wrlock(&st->rwlock);
1077
1078             for( rd = st->dimensions, last = NULL ; likely(rd) ; ) {
1079                 // remove it only it is not updated in rrd_delete_unupdated_dimensions seconds
1080
1081                 if(unlikely((rd->last_collected_time.tv_sec + (rrd_delete_unupdated_dimensions * st->update_every)) < st->last_collected_time.tv_sec)) {
1082                     info("Removing obsolete dimension '%s' (%s) of '%s' (%s).", rd->name, rd->id, st->name, st->id);
1083
1084                     if(unlikely(!last)) {
1085                         st->dimensions = rd->next;
1086                         rd->next = NULL;
1087                         rrddim_free(st, rd);
1088                         rd = st->dimensions;
1089                         continue;
1090                     }
1091                     else {
1092                         last->next = rd->next;
1093                         rd->next = NULL;
1094                         rrddim_free(st, rd);
1095                         rd = last->next;
1096                         continue;
1097                     }
1098                 }
1099
1100                 last = rd;
1101                 rd = rd->next;
1102             }
1103
1104             if(unlikely(!st->dimensions)) {
1105                 info("Disabling chart %s (%s) since it does not have any dimensions", st->name, st->id);
1106                 st->enabled = 0;
1107             }
1108         }
1109     }
1110 */
1111
1112     pthread_rwlock_unlock(&st->rwlock);
1113
1114     if(unlikely(pthread_setcancelstate(pthreadoldcancelstate, NULL) != 0))
1115         error("Cannot set pthread cancel state to RESTORE (%d).", pthreadoldcancelstate);
1116
1117     return(st->usec_since_last_update);
1118 }
1119