Increment the DataSetsWritten counter before freeing the RRD values. --kevin

[rrdtool.git] / src / rrd_daemon.c

diff --git a/src/rrd_daemon.c b/src/rrd_daemon.c
index d8f140e..0e95e90 100644 (file)
--- a/src/rrd_daemon.c
+++ b/src/rrd_daemon.c
@@ -215,7 +215,11 @@ static uid_t daemon_uid;
 static listen_socket_t *listen_fds = NULL;
 static size_t listen_fds_num = 0;
 
-static int do_shutdown = 0;
+enum {
+  RUNNING,             /* normal operation */
+  FLUSHING,            /* flushing remaining values */
+  SHUTDOWN             /* shutting down */
+} state = RUNNING;
 
 static pthread_t *queue_threads;
 static pthread_cond_t queue_cond = PTHREAD_COND_INITIALIZER;
@@ -273,7 +277,7 @@ static int handle_request_help (HANDLER_PROTO);
 static void sig_common (const char *sig) /* {{{ */
 {
   RRDD_LOG(LOG_NOTICE, "caught SIG%s", sig);
-  do_shutdown++;
+  state = FLUSHING;
   pthread_cond_broadcast(&flush_cond);
   pthread_cond_broadcast(&queue_cond);
 } /* }}} void sig_common */
