X-Git-Url: https://git.octo.it/?p=collectd.git;a=blobdiff_plain;f=src%2Fplugin.c;h=30a1ff1a304911f91e571fe56cd22c4d15535078;hp=22f89696d6d0bba6d51c168fdf0ebaebd836e9b7;hb=633c3966f770e4d46651a2fe219a18d8a9907a9f;hpb=cd35a6a4528cf78383266fbae770b9c7161fda25

diff --git a/src/plugin.c b/src/plugin.c
index 22f89696..30a1ff1a 100644
--- a/src/plugin.c
+++ b/src/plugin.c
@@ -1,19 +1,24 @@
 /**
  * collectd - src/plugin.c
- * Copyright (C) 2005-2011  Florian octo Forster
+ * Copyright (C) 2005-2014  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
- * Free Software Foundation; only version 2 of the License is applicable.
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
  *
- * This program is distributed in the hope that it will be useful, but
- * WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
- * General Public License for more details.
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
  *
- * You should have received a copy of the GNU General Public License along
- * with this program; if not, write to the Free Software Foundation, Inc.,
- * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
  *
  * Authors:
  *   Florian octo Forster <octo at collectd.org>
@@ -21,22 +26,23 @@
  **/
 
 #include "collectd.h"
-#include "utils_complain.h"
-
-#include <ltdl.h>
-
-#if HAVE_PTHREAD_H
-# include <pthread.h>
-#endif
-
 #include "common.h"
 #include "plugin.h"
 #include "configfile.h"
+#include "filter_chain.h"
 #include "utils_avltree.h"
+#include "utils_cache.h"
+#include "utils_complain.h"
 #include "utils_llist.h"
 #include "utils_heap.h"
-#include "utils_cache.h"
-#include "filter_chain.h"
+#include "utils_time.h"
+#include "utils_random.h"
+
+#if HAVE_PTHREAD_H
+# include <pthread.h>
+#endif
+
+#include <ltdl.h>
 
 /*
  * Private structures
@@ -61,17 +67,28 @@ struct read_func_s
 #define rf_ctx rf_super.cf_ctx
 	callback_func_t rf_super;
 	char rf_group[DATA_MAX_NAME_LEN];
-	char rf_name[DATA_MAX_NAME_LEN];
+	char *rf_name;
 	int rf_type;
-	struct timespec rf_interval;
-	struct timespec rf_effective_interval;
-	struct timespec rf_next_read;
+	cdtime_t rf_interval;
+	cdtime_t rf_effective_interval;
+	cdtime_t rf_next_read;
 };
 typedef struct read_func_s read_func_t;
 
+struct write_queue_s;
+typedef struct write_queue_s write_queue_t;
+struct write_queue_s
+{
+	value_list_t *vl;
+	plugin_ctx_t ctx;
+	write_queue_t *next;
+};
+
 /*
  * Private variables
  */
+static c_avl_tree_t *plugins_loaded = NULL;
+
 static llist_t *list_init;
 static llist_t *list_write;
 static llist_t *list_flush;
@@ -95,12 +112,26 @@ static pthread_cond_t  read_cond = PTHREAD_COND_INITIALIZER;
 static pthread_t      *read_threads = NULL;
 static int             read_threads_num = 0;
 
+static write_queue_t  *write_queue_head;
+static write_queue_t  *write_queue_tail;
+static long            write_queue_length = 0;
+static _Bool           write_loop = 1;
+static pthread_mutex_t write_lock = PTHREAD_MUTEX_INITIALIZER;
+static pthread_cond_t  write_cond = PTHREAD_COND_INITIALIZER;
+static pthread_t      *write_threads = NULL;
+static size_t          write_threads_num = 0;
+
 static pthread_key_t   plugin_ctx_key;
 static _Bool           plugin_ctx_key_initialized = 0;
 
+static long            write_limit_high = 0;
+static long            write_limit_low = 0;
+
 /*
  * Static functions
  */
+static int plugin_dispatch_values_internal (value_list_t *vl);
+
 static const char *plugin_get_dir (void)
 {
 	if (plugindir == NULL)
@@ -341,13 +372,6 @@ static int plugin_load_file (char *file, uint32_t flags)
 	return (0);
 }
 
-static _Bool timeout_reached(struct timespec timeout)
-{
-	struct timeval now;
-	gettimeofday(&now, NULL);
-	return (now.tv_sec >= timeout.tv_sec && now.tv_usec >= (timeout.tv_nsec / 1000));
-}
-
 static void *plugin_read_thread (void __attribute__((unused)) *args)
 {
 	while (read_loop != 0)
@@ -373,18 +397,15 @@ static void *plugin_read_thread (void __attribute__((unused)) *args)
 		}
 		pthread_mutex_unlock (&read_lock);
 
