X-Git-Url: https://git.octo.it/?p=collectd.git;a=blobdiff_plain;f=src%2Fnetwork.c;h=95f54641ae6e96af855494190272472b67458cd2;hp=a00955133fb3e951cd318f9597acb3ccfca13b86;hb=61a1fa91ba73e4fe3a34949f77c5f017056f2b7a;hpb=7ec4487eaf46b35d3860d177c65cae9cf9ab2b89

diff --git a/src/network.c b/src/network.c
index a0095513..95f54641 100644
--- a/src/network.c
+++ b/src/network.c
@@ -1,6 +1,7 @@
 /**
  * collectd - src/network.c
  * Copyright (C) 2005-2009  Florian octo Forster
+ * Copyright (C) 2009       Aman Gupta
  *
  * This program is free software; you can redistribute it and/or modify it
  * under the terms of the GNU Lesser General Public License as published by
@@ -18,6 +19,7 @@
  *
  * Authors:
  *   Florian octo Forster <octo at verplant.org>
+ *   Aman Gupta <aman at tmm1.net>
  **/
 
 #define _BSD_SOURCE /* For struct ip_mreq */
@@ -28,6 +30,8 @@
 #include "configfile.h"
 #include "utils_fbhash.h"
 #include "utils_avltree.h"
+#include "utils_cache.h"
+#include "utils_complain.h"
 
 #include "network.h"
 
@@ -49,15 +53,15 @@
 #if HAVE_POLL_H
 # include <poll.h>
 #endif
+#if HAVE_NET_IF_H
+# include <net/if.h>
+#endif
 
 #if HAVE_LIBGCRYPT
 # include <gcrypt.h>
 GCRY_THREAD_OPTION_PTHREAD_IMPL;
 #endif
 
-/* 1500 - 40 - 8  =  Ethernet packet - IPv6 header - UDP header */
-/* #define BUFF_SIZE 1452 */
-
 #ifndef IPV6_ADD_MEMBERSHIP
 # ifdef IPV6_JOIN_GROUP
 #  define IPV6_ADD_MEMBERSHIP IPV6_JOIN_GROUP
@@ -66,9 +70,6 @@ GCRY_THREAD_OPTION_PTHREAD_IMPL;
 # endif
 #endif /* !IP_ADD_MEMBERSHIP */
 
-/* Buffer size to allocate. */
-#define BUFF_SIZE 1024
-
 /*
  * Maximum size required for encryption / signing:
  *
@@ -121,6 +122,7 @@ typedef struct sockent
 
 	char *node;
 	char *service;
+	int interface;
 
 	union
 	{
@@ -246,7 +248,7 @@ typedef struct part_encryption_aes256_s part_encryption_aes256_t;
 
 struct receive_list_entry_s
 {
-  char data[BUFF_SIZE];
+  char *data;
   int  data_len;
   int  fd;
   struct receive_list_entry_s *next;
@@ -257,7 +259,9 @@ typedef struct receive_list_entry_s receive_list_entry_t;
  * Private variables
  */
 static int network_config_ttl = 0;
+static size_t network_config_packet_size = 1024;
 static int network_config_forward = 0;
+static int network_config_stats = 0;
 
 static sockent_t *sending_sockets = NULL;
 
@@ -265,6 +269,7 @@ static receive_list_entry_t *receive_list_head = NULL;
 static receive_list_entry_t *receive_list_tail = NULL;
 static pthread_mutex_t       receive_list_lock = PTHREAD_MUTEX_INITIALIZER;
 static pthread_cond_t        receive_list_cond = PTHREAD_COND_INITIALIZER;
+static uint64_t              receive_list_length = 0;
 
 static sockent_t     *listen_sockets = NULL;
 static struct pollfd *listen_sockets_pollfd = NULL;
@@ -279,133 +284,135 @@ static int       dispatch_thread_running = 0;
 static pthread_t dispatch_thread_id;
 
 /* Buffer in which to-be-sent network packets are constructed. */
-static char             send_buffer[BUFF_SIZE];
+static char            *send_buffer;
 static char            *send_buffer_ptr;
 static int              send_buffer_fill;
 static value_list_t     send_buffer_vl = VALUE_LIST_STATIC;
 static pthread_mutex_t  send_buffer_lock = PTHREAD_MUTEX_INITIALIZER;
 
-/* In this cache we store all the values we received, so we can send out only
- * those values which were *not* received via the network plugin, too. This is
- * used for the `Forward false' option. */
-static c_avl_tree_t    *cache_tree = NULL;
-static pthread_mutex_t  cache_lock = PTHREAD_MUTEX_INITIALIZER;
-static time_t           cache_flush_last = 0;
-static int              cache_flush_interval = 1800;
+/* XXX: These counters are incremented from one place only. The spot in which
+ * the values are incremented is either only reachable by one thread (the
+ * dispatch thread, for example) or locked by some lock (send_buffer_lock for
+ * example). Only if neither is true, the stats_lock is acquired. The counters
+ * are always read without holding a lock in the hope that writing 8 bytes to
+ * memory is an atomic operation. */
+static uint64_t stats_octets_rx  = 0;
+static uint64_t stats_octets_tx  = 0;
+static uint64_t stats_packets_rx = 0;
+static uint64_t stats_packets_tx = 0;
+static uint64_t stats_values_dispatched = 0;
+static uint64_t stats_values_not_dispatched = 0;
+static uint64_t stats_values_sent = 0;
+static uint64_t stats_values_not_sent = 0;
+static pthread_mutex_t stats_lock = PTHREAD_MUTEX_INITIALIZER;
 
 /*
  * Private functions
  */
-static int cache_flush (void)
+static _Bool check_receive_okay (const value_list_t *vl) /* {{{ */
 {
-	char **keys = NULL;
-	int    keys_num = 0;
+  uint64_t time_sent = 0;
+  int status;
 
-	char **tmp;
-	int    i;
+  status = uc_meta_data_get_unsigned_int (vl,
+      "network:time_sent", &time_sent);
 
-	char   *key;
-	time_t *value;
-	c_avl_iterator_t *iter;
+  /* This is a value we already sent. Don't allow it to be received again in
+   * order to avoid looping. */
+  if ((status == 0) && (time_sent >= ((uint64_t) vl->time)))
+    return (false);
 
-	time_t curtime = time (NULL);
+  return (true);
+} /* }}} _Bool check_receive_okay */
 
