Merge branch 'collectd-4.2'

[collectd.git] / src / network.c

diff --git a/src/network.c b/src/network.c
index a426c13..c347552 100644 (file)
--- a/src/network.c
+++ b/src/network.c
@@ -162,11 +162,11 @@ static int listen_loop = 0;
 static char         send_buffer[BUFF_SIZE];
 static char        *send_buffer_ptr;
 static int          send_buffer_fill;
-static value_list_t send_buffer_vl = VALUE_LIST_INIT;
+static value_list_t send_buffer_vl = VALUE_LIST_STATIC;
 static char         send_buffer_type[DATA_MAX_NAME_LEN];
 static pthread_mutex_t send_buffer_lock = PTHREAD_MUTEX_INITIALIZER;
 
-static avl_tree_t      *cache_tree = NULL;
+static c_avl_tree_t      *cache_tree = NULL;
 static pthread_mutex_t  cache_lock = PTHREAD_MUTEX_INITIALIZER;
 static time_t           cache_flush_last;
 static int              cache_flush_interval = 1800;
@@ -184,12 +184,12 @@ static int cache_flush (void)
 
        char   *key;
        time_t *value;
-       avl_iterator_t *iter;
+       c_avl_iterator_t *iter;
 
        time_t curtime = time (NULL);
 
-       iter = avl_get_iterator (cache_tree);
-       while (avl_iterator_next (iter, (void *) &key, (void *) &value) == 0)
+       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;
@@ -198,7 +198,7 @@ static int cache_flush (void)
                if (tmp == NULL)
                {
                        sfree (keys);
-                       avl_iterator_destroy (iter);
+                       c_avl_iterator_destroy (iter);
                        ERROR ("network plugin: cache_flush: realloc"
                                        " failed.");
                        return (-1);
@@ -206,15 +206,15 @@ static int cache_flush (void)
                keys = tmp;
                keys[keys_num] = key;
                keys_num++;
-       } /* while (avl_iterator_next) */
-       avl_iterator_destroy (iter);
+       } /* while (c_avl_iterator_next) */
+       c_avl_iterator_destroy (iter);
 
        for (i = 0; i < keys_num; i++)
        {
-               if (avl_remove (cache_tree, keys[i], (void *) &key,
+               if (c_avl_remove (cache_tree, keys[i], (void *) &key,
                                        (void *) &value) != 0)
                {
-                       WARNING ("network plugin: cache_flush: avl_remove"
+                       WARNING ("network plugin: cache_flush: c_avl_remove"
                                        " (%s) failed.", keys[i]);
                        continue;
                }
@@ -246,7 +246,7 @@ static int cache_check (const char *type, const value_list_t *vl)
 
        pthread_mutex_lock (&cache_lock);
 
-       if (avl_get (cache_tree, key, (void *) &value) == 0)
+       if (c_avl_get (cache_tree, key, (void *) &value) == 0)
        {
                if (*value < vl->time)
                {
@@ -267,7 +267,7 @@ static int cache_check (const char *type, const value_list_t *vl)
                if ((key_copy != NULL) && (value != NULL))
                {
                        *value = vl->time;
-                       avl_insert (cache_tree, key_copy, value);
+                       c_avl_insert (cache_tree, key_copy, value);
                        retval = 0;
                }
                else
@@ -385,7 +385,8 @@ static int parse_part_values (void **ret_buffer, int *ret_buffer_len,
 
        if (buffer_len < (15))
        {
-               DEBUG ("packet is too short: buffer_len = %i", buffer_len);
+               DEBUG ("network plugin: packet is too short: buffer_len = %i",
+                               buffer_len);
                return (-1);
        }
 
@@ -556,6 +557,13 @@ static int parse_packet (void *buffer, int buffer_len)
                        if (status == 0)
                                vl.time = (time_t) tmp;
                }
+               else if (ntohs (header->type) == TYPE_INTERVAL)
+               {
+                       uint64_t tmp = 0;
+                       status = parse_part_number (&buffer, &buffer_len, &tmp);
+                       if (status == 0)
+                               vl.interval = (int) tmp;
+               }
                else if (ntohs (header->type) == TYPE_HOST)
                {
                        status = parse_part_string (&buffer, &buffer_len,
@@ -674,6 +682,16 @@ static int network_set_ttl (const sockent_t *se, const struct addrinfo *ai)
 static int network_bind_socket (const sockent_t *se, const struct addrinfo *ai)
 {
        int loop = 0;
+       int yes  = 1;
+
+       /* allow multiple sockets to use the same PORT number */
+       if (setsockopt(se->fd, SOL_SOCKET, SO_REUSEADDR,
+                               &yes, sizeof(yes)) == -1) {
+                char errbuf[1024];
+                ERROR ("setsockopt: %s", 
+                                sstrerror (errno, errbuf, sizeof (errbuf)));
+               return (-1);
+       }
 
        DEBUG ("fd = %i; calling `bind'", se->fd);
 
@@ -981,7 +999,7 @@ static int network_add_sending_socket (const char *node, const char *service)
        return (0);
 } /* int network_get_listen_socket */
 
-int network_receive (void)
+static int network_receive (void)
 {
        char buffer[BUFF_SIZE];
        int  buffer_len;
@@ -1047,7 +1065,7 @@ static void network_send_buffer (const char *buffer, int buffer_len)
        sockent_t *se;
        int status;
 
-       DEBUG ("buffer_len = %i", buffer_len);
+       DEBUG ("network plugin: network_send_buffer: buffer_len = %i", buffer_len);
 
        for (se = sending_sockets; se != NULL; se = se->next)
        {
@@ -1083,7 +1101,6 @@ static int add_to_buffer (char *buffer, int buffer_size,
                                        vl->host, strlen (vl->host)) != 0)
                        return (-1);
                strcpy (vl_def->host, vl->host);
-               DEBUG ("network plugin: add_to_buffer: host = %s", vl->host);
        }
 
        if (vl_def->time != vl->time)
@@ -1092,8 +1109,14 @@ static int add_to_buffer (char *buffer, int buffer_size,
                                        (uint64_t) vl->time))
                        return (-1);
                vl_def->time = vl->time;
-               DEBUG ("network plugin: add_to_buffer: time = %u",
-                               (unsigned int) vl->time);
+       }
+
+       if (vl_def->interval != vl->interval)
+       {
+               if (write_part_number (&buffer, &buffer_size, TYPE_INTERVAL,
+                                       (uint64_t) vl->interval))
+                       return (-1);
+               vl_def->interval = vl->interval;
        }
 
        if (strcmp (vl_def->plugin, vl->plugin) != 0)
@@ -1102,8 +1125,6 @@ static int add_to_buffer (char *buffer, int buffer_size,
                                        vl->plugin, strlen (vl->plugin)) != 0)
                        return (-1);
                strcpy (vl_def->plugin, vl->plugin);
-               DEBUG ("network plugin: add_to_buffer: plugin = %s",
-                               vl->plugin);
        }
 
        if (strcmp (vl_def->plugin_instance, vl->plugin_instance) != 0)
@@ -1113,8 +1134,6 @@ static int add_to_buffer (char *buffer, int buffer_size,
                                        strlen (vl->plugin_instance)) != 0)
                        return (-1);
                strcpy (vl_def->plugin_instance, vl->plugin_instance);
-               DEBUG ("network plugin: add_to_buffer: plugin_instance = %s",
-                               vl->plugin_instance);
        }
 
        if (strcmp (type_def, ds->type) != 0)
@@ -1123,7 +1142,6 @@ static int add_to_buffer (char *buffer, int buffer_size,
                                        ds->type, strlen (ds->type)) != 0)
                        return (-1);
                strcpy (type_def, ds->type);
-               DEBUG ("network plugin: add_to_buffer: type = %s", ds->type);
        }
 
        if (strcmp (vl_def->type_instance, vl->type_instance) != 0)
