Merge branch 'ff/rrd-async'
authorFlorian Forster <octo@collectd.org>
Wed, 27 Feb 2013 07:37:42 +0000 (08:37 +0100)
committerFlorian Forster <octo@collectd.org>
Wed, 27 Feb 2013 07:37:42 +0000 (08:37 +0100)
src/collectd.conf.in
src/collectd.conf.pod
src/rrdcached.c
src/rrdtool.c
src/utils_rrdcreate.c
src/utils_rrdcreate.h

index 8744b21..9d8e688 100644 (file)
 #      DaemonAddress "unix:/tmp/rrdcached.sock"
 #      DataDir "@localstatedir@/lib/@PACKAGE_NAME@/rrd"
 #      CreateFiles true
+#      CreateFilesAsync false
 #      CollectStatistics true
 #</Plugin>
 
 #<Plugin rrdtool>
 #      DataDir "@localstatedir@/lib/@PACKAGE_NAME@/rrd"
+#      CreateFilesAsync false
 #      CacheTimeout 120
 #      CacheFlush   900
+#      WritesPerSecond 50
 #</Plugin>
 
 #<Plugin sensors>
index 70bc997..7626a46 100644 (file)
@@ -4644,6 +4644,16 @@ Enables or disables the creation of RRD files. If the daemon is not running
 locally, or B<DataDir> is set to a relative path, this will not work as
 expected. Default is B<true>.
 
+=item B<CreateFilesAsync> B<false>|B<true>
+
+When enabled, new RRD files are enabled asynchronously, using a separate thread
+that runs in the background. This prevents writes to block, which is a problem
+especially when many hundreds of files need to be created at once. However,
+since the purpose of creating the files asynchronously is I<not> to block until
+the file is available, values before the file is available will be discarded.
+When disabled (the default) files are created synchronously, blocking for a
+short while, while the file is being written.
+
 =item B<StepSize> I<Seconds>
 
 B<Force> the stepsize of newly created RRD-files. Ideally (and per default)
@@ -4701,8 +4711,18 @@ can safely ignore these settings.
 
 =item B<DataDir> I<Directory>
 
-Set the directory to store RRD-files under. Per default RRD-files are generated
-beneath the daemon's working directory, i.E<nbsp>e. the B<BaseDir>.
+Set the directory to store RRD files under. By default RRD files are generated
+beneath the daemon's working directory, i.e. the B<BaseDir>.
+
+=item B<CreateFilesAsync> B<false>|B<true>
+
+When enabled, new RRD files are enabled asynchronously, using a separate thread
+that runs in the background. This prevents writes to block, which is a problem
+especially when many hundreds of files need to be created at once. However,
+since the purpose of creating the files asynchronously is I<not> to block until
+the file is available, values before the file is available will be discarded.
+When disabled (the default) files are created synchronously, blocking for a
+short while, while the file is being written.
 
 =item B<StepSize> I<Seconds>
 
@@ -4723,7 +4743,7 @@ a very good reason to do so.
 
 The C<rrdtool plugin> calculates the number of PDPs per CDP based on the
 B<StepSize>, this setting and a timespan. This plugin creates RRD-files with
-three times five RRAs, i. e. five RRAs with the CFs B<MIN>, B<AVERAGE>, and
+three times five RRAs, i.e. five RRAs with the CFs B<MIN>, B<AVERAGE>, and
 B<MAX>. The five RRAs are optimized for graphs covering one hour, one day, one
 week, one month, and one year.
 
index 45553b7..0a98035 100644 (file)
@@ -46,7 +46,9 @@ static rrdcreate_config_t rrdcreate_config =
        /* timespans_num = */ 0,
 
        /* consolidation_functions = */ NULL,
-       /* consolidation_functions_num = */ 0
+       /* consolidation_functions_num = */ 0,
+
+       /* async = */ 0
 };
 
 /*
@@ -250,6 +252,8 @@ static int rc_config (oconfig_item_t *ci)
       status = cf_util_get_string (child, &daemon_address);
     else if (strcasecmp ("CreateFiles", key) == 0)
       status = cf_util_get_boolean (child, &config_create_files);
+    else if (strcasecmp ("CreateFilesAsync", key) == 0)
+      status = cf_util_get_boolean (child, &rrdcreate_config.async);
     else if (strcasecmp ("CollectStatistics", key) == 0)
       status = cf_util_get_boolean (child, &config_collect_stats);
     else if (strcasecmp ("StepSize", key) == 0)
@@ -459,6 +463,8 @@ static int rc_write (const data_set_t *ds, const value_list_t *vl,
             filename);
         return (-1);
       }
+      else if (rrdcreate_config.async)
+        return (0);
     }
   }
 
index b1d13ee..e7c48c0 100644 (file)
@@ -75,6 +75,7 @@ static const char *config_keys[] =
 {
        "CacheTimeout",
        "CacheFlush",
+       "CreateFilesAsync",
        "DataDir",
        "StepSize",
        "HeartBeat",
@@ -102,7 +103,9 @@ static rrdcreate_config_t rrdcreate_config =
        /* timespans_num = */ 0,
 
        /* consolidation_functions = */ NULL,
