X-Git-Url: https://git.octo.it/?a=blobdiff_plain;f=src%2Fplugin.c;h=24e5e01e7bb06ceb1e12f1a45881a16505bd761a;hb=00a87614c6652bf67be9d717dc4ef9efd703f552;hp=265bf46ebce99dfe2f30d6413cea3f012eaa9e6c;hpb=a261da68f819bec318f819238f7fe5c428d50af1;p=collectd.git diff --git a/src/plugin.c b/src/plugin.c index 265bf46e..24e5e01e 100644 --- a/src/plugin.c +++ b/src/plugin.c @@ -34,6 +34,7 @@ #include "configfile.h" #include "utils_avltree.h" #include "utils_llist.h" +#include "utils_heap.h" #include "utils_cache.h" #include "utils_threshold.h" #include "filter_chain.h" @@ -50,6 +51,7 @@ typedef struct callback_func_s callback_func_t; #define RF_SIMPLE 0 #define RF_COMPLEX 1 +#define RF_REMOVE 65535 struct read_func_s { /* `read_func_t' "inherits" from `callback_func_t'. @@ -57,10 +59,11 @@ struct read_func_s #define rf_callback rf_super.cf_callback #define rf_udata rf_super.cf_udata callback_func_t rf_super; + char rf_name[DATA_MAX_NAME_LEN]; int rf_type; - int rf_wait_time; - int rf_wait_left; - enum { DONE = 0, TODO = 1, ACTIVE = 2 } rf_needs_read; + struct timespec rf_interval; + struct timespec rf_effective_interval; + struct timespec rf_next_read; }; typedef struct read_func_s read_func_t; @@ -68,7 +71,6 @@ typedef struct read_func_s read_func_t; * Private variables */ static llist_t *list_init; -static llist_t *list_read; static llist_t *list_write; static llist_t *list_flush; static llist_t *list_shutdown; @@ -82,6 +84,8 @@ static c_avl_tree_t *data_sets; static char *plugindir = NULL; +static c_heap_t *read_heap = NULL; +static llist_t *read_list; static int read_loop = 1; static pthread_mutex_t read_lock = PTHREAD_MUTEX_INITIALIZER; static pthread_cond_t read_cond = PTHREAD_COND_INITIALIZER; @@ -138,6 +142,26 @@ static void destroy_all_callbacks (llist_t **list) /* {{{ */ *list = NULL; } /* }}} void destroy_all_callbacks */ +static void destroy_read_heap (void) /* {{{ */ +{ + if (read_heap == NULL) + return; + + while (42) + { + callback_func_t *cf; + + cf = c_head_get_root (read_heap); + if (cf == NULL) + break; + + destroy_callback (cf); + } + + c_heap_destroy (read_heap); + read_heap = NULL; +} /* }}} void destroy_read_heap */ + static int register_callback (llist_t **list, /* {{{ */ const char *name, callback_func_t *cf) { @@ -196,14 +220,12 @@ static int register_callback (llist_t **list, /* {{{ */ static int create_register_callback (llist_t **list, /* {{{ */ const char *name, void *callback, user_data_t *ud) { - char *key; callback_func_t *cf; cf = (callback_func_t *) malloc (sizeof (*cf)); if (cf == NULL) { ERROR ("plugin: create_register_callback: malloc failed."); - sfree (key); return (-1); } memset (cf, 0, sizeof (*cf)); @@ -282,86 +304,156 @@ static int plugin_load_file (char *file) static void *plugin_read_thread (void __attribute__((unused)) *args) { - llentry_t *le; - read_func_t *rf; - int status; - int done; - - pthread_mutex_lock (&read_lock); - while (read_loop != 0) { - le = llist_head (list_read); - done = 0; + read_func_t *rf; + struct timeval now; + int status; + int rf_type; - while ((read_loop != 0) && (le != NULL)) + /* Get the read function that needs to be read next. */ + rf = c_head_get_root (read_heap); + if (rf == NULL) { - rf = (read_func_t *) le->value; + struct timespec abstime; - if (rf->rf_needs_read != TODO) - { - le = le->next; - continue; - } + gettimeofday (&now, /* timezone = */ NULL); - /* We will do this read function */ - rf->rf_needs_read = ACTIVE; + abstime.tv_sec = now.tv_sec + interval_g; + abstime.tv_nsec = 1000 * now.tv_usec; - DEBUG ("[thread #%5lu] plugin: plugin_read_thread: Handling %s", - (unsigned long int) pthread_self (), le->key); + pthread_mutex_lock (&read_lock); + pthread_cond_timedwait (&read_cond, &read_lock, + &abstime); pthread_mutex_unlock (&read_lock); + continue; + } - if (rf->rf_type == RF_SIMPLE) - { - int (*callback) (void); + if ((rf->rf_interval.tv_sec == 0) && (rf->rf_interval.tv_nsec == 0)) + { + gettimeofday (&now, /* timezone = */ NULL); - callback = rf->rf_callback; - status = (*callback) (); - } - else - { - plugin_read_cb callback; + rf->rf_interval.tv_sec = interval_g; + rf->rf_interval.tv_nsec = 0; - callback = rf->rf_callback; - status = (*callback) (&rf->rf_udata); - } + rf->rf_effective_interval = rf->rf_interval; - done++; + rf->rf_next_read.tv_sec = now.tv_sec; + rf->rf_next_read.tv_nsec = 1000 * now.tv_usec; + } - if (status != 0) - { - if (rf->rf_wait_time < interval_g) - rf->rf_wait_time = interval_g; - rf->rf_wait_left = rf->rf_wait_time; - rf->rf_wait_time = rf->rf_wait_time * 2; - if (rf->rf_wait_time > 86400) - rf->rf_wait_time = 86400; - - NOTICE ("read-function of plugin `%s' " - "failed. Will suspend it for %i " - "seconds.", le->key, rf->rf_wait_left); - } - else + /* sleep until this entry is due, + * using pthread_cond_timedwait */ + pthread_mutex_lock (&read_lock); + pthread_cond_timedwait (&read_cond, &read_lock, + &rf->rf_next_read); + /* Must hold `real_lock' when accessing `rf->rf_type'. */ + rf_type = rf->rf_type; + pthread_mutex_unlock (&read_lock); + + /* Check if we're supposed to stop.. This may have interrupted + * the sleep, too. */ + if (read_loop == 0) + { + /* Insert `rf' again, so it can be free'd correctly */ + c_heap_insert (read_heap, rf); + break; + } + + /* The entry has been marked for deletion. The linked list + * entry has already been removed by `plugin_unregister_read'. + * All we have to do here is free the `read_func_t' and + * continue. */ + if (rf_type == RF_REMOVE) + { + DEBUG ("plugin_read_thread: Destroying the `%s' " + "callback.", rf->rf_name); + destroy_callback ((callback_func_t *) rf); + rf = NULL; + continue; + } + + DEBUG ("plugin_read_thread: Handling `%s'.", rf->rf_name); + + if (rf_type == RF_SIMPLE) + { + int (*callback) (void); + + callback = rf->rf_callback; + status = (*callback) (); + } + else + { + plugin_read_cb callback; + + assert (rf_type == RF_COMPLEX); + + callback = rf->rf_callback; + status = (*callback) (&rf->rf_udata); + } + + /* If the function signals failure, we will increase the + * intervals in which it will be called. */ + if (status != 0) + { + rf->rf_effective_interval.tv_sec *= 2; + rf->rf_effective_interval.tv_nsec *= 2; + NORMALIZE_TIMESPEC (rf->rf_effective_interval); + + if (rf->rf_effective_interval.tv_sec >= 86400) { - rf->rf_wait_left = 0; - rf->rf_wait_time = interval_g; + rf->rf_effective_interval.tv_sec = 86400; + rf->rf_effective_interval.tv_nsec = 0; } - pthread_mutex_lock (&read_lock); - - rf->rf_needs_read = DONE; - le = le->next; - } /* while (le != NULL) */ + NOTICE ("read-function of plugin `%s' failed. " + "Will suspend it for %i seconds.", + rf->rf_name, + (int) rf->rf_effective_interval.tv_sec); + } + else + { + /* Success: Restore the interval, if it was changed. */ + rf->rf_effective_interval = rf->rf_interval; + } - if ((read_loop != 0) && (done == 0)) + /* update the ``next read due'' field */ + gettimeofday (&now, /* timezone = */ NULL); + + DEBUG ("plugin_read_thread: Effective interval of the " + "%s plugin is %i.%09i.", + rf->rf_name, + (int) rf->rf_effective_interval.tv_sec, + (int) rf->rf_effective_interval.tv_nsec); + + /* Calculate the next (absolute) time at which this function + * should be called. */ + rf->rf_next_read.tv_sec = rf->rf_next_read.tv_sec + + rf->rf_effective_interval.tv_sec; + rf->rf_next_read.tv_nsec = rf->rf_next_read.tv_nsec + + rf->rf_effective_interval.tv_nsec; + NORMALIZE_TIMESPEC (rf->rf_next_read); + + /* Check, if `rf_next_read' is in the past. */ + if ((rf->rf_next_read.tv_sec < now.tv_sec) + || ((rf->rf_next_read.tv_sec == now.tv_sec) + && (rf->rf_next_read.tv_nsec < (1000 * now.tv_usec)))) { - DEBUG ("[thread #%5lu] plugin: plugin_read_thread: Waiting on read_cond.", - (unsigned long int) pthread_self ()); - pthread_cond_wait (&read_cond, &read_lock); + /* `rf_next_read' is in the past. Insert `now' + * so this value doesn't trail off into the + * past too much. */ + rf->rf_next_read.tv_sec = now.tv_sec; + rf->rf_next_read.tv_nsec = 1000 * now.tv_usec; } - } /* while (read_loop) */ - pthread_mutex_unlock (&read_lock); + DEBUG ("plugin_read_thread: Next read of the %s plugin at %i.%09i.", + rf->rf_name, + (int) rf->rf_next_read.tv_sec, + (int) rf->rf_next_read.tv_nsec); + + /* Re-insert this read function into the heap again. */ + c_heap_insert (read_heap, rf); + } /* while (read_loop) */ pthread_exit (NULL); return ((void *) 0); @@ -501,6 +593,7 @@ int plugin_load (const char *type) else if (!S_ISREG (statbuf.st_mode)) { /* don't follow symlinks */ + WARNING ("stat %s: not a regular file", filename); continue; } @@ -548,6 +641,82 @@ int plugin_register_init (const char *name, /* user_data = */ NULL)); } /* plugin_register_init */ +static int plugin_compare_read_func (const void *arg0, const void *arg1) +{ + const read_func_t *rf0; + const read_func_t *rf1; + + rf0 = arg0; + rf1 = arg1; + + if (rf0->rf_next_read.tv_sec < rf1->rf_next_read.tv_sec) + return (-1); + else if (rf0->rf_next_read.tv_sec > rf1->rf_next_read.tv_sec) + return (1); + else if (rf0->rf_next_read.tv_nsec < rf1->rf_next_read.tv_nsec) + return (-1); + else if (rf0->rf_next_read.tv_nsec > rf1->rf_next_read.tv_nsec) + return (1); + else + return (0); +} /* int plugin_compare_read_func */ + +/* Add a read function to both, the heap and a linked list. The linked list if + * used to look-up read functions, especially for the remove function. The heap + * is used to determine which plugin to read next. */ +static int plugin_insert_read (read_func_t *rf) +{ + int status; + llentry_t *le; + + pthread_mutex_lock (&read_lock); + + if (read_list == NULL) + { + read_list = llist_create (); + if (read_list == NULL) + { + pthread_mutex_unlock (&read_lock); + ERROR ("plugin_insert_read: read_list failed."); + return (-1); + } + } + + if (read_heap == NULL) + { + read_heap = c_heap_create (plugin_compare_read_func); + if (read_heap == NULL) + { + pthread_mutex_unlock (&read_lock); + ERROR ("plugin_insert_read: c_heap_create failed."); + return (-1); + } + } + + le = llentry_create (rf->rf_name, rf); + if (le == NULL) + { + pthread_mutex_unlock (&read_lock); + ERROR ("plugin_insert_read: llentry_create failed."); + return (-1); + } + + status = c_heap_insert (read_heap, rf); + if (status != 0) + { + pthread_mutex_unlock (&read_lock); + ERROR ("plugin_insert_read: c_heap_insert failed."); + llentry_destroy (le); + return (-1); + } + + /* This does not fail. */ + llist_append (read_list, le); + + pthread_mutex_unlock (&read_lock); + return (0); +} /* int plugin_insert_read */ + int plugin_register_read (const char *name, int (*callback) (void)) { @@ -566,16 +735,19 @@ int plugin_register_read (const char *name, rf->rf_callback = (void *) callback; rf->rf_udata.data = NULL; rf->rf_udata.free_func = NULL; - rf->rf_wait_time = interval_g; - rf->rf_wait_left = 0; + sstrncpy (rf->rf_name, name, sizeof (rf->rf_name)); rf->rf_type = RF_SIMPLE; - rf->rf_needs_read = DONE; + rf->rf_interval.tv_sec = 0; + rf->rf_interval.tv_nsec = 0; + rf->rf_effective_interval = rf->rf_interval; - return (register_callback (&list_read, name, (callback_func_t *) rf)); + return (plugin_insert_read (rf)); } /* int plugin_register_read */ int plugin_register_complex_read (const char *name, - plugin_read_cb callback, user_data_t *user_data) + plugin_read_cb callback, + const struct timespec *interval, + user_data_t *user_data) { read_func_t *rf; @@ -588,10 +760,13 @@ int plugin_register_complex_read (const char *name, memset (rf, 0, sizeof (read_func_t)); rf->rf_callback = (void *) callback; - rf->rf_wait_time = interval_g; - rf->rf_wait_left = 0; + sstrncpy (rf->rf_name, name, sizeof (rf->rf_name)); rf->rf_type = RF_COMPLEX; - rf->rf_needs_read = DONE; + if (interval != NULL) + { + rf->rf_interval = *interval; + } + rf->rf_effective_interval = rf->rf_interval; /* Set user data */ if (user_data == NULL) @@ -604,7 +779,7 @@ int plugin_register_complex_read (const char *name, rf->rf_udata = *user_data; } - return (register_callback (&list_read, name, (callback_func_t *) rf)); + return (plugin_insert_read (rf)); } /* int plugin_register_complex_read */ int plugin_register_write (const char *name, @@ -696,10 +871,45 @@ int plugin_unregister_init (const char *name) return (plugin_unregister (list_init, name)); } -int plugin_unregister_read (const char *name) +int plugin_unregister_read (const char *name) /* {{{ */ { - return (plugin_unregister (list_read, name)); -} + llentry_t *le; + read_func_t *rf; + + if (name == NULL) + return (-ENOENT); + + pthread_mutex_lock (&read_lock); + + if (read_list == NULL) + { + pthread_mutex_unlock (&read_lock); + return (-ENOENT); + } + + le = llist_search (read_list, name); + if (le == NULL) + { + pthread_mutex_unlock (&read_lock); + WARNING ("plugin_unregister_read: No such read function: %s", + name); + return (-ENOENT); + } + + llist_remove (read_list, le); + + rf = le->value; + assert (rf != NULL); + rf->rf_type = RF_REMOVE; + + pthread_mutex_unlock (&read_lock); + + llentry_destroy (le); + + DEBUG ("plugin_unregister_read: Marked `%s' for removal.", name); + + return (0); +} /* }}} int plugin_unregister_read */ int plugin_unregister_write (const char *name) { @@ -758,7 +968,7 @@ void plugin_init_all (void) post_cache_chain = fc_chain_get_by_name (chain_name); - if ((list_init == NULL) && (list_read == NULL)) + if ((list_init == NULL) && (read_heap == NULL)) return; /* Calling all init callbacks before checking if read callbacks @@ -791,7 +1001,7 @@ void plugin_init_all (void) } /* Start read-threads */ - if (list_read != NULL) + if (read_heap != NULL) { const char *rt; int num; @@ -802,64 +1012,33 @@ void plugin_init_all (void) } } /* void plugin_init_all */ +/* TODO: Rename this function. */ void plugin_read_all (void) { - llentry_t *le; - read_func_t *rf; - uc_check_timeout (); - if (list_read == NULL) - return; - - pthread_mutex_lock (&read_lock); - - le = llist_head (list_read); - while (le != NULL) - { - rf = (read_func_t *) le->value; - - if (rf->rf_needs_read != DONE) - { - le = le->next; - continue; - } - - if (rf->rf_wait_left > 0) - rf->rf_wait_left -= interval_g; - - if (rf->rf_wait_left <= 0) - { - rf->rf_needs_read = TODO; - } - - le = le->next; - } - - DEBUG ("plugin: plugin_read_all: Signalling `read_cond'"); - pthread_cond_broadcast (&read_cond); - pthread_mutex_unlock (&read_lock); + return; } /* void plugin_read_all */ /* Read function called when the `-T' command line argument is given. */ int plugin_read_all_once (void) { - llentry_t *le; - read_func_t *rf; int status; int return_status = 0; - if (list_read == NULL) + if (read_heap == NULL) { NOTICE ("No read-functions are registered."); return (0); } - for (le = llist_head (list_read); - le != NULL; - le = le->next) + while (42) { - rf = (read_func_t *) le->value; + read_func_t *rf; + + rf = c_head_get_root (read_heap); + if (rf == NULL) + break; if (rf->rf_type == RF_SIMPLE) { @@ -879,9 +1058,11 @@ int plugin_read_all_once (void) if (status != 0) { NOTICE ("read-function of plugin `%s' failed.", - le->key); + rf->rf_name); return_status = -1; } + + destroy_callback ((void *) rf); } return (return_status); @@ -1000,7 +1181,13 @@ void plugin_shutdown_all (void) stop_read_threads (); destroy_all_callbacks (&list_init); - destroy_all_callbacks (&list_read); + + pthread_mutex_lock (&read_lock); + llist_destroy (read_list); + read_list = NULL; + pthread_mutex_unlock (&read_lock); + + destroy_read_heap (); plugin_flush (/* plugin = */ NULL, /* timeout = */ -1, /* identifier = */ NULL); @@ -1073,6 +1260,9 @@ int plugin_dispatch_values (value_list_t *vl) if (vl->time == 0) vl->time = time (NULL); + if (vl->interval <= 0) + vl->interval = interval_g; + DEBUG ("plugin_dispatch_values: time = %u; interval = %i; " "host = %s; " "plugin = %s; plugin_instance = %s; " @@ -1161,6 +1351,9 @@ int plugin_dispatch_values (value_list_t *vl) /* Update the value cache */ uc_update (ds, vl); + /* Initiate threshold checking */ + ut_check_threshold (ds, vl); + if (post_cache_chain != NULL) { status = fc_process_chain (ds, vl, post_cache_chain); @@ -1231,7 +1424,12 @@ void plugin_log (int level, const char *format, ...) llentry_t *le; if (list_log == NULL) + { + va_start (ap, format); + vfprintf (stderr, format, ap); + va_end (ap); return; + } #if !COLLECT_DEBUG if (level >= LOG_DEBUG)