src/plugin.c

   1 /**
   2  * collectd - src/plugin.c
   3  * Copyright (C) 2005-2013  Florian octo Forster
   4  *
   5  * This program is free software; you can redistribute it and/or modify it
   6  * under the terms of the GNU General Public License as published by the
   7  * Free Software Foundation; only version 2 of the License is applicable.
   8  *
   9  * This program is distributed in the hope that it will be useful, but
  10  * WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12  * General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU General Public License along
  15  * with this program; if not, write to the Free Software Foundation, Inc.,
  16  * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
  17  *
  18  * Authors:
  19  *   Florian octo Forster <octo at collectd.org>
  20  *   Sebastian Harl <sh at tokkee.org>
  21  **/
  22
  23 #include "collectd.h"
  24 #include "utils_complain.h"
  25
  26 #include <ltdl.h>
  27
  28 #if HAVE_PTHREAD_H
  29 # include <pthread.h>
  30 #endif
  31
  32 #include "common.h"
  33 #include "plugin.h"
  34 #include "configfile.h"
  35 #include "utils_avltree.h"
  36 #include "utils_llist.h"
  37 #include "utils_heap.h"
  38 #include "utils_cache.h"
  39 #include "filter_chain.h"
  40
  41 /*
  42  * Private structures
  43  */
  44 struct callback_func_s
  45 {
  46         void *cf_callback;
  47         user_data_t cf_udata;
  48         plugin_ctx_t cf_ctx;
  49 };
  50 typedef struct callback_func_s callback_func_t;
  51
  52 #define RF_SIMPLE  0
  53 #define RF_COMPLEX 1
  54 #define RF_REMOVE  65535
  55 struct read_func_s
  56 {
  57         /* `read_func_t' "inherits" from `callback_func_t'.
  58          * The `rf_super' member MUST be the first one in this structure! */
  59 #define rf_callback rf_super.cf_callback
  60 #define rf_udata rf_super.cf_udata
  61 #define rf_ctx rf_super.cf_ctx
  62         callback_func_t rf_super;
  63         char rf_group[DATA_MAX_NAME_LEN];
  64         char rf_name[DATA_MAX_NAME_LEN];
  65         int rf_type;
  66         struct timespec rf_interval;
  67         struct timespec rf_effective_interval;
  68         struct timespec rf_next_read;
  69 };
  70 typedef struct read_func_s read_func_t;
  71
  72 struct write_queue_s;
  73 typedef struct write_queue_s write_queue_t;
  74 struct write_queue_s
  75 {
  76         value_list_t *vl;
  77         write_queue_t *next;
  78 };
  79
  80 /*
  81  * Private variables
  82  */
  83 static llist_t *list_init;
  84 static llist_t *list_write;
  85 static llist_t *list_flush;
  86 static llist_t *list_missing;
  87 static llist_t *list_shutdown;
  88 static llist_t *list_log;
  89 static llist_t *list_notification;
  90
  91 static fc_chain_t *pre_cache_chain = NULL;
  92 static fc_chain_t *post_cache_chain = NULL;
  93
  94 static c_avl_tree_t *data_sets;
  95
  96 static char *plugindir = NULL;
  97
  98 static c_heap_t       *read_heap = NULL;
  99 static llist_t        *read_list;
 100 static int             read_loop = 1;
 101 static pthread_mutex_t read_lock = PTHREAD_MUTEX_INITIALIZER;
 102 static pthread_cond_t  read_cond = PTHREAD_COND_INITIALIZER;
 103 static pthread_t      *read_threads = NULL;
 104 static int             read_threads_num = 0;
 105
 106 static write_queue_t  *write_queue_head;
 107 static write_queue_t  *write_queue_tail;
 108 static _Bool           write_loop = 1;
 109 static pthread_mutex_t write_lock = PTHREAD_MUTEX_INITIALIZER;
 110 static pthread_cond_t  write_cond = PTHREAD_COND_INITIALIZER;
 111 static pthread_t      *write_threads = NULL;
 112 static size_t          write_threads_num = 0;
 113
 114 static pthread_key_t   plugin_ctx_key;
 115 static _Bool           plugin_ctx_key_initialized = 0;
 116
 117 /*
 118  * Static functions
 119  */
 120 static int plugin_dispatch_values_internal (value_list_t *vl);
 121
 122 static const char *plugin_get_dir (void)
 123 {
 124         if (plugindir == NULL)
 125                 return (PLUGINDIR);
 126         else
 127                 return (plugindir);
 128 }
 129
 130 static void destroy_callback (callback_func_t *cf) /* {{{ */
 131 {
 132         if (cf == NULL)
 133                 return;
 134
 135         if ((cf->cf_udata.data != NULL) && (cf->cf_udata.free_func != NULL))
 136         {
 137                 cf->cf_udata.free_func (cf->cf_udata.data);
 138                 cf->cf_udata.data = NULL;
 139                 cf->cf_udata.free_func = NULL;
 140         }
 141         sfree (cf);
 142 } /* }}} void destroy_callback */
 143
 144 static void destroy_all_callbacks (llist_t **list) /* {{{ */
 145 {
 146         llentry_t *le;
 147
 148         if (*list == NULL)
 149                 return;
 150
 151         le = llist_head (*list);
 152         while (le != NULL)
 153         {
 154                 llentry_t *le_next;
 155
 156                 le_next = le->next;
 157
 158                 sfree (le->key);
 159                 destroy_callback (le->value);
 160                 le->value = NULL;
 161
 162                 le = le_next;
 163         }
 164
 165         llist_destroy (*list);
 166         *list = NULL;
 167 } /* }}} void destroy_all_callbacks */
 168
 169 static void destroy_read_heap (void) /* {{{ */
 170 {
 171         if (read_heap == NULL)
 172                 return;
 173
 174         while (42)
 175         {
 176                 callback_func_t *cf;
 177
 178                 cf = c_heap_get_root (read_heap);
 179                 if (cf == NULL)
 180                         break;
 181
 182                 destroy_callback (cf);
 183         }
 184
 185         c_heap_destroy (read_heap);
 186         read_heap = NULL;
 187 } /* }}} void destroy_read_heap */
 188
 189 static int register_callback (llist_t **list, /* {{{ */
 190                 const char *name, callback_func_t *cf)
 191 {
 192         llentry_t *le;
 193         char *key;
 194
 195         if (*list == NULL)
 196         {
 197                 *list = llist_create ();
 198                 if (*list == NULL)
 199                 {
 200                         ERROR ("plugin: register_callback: "
 201                                         "llist_create failed.");
 202                         destroy_callback (cf);
 203                         return (-1);
 204                 }
 205         }
 206
 207         key = strdup (name);
 208         if (key == NULL)
 209         {
 210                 ERROR ("plugin: register_callback: strdup failed.");
 211                 destroy_callback (cf);
 212                 return (-1);
 213         }
 214
 215         le = llist_search (*list, name);
 216         if (le == NULL)
 217         {
 218                 le = llentry_create (key, cf);
 219                 if (le == NULL)
 220                 {
 221                         ERROR ("plugin: register_callback: "
 222                                         "llentry_create failed.");
 223                         free (key);
 224                         destroy_callback (cf);
 225                         return (-1);
 226                 }
 227
 228                 llist_append (*list, le);
 229         }
 230         else
 231         {
 232                 callback_func_t *old_cf;
 233
 234                 old_cf = le->value;
 235                 le->value = cf;
 236
 237                 WARNING ("plugin: register_callback: "
 238                                 "a callback named `%s' already exists - "
 239                                 "overwriting the old entry!", name);
 240
 241                 destroy_callback (old_cf);
 242                 sfree (key);
 243         }
 244
 245         return (0);
 246 } /* }}} int register_callback */
 247
 248 static int create_register_callback (llist_t **list, /* {{{ */
 249                 const char *name, void *callback, user_data_t *ud)
 250 {
 251         callback_func_t *cf;
 252
 253         cf = (callback_func_t *) malloc (sizeof (*cf));
 254         if (cf == NULL)
 255         {
 256                 ERROR ("plugin: create_register_callback: malloc failed.");
 257                 return (-1);
 258         }
 259         memset (cf, 0, sizeof (*cf));
 260
 261         cf->cf_callback = callback;
 262         if (ud == NULL)
 263         {
 264                 cf->cf_udata.data = NULL;
 265                 cf->cf_udata.free_func = NULL;
 266         }
 267         else
 268         {
 269                 cf->cf_udata = *ud;
 270         }
 271
 272         cf->cf_ctx = plugin_get_ctx ();
 273
 274         return (register_callback (list, name, cf));
 275 } /* }}} int create_register_callback */
 276
 277 static int plugin_unregister (llist_t *list, const char *name) /* {{{ */
 278 {
 279         llentry_t *e;
 280
 281         if (list == NULL)
 282                 return (-1);
 283
 284         e = llist_search (list, name);
 285         if (e == NULL)
 286                 return (-1);
 287
 288         llist_remove (list, e);
 289
 290         sfree (e->key);
 291         destroy_callback (e->value);
 292
 293         llentry_destroy (e);
 294
 295         return (0);
 296 } /* }}} int plugin_unregister */
 297
 298 /*
 299  * (Try to) load the shared object `file'. Won't complain if it isn't a shared
 300  * object, but it will bitch about a shared object not having a
 301  * ``module_register'' symbol..
 302  */
 303 static int plugin_load_file (char *file, uint32_t flags)
 304 {
 305         lt_dlhandle dlh;
 306         void (*reg_handle) (void);
 307
 308         lt_dlinit ();
 309         lt_dlerror (); /* clear errors */
 310
 311 #if LIBTOOL_VERSION == 2
 312         if (flags & PLUGIN_FLAGS_GLOBAL) {
 313                 lt_dladvise advise;
 314                 lt_dladvise_init(&advise);
 315                 lt_dladvise_global(&advise);
 316                 dlh = lt_dlopenadvise(file, advise);
 317                 lt_dladvise_destroy(&advise);
 318         } else {
 319                 dlh = lt_dlopen (file);
 320         }
 321 #else /* if LIBTOOL_VERSION == 1 */
 322         if (flags & PLUGIN_FLAGS_GLOBAL)
 323                 WARNING ("plugin_load_file: The global flag is not supported, "
 324                                 "libtool 2 is required for this.");
 325         dlh = lt_dlopen (file);
 326 #endif
 327
 328         if (dlh == NULL)
 329         {
 330                 char errbuf[1024] = "";
 331
 332                 ssnprintf (errbuf, sizeof (errbuf),
 333                                 "lt_dlopen (\"%s\") failed: %s. "
 334                                 "The most common cause for this problem are "
 335                                 "missing dependencies. Use ldd(1) to check "
 336                                 "the dependencies of the plugin "
 337                                 "/ shared object.",
 338                                 file, lt_dlerror ());
 339
 340                 ERROR ("%s", errbuf);
 341                 /* Make sure this is printed to STDERR in any case, but also
 342                  * make sure it's printed only once. */
 343                 if (list_log != NULL)
 344                         fprintf (stderr, "ERROR: %s\n", errbuf);
 345
 346                 return (1);
 347         }
 348
 349         if ((reg_handle = (void (*) (void)) lt_dlsym (dlh, "module_register")) == NULL)
 350         {
 351                 WARNING ("Couldn't find symbol \"module_register\" in \"%s\": %s\n",
 352                                 file, lt_dlerror ());
 353                 lt_dlclose (dlh);
 354                 return (-1);
 355         }
 356
 357         (*reg_handle) ();
 358
 359         return (0);
 360 }
 361
 362 static _Bool timeout_reached(struct timespec timeout)
 363 {
 364         struct timeval now;
 365         gettimeofday(&now, NULL);
 366         return (now.tv_sec >= timeout.tv_sec && now.tv_usec >= (timeout.tv_nsec / 1000));
 367 }
 368
 369 static void *plugin_read_thread (void __attribute__((unused)) *args)
 370 {
 371         while (read_loop != 0)
 372         {
 373                 read_func_t *rf;
 374                 plugin_ctx_t old_ctx;
 375                 cdtime_t now;
 376                 int status;
 377                 int rf_type;
 378                 int rc;
 379
 380                 /* Get the read function that needs to be read next.
 381                  * We don't need to hold "read_lock" for the heap, but we need
 382                  * to call c_heap_get_root() and pthread_cond_wait() in the
 383                  * same protected block. */
 384                 pthread_mutex_lock (&read_lock);
 385                 rf = c_heap_get_root (read_heap);
 386                 if (rf == NULL)
 387                 {
 388                         pthread_cond_wait (&read_cond, &read_lock);
 389                         pthread_mutex_unlock (&read_lock);
 390                         continue;
 391                 }
 392                 pthread_mutex_unlock (&read_lock);
 393
 394                 if ((rf->rf_interval.tv_sec == 0) && (rf->rf_interval.tv_nsec == 0))
 395                 {
 396                         /* this should not happen, because the interval is set
 397                          * for each plugin when loading it
 398                          * XXX: issue a warning? */
 399                         now = cdtime ();
 400
 401                         CDTIME_T_TO_TIMESPEC (plugin_get_interval (), &rf->rf_interval);
 402
 403                         rf->rf_effective_interval = rf->rf_interval;
 404
 405                         CDTIME_T_TO_TIMESPEC (now, &rf->rf_next_read);
 406                 }
 407
 408                 /* sleep until this entry is due,
 409                  * using pthread_cond_timedwait */
 410                 pthread_mutex_lock (&read_lock);
 411                 /* In pthread_cond_timedwait, spurious wakeups are possible
 412                  * (and really happen, at least on NetBSD with > 1 CPU), thus
 413                  * we need to re-evaluate the condition every time
 414                  * pthread_cond_timedwait returns. */
 415                 rc = 0;
 416                 while ((read_loop != 0)
 417                                 && !timeout_reached(rf->rf_next_read)
 418                                 && rc == 0)
 419                 {
 420                         rc = pthread_cond_timedwait (&read_cond, &read_lock,
 421                                 &rf->rf_next_read);
 422                 }
 423
 424                 /* Must hold `read_lock' when accessing `rf->rf_type'. */
 425                 rf_type = rf->rf_type;
 426                 pthread_mutex_unlock (&read_lock);
 427
 428                 /* Check if we're supposed to stop.. This may have interrupted
 429                  * the sleep, too. */
 430                 if (read_loop == 0)
 431                 {
 432                         /* Insert `rf' again, so it can be free'd correctly */
 433                         c_heap_insert (read_heap, rf);
 434                         break;
 435                 }
 436
 437                 /* The entry has been marked for deletion. The linked list
 438                  * entry has already been removed by `plugin_unregister_read'.
 439                  * All we have to do here is free the `read_func_t' and
 440                  * continue. */
 441                 if (rf_type == RF_REMOVE)
 442                 {
 443                         DEBUG ("plugin_read_thread: Destroying the `%s' "
 444                                         "callback.", rf->rf_name);
 445                         destroy_callback ((callback_func_t *) rf);
 446                         rf = NULL;
 447                         continue;
 448                 }
 449
 450                 DEBUG ("plugin_read_thread: Handling `%s'.", rf->rf_name);
 451
 452                 old_ctx = plugin_set_ctx (rf->rf_ctx);
 453
 454                 if (rf_type == RF_SIMPLE)
 455                 {
 456                         int (*callback) (void);
 457
 458                         callback = rf->rf_callback;
 459                         status = (*callback) ();
 460                 }
 461                 else
 462                 {
 463                         plugin_read_cb callback;
 464
 465                         assert (rf_type == RF_COMPLEX);
 466
 467                         callback = rf->rf_callback;
 468                         status = (*callback) (&rf->rf_udata);
 469                 }
 470
 471                 plugin_set_ctx (old_ctx);
 472
 473                 /* If the function signals failure, we will increase the
 474                  * intervals in which it will be called. */
 475                 if (status != 0)
 476                 {
 477                         rf->rf_effective_interval.tv_sec *= 2;
 478                         rf->rf_effective_interval.tv_nsec *= 2;
 479                         NORMALIZE_TIMESPEC (rf->rf_effective_interval);
 480
 481                         if (rf->rf_effective_interval.tv_sec >= 86400)
 482                         {
 483                                 rf->rf_effective_interval.tv_sec = 86400;
 484                                 rf->rf_effective_interval.tv_nsec = 0;
 485                         }
 486
 487                         NOTICE ("read-function of plugin `%s' failed. "
 488                                         "Will suspend it for %i seconds.",
 489                                         rf->rf_name,
 490                                         (int) rf->rf_effective_interval.tv_sec);
 491                 }
 492                 else
 493                 {
 494                         /* Success: Restore the interval, if it was changed. */
 495                         rf->rf_effective_interval = rf->rf_interval;
 496                 }
 497
 498                 /* update the ``next read due'' field */
 499                 now = cdtime ();
 500
 501                 DEBUG ("plugin_read_thread: Effective interval of the "
 502                                 "%s plugin is %i.%09i.",
 503                                 rf->rf_name,
 504                                 (int) rf->rf_effective_interval.tv_sec,
 505                                 (int) rf->rf_effective_interval.tv_nsec);
 506
 507                 /* Calculate the next (absolute) time at which this function
 508                  * should be called. */
 509                 rf->rf_next_read.tv_sec = rf->rf_next_read.tv_sec
 510                         + rf->rf_effective_interval.tv_sec;
 511                 rf->rf_next_read.tv_nsec = rf->rf_next_read.tv_nsec
 512                         + rf->rf_effective_interval.tv_nsec;
 513                 NORMALIZE_TIMESPEC (rf->rf_next_read);
 514
 515                 /* Check, if `rf_next_read' is in the past. */
 516                 if (TIMESPEC_TO_CDTIME_T (&rf->rf_next_read) < now)
 517                 {
 518                         /* `rf_next_read' is in the past. Insert `now'
 519                          * so this value doesn't trail off into the
 520                          * past too much. */
 521                         CDTIME_T_TO_TIMESPEC (now, &rf->rf_next_read);
 522                 }
 523
 524                 DEBUG ("plugin_read_thread: Next read of the %s plugin at %i.%09i.",
 525                                 rf->rf_name,
 526                                 (int) rf->rf_next_read.tv_sec,
 527                                 (int) rf->rf_next_read.tv_nsec);
 528
 529                 /* Re-insert this read function into the heap again. */
 530                 c_heap_insert (read_heap, rf);
 531         } /* while (read_loop) */
 532
 533         pthread_exit (NULL);
 534         return ((void *) 0);
 535 } /* void *plugin_read_thread */
 536
 537 static void start_read_threads (int num)
 538 {
 539         int i;
 540
 541         if (read_threads != NULL)
 542                 return;
 543
 544         read_threads = (pthread_t *) calloc (num, sizeof (pthread_t));
 545         if (read_threads == NULL)
 546         {
 547                 ERROR ("plugin: start_read_threads: calloc failed.");
 548                 return;
 549         }
 550
 551         read_threads_num = 0;
 552         for (i = 0; i < num; i++)
 553         {
 554                 if (pthread_create (read_threads + read_threads_num, NULL,
 555                                         plugin_read_thread, NULL) == 0)
 556                 {
 557                         read_threads_num++;
 558                 }
 559                 else
 560                 {
 561                         ERROR ("plugin: start_read_threads: pthread_create failed.");
 562                         return;
 563                 }
 564         } /* for (i) */
 565 } /* void start_read_threads */
 566
 567 static void stop_read_threads (void)
 568 {
 569         int i;
 570
 571         if (read_threads == NULL)
 572                 return;
 573
 574         INFO ("collectd: Stopping %i read threads.", read_threads_num);
 575
 576         pthread_mutex_lock (&read_lock);
 577         read_loop = 0;
 578         DEBUG ("plugin: stop_read_threads: Signalling `read_cond'");
 579         pthread_cond_broadcast (&read_cond);
 580         pthread_mutex_unlock (&read_lock);
 581
 582         for (i = 0; i < read_threads_num; i++)
 583         {
 584                 if (pthread_join (read_threads[i], NULL) != 0)
 585                 {
 586                         ERROR ("plugin: stop_read_threads: pthread_join failed.");
 587                 }
 588                 read_threads[i] = (pthread_t) 0;
 589         }
 590         sfree (read_threads);
 591         read_threads_num = 0;
 592 } /* void stop_read_threads */
 593
 594 static void plugin_value_list_free (value_list_t *vl) /* {{{ */
 595 {
 596         if (vl == NULL)
 597                 return;
 598
 599         meta_data_destroy (vl->meta);
 600         sfree (vl->values);
 601         sfree (vl);
 602 } /* }}} void plugin_value_list_free */
 603
 604 static value_list_t *plugin_value_list_clone (value_list_t const *vl_orig) /* {{{ */
 605 {
 606         value_list_t *vl;
 607
 608         if (vl_orig == NULL)
 609                 return (NULL);
 610
 611         vl = malloc (sizeof (*vl));
 612         if (vl == NULL)
 613                 return (NULL);
 614         memcpy (vl, vl_orig, sizeof (*vl));
 615
 616         vl->values = calloc (vl_orig->values_len, sizeof (*vl->values));
 617         if (vl->values == NULL)
 618         {
 619                 plugin_value_list_free (vl);
 620                 return (NULL);
 621         }
 622         memcpy (vl->values, vl_orig->values,
 623                         vl_orig->values_len * sizeof (*vl->values));
 624
 625         vl->meta = meta_data_clone (vl->meta);
 626         if ((vl_orig->meta != NULL) && (vl->meta == NULL))
 627         {
 628                 plugin_value_list_free (vl);
 629                 return (NULL);
 630         }
 631
 632         if (vl->time == 0)
 633                 vl->time = cdtime ();
 634
 635         /* Once this gets dequeued by a write thread, we don't have access to
 636          * the thread context anymore. We therefore fill in the interval here,
 637          * if required. An alternative would be to copy the context and clone
 638          * the context in the write thread, but that seems overly complicated
 639          * for the interval alone. */
 640         if (vl->interval == 0)
 641         {
 642                 plugin_ctx_t ctx = plugin_get_ctx ();
 643
 644                 if (ctx.interval != 0)
 645                         vl->interval = ctx.interval;
 646                 else
 647                 {
 648                         char name[6 * DATA_MAX_NAME_LEN];
 649                         FORMAT_VL (name, sizeof (name), vl);
 650                         ERROR ("plugin_value_list_clone: Unable to determine "
 651                                         "interval from context for "
 652                                         "value list \"%s\". "
 653                                         "This indicates a broken plugin. "
 654                                         "Please report this problem to the "
 655                                         "collectd mailing list or at "
 656                                         "<http://collectd.org/bugs/>.", name);
 657                         vl->interval = cf_get_default_interval ();
 658                 }
 659         }
 660
 661         return (vl);
 662 } /* }}} value_list_t *plugin_value_list_clone */
 663
 664 static int plugin_write_enqueue (value_list_t const *vl) /* {{{ */
 665 {
 666         write_queue_t *q;
 667
 668         q = malloc (sizeof (*q));
 669         if (q == NULL)
 670                 return (ENOMEM);
 671         q->next = NULL;
 672
 673         q->vl = plugin_value_list_clone (vl);
 674         if (q->vl == NULL)
 675         {
 676                 sfree (q);
 677                 return (ENOMEM);
 678         }
 679
 680         pthread_mutex_lock (&write_lock);
 681
 682         if (write_queue_tail == NULL)
 683         {
 684                 write_queue_head = q;
 685                 write_queue_tail = q;
 686         }
 687         else
 688         {
 689                 write_queue_tail->next = q;
 690                 write_queue_tail = q;
 691         }
 692
 693         pthread_cond_signal (&write_cond);
 694         pthread_mutex_unlock (&write_lock);
 695
 696         return (0);
 697 } /* }}} int plugin_write_enqueue */
 698
 699 static value_list_t *plugin_write_dequeue (void) /* {{{ */
 700 {
 701         write_queue_t *q;
 702         value_list_t *vl;
 703
 704         pthread_mutex_lock (&write_lock);
 705
 706         while (write_loop && (write_queue_head == NULL))
 707                 pthread_cond_wait (&write_cond, &write_lock);
 708
 709         if (write_queue_head == NULL)
 710         {
 711                 pthread_mutex_unlock (&write_lock);
 712                 return (NULL);
 713         }
 714
 715         q = write_queue_head;
 716         write_queue_head = q->next;
 717         if (write_queue_head == NULL)
 718                 write_queue_tail = NULL;
 719
 720         pthread_mutex_unlock (&write_lock);
 721
 722         vl = q->vl;
 723         sfree (q);
 724         return (vl);
 725 } /* }}} value_list_t *plugin_write_dequeue */
 726
 727 static void *plugin_write_thread (void __attribute__((unused)) *args) /* {{{ */
 728 {
 729         while (write_loop)
 730         {
 731                 value_list_t *vl = plugin_write_dequeue ();
 732                 if (vl == NULL)
 733                         continue;
 734
 735                 plugin_dispatch_values_internal (vl);
 736
 737                 plugin_value_list_free (vl);
 738         }
 739
 740         pthread_exit (NULL);
 741         return ((void *) 0);
 742 } /* }}} void *plugin_write_thread */
 743
 744 static void start_write_threads (size_t num) /* {{{ */
 745 {
 746         size_t i;
 747
 748         if (write_threads != NULL)
 749                 return;
 750
 751         write_threads = (pthread_t *) calloc (num, sizeof (pthread_t));
 752         if (write_threads == NULL)
 753         {
 754                 ERROR ("plugin: start_write_threads: calloc failed.");
 755                 return;
 756         }
 757
 758         write_threads_num = 0;
 759         for (i = 0; i < num; i++)
 760         {
 761                 int status;
 762
 763                 status = pthread_create (write_threads + write_threads_num,
 764                                 /* attr = */ NULL,
 765                                 plugin_write_thread,
 766                                 /* arg = */ NULL);
 767                 if (status != 0)
 768                 {
 769                         char errbuf[1024];
 770                         ERROR ("plugin: start_write_threads: pthread_create failed "
 771                                         "with status %i (%s).", status,
 772                                         sstrerror (status, errbuf, sizeof (errbuf)));
 773                         return;
 774                 }
 775
 776                 write_threads_num++;
 777         } /* for (i) */
 778 } /* }}} void start_write_threads */
 779
 780 static void stop_write_threads (void) /* {{{ */
 781 {
 782         write_queue_t *q;
 783         int i;
 784
 785         if (write_threads == NULL)
 786                 return;
 787
 788         INFO ("collectd: Stopping %zu write threads.", write_threads_num);
 789
 790         pthread_mutex_lock (&write_lock);
 791         write_loop = 0;
 792         DEBUG ("plugin: stop_write_threads: Signalling `write_cond'");
 793         pthread_cond_broadcast (&write_cond);
 794         pthread_mutex_unlock (&write_lock);
 795
 796         for (i = 0; i < write_threads_num; i++)
 797         {
 798                 if (pthread_join (write_threads[i], NULL) != 0)
 799                 {
 800                         ERROR ("plugin: stop_write_threads: pthread_join failed.");
 801                 }
 802                 write_threads[i] = (pthread_t) 0;
 803         }
 804         sfree (write_threads);
 805         write_threads_num = 0;
 806
 807         pthread_mutex_lock (&write_lock);
 808         i = 0;
 809         for (q = write_queue_head; q != NULL; q = q->next)
 810         {
 811                 plugin_value_list_free (q->vl);
 812                 sfree (q);
 813                 i++;
 814         }
 815         write_queue_head = NULL;
 816         write_queue_tail = NULL;
 817         pthread_mutex_unlock (&write_lock);
 818
 819         if (i > 0)
 820         {
 821                 WARNING ("plugin: %i value list%s left after shutting down "
 822                                 "the write threads.",
 823                                 i, (i == 1) ? " was" : "s were");
 824         }
 825 } /* }}} void stop_write_threads */
 826
 827 /*
 828  * Public functions
 829  */
 830 void plugin_set_dir (const char *dir)
 831 {
 832         if (plugindir != NULL)
 833                 free (plugindir);
 834
 835         if (dir == NULL)
 836                 plugindir = NULL;
 837         else if ((plugindir = strdup (dir)) == NULL)
 838         {
 839                 char errbuf[1024];
 840                 ERROR ("strdup failed: %s",
 841                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 842         }
 843 }
 844
 845 #define BUFSIZE 512
 846 int plugin_load (const char *type, uint32_t flags)
 847 {
 848         DIR  *dh;
 849         const char *dir;
 850         char  filename[BUFSIZE] = "";
 851         char  typename[BUFSIZE];
 852         int   typename_len;
 853         int   ret;
 854         struct stat    statbuf;
 855         struct dirent *de;
 856         int status;
 857
 858         dir = plugin_get_dir ();
 859         ret = 1;
 860
 861         /* `cpu' should not match `cpufreq'. To solve this we add `.so' to the
 862          * type when matching the filename */
 863         status = ssnprintf (typename, sizeof (typename), "%s.so", type);
 864         if ((status < 0) || ((size_t) status >= sizeof (typename)))
 865         {
 866                 WARNING ("plugin_load: Filename too long: \"%s.so\"", type);
 867                 return (-1);
 868         }
 869         typename_len = strlen (typename);
 870
 871         if ((dh = opendir (dir)) == NULL)
 872         {
 873                 char errbuf[1024];
 874                 ERROR ("plugin_load: opendir (%s) failed: %s", dir,
 875                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 876                 return (-1);
 877         }
 878
 879         while ((de = readdir (dh)) != NULL)
 880         {
 881                 if (strncasecmp (de->d_name, typename, typename_len))
 882                         continue;
 883
 884                 status = ssnprintf (filename, sizeof (filename),
 885                                 "%s/%s", dir, de->d_name);
 886                 if ((status < 0) || ((size_t) status >= sizeof (filename)))
 887                 {
 888                         WARNING ("plugin_load: Filename too long: \"%s/%s\"",
 889                                         dir, de->d_name);
 890                         continue;
 891                 }
 892
 893                 if (lstat (filename, &statbuf) == -1)
 894                 {
 895                         char errbuf[1024];
 896                         WARNING ("plugin_load: stat (\"%s\") failed: %s",
 897                                         filename,
 898                                         sstrerror (errno, errbuf, sizeof (errbuf)));
 899                         continue;
 900                 }
 901                 else if (!S_ISREG (statbuf.st_mode))
 902                 {
 903                         /* don't follow symlinks */
 904                         WARNING ("plugin_load: %s is not a regular file.",
 905                                         filename);
 906                         continue;
 907                 }
 908
 909                 status = plugin_load_file (filename, flags);
 910                 if (status == 0)
 911                 {
 912                         /* success */
 913                         ret = 0;
 914                         break;
 915                 }
 916                 else
 917                 {
 918                         ERROR ("plugin_load: Load plugin \"%s\" failed with "
 919                                         "status %i.", type, status);
 920                 }
 921         }
 922
 923         closedir (dh);
 924
 925         if (filename[0] == 0)
 926                 ERROR ("plugin_load: Could not find plugin \"%s\" in %s",
 927                                 type, dir);
 928
 929         return (ret);
 930 }
 931
 932 /*
 933  * The `register_*' functions follow
 934  */
 935 int plugin_register_config (const char *name,
 936                 int (*callback) (const char *key, const char *val),
 937                 const char **keys, int keys_num)
 938 {
 939         cf_register (name, callback, keys, keys_num);
 940         return (0);
 941 } /* int plugin_register_config */
 942
 943 int plugin_register_complex_config (const char *type,
 944                 int (*callback) (oconfig_item_t *))
 945 {
 946         return (cf_register_complex (type, callback));
 947 } /* int plugin_register_complex_config */
 948
 949 int plugin_register_init (const char *name,
 950                 int (*callback) (void))
 951 {
 952         return (create_register_callback (&list_init, name, (void *) callback,
 953                                 /* user_data = */ NULL));
 954 } /* plugin_register_init */
 955
 956 static int plugin_compare_read_func (const void *arg0, const void *arg1)
 957 {
 958         const read_func_t *rf0;
 959         const read_func_t *rf1;
 960
 961         rf0 = arg0;
 962         rf1 = arg1;
 963
 964         if (rf0->rf_next_read.tv_sec < rf1->rf_next_read.tv_sec)
 965                 return (-1);
 966         else if (rf0->rf_next_read.tv_sec > rf1->rf_next_read.tv_sec)
 967                 return (1);
 968         else if (rf0->rf_next_read.tv_nsec < rf1->rf_next_read.tv_nsec)
 969                 return (-1);
 970         else if (rf0->rf_next_read.tv_nsec > rf1->rf_next_read.tv_nsec)
 971                 return (1);
 972         else
 973                 return (0);
 974 } /* int plugin_compare_read_func */
 975
 976 /* Add a read function to both, the heap and a linked list. The linked list if
 977  * used to look-up read functions, especially for the remove function. The heap
 978  * is used to determine which plugin to read next. */
 979 static int plugin_insert_read (read_func_t *rf)
 980 {
 981         int status;
 982         llentry_t *le;
 983
 984         pthread_mutex_lock (&read_lock);
 985
 986         if (read_list == NULL)
 987         {
 988                 read_list = llist_create ();
 989                 if (read_list == NULL)
 990                 {
 991                         pthread_mutex_unlock (&read_lock);
 992                         ERROR ("plugin_insert_read: read_list failed.");
 993                         return (-1);
 994                 }
 995         }
 996
 997         if (read_heap == NULL)
 998         {
 999                 read_heap = c_heap_create (plugin_compare_read_func);
1000                 if (read_heap == NULL)
1001                 {
1002                         pthread_mutex_unlock (&read_lock);
1003                         ERROR ("plugin_insert_read: c_heap_create failed.");
1004                         return (-1);
1005                 }
1006         }
1007
1008         le = llist_search (read_list, rf->rf_name);
1009         if (le != NULL)
1010         {
1011                 pthread_mutex_unlock (&read_lock);
1012                 WARNING ("The read function \"%s\" is already registered. "
1013                                 "Check for duplicate \"LoadPlugin\" lines "
1014                                 "in your configuration!",
1015                                 rf->rf_name);
1016                 return (EINVAL);
1017         }
1018
1019         le = llentry_create (rf->rf_name, rf);
1020         if (le == NULL)
1021         {
1022                 pthread_mutex_unlock (&read_lock);
1023                 ERROR ("plugin_insert_read: llentry_create failed.");
1024                 return (-1);
1025         }
1026
1027         status = c_heap_insert (read_heap, rf);
1028         if (status != 0)
1029         {
1030                 pthread_mutex_unlock (&read_lock);
1031                 ERROR ("plugin_insert_read: c_heap_insert failed.");
1032                 llentry_destroy (le);
1033                 return (-1);
1034         }
1035
1036         /* This does not fail. */
1037         llist_append (read_list, le);
1038
1039         /* Wake up all the read threads. */
1040         pthread_cond_broadcast (&read_cond);
1041         pthread_mutex_unlock (&read_lock);
1042         return (0);
1043 } /* int plugin_insert_read */
1044
1045 static int read_cb_wrapper (user_data_t *ud)
1046 {
1047         int (*callback) (void);
1048
1049         if (ud == NULL)
1050                 return -1;
1051
1052         callback = ud->data;
1053         return callback();
1054 } /* int read_cb_wrapper */
1055
1056 int plugin_register_read (const char *name,
1057                 int (*callback) (void))
1058 {
1059         read_func_t *rf;
1060         plugin_ctx_t ctx = plugin_get_ctx ();
1061         int status;
1062
1063         if (ctx.interval != 0) {
1064                 /* If ctx.interval is not zero (== use the plugin or global
1065                  * interval), we need to use the "complex" read callback,
1066                  * because only that allows to specify a different interval.
1067                  * Wrap the callback using read_cb_wrapper(). */
1068                 struct timespec interval;
1069                 user_data_t user_data;
1070
1071                 user_data.data = callback;
1072                 user_data.free_func = NULL;
1073
1074                 CDTIME_T_TO_TIMESPEC (ctx.interval, &interval);
1075                 return plugin_register_complex_read (/* group = */ NULL,
1076                                 name, read_cb_wrapper, &interval, &user_data);
1077         }
1078
1079         DEBUG ("plugin_register_read: default_interval = %.3f",
1080                         CDTIME_T_TO_DOUBLE(plugin_get_interval ()));
1081
1082         rf = malloc (sizeof (*rf));
1083         if (rf == NULL)
1084         {
1085                 ERROR ("plugin_register_read: malloc failed.");
1086                 return (ENOMEM);
1087         }
1088
1089         memset (rf, 0, sizeof (read_func_t));
1090         rf->rf_callback = (void *) callback;
1091         rf->rf_udata.data = NULL;
1092         rf->rf_udata.free_func = NULL;
1093         rf->rf_ctx = ctx;
1094         rf->rf_group[0] = '\0';
1095         sstrncpy (rf->rf_name, name, sizeof (rf->rf_name));
1096         rf->rf_type = RF_SIMPLE;
1097         rf->rf_interval.tv_sec = 0;
1098         rf->rf_interval.tv_nsec = 0;
1099         rf->rf_effective_interval = rf->rf_interval;
1100
1101         status = plugin_insert_read (rf);
1102         if (status != 0)
1103                 sfree (rf);
1104
1105         return (status);
1106 } /* int plugin_register_read */
1107
1108 int plugin_register_complex_read (const char *group, const char *name,
1109                 plugin_read_cb callback,
1110                 const struct timespec *interval,
1111                 user_data_t *user_data)
1112 {
1113         read_func_t *rf;
1114         plugin_ctx_t ctx = plugin_get_ctx ();
1115         int status;
1116
1117         rf = malloc (sizeof (*rf));
1118         if (rf == NULL)
1119         {
1120                 ERROR ("plugin_register_complex_read: malloc failed.");
1121                 return (ENOMEM);
1122         }
1123
1124         memset (rf, 0, sizeof (read_func_t));
1125         rf->rf_callback = (void *) callback;
1126         if (group != NULL)
1127                 sstrncpy (rf->rf_group, group, sizeof (rf->rf_group));
1128         else
1129                 rf->rf_group[0] = '\0';
1130         sstrncpy (rf->rf_name, name, sizeof (rf->rf_name));
1131         rf->rf_type = RF_COMPLEX;
1132         if (interval != NULL)
1133         {
1134                 rf->rf_interval = *interval;
1135         }
1136         else if (ctx.interval != 0)
1137         {
1138                 CDTIME_T_TO_TIMESPEC (ctx.interval, &rf->rf_interval);
1139         }
1140         rf->rf_effective_interval = rf->rf_interval;
1141
1142         DEBUG ("plugin_register_read: interval = %i.%09i",
1143                         (int) rf->rf_interval.tv_sec,
1144                         (int) rf->rf_interval.tv_nsec);
1145
1146         /* Set user data */
1147         if (user_data == NULL)
1148         {
1149                 rf->rf_udata.data = NULL;
1150                 rf->rf_udata.free_func = NULL;
1151         }
1152         else
1153         {
1154                 rf->rf_udata = *user_data;
1155         }
1156
1157         rf->rf_ctx = ctx;
1158
1159         status = plugin_insert_read (rf);
1160         if (status != 0)
1161                 sfree (rf);
1162
1163         return (status);
1164 } /* int plugin_register_complex_read */
1165
1166 int plugin_register_write (const char *name,
1167                 plugin_write_cb callback, user_data_t *ud)
1168 {
1169         return (create_register_callback (&list_write, name,
1170                                 (void *) callback, ud));
1171 } /* int plugin_register_write */
1172
1173 int plugin_register_flush (const char *name,
1174                 plugin_flush_cb callback, user_data_t *ud)
1175 {
1176         return (create_register_callback (&list_flush, name,
1177                                 (void *) callback, ud));
1178 } /* int plugin_register_flush */
1179
1180 int plugin_register_missing (const char *name,
1181                 plugin_missing_cb callback, user_data_t *ud)
1182 {
1183         return (create_register_callback (&list_missing, name,
1184                                 (void *) callback, ud));
1185 } /* int plugin_register_missing */
1186
1187 int plugin_register_shutdown (const char *name,
1188                 int (*callback) (void))
1189 {
1190         return (create_register_callback (&list_shutdown, name,
1191                                 (void *) callback, /* user_data = */ NULL));
1192 } /* int plugin_register_shutdown */
1193
1194 int plugin_register_data_set (const data_set_t *ds)
1195 {
1196         data_set_t *ds_copy;
1197         int i;
1198
1199         if ((data_sets != NULL)
1200                         && (c_avl_get (data_sets, ds->type, NULL) == 0))
1201         {
1202                 NOTICE ("Replacing DS `%s' with another version.", ds->type);
1203                 plugin_unregister_data_set (ds->type);
1204         }
1205         else if (data_sets == NULL)
1206         {
1207                 data_sets = c_avl_create ((int (*) (const void *, const void *)) strcmp);
1208                 if (data_sets == NULL)
1209                         return (-1);
1210         }
1211
1212         ds_copy = (data_set_t *) malloc (sizeof (data_set_t));
1213         if (ds_copy == NULL)
1214                 return (-1);
1215         memcpy(ds_copy, ds, sizeof (data_set_t));
1216
1217         ds_copy->ds = (data_source_t *) malloc (sizeof (data_source_t)
1218                         * ds->ds_num);
1219         if (ds_copy->ds == NULL)
1220         {
1221                 free (ds_copy);
1222                 return (-1);
1223         }
1224
1225         for (i = 0; i < ds->ds_num; i++)
1226                 memcpy (ds_copy->ds + i, ds->ds + i, sizeof (data_source_t));
1227
1228         return (c_avl_insert (data_sets, (void *) ds_copy->type, (void *) ds_copy));
1229 } /* int plugin_register_data_set */
1230
1231 int plugin_register_log (const char *name,
1232                 plugin_log_cb callback, user_data_t *ud)
1233 {
1234         return (create_register_callback (&list_log, name,
1235                                 (void *) callback, ud));
1236 } /* int plugin_register_log */
1237
1238 int plugin_register_notification (const char *name,
1239                 plugin_notification_cb callback, user_data_t *ud)
1240 {
1241         return (create_register_callback (&list_notification, name,
1242                                 (void *) callback, ud));
1243 } /* int plugin_register_log */
1244
1245 int plugin_unregister_config (const char *name)
1246 {
1247         cf_unregister (name);
1248         return (0);
1249 } /* int plugin_unregister_config */
1250
1251 int plugin_unregister_complex_config (const char *name)
1252 {
1253         cf_unregister_complex (name);
1254         return (0);
1255 } /* int plugin_unregister_complex_config */
1256
1257 int plugin_unregister_init (const char *name)
1258 {
1259         return (plugin_unregister (list_init, name));
1260 }
1261
1262 int plugin_unregister_read (const char *name) /* {{{ */
1263 {
1264         llentry_t *le;
1265         read_func_t *rf;
1266
1267         if (name == NULL)
1268                 return (-ENOENT);
1269
1270         pthread_mutex_lock (&read_lock);
1271
1272         if (read_list == NULL)
1273         {
1274                 pthread_mutex_unlock (&read_lock);
1275                 return (-ENOENT);
1276         }
1277
1278         le = llist_search (read_list, name);
1279         if (le == NULL)
1280         {
1281                 pthread_mutex_unlock (&read_lock);
1282                 WARNING ("plugin_unregister_read: No such read function: %s",
1283                                 name);
1284                 return (-ENOENT);
1285         }
1286
1287         llist_remove (read_list, le);
1288
1289         rf = le->value;
1290         assert (rf != NULL);
1291         rf->rf_type = RF_REMOVE;
1292
1293         pthread_mutex_unlock (&read_lock);
1294
1295         llentry_destroy (le);
1296
1297         DEBUG ("plugin_unregister_read: Marked `%s' for removal.", name);
1298
1299         return (0);
1300 } /* }}} int plugin_unregister_read */
1301
1302 static int compare_read_func_group (llentry_t *e, void *ud) /* {{{ */
1303 {
1304         read_func_t *rf    = e->value;
1305         char        *group = ud;
1306
1307         return strcmp (rf->rf_group, (const char *)group);
1308 } /* }}} int compare_read_func_group */
1309
1310 int plugin_unregister_read_group (const char *group) /* {{{ */
1311 {
1312         llentry_t *le;
1313         read_func_t *rf;
1314
1315         int found = 0;
1316
1317         if (group == NULL)
1318                 return (-ENOENT);
1319
1320         pthread_mutex_lock (&read_lock);
1321
1322         if (read_list == NULL)
1323         {
1324                 pthread_mutex_unlock (&read_lock);
1325                 return (-ENOENT);
1326         }
1327
1328         while (42)
1329         {
1330                 le = llist_search_custom (read_list,
1331                                 compare_read_func_group, (void *)group);
1332
1333                 if (le == NULL)
1334                         break;
1335
1336                 ++found;
1337
1338                 llist_remove (read_list, le);
1339
1340                 rf = le->value;
1341                 assert (rf != NULL);
1342                 rf->rf_type = RF_REMOVE;
1343
1344                 llentry_destroy (le);
1345
1346                 DEBUG ("plugin_unregister_read_group: "
1347                                 "Marked `%s' (group `%s') for removal.",
1348                                 rf->rf_name, group);
1349         }
1350
1351         pthread_mutex_unlock (&read_lock);
1352
1353         if (found == 0)
1354         {
1355                 WARNING ("plugin_unregister_read_group: No such "
1356                                 "group of read function: %s", group);
1357                 return (-ENOENT);
1358         }
1359
1360         return (0);
1361 } /* }}} int plugin_unregister_read_group */
1362
1363 int plugin_unregister_write (const char *name)
1364 {
1365         return (plugin_unregister (list_write, name));
1366 }
1367
1368 int plugin_unregister_flush (const char *name)
1369 {
1370         return (plugin_unregister (list_flush, name));
1371 }
1372
1373 int plugin_unregister_missing (const char *name)
1374 {
1375         return (plugin_unregister (list_missing, name));
1376 }
1377
1378 int plugin_unregister_shutdown (const char *name)
1379 {
1380         return (plugin_unregister (list_shutdown, name));
1381 }
1382
1383 int plugin_unregister_data_set (const char *name)
1384 {
1385         data_set_t *ds;
1386
1387         if (data_sets == NULL)
1388                 return (-1);
1389
1390         if (c_avl_remove (data_sets, name, NULL, (void *) &ds) != 0)
1391                 return (-1);
1392
1393         sfree (ds->ds);
1394         sfree (ds);
1395
1396         return (0);
1397 } /* int plugin_unregister_data_set */
1398
1399 int plugin_unregister_log (const char *name)
1400 {
1401         return (plugin_unregister (list_log, name));
1402 }
1403
1404 int plugin_unregister_notification (const char *name)
1405 {
1406         return (plugin_unregister (list_notification, name));
1407 }
1408
1409 void plugin_init_all (void)
1410 {
1411         const char *chain_name;
1412         llentry_t *le;
1413         int status;
1414
1415         /* Init the value cache */
1416         uc_init ();
1417
1418         chain_name = global_option_get ("PreCacheChain");
1419         pre_cache_chain = fc_chain_get_by_name (chain_name);
1420
1421         chain_name = global_option_get ("PostCacheChain");
1422         post_cache_chain = fc_chain_get_by_name (chain_name);
1423
1424         {
1425                 char const *tmp = global_option_get ("WriteThreads");
1426                 int num = atoi (tmp);
1427
1428                 if (num < 1)
1429                         num = 5;
1430
1431                 start_write_threads ((size_t) num);
1432         }
1433
1434         if ((list_init == NULL) && (read_heap == NULL))
1435                 return;
1436
1437         /* Calling all init callbacks before checking if read callbacks
1438          * are available allows the init callbacks to register the read
1439          * callback. */
1440         le = llist_head (list_init);
1441         while (le != NULL)
1442         {
1443                 callback_func_t *cf;
1444                 plugin_init_cb callback;
1445                 plugin_ctx_t old_ctx;
1446
1447                 cf = le->value;
1448                 old_ctx = plugin_set_ctx (cf->cf_ctx);
1449                 callback = cf->cf_callback;
1450                 status = (*callback) ();
1451                 plugin_set_ctx (old_ctx);
1452
1453                 if (status != 0)
1454                 {
1455                         ERROR ("Initialization of plugin `%s' "
1456                                         "failed with status %i. "
1457                                         "Plugin will be unloaded.",
1458                                         le->key, status);
1459                         /* Plugins that register read callbacks from the init
1460                          * callback should take care of appropriate error
1461                          * handling themselves. */
1462                         /* FIXME: Unload _all_ functions */
1463                         plugin_unregister_read (le->key);
1464                 }
1465
1466                 le = le->next;
1467         }
1468
1469         /* Start read-threads */
1470         if (read_heap != NULL)
1471         {
1472                 const char *rt;
1473                 int num;
1474                 rt = global_option_get ("ReadThreads");
1475                 num = atoi (rt);
1476                 if (num != -1)
1477                         start_read_threads ((num > 0) ? num : 5);
1478         }
1479 } /* void plugin_init_all */
1480
1481 /* TODO: Rename this function. */
1482 void plugin_read_all (void)
1483 {
1484         uc_check_timeout ();
1485
1486         return;
1487 } /* void plugin_read_all */
1488
1489 /* Read function called when the `-T' command line argument is given. */
1490 int plugin_read_all_once (void)
1491 {
1492         int status;
1493         int return_status = 0;
1494
1495         if (read_heap == NULL)
1496         {
1497                 NOTICE ("No read-functions are registered.");
1498                 return (0);
1499         }
1500
1501         while (42)
1502         {
1503                 read_func_t *rf;
1504                 plugin_ctx_t old_ctx;
1505
1506                 rf = c_heap_get_root (read_heap);
1507                 if (rf == NULL)
1508                         break;
1509
1510                 old_ctx = plugin_set_ctx (rf->rf_ctx);
1511
1512                 if (rf->rf_type == RF_SIMPLE)
1513                 {
1514                         int (*callback) (void);
1515
1516                         callback = rf->rf_callback;
1517                         status = (*callback) ();
1518                 }
1519                 else
1520                 {
1521                         plugin_read_cb callback;
1522
1523                         callback = rf->rf_callback;
1524                         status = (*callback) (&rf->rf_udata);
1525                 }
1526
1527                 plugin_set_ctx (old_ctx);
1528
1529                 if (status != 0)
1530                 {
1531                         NOTICE ("read-function of plugin `%s' failed.",
1532                                         rf->rf_name);
1533                         return_status = -1;
1534                 }
1535
1536                 destroy_callback ((void *) rf);
1537         }
1538
1539         return (return_status);
1540 } /* int plugin_read_all_once */
1541
1542 int plugin_write (const char *plugin, /* {{{ */
1543                 const data_set_t *ds, const value_list_t *vl)
1544 {
1545   llentry_t *le;
1546   int status;
1547
1548   if (vl == NULL)
1549     return (EINVAL);
1550
1551   if (list_write == NULL)
1552     return (ENOENT);
1553
1554   if (ds == NULL)
1555   {
1556     ds = plugin_get_ds (vl->type);
1557     if (ds == NULL)
1558     {
1559       ERROR ("plugin_write: Unable to lookup type `%s'.", vl->type);
1560       return (ENOENT);
1561     }
1562   }
1563
1564   if (plugin == NULL)
1565   {
1566     int success = 0;
1567     int failure = 0;
1568
1569     le = llist_head (list_write);
1570     while (le != NULL)
1571     {
1572       callback_func_t *cf = le->value;
1573       plugin_write_cb callback;
1574
1575       /* do not switch plugin context; rather keep the context (interval)
1576        * information of the calling read plugin */
1577
1578       DEBUG ("plugin: plugin_write: Writing values via %s.", le->key);
1579       callback = cf->cf_callback;
1580       status = (*callback) (ds, vl, &cf->cf_udata);
1581       if (status != 0)
1582         failure++;
1583       else
1584         success++;
1585
1586       le = le->next;
1587     }
1588
1589     if ((success == 0) && (failure != 0))
1590       status = -1;
1591     else
1592       status = 0;
1593   }
1594   else /* plugin != NULL */
1595   {
1596     callback_func_t *cf;
1597     plugin_write_cb callback;
1598
1599     le = llist_head (list_write);
1600     while (le != NULL)
1601     {
1602       if (strcasecmp (plugin, le->key) == 0)
1603         break;
1604
1605       le = le->next;
1606     }
1607
1608     if (le == NULL)
1609       return (ENOENT);
1610
1611     cf = le->value;
1612
1613     /* do not switch plugin context; rather keep the context (interval)
1614      * information of the calling read plugin */
1615
1616     DEBUG ("plugin: plugin_write: Writing values via %s.", le->key);
1617     callback = cf->cf_callback;
1618     status = (*callback) (ds, vl, &cf->cf_udata);
1619   }
1620
1621   return (status);
1622 } /* }}} int plugin_write */
1623
1624 int plugin_flush (const char *plugin, cdtime_t timeout, const char *identifier)
1625 {
1626   llentry_t *le;
1627
1628   if (list_flush == NULL)
1629     return (0);
1630
1631   le = llist_head (list_flush);
1632   while (le != NULL)
1633   {
1634     callback_func_t *cf;
1635     plugin_flush_cb callback;
1636     plugin_ctx_t old_ctx;
1637
1638     if ((plugin != NULL)
1639         && (strcmp (plugin, le->key) != 0))
1640     {
1641       le = le->next;
1642       continue;
1643     }
1644
1645     cf = le->value;
1646     old_ctx = plugin_set_ctx (cf->cf_ctx);
1647     callback = cf->cf_callback;
1648
1649     (*callback) (timeout, identifier, &cf->cf_udata);
1650
1651     plugin_set_ctx (old_ctx);
1652
1653     le = le->next;
1654   }
1655   return (0);
1656 } /* int plugin_flush */
1657
1658 void plugin_shutdown_all (void)
1659 {
1660         llentry_t *le;
1661
1662         stop_read_threads ();
1663
1664         destroy_all_callbacks (&list_init);
1665
1666         pthread_mutex_lock (&read_lock);
1667         llist_destroy (read_list);
1668         read_list = NULL;
1669         pthread_mutex_unlock (&read_lock);
1670
1671         destroy_read_heap ();
1672
1673         plugin_flush (/* plugin = */ NULL,
1674                         /* timeout = */ 0,
1675                         /* identifier = */ NULL);
1676
1677         le = NULL;
1678         if (list_shutdown != NULL)
1679                 le = llist_head (list_shutdown);
1680
1681         while (le != NULL)
1682         {
1683                 callback_func_t *cf;
1684                 plugin_shutdown_cb callback;
1685                 plugin_ctx_t old_ctx;
1686
1687                 cf = le->value;
1688                 old_ctx = plugin_set_ctx (cf->cf_ctx);
1689                 callback = cf->cf_callback;
1690
1691                 /* Advance the pointer before calling the callback allows
1692                  * shutdown functions to unregister themselves. If done the
1693                  * other way around the memory `le' points to will be freed
1694                  * after callback returns. */
1695                 le = le->next;
1696
1697                 (*callback) ();
1698
1699                 plugin_set_ctx (old_ctx);
1700         }
1701
1702         stop_write_threads ();
1703
1704         /* Write plugins which use the `user_data' pointer usually need the
1705          * same data available to the flush callback. If this is the case, set
1706          * the free_function to NULL when registering the flush callback and to
1707          * the real free function when registering the write callback. This way
1708          * the data isn't freed twice. */
1709         destroy_all_callbacks (&list_flush);
1710         destroy_all_callbacks (&list_missing);
1711         destroy_all_callbacks (&list_write);
1712
1713         destroy_all_callbacks (&list_notification);
1714         destroy_all_callbacks (&list_shutdown);
1715         destroy_all_callbacks (&list_log);
1716 } /* void plugin_shutdown_all */
1717
1718 int plugin_dispatch_missing (const value_list_t *vl) /* {{{ */
1719 {
1720   llentry_t *le;
1721
1722   if (list_missing == NULL)
1723     return (0);
1724
1725   le = llist_head (list_missing);
1726   while (le != NULL)
1727   {
1728     callback_func_t *cf;
1729     plugin_missing_cb callback;
1730     plugin_ctx_t old_ctx;
1731     int status;
1732
1733     cf = le->value;
1734     old_ctx = plugin_set_ctx (cf->cf_ctx);
1735     callback = cf->cf_callback;
1736
1737     status = (*callback) (vl, &cf->cf_udata);
1738     plugin_set_ctx (old_ctx);
1739     if (status != 0)
1740     {
1741       if (status < 0)
1742       {
1743         ERROR ("plugin_dispatch_missing: Callback function \"%s\" "
1744             "failed with status %i.",
1745             le->key, status);
1746         return (status);
1747       }
1748       else
1749       {
1750         return (0);
1751       }
1752     }
1753
1754     le = le->next;
1755   }
1756   return (0);
1757 } /* int }}} plugin_dispatch_missing */
1758
1759 static int plugin_dispatch_values_internal (value_list_t *vl)
1760 {
1761         int status;
1762         static c_complain_t no_write_complaint = C_COMPLAIN_INIT_STATIC;
1763
1764         value_t *saved_values;
1765         int      saved_values_len;
1766
1767         data_set_t *ds;
1768
1769         int free_meta_data = 0;
1770
1771         if ((vl == NULL) || (vl->type[0] == 0)
1772                         || (vl->values == NULL) || (vl->values_len < 1))
1773         {
1774                 ERROR ("plugin_dispatch_values: Invalid value list "
1775                                 "from plugin %s.", vl->plugin);
1776                 return (-1);
1777         }
1778
1779         /* Free meta data only if the calling function didn't specify any. In
1780          * this case matches and targets may add some and the calling function
1781          * may not expect (and therefore free) that data. */
1782         if (vl->meta == NULL)
1783                 free_meta_data = 1;
1784
1785         if (list_write == NULL)
1786                 c_complain_once (LOG_WARNING, &no_write_complaint,
1787                                 "plugin_dispatch_values: No write callback has been "
1788                                 "registered. Please load at least one output plugin, "
1789                                 "if you want the collected data to be stored.");
1790
1791         if (data_sets == NULL)
1792         {
1793                 ERROR ("plugin_dispatch_values: No data sets registered. "
1794                                 "Could the types database be read? Check "
1795                                 "your `TypesDB' setting!");
1796                 return (-1);
1797         }
1798
1799         if (c_avl_get (data_sets, vl->type, (void *) &ds) != 0)
1800         {
1801                 char ident[6 * DATA_MAX_NAME_LEN];
1802
1803                 FORMAT_VL (ident, sizeof (ident), vl);
1804                 INFO ("plugin_dispatch_values: Dataset not found: %s "
1805                                 "(from \"%s\"), check your types.db!",
1806                                 vl->type, ident);
1807                 return (-1);
1808         }
1809
1810         /* Assured by plugin_value_list_clone(). The write thread doesn't have
1811          * access to the thread context, so the interval has to be filled in
1812          * already. The time is determined at _enqueue_ time. */
1813         assert (vl->time != 0);
1814         assert (vl->interval != 0);
1815
1816         DEBUG ("plugin_dispatch_values: time = %.3f; interval = %.3f; "
1817                         "host = %s; "
1818                         "plugin = %s; plugin_instance = %s; "
1819                         "type = %s; type_instance = %s;",
1820                         CDTIME_T_TO_DOUBLE (vl->time),
1821                         CDTIME_T_TO_DOUBLE (vl->interval),
1822                         vl->host,
1823                         vl->plugin, vl->plugin_instance,
1824                         vl->type, vl->type_instance);
1825
1826 #if COLLECT_DEBUG
1827         assert (0 == strcmp (ds->type, vl->type));
1828 #else
1829         if (0 != strcmp (ds->type, vl->type))
1830                 WARNING ("plugin_dispatch_values: (ds->type = %s) != (vl->type = %s)",
1831                                 ds->type, vl->type);
1832 #endif
1833
1834 #if COLLECT_DEBUG
1835         assert (ds->ds_num == vl->values_len);
1836 #else
1837         if (ds->ds_num != vl->values_len)
1838         {
1839                 ERROR ("plugin_dispatch_values: ds->type = %s: "
1840                                 "(ds->ds_num = %i) != "
1841                                 "(vl->values_len = %i)",
1842                                 ds->type, ds->ds_num, vl->values_len);
1843                 return (-1);
1844         }
1845 #endif
1846
1847         escape_slashes (vl->host, sizeof (vl->host));
1848         escape_slashes (vl->plugin, sizeof (vl->plugin));
1849         escape_slashes (vl->plugin_instance, sizeof (vl->plugin_instance));
1850         escape_slashes (vl->type, sizeof (vl->type));
1851         escape_slashes (vl->type_instance, sizeof (vl->type_instance));
1852
1853         /* Copy the values. This way, we can assure `targets' that they get
1854          * dynamically allocated values, which they can free and replace if
1855          * they like. */
1856         if ((pre_cache_chain != NULL) || (post_cache_chain != NULL))
1857         {
1858                 saved_values     = vl->values;
1859                 saved_values_len = vl->values_len;
1860
1861                 vl->values = (value_t *) calloc (vl->values_len,
1862                                 sizeof (*vl->values));
1863                 if (vl->values == NULL)
1864                 {
1865                         ERROR ("plugin_dispatch_values: calloc failed.");
1866                         vl->values = saved_values;
1867                         return (-1);
1868                 }
1869                 memcpy (vl->values, saved_values,
1870                                 vl->values_len * sizeof (*vl->values));
1871         }
1872         else /* if ((pre == NULL) && (post == NULL)) */
1873         {
1874                 saved_values     = NULL;
1875                 saved_values_len = 0;
1876         }
1877
1878         if (pre_cache_chain != NULL)
1879         {
1880                 status = fc_process_chain (ds, vl, pre_cache_chain);
1881                 if (status < 0)
1882                 {
1883                         WARNING ("plugin_dispatch_values: Running the "
1884                                         "pre-cache chain failed with "
1885                                         "status %i (%#x).",
1886                                         status, status);
1887                 }
1888                 else if (status == FC_TARGET_STOP)
1889                 {
1890                         /* Restore the state of the value_list so that plugins
1891                          * don't get confused.. */
1892                         if (saved_values != NULL)
1893                         {
1894                                 free (vl->values);
1895                                 vl->values     = saved_values;
1896                                 vl->values_len = saved_values_len;
1897                         }
1898                         return (0);
1899                 }
1900         }
1901
1902         /* Update the value cache */
1903         uc_update (ds, vl);
1904
1905         if (post_cache_chain != NULL)
1906         {
1907                 status = fc_process_chain (ds, vl, post_cache_chain);
1908                 if (status < 0)
1909                 {
1910                         WARNING ("plugin_dispatch_values: Running the "
1911                                         "post-cache chain failed with "
1912                                         "status %i (%#x).",
1913                                         status, status);
1914                 }
1915         }
1916         else
1917                 fc_default_action (ds, vl);
1918
1919         /* Restore the state of the value_list so that plugins don't get
1920          * confused.. */
1921         if (saved_values != NULL)
1922         {
1923                 free (vl->values);
1924                 vl->values     = saved_values;
1925                 vl->values_len = saved_values_len;
1926         }
1927
1928         if ((free_meta_data != 0) && (vl->meta != NULL))
1929         {
1930                 meta_data_destroy (vl->meta);
1931                 vl->meta = NULL;
1932         }
1933
1934         return (0);
1935 } /* int plugin_dispatch_values_internal */
1936
1937 int plugin_dispatch_values (value_list_t const *vl)
1938 {
1939         int status;
1940
1941         status = plugin_write_enqueue (vl);
1942         if (status != 0)
1943         {
1944                 char errbuf[1024];
1945                 ERROR ("plugin_dispatch_values: plugin_write_enqueue failed "
1946                                 "with status %i (%s).", status,
1947                                 sstrerror (status, errbuf, sizeof (errbuf)));
1948                 return (status);
1949         }
1950
1951         return (0);
1952 }
1953
1954 int plugin_dispatch_notification (const notification_t *notif)
1955 {
1956         llentry_t *le;
1957         /* Possible TODO: Add flap detection here */
1958
1959         DEBUG ("plugin_dispatch_notification: severity = %i; message = %s; "
1960                         "time = %.3f; host = %s;",
1961                         notif->severity, notif->message,
1962                         CDTIME_T_TO_DOUBLE (notif->time), notif->host);
1963
1964         /* Nobody cares for notifications */
1965         if (list_notification == NULL)
1966                 return (-1);
1967
1968         le = llist_head (list_notification);
1969         while (le != NULL)
1970         {
1971                 callback_func_t *cf;
1972                 plugin_notification_cb callback;
1973                 int status;
1974
1975                 /* do not switch plugin context; rather keep the context
1976                  * (interval) information of the calling plugin */
1977
1978                 cf = le->value;
1979                 callback = cf->cf_callback;
1980                 status = (*callback) (notif, &cf->cf_udata);
1981                 if (status != 0)
1982                 {
1983                         WARNING ("plugin_dispatch_notification: Notification "
1984                                         "callback %s returned %i.",
1985                                         le->key, status);
1986                 }
1987
1988                 le = le->next;
1989         }
1990
1991         return (0);
1992 } /* int plugin_dispatch_notification */
1993
1994 void plugin_log (int level, const char *format, ...)
1995 {
1996         char msg[1024];
1997         va_list ap;
1998         llentry_t *le;
1999
2000 #if !COLLECT_DEBUG
2001         if (level >= LOG_DEBUG)
2002                 return;
2003 #endif
2004
2005         va_start (ap, format);
2006         vsnprintf (msg, sizeof (msg), format, ap);
2007         msg[sizeof (msg) - 1] = '\0';
2008         va_end (ap);
2009
2010         if (list_log == NULL)
2011         {
2012                 fprintf (stderr, "%s\n", msg);
2013                 return;
2014         }
2015
2016         le = llist_head (list_log);
2017         while (le != NULL)
2018         {
2019                 callback_func_t *cf;
2020                 plugin_log_cb callback;
2021
2022                 cf = le->value;
2023                 callback = cf->cf_callback;
2024
2025                 /* do not switch plugin context; rather keep the context
2026                  * (interval) information of the calling plugin */
2027
2028                 (*callback) (level, msg, &cf->cf_udata);
2029
2030                 le = le->next;
2031         }
2032 } /* void plugin_log */
2033
2034 int parse_log_severity (const char *severity)
2035 {
2036         int log_level = -1;
2037
2038         if ((0 == strcasecmp (severity, "emerg"))
2039                         || (0 == strcasecmp (severity, "alert"))
2040                         || (0 == strcasecmp (severity, "crit"))
2041                         || (0 == strcasecmp (severity, "err")))
2042                 log_level = LOG_ERR;
2043         else if (0 == strcasecmp (severity, "warning"))
2044                 log_level = LOG_WARNING;
2045         else if (0 == strcasecmp (severity, "notice"))
2046                 log_level = LOG_NOTICE;
2047         else if (0 == strcasecmp (severity, "info"))
2048                 log_level = LOG_INFO;
2049 #if COLLECT_DEBUG
2050         else if (0 == strcasecmp (severity, "debug"))
2051                 log_level = LOG_DEBUG;
2052 #endif /* COLLECT_DEBUG */
2053
2054         return (log_level);
2055 } /* int parse_log_severity */
2056
2057 int parse_notif_severity (const char *severity)
2058 {
2059         int notif_severity = -1;
2060
2061         if (strcasecmp (severity, "FAILURE") == 0)
2062                 notif_severity = NOTIF_FAILURE;
2063         else if (strcmp (severity, "OKAY") == 0)
2064                 notif_severity = NOTIF_OKAY;
2065         else if ((strcmp (severity, "WARNING") == 0)
2066                         || (strcmp (severity, "WARN") == 0))
2067                 notif_severity = NOTIF_WARNING;
2068
2069         return (notif_severity);
2070 } /* int parse_notif_severity */
2071
2072 const data_set_t *plugin_get_ds (const char *name)
2073 {
2074         data_set_t *ds;
2075
2076         if (c_avl_get (data_sets, name, (void *) &ds) != 0)
2077         {
2078                 DEBUG ("No such dataset registered: %s", name);
2079                 return (NULL);
2080         }
2081
2082         return (ds);
2083 } /* data_set_t *plugin_get_ds */
2084
2085 static int plugin_notification_meta_add (notification_t *n,
2086     const char *name,
2087     enum notification_meta_type_e type,
2088     const void *value)
2089 {
2090   notification_meta_t *meta;
2091   notification_meta_t *tail;
2092
2093   if ((n == NULL) || (name == NULL) || (value == NULL))
2094   {
2095     ERROR ("plugin_notification_meta_add: A pointer is NULL!");
2096     return (-1);
2097   }
2098
2099   meta = (notification_meta_t *) malloc (sizeof (notification_meta_t));
2100   if (meta == NULL)
2101   {
2102     ERROR ("plugin_notification_meta_add: malloc failed.");
2103     return (-1);
2104   }
2105   memset (meta, 0, sizeof (notification_meta_t));
2106
2107   sstrncpy (meta->name, name, sizeof (meta->name));
2108   meta->type = type;
2109
2110   switch (type)
2111   {
2112     case NM_TYPE_STRING:
2113     {
2114       meta->nm_value.nm_string = strdup ((const char *) value);
2115       if (meta->nm_value.nm_string == NULL)
2116       {
2117         ERROR ("plugin_notification_meta_add: strdup failed.");
2118         sfree (meta);
2119         return (-1);
2120       }
2121       break;
2122     }
2123     case NM_TYPE_SIGNED_INT:
2124     {
2125       meta->nm_value.nm_signed_int = *((int64_t *) value);
2126       break;
2127     }
2128     case NM_TYPE_UNSIGNED_INT:
2129     {
2130       meta->nm_value.nm_unsigned_int = *((uint64_t *) value);
2131       break;
2132     }
2133     case NM_TYPE_DOUBLE:
2134     {
2135       meta->nm_value.nm_double = *((double *) value);
2136       break;
2137     }
2138     case NM_TYPE_BOOLEAN:
2139     {
2140       meta->nm_value.nm_boolean = *((_Bool *) value);
2141       break;
2142     }
2143     default:
2144     {
2145       ERROR ("plugin_notification_meta_add: Unknown type: %i", type);
2146       sfree (meta);
2147       return (-1);
2148     }
2149   } /* switch (type) */
2150
2151   meta->next = NULL;
2152   tail = n->meta;
2153   while ((tail != NULL) && (tail->next != NULL))
2154     tail = tail->next;
2155
2156   if (tail == NULL)
2157     n->meta = meta;
2158   else
2159     tail->next = meta;
2160
2161   return (0);
2162 } /* int plugin_notification_meta_add */
2163
2164 int plugin_notification_meta_add_string (notification_t *n,
2165     const char *name,
2166     const char *value)
2167 {
2168   return (plugin_notification_meta_add (n, name, NM_TYPE_STRING, value));
2169 }
2170
2171 int plugin_notification_meta_add_signed_int (notification_t *n,
2172     const char *name,
2173     int64_t value)
2174 {
2175   return (plugin_notification_meta_add (n, name, NM_TYPE_SIGNED_INT, &value));
2176 }
2177
2178 int plugin_notification_meta_add_unsigned_int (notification_t *n,
2179     const char *name,
2180     uint64_t value)
2181 {
2182   return (plugin_notification_meta_add (n, name, NM_TYPE_UNSIGNED_INT, &value));
2183 }
2184
2185 int plugin_notification_meta_add_double (notification_t *n,
2186     const char *name,
2187     double value)
2188 {
2189   return (plugin_notification_meta_add (n, name, NM_TYPE_DOUBLE, &value));
2190 }
2191
2192 int plugin_notification_meta_add_boolean (notification_t *n,
2193     const char *name,
2194     _Bool value)
2195 {
2196   return (plugin_notification_meta_add (n, name, NM_TYPE_BOOLEAN, &value));
2197 }
2198
2199 int plugin_notification_meta_copy (notification_t *dst,
2200     const notification_t *src)
2201 {
2202   notification_meta_t *meta;
2203
2204   assert (dst != NULL);
2205   assert (src != NULL);
2206   assert (dst != src);
2207   assert ((src->meta == NULL) || (src->meta != dst->meta));
2208
2209   for (meta = src->meta; meta != NULL; meta = meta->next)
2210   {
2211     if (meta->type == NM_TYPE_STRING)
2212       plugin_notification_meta_add_string (dst, meta->name,
2213           meta->nm_value.nm_string);
2214     else if (meta->type == NM_TYPE_SIGNED_INT)
2215       plugin_notification_meta_add_signed_int (dst, meta->name,
2216           meta->nm_value.nm_signed_int);
2217     else if (meta->type == NM_TYPE_UNSIGNED_INT)
2218       plugin_notification_meta_add_unsigned_int (dst, meta->name,
2219           meta->nm_value.nm_unsigned_int);
2220     else if (meta->type == NM_TYPE_DOUBLE)
2221       plugin_notification_meta_add_double (dst, meta->name,
2222           meta->nm_value.nm_double);
2223     else if (meta->type == NM_TYPE_BOOLEAN)
2224       plugin_notification_meta_add_boolean (dst, meta->name,
2225           meta->nm_value.nm_boolean);
2226   }
2227
2228   return (0);
2229 } /* int plugin_notification_meta_copy */
2230
2231 int plugin_notification_meta_free (notification_meta_t *n)
2232 {
2233   notification_meta_t *this;
2234   notification_meta_t *next;
2235
2236   if (n == NULL)
2237   {
2238     ERROR ("plugin_notification_meta_free: n == NULL!");
2239     return (-1);
2240   }
2241
2242   this = n;
2243   while (this != NULL)
2244   {
2245     next = this->next;
2246
2247     if (this->type == NM_TYPE_STRING)
2248     {
2249       free ((char *)this->nm_value.nm_string);
2250       this->nm_value.nm_string = NULL;
2251     }
2252     sfree (this);
2253
2254     this = next;
2255   }
2256
2257   return (0);
2258 } /* int plugin_notification_meta_free */
2259
2260 static void plugin_ctx_destructor (void *ctx)
2261 {
2262         sfree (ctx);
2263 } /* void plugin_ctx_destructor */
2264
2265 static plugin_ctx_t ctx_init = { /* interval = */ 0 };
2266
2267 static plugin_ctx_t *plugin_ctx_create (void)
2268 {
2269         plugin_ctx_t *ctx;
2270
2271         ctx = malloc (sizeof (*ctx));
2272         if (ctx == NULL) {
2273                 char errbuf[1024];
2274                 ERROR ("Failed to allocate plugin context: %s",
2275                                 sstrerror (errno, errbuf, sizeof (errbuf)));
2276                 return NULL;
2277         }
2278
2279         *ctx = ctx_init;
2280         assert (plugin_ctx_key_initialized);
2281         pthread_setspecific (plugin_ctx_key, ctx);
2282         DEBUG("Created new plugin context.");
2283         return (ctx);
2284 } /* int plugin_ctx_create */
2285
2286 void plugin_init_ctx (void)
2287 {
2288         pthread_key_create (&plugin_ctx_key, plugin_ctx_destructor);
2289         plugin_ctx_key_initialized = 1;
2290 } /* void plugin_init_ctx */
2291
2292 plugin_ctx_t plugin_get_ctx (void)
2293 {
2294         plugin_ctx_t *ctx;
2295
2296         assert (plugin_ctx_key_initialized);
2297         ctx = pthread_getspecific (plugin_ctx_key);
2298
2299         if (ctx == NULL) {
2300                 ctx = plugin_ctx_create ();
2301                 /* this must no happen -- exit() instead? */
2302                 if (ctx == NULL)
2303                         return ctx_init;
2304         }
2305
2306         return (*ctx);
2307 } /* plugin_ctx_t plugin_get_ctx */
2308
2309 plugin_ctx_t plugin_set_ctx (plugin_ctx_t ctx)
2310 {
2311         plugin_ctx_t *c;
2312         plugin_ctx_t old;
2313
2314         assert (plugin_ctx_key_initialized);
2315         c = pthread_getspecific (plugin_ctx_key);
2316
2317         if (c == NULL) {
2318                 c = plugin_ctx_create ();
2319                 /* this must no happen -- exit() instead? */
2320                 if (c == NULL)
2321                         return ctx_init;
2322         }
2323
2324         old = *c;
2325         *c = ctx;
2326
2327         return (old);
2328 } /* void plugin_set_ctx */
2329
2330 cdtime_t plugin_get_interval (void)
2331 {
2332         cdtime_t interval;
2333
2334         interval = plugin_get_ctx().interval;
2335         if (interval > 0)
2336                 return interval;
2337
2338         return cf_get_default_interval ();
2339 } /* cdtime_t plugin_get_interval */
2340
2341 typedef struct {
2342         plugin_ctx_t ctx;
2343         void *(*start_routine) (void *);
2344         void *arg;
2345 } plugin_thread_t;
2346
2347 static void *plugin_thread_start (void *arg)
2348 {
2349         plugin_thread_t *plugin_thread = arg;
2350
2351         void *(*start_routine) (void *) = plugin_thread->start_routine;
2352         void *plugin_arg = plugin_thread->arg;
2353
2354         plugin_set_ctx (plugin_thread->ctx);
2355
2356         free (plugin_thread);
2357
2358         return start_routine (plugin_arg);
2359 } /* void *plugin_thread_start */
2360
2361 int plugin_thread_create (pthread_t *thread, const pthread_attr_t *attr,
2362                 void *(*start_routine) (void *), void *arg)
2363 {
2364         plugin_thread_t *plugin_thread;
2365
2366         plugin_thread = malloc (sizeof (*plugin_thread));
2367         if (plugin_thread == NULL)
2368                 return -1;
2369
2370         plugin_thread->ctx           = plugin_get_ctx ();
2371         plugin_thread->start_routine = start_routine;
2372         plugin_thread->arg           = arg;
2373
2374         return pthread_create (thread, attr,
2375                         plugin_thread_start, plugin_thread);
2376 } /* int plugin_thread_create */
2377
2378 /* vim: set sw=8 ts=8 noet fdm=marker : */