X-Git-Url: https://git.octo.it/?a=blobdiff_plain;f=src%2Fnetwork.c;h=34f89d96a39898cc4e27eb98024245fd686d1a65;hb=0b6de87f3d08326ee3200db8e16f3738df027ac0;hp=e15036424d880316ec58bcdffd7d9d3cc76ff2ae;hpb=af46a5f31a0e8d4279d63d8ca9232dbd433dfb25;p=collectd.git diff --git a/src/network.c b/src/network.c index e1503642..34f89d96 100644 --- a/src/network.c +++ b/src/network.c @@ -178,12 +178,11 @@ 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_STATIC; -static char send_buffer_type[DATA_MAX_NAME_LEN]; static pthread_mutex_t send_buffer_lock = PTHREAD_MUTEX_INITIALIZER; static c_avl_tree_t *cache_tree = NULL; static pthread_mutex_t cache_lock = PTHREAD_MUTEX_INITIALIZER; -static time_t cache_flush_last; +static time_t cache_flush_last = 0; static int cache_flush_interval = 1800; /* @@ -246,7 +245,7 @@ static int cache_flush (void) return (0); } /* int cache_flush */ -static int cache_check (const char *type, const value_list_t *vl) +static int cache_check (const value_list_t *vl) { char key[1024]; time_t *value = NULL; @@ -256,7 +255,7 @@ static int cache_check (const char *type, const value_list_t *vl) return (-1); if (format_name (key, sizeof (key), vl->host, vl->plugin, - vl->plugin_instance, type, vl->type_instance)) + vl->plugin_instance, vl->type, vl->type_instance)) return (-1); pthread_mutex_lock (&cache_lock); @@ -678,14 +677,12 @@ static int parse_packet (void *buffer, int buffer_len) int status; value_list_t vl = VALUE_LIST_INIT; - char type[DATA_MAX_NAME_LEN]; notification_t n; DEBUG ("network plugin: parse_packet: buffer = %p; buffer_len = %i;", buffer, buffer_len); memset (&vl, '\0', sizeof (vl)); - memset (&type, '\0', sizeof (type)); memset (&n, '\0', sizeof (n)); status = 0; @@ -722,10 +719,10 @@ static int parse_packet (void *buffer, int buffer_len) if ((vl.time > 0) && (strlen (vl.host) > 0) && (strlen (vl.plugin) > 0) - && (strlen (type) > 0) - && (cache_check (type, &vl) == 0)) + && (strlen (vl.type) > 0) + && (cache_check (&vl) == 0)) { - plugin_dispatch_values (type, &vl); + plugin_dispatch_values (&vl); } else { @@ -782,9 +779,9 @@ static int parse_packet (void *buffer, int buffer_len) else if (pkg_type == TYPE_TYPE) { status = parse_part_string (&buffer, &buffer_len, - type, sizeof (type)); + vl.type, sizeof (vl.type)); if (status == 0) - sstrncpy (n.type, type, sizeof (n.type)); + sstrncpy (n.type, vl.type, sizeof (n.type)); } else if (pkg_type == TYPE_TYPE_INSTANCE) { @@ -1282,6 +1279,9 @@ static int network_receive (void) int i; int status; + receive_list_entry_t *private_list_head; + receive_list_entry_t *private_list_tail; + if (listen_sockets_num == 0) network_add_listen_socket (NULL, NULL); @@ -1291,6 +1291,9 @@ static int network_receive (void) return (-1); } + private_list_head = NULL; + private_list_tail = NULL; + while (listen_loop == 0) { status = poll (listen_sockets, listen_sockets_num, -1); @@ -1331,7 +1334,8 @@ static int network_receive (void) ERROR ("network plugin: malloc failed."); return (-1); } - memset (ent, '\0', sizeof (receive_list_entry_t)); + memset (ent, 0, sizeof (receive_list_entry_t)); + ent->next = NULL; /* Hopefully this be optimized out by the compiler. It * might help prevent stupid bugs in the future though. @@ -1341,22 +1345,49 @@ static int network_receive (void) 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; - } + if (private_list_head == NULL) + private_list_head = ent; else + private_list_tail->next = ent; + private_list_tail = ent; + + /* Do not block here. Blocking here has led to + * insufficient performance in the past. */ + if (pthread_mutex_trylock (&receive_list_lock) == 0) { - receive_list_tail->next = ent; - receive_list_tail = ent; + 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; + + pthread_cond_signal (&receive_list_cond); + pthread_mutex_unlock (&receive_list_lock); } - pthread_cond_signal (&receive_list_cond); - pthread_mutex_unlock (&receive_list_lock); } /* for (listen_sockets) */ } /* while (listen_loop == 0) */ + /* Make sure everything is dispatched before exiting. */ + if (private_list_head != NULL) + { + pthread_mutex_lock (&receive_list_lock); + + 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; + + pthread_cond_signal (&receive_list_cond); + pthread_mutex_unlock (&receive_list_lock); + } + return (0); } @@ -1395,7 +1426,7 @@ static void network_send_buffer (const char *buffer, int buffer_len) } /* void network_send_buffer */ static int add_to_buffer (char *buffer, int buffer_size, - value_list_t *vl_def, char *type_def, + value_list_t *vl_def, const data_set_t *ds, const value_list_t *vl) { char *buffer_orig = buffer; @@ -1405,7 +1436,7 @@ static int add_to_buffer (char *buffer, int buffer_size, if (write_part_string (&buffer, &buffer_size, TYPE_HOST, vl->host, strlen (vl->host)) != 0) return (-1); - strcpy (vl_def->host, vl->host); + sstrncpy (vl_def->host, vl->host, sizeof (vl_def->host)); } if (vl_def->time != vl->time) @@ -1429,7 +1460,7 @@ static int add_to_buffer (char *buffer, int buffer_size, if (write_part_string (&buffer, &buffer_size, TYPE_PLUGIN, vl->plugin, strlen (vl->plugin)) != 0) return (-1); - strcpy (vl_def->plugin, vl->plugin); + sstrncpy (vl_def->plugin, vl->plugin, sizeof (vl_def->plugin)); } if (strcmp (vl_def->plugin_instance, vl->plugin_instance) != 0) @@ -1438,15 +1469,15 @@ static int add_to_buffer (char *buffer, int buffer_size, vl->plugin_instance, strlen (vl->plugin_instance)) != 0) return (-1); - strcpy (vl_def->plugin_instance, vl->plugin_instance); + sstrncpy (vl_def->plugin_instance, vl->plugin_instance, sizeof (vl_def->plugin_instance)); } - if (strcmp (type_def, ds->type) != 0) + if (strcmp (vl_def->type, vl->type) != 0) { if (write_part_string (&buffer, &buffer_size, TYPE_TYPE, - ds->type, strlen (ds->type)) != 0) + vl->type, strlen (vl->type)) != 0) return (-1); - strcpy (type_def, ds->type); + sstrncpy (vl_def->type, ds->type, sizeof (vl_def->type)); } if (strcmp (vl_def->type_instance, vl->type_instance) != 0) @@ -1455,7 +1486,7 @@ static int add_to_buffer (char *buffer, int buffer_size, vl->type_instance, strlen (vl->type_instance)) != 0) return (-1); - strcpy (vl_def->type_instance, vl->type_instance); + sstrncpy (vl_def->type_instance, vl->type_instance, sizeof (vl_def->type_instance)); } if (write_part_values (&buffer, &buffer_size, ds, vl) != 0) @@ -1472,8 +1503,7 @@ static void flush_buffer (void) network_send_buffer (send_buffer, send_buffer_fill); send_buffer_ptr = send_buffer; send_buffer_fill = 0; - memset (&send_buffer_vl, '\0', sizeof (send_buffer_vl)); - memset (send_buffer_type, '\0', sizeof (send_buffer_type)); + memset (&send_buffer_vl, 0, sizeof (send_buffer_vl)); } static int network_write (const data_set_t *ds, const value_list_t *vl) @@ -1483,7 +1513,7 @@ static int network_write (const data_set_t *ds, const value_list_t *vl) /* 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 (ds->type, vl); + status = cache_check (vl); if ((network_config_forward == 0) && (status != 0)) return (0); @@ -1492,7 +1522,7 @@ static int network_write (const data_set_t *ds, const value_list_t *vl) status = add_to_buffer (send_buffer_ptr, sizeof (send_buffer) - send_buffer_fill, - &send_buffer_vl, send_buffer_type, + &send_buffer_vl, ds, vl); if (status >= 0) { @@ -1506,7 +1536,7 @@ static int network_write (const data_set_t *ds, const value_list_t *vl) status = add_to_buffer (send_buffer_ptr, sizeof (send_buffer) - send_buffer_fill, - &send_buffer_vl, send_buffer_type, + &send_buffer_vl, ds, vl); if (status >= 0) @@ -1550,7 +1580,10 @@ static int network_config (const char *key, const char *val) fields_num = strsplit (val_cpy, fields, 3); if ((fields_num != 1) && (fields_num != 2)) + { + sfree (val_cpy); return (1); + } else if (fields_num == 2) { if ((service = strchr (fields[1], '.')) != NULL) @@ -1563,6 +1596,8 @@ static int network_config (const char *key, const char *val) network_add_listen_socket (node, service); else network_add_sending_socket (node, service); + + sfree (val_cpy); } else if (strcasecmp ("TimeToLive", key) == 0) { @@ -1706,17 +1741,24 @@ 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_init (void) { + /* 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) + return (0); + plugin_register_shutdown ("network", network_shutdown); send_buffer_ptr = send_buffer; send_buffer_fill = 0; - memset (&send_buffer_vl, '\0', sizeof (send_buffer_vl)); - memset (send_buffer_type, '\0', sizeof (send_buffer_type)); + memset (&send_buffer_vl, 0, sizeof (send_buffer_vl)); cache_tree = c_avl_create ((int (*) (const void *, const void *)) strcmp); cache_flush_last = time (NULL); @@ -1759,7 +1801,14 @@ static int network_init (void) return (0); } /* int network_init */ -static int network_flush (int timeout) +/* + * The flush option of the network plugin cannot flush individual identifiers. + * All the values are added to a buffer and sent when the buffer is full, the + * requested value may or may not be in there, it's not worth finding out. We + * just send the buffer if `flush' is called - if the requested value was in + * there, good. If not, well, then there is nothing to flush.. -octo + */ +static int network_flush (int timeout, const char *identifier) { pthread_mutex_lock (&send_buffer_lock);