-		if ((rf->rf_interval.tv_sec == 0) && (rf->rf_interval.tv_nsec == 0))
+		if (rf->rf_interval == 0)
 		{
 			/* this should not happen, because the interval is set
 			 * for each plugin when loading it
 			 * XXX: issue a warning? */
-			now = cdtime ();
-
-			CDTIME_T_TO_TIMESPEC (plugin_get_interval (), &rf->rf_interval);
-
+			rf->rf_interval = plugin_get_interval ();
 			rf->rf_effective_interval = rf->rf_interval;
 
-			CDTIME_T_TO_TIMESPEC (now, &rf->rf_next_read);
+			rf->rf_next_read = cdtime ();
 		}
 
 		/* sleep until this entry is due,
@@ -396,11 +417,15 @@ static void *plugin_read_thread (void __attribute__((unused)) *args)
 		 * pthread_cond_timedwait returns. */
 		rc = 0;
 		while ((read_loop != 0)
-				&& !timeout_reached(rf->rf_next_read)
+				&& (cdtime () < rf->rf_next_read)
 				&& rc == 0)
 		{
+			struct timespec ts = { 0 };
+
+			CDTIME_T_TO_TIMESPEC (rf->rf_next_read, &ts);
+
 			rc = pthread_cond_timedwait (&read_cond, &read_lock,
-				&rf->rf_next_read);
+				&ts);
 		}
 
 		/* Must hold `read_lock' when accessing `rf->rf_type'. */
@@ -424,6 +449,7 @@ static void *plugin_read_thread (void __attribute__((unused)) *args)
 		{
 			DEBUG ("plugin_read_thread: Destroying the `%s' "
 					"callback.", rf->rf_name);
+			sfree (rf->rf_name);
 			destroy_callback ((callback_func_t *) rf);
 			rf = NULL;
 			continue;
@@ -456,20 +482,14 @@ static void *plugin_read_thread (void __attribute__((unused)) *args)
 		 * intervals in which it will be called. */
 		if (status != 0)
 		{
-			rf->rf_effective_interval.tv_sec *= 2;
-			rf->rf_effective_interval.tv_nsec *= 2;
-			NORMALIZE_TIMESPEC (rf->rf_effective_interval);
-
-			if (rf->rf_effective_interval.tv_sec >= 86400)
-			{
-				rf->rf_effective_interval.tv_sec = 86400;
-				rf->rf_effective_interval.tv_nsec = 0;
-			}
+			rf->rf_effective_interval *= 2;
+			if (rf->rf_effective_interval > TIME_T_TO_CDTIME_T (86400))
+				rf->rf_effective_interval = TIME_T_TO_CDTIME_T (86400);
 
 			NOTICE ("read-function of plugin `%s' failed. "
-					"Will suspend it for %i seconds.",
+					"Will suspend it for %.3f seconds.",
 					rf->rf_name,
-					(int) rf->rf_effective_interval.tv_sec);
+					CDTIME_T_TO_DOUBLE (rf->rf_effective_interval));
 		}
 		else
 		{
@@ -481,32 +501,26 @@ static void *plugin_read_thread (void __attribute__((unused)) *args)
 		now = cdtime ();
 
 		DEBUG ("plugin_read_thread: Effective interval of the "
-				"%s plugin is %i.%09i.",
+				"%s plugin is %.3f seconds.",
 				rf->rf_name,
-				(int) rf->rf_effective_interval.tv_sec,
-				(int) rf->rf_effective_interval.tv_nsec);
+				CDTIME_T_TO_DOUBLE (rf->rf_effective_interval));
 
 		/* Calculate the next (absolute) time at which this function
 		 * should be called. */
-		rf->rf_next_read.tv_sec = rf->rf_next_read.tv_sec
-			+ rf->rf_effective_interval.tv_sec;
-		rf->rf_next_read.tv_nsec = rf->rf_next_read.tv_nsec
-			+ rf->rf_effective_interval.tv_nsec;
-		NORMALIZE_TIMESPEC (rf->rf_next_read);
+		rf->rf_next_read += rf->rf_effective_interval;
 
 		/* Check, if `rf_next_read' is in the past. */
-		if (TIMESPEC_TO_CDTIME_T (&rf->rf_next_read) < now)
+		if (rf->rf_next_read < now)
 		{
 			/* `rf_next_read' is in the past. Insert `now'
 			 * so this value doesn't trail off into the
 			 * past too much. */
-			CDTIME_T_TO_TIMESPEC (now, &rf->rf_next_read);
+			rf->rf_next_read = now;
 		}
 
-		DEBUG ("plugin_read_thread: Next read of the %s plugin at %i.%09i.",
+		DEBUG ("plugin_read_thread: Next read of the %s plugin at %.3f.",
 				rf->rf_name,
-				(int) rf->rf_next_read.tv_sec,
-				(int) rf->rf_next_read.tv_nsec);
+				CDTIME_T_TO_DOUBLE (rf->rf_next_read));
 
 		/* Re-insert this read function into the heap again. */
 		c_heap_insert (read_heap, rf);
@@ -573,6 +587,250 @@ static void stop_read_threads (void)
 	read_threads_num = 0;
 } /* void stop_read_threads */
 
