Code

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