X-Git-Url: https://git.octo.it/?a=blobdiff_plain;f=src%2Fplugin.c;h=9c2af5d7b8c6b3617512df94ee5dc27c1fbd380b;hb=d8825859c7b31f068899c2d5f8a147ea1ef65750;hp=91c40b6bad115203cb021f95bd919583310c9460;hpb=a24bf968e72f9a103a354dcc833619a40849cfcd;p=collectd.git diff --git a/src/plugin.c b/src/plugin.c index 91c40b6b..9c2af5d7 100644 --- a/src/plugin.c +++ b/src/plugin.c @@ -45,6 +45,7 @@ struct callback_func_s { void *cf_callback; user_data_t cf_udata; + plugin_ctx_t cf_ctx; }; typedef struct callback_func_s callback_func_t; @@ -57,6 +58,7 @@ struct read_func_s * The `rf_super' member MUST be the first one in this structure! */ #define rf_callback rf_super.cf_callback #define rf_udata rf_super.cf_udata +#define rf_ctx rf_super.cf_ctx callback_func_t rf_super; char rf_group[DATA_MAX_NAME_LEN]; char rf_name[DATA_MAX_NAME_LEN]; @@ -93,6 +95,9 @@ static pthread_cond_t read_cond = PTHREAD_COND_INITIALIZER; static pthread_t *read_threads = NULL; static int read_threads_num = 0; +static pthread_key_t plugin_ctx_key; +static _Bool plugin_ctx_key_initialized = 0; + /* * Static functions */ @@ -246,6 +251,8 @@ static int create_register_callback (llist_t **list, /* {{{ */ cf->cf_udata = *ud; } + cf->cf_ctx = plugin_get_ctx (); + return (register_callback (list, name, cf)); } /* }}} int create_register_callback */ @@ -291,7 +298,7 @@ static int plugin_load_file (char *file, uint32_t flags) dlh = lt_dlopenadvise(file, advise); lt_dladvise_destroy(&advise); } else { - dlh = lt_dlopen (file); + dlh = lt_dlopen (file); } #else /* if LIBTOOL_VERSION == 1 */ if (flags & PLUGIN_FLAGS_GLOBAL) @@ -346,6 +353,7 @@ static void *plugin_read_thread (void __attribute__((unused)) *args) while (read_loop != 0) { read_func_t *rf; + plugin_ctx_t old_ctx; cdtime_t now; int status; int rf_type; @@ -372,7 +380,7 @@ static void *plugin_read_thread (void __attribute__((unused)) *args) { now = cdtime (); - CDTIME_T_TO_TIMESPEC (interval_g, &rf->rf_interval); + CDTIME_T_TO_TIMESPEC (plugin_get_interval (), &rf->rf_interval); rf->rf_effective_interval = rf->rf_interval; @@ -423,6 +431,8 @@ static void *plugin_read_thread (void __attribute__((unused)) *args) DEBUG ("plugin_read_thread: Handling `%s'.", rf->rf_name); + old_ctx = plugin_set_ctx (rf->rf_ctx); + if (rf_type == RF_SIMPLE) { int (*callback) (void); @@ -440,6 +450,8 @@ static void *plugin_read_thread (void __attribute__((unused)) *args) status = (*callback) (&rf->rf_udata); } + plugin_set_ctx (old_ctx); + /* If the function signals failure, we will increase the * intervals in which it will be called. */ if (status != 0) @@ -773,12 +785,43 @@ static int plugin_insert_read (read_func_t *rf) return (0); } /* int plugin_insert_read */ +static int read_cb_wrapper (user_data_t *ud) +{ + int (*callback) (void); + + if (ud == NULL) + return -1; + + callback = ud->data; + return callback(); +} /* int read_cb_wrapper */ + int plugin_register_read (const char *name, int (*callback) (void)) { read_func_t *rf; + plugin_ctx_t ctx = plugin_get_ctx (); int status; + if (ctx.interval != 0) { + /* If ctx.interval is not zero (== use the plugin or global + * interval), we need to use the "complex" read callback, + * because only that allows to specify a different interval. + * Wrap the callback using read_cb_wrapper(). */ + struct timespec interval; + user_data_t user_data; + + DEBUG ("plugin_register_read: plugin_interval = %.3f", + CDTIME_T_TO_DOUBLE(plugin_interval)); + + user_data.data = callback; + user_data.free_func = NULL; + + CDTIME_T_TO_TIMESPEC (ctx.interval, &interval); + return plugin_register_complex_read (/* group = */ NULL, + name, read_cb_wrapper, &interval, &user_data); + } + rf = malloc (sizeof (*rf)); if (rf == NULL) { @@ -790,6 +833,7 @@ 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_ctx = ctx; rf->rf_group[0] = '\0'; sstrncpy (rf->rf_name, name, sizeof (rf->rf_name)); rf->rf_type = RF_SIMPLE; @@ -810,6 +854,7 @@ int plugin_register_complex_read (const char *group, const char *name, user_data_t *user_data) { read_func_t *rf; + plugin_ctx_t ctx = plugin_get_ctx (); int status; rf = malloc (sizeof (*rf)); @@ -831,6 +876,10 @@ int plugin_register_complex_read (const char *group, const char *name, { rf->rf_interval = *interval; } + else if (ctx.interval != 0) + { + CDTIME_T_TO_TIMESPEC (ctx.interval, &rf->rf_interval); + } rf->rf_effective_interval = rf->rf_interval; /* Set user data */ @@ -844,6 +893,8 @@ int plugin_register_complex_read (const char *group, const char *name, rf->rf_udata = *user_data; } + rf->rf_ctx = ctx; + status = plugin_insert_read (rf); if (status != 0) sfree (rf); @@ -1121,10 +1172,13 @@ void plugin_init_all (void) { callback_func_t *cf; plugin_init_cb callback; + plugin_ctx_t old_ctx; cf = le->value; + old_ctx = plugin_set_ctx (cf->cf_ctx); callback = cf->cf_callback; status = (*callback) (); + plugin_set_ctx (old_ctx); if (status != 0) { @@ -1177,11 +1231,14 @@ int plugin_read_all_once (void) while (42) { read_func_t *rf; + plugin_ctx_t old_ctx; rf = c_heap_get_root (read_heap); if (rf == NULL) break; + old_ctx = plugin_set_ctx (rf->rf_ctx); + if (rf->rf_type == RF_SIMPLE) { int (*callback) (void); @@ -1197,6 +1254,8 @@ int plugin_read_all_once (void) status = (*callback) (&rf->rf_udata); } + plugin_set_ctx (old_ctx); + if (status != 0) { NOTICE ("read-function of plugin `%s' failed.", @@ -1242,6 +1301,7 @@ int plugin_write (const char *plugin, /* {{{ */ { callback_func_t *cf = le->value; plugin_write_cb callback; + plugin_ctx_t old_ctx = plugin_set_ctx (cf->cf_ctx); DEBUG ("plugin: plugin_write: Writing values via %s.", le->key); callback = cf->cf_callback; @@ -1251,6 +1311,8 @@ int plugin_write (const char *plugin, /* {{{ */ else success++; + plugin_set_ctx (old_ctx); + le = le->next; } @@ -1263,6 +1325,7 @@ int plugin_write (const char *plugin, /* {{{ */ { callback_func_t *cf; plugin_write_cb callback; + plugin_ctx_t old_ctx; le = llist_head (list_write); while (le != NULL) @@ -1278,9 +1341,13 @@ int plugin_write (const char *plugin, /* {{{ */ cf = le->value; + old_ctx = plugin_set_ctx (cf->cf_ctx); + DEBUG ("plugin: plugin_write: Writing values via %s.", le->key); callback = cf->cf_callback; status = (*callback) (ds, vl, &cf->cf_udata); + + plugin_set_ctx (old_ctx); } return (status); @@ -1298,6 +1365,7 @@ int plugin_flush (const char *plugin, cdtime_t timeout, const char *identifier) { callback_func_t *cf; plugin_flush_cb callback; + plugin_ctx_t old_ctx; if ((plugin != NULL) && (strcmp (plugin, le->key) != 0)) @@ -1307,10 +1375,13 @@ int plugin_flush (const char *plugin, cdtime_t timeout, const char *identifier) } cf = le->value; + old_ctx = plugin_set_ctx (cf->cf_ctx); callback = cf->cf_callback; (*callback) (timeout, identifier, &cf->cf_udata); + plugin_set_ctx (old_ctx); + le = le->next; } return (0); @@ -1343,8 +1414,10 @@ void plugin_shutdown_all (void) { callback_func_t *cf; plugin_shutdown_cb callback; + plugin_ctx_t old_ctx; cf = le->value; + old_ctx = plugin_set_ctx (cf->cf_ctx); callback = cf->cf_callback; /* Advance the pointer before calling the callback allows @@ -1354,6 +1427,8 @@ void plugin_shutdown_all (void) le = le->next; (*callback) (); + + plugin_set_ctx (old_ctx); } /* Write plugins which use the `user_data' pointer usually need the @@ -1382,12 +1457,15 @@ int plugin_dispatch_missing (const value_list_t *vl) /* {{{ */ { callback_func_t *cf; plugin_missing_cb callback; + plugin_ctx_t old_ctx; int status; cf = le->value; + old_ctx = plugin_set_ctx (cf->cf_ctx); callback = cf->cf_callback; status = (*callback) (vl, &cf->cf_udata); + plugin_set_ctx (old_ctx); if (status != 0) { if (status < 0) @@ -1462,8 +1540,14 @@ int plugin_dispatch_values (value_list_t *vl) if (vl->time == 0) vl->time = cdtime (); - if (vl->interval <= 0) - vl->interval = interval_g; + if (vl->interval <= 0) { + plugin_ctx_t ctx = plugin_get_ctx (); + + if (ctx.interval != 0) + vl->interval = ctx.interval; + else + vl->interval = interval_g; + } DEBUG ("plugin_dispatch_values: time = %.3f; interval = %.3f; " "host = %s; " @@ -1651,11 +1735,14 @@ int plugin_dispatch_notification (const notification_t *notif) { callback_func_t *cf; plugin_notification_cb callback; + plugin_ctx_t old_ctx; int status; cf = le->value; + old_ctx = plugin_set_ctx (cf->cf_ctx); callback = cf->cf_callback; status = (*callback) (notif, &cf->cf_udata); + plugin_set_ctx (old_ctx); if (status != 0) { WARNING ("plugin_dispatch_notification: Notification " @@ -1696,12 +1783,15 @@ void plugin_log (int level, const char *format, ...) { callback_func_t *cf; plugin_log_cb callback; + plugin_ctx_t old_ctx; cf = le->value; + old_ctx = plugin_set_ctx (cf->cf_ctx); callback = cf->cf_callback; (*callback) (level, msg, &cf->cf_udata); + plugin_set_ctx (old_ctx); le = le->next; } } /* void plugin_log */ @@ -1894,4 +1984,136 @@ int plugin_notification_meta_free (notification_meta_t *n) return (0); } /* int plugin_notification_meta_free */ +static void plugin_ctx_destructor (void *ctx) +{ + sfree (ctx); +} /* void plugin_ctx_destructor */ + +static plugin_ctx_t ctx_init = { /* interval = */ 0 }; + +static plugin_ctx_t *plugin_ctx_create (void) +{ + plugin_ctx_t *ctx; + + ctx = malloc (sizeof (*ctx)); + if (ctx == NULL) { + char errbuf[1024]; + ERROR ("Failed to allocate plugin context: %s", + sstrerror (errno, errbuf, sizeof (errbuf))); + return NULL; + } + + *ctx = ctx_init; + assert (plugin_ctx_key_initialized); + pthread_setspecific (plugin_ctx_key, ctx); + DEBUG("Created new plugin context."); + return (ctx); +} /* int plugin_ctx_create */ + +void plugin_init_ctx (void) +{ + pthread_key_create (&plugin_ctx_key, plugin_ctx_destructor); + plugin_ctx_key_initialized = 1; +} /* void plugin_init_ctx */ + +plugin_ctx_t plugin_get_ctx (void) +{ + plugin_ctx_t *ctx; + + assert (plugin_ctx_key_initialized); + ctx = pthread_getspecific (plugin_ctx_key); + + if (ctx == NULL) { + ctx = plugin_ctx_create (); + /* this must no happen -- exit() instead? */ + if (ctx == NULL) + return ctx_init; + } + + return (*ctx); +} /* plugin_ctx_t plugin_get_ctx */ + +plugin_ctx_t plugin_set_ctx (plugin_ctx_t ctx) +{ + plugin_ctx_t *c; + plugin_ctx_t old; + + assert (plugin_ctx_key_initialized); + c = pthread_getspecific (plugin_ctx_key); + + if (c == NULL) { + c = plugin_ctx_create (); + /* this must no happen -- exit() instead? */ + if (c == NULL) + return ctx_init; + } + + old = *c; + *c = ctx; + + return (old); +} /* void plugin_set_ctx */ + +cdtime_t plugin_get_interval (void) +{ + cdtime_t interval; + + const char *interval_str; + double interval_dbl; + + interval = plugin_get_ctx().interval; + if (interval > 0) + return interval; + + /* this should happen during initialization only */ + interval_str = global_option_get ("Interval"); + if (interval_str != NULL) + { + interval_dbl = atof (interval_str); + if (interval_dbl > 0.0) + interval = DOUBLE_TO_CDTIME_T (interval_dbl); + } + + if (interval > 0) + return interval; + return TIME_T_TO_CDTIME_T (10); +} /* cdtime_t plugin_get_interval */ + +typedef struct { + plugin_ctx_t ctx; + void *(*start_routine) (void *); + void *arg; +} plugin_thread_t; + +static void *plugin_thread_start (void *arg) +{ + plugin_thread_t *plugin_thread = arg; + + void *(*start_routine) (void *) = plugin_thread->start_routine; + void *plugin_arg = plugin_thread->arg; + + plugin_set_ctx (plugin_thread->ctx); + + free (plugin_thread); + + return start_routine (plugin_arg); +} /* void *plugin_thread_start */ + +int plugin_thread_create (pthread_t *thread, const pthread_attr_t *attr, + void *(*start_routine) (void *), void *arg) +{ + plugin_thread_t *plugin_thread; + + plugin_thread = malloc (sizeof (*plugin_thread)); + if (plugin_thread == NULL) + return -1; + + plugin_thread->ctx = plugin_get_ctx (); + plugin_thread->start_routine = start_routine; + plugin_thread->arg = arg; + + return pthread_create (thread, attr, + plugin_thread_start, plugin_thread); +} /* int plugin_thread_create */ + /* vim: set sw=8 ts=8 noet fdm=marker : */