-       /* consolidation_functions_num = */ 0
+       /* consolidation_functions_num = */ 0,
+
+       /* async = */ 0
 };
 
 /* XXX: If you need to lock both, cache_lock and queue_lock, at the same time,
@@ -910,6 +913,8 @@ static int rrd_write (const data_set_t *ds, const value_list_t *vl,
                                        ds, vl, &rrdcreate_config);
                        if (status != 0)
                                return (-1);
+                       else if (rrdcreate_config.async)
+                               return (0);
                }
                else
                {
@@ -1008,6 +1013,13 @@ static int rrd_config (const char *key, const char *value)
                if (temp > 0)
                        rrdcreate_config.heartbeat = temp;
        }
+       else if (strcasecmp ("CreateFilesAsync", key) == 0)
+       {
+               if (IS_TRUE (value))
+                       rrdcreate_config.async = 1;
+               else
+                       rrdcreate_config.async = 0;
+       }
        else if (strcasecmp ("RRARows", key) == 0)
        {
                int tmp = atoi (value);
index 091b5fa..3b8f342 100644 (file)
@@ -1,6 +1,6 @@
 /**
  * collectd - src/utils_rrdcreate.c
- * Copyright (C) 2006-2008  Florian octo Forster
+ * Copyright (C) 2006-2013  Florian octo Forster
  *
  * This program is free software; you can redistribute it and/or modify it
  * under the terms of the GNU General Public License as published by the
@@ -16,7 +16,7 @@
  * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
  *
  * Authors:
- *   Florian octo Forster <octo at verplant.org>
+ *   Florian octo Forster <octo at collectd.org>
  **/
 
 #include "collectd.h"
 #include <pthread.h>
 #include <rrd.h>
 
+struct srrd_create_args_s
+{
+  char *filename;
+  unsigned long pdp_step;
+  time_t last_up;
+  int argc;
+  char **argv;
+};
+typedef struct srrd_create_args_s srrd_create_args_t;
+
+struct async_create_file_s;
+typedef struct async_create_file_s async_create_file_t;
+struct async_create_file_s
+{
+  char *filename;
+  async_create_file_t *next;
+};
+
 /*
  * Private variables
  */
@@ -51,6 +69,9 @@ static int rra_types_num = STATIC_ARRAY_SIZE (rra_types);
 static pthread_mutex_t librrd_lock = PTHREAD_MUTEX_INITIALIZER;
 #endif
 
+static async_create_file_t *async_creation_list = NULL;
+static pthread_mutex_t async_creation_lock = PTHREAD_MUTEX_INITIALIZER;
+
 /*
  * Private functions
  */
@@ -65,6 +86,71 @@ static void rra_free (int rra_num, char **rra_def) /* {{{ */
   sfree (rra_def);
 } /* }}} void rra_free */
 
+static void srrd_create_args_destroy (srrd_create_args_t *args)
+{
+  if (args == NULL)
+    return;
+
+  sfree (args->filename);
+  if (args->argv != NULL)
+  {
+    int i;
+    for (i = 0; i < args->argc; i++)
+      sfree (args->argv[i]);
+    sfree (args->argv);
+  }
+} /* void srrd_create_args_destroy */
+
+static srrd_create_args_t *srrd_create_args_create (const char *filename,
+    unsigned long pdp_step, time_t last_up,
+    int argc, const char **argv)
+{
+  srrd_create_args_t *args;
+
+  args = malloc (sizeof (*args));
+  if (args == NULL)
+  {
+    ERROR ("srrd_create_args_create: malloc failed.");
+    return (NULL);
+  }
+  memset (args, 0, sizeof (*args));
+  args->filename = NULL;
+  args->pdp_step = pdp_step;
+  args->last_up = last_up;
+  args->argv = NULL;
+
+  args->filename = strdup (filename);
+  if (args->filename == NULL)
+  {
+    ERROR ("srrd_create_args_create: strdup failed.");
+    srrd_create_args_destroy (args);
+    return (NULL);
+  }
+
+  args->argv = calloc ((size_t) (argc + 1), sizeof (*args->argv));
+  if (args->argv == NULL)
+  {
+    ERROR ("srrd_create_args_create: calloc failed.");
+    srrd_create_args_destroy (args);
+    return (NULL);
+  }
+
+  for (args->argc = 0; args->argc < argc; args->argc++)
+  {
+    args->argv[args->argc] = strdup (argv[args->argc]);
+    if (args->argv[args->argc] == NULL)
+    {
+      ERROR ("srrd_create_args_create: strdup failed.");
+      srrd_create_args_destroy (args);
+      return (NULL);
+    }
+  }
+  assert (args->argc == argc);
+  args->argv[args->argc] = NULL;
+
+  return (args);
+} /* srrd_create_args_t *srrd_create_args_create */
+
 /* * * * * * * * * *
  * WARNING:  Magic *
  * * * * * * * * * */
