fix for multi update - data corruption bug as reported (and patched) in #178 by kevin...
[rrdtool.git] / src / rrd_update.c
index f79b3cb..8fe1395 100644 (file)
@@ -1,6 +1,6 @@
 
 /*****************************************************************************
- * RRDtool 1.3rc6  Copyright by Tobi Oetiker, 1997-2008
+ * RRDtool 1.3.1  Copyright by Tobi Oetiker, 1997-2008
  *****************************************************************************
  * rrd_update.c  RRD Update Function
  *****************************************************************************
@@ -71,7 +71,7 @@ int       _rrd_update(
     const char *tmplt,
     int argc,
     const char **argv,
-    info_t *);
+    rrd_info_t *);
 
 static int allocate_data_structures(
     rrd_t *rrd,
@@ -104,7 +104,7 @@ static int process_arg(
     char **updvals,
     long *tmpl_idx,
     unsigned long tmpl_cnt,
-    info_t **pcdp_summary,
+    rrd_info_t ** pcdp_summary,
     int version,
     unsigned long *skip_update,
     int *schedule_smooth);
@@ -258,7 +258,7 @@ static int write_to_rras(
     unsigned long *rra_current,
     time_t current_time,
     unsigned long *skip_update,
-    info_t **pcdp_summary);
+    rrd_info_t ** pcdp_summary);
 
 static int write_RRA_row(
     rrd_file_t *rrd_file,
@@ -266,7 +266,7 @@ static int write_RRA_row(
     unsigned long rra_idx,
     unsigned long *rra_current,
     unsigned short CDP_scratch_idx,
-    info_t **pcdp_summary,
+    rrd_info_t ** pcdp_summary,
     time_t rra_time);
 
 static int smooth_all_rras(
@@ -317,13 +317,13 @@ static inline void initialize_time(
 
 #define IFDNAN(X,Y) (isnan(X) ? (Y) : (X));
 
-info_t   *rrd_update_v(
+rrd_info_t *rrd_update_v(
     int argc,
     char **argv)
 {
     char     *tmplt = NULL;
-    info_t   *result = NULL;
-    infoval   rc;
+    rrd_info_t *result = NULL;
+    rrd_infoval_t rc;
     struct option long_options[] = {
         {"template", required_argument, 0, 't'},
         {0, 0, 0, 0}
@@ -359,7 +359,7 @@ info_t   *rrd_update_v(
         goto end_tag;
     }
     rc.u_int = 0;
-    result = info_push(NULL, sprintf_alloc("return_value"), RD_I_INT, rc);
+    result = rrd_info_push(NULL, sprintf_alloc("return_value"), RD_I_INT, rc);
     rc.u_int = _rrd_update(argv[optind], tmplt,
                            argc - optind - 1,
                            (const char **) (argv + optind + 1), result);
@@ -428,7 +428,7 @@ int _rrd_update(
     const char *tmplt,
     int argc,
     const char **argv,
-    info_t *pcdp_summary)
+    rrd_info_t * pcdp_summary)
 {
 
     int       arg_i = 2;
@@ -479,7 +479,7 @@ int _rrd_update(
     /* get exclusive lock to whole file.
      * lock gets removed when we close the file.
      */
-    if (LockRRD(rrd_file->fd) != 0) {
+    if (rrd_lock(rrd_file) != 0) {
         rrd_set_error("could not lock RRD");
         goto err_close;
     }
@@ -502,6 +502,15 @@ int _rrd_update(
                         rra_step_cnt, updvals, tmpl_idx, tmpl_cnt,
                         &pcdp_summary, version, skip_update,
                         &schedule_smooth) == -1) {
+            if (rrd_test_error()) { /* Should have error string always here */
+                char     *save_error;
+
+                /* Prepend file name to error message */
+                if ((save_error = strdup(rrd_get_error())) != NULL) {
+                    rrd_set_error("%s: %s", filename, save_error);
+                    free(save_error);
+                }
+            }
             free(arg_copy);
             break;
         }