@@ -739,13 +743,8 @@ static gboolean tree_callback_flush (gpointer key, gpointer value, /* {{{ */
   if (ci->flags & CI_FLAGS_IN_QUEUE)
     return FALSE;
 
-  if ((ci->last_flush_time <= cfd->abs_timeout)
-      && (ci->values_num > 0))
-  {
-    enqueue_cache_item (ci, TAIL);
-  }
-  else if ((do_shutdown != 0)
-      && (ci->values_num > 0))
+  if (ci->values_num > 0
+      && (ci->last_flush_time <= cfd->abs_timeout || state != RUNNING))
   {
     enqueue_cache_item (ci, TAIL);
   }
@@ -814,21 +813,22 @@ static void *flush_thread_main (void *args __attribute__((unused))) /* {{{ */
 
   pthread_mutex_lock(&cache_lock);
 
-  while (!do_shutdown)
+  while (state == RUNNING)
   {
     gettimeofday (&now, NULL);
     if ((now.tv_sec > next_flush.tv_sec)
         || ((now.tv_sec == next_flush.tv_sec)
           && ((1000 * now.tv_usec) > next_flush.tv_nsec)))
     {
+      RRDD_LOG(LOG_DEBUG, "flushing old values");
+
+      /* Determine the time of the next cache flush. */
+      next_flush.tv_sec = now.tv_sec + config_flush_interval;
+
       /* Flush all values that haven't been written in the last
        * `config_write_interval' seconds. */
       flush_old_values (config_write_interval);
 
-      /* Determine the time of the next cache flush. */
-      next_flush.tv_sec =
-        now.tv_sec + next_flush.tv_sec % config_flush_interval;
-
       /* unlock the cache while we rotate so we don't block incoming
        * updates if the fsync() blocks on disk I/O */
       pthread_mutex_unlock(&cache_lock);
@@ -847,6 +847,8 @@ static void *flush_thread_main (void *args __attribute__((unused))) /* {{{ */
   if (config_flush_at_shutdown)
     flush_old_values (-1); /* flush everything */
 
+  state = SHUTDOWN;
+
   pthread_mutex_unlock(&cache_lock);
 
   return NULL;
@@ -856,7 +858,7 @@ static void *queue_thread_main (void *args __attribute__((unused))) /* {{{ */
 {
   pthread_mutex_lock (&cache_lock);
 
-  while (!do_shutdown
+  while (state != SHUTDOWN
          || (cache_queue_head != NULL && config_flush_at_shutdown))
   {
     cache_item_t *ci;
@@ -866,8 +868,8 @@ static void *queue_thread_main (void *args __attribute__((unused))) /* {{{ */
     int status;
 
     /* Now, check if there's something to store away. If not, wait until
-     * something comes in.  if we are shutting down, do not wait around.  */
-    if (cache_queue_head == NULL && !do_shutdown)
+     * something comes in. */
+    if (cache_queue_head == NULL)
     {
       status = pthread_cond_wait (&queue_cond, &cache_lock);
       if ((status != 0) && (status != ETIMEDOUT))
@@ -913,10 +915,14 @@ static void *queue_thread_main (void *args __attribute__((unused))) /* {{{ */
     }
 
     journal_write("wrote", file);
-    pthread_cond_broadcast(&ci->flushed);
 
-    rrd_free_ptrs((void ***) &values, &values_num);
-    free(file);
+    /* Search again in the tree.  It's possible someone issued a "FORGET"
+     * while we were writing the update values. */
+    pthread_mutex_lock(&cache_lock);
+    ci = (cache_item_t *) g_tree_lookup(cache_tree, file);
+    if (ci)
+      pthread_cond_broadcast(&ci->flushed);
+    pthread_mutex_unlock(&cache_lock);
 
     if (status == 0)
     {
@@ -926,6 +932,9 @@ static void *queue_thread_main (void *args __attribute__((unused))) /* {{{ */
       pthread_mutex_unlock (&stats_lock);
     }
 
+    rrd_free_ptrs((void ***) &values, &values_num);
+    free(file);
+
     pthread_mutex_lock (&cache_lock);
   }
   pthread_mutex_unlock (&cache_lock);
@@ -1307,6 +1316,7 @@ static int handle_request_update (HANDLER_PROTO) /* {{{ */
   if (ci == NULL) /* {{{ */
   {
     struct stat statbuf;
+    cache_item_t *tmp;
 
     /* don't hold the lock while we setup; stat(2) might block */
     pthread_mutex_unlock(&cache_lock);
@@ -1354,7 +1364,20 @@ static int handle_request_update (HANDLER_PROTO) /* {{{ */
     pthread_cond_init(&ci->flushed, NULL);
 
     pthread_mutex_lock(&cache_lock);
-    g_tree_replace (cache_tree, (void *) ci->file, (void *) ci);
+
+    /* another UPDATE might have added this entry in the meantime */
+    tmp = g_tree_lookup (cache_tree, file);
+    if (tmp == NULL)
+      g_tree_replace (cache_tree, (void *) ci->file, (void *) ci);
+    else
+    {
+      free_cache_item (ci);
+      ci = tmp;
+    }
+
+    /* state may have changed while we were unlocked */
+    if (state == SHUTDOWN)
+      return -1;
   } /* }}} */
   assert (ci != NULL);
 
@@ -1986,7 +2009,7 @@ static void *connection_thread_main (void *args) /* {{{ */
   connection_threads_num++;
   pthread_mutex_unlock (&connection_threads_lock);
 
-  while (do_shutdown == 0)
+  while (state == RUNNING)
   {
     char *cmd;
     ssize_t cmd_len;
@@ -2001,7 +2024,7 @@ static void *connection_thread_main (void *args) /* {{{ */
     pollfd.revents = 0;
 
     status = poll (&pollfd, 1, /* timeout = */ 500);
-    if (do_shutdown)
+    if (state != RUNNING)
       break;
     else if (status == 0) /* timeout */
       continue;
@@ -2306,7 +2329,7 @@ static void *listen_thread_main (void *args __attribute__((unused))) /* {{{ */
 
   RRDD_LOG(LOG_INFO, "listening for connections");
 
-  while (do_shutdown == 0)
+  while (state == RUNNING)
   {
     for (i = 0; i < pollfds_num; i++)
     {
@@ -2316,7 +2339,7 @@ static void *listen_thread_main (void *args __attribute__((unused))) /* {{{ */
     }
 
     status = poll (pollfds, pollfds_num, /* timeout = */ 1000);
-    if (do_shutdown)
+    if (state != RUNNING)
       break;
     else if (status == 0) /* timeout */
       continue;
@@ -2379,7 +2402,7 @@ static void *listen_thread_main (void *args __attribute__((unused))) /* {{{ */
         continue;
       }
     } /* for (pollfds_num) */
-  } /* while (do_shutdown == 0) */
+  } /* while (state == RUNNING) */
 
   RRDD_LOG(LOG_INFO, "starting shutdown");
 
@@ -2491,8 +2514,6 @@ error:
 
 static int cleanup (void) /* {{{ */
 {
-  do_shutdown++;
-
   pthread_cond_broadcast (&flush_cond);
   pthread_join (flush_thread, NULL);