Exec plugin: Reset the "PL_RUNNING" if "fork_child()" fails.
[collectd.git] / src / utils_cache.c
1 /**
2  * collectd - src/utils_cache.c
3  * Copyright (C) 2007,2008  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  * Author:
19  *   Florian octo Forster <octo at verplant.org>
20  **/
21
22 #include "collectd.h"
23 #include "common.h"
24 #include "plugin.h"
25 #include "utils_avltree.h"
26 #include "utils_cache.h"
27 #include "utils_threshold.h"
28 #include "meta_data.h"
29
30 #include <assert.h>
31 #include <pthread.h>
32
33 typedef struct cache_entry_s
34 {
35         char name[6 * DATA_MAX_NAME_LEN];
36         int        values_num;
37         gauge_t   *values_gauge;
38         value_t   *values_raw;
39         /* Time contained in the package
40          * (for calculating rates) */
41         time_t last_time;
42         /* Time according to the local clock
43          * (for purging old entries) */
44         time_t last_update;
45         /* Interval in which the data is collected
46          * (for purding old entries) */
47         int interval;
48         int state;
49         int hits;
50
51         /*
52          * +-----+-----+-----+-----+-----+-----+-----+-----+-----+----
53          * !  0  !  1  !  2  !  3  !  4  !  5  !  6  !  7  !  8  ! ...
54          * +-----+-----+-----+-----+-----+-----+-----+-----+-----+----
55          * ! ds0 ! ds1 ! ds2 ! ds0 ! ds1 ! ds2 ! ds0 ! ds1 ! ds2 ! ...
56          * +-----+-----+-----+-----+-----+-----+-----+-----+-----+----
57          * !      t = 0      !      t = 1      !      t = 2      ! ...
58          * +-----------------+-----------------+-----------------+----
59          */
60         gauge_t *history;
61         size_t   history_index; /* points to the next position to write to. */
62         size_t   history_length;
63
64         meta_data_t *meta;
65 } cache_entry_t;
66
67 static c_avl_tree_t   *cache_tree = NULL;
68 static pthread_mutex_t cache_lock = PTHREAD_MUTEX_INITIALIZER;
69
70 static int cache_compare (const cache_entry_t *a, const cache_entry_t *b)
71 {
72   assert ((a != NULL) && (b != NULL));
73   return (strcmp (a->name, b->name));
74 } /* int cache_compare */
75
76 static cache_entry_t *cache_alloc (int values_num)
77 {
78   cache_entry_t *ce;
79
80   ce = (cache_entry_t *) malloc (sizeof (cache_entry_t));
81   if (ce == NULL)
82   {
83     ERROR ("utils_cache: cache_alloc: malloc failed.");
84     return (NULL);
85   }
86   memset (ce, '\0', sizeof (cache_entry_t));
87   ce->values_num = values_num;
88
89   ce->values_gauge = calloc (values_num, sizeof (*ce->values_gauge));
90   ce->values_raw   = calloc (values_num, sizeof (*ce->values_raw));
91   if ((ce->values_gauge == NULL) || (ce->values_raw == NULL))
92   {
93     sfree (ce->values_gauge);
94     sfree (ce->values_raw);
95     sfree (ce);
96     ERROR ("utils_cache: cache_alloc: calloc failed.");
97     return (NULL);
98   }
99
100   ce->history = NULL;
101   ce->history_length = 0;
102   ce->meta = NULL;
103
104   return (ce);
105 } /* cache_entry_t *cache_alloc */
106
107 static void cache_free (cache_entry_t *ce)
108 {
109   if (ce == NULL)
110     return;
111
112   sfree (ce->values_gauge);
113   sfree (ce->values_raw);
114   sfree (ce->history);
115   if (ce->meta != NULL)
116   {
117     meta_data_destroy (ce->meta);
118     ce->meta = NULL;
119   }
120   sfree (ce);
121 } /* void cache_free */
122
123 static int uc_send_notification (const char *name)
124 {
125   cache_entry_t *ce = NULL;
126   int status;
127
128   char *name_copy;
129   char *host;
130   char *plugin;
131   char *plugin_instance;
132   char *type;
133   char *type_instance;
134
135   notification_t n;
136
137   name_copy = strdup (name);
138   if (name_copy == NULL)
139   {
140     ERROR ("uc_send_notification: strdup failed.");
141     return (-1);
142   }
143
144   status = parse_identifier (name_copy, &host,
145       &plugin, &plugin_instance,
146       &type, &type_instance);
147   if (status != 0)
148   {
149     ERROR ("uc_send_notification: Cannot parse name `%s'", name);
150     return (-1);
151   }
152
153   /* Copy the associative members */
154   notification_init (&n, NOTIF_FAILURE, /* host = */ NULL,
155       host, plugin, plugin_instance, type, type_instance);
156
157   sfree (name_copy);
158   name_copy = host = plugin = plugin_instance = type = type_instance = NULL;
159
160   pthread_mutex_lock (&cache_lock);
161
162   /*
163    * Set the time _after_ getting the lock because we don't know how long
164    * acquiring the lock takes and we will use this time later to decide
165    * whether or not the state is OKAY.
166    */
167   n.time = time (NULL);
168
169   status = c_avl_get (cache_tree, name, (void *) &ce);
170   if (status != 0)
171   {
172     pthread_mutex_unlock (&cache_lock);
173     sfree (name_copy);
174     return (-1);
175   }
176     
177   /* Check if the entry has been updated in the meantime */
178   if ((n.time - ce->last_update) < (2 * ce->interval))
179   {
180     ce->state = STATE_OKAY;
181     pthread_mutex_unlock (&cache_lock);
182     sfree (name_copy);
183     return (-1);
184   }
185
186   ssnprintf (n.message, sizeof (n.message),
187       "%s has not been updated for %i seconds.", name,
188       (int) (n.time - ce->last_update));
189
190   pthread_mutex_unlock (&cache_lock);
191
192   plugin_dispatch_notification (&n);
193
194   return (0);
195 } /* int uc_send_notification */
196
197 static void uc_check_range (const data_set_t *ds, cache_entry_t *ce)
198 {
199   int i;
200
201   for (i = 0; i < ds->ds_num; i++)
202   {
203     if (isnan (ce->values_gauge[i]))
204       continue;
205     else if (ce->values_gauge[i] < ds->ds[i].min)
206       ce->values_gauge[i] = NAN;
207     else if (ce->values_gauge[i] > ds->ds[i].max)
208       ce->values_gauge[i] = NAN;
209   }
210 } /* void uc_check_range */
211
212 static int uc_insert (const data_set_t *ds, const value_list_t *vl,
213     const char *key)
214 {
215   int i;
216   char *key_copy;
217   cache_entry_t *ce;
218
219   /* `cache_lock' has been locked by `uc_update' */
220
221   key_copy = strdup (key);
222   if (key_copy == NULL)
223   {
224     ERROR ("uc_insert: strdup failed.");
225     return (-1);
226   }
227
228   ce = cache_alloc (ds->ds_num);
229   if (ce == NULL)
230   {
231     sfree (key_copy);
232     ERROR ("uc_insert: cache_alloc (%i) failed.", ds->ds_num);
233     return (-1);
234   }
235
236   sstrncpy (ce->name, key, sizeof (ce->name));
237
238   for (i = 0; i < ds->ds_num; i++)
239   {
240     switch (ds->ds[i].type)
241     {
242       case DS_TYPE_COUNTER:
243         ce->values_gauge[i] = NAN;
244         ce->values_raw[i].counter = vl->values[i].counter;
245         break;
246
247       case DS_TYPE_GAUGE:
248         ce->values_gauge[i] = vl->values[i].gauge;
249         ce->values_raw[i].gauge = vl->values[i].gauge;
250         break;
251
252       case DS_TYPE_DERIVE:
253         ce->values_gauge[i] = NAN;
254         ce->values_raw[i].derive = vl->values[i].derive;
255         break;
256
257       case DS_TYPE_ABSOLUTE:
258         ce->values_gauge[i] = NAN;
259         if (vl->interval > 0)
260           ce->values_gauge[i] = ((double) vl->values[i].absolute)
261             / ((double) vl->interval);
262         ce->values_raw[i].absolute = vl->values[i].absolute;
263         break;
264         
265       default:
266         /* This shouldn't happen. */
267         ERROR ("uc_insert: Don't know how to handle data source type %i.",
268             ds->ds[i].type);
269         return (-1);
270     } /* switch (ds->ds[i].type) */
271   } /* for (i) */
272
273   /* Prune invalid gauge data */
274   uc_check_range (ds, ce);
275
276   ce->last_time = vl->time;
277   ce->last_update = time (NULL);
278   ce->interval = vl->interval;
279   ce->state = STATE_OKAY;
280
281   if (c_avl_insert (cache_tree, key_copy, ce) != 0)
282   {
283     sfree (key_copy);
284     ERROR ("uc_insert: c_avl_insert failed.");
285     return (-1);
286   }
287
288   DEBUG ("uc_insert: Added %s to the cache.", key);
289   return (0);
290 } /* int uc_insert */
291
292 int uc_init (void)
293 {
294   if (cache_tree == NULL)
295     cache_tree = c_avl_create ((int (*) (const void *, const void *))
296         cache_compare);
297
298   return (0);
299 } /* int uc_init */
300
301 int uc_check_timeout (void)
302 {
303   time_t now;
304   cache_entry_t *ce;
305
306   char **keys = NULL;
307   int keys_len = 0;
308
309   char *key;
310   c_avl_iterator_t *iter;
311   int i;
312   
313   pthread_mutex_lock (&cache_lock);
314
315   now = time (NULL);
316
317   /* Build a list of entries to be flushed */
318   iter = c_avl_get_iterator (cache_tree);
319   while (c_avl_iterator_next (iter, (void *) &key, (void *) &ce) == 0)
320   {
321     /* If entry has not been updated, add to `keys' array */
322     if ((now - ce->last_update) >= (2 * ce->interval))
323     {
324       char **tmp;
325
326       tmp = (char **) realloc ((void *) keys,
327           (keys_len + 1) * sizeof (char *));
328       if (tmp == NULL)
329       {
330         ERROR ("uc_check_timeout: realloc failed.");
331         c_avl_iterator_destroy (iter);
332         sfree (keys);
333         pthread_mutex_unlock (&cache_lock);
334         return (-1);
335       }
336
337       keys = tmp;
338       keys[keys_len] = strdup (key);
339       if (keys[keys_len] == NULL)
340       {
341         ERROR ("uc_check_timeout: strdup failed.");
342         continue;
343       }
344       keys_len++;
345     }
346   } /* while (c_avl_iterator_next) */
347
348   ce = NULL;
349
350   for (i = 0; i < keys_len; i++)
351   {
352     int status;
353
354     status = ut_check_interesting (keys[i]);
355
356     if (status < 0)
357     {
358       ERROR ("uc_check_timeout: ut_check_interesting failed.");
359       sfree (keys[i]);
360       continue;
361     }
362     else if (status == 0) /* ``service'' is uninteresting */
363     {
364       DEBUG ("uc_check_timeout: %s is missing but ``uninteresting''",
365           keys[i]);
366       ce = NULL;
367       status = c_avl_remove (cache_tree, keys[i],
368           (void *) &key, (void *) &ce);
369       if (status != 0)
370       {
371         ERROR ("uc_check_timeout: c_avl_remove (%s) failed.", keys[i]);
372       }
373       sfree (keys[i]);
374       sfree (key);
375       if (ce != NULL)
376         cache_free (ce);
377       continue;
378     }
379
380     /* If we get here, the value is ``interesting''. Query the record from the
381      * cache and update the state field. */
382     if (c_avl_get (cache_tree, keys[i], (void *) &ce) != 0)
383     {
384       ERROR ("uc_check_timeout: cannot get data for %s from cache", keys[i]);
385       /* Do not free `keys[i]' so a notification is sent further down. */
386       continue;
387     }
388     assert (ce != NULL);
389
390     if (status == 2) /* persist */
391     {
392       DEBUG ("uc_check_timeout: %s is missing, sending notification.",
393           keys[i]);
394       ce->state = STATE_MISSING;
395       /* Do not free `keys[i]' so a notification is sent further down. */
396     }
397     else if (status == 1) /* do not persist */
398     {
399       if (ce->state == STATE_MISSING)
400       {
401         DEBUG ("uc_check_timeout: %s is missing but "
402             "notification has already been sent.",
403             keys[i]);
404         /* Set `keys[i]' to NULL to no notification is sent. */
405         sfree (keys[i]);
406       }
407       else /* (ce->state != STATE_MISSING) */
408       {
409         DEBUG ("uc_check_timeout: %s is missing, sending one notification.",
410             keys[i]);
411         ce->state = STATE_MISSING;
412         /* Do not free `keys[i]' so a notification is sent further down. */
413       }
414     }
415     else
416     {
417       WARNING ("uc_check_timeout: ut_check_interesting (%s) returned "
418           "invalid status %i.",
419           keys[i], status);
420       sfree (keys[i]);
421     }
422
423     /* Make really sure the next iteration doesn't work with this pointer.
424      * There have been too many bugs in the past.. :/  -- octo */
425     ce = NULL;
426   } /* for (keys[i]) */
427
428   c_avl_iterator_destroy (iter);
429
430   pthread_mutex_unlock (&cache_lock);
431
432   for (i = 0; i < keys_len; i++)
433   {
434     if (keys[i] == NULL)
435       continue;
436
437     uc_send_notification (keys[i]);
438     sfree (keys[i]);
439   }
440
441   sfree (keys);
442
443   return (0);
444 } /* int uc_check_timeout */
445
446 int uc_update (const data_set_t *ds, const value_list_t *vl)
447 {
448   char name[6 * DATA_MAX_NAME_LEN];
449   cache_entry_t *ce = NULL;
450   int send_okay_notification = 0;
451   time_t update_delay = 0;
452   notification_t n;
453   int status;
454   int i;
455
456   if (FORMAT_VL (name, sizeof (name), vl) != 0)
457   {
458     ERROR ("uc_update: FORMAT_VL failed.");
459     return (-1);
460   }
461
462   pthread_mutex_lock (&cache_lock);
463
464   status = c_avl_get (cache_tree, name, (void *) &ce);
465   if (status != 0) /* entry does not yet exist */
466   {
467     status = uc_insert (ds, vl, name);
468     pthread_mutex_unlock (&cache_lock);
469     return (status);
470   }
471
472   assert (ce != NULL);
473   assert (ce->values_num == ds->ds_num);
474
475   if (ce->last_time >= vl->time)
476   {
477     pthread_mutex_unlock (&cache_lock);
478     NOTICE ("uc_update: Value too old: name = %s; value time = %u; "
479         "last cache update = %u;",
480         name, (unsigned int) vl->time, (unsigned int) ce->last_time);
481     return (-1);
482   }
483
484   /* Send a notification (after the lock has been released) if we switch the
485    * state from something else to `okay'. */
486   if (ce->state == STATE_MISSING)
487   {
488     send_okay_notification = 1;
489     ce->state = STATE_OKAY;
490     update_delay = time (NULL) - ce->last_update;
491   }
492
493   for (i = 0; i < ds->ds_num; i++)
494   {
495     switch (ds->ds[i].type)
496     {
497       case DS_TYPE_COUNTER:
498         {
499           counter_t diff;
500
501           /* check if the counter has wrapped around */
502           if (vl->values[i].counter < ce->values_raw[i].counter)
503           {
504             if (ce->values_raw[i].counter <= 4294967295U)
505               diff = (4294967295U - ce->values_raw[i].counter)
506                 + vl->values[i].counter;
507             else
508               diff = (18446744073709551615ULL - ce->values_raw[i].counter)
509                 + vl->values[i].counter;
510           }
511           else /* counter has NOT wrapped around */
512           {
513             diff = vl->values[i].counter - ce->values_raw[i].counter;
514           }
515
516           ce->values_gauge[i] = ((double) diff)
517             / ((double) (vl->time - ce->last_time));
518           ce->values_raw[i].counter = vl->values[i].counter;
519         }
520         break;
521
522       case DS_TYPE_GAUGE:
523         ce->values_raw[i].gauge = vl->values[i].gauge;
524         ce->values_gauge[i] = vl->values[i].gauge;
525         break;
526
527       case DS_TYPE_DERIVE:
528         {
529           derive_t diff;
530
531           diff = vl->values[i].derive - ce->values_raw[i].derive;
532
533           ce->values_gauge[i] = ((double) diff)
534             / ((double) (vl->time - ce->last_time));
535           ce->values_raw[i].derive = vl->values[i].derive;
536         }
537         break;
538
539       case DS_TYPE_ABSOLUTE:
540         ce->values_gauge[i] = ((double) vl->values[i].absolute)
541           / ((double) (vl->time - ce->last_time));
542         ce->values_raw[i].absolute = vl->values[i].absolute;
543         break;
544
545       default:
546         /* This shouldn't happen. */
547         pthread_mutex_unlock (&cache_lock);
548         ERROR ("uc_update: Don't know how to handle data source type %i.",
549             ds->ds[i].type);
550         return (-1);
551     } /* switch (ds->ds[i].type) */
552
553     DEBUG ("uc_update: %s: ds[%i] = %lf", name, i, ce->values_gauge[i]);
554   } /* for (i) */
555
556   /* Update the history if it exists. */
557   if (ce->history != NULL)
558   {
559     assert (ce->history_index < ce->history_length);
560     for (i = 0; i < ce->values_num; i++)
561     {
562       size_t hist_idx = (ce->values_num * ce->history_index) + i;
563       ce->history[hist_idx] = ce->values_gauge[i];
564     }
565
566     assert (ce->history_length > 0);
567     ce->history_index = (ce->history_index + 1) % ce->history_length;
568   }
569
570   /* Prune invalid gauge data */
571   uc_check_range (ds, ce);
572
573   ce->last_time = vl->time;
574   ce->last_update = time (NULL);
575   ce->interval = vl->interval;
576
577   pthread_mutex_unlock (&cache_lock);
578
579   if (send_okay_notification == 0)
580     return (0);
581
582   /* Do not send okay notifications for uninteresting values, i. e. values for
583    * which no threshold is configured. */
584   status = ut_check_interesting (name);
585   if (status <= 0)
586     return (0);
587
588   /* Initialize the notification */
589   memset (&n, '\0', sizeof (n));
590   NOTIFICATION_INIT_VL (&n, vl, ds);
591
592   n.severity = NOTIF_OKAY;
593   n.time = vl->time;
594
595   ssnprintf (n.message, sizeof (n.message),
596       "Received a value for %s. It was missing for %u seconds.",
597       name, (unsigned int) update_delay);
598
599   plugin_dispatch_notification (&n);
600
601   return (0);
602 } /* int uc_update */
603
604 int uc_get_rate_by_name (const char *name, gauge_t **ret_values, size_t *ret_values_num)
605 {
606   gauge_t *ret = NULL;
607   size_t ret_num = 0;
608   cache_entry_t *ce = NULL;
609   int status = 0;
610
611   pthread_mutex_lock (&cache_lock);
612
613   if (c_avl_get (cache_tree, name, (void *) &ce) == 0)
614   {
615     assert (ce != NULL);
616
617     /* remove missing values from getval */
618     if (ce->state == STATE_MISSING)
619     {
620       status = -1;
621     }
622     else
623     {
624       ret_num = ce->values_num;
625       ret = (gauge_t *) malloc (ret_num * sizeof (gauge_t));
626       if (ret == NULL)
627       {
628         ERROR ("utils_cache: uc_get_rate_by_name: malloc failed.");
629         status = -1;
630       }
631       else
632       {
633         memcpy (ret, ce->values_gauge, ret_num * sizeof (gauge_t));
634       }
635     }
636   }
637   else
638   {
639     DEBUG ("utils_cache: uc_get_rate_by_name: No such value: %s", name);
640     status = -1;
641   }
642
643   pthread_mutex_unlock (&cache_lock);
644
645   if (status == 0)
646   {
647     *ret_values = ret;
648     *ret_values_num = ret_num;
649   }
650
651   return (status);
652 } /* gauge_t *uc_get_rate_by_name */
653
654 gauge_t *uc_get_rate (const data_set_t *ds, const value_list_t *vl)
655 {
656   char name[6 * DATA_MAX_NAME_LEN];
657   gauge_t *ret = NULL;
658   size_t ret_num = 0;
659   int status;
660
661   if (FORMAT_VL (name, sizeof (name), vl) != 0)
662   {
663     ERROR ("utils_cache: uc_get_rate: FORMAT_VL failed.");
664     return (NULL);
665   }
666
667   status = uc_get_rate_by_name (name, &ret, &ret_num);
668   if (status != 0)
669     return (NULL);
670
671   /* This is important - the caller has no other way of knowing how many
672    * values are returned. */
673   if (ret_num != (size_t) ds->ds_num)
674   {
675     ERROR ("utils_cache: uc_get_rate: ds[%s] has %i values, "
676         "but uc_get_rate_by_name returned %zu.",
677         ds->type, ds->ds_num, ret_num);
678     sfree (ret);
679     return (NULL);
680   }
681
682   return (ret);
683 } /* gauge_t *uc_get_rate */
684
685 int uc_get_names (char ***ret_names, time_t **ret_times, size_t *ret_number)
686 {
687   c_avl_iterator_t *iter;
688   char *key;
689   cache_entry_t *value;
690
691   char **names = NULL;
692   time_t *times = NULL;
693   size_t number = 0;
694
695   int status = 0;
696
697   if ((ret_names == NULL) || (ret_number == NULL))
698     return (-1);
699
700   pthread_mutex_lock (&cache_lock);
701
702   iter = c_avl_get_iterator (cache_tree);
703   while (c_avl_iterator_next (iter, (void *) &key, (void *) &value) == 0)
704   {
705     char **temp;
706
707     /* remove missing values when list values */
708     if (value->state == STATE_MISSING)
709       continue;
710
711     if (ret_times != NULL)
712     {
713       time_t *tmp_times;
714
715       tmp_times = (time_t *) realloc (times, sizeof (time_t) * (number + 1));
716       if (tmp_times == NULL)
717       {
718         status = -1;
719         break;
720       }
721       times = tmp_times;
722       times[number] = value->last_time;
723     }
724
725     temp = (char **) realloc (names, sizeof (char *) * (number + 1));
726     if (temp == NULL)
727     {
728       status = -1;
729       break;
730     }
731     names = temp;
732     names[number] = strdup (key);
733     if (names[number] == NULL)
734     {
735       status = -1;
736       break;
737     }
738     number++;
739   } /* while (c_avl_iterator_next) */
740
741   c_avl_iterator_destroy (iter);
742   pthread_mutex_unlock (&cache_lock);
743
744   if (status != 0)
745   {
746     size_t i;
747     
748     for (i = 0; i < number; i++)
749     {
750       sfree (names[i]);
751     }
752     sfree (names);
753
754     return (-1);
755   }
756
757   *ret_names = names;
758   if (ret_times != NULL)
759     *ret_times = times;
760   *ret_number = number;
761
762   return (0);
763 } /* int uc_get_names */
764
765 int uc_get_state (const data_set_t *ds, const value_list_t *vl)
766 {
767   char name[6 * DATA_MAX_NAME_LEN];
768   cache_entry_t *ce = NULL;
769   int ret = STATE_ERROR;
770
771   if (FORMAT_VL (name, sizeof (name), vl) != 0)
772   {
773     ERROR ("uc_get_state: FORMAT_VL failed.");
774     return (STATE_ERROR);
775   }
776
777   pthread_mutex_lock (&cache_lock);
778
779   if (c_avl_get (cache_tree, name, (void *) &ce) == 0)
780   {
781     assert (ce != NULL);
782     ret = ce->state;
783   }
784
785   pthread_mutex_unlock (&cache_lock);
786
787   return (ret);
788 } /* int uc_get_state */
789
790 int uc_set_state (const data_set_t *ds, const value_list_t *vl, int state)
791 {
792   char name[6 * DATA_MAX_NAME_LEN];
793   cache_entry_t *ce = NULL;
794   int ret = -1;
795
796   if (FORMAT_VL (name, sizeof (name), vl) != 0)
797   {
798     ERROR ("uc_get_state: FORMAT_VL failed.");
799     return (STATE_ERROR);
800   }
801
802   pthread_mutex_lock (&cache_lock);
803
804   if (c_avl_get (cache_tree, name, (void *) &ce) == 0)
805   {
806     assert (ce != NULL);
807     ret = ce->state;
808     ce->state = state;
809   }
810
811   pthread_mutex_unlock (&cache_lock);
812
813   return (ret);
814 } /* int uc_set_state */
815
816 int uc_get_history_by_name (const char *name,
817     gauge_t *ret_history, size_t num_steps, size_t num_ds)
818 {
819   cache_entry_t *ce = NULL;
820   size_t i;
821   int status = 0;
822
823   pthread_mutex_lock (&cache_lock);
824
825   status = c_avl_get (cache_tree, name, (void *) &ce);
826   if (status != 0)
827   {
828     pthread_mutex_unlock (&cache_lock);
829     return (-ENOENT);
830   }
831
832   if (((size_t) ce->values_num) != num_ds)
833   {
834     pthread_mutex_unlock (&cache_lock);
835     return (-EINVAL);
836   }
837
838   /* Check if there are enough values available. If not, increase the buffer
839    * size. */
840   if (ce->history_length < num_steps)
841   {
842     gauge_t *tmp;
843     size_t i;
844
845     tmp = realloc (ce->history, sizeof (*ce->history)
846         * num_steps * ce->values_num);
847     if (tmp == NULL)
848     {
849       pthread_mutex_unlock (&cache_lock);
850       return (-ENOMEM);
851     }
852
853     for (i = ce->history_length * ce->values_num;
854         i < (num_steps * ce->values_num);
855         i++)
856       tmp[i] = NAN;
857
858     ce->history = tmp;
859     ce->history_length = num_steps;
860   } /* if (ce->history_length < num_steps) */
861
862   /* Copy the values to the output buffer. */
863   for (i = 0; i < num_steps; i++)
864   {
865     size_t src_index;
866     size_t dst_index;
867
868     if (i < ce->history_index)
869       src_index = ce->history_index - (i + 1);
870     else
871       src_index = ce->history_length + ce->history_index - (i + 1);
872     src_index = src_index * num_ds;
873
874     dst_index = i * num_ds;
875
876     memcpy (ret_history + dst_index, ce->history + src_index,
877         sizeof (*ret_history) * num_ds);
878   }
879
880   pthread_mutex_unlock (&cache_lock);
881
882   return (0);
883 } /* int uc_get_history_by_name */
884
885 int uc_get_history (const data_set_t *ds, const value_list_t *vl,
886     gauge_t *ret_history, size_t num_steps, size_t num_ds)
887 {
888   char name[6 * DATA_MAX_NAME_LEN];
889
890   if (FORMAT_VL (name, sizeof (name), vl) != 0)
891   {
892     ERROR ("utils_cache: uc_get_history: FORMAT_VL failed.");
893     return (-1);
894   }
895
896   return (uc_get_history_by_name (name, ret_history, num_steps, num_ds));
897 } /* int uc_get_history */
898
899 int uc_get_hits (const data_set_t *ds, const value_list_t *vl)
900 {
901   char name[6 * DATA_MAX_NAME_LEN];
902   cache_entry_t *ce = NULL;
903   int ret = STATE_ERROR;
904
905   if (FORMAT_VL (name, sizeof (name), vl) != 0)
906   {
907     ERROR ("uc_get_state: FORMAT_VL failed.");
908     return (STATE_ERROR);
909   }
910
911   pthread_mutex_lock (&cache_lock);
912
913   if (c_avl_get (cache_tree, name, (void *) &ce) == 0)
914   {
915     assert (ce != NULL);
916     ret = ce->hits;
917   }
918
919   pthread_mutex_unlock (&cache_lock);
920
921   return (ret);
922 } /* int uc_get_hits */
923
924 int uc_set_hits (const data_set_t *ds, const value_list_t *vl, int hits)
925 {
926   char name[6 * DATA_MAX_NAME_LEN];
927   cache_entry_t *ce = NULL;
928   int ret = -1;
929
930   if (FORMAT_VL (name, sizeof (name), vl) != 0)
931   {
932     ERROR ("uc_get_state: FORMAT_VL failed.");
933     return (STATE_ERROR);
934   }
935
936   pthread_mutex_lock (&cache_lock);
937
938   if (c_avl_get (cache_tree, name, (void *) &ce) == 0)
939   {
940     assert (ce != NULL);
941     ret = ce->hits;
942     ce->hits = hits;
943   }
944
945   pthread_mutex_unlock (&cache_lock);
946
947   return (ret);
948 } /* int uc_set_hits */
949
950 int uc_inc_hits (const data_set_t *ds, const value_list_t *vl, int step)
951 {
952   char name[6 * DATA_MAX_NAME_LEN];
953   cache_entry_t *ce = NULL;
954   int ret = -1;
955
956   if (FORMAT_VL (name, sizeof (name), vl) != 0)
957   {
958     ERROR ("uc_get_state: FORMAT_VL failed.");
959     return (STATE_ERROR);
960   }
961
962   pthread_mutex_lock (&cache_lock);
963
964   if (c_avl_get (cache_tree, name, (void *) &ce) == 0)
965   {
966     assert (ce != NULL);
967     ret = ce->hits;
968     ce->hits = ret + step;
969   }
970
971   pthread_mutex_unlock (&cache_lock);
972
973   return (ret);
974 } /* int uc_inc_hits */
975
976 /*
977  * Meta data interface
978  */
979 /* XXX: This function will acquire `cache_lock' but will not free it! */
980 static meta_data_t *uc_get_meta (const value_list_t *vl) /* {{{ */
981 {
982   char name[6 * DATA_MAX_NAME_LEN];
983   cache_entry_t *ce = NULL;
984   int status;
985
986   status = FORMAT_VL (name, sizeof (name), vl);
987   if (status != 0)
988   {
989     ERROR ("utils_cache: uc_get_meta: FORMAT_VL failed.");
990     return (NULL);
991   }
992
993   pthread_mutex_lock (&cache_lock);
994
995   status = c_avl_get (cache_tree, name, (void *) &ce);
996   if (status != 0)
997   {
998     pthread_mutex_unlock (&cache_lock);
999     return (NULL);
1000   }
1001   assert (ce != NULL);
1002
1003   if (ce->meta == NULL)
1004     ce->meta = meta_data_create ();
1005
1006   if (ce->meta == NULL)
1007     pthread_mutex_unlock (&cache_lock);
1008
1009   return (ce->meta);
1010 } /* }}} meta_data_t *uc_get_meta */
1011
1012 /* Sorry about this preprocessor magic, but it really makes this file much
1013  * shorter.. */
1014 #define UC_WRAP(wrap_function) { \
1015   meta_data_t *meta; \
1016   int status; \
1017   meta = uc_get_meta (vl); \
1018   if (meta == NULL) return (-1); \
1019   status = wrap_function (meta, key); \
1020   pthread_mutex_unlock (&cache_lock); \
1021   return (status); \
1022 }
1023 int uc_meta_data_exists (const value_list_t *vl, const char *key)
1024   UC_WRAP (meta_data_exists)
1025
1026 int uc_meta_data_delete (const value_list_t *vl, const char *key)
1027   UC_WRAP (meta_data_delete)
1028 #undef UC_WRAP
1029
1030 /* We need a new version of this macro because the following functions take
1031  * two argumetns. */
1032 #define UC_WRAP(wrap_function) { \
1033   meta_data_t *meta; \
1034   int status; \
1035   meta = uc_get_meta (vl); \
1036   if (meta == NULL) return (-1); \
1037   status = wrap_function (meta, key, value); \
1038   pthread_mutex_unlock (&cache_lock); \
1039   return (status); \
1040 }
1041 int uc_meta_data_add_string (const value_list_t *vl,
1042     const char *key,
1043     const char *value)
1044   UC_WRAP(meta_data_add_string)
1045 int uc_meta_data_add_signed_int (const value_list_t *vl,
1046     const char *key,
1047     int64_t value)
1048   UC_WRAP(meta_data_add_signed_int)
1049 int uc_meta_data_add_unsigned_int (const value_list_t *vl,
1050     const char *key,
1051     uint64_t value)
1052   UC_WRAP(meta_data_add_unsigned_int)
1053 int uc_meta_data_add_double (const value_list_t *vl,
1054     const char *key,
1055     double value)
1056   UC_WRAP(meta_data_add_double)
1057 int uc_meta_data_add_boolean (const value_list_t *vl,
1058     const char *key,
1059     _Bool value)
1060   UC_WRAP(meta_data_add_boolean)
1061
1062 int uc_meta_data_get_string (const value_list_t *vl,
1063     const char *key,
1064     char **value)
1065   UC_WRAP(meta_data_get_string)
1066 int uc_meta_data_get_signed_int (const value_list_t *vl,
1067     const char *key,
1068     int64_t *value)
1069   UC_WRAP(meta_data_get_signed_int)
1070 int uc_meta_data_get_unsigned_int (const value_list_t *vl,
1071     const char *key,
1072     uint64_t *value)
1073   UC_WRAP(meta_data_get_unsigned_int)
1074 int uc_meta_data_get_double (const value_list_t *vl,
1075     const char *key,
1076     double *value)
1077   UC_WRAP(meta_data_get_double)
1078 int uc_meta_data_get_boolean (const value_list_t *vl,
1079     const char *key,
1080     _Bool *value)
1081   UC_WRAP(meta_data_get_boolean)
1082 #undef UC_WRAP
1083
1084 /* vim: set sw=2 ts=8 sts=2 tw=78 : */