@@ -560,8 +569,8 @@ int _rrd_update(
  *
  * returns 0 on success
  */
-int LockRRD(
-    int in_file)
+int rrd_lock(
+    rrd_file_t *file)
 {
     int       rcstat;
 
@@ -569,8 +578,8 @@ int LockRRD(
 #if defined(_WIN32) && !defined(__CYGWIN__) && !defined(__CYGWIN32__)
         struct _stat st;
 
-        if (_fstat(in_file, &st) == 0) {
-            rcstat = _locking(in_file, _LK_NBLCK, st.st_size);
+        if (_fstat(file->fd, &st) == 0) {
+            rcstat = _locking(file->fd, _LK_NBLCK, st.st_size);
         } else {
             rcstat = -1;
         }
@@ -582,7 +591,7 @@ int LockRRD(
         lock.l_start = 0;   /* start of file */
         lock.l_whence = SEEK_SET;   /* end of file */
 
-        rcstat = fcntl(in_file, F_SETLK, &lock);
+        rcstat = fcntl(file->fd, F_SETLK, &lock);
 #endif
     }
 
@@ -750,7 +759,7 @@ static int process_arg(
     char **updvals,
     long *tmpl_idx,
     unsigned long tmpl_cnt,
-    info_t **pcdp_summary,
+    rrd_info_t ** pcdp_summary,
     int version,
     unsigned long *skip_update,
     int *schedule_smooth)
@@ -771,12 +780,10 @@ static int process_arg(
     }
     /* seek to the beginning of the rra's */
     if (*rra_current != rra_begin) {
-#ifndef HAVE_MMAP
         if (rrd_seek(rrd_file, rra_begin, SEEK_SET) != 0) {
             rrd_set_error("seek error in rrd");
             return -1;
         }
-#endif
         *rra_current = rra_begin;
     }
     rra_start = rra_begin;
@@ -832,7 +839,7 @@ static int process_arg(
     rrd->live_head->last_up = *current_time;
     rrd->live_head->last_up_usec = *current_time_usec;
 
-    if ( version < 3 ){
+    if (version < 3) {
         *rrd->legacy_last_up = rrd->live_head->last_up;
     }
     free(seasonal_coef);
@@ -926,12 +933,12 @@ static int get_time_from_reading(
     double    tmp;
     char     *parsetime_error = NULL;
     char     *old_locale;
-    struct rrd_time_value ds_tv;
+    rrd_time_value_t ds_tv;
     struct timeval tmp_time;    /* used for time conversion */
 
     /* get the time from the reading ... handle N */
     if (timesyntax == '@') {    /* at-style */
-        if ((parsetime_error = parsetime(updvals[0], &ds_tv))) {
+        if ((parsetime_error = rrd_parsetime(updvals[0], &ds_tv))) {
             rrd_set_error("ds time: %s: %s", updvals[0], parsetime_error);
             return -1;
         }
@@ -1855,99 +1862,74 @@ static int write_to_rras(
     unsigned long *rra_current,
     time_t current_time,
     unsigned long *skip_update,
-    info_t **pcdp_summary)
+    rrd_info_t ** pcdp_summary)
 {
     unsigned long rra_idx;
     unsigned long rra_start;
-    unsigned long rra_pos_tmp;  /* temporary byte pointer. */
     time_t    rra_time = 0; /* time of update for a RRA */
 
+    unsigned long ds_cnt = rrd->stat_head->ds_cnt;
+    
     /* Ready to write to disk */
     rra_start = rra_begin;
+
     for (rra_idx = 0; rra_idx < rrd->stat_head->rra_cnt; rra_idx++) {
-        /* skip unless there's something to write */
-        if (rra_step_cnt[rra_idx]) {
-            /* write the first row */
+        rra_def_t *rra_def = &rrd->rra_def[rra_idx];
+        rra_ptr_t *rra_ptr = &rrd->rra_ptr[rra_idx];
+
+        /* for cdp_prep */
+        unsigned short scratch_idx;
+        unsigned long step_subtract;
+
+        for (scratch_idx = CDP_primary_val,
+                 step_subtract = 1;
+             rra_step_cnt[rra_idx] > 0;
+             rra_step_cnt[rra_idx]--,
+                 scratch_idx = CDP_secondary_val,
+                 step_subtract = 2) {
+
+            unsigned long rra_pos_new;
 #ifdef DEBUG
             fprintf(stderr, "  -- RRA Preseek %ld\n", rrd_file->pos);
 #endif
-            rrd->rra_ptr[rra_idx].cur_row++;
-            if (rrd->rra_ptr[rra_idx].cur_row >=
-                rrd->rra_def[rra_idx].row_cnt)
-                rrd->rra_ptr[rra_idx].cur_row = 0;  /* wrap around */
-            /* position on the first row */
-            rra_pos_tmp = rra_start +
-                (rrd->stat_head->ds_cnt) * (rrd->rra_ptr[rra_idx].cur_row) *
-                sizeof(rrd_value_t);
-            if (rra_pos_tmp != *rra_current) {
-                if (rrd_seek(rrd_file, rra_pos_tmp, SEEK_SET) != 0) {
+            /* increment, with wrap-around */
+            if (++rra_ptr->cur_row >= rra_def->row_cnt)
+              rra_ptr->cur_row = 0;
+
+            /* we know what our position should be */
+            rra_pos_new = rra_start
+              + ds_cnt * rra_ptr->cur_row * sizeof(rrd_value_t);
+
+            /* re-seek if the position is wrong or we wrapped around */
+            if (rra_pos_new != *rra_current || rra_ptr->cur_row == 0) {
+                if (rrd_seek(rrd_file, rra_pos_new, SEEK_SET) != 0) {
                     rrd_set_error("seek error in rrd");
                     return -1;
                 }
-                *rra_current = rra_pos_tmp;
+                *rra_current = rra_pos_new;
             }
 #ifdef DEBUG
             fprintf(stderr, "  -- RRA Postseek %ld\n", rrd_file->pos);
 #endif
-            if (!skip_update[rra_idx]) {
-                if (*pcdp_summary != NULL) {
-                    rra_time = (current_time - current_time
-                                % (rrd->rra_def[rra_idx].pdp_cnt *
-                                   rrd->stat_head->pdp_step))
-                        -
-                        ((rra_step_cnt[rra_idx] -
-                          1) * rrd->rra_def[rra_idx].pdp_cnt *
-                         rrd->stat_head->pdp_step);
-                }
-                if (write_RRA_row
-                    (rrd_file, rrd, rra_idx, rra_current, CDP_primary_val,
-                     pcdp_summary, rra_time) == -1)
-                    return -1;
-            }
 
-            /* write other rows of the bulk update, if any */
-            for (; rra_step_cnt[rra_idx] > 1; rra_step_cnt[rra_idx]--) {
-                if (++rrd->rra_ptr[rra_idx].cur_row ==
-                    rrd->rra_def[rra_idx].row_cnt) {
-#ifdef DEBUG
-                    fprintf(stderr,
-                            "Wraparound for RRA %s, %lu updates left\n",
-                            rrd->rra_def[rra_idx].cf_nam,
-                            rra_step_cnt[rra_idx] - 1);
-#endif
-                    /* wrap */
-                    rrd->rra_ptr[rra_idx].cur_row = 0;
-                    /* seek back to beginning of current rra */
-                    if (rrd_seek(rrd_file, rra_start, SEEK_SET) != 0) {
-                        rrd_set_error("seek error in rrd");
-                        return -1;
-                    }
-#ifdef DEBUG
-                    fprintf(stderr, "  -- Wraparound Postseek %ld\n",
-                            rrd_file->pos);
-#endif
-                    *rra_current = rra_start;
-                }
-                if (!skip_update[rra_idx]) {
-                    if (*pcdp_summary != NULL) {
-                        rra_time = (current_time - current_time
-                                    % (rrd->rra_def[rra_idx].pdp_cnt *
-                                       rrd->stat_head->pdp_step))
-                            -
-                            ((rra_step_cnt[rra_idx] -
-                              2) * rrd->rra_def[rra_idx].pdp_cnt *
-                             rrd->stat_head->pdp_step);
-                    }
-                    if (write_RRA_row(rrd_file, rrd, rra_idx, rra_current,
-                                      CDP_secondary_val, pcdp_summary,
-                                      rra_time) == -1)
-                        return -1;
-                }
+            if (skip_update[rra_idx])
+                continue;
+
+            if (*pcdp_summary != NULL) {
+                unsigned long step_time = rra_def->pdp_cnt * rrd->stat_head->pdp_step;
+
+                rra_time = (current_time - current_time % step_time)
+                    - ((rra_step_cnt[rra_idx] - step_subtract) * step_time);
             }
+
+            if (write_RRA_row
+                (rrd_file, rrd, rra_idx, rra_current, scratch_idx,
+                 pcdp_summary, rra_time) == -1)
+                return -1;
         }
-        rra_start += rrd->rra_def[rra_idx].row_cnt * rrd->stat_head->ds_cnt *
-            sizeof(rrd_value_t);
-    }                   /* RRA LOOP */
+
+        rra_start += rra_def->row_cnt * ds_cnt * sizeof(rrd_value_t);
+    } /* RRA LOOP */
 
     return 0;
 }
@@ -1963,11 +1945,11 @@ static int write_RRA_row(
     unsigned long rra_idx,
     unsigned long *rra_current,
     unsigned short CDP_scratch_idx,
-    info_t **pcdp_summary,
+    rrd_info_t ** pcdp_summary,
     time_t rra_time)
 {
     unsigned long ds_idx, cdp_idx;
-    infoval   iv;
+    rrd_infoval_t iv;
 
     for (ds_idx = 0; ds_idx < rrd->stat_head->ds_cnt; ds_idx++) {
         /* compute the cdp index */
@@ -1980,15 +1962,13 @@ static int write_RRA_row(
         if (*pcdp_summary != NULL) {
             iv.u_val = rrd->cdp_prep[cdp_idx].scratch[CDP_scratch_idx].u_val;
             /* append info to the return hash */
-            *pcdp_summary = info_push(*pcdp_summary,
-                                      sprintf_alloc("[%d]RRA[%s][%lu]DS[%s]",
-                                                    rra_time,
-                                                    rrd->rra_def[rra_idx].
-                                                    cf_nam,
-                                                    rrd->rra_def[rra_idx].
-                                                    pdp_cnt,
-                                                    rrd->ds_def[ds_idx].
-                                                    ds_nam), RD_I_VAL, iv);
+            *pcdp_summary = rrd_info_push(*pcdp_summary,
+                                          sprintf_alloc
+                                          ("[%d]RRA[%s][%lu]DS[%s]", rra_time,
+                                           rrd->rra_def[rra_idx].cf_nam,
+                                           rrd->rra_def[rra_idx].pdp_cnt,
+                                           rrd->ds_def[ds_idx].ds_nam),
+                                          RD_I_VAL, iv);
         }
         if (rrd_write(rrd_file,
                       &(rrd->cdp_prep[cdp_idx].scratch[CDP_scratch_idx].