X-Git-Url: https://git.octo.it/?a=blobdiff_plain;f=src%2Fnetwork.c;h=17038a45a0a9e09416b0ce5d28f43bafd005469e;hb=b5c5890955fa19651ad8b3f48d99364d270a0d8f;hp=ad8d530e4780d4ff9ce8ab96d72b65455a243c42;hpb=4747e79c6e87d443e9f390f8a446521c29c66fb0;p=collectd.git diff --git a/src/network.c b/src/network.c index ad8d530e..17038a45 100644 --- a/src/network.c +++ b/src/network.c @@ -136,6 +136,14 @@ struct part_values_s }; typedef struct part_values_s part_values_t; +struct receive_list_entry_s +{ + char data[BUFF_SIZE]; + int data_len; + struct receive_list_entry_s *next; +}; +typedef struct receive_list_entry_s receive_list_entry_t; + /* * Private variables */ @@ -154,10 +162,17 @@ static int network_config_forward = 0; static sockent_t *sending_sockets = NULL; +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 struct pollfd *listen_sockets = NULL; static int listen_sockets_num = 0; -static pthread_t listen_thread = 0; + static int listen_loop = 0; +static pthread_t receive_thread_id = 0; +static pthread_t dispatch_thread_id = 0; static char send_buffer[BUFF_SIZE]; static char *send_buffer_ptr; @@ -166,7 +181,7 @@ 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 +199,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 +213,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 +221,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 +261,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 +282,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 @@ -989,7 +1004,38 @@ static int network_add_sending_socket (const char *node, const char *service) return (0); } /* int network_get_listen_socket */ -int network_receive (void) +static void *dispatch_thread (void *arg) +{ + while (42) + { + receive_list_entry_t *ent; + + /* Lock and wait for more data to come in */ + pthread_mutex_lock (&receive_list_lock); + while ((listen_loop == 0) + && (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; + pthread_mutex_unlock (&receive_list_lock); + + /* Check whether we are supposed to exit. We do NOT check `listen_loop' + * because we dispatch all missing packets before shutting down. */ + if (ent == NULL) + break; + + parse_packet (ent->data, ent->data_len); + + sfree (ent); + } /* while (42) */ + + return (NULL); +} /* void *receive_thread */ + +static int network_receive (void) { char buffer[BUFF_SIZE]; int buffer_len; @@ -1022,6 +1068,8 @@ int network_receive (void) for (i = 0; (i < listen_sockets_num) && (status > 0); i++) { + receive_list_entry_t *ent; + if ((listen_sockets[i].revents & (POLLIN | POLLPRI)) == 0) continue; status--; @@ -1038,7 +1086,35 @@ int network_receive (void) return (-1); } - parse_packet (buffer, buffer_len); + ent = malloc (sizeof (receive_list_entry_t)); + if (ent == NULL) + { + ERROR ("network plugin: malloc failed."); + return (-1); + } + memset (ent, '\0', sizeof (receive_list_entry_t)); + + /* 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; + + pthread_mutex_lock (&receive_list_lock); + if (receive_list_head == NULL) + { + receive_list_head = ent; + receive_list_tail = ent; + } + else + { + receive_list_tail->next = ent; + receive_list_tail = ent; + } + pthread_cond_signal (&receive_list_cond); + pthread_mutex_unlock (&receive_list_lock); } /* for (listen_sockets) */ } /* while (listen_loop == 0) */ @@ -1091,7 +1167,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) @@ -1100,8 +1175,6 @@ 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) @@ -1110,8 +1183,6 @@ static int add_to_buffer (char *buffer, int buffer_size, (uint64_t) vl->interval)) return (-1); vl_def->interval = vl->interval; - DEBUG ("network plugin: add_to_buffer: interval = %i", - (int) vl->interval); } if (strcmp (vl_def->plugin, vl->plugin) != 0) @@ -1120,8 +1191,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) @@ -1131,8 +1200,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) @@ -1141,7 +1208,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) @@ -1151,8 +1217,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) @@ -1249,7 +1313,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) @@ -1290,30 +1358,34 @@ 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) + /* Kill the listening thread */ + if (receive_thread_id != (pthread_t) 0) { - pthread_kill (listen_thread, SIGTERM); - pthread_join (listen_thread, NULL /* no return value */); - listen_thread = (pthread_t) 0; + pthread_kill (receive_thread_id, SIGTERM); + pthread_join (receive_thread_id, NULL /* no return value */); + receive_thread_id = (pthread_t) 0; } - listen_thread = 0; + /* Shutdown the dispatching thread */ + if (dispatch_thread_id != (pthread_t) 0) + pthread_cond_broadcast (&receive_list_cond); + + 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; } @@ -1336,20 +1408,33 @@ 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 */ if (sending_sockets != NULL) plugin_register_write ("network", network_write); - if ((listen_sockets_num != 0) && (listen_thread == 0)) + if ((listen_sockets_num != 0) && (receive_thread_id == 0)) { int status; - status = pthread_create (&listen_thread, NULL /* no attributes */, - receive_thread, NULL /* no argument */); + status = pthread_create (&dispatch_thread_id, + NULL /* no attributes */, + dispatch_thread, + NULL /* no argument */); + if (status != 0) + { + char errbuf[1024]; + ERROR ("network: pthread_create failed: %s", + sstrerror (errno, errbuf, + sizeof (errbuf))); + } + status = pthread_create (&receive_thread_id, + NULL /* no attributes */, + receive_thread, + NULL /* no argument */); if (status != 0) { char errbuf[1024];