![[tokkee]](http://tokkee.org/images/avatar.png){kind=avatar}
200f412071b42e75688b40b13139df15d5213d75
2 /*****************************************************************************
3 * RRDtool 1.3.9 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(¤t_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)
289 {
290 if (t->tv_usec < 0) {
291 t->tv_sec--;
292 t->tv_usec += 1e6L;
293 }
294 }
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)
304 {
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 }
315 }
317 #define IFDNAN(X,Y) (isnan(X) ? (Y) : (X));
319 rrd_info_t *rrd_update_v(
320 int argc,
321 char **argv)
322 {
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;
368 }
370 int rrd_update(
371 int argc,
372 char **argv)
373 {
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;
414 }
416 int rrd_update_r(
417 const char *filename,
418 const char *tmplt,
419 int argc,
420 const char **argv)
421 {
422 return _rrd_update(filename, tmplt, argc, argv, NULL);
423 }
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)
431 {
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(¤t_time, ¤t_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 ¤t_time, ¤t_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;
561 }
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)
571 {
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);
596 }
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)
616 {
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;
687 }
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)
699 {
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;
737 }
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)
762 {
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;
839 }
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)
856 {
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 else {
888 rrd_set_error("found extra data on update argument: %s",p+1);
889 return -1;
890 }
891 }
892 }
894 if (i != tmpl_cnt) {
895 rrd_set_error("expected %lu data source readings (got %lu) from %s",
896 tmpl_cnt - 1, i - 1, input);
897 return -1;
898 }
900 if (get_time_from_reading(rrd, timesyntax, updvals,
901 current_time, current_time_usec,
902 version) == -1) {
903 return -1;
904 }
905 return 0;
906 }
908 /*
909 * Parse the time in a DS string, store it in current_time and
910 * current_time_usec and verify that it's later than the last
911 * update for this DS.
912 *
913 * Returns 0 on success, -1 on error.
914 */
915 static int get_time_from_reading(
916 rrd_t *rrd,
917 char timesyntax,
918 char **updvals,
919 time_t *current_time,
920 unsigned long *current_time_usec,
921 int version)
922 {
923 double tmp;
924 char *parsetime_error = NULL;
925 char *old_locale;
926 rrd_time_value_t ds_tv;
927 struct timeval tmp_time; /* used for time conversion */
929 /* get the time from the reading ... handle N */
930 if (timesyntax == '@') { /* at-style */
931 if ((parsetime_error = rrd_parsetime(updvals[0], &ds_tv))) {
932 rrd_set_error("ds time: %s: %s", updvals[0], parsetime_error);
933 return -1;
934 }
935 if (ds_tv.type == RELATIVE_TO_END_TIME ||
936 ds_tv.type == RELATIVE_TO_START_TIME) {
937 rrd_set_error("specifying time relative to the 'start' "
938 "or 'end' makes no sense here: %s", updvals[0]);
939 return -1;
940 }
941 *current_time = mktime(&ds_tv.tm) +ds_tv.offset;
942 *current_time_usec = 0; /* FIXME: how to handle usecs here ? */
943 } else if (strcmp(updvals[0], "N") == 0) {
944 gettimeofday(&tmp_time, 0);
945 normalize_time(&tmp_time);
946 *current_time = tmp_time.tv_sec;
947 *current_time_usec = tmp_time.tv_usec;
948 } else {
949 old_locale = setlocale(LC_NUMERIC, "C");
950 errno = 0;
951 tmp = strtod(updvals[0], 0);
952 if (errno > 0) {
953 rrd_set_error("converting '%s' to float: %s",
954 updvals[0], rrd_strerror(errno));
955 return -1;
956 };
957 setlocale(LC_NUMERIC, old_locale);
958 if (tmp < 0.0){
959 gettimeofday(&tmp_time, 0);
960 tmp = (double)tmp_time.tv_sec + (double)tmp_time.tv_usec * 1e-6f + tmp;
961 }
963 *current_time = floor(tmp);
964 *current_time_usec = (long) ((tmp - (double) *current_time) * 1e6f);
965 }
966 /* dont do any correction for old version RRDs */
967 if (version < 3)
968 *current_time_usec = 0;
970 if (*current_time < rrd->live_head->last_up ||
971 (*current_time == rrd->live_head->last_up &&
972 (long) *current_time_usec <= (long) rrd->live_head->last_up_usec)) {
973 rrd_set_error("illegal attempt to update using time %ld when "
974 "last update time is %ld (minimum one second step)",
975 *current_time, rrd->live_head->last_up);
976 return -1;
977 }
978 return 0;
979 }
981 /*
982 * Update pdp_new by interpreting the updvals according to the DS type
983 * (COUNTER, GAUGE, etc.).
984 *
985 * Returns 0 on success, -1 on error.
986 */
987 static int update_pdp_prep(
988 rrd_t *rrd,
989 char **updvals,
990 rrd_value_t *pdp_new,
991 double interval)
992 {
993 unsigned long ds_idx;
994 int ii;
995 char *endptr; /* used in the conversion */
996 double rate;
997 char *old_locale;
998 enum dst_en dst_idx;
1000 for (ds_idx = 0; ds_idx < rrd->stat_head->ds_cnt; ds_idx++) {
1001 dst_idx = dst_conv(rrd->ds_def[ds_idx].dst);
1003 /* make sure we do not build diffs with old last_ds values */
1004 if (rrd->ds_def[ds_idx].par[DS_mrhb_cnt].u_cnt < interval) {
1005 strncpy(rrd->pdp_prep[ds_idx].last_ds, "U", LAST_DS_LEN - 1);
1006 rrd->pdp_prep[ds_idx].last_ds[LAST_DS_LEN - 1] = '\0';
1007 }
1009 /* NOTE: DST_CDEF should never enter this if block, because
1010 * updvals[ds_idx+1][0] is initialized to 'U'; unless the caller
1011 * accidently specified a value for the DST_CDEF. To handle this case,
1012 * an extra check is required. */
1014 if ((updvals[ds_idx + 1][0] != 'U') &&
1015 (dst_idx != DST_CDEF) &&
1016 rrd->ds_def[ds_idx].par[DS_mrhb_cnt].u_cnt >= interval) {
1017 rate = DNAN;
1019 /* pdp_new contains rate * time ... eg the bytes transferred during
1020 * the interval. Doing it this way saves a lot of math operations
1021 */
1022 switch (dst_idx) {
1023 case DST_COUNTER:
1024 case DST_DERIVE:
1025 /* Check if this is a valid integer. `U' is already handled in
1026 * another branch. */
1027 for (ii = 0; updvals[ds_idx + 1][ii] != 0; ii++) {
1028 if ((ii == 0) && (dst_idx == DST_DERIVE)
1029 && (updvals[ds_idx + 1][ii] == '-'))
1030 continue;
1032 if ((updvals[ds_idx + 1][ii] < '0')
1033 || (updvals[ds_idx + 1][ii] > '9')) {
1034 rrd_set_error("not a simple %s integer: '%s'",
1035 (dst_idx == DST_DERIVE) ? "signed" : "unsigned",
1036 updvals[ds_idx + 1]);
1037 return -1;
1038 }
1039 } /* for (ii = 0; updvals[ds_idx + 1][ii] != 0; ii++) */
1041 if (rrd->pdp_prep[ds_idx].last_ds[0] != 'U') {
1042 pdp_new[ds_idx] =
1043 rrd_diff(updvals[ds_idx + 1],
1044 rrd->pdp_prep[ds_idx].last_ds);
1045 if (dst_idx == DST_COUNTER) {
1046 /* simple overflow catcher. This will fail
1047 * terribly for non 32 or 64 bit counters
1048 * ... are there any others in SNMP land?
1049 */
1050 if (pdp_new[ds_idx] < (double) 0.0)
1051 pdp_new[ds_idx] += (double) 4294967296.0; /* 2^32 */
1052 if (pdp_new[ds_idx] < (double) 0.0)
1053 pdp_new[ds_idx] += (double) 18446744069414584320.0; /* 2^64-2^32 */
1054 }
1055 rate = pdp_new[ds_idx] / interval;
1056 } else {
1057 pdp_new[ds_idx] = DNAN;
1058 }
1059 break;
1060 case DST_ABSOLUTE:
1061 old_locale = setlocale(LC_NUMERIC, "C");
1062 errno = 0;
1063 pdp_new[ds_idx] = strtod(updvals[ds_idx + 1], &endptr);
1064 if (errno > 0) {
1065 rrd_set_error("converting '%s' to float: %s",
1066 updvals[ds_idx + 1], rrd_strerror(errno));
1067 return -1;
1068 };
1069 setlocale(LC_NUMERIC, old_locale);
1070 if (endptr[0] != '\0') {
1071 rrd_set_error
1072 ("conversion of '%s' to float not complete: tail '%s'",
1073 updvals[ds_idx + 1], endptr);
1074 return -1;
1075 }
1076 rate = pdp_new[ds_idx] / interval;
1077 break;
1078 case DST_GAUGE:
1079 old_locale = setlocale(LC_NUMERIC, "C");
1080 errno = 0;
1081 pdp_new[ds_idx] =
1082 strtod(updvals[ds_idx + 1], &endptr) * interval;
1083 if (errno) {
1084 rrd_set_error("converting '%s' to float: %s",
1085 updvals[ds_idx + 1], rrd_strerror(errno));
1086 return -1;
1087 };
1088 setlocale(LC_NUMERIC, old_locale);
1089 if (endptr[0] != '\0') {
1090 rrd_set_error
1091 ("conversion of '%s' to float not complete: tail '%s'",
1092 updvals[ds_idx + 1], endptr);
1093 return -1;
1094 }
1095 rate = pdp_new[ds_idx] / interval;
1096 break;
1097 default:
1098 rrd_set_error("rrd contains unknown DS type : '%s'",
1099 rrd->ds_def[ds_idx].dst);
1100 return -1;
1101 }
1102 /* break out of this for loop if the error string is set */
1103 if (rrd_test_error()) {
1104 return -1;
1105 }
1106 /* make sure pdp_temp is neither too large or too small
1107 * if any of these occur it becomes unknown ...
1108 * sorry folks ... */
1109 if (!isnan(rate) &&
1110 ((!isnan(rrd->ds_def[ds_idx].par[DS_max_val].u_val) &&
1111 rate > rrd->ds_def[ds_idx].par[DS_max_val].u_val) ||
1112 (!isnan(rrd->ds_def[ds_idx].par[DS_min_val].u_val) &&
1113 rate < rrd->ds_def[ds_idx].par[DS_min_val].u_val))) {
1114 pdp_new[ds_idx] = DNAN;
1115 }
1116 } else {
1117 /* no news is news all the same */
1118 pdp_new[ds_idx] = DNAN;
1119 }
1122 /* make a copy of the command line argument for the next run */
1123 #ifdef DEBUG
1124 fprintf(stderr, "prep ds[%lu]\t"
1125 "last_arg '%s'\t"
1126 "this_arg '%s'\t"
1127 "pdp_new %10.2f\n",
1128 ds_idx, rrd->pdp_prep[ds_idx].last_ds, updvals[ds_idx + 1],
1129 pdp_new[ds_idx]);
1130 #endif
1131 strncpy(rrd->pdp_prep[ds_idx].last_ds, updvals[ds_idx + 1],
1132 LAST_DS_LEN - 1);
1133 rrd->pdp_prep[ds_idx].last_ds[LAST_DS_LEN - 1] = '\0';
1134 }
1135 return 0;
1136 }
1138 /*
1139 * How many PDP steps have elapsed since the last update? Returns the answer,
1140 * and stores the time between the last update and the last PDP in pre_time,
1141 * and the time between the last PDP and the current time in post_int.
1142 */
1143 static int calculate_elapsed_steps(
1144 rrd_t *rrd,
1145 unsigned long current_time,
1146 unsigned long current_time_usec,
1147 double interval,
1148 double *pre_int,
1149 double *post_int,
1150 unsigned long *proc_pdp_cnt)
1151 {
1152 unsigned long proc_pdp_st; /* which pdp_st was the last to be processed */
1153 unsigned long occu_pdp_st; /* when was the pdp_st before the last update
1154 * time */
1155 unsigned long proc_pdp_age; /* how old was the data in the pdp prep area
1156 * when it was last updated */
1157 unsigned long occu_pdp_age; /* how long ago was the last pdp_step time */
1159 /* when was the current pdp started */
1160 proc_pdp_age = rrd->live_head->last_up % rrd->stat_head->pdp_step;
1161 proc_pdp_st = rrd->live_head->last_up - proc_pdp_age;
1163 /* when did the last pdp_st occur */
1164 occu_pdp_age = current_time % rrd->stat_head->pdp_step;
1165 occu_pdp_st = current_time - occu_pdp_age;
1167 if (occu_pdp_st > proc_pdp_st) {
1168 /* OK we passed the pdp_st moment */
1169 *pre_int = (long) occu_pdp_st - rrd->live_head->last_up; /* how much of the input data
1170 * occurred before the latest
1171 * pdp_st moment*/
1172 *pre_int -= ((double) rrd->live_head->last_up_usec) / 1e6f; /* adjust usecs */
1173 *post_int = occu_pdp_age; /* how much after it */
1174 *post_int += ((double) current_time_usec) / 1e6f; /* adjust usecs */
1175 } else {
1176 *pre_int = interval;
1177 *post_int = 0;
1178 }
1180 *proc_pdp_cnt = proc_pdp_st / rrd->stat_head->pdp_step;
1182 #ifdef DEBUG
1183 printf("proc_pdp_age %lu\t"
1184 "proc_pdp_st %lu\t"
1185 "occu_pfp_age %lu\t"
1186 "occu_pdp_st %lu\t"
1187 "int %lf\t"
1188 "pre_int %lf\t"
1189 "post_int %lf\n", proc_pdp_age, proc_pdp_st,
1190 occu_pdp_age, occu_pdp_st, interval, *pre_int, *post_int);
1191 #endif
1193 /* compute the number of elapsed pdp_st moments */
1194 return (occu_pdp_st - proc_pdp_st) / rrd->stat_head->pdp_step;
1195 }
1197 /*
1198 * Increment the PDP values by the values in pdp_new, or else initialize them.
1199 */
1200 static void simple_update(
1201 rrd_t *rrd,
1202 double interval,
1203 rrd_value_t *pdp_new)
1204 {
1205 int i;
1207 for (i = 0; i < (signed) rrd->stat_head->ds_cnt; i++) {
1208 if (isnan(pdp_new[i])) {
1209 /* this is not really accurate if we use subsecond data arrival time
1210 should have thought of it when going subsecond resolution ...
1211 sorry next format change we will have it! */
1212 rrd->pdp_prep[i].scratch[PDP_unkn_sec_cnt].u_cnt +=
1213 floor(interval);
1214 } else {
1215 if (isnan(rrd->pdp_prep[i].scratch[PDP_val].u_val)) {
1216 rrd->pdp_prep[i].scratch[PDP_val].u_val = pdp_new[i];
1217 } else {
1218 rrd->pdp_prep[i].scratch[PDP_val].u_val += pdp_new[i];
1219 }
1220 }
1221 #ifdef DEBUG
1222 fprintf(stderr,
1223 "NO PDP ds[%i]\t"
1224 "value %10.2f\t"
1225 "unkn_sec %5lu\n",
1226 i,
1227 rrd->pdp_prep[i].scratch[PDP_val].u_val,
1228 rrd->pdp_prep[i].scratch[PDP_unkn_sec_cnt].u_cnt);
1229 #endif
1230 }
1231 }
1233 /*
1234 * Call process_pdp_st for each DS.
1235 *
1236 * Returns 0 on success, -1 on error.
1237 */
1238 static int process_all_pdp_st(
1239 rrd_t *rrd,
1240 double interval,
1241 double pre_int,
1242 double post_int,
1243 unsigned long elapsed_pdp_st,
1244 rrd_value_t *pdp_new,
1245 rrd_value_t *pdp_temp)
1246 {
1247 unsigned long ds_idx;
1249 /* in pdp_prep[].scratch[PDP_val].u_val we have collected
1250 rate*seconds which occurred up to the last run.
1251 pdp_new[] contains rate*seconds from the latest run.
1252 pdp_temp[] will contain the rate for cdp */
1254 for (ds_idx = 0; ds_idx < rrd->stat_head->ds_cnt; ds_idx++) {
1255 if (process_pdp_st(rrd, ds_idx, interval, pre_int, post_int,
1256 elapsed_pdp_st * rrd->stat_head->pdp_step,
1257 pdp_new, pdp_temp) == -1) {
1258 return -1;
1259 }
1260 #ifdef DEBUG
1261 fprintf(stderr, "PDP UPD ds[%lu]\t"
1262 "elapsed_pdp_st %lu\t"
1263 "pdp_temp %10.2f\t"
1264 "new_prep %10.2f\t"
1265 "new_unkn_sec %5lu\n",
1266 ds_idx,
1267 elapsed_pdp_st,
1268 pdp_temp[ds_idx],
1269 rrd->pdp_prep[ds_idx].scratch[PDP_val].u_val,
1270 rrd->pdp_prep[ds_idx].scratch[PDP_unkn_sec_cnt].u_cnt);
1271 #endif
1272 }
1273 return 0;
1274 }
1276 /*
1277 * Process an update that occurs after one of the PDP moments.
1278 * Increments the PDP value, sets NAN if time greater than the
1279 * heartbeats have elapsed, processes CDEFs.
1280 *
1281 * Returns 0 on success, -1 on error.
1282 */
1283 static int process_pdp_st(
1284 rrd_t *rrd,
1285 unsigned long ds_idx,
1286 double interval,
1287 double pre_int,
1288 double post_int,
1289 long diff_pdp_st, /* number of seconds in full steps passed since last update */
1290 rrd_value_t *pdp_new,
1291 rrd_value_t *pdp_temp)
1292 {
1293 int i;
1295 /* update pdp_prep to the current pdp_st. */
1296 double pre_unknown = 0.0;
1297 unival *scratch = rrd->pdp_prep[ds_idx].scratch;
1298 unsigned long mrhb = rrd->ds_def[ds_idx].par[DS_mrhb_cnt].u_cnt;
1300 rpnstack_t rpnstack; /* used for COMPUTE DS */
1302 rpnstack_init(&rpnstack);
1305 if (isnan(pdp_new[ds_idx])) {
1306 /* a final bit of unknown to be added before calculation
1307 we use a temporary variable for this so that we
1308 don't have to turn integer lines before using the value */
1309 pre_unknown = pre_int;
1310 } else {
1311 if (isnan(scratch[PDP_val].u_val)) {
1312 scratch[PDP_val].u_val = 0;
1313 }
1314 scratch[PDP_val].u_val += pdp_new[ds_idx] / interval * pre_int;
1315 }
1317 /* if too much of the pdp_prep is unknown we dump it */
1318 /* if the interval is larger thatn mrhb we get NAN */
1319 if ((interval > mrhb) ||
1320 (rrd->stat_head->pdp_step / 2.0 <
1321 (signed) scratch[PDP_unkn_sec_cnt].u_cnt)) {
1322 pdp_temp[ds_idx] = DNAN;
1323 } else {
1324 pdp_temp[ds_idx] = scratch[PDP_val].u_val /
1325 ((double) (diff_pdp_st - scratch[PDP_unkn_sec_cnt].u_cnt) -
1326 pre_unknown);
1327 }
1329 /* process CDEF data sources; remember each CDEF DS can
1330 * only reference other DS with a lower index number */
1331 if (dst_conv(rrd->ds_def[ds_idx].dst) == DST_CDEF) {
1332 rpnp_t *rpnp;
1334 rpnp =
1335 rpn_expand((rpn_cdefds_t *) &(rrd->ds_def[ds_idx].par[DS_cdef]));
1336 if(rpnp == NULL) {
1337 rpnstack_free(&rpnstack);
1338 return -1;
1339 }
1340 /* substitute data values for OP_VARIABLE nodes */
1341 for (i = 0; rpnp[i].op != OP_END; i++) {
1342 if (rpnp[i].op == OP_VARIABLE) {
1343 rpnp[i].op = OP_NUMBER;
1344 rpnp[i].val = pdp_temp[rpnp[i].ptr];
1345 }
1346 }
1347 /* run the rpn calculator */
1348 if (rpn_calc(rpnp, &rpnstack, 0, pdp_temp, ds_idx) == -1) {
1349 free(rpnp);
1350 rpnstack_free(&rpnstack);
1351 return -1;
1352 }
1353 free(rpnp);
1354 }
1356 /* make pdp_prep ready for the next run */
1357 if (isnan(pdp_new[ds_idx])) {
1358 /* this is not realy accurate if we use subsecond data arival time
1359 should have thought of it when going subsecond resolution ...
1360 sorry next format change we will have it! */
1361 scratch[PDP_unkn_sec_cnt].u_cnt = floor(post_int);
1362 scratch[PDP_val].u_val = DNAN;
1363 } else {
1364 scratch[PDP_unkn_sec_cnt].u_cnt = 0;
1365 scratch[PDP_val].u_val = pdp_new[ds_idx] / interval * post_int;
1366 }
1367 rpnstack_free(&rpnstack);
1368 return 0;
1369 }
1371 /*
1372 * Iterate over all the RRAs for a given DS and:
1373 * 1. Decide whether to schedule a smooth later
1374 * 2. Decide whether to skip updating SEASONAL and DEVSEASONAL
1375 * 3. Update the CDP
1376 *
1377 * Returns 0 on success, -1 on error
1378 */
1379 static int update_all_cdp_prep(
1380 rrd_t *rrd,
1381 unsigned long *rra_step_cnt,
1382 unsigned long rra_begin,
1383 rrd_file_t *rrd_file,
1384 unsigned long elapsed_pdp_st,
1385 unsigned long proc_pdp_cnt,
1386 rrd_value_t **last_seasonal_coef,
1387 rrd_value_t **seasonal_coef,
1388 rrd_value_t *pdp_temp,
1389 unsigned long *skip_update,
1390 int *schedule_smooth)
1391 {
1392 unsigned long rra_idx;
1394 /* index into the CDP scratch array */
1395 enum cf_en current_cf;
1396 unsigned long rra_start;
1398 /* number of rows to be updated in an RRA for a data value. */
1399 unsigned long start_pdp_offset;
1401 rra_start = rra_begin;
1402 for (rra_idx = 0; rra_idx < rrd->stat_head->rra_cnt; rra_idx++) {
1403 current_cf = cf_conv(rrd->rra_def[rra_idx].cf_nam);
1404 start_pdp_offset =
1405 rrd->rra_def[rra_idx].pdp_cnt -
1406 proc_pdp_cnt % rrd->rra_def[rra_idx].pdp_cnt;
1407 skip_update[rra_idx] = 0;
1408 if (start_pdp_offset <= elapsed_pdp_st) {
1409 rra_step_cnt[rra_idx] = (elapsed_pdp_st - start_pdp_offset) /
1410 rrd->rra_def[rra_idx].pdp_cnt + 1;
1411 } else {
1412 rra_step_cnt[rra_idx] = 0;
1413 }
1415 if (current_cf == CF_SEASONAL || current_cf == CF_DEVSEASONAL) {
1416 /* If this is a bulk update, we need to skip ahead in the seasonal arrays
1417 * so that they will be correct for the next observed value; note that for
1418 * the bulk update itself, no update will occur to DEVSEASONAL or SEASONAL;
1419 * futhermore, HWPREDICT and DEVPREDICT will be set to DNAN. */
1420 if (rra_step_cnt[rra_idx] > 1) {
1421 skip_update[rra_idx] = 1;
1422 lookup_seasonal(rrd, rra_idx, rra_start, rrd_file,
1423 elapsed_pdp_st, last_seasonal_coef);
1424 lookup_seasonal(rrd, rra_idx, rra_start, rrd_file,
1425 elapsed_pdp_st + 1, seasonal_coef);
1426 }
1427 /* periodically run a smoother for seasonal effects */
1428 if (do_schedule_smooth(rrd, rra_idx, elapsed_pdp_st)) {
1429 #ifdef DEBUG
1430 fprintf(stderr,
1431 "schedule_smooth: cur_row %lu, elapsed_pdp_st %lu, smooth idx %lu\n",
1432 rrd->rra_ptr[rra_idx].cur_row, elapsed_pdp_st,
1433 rrd->rra_def[rra_idx].par[RRA_seasonal_smooth_idx].
1434 u_cnt);
1435 #endif
1436 *schedule_smooth = 1;
1437 }
1438 }
1439 if (rrd_test_error())
1440 return -1;
1442 if (update_cdp_prep
1443 (rrd, elapsed_pdp_st, start_pdp_offset, rra_step_cnt, rra_idx,
1444 pdp_temp, *last_seasonal_coef, *seasonal_coef,
1445 current_cf) == -1) {
1446 return -1;
1447 }
1448 rra_start +=
1449 rrd->rra_def[rra_idx].row_cnt * rrd->stat_head->ds_cnt *
1450 sizeof(rrd_value_t);
1451 }
1452 return 0;
1453 }
1455 /*
1456 * Are we due for a smooth? Also increments our position in the burn-in cycle.
1457 */
1458 static int do_schedule_smooth(
1459 rrd_t *rrd,
1460 unsigned long rra_idx,
1461 unsigned long elapsed_pdp_st)
1462 {
1463 unsigned long cdp_idx = rra_idx * (rrd->stat_head->ds_cnt);
1464 unsigned long cur_row = rrd->rra_ptr[rra_idx].cur_row;
1465 unsigned long row_cnt = rrd->rra_def[rra_idx].row_cnt;
1466 unsigned long seasonal_smooth_idx =
1467 rrd->rra_def[rra_idx].par[RRA_seasonal_smooth_idx].u_cnt;
1468 unsigned long *init_seasonal =
1469 &(rrd->cdp_prep[cdp_idx].scratch[CDP_init_seasonal].u_cnt);
1471 /* Need to use first cdp parameter buffer to track burnin (burnin requires
1472 * a specific smoothing schedule). The CDP_init_seasonal parameter is
1473 * really an RRA level, not a data source within RRA level parameter, but
1474 * the rra_def is read only for rrd_update (not flushed to disk). */
1475 if (*init_seasonal > BURNIN_CYCLES) {
1476 /* someone has no doubt invented a trick to deal with this wrap around,
1477 * but at least this code is clear. */
1478 if (seasonal_smooth_idx > cur_row) {
1479 /* here elapsed_pdp_st = rra_step_cnt[rra_idx] because of 1-1 mapping
1480 * between PDP and CDP */
1481 return (cur_row + elapsed_pdp_st >= seasonal_smooth_idx);
1482 }
1483 /* can't rely on negative numbers because we are working with
1484 * unsigned values */
1485 return (cur_row + elapsed_pdp_st >= row_cnt
1486 && cur_row + elapsed_pdp_st >= row_cnt + seasonal_smooth_idx);
1487 }
1488 /* mark off one of the burn-in cycles */
1489 return (cur_row + elapsed_pdp_st >= row_cnt && ++(*init_seasonal));
1490 }
1492 /*
1493 * For a given RRA, iterate over the data sources and call the appropriate
1494 * consolidation function.
1495 *
1496 * Returns 0 on success, -1 on error.
1497 */
1498 static int update_cdp_prep(
1499 rrd_t *rrd,
1500 unsigned long elapsed_pdp_st,
1501 unsigned long start_pdp_offset,
1502 unsigned long *rra_step_cnt,
1503 int rra_idx,
1504 rrd_value_t *pdp_temp,
1505 rrd_value_t *last_seasonal_coef,
1506 rrd_value_t *seasonal_coef,
1507 int current_cf)
1508 {
1509 unsigned long ds_idx, cdp_idx;
1511 /* update CDP_PREP areas */
1512 /* loop over data soures within each RRA */
1513 for (ds_idx = 0; ds_idx < rrd->stat_head->ds_cnt; ds_idx++) {
1515 cdp_idx = rra_idx * rrd->stat_head->ds_cnt + ds_idx;
1517 if (rrd->rra_def[rra_idx].pdp_cnt > 1) {
1518 update_cdp(rrd->cdp_prep[cdp_idx].scratch, current_cf,
1519 pdp_temp[ds_idx], rra_step_cnt[rra_idx],
1520 elapsed_pdp_st, start_pdp_offset,
1521 rrd->rra_def[rra_idx].pdp_cnt,
1522 rrd->rra_def[rra_idx].par[RRA_cdp_xff_val].u_val,
1523 rra_idx, ds_idx);
1524 } else {
1525 /* Nothing to consolidate if there's one PDP per CDP. However, if
1526 * we've missed some PDPs, let's update null counters etc. */
1527 if (elapsed_pdp_st > 2) {
1528 reset_cdp(rrd, elapsed_pdp_st, pdp_temp, last_seasonal_coef,
1529 seasonal_coef, rra_idx, ds_idx, cdp_idx,
1530 (enum cf_en)current_cf);
1531 }
1532 }
1534 if (rrd_test_error())
1535 return -1;
1536 } /* endif data sources loop */
1537 return 0;
1538 }
1540 /*
1541 * Given the new reading (pdp_temp_val), update or initialize the CDP value,
1542 * primary value, secondary value, and # of unknowns.
1543 */
1544 static void update_cdp(
1545 unival *scratch,
1546 int current_cf,
1547 rrd_value_t pdp_temp_val,
1548 unsigned long rra_step_cnt,
1549 unsigned long elapsed_pdp_st,
1550 unsigned long start_pdp_offset,
1551 unsigned long pdp_cnt,
1552 rrd_value_t xff,
1553 int i,
1554 int ii)
1555 {
1556 /* shorthand variables */
1557 rrd_value_t *cdp_val = &scratch[CDP_val].u_val;
1558 rrd_value_t *cdp_primary_val = &scratch[CDP_primary_val].u_val;
1559 rrd_value_t *cdp_secondary_val = &scratch[CDP_secondary_val].u_val;
1560 unsigned long *cdp_unkn_pdp_cnt = &scratch[CDP_unkn_pdp_cnt].u_cnt;
1562 if (rra_step_cnt) {
1563 /* If we are in this block, as least 1 CDP value will be written to
1564 * disk, this is the CDP_primary_val entry. If more than 1 value needs
1565 * to be written, then the "fill in" value is the CDP_secondary_val
1566 * entry. */
1567 if (isnan(pdp_temp_val)) {
1568 *cdp_unkn_pdp_cnt += start_pdp_offset;
1569 *cdp_secondary_val = DNAN;
1570 } else {
1571 /* CDP_secondary value is the RRA "fill in" value for intermediary
1572 * CDP data entries. No matter the CF, the value is the same because
1573 * the average, max, min, and last of a list of identical values is
1574 * the same, namely, the value itself. */
1575 *cdp_secondary_val = pdp_temp_val;
1576 }
1578 if (*cdp_unkn_pdp_cnt > pdp_cnt * xff) {
1579 *cdp_primary_val = DNAN;
1580 if (current_cf == CF_AVERAGE) {
1581 *cdp_val =
1582 initialize_average_carry_over(pdp_temp_val,
1583 elapsed_pdp_st,
1584 start_pdp_offset, pdp_cnt);
1585 } else {
1586 *cdp_val = pdp_temp_val;
1587 }
1588 } else {
1589 initialize_cdp_val(scratch, current_cf, pdp_temp_val,
1590 elapsed_pdp_st, start_pdp_offset, pdp_cnt);
1591 } /* endif meets xff value requirement for a valid value */
1592 /* initialize carry over CDP_unkn_pdp_cnt, this must after CDP_primary_val
1593 * is set because CDP_unkn_pdp_cnt is required to compute that value. */
1594 if (isnan(pdp_temp_val))
1595 *cdp_unkn_pdp_cnt = (elapsed_pdp_st - start_pdp_offset) % pdp_cnt;
1596 else
1597 *cdp_unkn_pdp_cnt = 0;
1598 } else { /* rra_step_cnt[i] == 0 */
1600 #ifdef DEBUG
1601 if (isnan(*cdp_val)) {
1602 fprintf(stderr, "schedule CDP_val update, RRA %d DS %d, DNAN\n",
1603 i, ii);
1604 } else {
1605 fprintf(stderr, "schedule CDP_val update, RRA %d DS %d, %10.2f\n",
1606 i, ii, *cdp_val);
1607 }
1608 #endif
1609 if (isnan(pdp_temp_val)) {
1610 *cdp_unkn_pdp_cnt += elapsed_pdp_st;
1611 } else {
1612 *cdp_val =
1613 calculate_cdp_val(*cdp_val, pdp_temp_val, elapsed_pdp_st,
1614 current_cf, i, ii);
1615 }
1616 }
1617 }
1619 /*
1620 * Set the CDP_primary_val and CDP_val to the appropriate initial value based
1621 * on the type of consolidation function.
1622 */
1623 static void initialize_cdp_val(
1624 unival *scratch,
1625 int current_cf,
1626 rrd_value_t pdp_temp_val,
1627 unsigned long elapsed_pdp_st,
1628 unsigned long start_pdp_offset,
1629 unsigned long pdp_cnt)
1630 {
1631 rrd_value_t cum_val, cur_val;
1633 switch (current_cf) {
1634 case CF_AVERAGE:
1635 cum_val = IFDNAN(scratch[CDP_val].u_val, 0.0);
1636 cur_val = IFDNAN(pdp_temp_val, 0.0);
1637 scratch[CDP_primary_val].u_val =
1638 (cum_val + cur_val * start_pdp_offset) /
1639 (pdp_cnt - scratch[CDP_unkn_pdp_cnt].u_cnt);
1640 scratch[CDP_val].u_val =
1641 initialize_average_carry_over(pdp_temp_val, elapsed_pdp_st,
1642 start_pdp_offset, pdp_cnt);
1643 break;
1644 case CF_MAXIMUM:
1645 cum_val = IFDNAN(scratch[CDP_val].u_val, -DINF);
1646 cur_val = IFDNAN(pdp_temp_val, -DINF);
1647 #if 0
1648 #ifdef DEBUG
1649 if (isnan(scratch[CDP_val].u_val) && isnan(pdp_temp)) {
1650 fprintf(stderr,
1651 "RRA %lu, DS %lu, both CDP_val and pdp_temp are DNAN!",
1652 i, ii);
1653 exit(-1);
1654 }
1655 #endif
1656 #endif
1657 if (cur_val > cum_val)
1658 scratch[CDP_primary_val].u_val = cur_val;
1659 else
1660 scratch[CDP_primary_val].u_val = cum_val;
1661 /* initialize carry over value */
1662 scratch[CDP_val].u_val = pdp_temp_val;
1663 break;
1664 case CF_MINIMUM:
1665 cum_val = IFDNAN(scratch[CDP_val].u_val, DINF);
1666 cur_val = IFDNAN(pdp_temp_val, DINF);
1667 #if 0
1668 #ifdef DEBUG
1669 if (isnan(scratch[CDP_val].u_val) && isnan(pdp_temp)) {
1670 fprintf(stderr,
1671 "RRA %lu, DS %lu, both CDP_val and pdp_temp are DNAN!", i,
1672 ii);
1673 exit(-1);
1674 }
1675 #endif
1676 #endif
1677 if (cur_val < cum_val)
1678 scratch[CDP_primary_val].u_val = cur_val;
1679 else
1680 scratch[CDP_primary_val].u_val = cum_val;
1681 /* initialize carry over value */
1682 scratch[CDP_val].u_val = pdp_temp_val;
1683 break;
1684 case CF_LAST:
1685 default:
1686 scratch[CDP_primary_val].u_val = pdp_temp_val;
1687 /* initialize carry over value */
1688 scratch[CDP_val].u_val = pdp_temp_val;
1689 break;
1690 }
1691 }
1693 /*
1694 * Update the consolidation function for Holt-Winters functions as
1695 * well as other functions that don't actually consolidate multiple
1696 * PDPs.
1697 */
1698 static void reset_cdp(
1699 rrd_t *rrd,
1700 unsigned long elapsed_pdp_st,
1701 rrd_value_t *pdp_temp,
1702 rrd_value_t *last_seasonal_coef,
1703 rrd_value_t *seasonal_coef,
1704 int rra_idx,
1705 int ds_idx,
1706 int cdp_idx,
1707 enum cf_en current_cf)
1708 {
1709 unival *scratch = rrd->cdp_prep[cdp_idx].scratch;
1711 switch (current_cf) {
1712 case CF_AVERAGE:
1713 default:
1714 scratch[CDP_primary_val].u_val = pdp_temp[ds_idx];
1715 scratch[CDP_secondary_val].u_val = pdp_temp[ds_idx];
1716 break;
1717 case CF_SEASONAL:
1718 case CF_DEVSEASONAL:
1719 /* need to update cached seasonal values, so they are consistent
1720 * with the bulk update */
1721 /* WARNING: code relies on the fact that CDP_hw_last_seasonal and
1722 * CDP_last_deviation are the same. */
1723 scratch[CDP_hw_last_seasonal].u_val = last_seasonal_coef[ds_idx];
1724 scratch[CDP_hw_seasonal].u_val = seasonal_coef[ds_idx];
1725 break;
1726 case CF_HWPREDICT:
1727 case CF_MHWPREDICT:
1728 /* need to update the null_count and last_null_count.
1729 * even do this for non-DNAN pdp_temp because the
1730 * algorithm is not learning from batch updates. */
1731 scratch[CDP_null_count].u_cnt += elapsed_pdp_st;
1732 scratch[CDP_last_null_count].u_cnt += elapsed_pdp_st - 1;
1733 /* fall through */
1734 case CF_DEVPREDICT:
1735 scratch[CDP_primary_val].u_val = DNAN;
1736 scratch[CDP_secondary_val].u_val = DNAN;
1737 break;
1738 case CF_FAILURES:
1739 /* do not count missed bulk values as failures */
1740 scratch[CDP_primary_val].u_val = 0;
1741 scratch[CDP_secondary_val].u_val = 0;
1742 /* need to reset violations buffer.
1743 * could do this more carefully, but for now, just
1744 * assume a bulk update wipes away all violations. */
1745 erase_violations(rrd, cdp_idx, rra_idx);
1746 break;
1747 }
1748 }
1750 static rrd_value_t initialize_average_carry_over(
1751 rrd_value_t pdp_temp_val,
1752 unsigned long elapsed_pdp_st,
1753 unsigned long start_pdp_offset,
1754 unsigned long pdp_cnt)
1755 {
1756 /* initialize carry over value */
1757 if (isnan(pdp_temp_val)) {
1758 return DNAN;
1759 }
1760 return pdp_temp_val * ((elapsed_pdp_st - start_pdp_offset) % pdp_cnt);
1761 }
1763 /*
1764 * Update or initialize a CDP value based on the consolidation
1765 * function.
1766 *
1767 * Returns the new value.
1768 */
1769 static rrd_value_t calculate_cdp_val(
1770 rrd_value_t cdp_val,
1771 rrd_value_t pdp_temp_val,
1772 unsigned long elapsed_pdp_st,
1773 int current_cf,
1774 #ifdef DEBUG
1775 int i,
1776 int ii
1777 #else
1778 int UNUSED(i),
1779 int UNUSED(ii)
1780 #endif
1781 )
1782 {
1783 if (isnan(cdp_val)) {
1784 if (current_cf == CF_AVERAGE) {
1785 pdp_temp_val *= elapsed_pdp_st;
1786 }
1787 #ifdef DEBUG
1788 fprintf(stderr, "Initialize CDP_val for RRA %d DS %d: %10.2f\n",
1789 i, ii, pdp_temp_val);
1790 #endif
1791 return pdp_temp_val;
1792 }
1793 if (current_cf == CF_AVERAGE)
1794 return cdp_val + pdp_temp_val * elapsed_pdp_st;
1795 if (current_cf == CF_MINIMUM)
1796 return (pdp_temp_val < cdp_val) ? pdp_temp_val : cdp_val;
1797 if (current_cf == CF_MAXIMUM)
1798 return (pdp_temp_val > cdp_val) ? pdp_temp_val : cdp_val;
1800 return pdp_temp_val;
1801 }
1803 /*
1804 * For each RRA, update the seasonal values and then call update_aberrant_CF
1805 * for each data source.
1806 *
1807 * Return 0 on success, -1 on error.
1808 */
1809 static int update_aberrant_cdps(
1810 rrd_t *rrd,
1811 rrd_file_t *rrd_file,
1812 unsigned long rra_begin,
1813 unsigned long elapsed_pdp_st,
1814 rrd_value_t *pdp_temp,
1815 rrd_value_t **seasonal_coef)
1816 {
1817 unsigned long rra_idx, ds_idx, j;
1819 /* number of PDP steps since the last update that
1820 * are assigned to the first CDP to be generated
1821 * since the last update. */
1822 unsigned short scratch_idx;
1823 unsigned long rra_start;
1824 enum cf_en current_cf;
1826 /* this loop is only entered if elapsed_pdp_st < 3 */
1827 for (j = elapsed_pdp_st, scratch_idx = CDP_primary_val;
1828 j > 0 && j < 3; j--, scratch_idx = CDP_secondary_val) {
1829 rra_start = rra_begin;
1830 for (rra_idx = 0; rra_idx < rrd->stat_head->rra_cnt; rra_idx++) {
1831 if (rrd->rra_def[rra_idx].pdp_cnt == 1) {
1832 current_cf = cf_conv(rrd->rra_def[rra_idx].cf_nam);
1833 if (current_cf == CF_SEASONAL || current_cf == CF_DEVSEASONAL) {
1834 if (scratch_idx == CDP_primary_val) {
1835 lookup_seasonal(rrd, rra_idx, rra_start, rrd_file,
1836 elapsed_pdp_st + 1, seasonal_coef);
1837 } else {
1838 lookup_seasonal(rrd, rra_idx, rra_start, rrd_file,
1839 elapsed_pdp_st + 2, seasonal_coef);
1840 }
1841 }
1842 if (rrd_test_error())
1843 return -1;
1844 /* loop over data soures within each RRA */
1845 for (ds_idx = 0; ds_idx < rrd->stat_head->ds_cnt; ds_idx++) {
1846 update_aberrant_CF(rrd, pdp_temp[ds_idx], current_cf,
1847 rra_idx * (rrd->stat_head->ds_cnt) +
1848 ds_idx, rra_idx, ds_idx, scratch_idx,
1849 *seasonal_coef);
1850 }
1851 }
1852 rra_start += rrd->rra_def[rra_idx].row_cnt
1853 * rrd->stat_head->ds_cnt * sizeof(rrd_value_t);
1854 }
1855 }
1856 return 0;
1857 }
1859 /*
1860 * Move sequentially through the file, writing one RRA at a time. Note this
1861 * architecture divorces the computation of CDP with flushing updated RRA
1862 * entries to disk.
1863 *
1864 * Return 0 on success, -1 on error.
1865 */
1866 static int write_to_rras(
1867 rrd_t *rrd,
1868 rrd_file_t *rrd_file,
1869 unsigned long *rra_step_cnt,
1870 unsigned long rra_begin,
1871 time_t current_time,
1872 unsigned long *skip_update,
1873 rrd_info_t ** pcdp_summary)
1874 {
1875 unsigned long rra_idx;
1876 unsigned long rra_start;
1877 time_t rra_time = 0; /* time of update for a RRA */
1879 unsigned long ds_cnt = rrd->stat_head->ds_cnt;
1881 /* Ready to write to disk */
1882 rra_start = rra_begin;
1884 for (rra_idx = 0; rra_idx < rrd->stat_head->rra_cnt; rra_idx++) {
1885 rra_def_t *rra_def = &rrd->rra_def[rra_idx];
1886 rra_ptr_t *rra_ptr = &rrd->rra_ptr[rra_idx];
1888 /* for cdp_prep */
1889 unsigned short scratch_idx;
1890 unsigned long step_subtract;
1892 for (scratch_idx = CDP_primary_val,
1893 step_subtract = 1;
1894 rra_step_cnt[rra_idx] > 0;
1895 rra_step_cnt[rra_idx]--,
1896 scratch_idx = CDP_secondary_val,
1897 step_subtract = 2) {
1899 size_t rra_pos_new;
1900 #ifdef DEBUG
1901 fprintf(stderr, " -- RRA Preseek %ld\n", rrd_file->pos);
1902 #endif
1903 /* increment, with wrap-around */
1904 if (++rra_ptr->cur_row >= rra_def->row_cnt)
1905 rra_ptr->cur_row = 0;
1907 /* we know what our position should be */
1908 rra_pos_new = rra_start
1909 + ds_cnt * rra_ptr->cur_row * sizeof(rrd_value_t);
1911 /* re-seek if the position is wrong or we wrapped around */
1912 if (rra_pos_new != rrd_file->pos) {
1913 if (rrd_seek(rrd_file, rra_pos_new, SEEK_SET) != 0) {
1914 rrd_set_error("seek error in rrd");
1915 return -1;
1916 }
1917 }
1918 #ifdef DEBUG
1919 fprintf(stderr, " -- RRA Postseek %ld\n", rrd_file->pos);
1920 #endif
1922 if (skip_update[rra_idx])
1923 continue;
1925 if (*pcdp_summary != NULL) {
1926 unsigned long step_time = rra_def->pdp_cnt * rrd->stat_head->pdp_step;
1928 rra_time = (current_time - current_time % step_time)
1929 - ((rra_step_cnt[rra_idx] - step_subtract) * step_time);
1930 }
1932 if (write_RRA_row
1933 (rrd_file, rrd, rra_idx, scratch_idx,
1934 pcdp_summary, rra_time) == -1)
1935 return -1;
1936 }
1938 rra_start += rra_def->row_cnt * ds_cnt * sizeof(rrd_value_t);
1939 } /* RRA LOOP */
1941 return 0;
1942 }
1944 /*
1945 * Write out one row of values (one value per DS) to the archive.
1946 *
1947 * Returns 0 on success, -1 on error.
1948 */
1949 static int write_RRA_row(
1950 rrd_file_t *rrd_file,
1951 rrd_t *rrd,
1952 unsigned long rra_idx,
1953 unsigned short CDP_scratch_idx,
1954 rrd_info_t ** pcdp_summary,
1955 time_t rra_time)
1956 {
1957 unsigned long ds_idx, cdp_idx;
1958 rrd_infoval_t iv;
1960 for (ds_idx = 0; ds_idx < rrd->stat_head->ds_cnt; ds_idx++) {
1961 /* compute the cdp index */
1962 cdp_idx = rra_idx * (rrd->stat_head->ds_cnt) + ds_idx;
1963 #ifdef DEBUG
1964 fprintf(stderr, " -- RRA WRITE VALUE %e, at %ld CF:%s\n",
1965 rrd->cdp_prep[cdp_idx].scratch[CDP_scratch_idx].u_val,
1966 rrd_file->pos, rrd->rra_def[rra_idx].cf_nam);
1967 #endif
1968 if (*pcdp_summary != NULL) {
1969 iv.u_val = rrd->cdp_prep[cdp_idx].scratch[CDP_scratch_idx].u_val;
1970 /* append info to the return hash */
1971 *pcdp_summary = rrd_info_push(*pcdp_summary,
1972 sprintf_alloc
1973 ("[%lli]RRA[%s][%lu]DS[%s]", (long long)rra_time,
1974 rrd->rra_def[rra_idx].cf_nam,
1975 rrd->rra_def[rra_idx].pdp_cnt,
1976 rrd->ds_def[ds_idx].ds_nam),
1977 RD_I_VAL, iv);
1978 }
1979 errno = 0;
1980 if (rrd_write(rrd_file,
1981 &(rrd->cdp_prep[cdp_idx].scratch[CDP_scratch_idx].
1982 u_val), sizeof(rrd_value_t)) != sizeof(rrd_value_t)) {
1983 rrd_set_error("writing rrd: %s", rrd_strerror(errno));
1984 return -1;
1985 }
1986 }
1987 return 0;
1988 }
1990 /*
1991 * Call apply_smoother for all DEVSEASONAL and SEASONAL RRAs.
1992 *
1993 * Returns 0 on success, -1 otherwise
1994 */
1995 static int smooth_all_rras(
1996 rrd_t *rrd,
1997 rrd_file_t *rrd_file,
1998 unsigned long rra_begin)
1999 {
2000 unsigned long rra_start = rra_begin;
2001 unsigned long rra_idx;
2003 for (rra_idx = 0; rra_idx < rrd->stat_head->rra_cnt; ++rra_idx) {
2004 if (cf_conv(rrd->rra_def[rra_idx].cf_nam) == CF_DEVSEASONAL ||
2005 cf_conv(rrd->rra_def[rra_idx].cf_nam) == CF_SEASONAL) {
2006 #ifdef DEBUG
2007 fprintf(stderr, "Running smoother for rra %lu\n", rra_idx);
2008 #endif
2009 apply_smoother(rrd, rra_idx, rra_start, rrd_file);
2010 if (rrd_test_error())
2011 return -1;
2012 }
2013 rra_start += rrd->rra_def[rra_idx].row_cnt
2014 * rrd->stat_head->ds_cnt * sizeof(rrd_value_t);
2015 }
2016 return 0;
2017 }
2019 #ifndef HAVE_MMAP
2020 /*
2021 * Flush changes to disk (unless we're using mmap)
2022 *
2023 * Returns 0 on success, -1 otherwise
2024 */
2025 static int write_changes_to_disk(
2026 rrd_t *rrd,
2027 rrd_file_t *rrd_file,
2028 int version)
2029 {
2030 /* we just need to write back the live header portion now */
2031 if (rrd_seek(rrd_file, (sizeof(stat_head_t)
2032 + sizeof(ds_def_t) * rrd->stat_head->ds_cnt
2033 + sizeof(rra_def_t) * rrd->stat_head->rra_cnt),
2034 SEEK_SET) != 0) {
2035 rrd_set_error("seek rrd for live header writeback");
2036 return -1;
2037 }
2038 if (version >= 3) {
2039 if (rrd_write(rrd_file, rrd->live_head,
2040 sizeof(live_head_t) * 1) != sizeof(live_head_t) * 1) {
2041 rrd_set_error("rrd_write live_head to rrd");
2042 return -1;
2043 }
2044 } else {
2045 if (rrd_write(rrd_file, rrd->legacy_last_up,
2046 sizeof(time_t) * 1) != sizeof(time_t) * 1) {
2047 rrd_set_error("rrd_write live_head to rrd");
2048 return -1;
2049 }
2050 }
2053 if (rrd_write(rrd_file, rrd->pdp_prep,
2054 sizeof(pdp_prep_t) * rrd->stat_head->ds_cnt)
2055 != (ssize_t) (sizeof(pdp_prep_t) * rrd->stat_head->ds_cnt)) {
2056 rrd_set_error("rrd_write pdp_prep to rrd");
2057 return -1;
2058 }
2060 if (rrd_write(rrd_file, rrd->cdp_prep,
2061 sizeof(cdp_prep_t) * rrd->stat_head->rra_cnt *
2062 rrd->stat_head->ds_cnt)
2063 != (ssize_t) (sizeof(cdp_prep_t) * rrd->stat_head->rra_cnt *
2064 rrd->stat_head->ds_cnt)) {
2066 rrd_set_error("rrd_write cdp_prep to rrd");
2067 return -1;
2068 }
2070 if (rrd_write(rrd_file, rrd->rra_ptr,
2071 sizeof(rra_ptr_t) * rrd->stat_head->rra_cnt)
2072 != (ssize_t) (sizeof(rra_ptr_t) * rrd->stat_head->rra_cnt)) {
2073 rrd_set_error("rrd_write rra_ptr to rrd");
2074 return -1;
2075 }
2076 return 0;
2077 }
2078 #endif