X-Git-Url: https://git.octo.it/?p=rrdtool.git;a=blobdiff_plain;f=src%2Frrd_daemon.c;h=7a72b070e088191af853c1b43dfc896eee515acb;hp=d56cf06da51ddde4126d302557b9a3915f257244;hb=85cd3f3afae48026843bea34d2d0d1056ee9dda5;hpb=8df853711a3c776d77fed52b6393b09e6289963a diff --git a/src/rrd_daemon.c b/src/rrd_daemon.c index d56cf06..7a72b07 100644 --- a/src/rrd_daemon.c +++ b/src/rrd_daemon.c @@ -75,19 +75,24 @@ #include "rrd_client.h" #include + +#ifndef WIN32 #include -#include #include -#include #include -#include #include +# include + +#else + +#endif +#include +#include #include #include #include #include -#include #include #include #include @@ -187,7 +192,12 @@ static size_t listen_fds_num = 0; static int do_shutdown = 0; -static pthread_t queue_thread; +static pthread_t *queue_threads; +static pthread_cond_t queue_cond = PTHREAD_COND_INITIALIZER; +static int config_queue_threads = 4; + +static pthread_t flush_thread; +static pthread_cond_t flush_cond = PTHREAD_COND_INITIALIZER; static pthread_t *connection_threads = NULL; static pthread_mutex_t connection_threads_lock = PTHREAD_MUTEX_INITIALIZER; @@ -198,7 +208,6 @@ static GTree *cache_tree = NULL; static cache_item_t *cache_queue_head = NULL; static cache_item_t *cache_queue_tail = NULL; static pthread_mutex_t cache_lock = PTHREAD_MUTEX_INITIALIZER; -static pthread_cond_t cache_cond = PTHREAD_COND_INITIALIZER; static int config_write_interval = 300; static int config_write_jitter = 0; @@ -237,7 +246,8 @@ static void sig_common (const char *sig) /* {{{ */ { RRDD_LOG(LOG_NOTICE, "caught SIG%s", sig); do_shutdown++; - pthread_cond_broadcast(&cache_cond); + pthread_cond_broadcast(&flush_cond); + pthread_cond_broadcast(&queue_cond); } /* }}} void sig_common */ static void sig_int_handler (int s __attribute__((unused))) /* {{{ */ @@ -456,7 +466,7 @@ static int add_response_info(listen_socket_t *sock, char *fmt, ...) /* {{{ */ va_start(argp, fmt); #ifdef HAVE_VSNPRINTF - len = vsnprintf(buffer, sizeof(buffer)-1, fmt, argp); + len = vsnprintf(buffer, sizeof(buffer), fmt, argp); #else len = vsprintf(buffer, fmt, argp); #endif @@ -514,7 +524,7 @@ static int send_response (listen_socket_t *sock, response_code rc, rclen = sprintf(buffer, "%d ", lines); va_start(argp, fmt); #ifdef HAVE_VSNPRINTF - len = vsnprintf(buffer+rclen, sizeof(buffer)-rclen-1, fmt, argp); + len = vsnprintf(buffer+rclen, sizeof(buffer)-rclen, fmt, argp); #else len = vsprintf(buffer+rclen, fmt, argp); #endif @@ -587,6 +597,12 @@ static void remove_from_queue(cache_item_t *ci) /* {{{ */ ci->next = ci->prev = NULL; ci->flags &= ~CI_FLAGS_IN_QUEUE; + + pthread_mutex_lock (&stats_lock); + assert (stats_queue_length > 0); + stats_queue_length--; + pthread_mutex_unlock (&stats_lock); + } /* }}} static void remove_from_queue */ /* free the resources associated with the cache_item_t @@ -663,7 +679,7 @@ static int enqueue_cache_item (cache_item_t *ci, /* {{{ */ ci->flags |= CI_FLAGS_IN_QUEUE; - pthread_cond_broadcast(&cache_cond); + pthread_cond_signal(&queue_cond); pthread_mutex_lock (&stats_lock); stats_queue_length++; pthread_mutex_unlock (&stats_lock); @@ -673,7 +689,7 @@ static int enqueue_cache_item (cache_item_t *ci, /* {{{ */ /* * tree_callback_flush: - * Called via `g_tree_foreach' in `queue_thread_main'. `cache_lock' is held + * Called via `g_tree_foreach' in `flush_thread_main'. `cache_lock' is held * while this is in progress. */ static gboolean tree_callback_flush (gpointer key, gpointer value, /* {{{ */ @@ -759,27 +775,20 @@ static int flush_old_values (int max_age) return (0); } /* int flush_old_values */ -static void *queue_thread_main (void *args __attribute__((unused))) /* {{{ */ +static void *flush_thread_main (void *args __attribute__((unused))) /* {{{ */ { struct timeval now; struct timespec next_flush; - int final_flush = 0; /* make sure we only flush once on shutdown */ + int status; gettimeofday (&now, NULL); next_flush.tv_sec = now.tv_sec + config_flush_interval; next_flush.tv_nsec = 1000 * now.tv_usec; - pthread_mutex_lock (&cache_lock); - while ((do_shutdown == 0) || (cache_queue_head != NULL)) - { - cache_item_t *ci; - char *file; - char **values; - int values_num; - int status; - int i; + pthread_mutex_lock(&cache_lock); - /* First, check if it's time to do the cache flush. */ + while (!do_shutdown) + { gettimeofday (&now, NULL); if ((now.tv_sec > next_flush.tv_sec) || ((now.tv_sec == next_flush.tv_sec) @@ -800,28 +809,48 @@ static void *queue_thread_main (void *args __attribute__((unused))) /* {{{ */ pthread_mutex_lock(&cache_lock); } + status = pthread_cond_timedwait(&flush_cond, &cache_lock, &next_flush); + if (status != 0 && status != ETIMEDOUT) + { + RRDD_LOG (LOG_ERR, "flush_thread_main: " + "pthread_cond_timedwait returned %i.", status); + } + } + + if (config_flush_at_shutdown) + flush_old_values (-1); /* flush everything */ + + pthread_mutex_unlock(&cache_lock); + + return NULL; +} /* void *flush_thread_main */ + +static void *queue_thread_main (void *args __attribute__((unused))) /* {{{ */ +{ + pthread_mutex_lock (&cache_lock); + + while (!do_shutdown + || (cache_queue_head != NULL && config_flush_at_shutdown)) + { + cache_item_t *ci; + char *file; + char **values; + int values_num; + int status; + int i; + /* Now, check if there's something to store away. If not, wait until - * something comes in or it's time to do the cache flush. if we are - * shutting down, do not wait around. */ + * something comes in. if we are shutting down, do not wait around. */ if (cache_queue_head == NULL && !do_shutdown) { - status = pthread_cond_timedwait (&cache_cond, &cache_lock, &next_flush); + status = pthread_cond_wait (&queue_cond, &cache_lock); if ((status != 0) && (status != ETIMEDOUT)) { RRDD_LOG (LOG_ERR, "queue_thread_main: " - "pthread_cond_timedwait returned %i.", status); + "pthread_cond_wait returned %i.", status); } } - /* We're about to shut down */ - if (do_shutdown != 0 && !final_flush++) - { - if (config_flush_at_shutdown) - flush_old_values (-1); /* flush everything */ - else - break; - } - /* Check if a value has arrived. This may be NULL if we timed out or there * was an interrupt such as a signal. */ if (cache_queue_head == NULL) @@ -846,11 +875,6 @@ static void *queue_thread_main (void *args __attribute__((unused))) /* {{{ */ wipe_ci_values(ci, time(NULL)); remove_from_queue(ci); - pthread_mutex_lock (&stats_lock); - assert (stats_queue_length > 0); - stats_queue_length--; - pthread_mutex_unlock (&stats_lock); - pthread_mutex_unlock (&cache_lock); rrd_clear_error (); @@ -880,25 +904,8 @@ static void *queue_thread_main (void *args __attribute__((unused))) /* {{{ */ } pthread_mutex_lock (&cache_lock); - - /* We're about to shut down */ - if (do_shutdown != 0 && !final_flush++) - { - if (config_flush_at_shutdown) - flush_old_values (-1); /* flush everything */ - else - break; - } - } /* while ((do_shutdown == 0) || (cache_queue_head != NULL)) */ - pthread_mutex_unlock (&cache_lock); - - if (config_flush_at_shutdown) - { - assert(cache_queue_head == NULL); - RRDD_LOG(LOG_INFO, "clean shutdown; all RRDs flushed"); } - - journal_done(); + pthread_mutex_unlock (&cache_lock); return (NULL); } /* }}} void *queue_thread_main */ @@ -1090,7 +1097,7 @@ static int handle_request_help (listen_socket_t *sock, /* {{{ */ "Usage: FLUSH \n" "\n" "Adds the given filename to the head of the update queue and returns\n" - "after is has been dequeued.\n" + "after it has been dequeued.\n" }; char *help_flushall[2] = @@ -2152,7 +2159,7 @@ static int open_listen_socket_network(const listen_socket_t *sock) /* {{{ */ char *port; int status; - strncpy (addr_copy, sock->addr, sizeof (addr_copy)); + strncpy (addr_copy, sock->addr, sizeof(addr_copy)-1); addr_copy[sizeof (addr_copy) - 1] = 0; addr = addr_copy; @@ -2444,7 +2451,7 @@ static int daemonize (void) /* {{{ */ { listen_socket_t sock; memset(&sock, 0, sizeof(sock)); - strncpy(sock.addr, RRDCACHED_DEFAULT_ADDRESS, sizeof(sock.addr)); + strncpy(sock.addr, RRDCACHED_DEFAULT_ADDRESS, sizeof(sock.addr)-1); open_listen_socket (&sock); } @@ -2515,11 +2522,23 @@ static int cleanup (void) /* {{{ */ { do_shutdown++; - pthread_cond_signal (&cache_cond); - pthread_join (queue_thread, /* return = */ NULL); + pthread_cond_broadcast (&flush_cond); + pthread_join (flush_thread, NULL); + + pthread_cond_broadcast (&queue_cond); + for (int i = 0; i < config_queue_threads; i++) + pthread_join (queue_threads[i], NULL); + if (config_flush_at_shutdown) + { + assert(cache_queue_head == NULL); + RRDD_LOG(LOG_INFO, "clean shutdown; all RRDs flushed"); + } + + journal_done(); remove_pidfile (); + free(queue_threads); free(config_base_dir); free(config_pid_file); free(journal_cur); @@ -2539,7 +2558,7 @@ static int read_options (int argc, char **argv) /* {{{ */ int option; int status = 0; - while ((option = getopt(argc, argv, "gl:L:f:w:b:Bz:p:j:h?F")) != -1) + while ((option = getopt(argc, argv, "gl:L:f:w:z:t:Bb:p:Fj:h?")) != -1) { switch (option) { @@ -2624,6 +2643,20 @@ static int read_options (int argc, char **argv) /* {{{ */ break; } + case 't': + { + int threads; + threads = atoi(optarg); + if (threads >= 1) + config_queue_threads = threads; + else + { + fprintf (stderr, "Invalid thread count: -t %s\n", optarg); + return 1; + } + } + break; + case 'B': config_write_base_only = 1; break; @@ -2743,6 +2776,7 @@ static int read_options (int argc, char **argv) /* {{{ */ " -L
Socket address to listen to ('FLUSH' only).\n" " -w Interval in which to write data.\n" " -z Delay writes up to seconds to spread load\n" + " -t Number of write threads.\n" " -f Interval in which to flush dead data.\n" " -p Location of the PID-file.\n" " -b Base directory to change to.\n" @@ -2799,15 +2833,32 @@ int main (int argc, char **argv) journal_init(); - /* start the queue thread */ - memset (&queue_thread, 0, sizeof (queue_thread)); - status = pthread_create (&queue_thread, - NULL, /* attr */ - queue_thread_main, - NULL); /* args */ + /* start the queue threads */ + queue_threads = calloc(config_queue_threads, sizeof(*queue_threads)); + if (queue_threads == NULL) + { + RRDD_LOG (LOG_ERR, "FATAL: cannot calloc queue threads"); + cleanup(); + return (1); + } + for (int i = 0; i < config_queue_threads; i++) + { + memset (&queue_threads[i], 0, sizeof (*queue_threads)); + status = pthread_create (&queue_threads[i], NULL, queue_thread_main, NULL); + if (status != 0) + { + RRDD_LOG (LOG_ERR, "FATAL: cannot create queue thread"); + cleanup(); + return (1); + } + } + + /* start the flush thread */ + memset(&flush_thread, 0, sizeof(flush_thread)); + status = pthread_create (&flush_thread, NULL, flush_thread_main, NULL); if (status != 0) { - RRDD_LOG (LOG_ERR, "FATAL: cannot create queue thread"); + RRDD_LOG (LOG_ERR, "FATAL: cannot create flush thread"); cleanup(); return (1); }