+static void plugin_value_list_free (value_list_t *vl) /* {{{ */
+{
+	if (vl == NULL)
+		return;
+
+	meta_data_destroy (vl->meta);
+	sfree (vl->values);
+	sfree (vl);
+} /* }}} void plugin_value_list_free */
+
+static value_list_t *plugin_value_list_clone (value_list_t const *vl_orig) /* {{{ */
+{
+	value_list_t *vl;
+
+	if (vl_orig == NULL)
+		return (NULL);
+
+	vl = malloc (sizeof (*vl));
+	if (vl == NULL)
+		return (NULL);
+	memcpy (vl, vl_orig, sizeof (*vl));
+
+	vl->values = calloc (vl_orig->values_len, sizeof (*vl->values));
+	if (vl->values == NULL)
+	{
+		plugin_value_list_free (vl);
+		return (NULL);
+	}
+	memcpy (vl->values, vl_orig->values,
+			vl_orig->values_len * sizeof (*vl->values));
+
+	vl->meta = meta_data_clone (vl->meta);
+	if ((vl_orig->meta != NULL) && (vl->meta == NULL))
+	{
+		plugin_value_list_free (vl);
+		return (NULL);
+	}
+
+	if (vl->time == 0)
+		vl->time = cdtime ();
+
+	/* Fill in the interval from the thread context, if it is zero. */
+	if (vl->interval == 0)
+	{
+		plugin_ctx_t ctx = plugin_get_ctx ();
+
+		if (ctx.interval != 0)
+			vl->interval = ctx.interval;
+		else
+		{
+			char name[6 * DATA_MAX_NAME_LEN];
+			FORMAT_VL (name, sizeof (name), vl);
+			ERROR ("plugin_value_list_clone: Unable to determine "
+					"interval from context for "
+					"value list \"%s\". "
+					"This indicates a broken plugin. "
+					"Please report this problem to the "
+					"collectd mailing list or at "
+					"<http://collectd.org/bugs/>.", name);
+			vl->interval = cf_get_default_interval ();
+		}
+	}
+
+	return (vl);
+} /* }}} value_list_t *plugin_value_list_clone */
+
+static int plugin_write_enqueue (value_list_t const *vl) /* {{{ */
+{
+	write_queue_t *q;
+
+	q = malloc (sizeof (*q));
+	if (q == NULL)
+		return (ENOMEM);
+	q->next = NULL;
+
+	q->vl = plugin_value_list_clone (vl);
+	if (q->vl == NULL)
+	{
+		sfree (q);
+		return (ENOMEM);
+	}
+
+	/* Store context of caller (read plugin); otherwise, it would not be
+	 * available to the write plugins when actually dispatching the
+	 * value-list later on. */
+	q->ctx = plugin_get_ctx ();
+
+	pthread_mutex_lock (&write_lock);
+
+	if (write_queue_tail == NULL)
+	{
+		write_queue_head = q;
+		write_queue_tail = q;
+		write_queue_length = 1;
+	}
+	else
+	{
+		write_queue_tail->next = q;
+		write_queue_tail = q;
+		write_queue_length += 1;
+	}
+
+	pthread_cond_signal (&write_cond);
+	pthread_mutex_unlock (&write_lock);
+
+	return (0);
+} /* }}} int plugin_write_enqueue */
+
+static value_list_t *plugin_write_dequeue (void) /* {{{ */
+{
+	write_queue_t *q;
+	value_list_t *vl;
+
+	pthread_mutex_lock (&write_lock);
+
+	while (write_loop && (write_queue_head == NULL))
+		pthread_cond_wait (&write_cond, &write_lock);
+
+	if (write_queue_head == NULL)
+	{
+		pthread_mutex_unlock (&write_lock);
+		return (NULL);
+	}
+
+	q = write_queue_head;
+	write_queue_head = q->next;
+	write_queue_length -= 1;
+	if (write_queue_head == NULL) {
+		write_queue_tail = NULL;
+		assert(0 == write_queue_length);
+		}
+
+	pthread_mutex_unlock (&write_lock);
+
+	(void) plugin_set_ctx (q->ctx);
+
+	vl = q->vl;
+	sfree (q);
+	return (vl);
+} /* }}} value_list_t *plugin_write_dequeue */
+
+static void *plugin_write_thread (void __attribute__((unused)) *args) /* {{{ */
+{
+	while (write_loop)
+	{
+		value_list_t *vl = plugin_write_dequeue ();
+		if (vl == NULL)
+			continue;
+
+		plugin_dispatch_values_internal (vl);
+
+		plugin_value_list_free (vl);
+	}
+
+	pthread_exit (NULL);
+	return ((void *) 0);
+} /* }}} void *plugin_write_thread */
+
+static void start_write_threads (size_t num) /* {{{ */
+{
+	size_t i;
+
+	if (write_threads != NULL)
+		return;
+
+	write_threads = (pthread_t *) calloc (num, sizeof (pthread_t));
+	if (write_threads == NULL)
+	{
+		ERROR ("plugin: start_write_threads: calloc failed.");
+		return;
+	}
+
+	write_threads_num = 0;
+	for (i = 0; i < num; i++)
+	{
+		int status;
+
+		status = pthread_create (write_threads + write_threads_num,
+				/* attr = */ NULL,
+				plugin_write_thread,
+				/* arg = */ NULL);
+		if (status != 0)
+		{
+			char errbuf[1024];
+			ERROR ("plugin: start_write_threads: pthread_create failed "
+					"with status %i (%s).", status,
+					sstrerror (status, errbuf, sizeof (errbuf)));
+			return;
+		}
+
+		write_threads_num++;
+	} /* for (i) */
+} /* }}} void start_write_threads */
+
+static void stop_write_threads (void) /* {{{ */
+{
+	write_queue_t *q;
+	int i;
+
+	if (write_threads == NULL)
+		return;
+
+	INFO ("collectd: Stopping %zu write threads.", write_threads_num);
+
+	pthread_mutex_lock (&write_lock);
+	write_loop = 0;
+	DEBUG ("plugin: stop_write_threads: Signalling `write_cond'");
+	pthread_cond_broadcast (&write_cond);
+	pthread_mutex_unlock (&write_lock);
+
+	for (i = 0; i < write_threads_num; i++)
+	{
+		if (pthread_join (write_threads[i], NULL) != 0)
+		{
+			ERROR ("plugin: stop_write_threads: pthread_join failed.");
+		}
+		write_threads[i] = (pthread_t) 0;
+	}
+	sfree (write_threads);
+	write_threads_num = 0;
+
+	pthread_mutex_lock (&write_lock);
+	i = 0;
+	for (q = write_queue_head; q != NULL; )
+	{
+		write_queue_t *q1 = q;
+		plugin_value_list_free (q->vl);
+		q = q->next;
+		sfree (q1);
+		i++;
+	}
+	write_queue_head = NULL;
+	write_queue_tail = NULL;
+	write_queue_length = 0;
+	pthread_mutex_unlock (&write_lock);
+
+	if (i > 0)
+	{
+		WARNING ("plugin: %i value list%s left after shutting down "
+				"the write threads.",
+				i, (i == 1) ? " was" : "s were");
+	}
+} /* }}} void stop_write_threads */
+
 /*
  * Public functions
  */
