X-Git-Url: https://git.tokkee.org/?a=blobdiff_plain;f=src%2Fnetwork.c;h=a659189f3e7960d312d0d608f9f8af55d1e3c5bb;hb=acbd25344a59c091b087311804d375b759077e32;hp=96d48f8bbcf790b7bba7c5bd9713479c93c5a7eb;hpb=5e75ddfba7eb16dd035478d5267d6a86a13d8055;p=collectd.git diff --git a/src/network.c b/src/network.c index 96d48f8b..a659189f 100644 --- a/src/network.c +++ b/src/network.c @@ -141,6 +141,7 @@ typedef struct part_values_s part_values_t; */ static const char *config_keys[] = { + "CacheFlush", "Listen", "Server", "TimeToLive", @@ -161,16 +162,75 @@ 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 pthread_mutex_t cache_lock = PTHREAD_MUTEX_INITIALIZER; +static time_t cache_flush_last; +static int cache_flush_interval = 1800; /* * Private functions */ +static int cache_flush (void) +{ + char **keys = NULL; + int keys_num = 0; + + char **tmp; + int i; + + char *key; + time_t *value; + 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) + { + if ((curtime - *value) <= cache_flush_interval) + continue; + tmp = (char **) realloc (keys, + (keys_num + 1) * sizeof (char *)); + if (tmp == NULL) + { + sfree (keys); + avl_iterator_destroy (iter); + ERROR ("network plugin: cache_flush: realloc" + " failed."); + return (-1); + } + keys = tmp; + keys[keys_num] = key; + keys_num++; + } /* while (avl_iterator_next) */ + avl_iterator_destroy (iter); + + for (i = 0; i < keys_num; i++) + { + if (avl_remove (cache_tree, keys[i], (void *) &key, + (void *) &value) != 0) + { + WARNING ("network plugin: cache_flush: avl_remove" + " (%s) failed.", keys[i]); + continue; + } + + sfree (key); + sfree (value); + } + + sfree (keys); + + DEBUG ("network plugin: cache_flush: Removed %i %s", + keys_num, (keys_num == 1) ? "entry" : "entries"); + cache_flush_last = curtime; + return (0); +} /* int cache_flush */ + static int cache_check (const char *type, const value_list_t *vl) { char key[1024]; @@ -217,7 +277,8 @@ static int cache_check (const char *type, const value_list_t *vl) } } - /* TODO: Flush cache */ + if ((time (NULL) - cache_flush_last) > cache_flush_interval) + cache_flush (); pthread_mutex_unlock (&cache_lock); @@ -324,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); } @@ -495,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, @@ -613,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); @@ -986,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) { @@ -1022,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) @@ -1031,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) @@ -1041,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) @@ -1052,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) @@ -1062,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) @@ -1072,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) @@ -1170,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) @@ -1195,17 +1276,22 @@ static int network_config (const char *key, const char *val) else network_config_forward = 0; } + else if (strcasecmp ("CacheFlush", key) == 0) + { + int tmp = atoi (val); + if (tmp > 0) + cache_flush_interval = tmp; + else return (1); + } else { return (-1); } return (0); -} +} /* int network_config */ static int network_shutdown (void) { - DEBUG ("Shutting down."); - listen_loop++; if (listen_thread != (pthread_t) 0) @@ -1215,7 +1301,8 @@ static int network_shutdown (void) listen_thread = (pthread_t) 0; } - listen_thread = 0; + if (send_buffer_fill > 0) + flush_buffer (); if (cache_tree != NULL) { @@ -1251,6 +1338,7 @@ static int network_init (void) memset (send_buffer_type, '\0', sizeof (send_buffer_type)); cache_tree = avl_create ((int (*) (const void *, const void *)) strcmp); + cache_flush_last = time (NULL); /* setup socket(s) and so on */ if (sending_sockets != NULL) @@ -1279,4 +1367,4 @@ void module_register (void) plugin_register_config ("network", network_config, config_keys, config_keys_num); plugin_register_init ("network", network_init); -} +} /* void module_register */