@@ -1133,8 +1151,6 @@ static int add_to_buffer (char *buffer, int buffer_size,
                                        strlen (vl->type_instance)) != 0)
                        return (-1);
                strcpy (vl_def->type_instance, vl->type_instance);
-               DEBUG ("network plugin: add_to_buffer: type_instance = %s",
-                               vl->type_instance);
        }
        
        if (write_part_values (&buffer, &buffer_size, ds, vl) != 0)
@@ -1231,7 +1247,11 @@ static int network_config (const char *key, const char *val)
                                && (fields_num != 2))
                        return (1);
                else if (fields_num == 2)
+               {
+                       if ((service = strchr (fields[1], '.')) != NULL)
+                               *service = '\0';
                        service = fields[1];
+               }
                node = fields[0];
 
                if (strcasecmp ("Listen", key) == 0)
@@ -1272,8 +1292,6 @@ static int network_config (const char *key, const char *val)
 
 static int network_shutdown (void)
 {
-       DEBUG ("Shutting down.");
-
        listen_loop++;
 
        if (listen_thread != (pthread_t) 0)
@@ -1283,19 +1301,20 @@ static int network_shutdown (void)
                listen_thread = (pthread_t) 0;
        }
 
-       listen_thread = 0;
+       if (send_buffer_fill > 0)
+               flush_buffer ();
 
        if (cache_tree != NULL)
        {
                void *key;
                void *value;
 
-               while (avl_pick (cache_tree, &key, &value) == 0)
+               while (c_avl_pick (cache_tree, &key, &value) == 0)
                {
                        sfree (key);
                        sfree (value);
                }
-               avl_destroy (cache_tree);
+               c_avl_destroy (cache_tree);
                cache_tree = NULL;
        }
 
@@ -1318,7 +1337,7 @@ static int network_init (void)
        memset (&send_buffer_vl, '\0', sizeof (send_buffer_vl));
        memset (send_buffer_type, '\0', sizeof (send_buffer_type));
 
-       cache_tree = avl_create ((int (*) (const void *, const void *)) strcmp);
+       cache_tree = c_avl_create ((int (*) (const void *, const void *)) strcmp);
        cache_flush_last = time (NULL);
 
        /* setup socket(s) and so on */