X-Git-Url: https://git.octo.it/?a=blobdiff_plain;f=src%2Frrdtool.c;h=1e9e52a93cbd8e52b774f48c42c2902bb01b3b03;hb=96e0f2341bace029acefe0a88bab96ae326c0ff5;hp=a205ef2ce2ff0465360df9a00d99fa9a0b5e0abc;hpb=f331f9975e8d90a8c430ee7c739c4c3ed8bbb1ec;p=collectd.git

diff --git a/src/rrdtool.c b/src/rrdtool.c
index a205ef2c..71da5474 100644
--- a/src/rrdtool.c
+++ b/src/rrdtool.c
@@ -1,6 +1,8 @@
 /**
  * collectd - src/rrdtool.c
- * Copyright (C) 2006-2008  Florian octo Forster
+ * Copyright (C) 2006-2013  Florian octo Forster
+ * Copyright (C) 2008-2008  Sebastian Harl
+ * Copyright (C) 2009       Mariusz Gronczewski
  *
  * 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,13 +18,16 @@
  * 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>
+ *   Sebastian Harl <sh at tokkee.org>
+ *   Mariusz Gronczewski <xani666 at gmail.com>
  **/
 
 #include "collectd.h"
 #include "plugin.h"
 #include "common.h"
 #include "utils_avltree.h"
+#include "utils_random.h"
 #include "utils_rrdcreate.h"
 
 #include <rrd.h>