-	iter = c_avl_get_iterator (cache_tree);
-	while (c_avl_iterator_next (iter, (void *) &key, (void *) &value) == 0)
-	{
-		if ((curtime - *value) <= cache_flush_interval)
-			continue;
-		tmp = (char **) realloc (keys,
-				(keys_num + 1) * sizeof (char *));
-		if (tmp == NULL)
-		{
-			sfree (keys);
-			c_avl_iterator_destroy (iter);
-			ERROR ("network plugin: cache_flush: realloc"
-					" failed.");
-			return (-1);
-		}
-		keys = tmp;
-		keys[keys_num] = key;
-		keys_num++;
-	} /* while (c_avl_iterator_next) */
-	c_avl_iterator_destroy (iter);
+static _Bool check_send_okay (const value_list_t *vl) /* {{{ */
+{
+  _Bool received = false;
+  int status;
 
-	for (i = 0; i < keys_num; i++)
-	{
-		if (c_avl_remove (cache_tree, keys[i], (void *) &key,
-					(void *) &value) != 0)
-		{
-			WARNING ("network plugin: cache_flush: c_avl_remove"
-					" (%s) failed.", keys[i]);
-			continue;
-		}
+  if (network_config_forward != 0)
+    return (true);
 
-		sfree (key);
-		sfree (value);
-	}
+  if (vl->meta == NULL)
+    return (true);
 
-	sfree (keys);
+  status = meta_data_get_boolean (vl->meta, "network:received", &received);
+  if (status == -ENOENT)
+    return (true);
+  else if (status != 0)
+  {
+    ERROR ("network plugin: check_send_okay: meta_data_get_boolean failed "
+	"with status %i.", status);
+    return (true);
+  }
 
-	DEBUG ("network plugin: cache_flush: Removed %i %s",
-			keys_num, (keys_num == 1) ? "entry" : "entries");
-	cache_flush_last = curtime;
-	return (0);
-} /* int cache_flush */
+  /* By default, only *send* value lists that were not *received* by the
+   * network plugin. */
+  return (!received);
+} /* }}} _Bool check_send_okay */
 
-static int cache_check (const value_list_t *vl)
+static int network_dispatch_values (value_list_t *vl, /* {{{ */
+    const char *username)
 {
-	char key[1024];
-	time_t *value = NULL;
-	int retval = -1;
+  int status;
 
-	if (cache_tree == NULL)
-		return (-1);
+  if ((vl->time <= 0)
+      || (strlen (vl->host) <= 0)
+      || (strlen (vl->plugin) <= 0)
+      || (strlen (vl->type) <= 0))
+    return (-EINVAL);
 
-	if (format_name (key, sizeof (key), vl->host, vl->plugin,
-				vl->plugin_instance, vl->type, vl->type_instance))
-		return (-1);
+  if (!check_receive_okay (vl))
+  {
+#if COLLECT_DEBUG
+    char name[6*DATA_MAX_NAME_LEN];
+    FORMAT_VL (name, sizeof (name), vl);
+    name[sizeof (name) - 1] = 0;
+    DEBUG ("network plugin: network_dispatch_values: "
+	"NOT dispatching %s.", name);
+#endif
+    stats_values_not_dispatched++;
+    return (0);
+  }
 
-	pthread_mutex_lock (&cache_lock);
+  assert (vl->meta == NULL);
 
-	if (c_avl_get (cache_tree, key, (void *) &value) == 0)
-	{
-		if (*value < vl->time)
-		{
-			*value = vl->time;
-			retval = 0;
-		}
-		else
-		{
-			DEBUG ("network plugin: cache_check: *value = %i >= vl->time = %i",
-					(int) *value, (int) vl->time);
-			retval = 1;
-		}
-	}
-	else
-	{
-		char *key_copy = strdup (key);
-		value = malloc (sizeof (time_t));
-		if ((key_copy != NULL) && (value != NULL))
-		{
-			*value = vl->time;
-			c_avl_insert (cache_tree, key_copy, value);
-			retval = 0;
-		}
-		else
-		{
-			sfree (key_copy);
-			sfree (value);
-		}
-	}
+  vl->meta = meta_data_create ();
+  if (vl->meta == NULL)
+  {
+    ERROR ("network plugin: meta_data_create failed.");
+    return (-ENOMEM);
+  }
 
-	if ((time (NULL) - cache_flush_last) > cache_flush_interval)
-		cache_flush ();
+  status = meta_data_add_boolean (vl->meta, "network:received", true);
+  if (status != 0)
+  {
+    ERROR ("network plugin: meta_data_add_boolean failed.");
+    meta_data_destroy (vl->meta);
+    vl->meta = NULL;
+    return (status);
+  }
 
-	pthread_mutex_unlock (&cache_lock);
+  if (username != NULL)
+  {
+    status = meta_data_add_string (vl->meta, "network:username", username);
+    if (status != 0)
+    {
+      ERROR ("network plugin: meta_data_add_string failed.");
+      meta_data_destroy (vl->meta);
+      vl->meta = NULL;
+      return (status);
+    }
+  }
 
-	return (retval);
-} /* int cache_check */
+  plugin_dispatch_values (vl);
+  stats_values_dispatched++;
+
+  meta_data_destroy (vl->meta);
+  vl->meta = NULL;
+
+  return (0);
+} /* }}} int network_dispatch_values */
 
 #if HAVE_LIBGCRYPT
 static gcry_cipher_hd_t network_get_aes256_cypher (sockent_t *se, /* {{{ */
@@ -529,17 +536,34 @@ static int write_part_values (char **ret_buffer, int *ret_buffer_len,
 
 	for (i = 0; i < num_values; i++)
 	{
-		if (ds->ds[i].type == DS_TYPE_COUNTER)
-		{
-			pkg_values_types[i] = DS_TYPE_COUNTER;
-			pkg_values[i].counter = htonll (vl->values[i].counter);
-		}
-		else
+		pkg_values_types[i] = (uint8_t) ds->ds[i].type;
+		switch (ds->ds[i].type)
 		{
-			pkg_values_types[i] = DS_TYPE_GAUGE;
-			pkg_values[i].gauge = htond (vl->values[i].gauge);
-		}
-	}
+			case DS_TYPE_COUNTER:
+				pkg_values[i].counter = htonll (vl->values[i].counter);
+				break;
+
+			case DS_TYPE_GAUGE:
+				pkg_values[i].gauge = htond (vl->values[i].gauge);
+				break;
+
+			case DS_TYPE_DERIVE:
+				pkg_values[i].derive = htonll (vl->values[i].derive);
+				break;
+
+			case DS_TYPE_ABSOLUTE:
+				pkg_values[i].absolute = htonll (vl->values[i].absolute);
+				break;
+
+			default:
+				free (pkg_values_types);
+				free (pkg_values);
+				ERROR ("network plugin: write_part_values: "
+						"Unknown data source type: %i",
+						ds->ds[i].type);
+				return (-1);
+		} /* switch (ds->ds[i].type) */
+	} /* for (num_values) */
 
 	/*
 	 * Use `memcpy' to write everything to the buffer, because the pointer
@@ -715,10 +739,32 @@ static int parse_part_values (void **ret_buffer, size_t *ret_buffer_len,
 
 	for (i = 0; i < pkg_numval; i++)
 	{
-		if (pkg_types[i] == DS_TYPE_COUNTER)
-			pkg_values[i].counter = ntohll (pkg_values[i].counter);
-		else if (pkg_types[i] == DS_TYPE_GAUGE)
-			pkg_values[i].gauge = ntohd (pkg_values[i].gauge);
+		switch (pkg_types[i])
+		{
+		  case DS_TYPE_COUNTER:
+		    pkg_values[i].counter = (counter_t) ntohll (pkg_values[i].counter);
+		    break;
+
+		  case DS_TYPE_GAUGE:
+		    pkg_values[i].gauge = (gauge_t) ntohd (pkg_values[i].gauge);
+		    break;
+
+		  case DS_TYPE_DERIVE:
+		    pkg_values[i].derive = (derive_t) ntohll (pkg_values[i].derive);
+		    break;
+
+		  case DS_TYPE_ABSOLUTE:
+		    pkg_values[i].absolute = (absolute_t) ntohll (pkg_values[i].absolute);
+		    break;
+
+		  default:
+		    NOTICE ("network plugin: parse_part_values: "
+			"Don't know how to handle data source type %"PRIu8,
+			pkg_types[i]);
+		    sfree (pkg_types);
+		    sfree (pkg_values);
+		    return (-1);
+		} /* switch (pkg_types[i]) */
 	}
 
 	*ret_buffer     = buffer;
