src/rrd_update.c

   1 /*****************************************************************************
   2  * RRDtool 1.4.2  Copyright by Tobi Oetiker, 1997-2009
   3  *                Copyright by Florian Forster, 2008
   4  *****************************************************************************
   5  * rrd_update.c  RRD Update Function
   6  *****************************************************************************
   7  * $Id$
   8  *****************************************************************************/
   9
  10 #include "rrd_tool.h"
  11
  12 #if defined(_WIN32) && !defined(__CYGWIN__) && !defined(__CYGWIN32__)
  13 #include <sys/locking.h>
  14 #include <sys/stat.h>
  15 #include <io.h>
  16 #endif
  17
  18 #include <locale.h>
  19
  20 #include "rrd_hw.h"
  21 #include "rrd_rpncalc.h"
  22
  23 #include "rrd_is_thread_safe.h"
  24 #include "unused.h"
  25
  26 #include "rrd_client.h"
  27
  28 #if defined(_WIN32) && !defined(__CYGWIN__) && !defined(__CYGWIN32__)
  29 /*
  30  * WIN32 does not have gettimeofday     and struct timeval. This is a quick and dirty
  31  * replacement.
  32  */
  33 #include <sys/timeb.h>
  34
  35 #ifndef __MINGW32__
  36 struct timeval {
  37     time_t    tv_sec;   /* seconds */
  38     long      tv_usec;  /* microseconds */
  39 };
  40 #endif
  41
  42 struct __timezone {
  43     int       tz_minuteswest;   /* minutes W of Greenwich */
  44     int       tz_dsttime;   /* type of dst correction */
  45 };
  46
  47 static int gettimeofday(
  48     struct timeval *t,
  49     struct __timezone *tz)
  50 {
  51
  52     struct _timeb current_time;
  53
  54     _ftime(&current_time);
  55
  56     t->tv_sec = current_time.time;
  57     t->tv_usec = current_time.millitm * 1000;
  58
  59     return 0;
  60 }
  61
  62 #endif
  63
  64 /* FUNCTION PROTOTYPES */
  65
  66 int       rrd_update_r(
  67     const char *filename,
  68     const char *tmplt,
  69     int argc,
  70     const char **argv);
  71 int       _rrd_update(
  72     const char *filename,
  73     const char *tmplt,
  74     int argc,
  75     const char **argv,
  76     rrd_info_t *);
  77
  78 static int allocate_data_structures(
  79     rrd_t *rrd,
  80     char ***updvals,
  81     rrd_value_t **pdp_temp,
  82     const char *tmplt,
  83     long **tmpl_idx,
  84     unsigned long *tmpl_cnt,
  85     unsigned long **rra_step_cnt,
  86     unsigned long **skip_update,
  87     rrd_value_t **pdp_new);
  88
  89 static int parse_template(
  90     rrd_t *rrd,
  91     const char *tmplt,
  92     unsigned long *tmpl_cnt,
  93     long *tmpl_idx);
  94
  95 static int process_arg(
  96     char *step_start,
  97     rrd_t *rrd,
  98     rrd_file_t *rrd_file,
  99     unsigned long rra_begin,
 100     time_t *current_time,
 101     unsigned long *current_time_usec,
 102     rrd_value_t *pdp_temp,
 103     rrd_value_t *pdp_new,
 104     unsigned long *rra_step_cnt,
 105     char **updvals,
 106     long *tmpl_idx,
 107     unsigned long tmpl_cnt,
 108     rrd_info_t ** pcdp_summary,
 109     int version,
 110     unsigned long *skip_update,
 111     int *schedule_smooth);
 112
 113 static int parse_ds(
 114     rrd_t *rrd,
 115     char **updvals,
 116     long *tmpl_idx,
 117     char *input,
 118     unsigned long tmpl_cnt,
 119     time_t *current_time,
 120     unsigned long *current_time_usec,
 121     int version);
 122
 123 static int get_time_from_reading(
 124     rrd_t *rrd,
 125     char timesyntax,
 126     char **updvals,
 127     time_t *current_time,
 128     unsigned long *current_time_usec,
 129     int version);
 130
 131 static int update_pdp_prep(
 132     rrd_t *rrd,
 133     char **updvals,
 134     rrd_value_t *pdp_new,
 135     double interval);
 136
 137 static int calculate_elapsed_steps(
 138     rrd_t *rrd,
 139     unsigned long current_time,
 140     unsigned long current_time_usec,
 141     double interval,
 142     double *pre_int,
 143     double *post_int,
 144     unsigned long *proc_pdp_cnt);
 145
 146 static void simple_update(
 147     rrd_t *rrd,
 148     double interval,
 149     rrd_value_t *pdp_new);
 150
 151 static int process_all_pdp_st(
 152     rrd_t *rrd,
 153     double interval,
 154     double pre_int,
 155     double post_int,
 156     unsigned long elapsed_pdp_st,
 157     rrd_value_t *pdp_new,
 158     rrd_value_t *pdp_temp);
 159
 160 static int process_pdp_st(
 161     rrd_t *rrd,
 162     unsigned long ds_idx,
 163     double interval,
 164     double pre_int,
 165     double post_int,
 166     long diff_pdp_st,
 167     rrd_value_t *pdp_new,
 168     rrd_value_t *pdp_temp);
 169
 170 static int update_all_cdp_prep(
 171     rrd_t *rrd,
 172     unsigned long *rra_step_cnt,
 173     unsigned long rra_begin,
 174     rrd_file_t *rrd_file,
 175     unsigned long elapsed_pdp_st,
 176     unsigned long proc_pdp_cnt,
 177     rrd_value_t **last_seasonal_coef,
 178     rrd_value_t **seasonal_coef,
 179     rrd_value_t *pdp_temp,
 180     unsigned long *skip_update,
 181     int *schedule_smooth);
 182
 183 static int do_schedule_smooth(
 184     rrd_t *rrd,
 185     unsigned long rra_idx,
 186     unsigned long elapsed_pdp_st);
 187
 188 static int update_cdp_prep(
 189     rrd_t *rrd,
 190     unsigned long elapsed_pdp_st,
 191     unsigned long start_pdp_offset,
 192     unsigned long *rra_step_cnt,
 193     int rra_idx,
 194     rrd_value_t *pdp_temp,
 195     rrd_value_t *last_seasonal_coef,
 196     rrd_value_t *seasonal_coef,
 197     int current_cf);
 198
 199 static void update_cdp(
 200     unival *scratch,
 201     int current_cf,
 202     rrd_value_t pdp_temp_val,
 203     unsigned long rra_step_cnt,
 204     unsigned long elapsed_pdp_st,
 205     unsigned long start_pdp_offset,
 206     unsigned long pdp_cnt,
 207     rrd_value_t xff,
 208     int i,
 209     int ii);
 210
 211 static void initialize_cdp_val(
 212     unival *scratch,
 213     int current_cf,
 214     rrd_value_t pdp_temp_val,
 215     unsigned long elapsed_pdp_st,
 216     unsigned long start_pdp_offset,
 217     unsigned long pdp_cnt);
 218
 219 static void reset_cdp(
 220     rrd_t *rrd,
 221     unsigned long elapsed_pdp_st,
 222     rrd_value_t *pdp_temp,
 223     rrd_value_t *last_seasonal_coef,
 224     rrd_value_t *seasonal_coef,
 225     int rra_idx,
 226     int ds_idx,
 227     int cdp_idx,
 228     enum cf_en current_cf);
 229
 230 static rrd_value_t initialize_average_carry_over(
 231     rrd_value_t pdp_temp_val,
 232     unsigned long elapsed_pdp_st,
 233     unsigned long start_pdp_offset,
 234     unsigned long pdp_cnt);
 235
 236 static rrd_value_t calculate_cdp_val(
 237     rrd_value_t cdp_val,
 238     rrd_value_t pdp_temp_val,
 239     unsigned long elapsed_pdp_st,
 240     int current_cf,
 241     int i,
 242     int ii);
 243
 244 static int update_aberrant_cdps(
 245     rrd_t *rrd,
 246     rrd_file_t *rrd_file,
 247     unsigned long rra_begin,
 248     unsigned long elapsed_pdp_st,
 249     rrd_value_t *pdp_temp,
 250     rrd_value_t **seasonal_coef);
 251
 252 static int write_to_rras(
 253     rrd_t *rrd,
 254     rrd_file_t *rrd_file,
 255     unsigned long *rra_step_cnt,
 256     unsigned long rra_begin,
 257     time_t current_time,
 258     unsigned long *skip_update,
 259     rrd_info_t ** pcdp_summary);
 260
 261 static int write_RRA_row(
 262     rrd_file_t *rrd_file,
 263     rrd_t *rrd,
 264     unsigned long rra_idx,
 265     unsigned short CDP_scratch_idx,
 266     rrd_info_t ** pcdp_summary,
 267     time_t rra_time);
 268
 269 static int smooth_all_rras(
 270     rrd_t *rrd,
 271     rrd_file_t *rrd_file,
 272     unsigned long rra_begin);
 273
 274 #ifndef HAVE_MMAP
 275 static int write_changes_to_disk(
 276     rrd_t *rrd,
 277     rrd_file_t *rrd_file,
 278     int version);
 279 #endif
 280
 281 /*
 282  * normalize time as returned by gettimeofday. usec part must
 283  * be always >= 0
 284  */
 285 static void normalize_time(
 286     struct timeval *t)
 287 {
 288     if (t->tv_usec < 0) {
 289         t->tv_sec--;
 290         t->tv_usec += 1e6L;
 291     }
 292 }
 293
 294 /*
 295  * Sets current_time and current_time_usec based on the current time.
 296  * current_time_usec is set to 0 if the version number is 1 or 2.
 297  */
 298 static void initialize_time(
 299     time_t *current_time,
 300     unsigned long *current_time_usec,
 301     int version)
 302 {
 303     struct timeval tmp_time;    /* used for time conversion */
 304
 305     gettimeofday(&tmp_time, 0);
 306     normalize_time(&tmp_time);
 307     *current_time = tmp_time.tv_sec;
 308     if (version >= 3) {
 309         *current_time_usec = tmp_time.tv_usec;
 310     } else {
 311         *current_time_usec = 0;
 312     }
 313 }
 314
 315 #define IFDNAN(X,Y) (isnan(X) ? (Y) : (X));
 316
 317 rrd_info_t *rrd_update_v(
 318     int argc,
 319     char **argv)
 320 {
 321     char     *tmplt = NULL;
 322     rrd_info_t *result = NULL;
 323     rrd_infoval_t rc;
 324     char *opt_daemon = NULL;
 325     struct option long_options[] = {
 326         {"template", required_argument, 0, 't'},
 327         {0, 0, 0, 0}
 328     };
 329
 330     rc.u_int = -1;
 331     optind = 0;
 332     opterr = 0;         /* initialize getopt */
 333
 334     while (1) {
 335         int       option_index = 0;
 336         int       opt;
 337
 338         opt = getopt_long(argc, argv, "t:", long_options, &option_index);
 339
 340         if (opt == EOF)
 341             break;
 342
 343         switch (opt) {
 344         case 't':
 345             tmplt = optarg;
 346             break;
 347
 348         case '?':
 349             rrd_set_error("unknown option '%s'", argv[optind - 1]);
 350             goto end_tag;
 351         }
 352     }
 353
 354     opt_daemon = getenv (ENV_RRDCACHED_ADDRESS);
 355     if (opt_daemon != NULL) {
 356         rrd_set_error ("The \"%s\" environment variable is defined, "
 357                 "but \"%s\" cannot work with rrdcached. Either unset "
 358                 "the environment variable or use \"update\" instead.",
 359                 ENV_RRDCACHED_ADDRESS, argv[0]);
 360         goto end_tag;
 361     }
 362
 363     /* need at least 2 arguments: filename, data. */
 364     if (argc - optind < 2) {
 365         rrd_set_error("Not enough arguments");
 366         goto end_tag;
 367     }
 368     rc.u_int = 0;
 369     result = rrd_info_push(NULL, sprintf_alloc("return_value"), RD_I_INT, rc);
 370     rc.u_int = _rrd_update(argv[optind], tmplt,
 371                            argc - optind - 1,
 372                            (const char **) (argv + optind + 1), result);
 373     result->value.u_int = rc.u_int;
 374   end_tag:
 375     return result;
 376 }
 377
 378 int rrd_update(
 379     int argc,
 380     char **argv)
 381 {
 382     struct option long_options[] = {
 383         {"template", required_argument, 0, 't'},
 384         {"daemon",   required_argument, 0, 'd'},
 385         {0, 0, 0, 0}
 386     };
 387     int       option_index = 0;
 388     int       opt;
 389     char     *tmplt = NULL;
 390     int       rc = -1;
 391     char     *opt_daemon = NULL;
 392
 393     optind = 0;
 394     opterr = 0;         /* initialize getopt */
 395
 396     while (1) {
 397         opt = getopt_long(argc, argv, "t:d:", long_options, &option_index);
 398
 399         if (opt == EOF)
 400             break;
 401
 402         switch (opt) {
 403         case 't':
 404             tmplt = strdup(optarg);
 405             break;
 406
 407         case 'd':
 408             if (opt_daemon != NULL)
 409                 free (opt_daemon);
 410             opt_daemon = strdup (optarg);
 411             if (opt_daemon == NULL)
 412             {
 413                 rrd_set_error("strdup failed.");
 414                 goto out;
 415             }
 416             break;
 417
 418         case '?':
 419             rrd_set_error("unknown option '%s'", argv[optind - 1]);
 420             goto out;
 421         }
 422     }
 423
 424     /* need at least 2 arguments: filename, data. */
 425     if (argc - optind < 2) {
 426         rrd_set_error("Not enough arguments");
 427         goto out;
 428     }
 429
 430     {   /* try to connect to rrdcached */
 431         int status = rrdc_connect(opt_daemon);
 432         if (status != 0) return status;
 433     }
 434
 435     if ((tmplt != NULL) && rrdc_is_connected(opt_daemon))
 436     {
 437         rrd_set_error("The caching daemon cannot be used together with "
 438                 "templates yet.");
 439         goto out;
 440     }
 441
 442     if (! rrdc_is_connected(opt_daemon))
 443     {
 444       rc = rrd_update_r(argv[optind], tmplt,
 445                         argc - optind - 1, (const char **) (argv + optind + 1));
 446     }
 447     else /* we are connected */
 448     {
 449         rc = rrdc_update (argv[optind], /* file */
 450                           argc - optind - 1, /* values_num */
 451                           (const char *const *) (argv + optind + 1)); /* values */
 452         if (rc > 0)
 453             rrd_set_error("Failed sending the values to rrdcached: %s",
 454                           rrd_strerror (rc));
 455     }
 456
 457   out:
 458     if (tmplt != NULL)
 459     {
 460         free(tmplt);
 461         tmplt = NULL;
 462     }
 463     if (opt_daemon != NULL)
 464     {
 465         free (opt_daemon);
 466         opt_daemon = NULL;
 467     }
 468     return rc;
 469 }
 470
 471 int rrd_update_r(
 472     const char *filename,
 473     const char *tmplt,
 474     int argc,
 475     const char **argv)
 476 {
 477     return _rrd_update(filename, tmplt, argc, argv, NULL);
 478 }
 479
 480 int _rrd_update(
 481     const char *filename,
 482     const char *tmplt,
 483     int argc,
 484     const char **argv,
 485     rrd_info_t * pcdp_summary)
 486 {
 487
 488     int       arg_i = 2;
 489
 490     unsigned long rra_begin;    /* byte pointer to the rra
 491                                  * area in the rrd file.  this
 492                                  * pointer never changes value */
 493     rrd_value_t *pdp_new;   /* prepare the incoming data to be added
 494                              * to the existing entry */
 495     rrd_value_t *pdp_temp;  /* prepare the pdp values to be added
 496                              * to the cdp values */
 497
 498     long     *tmpl_idx; /* index representing the settings
 499                          * transported by the tmplt index */
 500     unsigned long tmpl_cnt = 2; /* time and data */
 501     rrd_t     rrd;
 502     time_t    current_time = 0;
 503     unsigned long current_time_usec = 0;    /* microseconds part of current time */
 504     char    **updvals;
 505     int       schedule_smooth = 0;
 506
 507     /* number of elapsed PDP steps since last update */
 508     unsigned long *rra_step_cnt = NULL;
 509
 510     int       version;  /* rrd version */
 511     rrd_file_t *rrd_file;
 512     char     *arg_copy; /* for processing the argv */
 513     unsigned long *skip_update; /* RRAs to advance but not write */
 514
 515     /* need at least 1 arguments: data. */
 516     if (argc < 1) {
 517         rrd_set_error("Not enough arguments");
 518         goto err_out;
 519     }
 520
 521     rrd_init(&rrd);
 522     if ((rrd_file = rrd_open(filename, &rrd, RRD_READWRITE)) == NULL) {
 523         goto err_free;
 524     }
 525     /* We are now at the beginning of the rra's */
 526     rra_begin = rrd_file->header_len;
 527
 528     version = atoi(rrd.stat_head->version);
 529
 530     initialize_time(&current_time, &current_time_usec, version);
 531
 532     /* get exclusive lock to whole file.
 533      * lock gets removed when we close the file.
 534      */
 535     if (rrd_lock(rrd_file) != 0) {
 536         rrd_set_error("could not lock RRD");
 537         goto err_close;
 538     }
 539
 540     if (allocate_data_structures(&rrd, &updvals,
 541                                  &pdp_temp, tmplt, &tmpl_idx, &tmpl_cnt,
 542                                  &rra_step_cnt, &skip_update,
 543                                  &pdp_new) == -1) {
 544         goto err_close;
 545     }
 546
 547     /* loop through the arguments. */
 548     for (arg_i = 0; arg_i < argc; arg_i++) {
 549         if ((arg_copy = strdup(argv[arg_i])) == NULL) {
 550             rrd_set_error("failed duplication argv entry");
 551             break;
 552         }
 553         if (process_arg(arg_copy, &rrd, rrd_file, rra_begin,
 554                         &current_time, &current_time_usec, pdp_temp, pdp_new,
 555                         rra_step_cnt, updvals, tmpl_idx, tmpl_cnt,
 556                         &pcdp_summary, version, skip_update,
 557                         &schedule_smooth) == -1) {
 558             if (rrd_test_error()) { /* Should have error string always here */
 559                 char     *save_error;
 560
 561                 /* Prepend file name to error message */
 562                 if ((save_error = strdup(rrd_get_error())) != NULL) {
 563                     rrd_set_error("%s: %s", filename, save_error);
 564                     free(save_error);
 565                 }
 566             }
 567             free(arg_copy);
 568             break;
 569         }
 570         free(arg_copy);
 571     }
 572
 573     free(rra_step_cnt);
 574
 575     /* if we got here and if there is an error and if the file has not been
 576      * written to, then close things up and return. */
 577     if (rrd_test_error()) {
 578         goto err_free_structures;
 579     }
 580 #ifndef HAVE_MMAP
 581     if (write_changes_to_disk(&rrd, rrd_file, version) == -1) {
 582         goto err_free_structures;
 583     }
 584 #endif
 585
 586     /* calling the smoothing code here guarantees at most one smoothing
 587      * operation per rrd_update call. Unfortunately, it is possible with bulk
 588      * updates, or a long-delayed update for smoothing to occur off-schedule.
 589      * This really isn't critical except during the burn-in cycles. */
 590     if (schedule_smooth) {
 591         smooth_all_rras(&rrd, rrd_file, rra_begin);
 592     }
 593
 594 /*    rrd_dontneed(rrd_file,&rrd); */
 595     rrd_free(&rrd);
 596     rrd_close(rrd_file);
 597
 598     free(pdp_new);
 599     free(tmpl_idx);
 600     free(pdp_temp);
 601     free(skip_update);
 602     free(updvals);
 603     return 0;
 604
 605   err_free_structures:
 606     free(pdp_new);
 607     free(tmpl_idx);
 608     free(pdp_temp);
 609     free(skip_update);
 610     free(updvals);
 611   err_close:
 612     rrd_close(rrd_file);
 613   err_free:
 614     rrd_free(&rrd);
 615   err_out:
 616     return -1;
 617 }
 618
 619 /*
 620  * Allocate some important arrays used, and initialize the template.
 621  *
 622  * When it returns, either all of the structures are allocated
 623  * or none of them are.
 624  *
 625  * Returns 0 on success, -1 on error.
 626  */
 627 static int allocate_data_structures(
 628     rrd_t *rrd,
 629     char ***updvals,
 630     rrd_value_t **pdp_temp,
 631     const char *tmplt,
 632     long **tmpl_idx,
 633     unsigned long *tmpl_cnt,
 634     unsigned long **rra_step_cnt,
 635     unsigned long **skip_update,
 636     rrd_value_t **pdp_new)
 637 {
 638     unsigned  i, ii;
 639     if ((*updvals = (char **) malloc(sizeof(char *)
 640                                      * (rrd->stat_head->ds_cnt + 1))) == NULL) {
 641         rrd_set_error("allocating updvals pointer array.");
 642         return -1;
 643     }
 644     if ((*pdp_temp = (rrd_value_t *) malloc(sizeof(rrd_value_t)
 645                                             * rrd->stat_head->ds_cnt)) ==
 646         NULL) {
 647         rrd_set_error("allocating pdp_temp.");
 648         goto err_free_updvals;
 649     }
 650     if ((*skip_update = (unsigned long *) malloc(sizeof(unsigned long)
 651                                                  *
 652                                                  rrd->stat_head->rra_cnt)) ==
 653         NULL) {
 654         rrd_set_error("allocating skip_update.");
 655         goto err_free_pdp_temp;
 656     }
 657     if ((*tmpl_idx = (long *) malloc(sizeof(unsigned long)
 658                                      * (rrd->stat_head->ds_cnt + 1))) == NULL) {
 659         rrd_set_error("allocating tmpl_idx.");
 660         goto err_free_skip_update;
 661     }
 662     if ((*rra_step_cnt = (unsigned long *) malloc(sizeof(unsigned long)
 663                                                   *
 664                                                   (rrd->stat_head->
 665                                                    rra_cnt))) == NULL) {
 666         rrd_set_error("allocating rra_step_cnt.");
 667         goto err_free_tmpl_idx;
 668     }
 669
 670     /* initialize tmplt redirector */
 671     /* default config example (assume DS 1 is a CDEF DS)
 672        tmpl_idx[0] -> 0; (time)
 673        tmpl_idx[1] -> 1; (DS 0)
 674        tmpl_idx[2] -> 3; (DS 2)
 675        tmpl_idx[3] -> 4; (DS 3) */
 676     (*tmpl_idx)[0] = 0; /* time */
 677     for (i = 1, ii = 1; i <= rrd->stat_head->ds_cnt; i++) {
 678         if (dst_conv(rrd->ds_def[i - 1].dst) != DST_CDEF)
 679             (*tmpl_idx)[ii++] = i;
 680     }
 681     *tmpl_cnt = ii;
 682
 683     if (tmplt != NULL) {
 684         if (parse_template(rrd, tmplt, tmpl_cnt, *tmpl_idx) == -1) {
 685             goto err_free_rra_step_cnt;
 686         }
 687     }
 688
 689     if ((*pdp_new = (rrd_value_t *) malloc(sizeof(rrd_value_t)
 690                                            * rrd->stat_head->ds_cnt)) == NULL) {
 691         rrd_set_error("allocating pdp_new.");
 692         goto err_free_rra_step_cnt;
 693     }
 694
 695     return 0;
 696
 697   err_free_rra_step_cnt:
 698     free(*rra_step_cnt);
 699   err_free_tmpl_idx:
 700     free(*tmpl_idx);
 701   err_free_skip_update:
 702     free(*skip_update);
 703   err_free_pdp_temp:
 704     free(*pdp_temp);
 705   err_free_updvals:
 706     free(*updvals);
 707     return -1;
 708 }
 709
 710 /*
 711  * Parses tmplt and puts an ordered list of DS's into tmpl_idx.
 712  *
 713  * Returns 0 on success.
 714  */
 715 static int parse_template(
 716     rrd_t *rrd,
 717     const char *tmplt,
 718     unsigned long *tmpl_cnt,
 719     long *tmpl_idx)
 720 {
 721     char     *dsname, *tmplt_copy;
 722     unsigned int tmpl_len, i;
 723     int       ret = 0;
 724
 725     *tmpl_cnt = 1;      /* the first entry is the time */
 726
 727     /* we should work on a writeable copy here */
 728     if ((tmplt_copy = strdup(tmplt)) == NULL) {
 729         rrd_set_error("error copying tmplt '%s'", tmplt);
 730         ret = -1;
 731         goto out;
 732     }
 733
 734     dsname = tmplt_copy;
 735     tmpl_len = strlen(tmplt_copy);
 736     for (i = 0; i <= tmpl_len; i++) {
 737         if (tmplt_copy[i] == ':' || tmplt_copy[i] == '\0') {
 738             tmplt_copy[i] = '\0';
 739             if (*tmpl_cnt > rrd->stat_head->ds_cnt) {
 740                 rrd_set_error("tmplt contains more DS definitions than RRD");
 741                 ret = -1;
 742                 goto out_free_tmpl_copy;
 743             }
 744             if ((tmpl_idx[(*tmpl_cnt)++] = ds_match(rrd, dsname) + 1) == 0) {
 745                 rrd_set_error("unknown DS name '%s'", dsname);
 746                 ret = -1;
 747                 goto out_free_tmpl_copy;
 748             }
 749             /* go to the next entry on the tmplt_copy */
 750             if (i < tmpl_len)
 751                 dsname = &tmplt_copy[i + 1];
 752         }
 753     }
 754   out_free_tmpl_copy:
 755     free(tmplt_copy);
 756   out:
 757     return ret;
 758 }
 759
 760 /*
 761  * Parse an update string, updates the primary data points (PDPs)
 762  * and consolidated data points (CDPs), and writes changes to the RRAs.
 763  *
 764  * Returns 0 on success, -1 on error.
 765  */
 766 static int process_arg(
 767     char *step_start,
 768     rrd_t *rrd,
 769     rrd_file_t *rrd_file,
 770     unsigned long rra_begin,
 771     time_t *current_time,
 772     unsigned long *current_time_usec,
 773     rrd_value_t *pdp_temp,
 774     rrd_value_t *pdp_new,
 775     unsigned long *rra_step_cnt,
 776     char **updvals,
 777     long *tmpl_idx,
 778     unsigned long tmpl_cnt,
 779     rrd_info_t ** pcdp_summary,
 780     int version,
 781     unsigned long *skip_update,
 782     int *schedule_smooth)
 783 {
 784     rrd_value_t *seasonal_coef = NULL, *last_seasonal_coef = NULL;
 785
 786     /* a vector of future Holt-Winters seasonal coefs */
 787     unsigned long elapsed_pdp_st;
 788
 789     double    interval, pre_int, post_int;  /* interval between this and
 790                                              * the last run */
 791     unsigned long proc_pdp_cnt;
 792
 793     if (parse_ds(rrd, updvals, tmpl_idx, step_start, tmpl_cnt,
 794                  current_time, current_time_usec, version) == -1) {
 795         return -1;
 796     }
 797
 798     interval = (double) (*current_time - rrd->live_head->last_up)
 799         + (double) ((long) *current_time_usec -
 800                     (long) rrd->live_head->last_up_usec) / 1e6f;
 801
 802     /* process the data sources and update the pdp_prep
 803      * area accordingly */
 804     if (update_pdp_prep(rrd, updvals, pdp_new, interval) == -1) {
 805         return -1;
 806     }
 807
 808     elapsed_pdp_st = calculate_elapsed_steps(rrd,
 809                                              *current_time,
 810                                              *current_time_usec, interval,
 811                                              &pre_int, &post_int,
 812                                              &proc_pdp_cnt);
 813
 814     /* has a pdp_st moment occurred since the last run ? */
 815     if (elapsed_pdp_st == 0) {
 816         /* no we have not passed a pdp_st moment. therefore update is simple */
 817         simple_update(rrd, interval, pdp_new);
 818     } else {
 819         /* an pdp_st has occurred. */
 820         if (process_all_pdp_st(rrd, interval,
 821                                pre_int, post_int,
 822                                elapsed_pdp_st, pdp_new, pdp_temp) == -1) {
 823             return -1;
 824         }
 825         if (update_all_cdp_prep(rrd, rra_step_cnt,
 826                                 rra_begin, rrd_file,
 827                                 elapsed_pdp_st,
 828                                 proc_pdp_cnt,
 829                                 &last_seasonal_coef,
 830                                 &seasonal_coef,
 831                                 pdp_temp,
 832                                 skip_update, schedule_smooth) == -1) {
 833             goto err_free_coefficients;
 834         }
 835         if (update_aberrant_cdps(rrd, rrd_file, rra_begin,
 836                                  elapsed_pdp_st, pdp_temp,
 837                                  &seasonal_coef) == -1) {
 838             goto err_free_coefficients;
 839         }
 840         if (write_to_rras(rrd, rrd_file, rra_step_cnt, rra_begin,
 841                           *current_time, skip_update,
 842                           pcdp_summary) == -1) {
 843             goto err_free_coefficients;
 844         }
 845     }                   /* endif a pdp_st has occurred */
 846     rrd->live_head->last_up = *current_time;
 847     rrd->live_head->last_up_usec = *current_time_usec;
 848
 849     if (version < 3) {
 850         *rrd->legacy_last_up = rrd->live_head->last_up;
 851     }
 852     free(seasonal_coef);
 853     free(last_seasonal_coef);
 854     return 0;
 855
 856   err_free_coefficients:
 857     free(seasonal_coef);
 858     free(last_seasonal_coef);
 859     return -1;
 860 }
 861
 862 /*
 863  * Parse a DS string (time + colon-separated values), storing the
 864  * results in current_time, current_time_usec, and updvals.
 865  *
 866  * Returns 0 on success, -1 on error.
 867  */
 868 static int parse_ds(
 869     rrd_t *rrd,
 870     char **updvals,
 871     long *tmpl_idx,
 872     char *input,
 873     unsigned long tmpl_cnt,
 874     time_t *current_time,
 875     unsigned long *current_time_usec,
 876     int version)
 877 {
 878     char     *p;
 879     unsigned long i;
 880     char      timesyntax;
 881
 882     updvals[0] = input;
 883     /* initialize all ds input to unknown except the first one
 884        which has always got to be set */
 885     for (i = 1; i <= rrd->stat_head->ds_cnt; i++)
 886         updvals[i] = "U";
 887
 888     /* separate all ds elements; first must be examined separately
 889        due to alternate time syntax */
 890     if ((p = strchr(input, '@')) != NULL) {
 891         timesyntax = '@';
 892     } else if ((p = strchr(input, ':')) != NULL) {
 893         timesyntax = ':';
 894     } else {
 895         rrd_set_error("expected timestamp not found in data source from %s",
 896                       input);
 897         return -1;
 898     }
 899     *p = '\0';
 900     i = 1;
 901     updvals[tmpl_idx[i++]] = p + 1;
 902     while (*(++p)) {
 903         if (*p == ':') {
 904             *p = '\0';
 905             if (i < tmpl_cnt) {
 906                 updvals[tmpl_idx[i++]] = p + 1;
 907             }
 908             else {
 909                 rrd_set_error("found extra data on update argument: %s",p+1);
 910                 return -1;
 911             }
 912         }
 913     }
 914
 915     if (i != tmpl_cnt) {
 916         rrd_set_error("expected %lu data source readings (got %lu) from %s",
 917                       tmpl_cnt - 1, i - 1, input);
 918         return -1;
 919     }
 920
 921     if (get_time_from_reading(rrd, timesyntax, updvals,
 922                               current_time, current_time_usec,
 923                               version) == -1) {
 924         return -1;
 925     }
 926     return 0;
 927 }
 928
 929 /*
 930  * Parse the time in a DS string, store it in current_time and
 931  * current_time_usec and verify that it's later than the last
 932  * update for this DS.
 933  *
 934  * Returns 0 on success, -1 on error.
 935  */
 936 static int get_time_from_reading(
 937     rrd_t *rrd,
 938     char timesyntax,
 939     char **updvals,
 940     time_t *current_time,
 941     unsigned long *current_time_usec,
 942     int version)
 943 {
 944     double    tmp;
 945     char     *parsetime_error = NULL;
 946     char     *old_locale;
 947     rrd_time_value_t ds_tv;
 948     struct timeval tmp_time;    /* used for time conversion */
 949
 950     /* get the time from the reading ... handle N */
 951     if (timesyntax == '@') {    /* at-style */
 952         if ((parsetime_error = rrd_parsetime(updvals[0], &ds_tv))) {
 953             rrd_set_error("ds time: %s: %s", updvals[0], parsetime_error);
 954             return -1;
 955         }
 956         if (ds_tv.type == RELATIVE_TO_END_TIME ||
 957             ds_tv.type == RELATIVE_TO_START_TIME) {
 958             rrd_set_error("specifying time relative to the 'start' "
 959                           "or 'end' makes no sense here: %s", updvals[0]);
 960             return -1;
 961         }
 962         *current_time = mktime(&ds_tv.tm) +ds_tv.offset;
 963         *current_time_usec = 0; /* FIXME: how to handle usecs here ? */
 964     } else if (strcmp(updvals[0], "N") == 0) {
 965         gettimeofday(&tmp_time, 0);
 966         normalize_time(&tmp_time);
 967         *current_time = tmp_time.tv_sec;
 968         *current_time_usec = tmp_time.tv_usec;
 969     } else {
 970         old_locale = setlocale(LC_NUMERIC, "C");
 971         errno = 0;
 972         tmp = strtod(updvals[0], 0);
 973         if (errno > 0) {
 974             rrd_set_error("converting '%s' to float: %s",
 975                 updvals[0], rrd_strerror(errno));
 976             return -1;
 977         };
 978         setlocale(LC_NUMERIC, old_locale);
 979         if (tmp < 0.0){
 980             gettimeofday(&tmp_time, 0);
 981             tmp = (double)tmp_time.tv_sec + (double)tmp_time.tv_usec * 1e-6f + tmp;
 982         }
 983
 984         *current_time = floor(tmp);
 985         *current_time_usec = (long) ((tmp - (double) *current_time) * 1e6f);
 986     }
 987     /* dont do any correction for old version RRDs */
 988     if (version < 3)
 989         *current_time_usec = 0;
 990
 991     if (*current_time < rrd->live_head->last_up ||
 992         (*current_time == rrd->live_head->last_up &&
 993          (long) *current_time_usec <= (long) rrd->live_head->last_up_usec)) {
 994         rrd_set_error("illegal attempt to update using time %ld when "
 995                       "last update time is %ld (minimum one second step)",
 996                       *current_time, rrd->live_head->last_up);
 997         return -1;
 998     }
 999     return 0;
1000 }
1001
1002 /*
1003  * Update pdp_new by interpreting the updvals according to the DS type
1004  * (COUNTER, GAUGE, etc.).
1005  *
1006  * Returns 0 on success, -1 on error.
1007  */
1008 static int update_pdp_prep(
1009     rrd_t *rrd,
1010     char **updvals,
1011     rrd_value_t *pdp_new,
1012     double interval)
1013 {
1014     unsigned long ds_idx;
1015     int       ii;
1016     char     *endptr;   /* used in the conversion */
1017     double    rate;
1018     char     *old_locale;
1019     enum dst_en dst_idx;
1020
1021     for (ds_idx = 0; ds_idx < rrd->stat_head->ds_cnt; ds_idx++) {
1022         dst_idx = dst_conv(rrd->ds_def[ds_idx].dst);
1023
1024         /* make sure we do not build diffs with old last_ds values */
1025         if (rrd->ds_def[ds_idx].par[DS_mrhb_cnt].u_cnt < interval) {
1026             strncpy(rrd->pdp_prep[ds_idx].last_ds, "U", LAST_DS_LEN - 1);
1027             rrd->pdp_prep[ds_idx].last_ds[LAST_DS_LEN - 1] = '\0';
1028         }
1029
1030         /* NOTE: DST_CDEF should never enter this if block, because
1031          * updvals[ds_idx+1][0] is initialized to 'U'; unless the caller
1032          * accidently specified a value for the DST_CDEF. To handle this case,
1033          * an extra check is required. */
1034
1035         if ((updvals[ds_idx + 1][0] != 'U') &&
1036             (dst_idx != DST_CDEF) &&
1037             rrd->ds_def[ds_idx].par[DS_mrhb_cnt].u_cnt >= interval) {
1038             rate = DNAN;
1039
1040             /* pdp_new contains rate * time ... eg the bytes transferred during
1041              * the interval. Doing it this way saves a lot of math operations
1042              */
1043             switch (dst_idx) {
1044             case DST_COUNTER:
1045             case DST_DERIVE:
1046                 if ( (   updvals[ds_idx + 1][0] < '0'
1047                       || updvals[ds_idx + 1][0] > '9' )
1048                      && updvals[ds_idx + 1][0] != '-'
1049                      && updvals[ds_idx + 1][0] != 'U'
1050                      && updvals[ds_idx + 1][0] == '\0'
1051                    ) {
1052                     rrd_set_error("not a simple integer: '%s'",
1053                                   updvals[ds_idx + 1]);
1054                     return -1;
1055                 }
1056                 for (ii = 1; updvals[ds_idx + 1][ii] != '\0'; ii++) {
1057                     if (    updvals[ds_idx + 1][ii] < '0'
1058                          || updvals[ds_idx + 1][ii] > '9'
1059                        ) {
1060                         rrd_set_error("not a simple integer: '%s'",
1061                                       updvals[ds_idx + 1]);
1062                         return -1;
1063                     }
1064                 }
1065                 if (rrd->pdp_prep[ds_idx].last_ds[0] != 'U') {
1066                     pdp_new[ds_idx] =
1067                         rrd_diff(updvals[ds_idx + 1],
1068                                  rrd->pdp_prep[ds_idx].last_ds);
1069                     if (dst_idx == DST_COUNTER) {
1070                         /* simple overflow catcher. This will fail
1071                          * terribly for non 32 or 64 bit counters
1072                          * ... are there any others in SNMP land?
1073                          */
1074                         if (pdp_new[ds_idx] < (double) 0.0)
1075                             pdp_new[ds_idx] += (double) 4294967296.0;   /* 2^32 */
1076                         if (pdp_new[ds_idx] < (double) 0.0)
1077                             pdp_new[ds_idx] += (double) 18446744069414584320.0; /* 2^64-2^32 */
1078                     }
1079                     rate = pdp_new[ds_idx] / interval;
1080                 } else {
1081                     pdp_new[ds_idx] = DNAN;
1082                 }
1083                 break;
1084             case DST_ABSOLUTE:
1085                 old_locale = setlocale(LC_NUMERIC, "C");
1086                 errno = 0;
1087                 pdp_new[ds_idx] = strtod(updvals[ds_idx + 1], &endptr);
1088                 if (errno > 0) {
1089                     rrd_set_error("converting '%s' to float: %s",
1090                                   updvals[ds_idx + 1], rrd_strerror(errno));
1091                     return -1;
1092                 };
1093                 setlocale(LC_NUMERIC, old_locale);
1094                 if (endptr[0] != '\0') {
1095                     rrd_set_error
1096                         ("conversion of '%s' to float not complete: tail '%s'",
1097                          updvals[ds_idx + 1], endptr);
1098                     return -1;
1099                 }
1100                 rate = pdp_new[ds_idx] / interval;
1101                 break;
1102             case DST_GAUGE:
1103                 old_locale = setlocale(LC_NUMERIC, "C");
1104                 errno = 0;
1105                 pdp_new[ds_idx] =
1106                     strtod(updvals[ds_idx + 1], &endptr) * interval;
1107                 if (errno) {
1108                     rrd_set_error("converting '%s' to float: %s",
1109                                   updvals[ds_idx + 1], rrd_strerror(errno));
1110                     return -1;
1111                 };
1112                 setlocale(LC_NUMERIC, old_locale);
1113                 if (endptr[0] != '\0') {
1114                     rrd_set_error
1115                         ("conversion of '%s' to float not complete: tail '%s'",
1116                          updvals[ds_idx + 1], endptr);
1117                     return -1;
1118                 }
1119                 rate = pdp_new[ds_idx] / interval;
1120                 break;
1121             default:
1122                 rrd_set_error("rrd contains unknown DS type : '%s'",
1123                               rrd->ds_def[ds_idx].dst);
1124                 return -1;
1125             }
1126             /* break out of this for loop if the error string is set */
1127             if (rrd_test_error()) {
1128                 return -1;
1129             }
1130             /* make sure pdp_temp is neither too large or too small
1131              * if any of these occur it becomes unknown ...
1132              * sorry folks ... */
1133             if (!isnan(rate) &&
1134                 ((!isnan(rrd->ds_def[ds_idx].par[DS_max_val].u_val) &&
1135                   rate > rrd->ds_def[ds_idx].par[DS_max_val].u_val) ||
1136                  (!isnan(rrd->ds_def[ds_idx].par[DS_min_val].u_val) &&
1137                   rate < rrd->ds_def[ds_idx].par[DS_min_val].u_val))) {
1138                 pdp_new[ds_idx] = DNAN;
1139             }
1140         } else {
1141             /* no news is news all the same */
1142             pdp_new[ds_idx] = DNAN;
1143         }
1144
1145
1146         /* make a copy of the command line argument for the next run */
1147 #ifdef DEBUG
1148         fprintf(stderr, "prep ds[%lu]\t"
1149                 "last_arg '%s'\t"
1150                 "this_arg '%s'\t"
1151                 "pdp_new %10.2f\n",
1152                 ds_idx, rrd->pdp_prep[ds_idx].last_ds, updvals[ds_idx + 1],
1153                 pdp_new[ds_idx]);
1154 #endif
1155         strncpy(rrd->pdp_prep[ds_idx].last_ds, updvals[ds_idx + 1],
1156                 LAST_DS_LEN - 1);
1157         rrd->pdp_prep[ds_idx].last_ds[LAST_DS_LEN - 1] = '\0';
1158     }
1159     return 0;
1160 }
1161
1162 /*
1163  * How many PDP steps have elapsed since the last update? Returns the answer,
1164  * and stores the time between the last update and the last PDP in pre_time,
1165  * and the time between the last PDP and the current time in post_int.
1166  */
1167 static int calculate_elapsed_steps(
1168     rrd_t *rrd,
1169     unsigned long current_time,
1170     unsigned long current_time_usec,
1171     double interval,
1172     double *pre_int,
1173     double *post_int,
1174     unsigned long *proc_pdp_cnt)
1175 {
1176     unsigned long proc_pdp_st;  /* which pdp_st was the last to be processed */
1177     unsigned long occu_pdp_st;  /* when was the pdp_st before the last update
1178                                  * time */
1179     unsigned long proc_pdp_age; /* how old was the data in the pdp prep area
1180                                  * when it was last updated */
1181     unsigned long occu_pdp_age; /* how long ago was the last pdp_step time */
1182
1183     /* when was the current pdp started */
1184     proc_pdp_age = rrd->live_head->last_up % rrd->stat_head->pdp_step;
1185     proc_pdp_st = rrd->live_head->last_up - proc_pdp_age;
1186
1187     /* when did the last pdp_st occur */
1188     occu_pdp_age = current_time % rrd->stat_head->pdp_step;
1189     occu_pdp_st = current_time - occu_pdp_age;
1190
1191     if (occu_pdp_st > proc_pdp_st) {
1192         /* OK we passed the pdp_st moment */
1193         *pre_int = (long) occu_pdp_st - rrd->live_head->last_up;    /* how much of the input data
1194                                                                      * occurred before the latest
1195                                                                      * pdp_st moment*/
1196         *pre_int -= ((double) rrd->live_head->last_up_usec) / 1e6f; /* adjust usecs */
1197         *post_int = occu_pdp_age;   /* how much after it */
1198         *post_int += ((double) current_time_usec) / 1e6f;   /* adjust usecs */
1199     } else {
1200         *pre_int = interval;
1201         *post_int = 0;
1202     }
1203
1204     *proc_pdp_cnt = proc_pdp_st / rrd->stat_head->pdp_step;
1205
1206 #ifdef DEBUG
1207     printf("proc_pdp_age %lu\t"
1208            "proc_pdp_st %lu\t"
1209            "occu_pfp_age %lu\t"
1210            "occu_pdp_st %lu\t"
1211            "int %lf\t"
1212            "pre_int %lf\t"
1213            "post_int %lf\n", proc_pdp_age, proc_pdp_st,
1214            occu_pdp_age, occu_pdp_st, interval, *pre_int, *post_int);
1215 #endif
1216
1217     /* compute the number of elapsed pdp_st moments */
1218     return (occu_pdp_st - proc_pdp_st) / rrd->stat_head->pdp_step;
1219 }
1220
1221 /*
1222  * Increment the PDP values by the values in pdp_new, or else initialize them.
1223  */
1224 static void simple_update(
1225     rrd_t *rrd,
1226     double interval,
1227     rrd_value_t *pdp_new)
1228 {
1229     int       i;
1230
1231     for (i = 0; i < (signed) rrd->stat_head->ds_cnt; i++) {
1232         if (isnan(pdp_new[i])) {
1233             /* this is not really accurate if we use subsecond data arrival time
1234                should have thought of it when going subsecond resolution ...
1235                sorry next format change we will have it! */
1236             rrd->pdp_prep[i].scratch[PDP_unkn_sec_cnt].u_cnt +=
1237                 floor(interval);
1238         } else {
1239             if (isnan(rrd->pdp_prep[i].scratch[PDP_val].u_val)) {
1240                 rrd->pdp_prep[i].scratch[PDP_val].u_val = pdp_new[i];
1241             } else {
1242                 rrd->pdp_prep[i].scratch[PDP_val].u_val += pdp_new[i];
1243             }
1244         }
1245 #ifdef DEBUG
1246         fprintf(stderr,
1247                 "NO PDP  ds[%i]\t"
1248                 "value %10.2f\t"
1249                 "unkn_sec %5lu\n",
1250                 i,
1251                 rrd->pdp_prep[i].scratch[PDP_val].u_val,
1252                 rrd->pdp_prep[i].scratch[PDP_unkn_sec_cnt].u_cnt);
1253 #endif
1254     }
1255 }
1256
1257 /*
1258  * Call process_pdp_st for each DS.
1259  *
1260  * Returns 0 on success, -1 on error.
1261  */
1262 static int process_all_pdp_st(
1263     rrd_t *rrd,
1264     double interval,
1265     double pre_int,
1266     double post_int,
1267     unsigned long elapsed_pdp_st,
1268     rrd_value_t *pdp_new,
1269     rrd_value_t *pdp_temp)
1270 {
1271     unsigned long ds_idx;
1272
1273     /* in pdp_prep[].scratch[PDP_val].u_val we have collected
1274        rate*seconds which occurred up to the last run.
1275        pdp_new[] contains rate*seconds from the latest run.
1276        pdp_temp[] will contain the rate for cdp */
1277
1278     for (ds_idx = 0; ds_idx < rrd->stat_head->ds_cnt; ds_idx++) {
1279         if (process_pdp_st(rrd, ds_idx, interval, pre_int, post_int,
1280                            elapsed_pdp_st * rrd->stat_head->pdp_step,
1281                            pdp_new, pdp_temp) == -1) {
1282             return -1;
1283         }
1284 #ifdef DEBUG
1285         fprintf(stderr, "PDP UPD ds[%lu]\t"
1286                 "elapsed_pdp_st %lu\t"
1287                 "pdp_temp %10.2f\t"
1288                 "new_prep %10.2f\t"
1289                 "new_unkn_sec %5lu\n",
1290                 ds_idx,
1291                 elapsed_pdp_st,
1292                 pdp_temp[ds_idx],
1293                 rrd->pdp_prep[ds_idx].scratch[PDP_val].u_val,
1294                 rrd->pdp_prep[ds_idx].scratch[PDP_unkn_sec_cnt].u_cnt);
1295 #endif
1296     }
1297     return 0;
1298 }
1299
1300 /*
1301  * Process an update that occurs after one of the PDP moments.
1302  * Increments the PDP value, sets NAN if time greater than the
1303  * heartbeats have elapsed, processes CDEFs.
1304  *
1305  * Returns 0 on success, -1 on error.
1306  */
1307 static int process_pdp_st(
1308     rrd_t *rrd,
1309     unsigned long ds_idx,
1310     double interval,
1311     double pre_int,
1312     double post_int,
1313     long diff_pdp_st,   /* number of seconds in full steps passed since last update */
1314     rrd_value_t *pdp_new,
1315     rrd_value_t *pdp_temp)
1316 {
1317     int       i;
1318
1319     /* update pdp_prep to the current pdp_st. */
1320     double    pre_unknown = 0.0;
1321     unival   *scratch = rrd->pdp_prep[ds_idx].scratch;
1322     unsigned long mrhb = rrd->ds_def[ds_idx].par[DS_mrhb_cnt].u_cnt;
1323
1324     rpnstack_t rpnstack;    /* used for COMPUTE DS */
1325
1326     rpnstack_init(&rpnstack);
1327
1328
1329     if (isnan(pdp_new[ds_idx])) {
1330         /* a final bit of unknown to be added before calculation
1331            we use a temporary variable for this so that we
1332            don't have to turn integer lines before using the value */
1333         pre_unknown = pre_int;
1334     } else {
1335         if (isnan(scratch[PDP_val].u_val)) {
1336             scratch[PDP_val].u_val = 0;
1337         }
1338         scratch[PDP_val].u_val += pdp_new[ds_idx] / interval * pre_int;
1339     }
1340
1341     /* if too much of the pdp_prep is unknown we dump it */
1342     /* if the interval is larger thatn mrhb we get NAN */
1343     if ((interval > mrhb) ||
1344         (rrd->stat_head->pdp_step / 2.0 <
1345          (signed) scratch[PDP_unkn_sec_cnt].u_cnt)) {
1346         pdp_temp[ds_idx] = DNAN;
1347     } else {
1348         pdp_temp[ds_idx] = scratch[PDP_val].u_val /
1349             ((double) (diff_pdp_st - scratch[PDP_unkn_sec_cnt].u_cnt) -
1350              pre_unknown);
1351     }
1352
1353     /* process CDEF data sources; remember each CDEF DS can
1354      * only reference other DS with a lower index number */
1355     if (dst_conv(rrd->ds_def[ds_idx].dst) == DST_CDEF) {
1356         rpnp_t   *rpnp;
1357
1358         rpnp =
1359             rpn_expand((rpn_cdefds_t *) &(rrd->ds_def[ds_idx].par[DS_cdef]));
1360         /* substitute data values for OP_VARIABLE nodes */
1361         for (i = 0; rpnp[i].op != OP_END; i++) {
1362             if (rpnp[i].op == OP_VARIABLE) {
1363                 rpnp[i].op = OP_NUMBER;
1364                 rpnp[i].val = pdp_temp[rpnp[i].ptr];
1365             }
1366         }
1367         /* run the rpn calculator */
1368         if (rpn_calc(rpnp, &rpnstack, 0, pdp_temp, ds_idx) == -1) {
1369             free(rpnp);
1370             rpnstack_free(&rpnstack);
1371             return -1;
1372         }
1373     }
1374
1375     /* make pdp_prep ready for the next run */
1376     if (isnan(pdp_new[ds_idx])) {
1377         /* this is not realy accurate if we use subsecond data arival time
1378            should have thought of it when going subsecond resolution ...
1379            sorry next format change we will have it! */
1380         scratch[PDP_unkn_sec_cnt].u_cnt = floor(post_int);
1381         scratch[PDP_val].u_val = DNAN;
1382     } else {
1383         scratch[PDP_unkn_sec_cnt].u_cnt = 0;
1384         scratch[PDP_val].u_val = pdp_new[ds_idx] / interval * post_int;
1385     }
1386     rpnstack_free(&rpnstack);
1387     return 0;
1388 }
1389
1390 /*
1391  * Iterate over all the RRAs for a given DS and:
1392  * 1. Decide whether to schedule a smooth later
1393  * 2. Decide whether to skip updating SEASONAL and DEVSEASONAL
1394  * 3. Update the CDP
1395  *
1396  * Returns 0 on success, -1 on error
1397  */
1398 static int update_all_cdp_prep(
1399     rrd_t *rrd,
1400     unsigned long *rra_step_cnt,
1401     unsigned long rra_begin,
1402     rrd_file_t *rrd_file,
1403     unsigned long elapsed_pdp_st,
1404     unsigned long proc_pdp_cnt,
1405     rrd_value_t **last_seasonal_coef,
1406     rrd_value_t **seasonal_coef,
1407     rrd_value_t *pdp_temp,
1408     unsigned long *skip_update,
1409     int *schedule_smooth)
1410 {
1411     unsigned long rra_idx;
1412
1413     /* index into the CDP scratch array */
1414     enum cf_en current_cf;
1415     unsigned long rra_start;
1416
1417     /* number of rows to be updated in an RRA for a data value. */
1418     unsigned long start_pdp_offset;
1419
1420     rra_start = rra_begin;
1421     for (rra_idx = 0; rra_idx < rrd->stat_head->rra_cnt; rra_idx++) {
1422         current_cf = cf_conv(rrd->rra_def[rra_idx].cf_nam);
1423         start_pdp_offset =
1424             rrd->rra_def[rra_idx].pdp_cnt -
1425             proc_pdp_cnt % rrd->rra_def[rra_idx].pdp_cnt;
1426         skip_update[rra_idx] = 0;
1427         if (start_pdp_offset <= elapsed_pdp_st) {
1428             rra_step_cnt[rra_idx] = (elapsed_pdp_st - start_pdp_offset) /
1429                 rrd->rra_def[rra_idx].pdp_cnt + 1;
1430         } else {
1431             rra_step_cnt[rra_idx] = 0;
1432         }
1433
1434         if (current_cf == CF_SEASONAL || current_cf == CF_DEVSEASONAL) {
1435             /* If this is a bulk update, we need to skip ahead in the seasonal arrays
1436              * so that they will be correct for the next observed value; note that for
1437              * the bulk update itself, no update will occur to DEVSEASONAL or SEASONAL;
1438              * futhermore, HWPREDICT and DEVPREDICT will be set to DNAN. */
1439             if (rra_step_cnt[rra_idx] > 1) {
1440                 skip_update[rra_idx] = 1;
1441                 lookup_seasonal(rrd, rra_idx, rra_start, rrd_file,
1442                                 elapsed_pdp_st, last_seasonal_coef);
1443                 lookup_seasonal(rrd, rra_idx, rra_start, rrd_file,
1444                                 elapsed_pdp_st + 1, seasonal_coef);
1445             }
1446             /* periodically run a smoother for seasonal effects */
1447             if (do_schedule_smooth(rrd, rra_idx, elapsed_pdp_st)) {
1448 #ifdef DEBUG
1449                 fprintf(stderr,
1450                         "schedule_smooth: cur_row %lu, elapsed_pdp_st %lu, smooth idx %lu\n",
1451                         rrd->rra_ptr[rra_idx].cur_row, elapsed_pdp_st,
1452                         rrd->rra_def[rra_idx].par[RRA_seasonal_smooth_idx].
1453                         u_cnt);
1454 #endif
1455                 *schedule_smooth = 1;
1456             }
1457         }
1458         if (rrd_test_error())
1459             return -1;
1460
1461         if (update_cdp_prep
1462             (rrd, elapsed_pdp_st, start_pdp_offset, rra_step_cnt, rra_idx,
1463              pdp_temp, *last_seasonal_coef, *seasonal_coef,
1464              current_cf) == -1) {
1465             return -1;
1466         }
1467         rra_start +=
1468             rrd->rra_def[rra_idx].row_cnt * rrd->stat_head->ds_cnt *
1469             sizeof(rrd_value_t);
1470     }
1471     return 0;
1472 }
1473
1474 /*
1475  * Are we due for a smooth? Also increments our position in the burn-in cycle.
1476  */
1477 static int do_schedule_smooth(
1478     rrd_t *rrd,
1479     unsigned long rra_idx,
1480     unsigned long elapsed_pdp_st)
1481 {
1482     unsigned long cdp_idx = rra_idx * (rrd->stat_head->ds_cnt);
1483     unsigned long cur_row = rrd->rra_ptr[rra_idx].cur_row;
1484     unsigned long row_cnt = rrd->rra_def[rra_idx].row_cnt;
1485     unsigned long seasonal_smooth_idx =
1486         rrd->rra_def[rra_idx].par[RRA_seasonal_smooth_idx].u_cnt;
1487     unsigned long *init_seasonal =
1488         &(rrd->cdp_prep[cdp_idx].scratch[CDP_init_seasonal].u_cnt);
1489
1490     /* Need to use first cdp parameter buffer to track burnin (burnin requires
1491      * a specific smoothing schedule).  The CDP_init_seasonal parameter is
1492      * really an RRA level, not a data source within RRA level parameter, but
1493      * the rra_def is read only for rrd_update (not flushed to disk). */
1494     if (*init_seasonal > BURNIN_CYCLES) {
1495         /* someone has no doubt invented a trick to deal with this wrap around,
1496          * but at least this code is clear. */
1497         if (seasonal_smooth_idx > cur_row) {
1498             /* here elapsed_pdp_st = rra_step_cnt[rra_idx] because of 1-1 mapping
1499              * between PDP and CDP */
1500             return (cur_row + elapsed_pdp_st >= seasonal_smooth_idx);
1501         }
1502         /* can't rely on negative numbers because we are working with
1503          * unsigned values */
1504         return (cur_row + elapsed_pdp_st >= row_cnt
1505                 && cur_row + elapsed_pdp_st >= row_cnt + seasonal_smooth_idx);
1506     }
1507     /* mark off one of the burn-in cycles */
1508     return (cur_row + elapsed_pdp_st >= row_cnt && ++(*init_seasonal));
1509 }
1510
1511 /*
1512  * For a given RRA, iterate over the data sources and call the appropriate
1513  * consolidation function.
1514  *
1515  * Returns 0 on success, -1 on error.
1516  */
1517 static int update_cdp_prep(
1518     rrd_t *rrd,
1519     unsigned long elapsed_pdp_st,
1520     unsigned long start_pdp_offset,
1521     unsigned long *rra_step_cnt,
1522     int rra_idx,
1523     rrd_value_t *pdp_temp,
1524     rrd_value_t *last_seasonal_coef,
1525     rrd_value_t *seasonal_coef,
1526     int current_cf)
1527 {
1528     unsigned long ds_idx, cdp_idx;
1529
1530     /* update CDP_PREP areas */
1531     /* loop over data soures within each RRA */
1532     for (ds_idx = 0; ds_idx < rrd->stat_head->ds_cnt; ds_idx++) {
1533
1534         cdp_idx = rra_idx * rrd->stat_head->ds_cnt + ds_idx;
1535
1536         if (rrd->rra_def[rra_idx].pdp_cnt > 1) {
1537             update_cdp(rrd->cdp_prep[cdp_idx].scratch, current_cf,
1538                        pdp_temp[ds_idx], rra_step_cnt[rra_idx],
1539                        elapsed_pdp_st, start_pdp_offset,
1540                        rrd->rra_def[rra_idx].pdp_cnt,
1541                        rrd->rra_def[rra_idx].par[RRA_cdp_xff_val].u_val,
1542                        rra_idx, ds_idx);
1543         } else {
1544             /* Nothing to consolidate if there's one PDP per CDP. However, if
1545              * we've missed some PDPs, let's update null counters etc. */
1546             if (elapsed_pdp_st > 2) {
1547                 reset_cdp(rrd, elapsed_pdp_st, pdp_temp, last_seasonal_coef,
1548                           seasonal_coef, rra_idx, ds_idx, cdp_idx,
1549                           current_cf);
1550             }
1551         }
1552
1553         if (rrd_test_error())
1554             return -1;
1555     }                   /* endif data sources loop */
1556     return 0;
1557 }
1558
1559 /*
1560  * Given the new reading (pdp_temp_val), update or initialize the CDP value,
1561  * primary value, secondary value, and # of unknowns.
1562  */
1563 static void update_cdp(
1564     unival *scratch,
1565     int current_cf,
1566     rrd_value_t pdp_temp_val,
1567     unsigned long rra_step_cnt,
1568     unsigned long elapsed_pdp_st,
1569     unsigned long start_pdp_offset,
1570     unsigned long pdp_cnt,
1571     rrd_value_t xff,
1572     int i,
1573     int ii)
1574 {
1575     /* shorthand variables */
1576     rrd_value_t *cdp_val = &scratch[CDP_val].u_val;
1577     rrd_value_t *cdp_primary_val = &scratch[CDP_primary_val].u_val;
1578     rrd_value_t *cdp_secondary_val = &scratch[CDP_secondary_val].u_val;
1579     unsigned long *cdp_unkn_pdp_cnt = &scratch[CDP_unkn_pdp_cnt].u_cnt;
1580
1581     if (rra_step_cnt) {
1582         /* If we are in this block, as least 1 CDP value will be written to
1583          * disk, this is the CDP_primary_val entry. If more than 1 value needs
1584          * to be written, then the "fill in" value is the CDP_secondary_val
1585          * entry. */
1586         if (isnan(pdp_temp_val)) {
1587             *cdp_unkn_pdp_cnt += start_pdp_offset;
1588             *cdp_secondary_val = DNAN;
1589         } else {
1590             /* CDP_secondary value is the RRA "fill in" value for intermediary
1591              * CDP data entries. No matter the CF, the value is the same because
1592              * the average, max, min, and last of a list of identical values is
1593              * the same, namely, the value itself. */
1594             *cdp_secondary_val = pdp_temp_val;
1595         }
1596
1597         if (*cdp_unkn_pdp_cnt > pdp_cnt * xff) {
1598             *cdp_primary_val = DNAN;
1599             if (current_cf == CF_AVERAGE) {
1600                 *cdp_val =
1601                     initialize_average_carry_over(pdp_temp_val,
1602                                                   elapsed_pdp_st,
1603                                                   start_pdp_offset, pdp_cnt);
1604             } else {
1605                 *cdp_val = pdp_temp_val;
1606             }
1607         } else {
1608             initialize_cdp_val(scratch, current_cf, pdp_temp_val,
1609                                elapsed_pdp_st, start_pdp_offset, pdp_cnt);
1610         }               /* endif meets xff value requirement for a valid value */
1611         /* initialize carry over CDP_unkn_pdp_cnt, this must after CDP_primary_val
1612          * is set because CDP_unkn_pdp_cnt is required to compute that value. */
1613         if (isnan(pdp_temp_val))
1614             *cdp_unkn_pdp_cnt = (elapsed_pdp_st - start_pdp_offset) % pdp_cnt;
1615         else
1616             *cdp_unkn_pdp_cnt = 0;
1617     } else {            /* rra_step_cnt[i]  == 0 */
1618
1619 #ifdef DEBUG
1620         if (isnan(*cdp_val)) {
1621             fprintf(stderr, "schedule CDP_val update, RRA %d DS %d, DNAN\n",
1622                     i, ii);
1623         } else {
1624             fprintf(stderr, "schedule CDP_val update, RRA %d DS %d, %10.2f\n",
1625                     i, ii, *cdp_val);
1626         }
1627 #endif
1628         if (isnan(pdp_temp_val)) {
1629             *cdp_unkn_pdp_cnt += elapsed_pdp_st;
1630         } else {
1631             *cdp_val =
1632                 calculate_cdp_val(*cdp_val, pdp_temp_val, elapsed_pdp_st,
1633                                   current_cf, i, ii);
1634         }
1635     }
1636 }
1637
1638 /*
1639  * Set the CDP_primary_val and CDP_val to the appropriate initial value based
1640  * on the type of consolidation function.
1641  */
1642 static void initialize_cdp_val(
1643     unival *scratch,
1644     int current_cf,
1645     rrd_value_t pdp_temp_val,
1646     unsigned long elapsed_pdp_st,
1647     unsigned long start_pdp_offset,
1648     unsigned long pdp_cnt)
1649 {
1650     rrd_value_t cum_val, cur_val;
1651
1652     switch (current_cf) {
1653     case CF_AVERAGE:
1654         cum_val = IFDNAN(scratch[CDP_val].u_val, 0.0);
1655         cur_val = IFDNAN(pdp_temp_val, 0.0);
1656         scratch[CDP_primary_val].u_val =
1657             (cum_val + cur_val * start_pdp_offset) /
1658             (pdp_cnt - scratch[CDP_unkn_pdp_cnt].u_cnt);
1659         scratch[CDP_val].u_val =
1660             initialize_average_carry_over(pdp_temp_val, elapsed_pdp_st,
1661                                           start_pdp_offset, pdp_cnt);
1662         break;
1663     case CF_MAXIMUM:
1664         cum_val = IFDNAN(scratch[CDP_val].u_val, -DINF);
1665         cur_val = IFDNAN(pdp_temp_val, -DINF);
1666 #if 0
1667 #ifdef DEBUG
1668         if (isnan(scratch[CDP_val].u_val) && isnan(pdp_temp)) {
1669             fprintf(stderr,
1670                     "RRA %lu, DS %lu, both CDP_val and pdp_temp are DNAN!",
1671                     i, ii);
1672             exit(-1);
1673         }
1674 #endif
1675 #endif
1676         if (cur_val > cum_val)
1677             scratch[CDP_primary_val].u_val = cur_val;
1678         else
1679             scratch[CDP_primary_val].u_val = cum_val;
1680         /* initialize carry over value */
1681         scratch[CDP_val].u_val = pdp_temp_val;
1682         break;
1683     case CF_MINIMUM:
1684         cum_val = IFDNAN(scratch[CDP_val].u_val, DINF);
1685         cur_val = IFDNAN(pdp_temp_val, DINF);
1686 #if 0
1687 #ifdef DEBUG
1688         if (isnan(scratch[CDP_val].u_val) && isnan(pdp_temp)) {
1689             fprintf(stderr,
1690                     "RRA %lu, DS %lu, both CDP_val and pdp_temp are DNAN!", i,
1691                     ii);
1692             exit(-1);
1693         }
1694 #endif
1695 #endif
1696         if (cur_val < cum_val)
1697             scratch[CDP_primary_val].u_val = cur_val;
1698         else
1699             scratch[CDP_primary_val].u_val = cum_val;
1700         /* initialize carry over value */
1701         scratch[CDP_val].u_val = pdp_temp_val;
1702         break;
1703     case CF_LAST:
1704     default:
1705         scratch[CDP_primary_val].u_val = pdp_temp_val;
1706         /* initialize carry over value */
1707         scratch[CDP_val].u_val = pdp_temp_val;
1708         break;
1709     }
1710 }
1711
1712 /*
1713  * Update the consolidation function for Holt-Winters functions as
1714  * well as other functions that don't actually consolidate multiple
1715  * PDPs.
1716  */
1717 static void reset_cdp(
1718     rrd_t *rrd,
1719     unsigned long elapsed_pdp_st,
1720     rrd_value_t *pdp_temp,
1721     rrd_value_t *last_seasonal_coef,
1722     rrd_value_t *seasonal_coef,
1723     int rra_idx,
1724     int ds_idx,
1725     int cdp_idx,
1726     enum cf_en current_cf)
1727 {
1728     unival   *scratch = rrd->cdp_prep[cdp_idx].scratch;
1729
1730     switch (current_cf) {
1731     case CF_AVERAGE:
1732     default:
1733         scratch[CDP_primary_val].u_val = pdp_temp[ds_idx];
1734         scratch[CDP_secondary_val].u_val = pdp_temp[ds_idx];
1735         break;
1736     case CF_SEASONAL:
1737     case CF_DEVSEASONAL:
1738         /* need to update cached seasonal values, so they are consistent
1739          * with the bulk update */
1740         /* WARNING: code relies on the fact that CDP_hw_last_seasonal and
1741          * CDP_last_deviation are the same. */
1742         scratch[CDP_hw_last_seasonal].u_val = last_seasonal_coef[ds_idx];
1743         scratch[CDP_hw_seasonal].u_val = seasonal_coef[ds_idx];
1744         break;
1745     case CF_HWPREDICT:
1746     case CF_MHWPREDICT:
1747         /* need to update the null_count and last_null_count.
1748          * even do this for non-DNAN pdp_temp because the
1749          * algorithm is not learning from batch updates. */
1750         scratch[CDP_null_count].u_cnt += elapsed_pdp_st;
1751         scratch[CDP_last_null_count].u_cnt += elapsed_pdp_st - 1;
1752         /* fall through */
1753     case CF_DEVPREDICT:
1754         scratch[CDP_primary_val].u_val = DNAN;
1755         scratch[CDP_secondary_val].u_val = DNAN;
1756         break;
1757     case CF_FAILURES:
1758         /* do not count missed bulk values as failures */
1759         scratch[CDP_primary_val].u_val = 0;
1760         scratch[CDP_secondary_val].u_val = 0;
1761         /* need to reset violations buffer.
1762          * could do this more carefully, but for now, just
1763          * assume a bulk update wipes away all violations. */
1764         erase_violations(rrd, cdp_idx, rra_idx);
1765         break;
1766     }
1767 }
1768
1769 static rrd_value_t initialize_average_carry_over(
1770     rrd_value_t pdp_temp_val,
1771     unsigned long elapsed_pdp_st,
1772     unsigned long start_pdp_offset,
1773     unsigned long pdp_cnt)
1774 {
1775     /* initialize carry over value */
1776     if (isnan(pdp_temp_val)) {
1777         return DNAN;
1778     }
1779     return pdp_temp_val * ((elapsed_pdp_st - start_pdp_offset) % pdp_cnt);
1780 }
1781
1782 /*
1783  * Update or initialize a CDP value based on the consolidation
1784  * function.
1785  *
1786  * Returns the new value.
1787  */
1788 static rrd_value_t calculate_cdp_val(
1789     rrd_value_t cdp_val,
1790     rrd_value_t pdp_temp_val,
1791     unsigned long elapsed_pdp_st,
1792     int current_cf,
1793 #ifdef DEBUG
1794     int i,
1795     int ii
1796 #else
1797     int UNUSED(i),
1798     int UNUSED(ii)
1799 #endif
1800     )
1801 {
1802     if (isnan(cdp_val)) {
1803         if (current_cf == CF_AVERAGE) {
1804             pdp_temp_val *= elapsed_pdp_st;
1805         }
1806 #ifdef DEBUG
1807         fprintf(stderr, "Initialize CDP_val for RRA %d DS %d: %10.2f\n",
1808                 i, ii, pdp_temp_val);
1809 #endif
1810         return pdp_temp_val;
1811     }
1812     if (current_cf == CF_AVERAGE)
1813         return cdp_val + pdp_temp_val * elapsed_pdp_st;
1814     if (current_cf == CF_MINIMUM)
1815         return (pdp_temp_val < cdp_val) ? pdp_temp_val : cdp_val;
1816     if (current_cf == CF_MAXIMUM)
1817         return (pdp_temp_val > cdp_val) ? pdp_temp_val : cdp_val;
1818
1819     return pdp_temp_val;
1820 }
1821
1822 /*
1823  * For each RRA, update the seasonal values and then call update_aberrant_CF
1824  * for each data source.
1825  *
1826  * Return 0 on success, -1 on error.
1827  */
1828 static int update_aberrant_cdps(
1829     rrd_t *rrd,
1830     rrd_file_t *rrd_file,
1831     unsigned long rra_begin,
1832     unsigned long elapsed_pdp_st,
1833     rrd_value_t *pdp_temp,
1834     rrd_value_t **seasonal_coef)
1835 {
1836     unsigned long rra_idx, ds_idx, j;
1837
1838     /* number of PDP steps since the last update that
1839      * are assigned to the first CDP to be generated
1840      * since the last update. */
1841     unsigned short scratch_idx;
1842     unsigned long rra_start;
1843     enum cf_en current_cf;
1844
1845     /* this loop is only entered if elapsed_pdp_st < 3 */
1846     for (j = elapsed_pdp_st, scratch_idx = CDP_primary_val;
1847          j > 0 && j < 3; j--, scratch_idx = CDP_secondary_val) {
1848         rra_start = rra_begin;
1849         for (rra_idx = 0; rra_idx < rrd->stat_head->rra_cnt; rra_idx++) {
1850             if (rrd->rra_def[rra_idx].pdp_cnt == 1) {
1851                 current_cf = cf_conv(rrd->rra_def[rra_idx].cf_nam);
1852                 if (current_cf == CF_SEASONAL || current_cf == CF_DEVSEASONAL) {
1853                     if (scratch_idx == CDP_primary_val) {
1854                         lookup_seasonal(rrd, rra_idx, rra_start, rrd_file,
1855                                         elapsed_pdp_st + 1, seasonal_coef);
1856                     } else {
1857                         lookup_seasonal(rrd, rra_idx, rra_start, rrd_file,
1858                                         elapsed_pdp_st + 2, seasonal_coef);
1859                     }
1860                 }
1861                 if (rrd_test_error())
1862                     return -1;
1863                 /* loop over data soures within each RRA */
1864                 for (ds_idx = 0; ds_idx < rrd->stat_head->ds_cnt; ds_idx++) {
1865                     update_aberrant_CF(rrd, pdp_temp[ds_idx], current_cf,
1866                                        rra_idx * (rrd->stat_head->ds_cnt) +
1867                                        ds_idx, rra_idx, ds_idx, scratch_idx,
1868                                        *seasonal_coef);
1869                 }
1870             }
1871             rra_start += rrd->rra_def[rra_idx].row_cnt
1872                 * rrd->stat_head->ds_cnt * sizeof(rrd_value_t);
1873         }
1874     }
1875     return 0;
1876 }
1877
1878 /*
1879  * Move sequentially through the file, writing one RRA at a time.  Note this
1880  * architecture divorces the computation of CDP with flushing updated RRA
1881  * entries to disk.
1882  *
1883  * Return 0 on success, -1 on error.
1884  */
1885 static int write_to_rras(
1886     rrd_t *rrd,
1887     rrd_file_t *rrd_file,
1888     unsigned long *rra_step_cnt,
1889     unsigned long rra_begin,
1890     time_t current_time,
1891     unsigned long *skip_update,
1892     rrd_info_t ** pcdp_summary)
1893 {
1894     unsigned long rra_idx;
1895     unsigned long rra_start;
1896     time_t    rra_time = 0; /* time of update for a RRA */
1897
1898     unsigned long ds_cnt = rrd->stat_head->ds_cnt;
1899
1900     /* Ready to write to disk */
1901     rra_start = rra_begin;
1902
1903     for (rra_idx = 0; rra_idx < rrd->stat_head->rra_cnt; rra_idx++) {
1904         rra_def_t *rra_def = &rrd->rra_def[rra_idx];
1905         rra_ptr_t *rra_ptr = &rrd->rra_ptr[rra_idx];
1906
1907         /* for cdp_prep */
1908         unsigned short scratch_idx;
1909         unsigned long step_subtract;
1910
1911         for (scratch_idx = CDP_primary_val,
1912                  step_subtract = 1;
1913              rra_step_cnt[rra_idx] > 0;
1914              rra_step_cnt[rra_idx]--,
1915                  scratch_idx = CDP_secondary_val,
1916                  step_subtract = 2) {
1917
1918             size_t rra_pos_new;
1919 #ifdef DEBUG
1920             fprintf(stderr, "  -- RRA Preseek %ld\n", rrd_file->pos);
1921 #endif
1922             /* increment, with wrap-around */
1923             if (++rra_ptr->cur_row >= rra_def->row_cnt)
1924               rra_ptr->cur_row = 0;
1925
1926             /* we know what our position should be */
1927             rra_pos_new = rra_start
1928               + ds_cnt * rra_ptr->cur_row * sizeof(rrd_value_t);
1929
1930             /* re-seek if the position is wrong or we wrapped around */
1931             if ((size_t)rra_pos_new != rrd_file->pos) {
1932                 if (rrd_seek(rrd_file, rra_pos_new, SEEK_SET) != 0) {
1933                     rrd_set_error("seek error in rrd");
1934                     return -1;
1935                 }
1936             }
1937 #ifdef DEBUG
1938             fprintf(stderr, "  -- RRA Postseek %ld\n", rrd_file->pos);
1939 #endif
1940
1941             if (skip_update[rra_idx])
1942                 continue;
1943
1944             if (*pcdp_summary != NULL) {
1945                 unsigned long step_time = rra_def->pdp_cnt * rrd->stat_head->pdp_step;
1946
1947                 rra_time = (current_time - current_time % step_time)
1948                     - ((rra_step_cnt[rra_idx] - step_subtract) * step_time);
1949             }
1950
1951             if (write_RRA_row
1952                 (rrd_file, rrd, rra_idx, scratch_idx,
1953                  pcdp_summary, rra_time) == -1)
1954                 return -1;
1955
1956             rrd_notify_row(rrd_file, rra_idx, rra_pos_new, rra_time);
1957         }
1958
1959         rra_start += rra_def->row_cnt * ds_cnt * sizeof(rrd_value_t);
1960     } /* RRA LOOP */
1961
1962     return 0;
1963 }
1964
1965 /*
1966  * Write out one row of values (one value per DS) to the archive.
1967  *
1968  * Returns 0 on success, -1 on error.
1969  */
1970 static int write_RRA_row(
1971     rrd_file_t *rrd_file,
1972     rrd_t *rrd,
1973     unsigned long rra_idx,
1974     unsigned short CDP_scratch_idx,
1975     rrd_info_t ** pcdp_summary,
1976     time_t rra_time)
1977 {
1978     unsigned long ds_idx, cdp_idx;
1979     rrd_infoval_t iv;
1980
1981     for (ds_idx = 0; ds_idx < rrd->stat_head->ds_cnt; ds_idx++) {
1982         /* compute the cdp index */
1983         cdp_idx = rra_idx * (rrd->stat_head->ds_cnt) + ds_idx;
1984 #ifdef DEBUG
1985         fprintf(stderr, "  -- RRA WRITE VALUE %e, at %ld CF:%s\n",
1986                 rrd->cdp_prep[cdp_idx].scratch[CDP_scratch_idx].u_val,
1987                 rrd_file->pos, rrd->rra_def[rra_idx].cf_nam);
1988 #endif
1989         if (*pcdp_summary != NULL) {
1990             iv.u_val = rrd->cdp_prep[cdp_idx].scratch[CDP_scratch_idx].u_val;
1991             /* append info to the return hash */
1992             *pcdp_summary = rrd_info_push(*pcdp_summary,
1993                                           sprintf_alloc
1994                                           ("[%lli]RRA[%s][%lu]DS[%s]",
1995                                            (long long)rra_time,
1996                                            rrd->rra_def[rra_idx].cf_nam,
1997                                            rrd->rra_def[rra_idx].pdp_cnt,
1998                                            rrd->ds_def[ds_idx].ds_nam),
1999                                            RD_I_VAL, iv);
2000         }
2001         errno = 0;
2002         if (rrd_write(rrd_file,
2003                       &(rrd->cdp_prep[cdp_idx].scratch[CDP_scratch_idx].
2004                         u_val), sizeof(rrd_value_t)) != sizeof(rrd_value_t)) {
2005             rrd_set_error("writing rrd: %s", rrd_strerror(errno));
2006             return -1;
2007         }
2008     }
2009     return 0;
2010 }
2011
2012 /*
2013  * Call apply_smoother for all DEVSEASONAL and SEASONAL RRAs.
2014  *
2015  * Returns 0 on success, -1 otherwise
2016  */
2017 static int smooth_all_rras(
2018     rrd_t *rrd,
2019     rrd_file_t *rrd_file,
2020     unsigned long rra_begin)
2021 {
2022     unsigned long rra_start = rra_begin;
2023     unsigned long rra_idx;
2024
2025     for (rra_idx = 0; rra_idx < rrd->stat_head->rra_cnt; ++rra_idx) {
2026         if (cf_conv(rrd->rra_def[rra_idx].cf_nam) == CF_DEVSEASONAL ||
2027             cf_conv(rrd->rra_def[rra_idx].cf_nam) == CF_SEASONAL) {
2028 #ifdef DEBUG
2029             fprintf(stderr, "Running smoother for rra %lu\n", rra_idx);
2030 #endif
2031             apply_smoother(rrd, rra_idx, rra_start, rrd_file);
2032             if (rrd_test_error())
2033                 return -1;
2034         }
2035         rra_start += rrd->rra_def[rra_idx].row_cnt
2036             * rrd->stat_head->ds_cnt * sizeof(rrd_value_t);
2037     }
2038     return 0;
2039 }
2040
2041 #ifndef HAVE_MMAP
2042 /*
2043  * Flush changes to disk (unless we're using mmap)
2044  *
2045  * Returns 0 on success, -1 otherwise
2046  */
2047 static int write_changes_to_disk(
2048     rrd_t *rrd,
2049     rrd_file_t *rrd_file,
2050     int version)
2051 {
2052     /* we just need to write back the live header portion now */
2053     if (rrd_seek(rrd_file, (sizeof(stat_head_t)
2054                             + sizeof(ds_def_t) * rrd->stat_head->ds_cnt
2055                             + sizeof(rra_def_t) * rrd->stat_head->rra_cnt),
2056                  SEEK_SET) != 0) {
2057         rrd_set_error("seek rrd for live header writeback");
2058         return -1;
2059     }
2060     if (version >= 3) {
2061         if (rrd_write(rrd_file, rrd->live_head,
2062                       sizeof(live_head_t) * 1) != sizeof(live_head_t) * 1) {
2063             rrd_set_error("rrd_write live_head to rrd");
2064             return -1;
2065         }
2066     } else {
2067         if (rrd_write(rrd_file, rrd->legacy_last_up,
2068                       sizeof(time_t) * 1) != sizeof(time_t) * 1) {
2069             rrd_set_error("rrd_write live_head to rrd");
2070             return -1;
2071         }
2072     }
2073
2074
2075     if (rrd_write(rrd_file, rrd->pdp_prep,
2076                   sizeof(pdp_prep_t) * rrd->stat_head->ds_cnt)
2077         != (ssize_t) (sizeof(pdp_prep_t) * rrd->stat_head->ds_cnt)) {
2078         rrd_set_error("rrd_write pdp_prep to rrd");
2079         return -1;
2080     }
2081
2082     if (rrd_write(rrd_file, rrd->cdp_prep,
2083                   sizeof(cdp_prep_t) * rrd->stat_head->rra_cnt *
2084                   rrd->stat_head->ds_cnt)
2085         != (ssize_t) (sizeof(cdp_prep_t) * rrd->stat_head->rra_cnt *
2086                       rrd->stat_head->ds_cnt)) {
2087
2088         rrd_set_error("rrd_write cdp_prep to rrd");
2089         return -1;
2090     }
2091
2092     if (rrd_write(rrd_file, rrd->rra_ptr,
2093                   sizeof(rra_ptr_t) * rrd->stat_head->rra_cnt)
2094         != (ssize_t) (sizeof(rra_ptr_t) * rrd->stat_head->rra_cnt)) {
2095         rrd_set_error("rrd_write rra_ptr to rrd");
2096         return -1;
2097     }
2098     return 0;
2099 }
2100 #endif