added support for FLUSHALL command -- kevin brintnall

[rrdtool.git] / src / rrd_daemon.c

diff --git a/src/rrd_daemon.c b/src/rrd_daemon.c
index d30893e..0e29f13 100644 (file)
--- a/src/rrd_daemon.c
+++ b/src/rrd_daemon.c
@@ -236,7 +236,7 @@ static int write_pidfile (int fd) /* {{{ */
   fh = fdopen (fd, "w");
   if (fh == NULL)
   {
-    RRDD_LOG (LOG_ERR, "write_pidfile: fopen() failed.");
+    RRDD_LOG (LOG_ERR, "write_pidfile: fdopen() failed.");
     close(fd);
     return (-1);
   }
@@ -428,6 +428,7 @@ static int enqueue_cache_item (cache_item_t *ci, /* {{{ */
 
   if (did_insert)
   {
+    pthread_cond_broadcast(&cache_cond);
     pthread_mutex_lock (&stats_lock);
     stats_queue_length++;
     pthread_mutex_unlock (&stats_lock);
@@ -500,7 +501,7 @@ static int flush_old_values (int max_age)
   if (max_age > 0)
     cfd.abs_timeout = cfd.now - max_age;
   else
-    cfd.abs_timeout = cfd.now + 1;
+    cfd.abs_timeout = cfd.now + 2*config_write_jitter + 1;
 
   /* `tree_callback_flush' will return the keys of all values that haven't
    * been touched in the last `config_flush_interval' seconds in `cfd'.
@@ -750,7 +751,6 @@ static int flush_file (const char *filename) /* {{{ */
 
   /* Enqueue at head */
   enqueue_cache_item (ci, HEAD);
-  pthread_cond_signal (&cache_cond);
 
   pthread_cond_wait(&ci->flushed, &cache_lock);
   pthread_mutex_unlock(&cache_lock);
@@ -769,8 +769,9 @@ static int handle_request_help (int fd, /* {{{ */
 
   char *help_help[] =
   {
-    "4 Command overview\n",
+    "5 Command overview\n",
     "FLUSH <filename>\n",
+    "FLUSHALL\n",
     "HELP [<command>]\n",
     "UPDATE <filename> <values> [<values> ...]\n",
     "STATS\n"
@@ -787,6 +788,15 @@ static int handle_request_help (int fd, /* {{{ */
   };
   size_t help_flush_len = sizeof (help_flush) / sizeof (help_flush[0]);
 
+  char *help_flushall[] =
+  {
+    "3 Help for FLUSHALL\n",
+    "Usage: FLUSHALL\n",
+    "\n",
+    "Triggers writing of all pending updates.  Returns immediately.\n"
+  };
+  size_t help_flushall_len = sizeof(help_flushall) / sizeof(help_flushall[0]);
+
   char *help_update[] =
   {
     "9 Help for UPDATE\n",
@@ -830,6 +840,11 @@ static int handle_request_help (int fd, /* {{{ */
       help_text = help_flush;
       help_text_len = help_flush_len;
     }
+    else if (strcasecmp (command, "flushall") == 0)
+    {
+      help_text = help_flushall;
+      help_text_len = help_flushall_len;
+    }
     else if (strcasecmp (command, "stats") == 0)
     {
       help_text = help_stats;
@@ -992,6 +1007,27 @@ static int handle_request_flush (int fd, /* {{{ */
   return (0);
 } /* }}} int handle_request_flush */
 
+static int handle_request_flushall(int fd) /* {{{ */
+{
+  int status;
+  char answer[] ="0 Started flush.\n";
+
+  RRDD_LOG(LOG_DEBUG, "Received FLUSHALL");
+
+  pthread_mutex_lock(&cache_lock);
+  flush_old_values(-1);
+  pthread_mutex_unlock(&cache_lock);
+
+  status = swrite(fd, answer, strlen(answer));
+  if (status < 0)
+  {
+    status = errno;
+    RRDD_LOG(LOG_INFO, "handle_request_flushall: swrite returned an error.");
+  }
+
+  return (status);
+}
+
 static int handle_request_update (int fd, /* {{{ */
     char *buffer, size_t buffer_size)
 {
@@ -1135,7 +1171,6 @@ static int handle_request_update (int fd, /* {{{ */
       && (ci->values_num > 0))
   {
     enqueue_cache_item (ci, TAIL);
-    pthread_cond_signal (&cache_cond);
   }
 
   pthread_mutex_unlock (&cache_lock);
@@ -1223,6 +1258,10 @@ static int handle_request (int fd, char *buffer, size_t buffer_size) /* {{{ */
   {
     return (handle_request_flush (fd, buffer_ptr, buffer_size));
   }
+  else if (strcasecmp (command, "flushall") == 0)
+  {
+    return (handle_request_flushall(fd));
+  }
   else if (strcasecmp (command, "stats") == 0)
   {
     return (handle_request_stats (fd, buffer_ptr, buffer_size));