@@ -359,6 +445,198 @@ static int srrd_create (const char *filename, /* {{{ */
 } /* }}} int srrd_create */
 #endif /* !HAVE_THREADSAFE_LIBRRD */
 
+static int lock_file (char const *filename) /* {{{ */
+{
+  async_create_file_t *ptr;
+  struct stat sb;
+  int status;
+
+  pthread_mutex_lock (&async_creation_lock);
+
+  for (ptr = async_creation_list; ptr != NULL; ptr = ptr->next)
+    if (strcmp (filename, ptr->filename) == 0)
+      break;
+
+  if (ptr != NULL)
+  {
+    pthread_mutex_unlock (&async_creation_lock);
+    return (EEXIST);
+  }
+
+  errno = 0;
+  status = stat (filename, &sb);
+  if (errno != ENOENT)
+  {
+    pthread_mutex_unlock (&async_creation_lock);
+    return (EEXIST);
+  }
+
+  ptr = malloc (sizeof (*ptr));
+  if (ptr == NULL)
+  {
+    pthread_mutex_unlock (&async_creation_lock);
+    return (ENOMEM);
+  }
+
+  ptr->filename = strdup (filename);
+  if (ptr->filename == NULL)
+  {
+    pthread_mutex_unlock (&async_creation_lock);
+    sfree (ptr);
+    return (ENOMEM);
+  }
+
+  ptr->next = async_creation_list;
+  async_creation_list = ptr;
+
+  pthread_mutex_unlock (&async_creation_lock);
+
+  return (0);
+} /* }}} int lock_file */
+
+static int unlock_file (char const *filename) /* {{{ */
+{
+  async_create_file_t *this;
+  async_create_file_t *prev;
+
+
+  pthread_mutex_lock (&async_creation_lock);
+
+  prev = NULL;
+  for (this = async_creation_list; this != NULL; this = this->next)
+  {
+    if (strcmp (filename, this->filename) == 0)
+      break;
+    prev = this;
+  }
+
+  if (this == NULL)
+  {
+    pthread_mutex_unlock (&async_creation_lock);
+    return (ENOENT);
+  }
+
+  if (prev == NULL)
+  {
+    assert (this == async_creation_list);
+    async_creation_list = this->next;
+  }
+  else
+  {
+    assert (this == prev->next);
+    prev->next = this->next;
+  }
+  this->next = NULL;
+
+  pthread_mutex_unlock (&async_creation_lock);
+
+  sfree (this->filename);
+  sfree (this);
+
+  return (0);
+} /* }}} int unlock_file */
+
+static void *srrd_create_thread (void *targs) /* {{{ */
+{
+  srrd_create_args_t *args = targs;
+  char tmpfile[PATH_MAX];
+  int status;
+
+  status = lock_file (args->filename);
+  if (status != 0)
+  {
+    if (status == EEXIST)
+      NOTICE ("srrd_create_thread: File \"%s\" is already being created.",
+          args->filename);
+    else
+      ERROR ("srrd_create_thread: Unable to lock file \"%s\".",
+          args->filename);
+    srrd_create_args_destroy (args);
+    return (0);
+  }
+
+  ssnprintf (tmpfile, sizeof (tmpfile), "%s.async", args->filename);
+
+  status = srrd_create (tmpfile, args->pdp_step, args->last_up,
+      args->argc, (void *) args->argv);
+  if (status != 0)
+  {
+    WARNING ("srrd_create_thread: srrd_create (%s) returned status %i.",
+        args->filename, status);
+    unlink (tmpfile);
+    unlock_file (args->filename);
+    srrd_create_args_destroy (args);
+    return (0);
+  }
+
+  status = rename (tmpfile, args->filename);
+  if (status != 0)
+  {
+    char errbuf[1024];
+    ERROR ("srrd_create_thread: rename (\"%s\", \"%s\") failed: %s",
+        tmpfile, args->filename,
+        sstrerror (errno, errbuf, sizeof (errbuf)));
+    unlink (tmpfile);
+    unlock_file (args->filename);
+    srrd_create_args_destroy (args);
+    return (0);
+  }
+
+  DEBUG ("srrd_create_thread: Successfully created RRD file \"%s\".",
+      args->filename);
+
+  unlock_file (args->filename);
+  srrd_create_args_destroy (args);
+
+  return (0);
+} /* }}} void *srrd_create_thread */
+
+static int srrd_create_async (const char *filename, /* {{{ */
+    unsigned long pdp_step, time_t last_up,
+    int argc, const char **argv)
+{
+  srrd_create_args_t *args;
+  pthread_t thread;
+  pthread_attr_t attr;
+  int status;
+
+  DEBUG ("srrd_create_async: Creating \"%s\" in the background.", filename);
+
+  args = srrd_create_args_create (filename, pdp_step, last_up, argc, argv);
+  if (args == NULL)
+    return (-1);
+
+  status = pthread_attr_init (&attr);
+  if (status != 0)
+  {
+    srrd_create_args_destroy (args);
+    return (-1);
+  }
+
+  status = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
+  if (status != 0)
+  {
+    pthread_attr_destroy (&attr);
+    srrd_create_args_destroy (args);
+    return (-1);
+  }
+
+  status = pthread_create (&thread, &attr, srrd_create_thread, args);
+  if (status != 0)
+  {
+    char errbuf[1024];
+    ERROR ("srrd_create_async: pthread_create failed: %s",
+        sstrerror (status, errbuf, sizeof (errbuf)));
+    pthread_attr_destroy (&attr);
+    srrd_create_args_destroy (args);
+    return (status);
+  }
+
+  pthread_attr_destroy (&attr);
+  /* args is freed in srrd_create_thread(). */
+  return (0);
+} /* }}} int srrd_create_async */
+
 /*
  * Public functions
  */
