X-Git-Url: https://git.octo.it/?a=blobdiff_plain;f=src%2Frrdtool.c;h=98ec66621c93460197891d67b838b5ff8fad8803;hb=6b89dd24f42ea2ee6ac5ce05a2fb3f18ea5331aa;hp=debb7bd4ff4416386d5881ec814e1dbd63f7313a;hpb=6597f3a6584704f92f824f3cf7bac3369102e8a0;p=collectd.git

diff --git a/src/rrdtool.c b/src/rrdtool.c
index debb7bd4..98ec6662 100644
--- a/src/rrdtool.c
+++ b/src/rrdtool.c
@@ -111,7 +111,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;
 
@@ -279,14 +280,17 @@ static void *rrd_queue_thread (void __attribute__((unused)) *data)
 		rrd_cache_t *cache_entry;
 		char **values;
 		int    values_num;
+		int    status;
 		int    i;
 
+		values = NULL;
+		values_num = 0;
+
                 pthread_mutex_lock (&queue_lock);
                 /* Wait for values to arrive */
                 while (true)
                 {
                   struct timespec ts_wait;
-                  int status;
 
                   while ((flushq_head == NULL) && (queue_head == NULL)
                       && (do_shutdown == 0))
@@ -361,17 +365,28 @@ static void *rrd_queue_thread (void __attribute__((unused)) *data)
 		 * we make a copy of it's values */
 		pthread_mutex_lock (&cache_lock);
 
-		c_avl_get (cache, queue_entry->filename, (void *) &cache_entry);
+		status = c_avl_get (cache, queue_entry->filename,
+				(void *) &cache_entry);
 
-		values = cache_entry->values;
-		values_num = cache_entry->values_num;
+		if (status == 0)
+		{
+			values = cache_entry->values;
+			values_num = cache_entry->values_num;
 
-		cache_entry->values = NULL;
-		cache_entry->values_num = 0;
-		cache_entry->flags = FLAG_NONE;
+			cache_entry->values = NULL;
+			cache_entry->values_num = 0;
+			cache_entry->flags = FLAG_NONE;
+		}
 
 		pthread_mutex_unlock (&cache_lock);
 
+		if (status != 0)
+		{
+			sfree (queue_entry->filename);
+			sfree (queue_entry);
+			continue;
+		}
+
 		/* Update `tv_next_update' */
 		if (write_rate > 0.0) 
                 {
@@ -389,8 +404,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++)
 		{
@@ -625,6 +641,15 @@ static int rrd_cache_insert (const char *filename,
 
 	pthread_mutex_lock (&cache_lock);
 
+	/* This shouldn't happen, but it did happen at least once, so we'll be
+	 * careful. */
+	if (cache == NULL)
+	{
+		pthread_mutex_unlock (&cache_lock);
+		WARNING ("rrdtool plugin: cache == NULL.");
+		return (-1);
+	}
+
 	c_avl_get (cache, filename, (void *) &rc);
 
 	if (rc == NULL)
@@ -746,7 +771,8 @@ 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];
@@ -794,7 +820,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 (int timeout, const char *identifier,
+		user_data_t __attribute__((unused)) *user_data)
 {
 	pthread_mutex_lock (&cache_lock);
 
@@ -971,14 +998,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.");
 	}
 
+	/* TODO: Maybe it'd be a good idea to free the cache here.. */
+
 	return (0);
 } /* int rrd_shutdown */
 
@@ -1023,12 +1064,14 @@ static int rrd_init (void)
 
 	pthread_mutex_unlock (&cache_lock);
 
-	status = pthread_create (&queue_thread, NULL, rrd_queue_thread, NULL);
+	status = pthread_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;"
 			" heartbeat = %i; rrarows = %i; xff = %lf;",
@@ -1046,7 +1089,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);
 }