src/plugin.c

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