X-Git-Url: https://git.octo.it/?a=blobdiff_plain;f=src%2Frrdtool.c;h=b80e1109cfcdfed8e27623a3390e548e0ca9e7e5;hb=6d86a97190a193863aadc252f20f6b9dba61c263;hp=1c4f5dd297b759d7bbc9c2071f3e5eb822555471;hpb=1b2b58c338a39a7c2e01bfd3273a381ed376284d;p=collectd.git diff --git a/src/rrdtool.c b/src/rrdtool.c index 1c4f5dd2..b80e1109 100644 --- a/src/rrdtool.c +++ b/src/rrdtool.c @@ -43,7 +43,8 @@ struct rrd_cache_s enum { FLAG_NONE = 0x00, - FLAG_QUEUED = 0x01 + FLAG_QUEUED = 0x01, + FLAG_FLUSHQ = 0x02 } flags; }; typedef struct rrd_cache_s rrd_cache_t; @@ -74,14 +75,16 @@ static const char *config_keys[] = "HeartBeat", "RRARows", "RRATimespan", - "XFF" + "XFF", + "WritesPerSecond" }; static int config_keys_num = STATIC_ARRAY_SIZE (config_keys); /* If datadir is zero, the daemon's basedir is used. If stepsize or heartbeat * is zero a default, depending on the `interval' member of the value list is * being used. */ -static char *datadir = NULL; +static char *datadir = NULL; +static double write_rate = 0.0; static rrdcreate_config_t rrdcreate_config = { /* stepsize = */ 0, @@ -106,6 +109,8 @@ static pthread_mutex_t cache_lock = PTHREAD_MUTEX_INITIALIZER; static rrd_queue_t *queue_head = NULL; static rrd_queue_t *queue_tail = NULL; +static rrd_queue_t *flushq_head = NULL; +static rrd_queue_t *flushq_tail = NULL; static pthread_t queue_thread = 0; static pthread_mutex_t queue_lock = PTHREAD_MUTEX_INITIALIZER; static pthread_cond_t queue_cond = PTHREAD_COND_INITIALIZER; @@ -218,7 +223,7 @@ static int value_list_to_string (char *buffer, int buffer_len, } /* int value_list_to_string */ static int value_list_to_filename (char *buffer, int buffer_len, - const data_set_t *ds, const value_list_t *vl) + const data_set_t __attribute__((unused)) *ds, const value_list_t *vl) { int offset = 0; int status; @@ -261,37 +266,93 @@ static int value_list_to_filename (char *buffer, int buffer_len, return (0); } /* int value_list_to_filename */ -static void *rrd_queue_thread (void *data) +static void *rrd_queue_thread (void __attribute__((unused)) *data) { + struct timeval tv_next_update; + struct timeval tv_now; + + gettimeofday (&tv_next_update, /* timezone = */ NULL); + while (42) { rrd_queue_t *queue_entry; rrd_cache_t *cache_entry; char **values; int values_num; + int status; int i; - /* XXX: If you need to lock both, cache_lock and queue_lock, at - * the same time, ALWAYS lock `cache_lock' first! */ - - /* wait until an entry is available */ - pthread_mutex_lock (&queue_lock); - while ((queue_head == NULL) && (do_shutdown == 0)) - pthread_cond_wait (&queue_cond, &queue_lock); - - /* We're in the shutdown phase */ - if (queue_head == NULL) - { - pthread_mutex_unlock (&queue_lock); - break; - } - - /* Dequeue the first entry */ - queue_entry = queue_head; - if (queue_head == queue_tail) - queue_head = queue_tail = NULL; - else - queue_head = queue_head->next; + pthread_mutex_lock (&queue_lock); + /* Wait for values to arrive */ + while (true) + { + struct timespec ts_wait; + + while ((flushq_head == NULL) && (queue_head == NULL) + && (do_shutdown == 0)) + pthread_cond_wait (&queue_cond, &queue_lock); + + if ((flushq_head == NULL) && (queue_head == NULL)) + break; + + /* Don't delay if there's something to flush */ + if (flushq_head != NULL) + break; + + /* Don't delay if we're shutting down */ + if (do_shutdown != 0) + break; + + /* Don't delay if no delay was configured. */ + if (write_rate <= 0.0) + break; + + gettimeofday (&tv_now, /* timezone = */ NULL); + status = timeval_cmp (tv_next_update, tv_now, NULL); + /* We're good to go */ + if (status <= 0) + break; + + /* We're supposed to wait a bit with this update, so we'll + * wait for the next addition to the queue or to the end of + * the wait period - whichever comes first. */ + ts_wait.tv_sec = tv_next_update.tv_sec; + ts_wait.tv_nsec = 1000 * tv_next_update.tv_usec; + + status = pthread_cond_timedwait (&queue_cond, &queue_lock, + &ts_wait); + if (status == ETIMEDOUT) + break; + } /* while (true) */ + + /* XXX: If you need to lock both, cache_lock and queue_lock, at + * the same time, ALWAYS lock `cache_lock' first! */ + + /* We're in the shutdown phase */ + if ((flushq_head == NULL) && (queue_head == NULL)) + { + pthread_mutex_unlock (&queue_lock); + break; + } + + if (flushq_head != NULL) + { + /* Dequeue the first flush entry */ + queue_entry = flushq_head; + if (flushq_head == flushq_tail) + flushq_head = flushq_tail = NULL; + else + flushq_head = flushq_head->next; + } + else /* if (queue_head != NULL) */ + { + /* Dequeue the first regular entry */ + queue_entry = queue_head; + if (queue_head == queue_tail) + queue_head = queue_tail = NULL; + else + queue_head = queue_head->next; + } /* Unlock the queue again */ pthread_mutex_unlock (&queue_lock); @@ -300,17 +361,42 @@ static void *rrd_queue_thread (void *data) * we make a copy of it's values */ pthread_mutex_lock (&cache_lock); - c_avl_get (cache, queue_entry->filename, (void *) &cache_entry); + status = c_avl_get (cache, queue_entry->filename, + (void *) &cache_entry); - values = cache_entry->values; - values_num = cache_entry->values_num; + if (status == 0) + { + values = cache_entry->values; + values_num = cache_entry->values_num; - cache_entry->values = NULL; - cache_entry->values_num = 0; - cache_entry->flags = FLAG_NONE; + cache_entry->values = NULL; + cache_entry->values_num = 0; + cache_entry->flags = FLAG_NONE; + } pthread_mutex_unlock (&cache_lock); + if (status != 0) + { + sfree (queue_entry->filename); + sfree (queue_entry); + continue; + } + + /* Update `tv_next_update' */ + if (write_rate > 0.0) + { + gettimeofday (&tv_now, /* timezone = */ NULL); + tv_next_update.tv_sec = tv_now.tv_sec; + tv_next_update.tv_usec = tv_now.tv_usec + + ((suseconds_t) (1000000 * write_rate)); + while (tv_next_update.tv_usec > 1000000) + { + tv_next_update.tv_sec++; + tv_next_update.tv_usec -= 1000000; + } + } + /* Write the values to the RRD-file */ srrd_update (queue_entry->filename, NULL, values_num, (const char **)values); @@ -335,7 +421,8 @@ static void *rrd_queue_thread (void *data) return ((void *) 0); } /* void *rrd_queue_thread */ -static int rrd_queue_cache_entry (const char *filename, rrd_queue_dir_t dir) +static int rrd_queue_enqueue (const char *filename, + rrd_queue_t **head, rrd_queue_t **tail) { rrd_queue_t *queue_entry; @@ -353,55 +440,60 @@ static int rrd_queue_cache_entry (const char *filename, rrd_queue_dir_t dir) queue_entry->next = NULL; pthread_mutex_lock (&queue_lock); - if (dir == QUEUE_INSERT_FRONT) - { - queue_entry->next = queue_head; - queue_head = queue_entry; - if (queue_tail == NULL) - queue_tail = queue_head; - } - else /* (dir == QUEUE_INSERT_BACK) */ - { - if (queue_tail == NULL) - queue_head = queue_entry; - else - queue_tail->next = queue_entry; - queue_tail = queue_entry; - } + + if (*tail == NULL) + *head = queue_entry; + else + (*tail)->next = queue_entry; + *tail = queue_entry; + pthread_cond_signal (&queue_cond); pthread_mutex_unlock (&queue_lock); - DEBUG ("rrdtool plugin: Put `%s' into the update queue", filename); - return (0); -} /* int rrd_queue_cache_entry */ +} /* int rrd_queue_enqueue */ -static int rrd_queue_move_to_front (const char *filename) +static int rrd_queue_dequeue (const char *filename, + rrd_queue_t **head, rrd_queue_t **tail) { rrd_queue_t *this; rrd_queue_t *prev; - this = NULL; - prev = NULL; pthread_mutex_lock (&queue_lock); - for (this = queue_head; this != NULL; this = this->next) + + prev = NULL; + this = *head; + + while (this != NULL) { if (strcmp (this->filename, filename) == 0) break; + prev = this; + this = this->next; } - /* Check if we found the entry and if it is NOT the first entry. */ - if ((this != NULL) && (prev != NULL)) + if (this == NULL) { - prev->next = this->next; - this->next = queue_head; - queue_head = this; + pthread_mutex_unlock (&queue_lock); + return (-1); } + + if (prev == NULL) + *head = this->next; + else + prev->next = this->next; + + if (this->next == NULL) + *tail = prev; + pthread_mutex_unlock (&queue_lock); + sfree (this->filename); + sfree (this); + return (0); -} /* int rrd_queue_move_to_front */ +} /* int rrd_queue_dequeue */ static void rrd_cache_flush (int timeout) { @@ -423,13 +515,16 @@ static void rrd_cache_flush (int timeout) iter = c_avl_get_iterator (cache); while (c_avl_iterator_next (iter, (void *) &key, (void *) &rc) == 0) { - if (rc->flags == FLAG_QUEUED) + if (rc->flags != FLAG_NONE) continue; else if ((now - rc->first_value) < timeout) continue; else if (rc->values_num > 0) { - if (rrd_queue_cache_entry (key, QUEUE_INSERT_BACK) == 0) + int status; + + status = rrd_queue_enqueue (key, &queue_head, &queue_tail); + if (status == 0) rc->flags = FLAG_QUEUED; } else /* ancient and no values -> waste of memory */ @@ -491,31 +586,42 @@ static int rrd_cache_flush_identifier (int timeout, const char *identifier) now = time (NULL); if (datadir == NULL) - snprintf (key, sizeof (key), "%s.rrd", - identifier); + snprintf (key, sizeof (key), "%s.rrd", + identifier); else - snprintf (key, sizeof (key), "%s/%s.rrd", - datadir, identifier); + snprintf (key, sizeof (key), "%s/%s.rrd", + datadir, identifier); key[sizeof (key) - 1] = 0; status = c_avl_get (cache, key, (void *) &rc); if (status != 0) { WARNING ("rrdtool plugin: rrd_cache_flush_identifier: " - "c_avl_get (%s) failed. Does that file really exist?", - key); + "c_avl_get (%s) failed. Does that file really exist?", + key); return (status); } - if (rc->flags == FLAG_QUEUED) - status = rrd_queue_move_to_front (key); + if (rc->flags == FLAG_FLUSHQ) + { + status = 0; + } + else if (rc->flags == FLAG_QUEUED) + { + rrd_queue_dequeue (key, &queue_head, &queue_tail); + status = rrd_queue_enqueue (key, &flushq_head, &flushq_tail); + if (status == 0) + rc->flags = FLAG_FLUSHQ; + } else if ((now - rc->first_value) < timeout) + { status = 0; + } else if (rc->values_num > 0) { - status = rrd_queue_cache_entry (key, QUEUE_INSERT_FRONT); + status = rrd_queue_enqueue (key, &flushq_head, &flushq_tail); if (status == 0) - rc->flags = FLAG_QUEUED; + rc->flags = FLAG_FLUSHQ; } return (status); @@ -530,6 +636,15 @@ static int rrd_cache_insert (const char *filename, pthread_mutex_lock (&cache_lock); + /* This shouldn't happen, but it did happen at least once, so we'll be + * careful. */ + if (cache == NULL) + { + pthread_mutex_unlock (&cache_lock); + WARNING ("rrdtool plugin: cache == NULL."); + return (-1); + } + c_avl_get (cache, filename, (void *) &rc); if (rc == NULL) @@ -615,9 +730,12 @@ static int rrd_cache_insert (const char *filename, { /* XXX: If you need to lock both, cache_lock and queue_lock, at * the same time, ALWAYS lock `cache_lock' first! */ - if (rc->flags != FLAG_QUEUED) + if (rc->flags == FLAG_NONE) { - if (rrd_queue_cache_entry (filename, QUEUE_INSERT_BACK) == 0) + int status; + + status = rrd_queue_enqueue (filename, &queue_head, &queue_tail); + if (status == 0) rc->flags = FLAG_QUEUED; } else @@ -630,7 +748,6 @@ static int rrd_cache_insert (const char *filename, ((time (NULL) - cache_flush_last) > cache_flush_timeout)) rrd_cache_flush (cache_flush_timeout); - pthread_mutex_unlock (&cache_lock); return (0); @@ -721,6 +838,8 @@ static int rrd_config (const char *key, const char *value) { fprintf (stderr, "rrdtool: `CacheTimeout' must " "be greater than 0.\n"); + ERROR ("rrdtool: `CacheTimeout' must " + "be greater than 0.\n"); return (1); } cache_timeout = tmp; @@ -732,6 +851,8 @@ static int rrd_config (const char *key, const char *value) { fprintf (stderr, "rrdtool: `CacheFlush' must " "be greater than 0.\n"); + ERROR ("rrdtool: `CacheFlush' must " + "be greater than 0.\n"); return (1); } cache_flush_timeout = tmp; @@ -775,6 +896,8 @@ static int rrd_config (const char *key, const char *value) { fprintf (stderr, "rrdtool: `RRARows' must " "be greater than 0.\n"); + ERROR ("rrdtool: `RRARows' must " + "be greater than 0.\n"); return (1); } rrdcreate_config.rrarows = tmp; @@ -801,6 +924,7 @@ static int rrd_config (const char *key, const char *value) if (tmp_alloc == NULL) { fprintf (stderr, "rrdtool: realloc failed.\n"); + ERROR ("rrdtool: realloc failed.\n"); free (value_copy); return (1); } @@ -824,10 +948,31 @@ static int rrd_config (const char *key, const char *value) { fprintf (stderr, "rrdtool: `XFF' must " "be in the range 0 to 1 (exclusive)."); + ERROR ("rrdtool: `XFF' must " + "be in the range 0 to 1 (exclusive)."); return (1); } rrdcreate_config.xff = tmp; } + else if (strcasecmp ("WritesPerSecond", key) == 0) + { + double wps = atof (value); + + if (wps < 0.0) + { + fprintf (stderr, "rrdtool: `WritesPerSecond' must be " + "greater than or equal to zero."); + return (1); + } + else if (wps == 0.0) + { + write_rate = 0.0; + } + else + { + write_rate = 1.0 / wps; + } + } else { return (-1);