@@ -591,8 +849,52 @@ void plugin_set_dir (const char *dir)
 	}
 }
 
+static _Bool plugin_is_loaded (char const *name)
+{
+	int status;
+
+	if (plugins_loaded == NULL)
+		plugins_loaded = c_avl_create ((void *) strcasecmp);
+	assert (plugins_loaded != NULL);
+
+	status = c_avl_get (plugins_loaded, name, /* ret_value = */ NULL);
+	return (status == 0);
+}
+
+static int plugin_mark_loaded (char const *name)
+{
+	char *name_copy;
+	int status;
+
+	name_copy = strdup (name);
+	if (name_copy == NULL)
+		return (ENOMEM);
+
+	status = c_avl_insert (plugins_loaded,
+			/* key = */ name_copy, /* value = */ NULL);
+	return (status);
+}
+
+static void plugin_free_loaded ()
+{
+	void *key;
+	void *value;
+
+	if (plugins_loaded == NULL)
+		return;
+
+	while (c_avl_pick (plugins_loaded, &key, &value) == 0)
+	{
+		sfree (key);
+		assert (value == NULL);
+	}
+
+	c_avl_destroy (plugins_loaded);
+	plugins_loaded = NULL;
+}
+
 #define BUFSIZE 512
-int plugin_load (const char *type, uint32_t flags)
+int plugin_load (char const *plugin_name, uint32_t flags)
 {
 	DIR  *dh;
 	const char *dir;
@@ -604,17 +906,38 @@ int plugin_load (const char *type, uint32_t flags)
 	struct dirent *de;
 	int status;
 
-	DEBUG ("type = %s", type);
+	if (plugin_name == NULL)
+		return (EINVAL);
+
+	/* Check if plugin is already loaded and don't do anything in this
+	 * case. */
+	if (plugin_is_loaded (plugin_name))
+		return (0);
 
 	dir = plugin_get_dir ();
 	ret = 1;
 
+	/*
+	 * XXX: Magic at work:
+	 *
+	 * Some of the language bindings, for example the Python and Perl
+	 * plugins, need to be able to export symbols to the scripts they run.
+	 * For this to happen, the "Globals" flag needs to be set.
+	 * Unfortunately, this technical detail is hard to explain to the
+	 * average user and she shouldn't have to worry about this, ideally.
+	 * So in order to save everyone's sanity use a different default for a
+	 * handful of special plugins. --octo
+	 */
+	if ((strcasecmp ("perl", plugin_name) == 0)
+			|| (strcasecmp ("python", plugin_name) == 0))
+		flags |= PLUGIN_FLAGS_GLOBAL;
+
 	/* `cpu' should not match `cpufreq'. To solve this we add `.so' to the
 	 * type when matching the filename */
-	status = ssnprintf (typename, sizeof (typename), "%s.so", type);
+	status = ssnprintf (typename, sizeof (typename), "%s.so", plugin_name);
 	if ((status < 0) || ((size_t) status >= sizeof (typename)))
 	{
-		WARNING ("snprintf: truncated: `%s.so'", type);
+		WARNING ("plugin_load: Filename too long: \"%s.so\"", plugin_name);
 		return (-1);
 	}
 	typename_len = strlen (typename);
@@ -622,7 +945,7 @@ int plugin_load (const char *type, uint32_t flags)
 	if ((dh = opendir (dir)) == NULL)
 	{
 		char errbuf[1024];
-		ERROR ("opendir (%s): %s", dir,
+		ERROR ("plugin_load: opendir (%s) failed: %s", dir,
 				sstrerror (errno, errbuf, sizeof (errbuf)));
 		return (-1);
 	}
@@ -636,40 +959,47 @@ int plugin_load (const char *type, uint32_t flags)
 				"%s/%s", dir, de->d_name);
 		if ((status < 0) || ((size_t) status >= sizeof (filename)))
 		{
-			WARNING ("snprintf: truncated: `%s/%s'", dir, de->d_name);
+			WARNING ("plugin_load: Filename too long: \"%s/%s\"",
+					dir, de->d_name);
 			continue;
 		}
 
 		if (lstat (filename, &statbuf) == -1)
 		{
 			char errbuf[1024];
-			WARNING ("stat %s: %s", filename,
+			WARNING ("plugin_load: stat (\"%s\") failed: %s",
+					filename,
 					sstrerror (errno, errbuf, sizeof (errbuf)));
 			continue;
 		}
 		else if (!S_ISREG (statbuf.st_mode))
 		{
 			/* don't follow symlinks */
-			WARNING ("stat %s: not a regular file", filename);
+			WARNING ("plugin_load: %s is not a regular file.",
+				       	filename);
 			continue;
 		}
 
-		if (plugin_load_file (filename, flags) == 0)
+		status = plugin_load_file (filename, flags);
+		if (status == 0)
 		{
 			/* success */
+			plugin_mark_loaded (plugin_name);
 			ret = 0;
 			break;
 		}
 		else
 		{
-			fprintf (stderr, "Unable to load plugin %s.\n", type);
+			ERROR ("plugin_load: Load plugin \"%s\" failed with "
+					"status %i.", plugin_name, status);
 		}
 	}
 
 	closedir (dh);
 
-	if (filename[0] == '\0')
-		fprintf (stderr, "Could not find plugin %s.\n", type);
+	if (filename[0] == 0)
+		ERROR ("plugin_load: Could not find plugin \"%s\" in %s",
+				plugin_name, dir);
 
 	return (ret);
 }