@@ -415,24 +693,36 @@ int cu_rrd_create_file (const char *filename, /* {{{ */
   else
     stepsize = (unsigned long) CDTIME_T_TO_TIME_T (vl->interval);
 
-  status = srrd_create (filename, stepsize, last_up,
-      argc, (const char **) argv);
-
-  free (argv);
-  ds_free (ds_num, ds_def);
-  rra_free (rra_num, rra_def);
-
-  if (status != 0)
+  if (cfg->async)
   {
-    WARNING ("cu_rrd_create_file: srrd_create (%s) returned status %i.",
-        filename, status);
+    status = srrd_create_async (filename, stepsize, last_up,
+        argc, (const char **) argv);
+    if (status != 0)
+      WARNING ("cu_rrd_create_file: srrd_create_async (%s) "
+          "returned status %i.",
+          filename, status);
   }
-  else
+  else /* synchronous */
   {
-    DEBUG ("cu_rrd_create_file: Successfully created RRD file \"%s\".",
-        filename);
+    status = srrd_create (filename, stepsize, last_up,
+        argc, (const char **) argv);
+
+    if (status != 0)
+    {
+      WARNING ("cu_rrd_create_file: srrd_create (%s) returned status %i.",
+          filename, status);
+    }
+    else
+    {
+      DEBUG ("cu_rrd_create_file: Successfully created RRD file \"%s\".",
+          filename);
+    }
   }
 
+  free (argv);
+  ds_free (ds_num, ds_def);
+  rra_free (rra_num, rra_def);
+
   return (status);
 } /* }}} int cu_rrd_create_file */
 
index 103ca57..fdfd6ec 100644 (file)
@@ -1,6 +1,6 @@
 /**
  * collectd - src/utils_rrdcreate.h
- * Copyright (C) 2008  Florian octo Forster
+ * Copyright (C) 2008-2013  Florian octo Forster
  *
  * This program is free software; you can redistribute it and/or modify it
  * under the terms of the GNU General Public License as published by the
@@ -16,7 +16,7 @@
  * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
  *
  * Authors:
- *   Florian octo Forster <octo at verplant.org>
+ *   Florian octo Forster <octo at collectd.org>
  **/
 
 #ifndef UTILS_RRDCREATE_H
@@ -38,6 +38,8 @@ struct rrdcreate_config_s
 
   char **consolidation_functions;
   size_t consolidation_functions_num;
+
+  _Bool async;
 };
 typedef struct rrdcreate_config_s rrdcreate_config_t;