static pthread_cond_t receive_list_cond = PTHREAD_COND_INITIALIZER;
static struct pollfd *listen_sockets = NULL;
-static int listen_sockets_num = 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;
-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;
-
-static c_avl_tree_t *cache_tree = NULL;
+static int listen_sockets_num = 0;
+
+/* The receive and dispatch threads will run as long as `listen_loop' is set to
+ * zero. */
+static int listen_loop = 0;
+static int receive_thread_running = 0;
+static pthread_t receive_thread_id;
+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_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;
exp_size = 3 * sizeof (uint16_t)
+ pkg_numval * (sizeof (uint8_t) + sizeof (value_t));
- if (buffer_len < exp_size)
+ if ((buffer_len < 0) || ((size_t) buffer_len < exp_size))
{
WARNING ("network plugin: parse_part_values: "
"Packet too short: "
uint16_t pkg_length;
uint16_t pkg_type;
- if (buffer_len < exp_size)
+ if ((buffer_len < 0) || ((size_t) buffer_len < exp_size))
{
WARNING ("network plugin: parse_part_number: "
"Packet too short: "
uint16_t pkg_length;
uint16_t pkg_type;
- if (buffer_len < header_size)
+ if ((buffer_len < 0) || ((size_t) buffer_len < header_size))
{
WARNING ("network plugin: parse_part_string: "
"Packet too short: "
/* Check that the package data fits into the output buffer.
* The previous if-statement ensures that:
* `pkg_length > header_size' */
- if ((pkg_length - header_size) > output_len)
+ if ((output_len < 0)
+ || ((size_t) output_len < ((size_t) pkg_length - header_size)))
{
WARNING ("network plugin: parse_part_string: "
"Output buffer too small.");
return (0);
} /* int network_get_listen_socket */
-static void *dispatch_thread (void *arg)
+static void *dispatch_thread (void __attribute__((unused)) *arg)
{
while (42)
{
return (-1);
}
+ /* 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
+ * more of these structures. */
ent = malloc (sizeof (receive_list_entry_t));
if (ent == NULL)
{
return (0);
}
-static void *receive_thread (void *arg)
+static void *receive_thread (void __attribute__((unused)) *arg)
{
return (network_receive () ? (void *) 1 : (void *) 0);
} /* void *receive_thread */
memset (&send_buffer_vl, 0, sizeof (send_buffer_vl));
}
-static int network_write (const data_set_t *ds, const value_list_t *vl)
+static int network_write (const data_set_t *ds, const value_list_t *vl,
+ user_data_t __attribute__((unused)) *user_data)
{
int status;
return (0);
} /* int network_config */
-static int network_notification (const notification_t *n)
+static int network_notification (const notification_t *n,
+ user_data_t __attribute__((unused)) *user_data)
{
char buffer[BUFF_SIZE];
char *buffer_ptr = buffer;
listen_loop++;
/* Kill the listening thread */
- if (receive_thread_id != (pthread_t) 0)
+ if (receive_thread_running != 0)
{
+ INFO ("network plugin: Stopping receive thread.");
pthread_kill (receive_thread_id, SIGTERM);
pthread_join (receive_thread_id, NULL /* no return value */);
- receive_thread_id = (pthread_t) 0;
+ memset (&receive_thread_id, 0, sizeof (receive_thread_id));
+ receive_thread_running = 0;
}
/* Shutdown the dispatching thread */
- if (dispatch_thread_id != (pthread_t) 0)
+ if (dispatch_thread_running != 0)
+ {
+ INFO ("network plugin: Stopping dispatch thread.");
+ pthread_join (dispatch_thread_id, /* ret = */ NULL);
pthread_cond_broadcast (&receive_list_cond);
+ dispatch_thread_running = 0;
+ }
if (send_buffer_fill > 0)
flush_buffer ();
/* setup socket(s) and so on */
if (sending_sockets != NULL)
{
- plugin_register_write ("network", network_write);
- plugin_register_notification ("network", network_notification);
+ plugin_register_write ("network", network_write,
+ /* user_data = */ NULL);
+ plugin_register_notification ("network", network_notification,
+ /* user_data = */ NULL);
}
- if ((listen_sockets_num != 0) && (receive_thread_id == 0))
+ /* If no threads need to be started, return here. */
+ if ((listen_sockets_num == 0)
+ || ((dispatch_thread_running != 0)
+ && (receive_thread_running != 0)))
+ return (0);
+
+ if (dispatch_thread_running == 0)
{
int status;
-
status = pthread_create (&dispatch_thread_id,
NULL /* no attributes */,
dispatch_thread,
sstrerror (errno, errbuf,
sizeof (errbuf)));
}
+ else
+ {
+ dispatch_thread_running = 1;
+ }
+ }
+ if (receive_thread_running == 0)
+ {
+ int status;
status = pthread_create (&receive_thread_id,
NULL /* no attributes */,
receive_thread,
sstrerror (errno, errbuf,
sizeof (errbuf)));
}
+ else
+ {
+ receive_thread_running = 1;
+ }
}
+
return (0);
} /* int network_init */
* 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)
+static int network_flush (int timeout,
+ const char __attribute__((unused)) *identifier,
+ user_data_t __attribute__((unused)) *user_data)
{
pthread_mutex_lock (&send_buffer_lock);
plugin_register_config ("network", network_config,
config_keys, config_keys_num);
plugin_register_init ("network", network_init);
- plugin_register_flush ("network", network_flush);
+ plugin_register_flush ("network", network_flush,
+ /* user_data = */ NULL);
} /* void module_register */
projects / collectd.git / blobdiff