@@ -706,13 +1036,9 @@ static int plugin_compare_read_func (const void *arg0, const void *arg1)
 	rf0 = arg0;
 	rf1 = arg1;
 
-	if (rf0->rf_next_read.tv_sec < rf1->rf_next_read.tv_sec)
-		return (-1);
-	else if (rf0->rf_next_read.tv_sec > rf1->rf_next_read.tv_sec)
-		return (1);
-	else if (rf0->rf_next_read.tv_nsec < rf1->rf_next_read.tv_nsec)
+	if (rf0->rf_next_read < rf1->rf_next_read)
 		return (-1);
-	else if (rf0->rf_next_read.tv_nsec > rf1->rf_next_read.tv_nsec)
+	else if (rf0->rf_next_read > rf1->rf_next_read)
 		return (1);
 	else
 		return (0);
@@ -726,6 +1052,9 @@ static int plugin_insert_read (read_func_t *rf)
 	int status;
 	llentry_t *le;
 
+	rf->rf_next_read = cdtime ();
+	rf->rf_effective_interval = rf->rf_interval;
+
 	pthread_mutex_lock (&read_lock);
 
 	if (read_list == NULL)
@@ -787,43 +1116,12 @@ static int plugin_insert_read (read_func_t *rf)
 	return (0);
 } /* int plugin_insert_read */
 