@@ -742,7 +788,6 @@ static int parse_part_number (void **ret_buffer, size_t *ret_buffer_len,
 	size_t exp_size = 2 * sizeof (uint16_t) + sizeof (uint64_t);
 
 	uint16_t pkg_length;
-	uint16_t pkg_type;
 
 	if ((buffer_len < 0) || ((size_t) buffer_len < exp_size))
 	{
@@ -756,7 +801,7 @@ static int parse_part_number (void **ret_buffer, size_t *ret_buffer_len,
 
 	memcpy ((void *) &tmp16, buffer, sizeof (tmp16));
 	buffer += sizeof (tmp16);
-	pkg_type = ntohs (tmp16);
+	/* pkg_type = ntohs (tmp16); */
 
 	memcpy ((void *) &tmp16, buffer, sizeof (tmp16));
 	buffer += sizeof (tmp16);
@@ -782,7 +827,6 @@ static int parse_part_string (void **ret_buffer, size_t *ret_buffer_len,
 	size_t header_size = 2 * sizeof (uint16_t);
 
 	uint16_t pkg_length;
-	uint16_t pkg_type;
 
 	if ((buffer_len < 0) || (buffer_len < header_size))
 	{
@@ -796,7 +840,7 @@ static int parse_part_string (void **ret_buffer, size_t *ret_buffer_len,
 
 	memcpy ((void *) &tmp16, buffer, sizeof (tmp16));
 	buffer += sizeof (tmp16);
-	pkg_type = ntohs (tmp16);
+	/* pkg_type = ntohs (tmp16); */
 
 	memcpy ((void *) &tmp16, buffer, sizeof (tmp16));
 	buffer += sizeof (tmp16);
@@ -860,7 +904,8 @@ static int parse_part_string (void **ret_buffer, size_t *ret_buffer_len,
 #define PP_SIGNED    0x01
 #define PP_ENCRYPTED 0x02
 static int parse_packet (sockent_t *se,
-		void *buffer, size_t buffer_size, int flags);
+		void *buffer, size_t buffer_size, int flags,
+		const char *username);
 
 #define BUFFER_READ(p,s) do { \
   memcpy ((p), buffer + buffer_offset, (s)); \
@@ -871,6 +916,8 @@ static int parse_packet (sockent_t *se,
 static int parse_part_sign_sha256 (sockent_t *se, /* {{{ */
     void **ret_buffer, size_t *ret_buffer_len, int flags)
 {
+  static c_complain_t complain_no_users = C_COMPLAIN_INIT_STATIC;
+
   char *buffer;
   size_t buffer_len;
   size_t buffer_offset;
@@ -892,8 +939,9 @@ static int parse_part_sign_sha256 (sockent_t *se, /* {{{ */
 
   if (se->data.server.userdb == NULL)
   {
-    NOTICE ("network plugin: Received signed network packet but can't verify "
-        "it because no user DB has been configured. Will accept it.");
+    c_complain (LOG_NOTICE, &complain_no_users,
+        "network plugin: Received signed network packet but can't verify it "
+        "because no user DB has been configured. Will accept it.");
     return (0);
   }
 
@@ -955,6 +1003,8 @@ static int parse_part_sign_sha256 (sockent_t *se, /* {{{ */
   {
     ERROR ("network plugin: gcry_md_setkey failed: %s", gcry_strerror (err));
     gcry_md_close (hd);
+    sfree (secret);
+    sfree (pss.username);
     return (-1);
   }
 
@@ -976,9 +1026,6 @@ static int parse_part_sign_sha256 (sockent_t *se, /* {{{ */
   gcry_md_close (hd);
   hd = NULL;
 
-  sfree (secret);
-  sfree (pss.username);
-
   if (memcmp (pss.hash, hash, sizeof (pss.hash)) != 0)
   {
     WARNING ("network plugin: Verifying HMAC-SHA-256 signature failed: "
@@ -987,9 +1034,12 @@ static int parse_part_sign_sha256 (sockent_t *se, /* {{{ */
   else
   {
     parse_packet (se, buffer + buffer_offset, buffer_len - buffer_offset,
-        flags | PP_SIGNED);
+        flags | PP_SIGNED, pss.username);
   }
 
+  sfree (secret);
+  sfree (pss.username);
+
   *ret_buffer = buffer + buffer_len;
   *ret_buffer_len = 0;
 
@@ -1033,7 +1083,8 @@ static int parse_part_sign_sha256 (sockent_t *se, /* {{{ */
     warning_has_been_printed = 1;
   }
 
-  parse_packet (se, buffer + part_len, buffer_size - part_len, flags);
+  parse_packet (se, buffer + part_len, buffer_size - part_len, flags,
+      /* username = */ NULL);
 
   *ret_buffer = buffer + buffer_size;
   *ret_buffer_size = 0;
@@ -1112,7 +1163,10 @@ static int parse_part_encr_aes256 (sockent_t *se, /* {{{ */
   cypher = network_get_aes256_cypher (se, pea.iv, sizeof (pea.iv),
       pea.username);
   if (cypher == NULL)
+  {
+    sfree (pea.username);
     return (-1);
+  }
 
   payload_len = part_size - (PART_ENCRYPTION_AES256_SIZE + username_len);
   assert (payload_len > 0);
@@ -1124,6 +1178,7 @@ static int parse_part_encr_aes256 (sockent_t *se, /* {{{ */
       /* in = */ NULL, /* in len = */ 0);
   if (err != 0)
   {
+    sfree (pea.username);
     ERROR ("network plugin: gcry_cipher_decrypt returned: %s",
         gcry_strerror (err));
     return (-1);
@@ -1142,17 +1197,22 @@ static int parse_part_encr_aes256 (sockent_t *se, /* {{{ */
       buffer + buffer_offset, payload_len);
   if (memcmp (hash, pea.hash, sizeof (hash)) != 0)
   {
+    sfree (pea.username);
     ERROR ("network plugin: Decryption failed: Checksum mismatch.");
     return (-1);
   }
 
   parse_packet (se, buffer + buffer_offset, payload_len,
-      flags | PP_ENCRYPTED);
+      flags | PP_ENCRYPTED, pea.username);
+
+  /* XXX: Free pea.username?!? */
 
   /* Update return values */
   *ret_buffer =     buffer     + part_size;
   *ret_buffer_len = buffer_len - part_size;
 
+  sfree (pea.username);
+
   return (0);
 } /* }}} int parse_part_encr_aes256 */
 /* #endif HAVE_LIBGCRYPT */
@@ -1207,7 +1267,8 @@ static int parse_part_encr_aes256 (sockent_t *se, /* {{{ */
 #undef BUFFER_READ
 
 static int parse_packet (sockent_t *se, /* {{{ */
-		void *buffer, size_t buffer_size, int flags)
+		void *buffer, size_t buffer_size, int flags,
+		const char *username)
 {
 	int status;
 
@@ -1304,23 +1365,10 @@ static int parse_packet (sockent_t *se, /* {{{ */
 		{
 			status = parse_part_values (&buffer, &buffer_size,
 					&vl.values, &vl.values_len);
-
 			if (status != 0)
 				break;
 
-			if ((vl.time > 0)
-					&& (strlen (vl.host) > 0)
-					&& (strlen (vl.plugin) > 0)
-					&& (strlen (vl.type) > 0)
-					&& (cache_check (&vl) == 0))
-			{
-				plugin_dispatch_values (&vl);
-			}
-			else
-			{
-				DEBUG ("network plugin: parse_packet:"
-						" NOT dispatching values");
-			}
+			network_dispatch_values (&vl, username);
 
 			sfree (vl.values);
 		}
@@ -1435,6 +1483,10 @@ static int parse_packet (sockent_t *se, /* {{{ */
 		}
 	} /* while (buffer_size > sizeof (part_header_t)) */
 
+	if (status == 0 && buffer_size > 0)
+		WARNING ("network plugin: parse_packet: Received truncated "
+				"packet, try increasing `MaxPacketSize'");
+
 	return (status);
 } /* }}} int parse_packet */
 
@@ -1530,7 +1582,7 @@ static int network_set_ttl (const sockent_t *se, const struct addrinfo *ai)
 
 		if (setsockopt (se->data.client.fd, IPPROTO_IP, optname,
 					&network_config_ttl,
-					sizeof (network_config_ttl)) == -1)
+					sizeof (network_config_ttl)) != 0)
 		{
 			char errbuf[1024];
 			ERROR ("setsockopt: %s",
@@ -1551,7 +1603,7 @@ static int network_set_ttl (const sockent_t *se, const struct addrinfo *ai)
 
 		if (setsockopt (se->data.client.fd, IPPROTO_IPV6, optname,
 					&network_config_ttl,
-					sizeof (network_config_ttl)) == -1)
+					sizeof (network_config_ttl)) != 0)
 		{
 			char errbuf[1024];
 			ERROR ("setsockopt: %s",
@@ -1564,7 +1616,110 @@ static int network_set_ttl (const sockent_t *se, const struct addrinfo *ai)
 	return (0);
 } /* int network_set_ttl */
 
-static int network_bind_socket (int fd, const struct addrinfo *ai)
+static int network_set_interface (const sockent_t *se, const struct addrinfo *ai) /* {{{ */
+{
+	DEBUG ("network plugin: network_set_interface: interface index = %i;",
+			se->interface);
+
+        assert (se->type == SOCKENT_TYPE_CLIENT);
+
+	if (ai->ai_family == AF_INET)
+	{
+		struct sockaddr_in *addr = (struct sockaddr_in *) ai->ai_addr;
+
+		if (IN_MULTICAST (ntohl (addr->sin_addr.s_addr)))
+		{
+#if HAVE_STRUCT_IP_MREQN_IMR_IFINDEX
+			/* If possible, use the "ip_mreqn" structure which has
+			 * an "interface index" member. Using the interface
+			 * index is preferred here, because of its similarity
+			 * to the way IPv6 handles this. Unfortunately, it
+			 * appears not to be portable. */
+			struct ip_mreqn mreq;
+
+			memset (&mreq, 0, sizeof (mreq));
+			mreq.imr_multiaddr.s_addr = addr->sin_addr.s_addr;
+			mreq.imr_address.s_addr = ntohl (INADDR_ANY);
+			mreq.imr_ifindex = se->interface;
+#else
+			struct ip_mreq mreq;
+
+			memset (&mreq, 0, sizeof (mreq));
+			mreq.imr_multiaddr.s_addr = addr->sin_addr.s_addr;
+			mreq.imr_interface.s_addr = ntohl (INADDR_ANY);
+#endif
+
+			if (setsockopt (se->data.client.fd, IPPROTO_IP, IP_MULTICAST_IF,
+						&mreq, sizeof (mreq)) != 0)
+			{
+				char errbuf[1024];
+				ERROR ("setsockopt: %s",
+						sstrerror (errno, errbuf, sizeof (errbuf)));
+				return (-1);
+			}
+
+			return (0);
+		}
+	}
+	else if (ai->ai_family == AF_INET6)
+	{
+		struct sockaddr_in6 *addr = (struct sockaddr_in6 *) ai->ai_addr;
+
+		if (IN6_IS_ADDR_MULTICAST (&addr->sin6_addr))
+		{
+			if (setsockopt (se->data.client.fd, IPPROTO_IPV6, IPV6_MULTICAST_IF,
+						&se->interface,
+						sizeof (se->interface)) != 0)
+			{
+				char errbuf[1024];
+				ERROR ("setsockopt: %s",
+						sstrerror (errno, errbuf,
+							sizeof (errbuf)));
+				return (-1);
+			}
+
+			return (0);
+		}
+	}
+
+	/* else: Not a multicast interface. */
+#if defined(HAVE_IF_INDEXTONAME) && HAVE_IF_INDEXTONAME && defined(SO_BINDTODEVICE)
+	if (se->interface != 0)
+	{
+		char interface_name[IFNAMSIZ];
+
+		if (if_indextoname (se->interface, interface_name) == NULL)
+			return (-1);
+
+		DEBUG ("network plugin: Binding socket to interface %s", interface_name);
+
+		if (setsockopt (se->data.client.fd, SOL_SOCKET, SO_BINDTODEVICE,
+					interface_name,
+					sizeof(interface_name)) == -1 )
+		{
+			char errbuf[1024];
+			ERROR ("setsockopt: %s",
+					sstrerror (errno, errbuf, sizeof (errbuf)));
+			return (-1);
+		}
+	}
+/* #endif HAVE_IF_INDEXTONAME && SO_BINDTODEVICE */
+
+#else
+	WARNING ("network plugin: Cannot set the interface on a unicast "
+			"socket because "
+# if !defined(SO_BINDTODEVICE)
+			"the the \"SO_BINDTODEVICE\" socket option "
+# else
+			"the \"if_indextoname\" function "
+# endif
+			"is not available on your system.");
+#endif
+
+	return (0);
+} /* }}} network_set_interface */
+
+static int network_bind_socket (int fd, const struct addrinfo *ai, const int interface_idx)
 {
 	int loop = 0;
 	int yes  = 1;
@@ -1593,12 +1748,24 @@ static int network_bind_socket (int fd, const struct addrinfo *ai)
 		struct sockaddr_in *addr = (struct sockaddr_in *) ai->ai_addr;
 		if (IN_MULTICAST (ntohl (addr->sin_addr.s_addr)))
 		{
+#if HAVE_STRUCT_IP_MREQN_IMR_IFINDEX
+			struct ip_mreqn mreq;
+#else
 			struct ip_mreq mreq;
+#endif
 
 			DEBUG ("fd = %i; IPv4 multicast address found", fd);
 
 			mreq.imr_multiaddr.s_addr = addr->sin_addr.s_addr;
-			mreq.imr_interface.s_addr = htonl (INADDR_ANY);
+#if HAVE_STRUCT_IP_MREQN_IMR_IFINDEX
+			/* Set the interface using the interface index if
+			 * possible (available). Unfortunately, the struct
+			 * ip_mreqn is not portable. */
+			mreq.imr_address.s_addr = ntohl (INADDR_ANY);
+			mreq.imr_ifindex = interface_idx;
+#else
+			mreq.imr_interface.s_addr = ntohl (INADDR_ANY);
+#endif
 
 			if (setsockopt (fd, IPPROTO_IP, IP_MULTICAST_LOOP,
 						&loop, sizeof (loop)) == -1)
@@ -1619,6 +1786,8 @@ static int network_bind_socket (int fd, const struct addrinfo *ai)
 							sizeof (errbuf)));
 				return (-1);
 			}
+
+			return (0);
 		}
 	}
 	else if (ai->ai_family == AF_INET6)
@@ -1644,7 +1813,7 @@ static int network_bind_socket (int fd, const struct addrinfo *ai)
 			 * single interface; programs running on
 			 * multihomed hosts may need to join the same
 			 * group on more than one interface.*/
-			mreq.ipv6mr_interface = 0;
+			mreq.ipv6mr_interface = interface_idx;
 
 			if (setsockopt (fd, IPPROTO_IPV6, IPV6_MULTICAST_LOOP,
 						&loop, sizeof (loop)) == -1)
@@ -1665,9 +1834,36 @@ static int network_bind_socket (int fd, const struct addrinfo *ai)
 							sizeof (errbuf)));
 				return (-1);
 			}
+
+			return (0);
 		}
 	}
 
+#if defined(HAVE_IF_INDEXTONAME) && HAVE_IF_INDEXTONAME && defined(SO_BINDTODEVICE)
+	/* if a specific interface was set, bind the socket to it. But to avoid
+ 	 * possible problems with multicast routing, only do that for non-multicast
+	 * addresses */
+	if (interface_idx != 0)
+	{
+		char interface_name[IFNAMSIZ];
+
+		if (if_indextoname (interface_idx, interface_name) == NULL)
+			return (-1);
+
+		DEBUG ("fd = %i; Binding socket to interface %s", fd, interface_name);
+
+		if (setsockopt (fd, SOL_SOCKET, SO_BINDTODEVICE,
+					interface_name,
+					sizeof(interface_name)) == -1 )
+		{
+			char errbuf[1024];
+			ERROR ("setsockopt: %s",
+					sstrerror (errno, errbuf, sizeof (errbuf)));
+			return (-1);
+		}
+	}
+#endif /* HAVE_IF_INDEXTONAME && SO_BINDTODEVICE */
+
 	return (0);
 } /* int network_bind_socket */
 
@@ -1683,6 +1879,7 @@ static int sockent_init (sockent_t *se, int type) /* {{{ */
 	se->type = SOCKENT_TYPE_CLIENT;
 	se->node = NULL;
 	se->service = NULL;
+	se->interface = 0;
 	se->next = NULL;
 
 	if (type == SOCKENT_TYPE_SERVER)
@@ -1831,7 +2028,7 @@ static int sockent_open (sockent_t *se) /* {{{ */
 				continue;
 			}
 
-			status = network_bind_socket (*tmp, ai_ptr);
+			status = network_bind_socket (*tmp, ai_ptr, se->interface);
 			if (status != 0)
 			{
 				close (*tmp);
@@ -1871,6 +2068,7 @@ static int sockent_open (sockent_t *se) /* {{{ */
 			se->data.client.addrlen = ai_ptr->ai_addrlen;
 
 			network_set_ttl (se, ai_ptr);
+			network_set_interface (se, ai_ptr);
 
 			/* We don't open more than one write-socket per
 			 * node/service pair.. */
@@ -1963,13 +2161,14 @@ static void *dispatch_thread (void __attribute__((unused)) *arg) /* {{{ */
     /* Lock and wait for more data to come in */
     pthread_mutex_lock (&receive_list_lock);
     while ((listen_loop == 0)
-	&& (receive_list_head == NULL))
+        && (receive_list_head == NULL))
       pthread_cond_wait (&receive_list_cond, &receive_list_lock);
 
     /* Remove the head entry and unlock */
     ent = receive_list_head;
     if (ent != NULL)
       receive_list_head = ent->next;
+    receive_list_length--;
     pthread_mutex_unlock (&receive_list_lock);
 
     /* Check whether we are supposed to exit. We do NOT check `listen_loop'
@@ -1995,14 +2194,17 @@ static void *dispatch_thread (void __attribute__((unused)) *arg) /* {{{ */
 
     if (se == NULL)
     {
-	    ERROR ("network plugin: Got packet from FD %i, but can't "
-			    "find an appropriate socket entry.",
-			    ent->fd);
-	    sfree (ent);
-	    continue;
+      ERROR ("network plugin: Got packet from FD %i, but can't "
+          "find an appropriate socket entry.",
+          ent->fd);
+      sfree (ent->data);
+      sfree (ent);
+      continue;
     }
 
-    parse_packet (se, ent->data, ent->data_len, /* flags = */ 0);
+    parse_packet (se, ent->data, ent->data_len, /* flags = */ 0,
+	/* username = */ NULL);
+    sfree (ent->data);
     sfree (ent);
   } /* while (42) */
 
@@ -2011,7 +2213,7 @@ static void *dispatch_thread (void __attribute__((unused)) *arg) /* {{{ */
 
 static int network_receive (void) /* {{{ */
 {
-	char buffer[BUFF_SIZE];
+	char buffer[network_config_packet_size];
 	int  buffer_len;
 
 	int i;
@@ -2019,11 +2221,13 @@ static int network_receive (void) /* {{{ */
 
 	receive_list_entry_t *private_list_head;
 	receive_list_entry_t *private_list_tail;
+	uint64_t              private_list_length;
 
         assert (listen_sockets_num > 0);
 
 	private_list_head = NULL;
 	private_list_tail = NULL;
+	private_list_length = 0;
 
 	while (listen_loop == 0)
 	{
@@ -2060,6 +2264,9 @@ static int network_receive (void) /* {{{ */
 				return (-1);
 			}
 
+			stats_octets_rx += ((uint64_t) buffer_len);
+			stats_packets_rx++;
+
 			/* TODO: Possible performance enhancement: Do not free
 			 * these entries in the dispatch thread but put them in
 			 * another list, so we don't have to allocate more and
@@ -2071,14 +2278,16 @@ static int network_receive (void) /* {{{ */
 				return (-1);
 			}
 			memset (ent, 0, sizeof (receive_list_entry_t));
+			ent->data = malloc (network_config_packet_size);
+			if (ent->data == NULL)
+			{
+				sfree (ent);
+				ERROR ("network plugin: malloc failed.");
+				return (-1);
+			}
 			ent->fd = listen_sockets_pollfd[i].fd;
 			ent->next = NULL;
 
-			/* Hopefully this be optimized out by the compiler. It
-			 * might help prevent stupid bugs in the future though.
-			 */
-			assert (sizeof (ent->data) == sizeof (buffer));
-
 			memcpy (ent->data, buffer, buffer_len);
 			ent->data_len = buffer_len;
 
@@ -2087,22 +2296,28 @@ static int network_receive (void) /* {{{ */
 			else
 				private_list_tail->next = ent;
 			private_list_tail = ent;
+			private_list_length++;
 
 			/* Do not block here. Blocking here has led to
 			 * insufficient performance in the past. */
 			if (pthread_mutex_trylock (&receive_list_lock) == 0)
 			{
+				assert (((receive_list_head == NULL) && (receive_list_length == 0))
+						|| ((receive_list_head != NULL) && (receive_list_length != 0)));
+
 				if (receive_list_head == NULL)
 					receive_list_head = private_list_head;
 				else
 					receive_list_tail->next = private_list_head;
 				receive_list_tail = private_list_tail;
-
-				private_list_head = NULL;
-				private_list_tail = NULL;
+				receive_list_length += private_list_length;
 
 				pthread_cond_signal (&receive_list_cond);
 				pthread_mutex_unlock (&receive_list_lock);
+
+				private_list_head = NULL;
+				private_list_tail = NULL;
+				private_list_length = 0;
 			}
 		} /* for (listen_sockets_pollfd) */
 	} /* while (listen_loop == 0) */