@@ -36,10 +41,11 @@
  */
 struct rrd_cache_s
 {
-	int    values_num;
-	char **values;
-	time_t first_value;
-	time_t last_value;
+	int      values_num;
+	char   **values;
+	cdtime_t first_value;
+	cdtime_t last_value;
+	int64_t  random_variation;
 	enum
 	{
 		FLAG_NONE   = 0x00,
@@ -70,13 +76,15 @@ static const char *config_keys[] =
 {
 	"CacheTimeout",
 	"CacheFlush",
+	"CreateFilesAsync",
 	"DataDir",
 	"StepSize",
 	"HeartBeat",
 	"RRARows",
 	"RRATimespan",
 	"XFF",
-	"WritesPerSecond"
+	"WritesPerSecond",
+	"RandomTimeout"
 };
 static int config_keys_num = STATIC_ARRAY_SIZE (config_keys);
 
@@ -96,14 +104,17 @@ 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,
  * ALWAYS lock `cache_lock' first! */
-static int         cache_timeout = 0;
-static int         cache_flush_timeout = 0;
-static time_t      cache_flush_last;
+static cdtime_t    cache_timeout = 0;
+static cdtime_t    cache_flush_timeout = 0;
+static cdtime_t    random_timeout = TIME_T_TO_CDTIME_T (1);
+static cdtime_t    cache_flush_last;
 static c_avl_tree_t *cache = NULL;
 static pthread_mutex_t cache_lock = PTHREAD_MUTEX_INITIALIZER;
 
@@ -111,7 +122,8 @@ static rrd_queue_t    *queue_head = NULL;
 static rrd_queue_t    *queue_tail = NULL;
 static rrd_queue_t    *flushq_head = NULL;
 static rrd_queue_t    *flushq_tail = NULL;
-static pthread_t       queue_thread = 0;
+static pthread_t       queue_thread;
+static int             queue_thread_running = 1;
 static pthread_mutex_t queue_lock = PTHREAD_MUTEX_INITIALIZER;
 static pthread_cond_t  queue_cond = PTHREAD_COND_INITIALIZER;
 
@@ -177,7 +189,7 @@ static int srrd_update (char *filename, char *template,
 	if (status != 0)
 	{
 		WARNING ("rrdtool plugin: rrd_update_r failed: %s: %s",
-				argv[1], rrd_get_error ());
+				filename, rrd_get_error ());
 	}
 
 	sfree (new_argv);
@@ -191,11 +203,13 @@ static int value_list_to_string (char *buffer, int buffer_len,
 {
 	int offset;
 	int status;
+	time_t tt;
 	int i;
 
 	memset (buffer, '\0', buffer_len);
 
-	status = ssnprintf (buffer, buffer_len, "%u", (unsigned int) vl->time);
+	tt = CDTIME_T_TO_TIME_T (vl->time);
+	status = ssnprintf (buffer, buffer_len, "%u", (unsigned int) tt);
 	if ((status < 1) || (status >= buffer_len))
 		return (-1);
 	offset = status;
@@ -203,15 +217,23 @@ static int value_list_to_string (char *buffer, int buffer_len,
 	for (i = 0; i < ds->ds_num; i++)
 	{
 		if ((ds->ds[i].type != DS_TYPE_COUNTER)
-				&& (ds->ds[i].type != DS_TYPE_GAUGE))
+				&& (ds->ds[i].type != DS_TYPE_GAUGE)
+				&& (ds->ds[i].type != DS_TYPE_DERIVE)
+				&& (ds->ds[i].type != DS_TYPE_ABSOLUTE))
 			return (-1);
 
 		if (ds->ds[i].type == DS_TYPE_COUNTER)
 			status = ssnprintf (buffer + offset, buffer_len - offset,
 					":%llu", vl->values[i].counter);
-		else
+		else if (ds->ds[i].type == DS_TYPE_GAUGE)
 			status = ssnprintf (buffer + offset, buffer_len - offset,
 					":%lf", vl->values[i].gauge);
+		else if (ds->ds[i].type == DS_TYPE_DERIVE)
+			status = ssnprintf (buffer + offset, buffer_len - offset,
+					":%"PRIi64, vl->values[i].derive);
+		else /*if (ds->ds[i].type == DS_TYPE_ABSOLUTE) */
+			status = ssnprintf (buffer + offset, buffer_len - offset,
+					":%"PRIu64, vl->values[i].absolute);
 
 		if ((status < 1) || (status >= (buffer_len - offset)))
 			return (-1);
@@ -222,47 +244,41 @@ static int value_list_to_string (char *buffer, int buffer_len,
 	return (0);
 } /* int value_list_to_string */
 
-static int value_list_to_filename (char *buffer, int buffer_len,
-		const data_set_t __attribute__((unused)) *ds, const value_list_t *vl)
+static int value_list_to_filename (char *buffer, size_t buffer_size,
+		value_list_t const *vl)
 {
-	int offset = 0;
+	char const suffix[] = ".rrd";
 	int status;
+	size_t len;
 
 	if (datadir != NULL)
 	{
-		status = ssnprintf (buffer + offset, buffer_len - offset,
-				"%s/", datadir);
-		if ((status < 1) || (status >= buffer_len - offset))
-			return (-1);
-		offset += status;
+		size_t datadir_len = strlen (datadir) + 1;
+
+		if (datadir_len >= buffer_size)
+			return (ENOMEM);
+
+		sstrncpy (buffer, datadir, buffer_size);
+		buffer[datadir_len - 1] = '/';
+		buffer[datadir_len] = 0;
+
+		buffer += datadir_len;
+		buffer_size -= datadir_len;
 	}
 
-	status = ssnprintf (buffer + offset, buffer_len - offset,
-			"%s/", vl->host);
-	if ((status < 1) || (status >= buffer_len - offset))
-		return (-1);
-	offset += status;
+	status = FORMAT_VL (buffer, buffer_size, vl);
+	if (status != 0)
+		return (status);
 
-	if (strlen (vl->plugin_instance) > 0)
-		status = ssnprintf (buffer + offset, buffer_len - offset,
-				"%s-%s/", vl->plugin, vl->plugin_instance);
-	else
-		status = ssnprintf (buffer + offset, buffer_len - offset,
-				"%s/", vl->plugin);
-	if ((status < 1) || (status >= buffer_len - offset))
-		return (-1);
-	offset += status;
+	len = strlen (buffer);
+	assert (len < buffer_size);
+	buffer += len;
+	buffer_size -= len;
 
-	if (strlen (vl->type_instance) > 0)
-		status = ssnprintf (buffer + offset, buffer_len - offset,
-				"%s-%s.rrd", vl->type, vl->type_instance);
-	else
-		status = ssnprintf (buffer + offset, buffer_len - offset,
-				"%s.rrd", vl->type);
-	if ((status < 1) || (status >= buffer_len - offset))
-		return (-1);
-	offset += status;
+	if (buffer_size <= sizeof (suffix))
+		return (ENOMEM);
 
+	memcpy (buffer, suffix, sizeof (suffix));
 	return (0);
 } /* int value_list_to_filename */
 
@@ -287,7 +303,7 @@ static void *rrd_queue_thread (void __attribute__((unused)) *data)
 
                 pthread_mutex_lock (&queue_lock);
                 /* Wait for values to arrive */
-                while (true)
+                while (42)
                 {
                   struct timespec ts_wait;
 
@@ -326,7 +342,7 @@ static void *rrd_queue_thread (void __attribute__((unused)) *data)
                       &ts_wait);
                   if (status == ETIMEDOUT)
                     break;
-                } /* while (true) */
+                } /* while (42) */
 
                 /* XXX: If you need to lock both, cache_lock and queue_lock, at
                  * the same time, ALWAYS lock `cache_lock' first! */
@@ -403,8 +419,9 @@ static void *rrd_queue_thread (void __attribute__((unused)) *data)
 		/* Write the values to the RRD-file */
 		srrd_update (queue_entry->filename, NULL,
 				values_num, (const char **)values);
-		DEBUG ("rrdtool plugin: queue thread: Wrote %i values to %s",
-				values_num, queue_entry->filename);
+		DEBUG ("rrdtool plugin: queue thread: Wrote %i value%s to %s",
+				values_num, (values_num == 1) ? "" : "s",
+				queue_entry->filename);
 
 		for (i = 0; i < values_num; i++)
 		{
@@ -415,11 +432,6 @@ static void *rrd_queue_thread (void __attribute__((unused)) *data)
 		sfree (queue_entry);
 	} /* while (42) */
 
-	pthread_mutex_lock (&cache_lock);
-	c_avl_destroy (cache);
-	cache = NULL;
-	pthread_mutex_unlock (&cache_lock);
-
 	pthread_exit ((void *) 0);
 	return ((void *) 0);
 } /* void *rrd_queue_thread */
@@ -498,10 +510,11 @@ static int rrd_queue_dequeue (const char *filename,
   return (0);
 } /* int rrd_queue_dequeue */
 
-static void rrd_cache_flush (int timeout)
+/* XXX: You must hold "cache_lock" when calling this function! */
+static void rrd_cache_flush (cdtime_t timeout)
 {
 	rrd_cache_t *rc;
-	time_t       now;
+	cdtime_t     now;
 
 	char **keys = NULL;
 	int    keys_num = 0;
@@ -510,9 +523,11 @@ static void rrd_cache_flush (int timeout)
 	c_avl_iterator_t *iter;
 	int i;
 
-	DEBUG ("rrdtool plugin: Flushing cache, timeout = %i", timeout);
+	DEBUG ("rrdtool plugin: Flushing cache, timeout = %.3f",
+			CDTIME_T_TO_DOUBLE (timeout));
 
-	now = time (NULL);
+	now = cdtime ();
+	timeout = TIME_T_TO_CDTIME_T (timeout);
 
 	/* Build a list of entries to be flushed */
 	iter = c_avl_get_iterator (cache);
@@ -520,7 +535,9 @@ static void rrd_cache_flush (int timeout)
 	{
 		if (rc->flags != FLAG_NONE)
 			continue;
-		else if ((now - rc->first_value) < timeout)
+		/* timeout == 0  =>  flush everything */
+		else if ((timeout != 0)
+				&& ((now - rc->first_value) < timeout))
 			continue;
 		else if (rc->values_num > 0)
 		{
@@ -573,10 +590,11 @@ static void rrd_cache_flush (int timeout)
 	cache_flush_last = now;
 } /* void rrd_cache_flush */
 
-static int rrd_cache_flush_identifier (int timeout, const char *identifier)
+static int rrd_cache_flush_identifier (cdtime_t timeout,
+    const char *identifier)
 {
   rrd_cache_t *rc;
-  time_t now;
+  cdtime_t now;
   int status;
   char key[2048];
 
@@ -586,7 +604,7 @@ static int rrd_cache_flush_identifier (int timeout, const char *identifier)
     return (0);
   }
 
-  now = time (NULL);
+  now = cdtime ();
 
   if (datadir == NULL)
     snprintf (key, sizeof (key), "%s.rrd",
@@ -599,7 +617,7 @@ static int rrd_cache_flush_identifier (int timeout, const char *identifier)
   status = c_avl_get (cache, key, (void *) &rc);
   if (status != 0)
   {
-    WARNING ("rrdtool plugin: rrd_cache_flush_identifier: "
+    INFO ("rrdtool plugin: rrd_cache_flush_identifier: "
         "c_avl_get (%s) failed. Does that file really exist?",
         key);
     return (status);
@@ -630,8 +648,30 @@ static int rrd_cache_flush_identifier (int timeout, const char *identifier)
   return (status);
 } /* int rrd_cache_flush_identifier */
 
+static int64_t rrd_get_random_variation (void)
+{
+  long min;
+  long max;
+
+  if (random_timeout <= 0)
+    return (0);
+
+  /* Assure that "cache_timeout + random_variation" is never negative. */
+  if (random_timeout > cache_timeout)
+  {
+	  INFO ("rrdtool plugin: Adjusting \"RandomTimeout\" to %.3f seconds.",
+			  CDTIME_T_TO_DOUBLE (cache_timeout));
+	  random_timeout = cache_timeout;
+  }
+
+  max = (long) (random_timeout / 2);
+  min = max - ((long) random_timeout);
+
+  return ((int64_t) cdrand_range (min, max));
+} /* int64_t rrd_get_random_variation */
+
 static int rrd_cache_insert (const char *filename,
-		const char *value, time_t value_time)
+		const char *value, cdtime_t value_time)
 {
 	rrd_cache_t *rc = NULL;
 	int new_rc = 0;
@@ -652,13 +692,14 @@ static int rrd_cache_insert (const char *filename,
 
 	if (rc == NULL)
 	{
-		rc = (rrd_cache_t *) malloc (sizeof (rrd_cache_t));
+		rc = malloc (sizeof (*rc));
 		if (rc == NULL)
 			return (-1);
 		rc->values_num = 0;
 		rc->values = NULL;
 		rc->first_value = 0;
 		rc->last_value = 0;
+		rc->random_variation = rrd_get_random_variation ();
 		rc->flags = FLAG_NONE;
 		new_rc = 1;
 	}
@@ -666,9 +707,9 @@ static int rrd_cache_insert (const char *filename,
 	if (rc->last_value >= value_time)
 	{
 		pthread_mutex_unlock (&cache_lock);
-		WARNING ("rrdtool plugin: (rc->last_value = %u) >= (value_time = %u)",
-				(unsigned int) rc->last_value,
-				(unsigned int) value_time);
+		DEBUG ("rrdtool plugin: (rc->last_value = %"PRIu64") "
+				">= (value_time = %"PRIu64")",
+				rc->last_value, value_time);
 		return (-1);
 	}
 
@@ -725,11 +766,11 @@ static int rrd_cache_insert (const char *filename,
 	}
 
 	DEBUG ("rrdtool plugin: rrd_cache_insert: file = %s; "
-			"values_num = %i; age = %lu;",
+			"values_num = %i; age = %.3f;",
 			filename, rc->values_num,
-			(unsigned long)(rc->last_value - rc->first_value));
+			CDTIME_T_TO_DOUBLE (rc->last_value - rc->first_value));
 
-	if ((rc->last_value - rc->first_value) >= cache_timeout)
+	if ((rc->last_value - rc->first_value) >= (cache_timeout + rc->random_variation))
 	{
 		/* XXX: If you need to lock both, cache_lock and queue_lock, at
 		 * the same time, ALWAYS lock `cache_lock' first! */
@@ -740,6 +781,8 @@ static int rrd_cache_insert (const char *filename,
 			status = rrd_queue_enqueue (filename, &queue_head, &queue_tail);
 			if (status == 0)
 				rc->flags = FLAG_QUEUED;
+
+                        rc->random_variation = rrd_get_random_variation ();
 		}
 		else
 		{
@@ -748,7 +791,7 @@ static int rrd_cache_insert (const char *filename,
 	}
 
 	if ((cache_timeout > 0) &&
-			((time (NULL) - cache_flush_last) > cache_flush_timeout))
+			((cdtime () - cache_flush_last) > cache_flush_timeout))
 		rrd_cache_flush (cache_flush_timeout);
 
 	pthread_mutex_unlock (&cache_lock);
@@ -756,6 +799,59 @@ static int rrd_cache_insert (const char *filename,
 	return (0);
 } /* int rrd_cache_insert */
 
+static int rrd_cache_destroy (void) /* {{{ */
+{
+  void *key = NULL;
+  void *value = NULL;
+
+  int non_empty = 0;
+
+  pthread_mutex_lock (&cache_lock);
+
+  if (cache == NULL)
+  {
+    pthread_mutex_unlock (&cache_lock);
+    return (0);
+  }
+
+  while (c_avl_pick (cache, &key, &value) == 0)
+  {
+    rrd_cache_t *rc;
+    int i;
+
+    sfree (key);
+    key = NULL;
+
+    rc = value;
+    value = NULL;
+
+    if (rc->values_num > 0)
+      non_empty++;
+
+    for (i = 0; i < rc->values_num; i++)
+      sfree (rc->values[i]);
+    sfree (rc->values);
+    sfree (rc);
+  }
+
+  c_avl_destroy (cache);
+  cache = NULL;
+
+  if (non_empty > 0)
+  {
+    INFO ("rrdtool plugin: %i cache %s had values when destroying the cache.",
+        non_empty, (non_empty == 1) ? "entry" : "entries");
+  }
+  else
+  {
+    DEBUG ("rrdtool plugin: No values have been lost "
+        "when destroying the cache.");
+  }
+
+  pthread_mutex_unlock (&cache_lock);
+  return (0);
+} /* }}} int rrd_cache_destroy */
+
 static int rrd_compare_numeric (const void *a_ptr, const void *b_ptr)
 {
 	int a = *((int *) a_ptr);
@@ -769,19 +865,23 @@ static int rrd_compare_numeric (const void *a_ptr, const void *b_ptr)
 		return (0);
 } /* int rrd_compare_numeric */
 
-static int rrd_write (const data_set_t *ds, const value_list_t *vl)
+static int rrd_write (const data_set_t *ds, const value_list_t *vl,
+		user_data_t __attribute__((unused)) *user_data)
 {
 	struct stat  statbuf;
 	char         filename[512];
 	char         values[512];
 	int          status;
 
+	if (do_shutdown)
+		return (0);
+
 	if (0 != strcmp (ds->type, vl->type)) {
 		ERROR ("rrdtool plugin: DS type does not match value list type");
 		return -1;
 	}
 
-	if (value_list_to_filename (filename, sizeof (filename), ds, vl) != 0)
+	if (value_list_to_filename (filename, sizeof (filename), vl) != 0)
 		return (-1);
 
 	if (value_list_to_string (values, sizeof (values), ds, vl) != 0)
@@ -795,6 +895,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
 		{
@@ -817,7 +919,8 @@ static int rrd_write (const data_set_t *ds, const value_list_t *vl)
 	return (status);
 } /* int rrd_write */
 
-static int rrd_flush (int timeout, const char *identifier)
+static int rrd_flush (cdtime_t timeout, const char *identifier,
+		__attribute__((unused)) user_data_t *user_data)
 {
 	pthread_mutex_lock (&cache_lock);
 
@@ -836,7 +939,7 @@ static int rrd_config (const char *key, const char *value)
 {
 	if (strcasecmp ("CacheTimeout", key) == 0)
 	{
-		int tmp = atoi (value);
+		double tmp = atof (value);
 		if (tmp < 0)
 		{
 			fprintf (stderr, "rrdtool: `CacheTimeout' must "
@@ -845,7 +948,7 @@ static int rrd_config (const char *key, const char *value)
 					"be greater than 0.\n");
 			return (1);
 		}
-		cache_timeout = tmp;
+		cache_timeout = DOUBLE_TO_CDTIME_T (tmp);
 	}
 	else if (strcasecmp ("CacheFlush", key) == 0)
 	{
@@ -882,7 +985,7 @@ static int rrd_config (const char *key, const char *value)
 	}
 	else if (strcasecmp ("StepSize", key) == 0)
 	{
-		int temp = atoi (value);
+		unsigned long temp = strtoul (value, NULL, 0);
 		if (temp > 0)
 			rrdcreate_config.stepsize = temp;
 	}
@@ -892,6 +995,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);
@@ -976,6 +1086,23 @@ static int rrd_config (const char *key, const char *value)
 			write_rate = 1.0 / wps;
 		}
 	}
+	else if (strcasecmp ("RandomTimeout", key) == 0)
+        {
+		double tmp;
+
+		tmp = atof (value);
+		if (tmp < 0.0)
+		{
+			fprintf (stderr, "rrdtool: `RandomTimeout' must "
+					"be greater than or equal to zero.\n");
+			ERROR ("rrdtool: `RandomTimeout' must "
+					"be greater then or equal to zero.");
+		}
+		else
+		{
+			random_timeout = DOUBLE_TO_CDTIME_T (tmp);
+		}
+	}
 	else
 	{
 		return (-1);
@@ -986,7 +1113,7 @@ static int rrd_config (const char *key, const char *value)
 static int rrd_shutdown (void)
 {
 	pthread_mutex_lock (&cache_lock);
-	rrd_cache_flush (-1);
+	rrd_cache_flush (0);
 	pthread_mutex_unlock (&cache_lock);
 
 	pthread_mutex_lock (&queue_lock);
@@ -994,14 +1121,28 @@ static int rrd_shutdown (void)
 	pthread_cond_signal (&queue_cond);
 	pthread_mutex_unlock (&queue_lock);
 
+	if ((queue_thread_running != 0)
+			&& ((queue_head != NULL) || (flushq_head != NULL)))
+	{
+		INFO ("rrdtool plugin: Shutting down the queue thread. "
+				"This may take a while.");
+	}
+	else if (queue_thread_running != 0)
+	{
+		INFO ("rrdtool plugin: Shutting down the queue thread.");
+	}
+
 	/* Wait for all the values to be written to disk before returning. */
-	if (queue_thread != 0)
+	if (queue_thread_running != 0)
 	{
 		pthread_join (queue_thread, NULL);
-		queue_thread = 0;
+		memset (&queue_thread, 0, sizeof (queue_thread));
+		queue_thread_running = 0;
 		DEBUG ("rrdtool plugin: queue_thread exited.");
 	}
 
+	rrd_cache_destroy ();
+
 	return (0);
 } /* int rrd_shutdown */
 
@@ -1014,22 +1155,9 @@ static int rrd_init (void)
 		return (0);
 	init_once = 1;
 
-	if (rrdcreate_config.stepsize < 0)
-		rrdcreate_config.stepsize = 0;
 	if (rrdcreate_config.heartbeat <= 0)
 		rrdcreate_config.heartbeat = 2 * rrdcreate_config.stepsize;
 
-	if ((rrdcreate_config.heartbeat > 0)
-			&& (rrdcreate_config.heartbeat < interval_g))
-		WARNING ("rrdtool plugin: Your `heartbeat' is "
-				"smaller than your `interval'. This will "
-				"likely cause problems.");
-	else if ((rrdcreate_config.stepsize > 0)
-			&& (rrdcreate_config.stepsize < interval_g))
-		WARNING ("rrdtool plugin: Your `stepsize' is "
-				"smaller than your `interval'. This will "
-				"create needlessly big RRD-files.");
-
 	/* Set the cache up */
 	pthread_mutex_lock (&cache_lock);
 
@@ -1040,10 +1168,9 @@ static int rrd_init (void)
 		return (-1);
 	}
 
-	cache_flush_last = time (NULL);
-	if (cache_timeout < 2)
+	cache_flush_last = cdtime ();
+	if (cache_timeout == 0)
 	{
-		cache_timeout = 0;
 		cache_flush_timeout = 0;
 	}
 	else if (cache_flush_timeout < cache_timeout)
@@ -1051,14 +1178,16 @@ static int rrd_init (void)
 
 	pthread_mutex_unlock (&cache_lock);
 
-	status = pthread_create (&queue_thread, NULL, rrd_queue_thread, NULL);
+	status = plugin_thread_create (&queue_thread, /* attr = */ NULL,
+			rrd_queue_thread, /* args = */ NULL);
 	if (status != 0)
 	{
 		ERROR ("rrdtool plugin: Cannot create queue-thread.");
 		return (-1);
 	}
+	queue_thread_running = 1;
 
-	DEBUG ("rrdtool plugin: rrd_init: datadir = %s; stepsize = %i;"
+	DEBUG ("rrdtool plugin: rrd_init: datadir = %s; stepsize = %lu;"
 			" heartbeat = %i; rrarows = %i; xff = %lf;",
 			(datadir == NULL) ? "(null)" : datadir,
 			rrdcreate_config.stepsize,
@@ -1074,7 +1203,7 @@ void module_register (void)
 	plugin_register_config ("rrdtool", rrd_config,
 			config_keys, config_keys_num);
 	plugin_register_init ("rrdtool", rrd_init);
-	plugin_register_write ("rrdtool", rrd_write);
-	plugin_register_flush ("rrdtool", rrd_flush);
+	plugin_register_write ("rrdtool", rrd_write, /* user_data = */ NULL);
+	plugin_register_flush ("rrdtool", rrd_flush, /* user_data = */ NULL);
 	plugin_register_shutdown ("rrdtool", rrd_shutdown);
 }