-static int read_cb_wrapper (user_data_t *ud)
-{
-	int (*callback) (void);
-
-	if (ud == NULL)
-		return -1;
-
-	callback = ud->data;
-	return callback();
-} /* int read_cb_wrapper */
-
 int plugin_register_read (const char *name,
 		int (*callback) (void))
 {
 	read_func_t *rf;
-	plugin_ctx_t ctx = plugin_get_ctx ();
 	int status;
 
-	if (ctx.interval != 0) {
-		/* If ctx.interval is not zero (== use the plugin or global
-		 * interval), we need to use the "complex" read callback,
-		 * because only that allows to specify a different interval.
-		 * Wrap the callback using read_cb_wrapper(). */
-		struct timespec interval;
-		user_data_t user_data;
-
-		user_data.data = callback;
-		user_data.free_func = NULL;
-
-		CDTIME_T_TO_TIMESPEC (ctx.interval, &interval);
-		return plugin_register_complex_read (/* group = */ NULL,
-				name, read_cb_wrapper, &interval, &user_data);
-	}
-
-	DEBUG ("plugin_register_read: default_interval = %.3f",
-			CDTIME_T_TO_DOUBLE(plugin_get_interval ()));
-
 	rf = malloc (sizeof (*rf));
 	if (rf == NULL)
 	{
@@ -835,13 +1133,11 @@ int plugin_register_read (const char *name,
 	rf->rf_callback = (void *) callback;
 	rf->rf_udata.data = NULL;
 	rf->rf_udata.free_func = NULL;
-	rf->rf_ctx = ctx;
+	rf->rf_ctx = plugin_get_ctx ();
 	rf->rf_group[0] = '\0';
-	sstrncpy (rf->rf_name, name, sizeof (rf->rf_name));
+	rf->rf_name = strdup (name);
 	rf->rf_type = RF_SIMPLE;
-	rf->rf_interval.tv_sec = 0;
-	rf->rf_interval.tv_nsec = 0;
-	rf->rf_effective_interval = rf->rf_interval;
+	rf->rf_interval = plugin_get_interval ();
 
 	status = plugin_insert_read (rf);
 	if (status != 0)
@@ -856,7 +1152,6 @@ int plugin_register_complex_read (const char *group, const char *name,
 		user_data_t *user_data)
 {
 	read_func_t *rf;
-	plugin_ctx_t ctx = plugin_get_ctx ();
 	int status;
 
 	rf = malloc (sizeof (*rf));
@@ -872,21 +1167,12 @@ int plugin_register_complex_read (const char *group, const char *name,
 		sstrncpy (rf->rf_group, group, sizeof (rf->rf_group));
 	else
 		rf->rf_group[0] = '\0';
-	sstrncpy (rf->rf_name, name, sizeof (rf->rf_name));
+	rf->rf_name = strdup (name);
 	rf->rf_type = RF_COMPLEX;
 	if (interval != NULL)
-	{
-		rf->rf_interval = *interval;
-	}
-	else if (ctx.interval != 0)
-	{
-		CDTIME_T_TO_TIMESPEC (ctx.interval, &rf->rf_interval);
-	}
-	rf->rf_effective_interval = rf->rf_interval;
-
-	DEBUG ("plugin_register_read: interval = %i.%09i",
-			(int) rf->rf_interval.tv_sec,
-			(int) rf->rf_interval.tv_nsec);
+		rf->rf_interval = TIMESPEC_TO_CDTIME_T (interval);
+	else
+		rf->rf_interval = plugin_get_interval ();
 
 	/* Set user data */
 	if (user_data == NULL)
@@ -899,7 +1185,7 @@ int plugin_register_complex_read (const char *group, const char *name,
 		rf->rf_udata = *user_data;
 	}
 
-	rf->rf_ctx = ctx;
+	rf->rf_ctx = plugin_get_ctx ();
 
 	status = plugin_insert_read (rf);
 	if (status != 0)
@@ -936,6 +1222,27 @@ int plugin_register_shutdown (const char *name,
 				(void *) callback, /* user_data = */ NULL));
 } /* int plugin_register_shutdown */
 
+static void plugin_free_data_sets (void)
+{
+	void *key;
+	void *value;
+
+	if (data_sets == NULL)
+		return;
+
+	while (c_avl_pick (data_sets, &key, &value) == 0)
+	{
+		data_set_t *ds = value;
+		/* key is a pointer to ds->type */
+
+		sfree (ds->ds);
+		sfree (ds);
+	}
+
+	c_avl_destroy (data_sets);
+	data_sets = NULL;
+} /* void plugin_free_data_sets */
+
 int plugin_register_data_set (const data_set_t *ds)
 {
 	data_set_t *ds_copy;
@@ -1153,7 +1460,8 @@ int plugin_unregister_notification (const char *name)
 
 void plugin_init_all (void)
 {
-	const char *chain_name;
+	char const *chain_name;
+	long write_threads_num;
 	llentry_t *le;
 	int status;
 
@@ -1166,6 +1474,37 @@ void plugin_init_all (void)
 	chain_name = global_option_get ("PostCacheChain");
 	post_cache_chain = fc_chain_get_by_name (chain_name);
 
+	write_limit_high = global_option_get_long ("WriteQueueLimitHigh",
+			/* default = */ 0);
+	if (write_limit_high < 0)
+	{
+		ERROR ("WriteQueueLimitHigh must be positive or zero.");
+		write_limit_high = 0;
+	}
+
+	write_limit_low = global_option_get_long ("WriteQueueLimitLow",
+			/* default = */ write_limit_high / 2);
+	if (write_limit_low < 0)
+	{
+		ERROR ("WriteQueueLimitLow must be positive or zero.");
+		write_limit_low = write_limit_high / 2;
+	}
+	else if (write_limit_low > write_limit_high)
+	{
+		ERROR ("WriteQueueLimitLow must not be larger than "
+				"WriteQueueLimitHigh.");
+		write_limit_low = write_limit_high;
+	}
+
+	write_threads_num = global_option_get_long ("WriteThreads",
+			/* default = */ 5);
+	if (write_threads_num < 1)
+	{
+		ERROR ("WriteThreads must be positive.");
+		write_threads_num = 5;
+	}
+
+	start_write_threads ((size_t) write_threads_num);
 
 	if ((list_init == NULL) && (read_heap == NULL))
 		return;
@@ -1435,6 +1774,8 @@ void plugin_shutdown_all (void)
 		plugin_set_ctx (old_ctx);
 	}
 
+	stop_write_threads ();
+
 	/* Write plugins which use the `user_data' pointer usually need the
 	 * same data available to the flush callback. If this is the case, set
 	 * the free_function to NULL when registering the flush callback and to
@@ -1447,6 +1788,9 @@ void plugin_shutdown_all (void)
 	destroy_all_callbacks (&list_notification);
 	destroy_all_callbacks (&list_shutdown);
 	destroy_all_callbacks (&list_log);
+
+	plugin_free_loaded ();
+	plugin_free_data_sets ();
 } /* void plugin_shutdown_all */
 
 int plugin_dispatch_missing (const value_list_t *vl) /* {{{ */
@@ -1490,7 +1834,7 @@ int plugin_dispatch_missing (const value_list_t *vl) /* {{{ */
   return (0);
 } /* int }}} plugin_dispatch_missing */
 
-int plugin_dispatch_values (value_list_t *vl)
+static int plugin_dispatch_values_internal (value_list_t *vl)
 {
 	int status;
 	static c_complain_t no_write_complaint = C_COMPLAIN_INIT_STATIC;
@@ -1541,29 +1885,10 @@ int plugin_dispatch_values (value_list_t *vl)
 		return (-1);
 	}
 
-	if (vl->time == 0)
-		vl->time = cdtime ();
-
-	if (vl->interval <= 0)
-	{
-		plugin_ctx_t ctx = plugin_get_ctx ();
-
-		if (ctx.interval != 0)
-			vl->interval = ctx.interval;
-		else
-		{
-			char name[6 * DATA_MAX_NAME_LEN];
-			FORMAT_VL (name, sizeof (name), vl);
-			ERROR ("plugin_dispatch_values: Unable to determine "
-					"interval from context for "
-					"value list \"%s\". "
-					"This indicates a broken plugin. "
-					"Please report this problem to the "
-					"collectd mailing list or at "
-					"<http://collectd.org/bugs/>.", name);
-			vl->interval = interval_g;
-		}
-	}
+	/* Assured by plugin_value_list_clone(). The time is determined at
+	 * _enqueue_ time. */
+	assert (vl->time != 0);
+	assert (vl->interval != 0);
 
 	DEBUG ("plugin_dispatch_values: time = %.3f; interval = %.3f; "
 			"host = %s; "
@@ -1684,53 +2009,149 @@ int plugin_dispatch_values (value_list_t *vl)
 	}
 
 	return (0);
-} /* int plugin_dispatch_values */
+} /* int plugin_dispatch_values_internal */
 
-int plugin_dispatch_values_secure (const value_list_t *vl)
+static double get_drop_probability (void) /* {{{ */
 {
-  value_list_t vl_copy;
-  int status;
+	long pos;
+	long size;
+	long wql;
 
-  if (vl == NULL)
-    return EINVAL;
+	pthread_mutex_lock (&write_lock);
+	wql = write_queue_length;
+	pthread_mutex_unlock (&write_lock);
 
-  memcpy (&vl_copy, vl, sizeof (vl_copy));
+	if (wql < write_limit_low)
+		return (0.0);
+	if (wql >= write_limit_high)
+		return (1.0);
 
-  /* Write callbacks must not change the values and meta pointers, so we can
-   * savely skip copying those and make this more efficient. */
-  if ((pre_cache_chain == NULL) && (post_cache_chain == NULL))
-    return (plugin_dispatch_values (&vl_copy));
+	pos = 1 + wql - write_limit_low;
+	size = 1 + write_limit_high - write_limit_low;
 
-  /* Set pointers to NULL, just to be on the save side. */
-  vl_copy.values = NULL;
-  vl_copy.meta = NULL;
+	return (((double) pos) / ((double) size));
+} /* }}} double get_drop_probability */
 
-  vl_copy.values = malloc (sizeof (*vl_copy.values) * vl->values_len);
-  if (vl_copy.values == NULL)
-  {
-    ERROR ("plugin_dispatch_values_secure: malloc failed.");
-    return (ENOMEM);
-  }
-  memcpy (vl_copy.values, vl->values, sizeof (*vl_copy.values) * vl->values_len);
+static _Bool check_drop_value (void) /* {{{ */
+{
+	static cdtime_t last_message_time = 0;
+	static pthread_mutex_t last_message_lock = PTHREAD_MUTEX_INITIALIZER;
 
-  if (vl->meta != NULL)
-  {
-    vl_copy.meta = meta_data_clone (vl->meta);
-    if (vl_copy.meta == NULL)
-    {
-      ERROR ("plugin_dispatch_values_secure: meta_data_clone failed.");
-      free (vl_copy.values);
-      return (ENOMEM);
-    }
-  } /* if (vl->meta) */
+	double p;
+	double q;
+	int status;
 
-  status = plugin_dispatch_values (&vl_copy);
+	if (write_limit_high == 0)
+		return (0);
 
-  meta_data_destroy (vl_copy.meta);
-  free (vl_copy.values);
+	p = get_drop_probability ();
+	if (p == 0.0)
+		return (0);
 
-  return (status);
-} /* int plugin_dispatch_values_secure */
+	status = pthread_mutex_trylock (&last_message_lock);
+	if (status == 0)
+	{
+		cdtime_t now;
+
+		now = cdtime ();
+		if ((now - last_message_time) > TIME_T_TO_CDTIME_T (1))
+		{
+			last_message_time = now;
+			ERROR ("plugin_dispatch_values: Low water mark "
+					"reached. Dropping %.0f%% of metrics.",
+					100.0 * p);
+		}
+		pthread_mutex_unlock (&last_message_lock);
+	}
+
+	if (p == 1.0)
+		return (1);
+
+	q = cdrand_d ();
+	if (q > p)
+		return (1);
+	else
+		return (0);
+} /* }}} _Bool check_drop_value */
+
+int plugin_dispatch_values (value_list_t const *vl)
+{
+	int status;
+
+	if (check_drop_value ())
+		return (0);
+
+	status = plugin_write_enqueue (vl);
+	if (status != 0)
+	{
+		char errbuf[1024];
+		ERROR ("plugin_dispatch_values: plugin_write_enqueue failed "
+				"with status %i (%s).", status,
+				sstrerror (status, errbuf, sizeof (errbuf)));
+		return (status);
+	}
+
+	return (0);
+}
+
+__attribute__((sentinel))
+int plugin_dispatch_multivalue (value_list_t const *template, /* {{{ */
+		_Bool store_percentage, ...)
+{
+	value_list_t *vl;
+	int failed = 0;
+	gauge_t sum = 0.0;
+	va_list ap;
+
+	assert (template->values_len == 1);
+
+	va_start (ap, store_percentage);
+	while (42)
+	{
+		char const *name;
+		gauge_t value;
+
+		name = va_arg (ap, char const *);
+		if (name == NULL)
+			break;
+
+		value = va_arg (ap, gauge_t);
+		if (!isnan (value))
+			sum += value;
+	}
+	va_end (ap);
+
+	vl = plugin_value_list_clone (template);
+	/* plugin_value_list_clone makes sure vl->time is set to non-zero. */
+	if (store_percentage)
+		sstrncpy (vl->type, "percent", sizeof (vl->type));
+
+	va_start (ap, store_percentage);
+	while (42)
+	{
+		char const *name;
+		int status;
+
+		/* Set the type instance. */
+		name = va_arg (ap, char const *);
+		if (name == NULL)
+			break;
+		sstrncpy (vl->type_instance, name, sizeof (vl->type_instance));
+
+		/* Set the value. */
+		vl->values[0].gauge = va_arg (ap, gauge_t);
+		if (store_percentage)
+			vl->values[0].gauge *= 100.0 / sum;
+
+		status = plugin_write_enqueue (vl);
+		if (status != 0)
+			failed++;
+	}
+	va_end (ap);
+
+	plugin_value_list_free (vl);
+	return (failed);
+} /* }}} int plugin_dispatch_multivalue */
 
 int plugin_dispatch_notification (const notification_t *notif)
 {
@@ -1812,10 +2233,54 @@ void plugin_log (int level, const char *format, ...)
 	}
 } /* void plugin_log */
 
