src/rrd_daemon.c: Implement the `flush' command.

diff --git a/src/rrd_daemon.c b/src/rrd_daemon.c
index 7423088..bd9db79 100644 (file)
--- a/src/rrd_daemon.c
+++ b/src/rrd_daemon.c
@@ -144,6 +144,8 @@ static cache_item_t   *cache_queue_tail = NULL;
 static pthread_mutex_t cache_lock = PTHREAD_MUTEX_INITIALIZER;
 static pthread_cond_t  cache_cond = PTHREAD_COND_INITIALIZER;
 
+static pthread_cond_t  flush_cond = PTHREAD_COND_INITIALIZER;
+
 static int config_write_interval = 300;
 static int config_flush_interval = 3600;
 
@@ -355,6 +357,7 @@ static void *queue_thread_main (void *args __attribute__((unused))) /* {{{ */
       free (values[i]);
 
     pthread_mutex_lock (&cache_lock);
+    pthread_cond_broadcast (&flush_cond);
   } /* while (do_shutdown == 0) */
   pthread_mutex_unlock (&cache_lock);
 
@@ -423,6 +426,79 @@ static int buffer_get_field (char **buffer_ret, /* {{{ */
   return (0);
 } /* }}} int buffer_get_field */
 
+static int flush_file (const char *filename) /* {{{ */
+{
+  cache_item_t *ci;
+
+  pthread_mutex_lock (&cache_lock);
+
+  ci = g_tree_lookup (cache_tree, filename);
+  if (ci == NULL)
+  {
+    pthread_mutex_unlock (&cache_lock);
+    return (ENOENT);
+  }
+
+  /* Enqueue at head */
+  enqueue_cache_item (ci, HEAD);
+  pthread_cond_signal (&cache_cond);
+
+  while ((ci->flags & CI_FLAGS_IN_QUEUE) != 0)
+  {
+    ci = NULL;
+
+    pthread_cond_wait (&flush_cond, &cache_lock);
+
+    ci = g_tree_lookup (cache_tree, filename);
+    if (ci == NULL)
+    {
+      RRDD_LOG (LOG_ERR, "flush_file: Tree node went away "
+          "while waiting for flush.");
+      pthread_mutex_unlock (&cache_lock);
+      return (-1);
+    }
+  }
+
+  pthread_mutex_unlock (&cache_lock);
+  return (0);
+} /* }}} int flush_file */
+
+static int handle_request_flush (int fd, /* {{{ */
+    char *buffer, size_t buffer_size)
+{
+  char *file;
+  int status;
+  char result[4096];
+
+  status = buffer_get_field (&buffer, &buffer_size, &file);
+  if (status != 0)
+  {
+    RRDD_LOG (LOG_INFO, "handle_request_flush: Cannot get file name.");
+    return (-1);
+  }
+
+  status = flush_file (file);
+  if (status == 0)
+    snprintf (result, sizeof (result), "0 Successfully flushed %s.\n", file);
+  else if (status == ENOENT)
+    snprintf (result, sizeof (result), "-1 No such file: %s.\n", file);
+  else if (status < 0)
+    strncpy (result, "-1 Internal error.\n", sizeof (result));
+  else
+    snprintf (result, sizeof (result), "-1 Failed with status %i.\n", status);
+  result[sizeof (result) - 1] = 0;
+
+  status = write (fd, result, strlen (result));
+  if (status < 0)
+  {
+    status = errno;
+    RRDD_LOG (LOG_INFO, "handle_request_flush: write(2) returned an error.");
+    return (status);
+  }
+
+  return (0);
+} /* }}} int handle_request_flush */
+
 static int handle_request_update (int fd, /* {{{ */
     char *buffer, size_t buffer_size)
 {
@@ -574,6 +650,10 @@ static int handle_request (int fd) /* {{{ */
   {
     return (handle_request_update (fd, buffer_ptr, buffer_size));
   }
+  else if (strcmp (command, "flush") == 0)
+  {
+    return (handle_request_flush (fd, buffer_ptr, buffer_size));
+  }
   else
   {
     RRDD_LOG (LOG_INFO, "handle_request: unknown command: %s.", buffer);