@@ -2117,9 +2332,11 @@ static int network_receive (void) /* {{{ */
 		else
 			receive_list_tail->next = private_list_head;
 		receive_list_tail = private_list_tail;
+		receive_list_length += private_list_length;
 
 		private_list_head = NULL;
 		private_list_tail = NULL;
+		private_list_length = 0;
 
 		pthread_cond_signal (&receive_list_cond);
 		pthread_mutex_unlock (&receive_list_lock);
@@ -2135,7 +2352,7 @@ static void *receive_thread (void __attribute__((unused)) *arg)
 
 static void network_init_buffer (void)
 {
-	memset (send_buffer, 0, sizeof (send_buffer));
+	memset (send_buffer, 0, network_config_packet_size);
 	send_buffer_ptr = send_buffer;
 	send_buffer_fill = 0;
 
@@ -2428,6 +2645,10 @@ static void flush_buffer (void)
 			send_buffer_fill);
 
 	network_send_buffer (send_buffer, (size_t) send_buffer_fill);
+
+	stats_octets_tx += ((uint64_t) send_buffer_fill);
+	stats_packets_tx++;
+
 	network_init_buffer ();
 }
 
@@ -2436,18 +2657,30 @@ static int network_write (const data_set_t *ds, const value_list_t *vl,
 {
 	int status;
 
-	/* If the value is already in the cache, we have received it via the
-	 * network. We write it again if forwarding is activated. It's then in
-	 * the cache and should we receive it again we will ignore it. */
-	status = cache_check (vl);
-	if ((network_config_forward == 0)
-			&& (status != 0))
-		return (0);
+	if (!check_send_okay (vl))
+	{
+#if COLLECT_DEBUG
+	  char name[6*DATA_MAX_NAME_LEN];
+	  FORMAT_VL (name, sizeof (name), vl);
+	  name[sizeof (name) - 1] = 0;
+	  DEBUG ("network plugin: network_write: "
+	      "NOT sending %s.", name);
+#endif
+	  /* Counter is not protected by another lock and may be reached by
+	   * multiple threads */
+	  pthread_mutex_lock (&stats_lock);
+	  stats_values_not_sent++;
+	  pthread_mutex_unlock (&stats_lock);
+	  return (0);
+	}
+
+	uc_meta_data_add_unsigned_int (vl,
+	    "network:time_sent", (uint64_t) vl->time);
 
 	pthread_mutex_lock (&send_buffer_lock);
 
 	status = add_to_buffer (send_buffer_ptr,
-			sizeof (send_buffer) - (send_buffer_fill + BUFF_SIG_SIZE),
+			network_config_packet_size - (send_buffer_fill + BUFF_SIG_SIZE),
 			&send_buffer_vl,
 			ds, vl);
 	if (status >= 0)
@@ -2455,13 +2688,15 @@ static int network_write (const data_set_t *ds, const value_list_t *vl,
 		/* status == bytes added to the buffer */
 		send_buffer_fill += status;
 		send_buffer_ptr  += status;
+
+		stats_values_sent++;
 	}
 	else
 	{
 		flush_buffer ();
 
 		status = add_to_buffer (send_buffer_ptr,
-				sizeof (send_buffer) - (send_buffer_fill + BUFF_SIG_SIZE),
+				network_config_packet_size - (send_buffer_fill + BUFF_SIG_SIZE),
 				&send_buffer_vl,
 				ds, vl);
 
@@ -2469,6 +2704,8 @@ static int network_write (const data_set_t *ds, const value_list_t *vl,
 		{
 			send_buffer_fill += status;
 			send_buffer_ptr  += status;
+
+			stats_values_sent++;
 		}
 	}
 
@@ -2477,7 +2714,7 @@ static int network_write (const data_set_t *ds, const value_list_t *vl,
 		ERROR ("network plugin: Unable to append to the "
 				"buffer for some weird reason");
 	}
-	else if ((sizeof (send_buffer) - send_buffer_fill) < 15)
+	else if ((network_config_packet_size - send_buffer_fill) < 15)
 	{
 		flush_buffer ();
 	}
@@ -2510,13 +2747,9 @@ static int network_config_set_boolean (const oconfig_item_t *ci, /* {{{ */
   {
     char *str = ci->values[0].value.string;
 
-    if ((strcasecmp ("true", str) == 0)
-        || (strcasecmp ("yes", str) == 0)
-        || (strcasecmp ("on", str) == 0))
+    if (IS_TRUE (str))
       *retval = 1;
-    else if ((strcasecmp ("false", str) == 0)
-        || (strcasecmp ("no", str) == 0)
-        || (strcasecmp ("off", str) == 0))
+    else if (IS_FALSE (str))
       *retval = 0;
     else
     {
@@ -2548,6 +2781,43 @@ static int network_config_set_ttl (const oconfig_item_t *ci) /* {{{ */
   return (0);
 } /* }}} int network_config_set_ttl */
 
+static int network_config_set_interface (const oconfig_item_t *ci, /* {{{ */
+    int *interface)
+{
+  if ((ci->values_num != 1)
+      || (ci->values[0].type != OCONFIG_TYPE_STRING))
+  {
+    WARNING ("network plugin: The `Interface' config option needs exactly "
+        "one string argument.");
+    return (-1);
+  }
+
+  if (interface == NULL)
+    return (-1);
+
+  *interface = if_nametoindex (ci->values[0].value.string);
+
+  return (0);
+} /* }}} int network_config_set_interface */
+
+static int network_config_set_buffer_size (const oconfig_item_t *ci) /* {{{ */
+{
+  int tmp;
+  if ((ci->values_num != 1)
+      || (ci->values[0].type != OCONFIG_TYPE_NUMBER))
+  {
+    WARNING ("network plugin: The `MaxPacketSize' config option needs exactly "
+        "one numeric argument.");
+    return (-1);
+  }
+
+  tmp = (int) ci->values[0].value.number;
+  if ((tmp >= 1024) && (tmp <= 65535))
+    network_config_packet_size = tmp;
+
+  return (0);
+} /* }}} int network_config_set_buffer_size */
+
 #if HAVE_LIBGCRYPT
 static int network_config_set_string (const oconfig_item_t *ci, /* {{{ */
     char **ret_string)
@@ -2641,6 +2911,10 @@ static int network_config_add_listen (const oconfig_item_t *ci) /* {{{ */
           &se->data.server.security_level);
     else
 #endif /* HAVE_LIBGCRYPT */
+    if (strcasecmp ("Interface", child->key) == 0)
+      network_config_set_interface (child,
+          &se->interface);
+    else
     {
       WARNING ("network plugin: Option `%s' is not allowed here.",
           child->key);
@@ -2719,6 +2993,10 @@ static int network_config_add_server (const oconfig_item_t *ci) /* {{{ */
           &se->data.client.security_level);
     else
 #endif /* HAVE_LIBGCRYPT */
+    if (strcasecmp ("Interface", child->key) == 0)
+      network_config_set_interface (child,
+          &se->interface);
+    else
     {
       WARNING ("network plugin: Option `%s' is not allowed here.",
           child->key);
@@ -2757,24 +3035,6 @@ static int network_config_add_server (const oconfig_item_t *ci) /* {{{ */
   return (0);
 } /* }}} int network_config_add_server */
 
-static int network_config_set_cache_flush (const oconfig_item_t *ci) /* {{{ */
-{
-  int tmp;
-  if ((ci->values_num != 1)
-      || (ci->values[0].type != OCONFIG_TYPE_NUMBER))
-  {
-    WARNING ("network plugin: The `CacheFlush' config option needs exactly "
-        "one numeric argument.");
-    return (-1);
-  }
-
-  tmp = (int) ci->values[0].value.number;
-  if (tmp > 0)
-    cache_flush_interval = tmp;
-
-  return (0);
-} /* }}} int network_config_set_cache_flush */
-
 static int network_config (oconfig_item_t *ci) /* {{{ */
 {
   int i;
@@ -2789,10 +3049,14 @@ static int network_config (oconfig_item_t *ci) /* {{{ */
       network_config_add_server (child);
     else if (strcasecmp ("TimeToLive", child->key) == 0)
       network_config_set_ttl (child);
+    else if (strcasecmp ("MaxPacketSize", child->key) == 0)
+      network_config_set_buffer_size (child);
     else if (strcasecmp ("Forward", child->key) == 0)
       network_config_set_boolean (child, &network_config_forward);
+    else if (strcasecmp ("ReportStats", child->key) == 0)
+      network_config_set_boolean (child, &network_config_stats);
     else if (strcasecmp ("CacheFlush", child->key) == 0)
-      network_config_set_cache_flush (child);
+      /* no op for backwards compatibility only */;
     else
     {
       WARNING ("network plugin: Option `%s' is not allowed here.",
@@ -2806,7 +3070,7 @@ static int network_config (oconfig_item_t *ci) /* {{{ */
 static int network_notification (const notification_t *n,
 		user_data_t __attribute__((unused)) *user_data)
 {
-  char  buffer[BUFF_SIZE];
+  char  buffer[network_config_packet_size];
   char *buffer_ptr = buffer;
   int   buffer_free = sizeof (buffer);
   int   status;
@@ -2905,19 +3169,7 @@ static int network_shutdown (void)
 	if (send_buffer_fill > 0)
 		flush_buffer ();
 
-	if (cache_tree != NULL)
-	{
-		void *key;
-		void *value;
-
-		while (c_avl_pick (cache_tree, &key, &value) == 0)
-		{
-			sfree (key);
-			sfree (value);
-		}
-		c_avl_destroy (cache_tree);
-		cache_tree = NULL;
-	}
+	sfree (send_buffer);
 
 	/* TODO: Close `sending_sockets' */
 
@@ -2926,18 +3178,95 @@ static int network_shutdown (void)
 	plugin_unregister_write ("network");
 	plugin_unregister_shutdown ("network");
 
-	/* Let the init function do it's move again ;) */
-	cache_flush_last = 0;
-
 	return (0);
 } /* int network_shutdown */
 
+static int network_stats_read (void) /* {{{ */
+{
+	uint64_t copy_octets_rx;
+	uint64_t copy_octets_tx;
+	uint64_t copy_packets_rx;
+	uint64_t copy_packets_tx;
+	uint64_t copy_values_dispatched;
+	uint64_t copy_values_not_dispatched;
+	uint64_t copy_values_sent;
+	uint64_t copy_values_not_sent;
+	uint64_t copy_receive_list_length;
+	value_list_t vl = VALUE_LIST_INIT;
+	value_t values[2];
+
+	copy_octets_rx = stats_octets_rx;
+	copy_octets_tx = stats_octets_tx;
+	copy_packets_rx = stats_packets_rx;
+	copy_packets_tx = stats_packets_tx;
+	copy_values_dispatched = stats_values_dispatched;
+	copy_values_not_dispatched = stats_values_not_dispatched;
+	copy_values_sent = stats_values_sent;
+	copy_values_not_sent = stats_values_not_sent;
+	copy_receive_list_length = receive_list_length;
+
+	/* Initialize `vl' */
+	vl.values = values;
+	vl.values_len = 2;
+	vl.time = 0;
+	vl.interval = interval_g;
+	sstrncpy (vl.host, hostname_g, sizeof (vl.host));
+	sstrncpy (vl.plugin, "network", sizeof (vl.plugin));
+
+	/* Octets received / sent */
+	vl.values[0].counter = (counter_t) copy_octets_rx;
+	vl.values[1].counter = (counter_t) copy_octets_tx;
+	sstrncpy (vl.type, "if_octets", sizeof (vl.type));
+	plugin_dispatch_values (&vl);
+
+	/* Packets received / send */
+	vl.values[0].counter = (counter_t) copy_packets_rx;
+	vl.values[1].counter = (counter_t) copy_packets_tx;
+	sstrncpy (vl.type, "if_packets", sizeof (vl.type));
+	plugin_dispatch_values (&vl);
+
+	/* Values (not) dispatched and (not) send */
+	sstrncpy (vl.type, "total_values", sizeof (vl.type));
+	vl.values_len = 1;
+
+	vl.values[0].derive = (derive_t) copy_values_dispatched;
+	sstrncpy (vl.type_instance, "dispatch-accepted",
+			sizeof (vl.type_instance));
+	plugin_dispatch_values (&vl);
+
+	vl.values[0].derive = (derive_t) copy_values_not_dispatched;
+	sstrncpy (vl.type_instance, "dispatch-rejected",
+			sizeof (vl.type_instance));
+	plugin_dispatch_values (&vl);
+
+	vl.values[0].derive = (derive_t) copy_values_sent;
+	sstrncpy (vl.type_instance, "send-accepted",
+			sizeof (vl.type_instance));
+	plugin_dispatch_values (&vl);
+
+	vl.values[0].derive = (derive_t) copy_values_not_sent;
+	sstrncpy (vl.type_instance, "send-rejected",
+			sizeof (vl.type_instance));
+	plugin_dispatch_values (&vl);
+
+	/* Receive queue length */
+	vl.values[0].gauge = (gauge_t) copy_receive_list_length;
+	sstrncpy (vl.type, "queue_length", sizeof (vl.type));
+	vl.type_instance[0] = 0;
+	plugin_dispatch_values (&vl);
+
+	return (0);
+} /* }}} int network_stats_read */
+
 static int network_init (void)
 {
+	static _Bool have_init = false;
+
 	/* Check if we were already initialized. If so, just return - there's
 	 * nothing more to do (for now, that is). */
-	if (cache_flush_last != 0)
+	if (have_init)
 		return (0);
+	have_init = true;
 
 #if HAVE_LIBGCRYPT
 	gcry_control (GCRYCTL_SET_THREAD_CBS, &gcry_threads_pthread);
@@ -2945,13 +3274,19 @@ static int network_init (void)
 	gcry_control (GCRYCTL_INITIALIZATION_FINISHED, 0);
 #endif
 
+	if (network_config_stats != 0)
+		plugin_register_read ("network", network_stats_read);
+
 	plugin_register_shutdown ("network", network_shutdown);
 
+	send_buffer = malloc (network_config_packet_size);
+	if (send_buffer == NULL)
+	{
+		ERROR ("network plugin: malloc failed.");
+		return (-1);
+	}
 	network_init_buffer ();
 
-	cache_tree = c_avl_create ((int (*) (const void *, const void *)) strcmp);
-	cache_flush_last = time (NULL);
-
 	/* setup socket(s) and so on */
 	if (sending_sockets != NULL)
 	{
@@ -3023,11 +3358,8 @@ static int network_flush (int timeout,
 {
 	pthread_mutex_lock (&send_buffer_lock);
 
-	if (((time (NULL) - cache_flush_last) >= timeout)
-			&& (send_buffer_fill > 0))
-	{
-		flush_buffer ();
-	}
+	if (send_buffer_fill > 0)
+	  flush_buffer ();
 
 	pthread_mutex_unlock (&send_buffer_lock);