+int parse_log_severity (const char *severity)
+{
+	int log_level = -1;
+
+	if ((0 == strcasecmp (severity, "emerg"))
+			|| (0 == strcasecmp (severity, "alert"))
+			|| (0 == strcasecmp (severity, "crit"))
+			|| (0 == strcasecmp (severity, "err")))
+		log_level = LOG_ERR;
+	else if (0 == strcasecmp (severity, "warning"))
+		log_level = LOG_WARNING;
+	else if (0 == strcasecmp (severity, "notice"))
+		log_level = LOG_NOTICE;
+	else if (0 == strcasecmp (severity, "info"))
+		log_level = LOG_INFO;
+#if COLLECT_DEBUG
+	else if (0 == strcasecmp (severity, "debug"))
+		log_level = LOG_DEBUG;
+#endif /* COLLECT_DEBUG */
+
+	return (log_level);
+} /* int parse_log_severity */
+
+int parse_notif_severity (const char *severity)
+{
+	int notif_severity = -1;
+
+	if (strcasecmp (severity, "FAILURE") == 0)
+		notif_severity = NOTIF_FAILURE;
+	else if (strcmp (severity, "OKAY") == 0)
+		notif_severity = NOTIF_OKAY;
+	else if ((strcmp (severity, "WARNING") == 0)
+			|| (strcmp (severity, "WARN") == 0))
+		notif_severity = NOTIF_WARNING;
+
+	return (notif_severity);
+} /* int parse_notif_severity */
+
 const data_set_t *plugin_get_ds (const char *name)
 {
 	data_set_t *ds;
 
+	if (data_sets == NULL)
+	{
+		ERROR ("plugin_get_ds: No data sets are defined yet.");
+		return (NULL);
+	}
+
 	if (c_avl_get (data_sets, name, (void *) &ds) != 0)
 	{
 		DEBUG ("No such dataset registered: %s", name);
@@ -2074,25 +2539,11 @@ cdtime_t plugin_get_interval (void)
 {
 	cdtime_t interval;
 
-	const char *interval_str;
-	double interval_dbl;
-
 	interval = plugin_get_ctx().interval;
 	if (interval > 0)
 		return interval;
 
-	/* this should happen during initialization only */
-	interval_str = global_option_get ("Interval");
-	if (interval_str != NULL)
-	{
-		interval_dbl = atof (interval_str);
-		if (interval_dbl > 0.0)
-			interval = DOUBLE_TO_CDTIME_T (interval_dbl);
-	}
-
-	if (interval > 0)
-		return interval;
-	return TIME_T_TO_CDTIME_T (10);
+	return cf_get_default_interval ();
 } /* cdtime_t plugin_get_interval */
 
 typedef struct {