X-Git-Url: https://git.octo.it/?p=rrdtool.git;a=blobdiff_plain;f=src%2Frrd_daemon.c;h=6445e55d6bad10ad207ea4d64ff73e4a413d4cee;hp=52e9f12ba06301157db4a80ea4dce0164d19d490;hb=d4110e29da41ce702bcc3327e86768c6f266915e;hpb=f942d3c825ad1073bb6d68edd9653b0c999bbe49 diff --git a/src/rrd_daemon.c b/src/rrd_daemon.c index 52e9f12..6445e55 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))) /* {{{ */ @@ -429,7 +439,7 @@ static int add_to_wbuf(listen_socket_t *sock, char *str, size_t len) /* {{{ */ assert(sock != NULL); - new_buf = realloc(sock->wbuf, sock->wbuf_len + len + 1); + new_buf = rrd_realloc(sock->wbuf, sock->wbuf_len + len + 1); if (new_buf == NULL) { RRDD_LOG(LOG_ERR, "add_to_wbuf: realloc failed"); @@ -587,20 +597,21 @@ 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 */ -/* remove an entry from the tree and free all its resources. - * must hold 'cache lock' while calling this. - * returns 0 on success, otherwise errno */ -static int forget_file(const char *file) +/* free the resources associated with the cache_item_t + * must hold cache_lock when calling this function + */ +static void *free_cache_item(cache_item_t *ci) /* {{{ */ { - cache_item_t *ci; - - ci = g_tree_lookup(cache_tree, file); - if (ci == NULL) - return ENOENT; + if (ci == NULL) return NULL; - g_tree_remove (cache_tree, file); remove_from_queue(ci); for (int i=0; i < ci->values_num; i++) @@ -614,8 +625,8 @@ static int forget_file(const char *file) free (ci); - return 0; -} /* }}} static int forget_file */ + return NULL; +} /* }}} static void *free_cache_item */ /* * enqueue_cache_item: @@ -668,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); @@ -678,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, /* {{{ */ @@ -708,7 +719,7 @@ static gboolean tree_callback_flush (gpointer key, gpointer value, /* {{{ */ { char **temp; - temp = (char **) realloc (cfd->keys, + temp = (char **) rrd_realloc (cfd->keys, sizeof (char *) * (cfd->keys_num + 1)); if (temp == NULL) { @@ -752,7 +763,7 @@ static int flush_old_values (int max_age) { /* should never fail, since we have held the cache_lock * the entire time */ - assert( forget_file(cfd.keys[k]) == 0 ); + assert( g_tree_remove(cache_tree, cfd.keys[k]) == TRUE ); } if (cfd.keys != NULL) @@ -764,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) @@ -805,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) @@ -851,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 (); @@ -885,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 */ @@ -1081,9 +1083,11 @@ static int handle_request_help (listen_socket_t *sock, /* {{{ */ "FLUSHALL\n" "PENDING \n" "FORGET \n" + "QUEUE\n" "UPDATE [ ...]\n" "BATCH\n" "STATS\n" + "QUIT\n" }; char *help_flush[2] = @@ -1093,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] = @@ -1125,6 +1129,18 @@ static int handle_request_help (listen_socket_t *sock, /* {{{ */ "Any pending updates for the file will be lost.\n" }; + char *help_queue[2] = + { + "Help for QUEUE\n" + , + "Shows all files in the output queue.\n" + "The output is zero or more lines in the following format:\n" + "(where is the number of values to be written)\n" + "\n" + " \n" + "\n" + }; + char *help_update[2] = { "Help for UPDATE\n" @@ -1172,6 +1188,13 @@ static int handle_request_help (listen_socket_t *sock, /* {{{ */ "For more information, consult the rrdcached(1) documentation.\n" }; + char *help_quit[2] = + { + "Help for QUIT\n" + , + "Disconnect from rrdcached.\n" + }; + status = buffer_get_field (&buffer, &buffer_size, &command); if (status != 0) help_text = help_help; @@ -1187,10 +1210,14 @@ static int handle_request_help (listen_socket_t *sock, /* {{{ */ help_text = help_pending; else if (strcasecmp (command, "forget") == 0) help_text = help_forget; + else if (strcasecmp (command, "queue") == 0) + help_text = help_queue; else if (strcasecmp (command, "stats") == 0) help_text = help_stats; else if (strcasecmp (command, "batch") == 0) help_text = help_batch; + else if (strcasecmp (command, "quit") == 0) + help_text = help_quit; else help_text = help_help; } @@ -1345,6 +1372,7 @@ static int handle_request_forget(listen_socket_t *sock, /* {{{ */ char *buffer, size_t buffer_size) { int status; + gboolean found; char *file, file_tmp[PATH_MAX]; status = buffer_get_field(&buffer, &buffer_size, &file); @@ -1360,10 +1388,10 @@ static int handle_request_forget(listen_socket_t *sock, /* {{{ */ if (!check_file_access(file, sock)) return 0; pthread_mutex_lock(&cache_lock); - status = forget_file(file); + found = g_tree_remove(cache_tree, file); pthread_mutex_unlock(&cache_lock); - if (status == 0) + if (found == TRUE) { if (sock != NULL) journal_write("forget", file); @@ -1371,13 +1399,30 @@ static int handle_request_forget(listen_socket_t *sock, /* {{{ */ return send_response(sock, RESP_OK, "Gone!\n"); } else - return send_response(sock, RESP_ERR, "cannot forget: %s\n", - status < 0 ? "Internal error" : rrd_strerror(status)); + return send_response(sock, RESP_ERR, "%s\n", rrd_strerror(ENOENT)); /* NOTREACHED */ assert(1==0); } /* }}} static int handle_request_forget */ +static int handle_request_queue (listen_socket_t *sock) /* {{{ */ +{ + cache_item_t *ci; + + pthread_mutex_lock(&cache_lock); + + ci = cache_queue_head; + while (ci != NULL) + { + add_response_info(sock, "%d %s\n", ci->values_num, ci->file); + ci = ci->next; + } + + pthread_mutex_unlock(&cache_lock); + + return send_response(sock, RESP_OK, "in queue.\n"); +} /* }}} int handle_request_queue */ + static int handle_request_update (listen_socket_t *sock, /* {{{ */ time_t now, char *buffer, size_t buffer_size) @@ -1461,7 +1506,7 @@ static int handle_request_update (listen_socket_t *sock, /* {{{ */ pthread_cond_init(&ci->flushed, NULL); pthread_mutex_lock(&cache_lock); - g_tree_insert (cache_tree, (void *) ci->file, (void *) ci); + g_tree_replace (cache_tree, (void *) ci->file, (void *) ci); } /* }}} */ assert (ci != NULL); @@ -1502,7 +1547,7 @@ static int handle_request_update (listen_socket_t *sock, /* {{{ */ else ci->last_update_stamp = stamp; - temp = (char **) realloc (ci->values, + temp = (char **) rrd_realloc (ci->values, sizeof (char *) * (ci->values_num + 1)); if (temp == NULL) { @@ -1637,6 +1682,8 @@ static int handle_request (listen_socket_t *sock, /* {{{ */ return (handle_request_pending(sock, buffer_ptr, buffer_size)); else if (strcasecmp (command, "forget") == 0) return (handle_request_forget(sock, buffer_ptr, buffer_size)); + else if (strcasecmp (command, "queue") == 0) + return (handle_request_queue(sock)); else if (strcasecmp (command, "stats") == 0) return (handle_request_stats (sock)); else if (strcasecmp (command, "help") == 0) @@ -1645,6 +1692,8 @@ static int handle_request (listen_socket_t *sock, /* {{{ */ return batch_start(sock); else if (strcasecmp (command, ".") == 0 && sock != NULL && sock->batch_start) return batch_done(sock); + else if (strcasecmp (command, "quit") == 0) + return -1; else return send_response(sock, RESP_ERR, "Unknown command: %s\n", command); @@ -1764,7 +1813,7 @@ static int journal_replay (const char *file) /* {{{ */ if (file == NULL) return 0; { - char *reason; + char *reason = "unknown error"; int status = 0; struct stat statbuf; @@ -1873,18 +1922,29 @@ static void journal_init(void) /* {{{ */ } /* }}} static void journal_init */ -static void close_connection(listen_socket_t *sock) +static void free_listen_socket(listen_socket_t *sock) /* {{{ */ { - close(sock->fd) ; sock->fd = -1; + assert(sock != NULL); + free(sock->rbuf); sock->rbuf = NULL; free(sock->wbuf); sock->wbuf = NULL; - free(sock); -} +} /* }}} void free_listen_socket */ + +static void close_connection(listen_socket_t *sock) /* {{{ */ +{ + if (sock->fd >= 0) + { + close(sock->fd); + sock->fd = -1; + } + + free_listen_socket(sock); + +} /* }}} void close_connection */ static void *connection_thread_main (void *args) /* {{{ */ { - pthread_t self; listen_socket_t *sock; int i; int fd; @@ -1906,11 +1966,11 @@ static void *connection_thread_main (void *args) /* {{{ */ { pthread_t *temp; - temp = (pthread_t *) realloc (connection_threads, + temp = (pthread_t *) rrd_realloc (connection_threads, sizeof (pthread_t) * (connection_threads_num + 1)); if (temp == NULL) { - RRDD_LOG (LOG_ERR, "connection_thread_main: realloc failed."); + RRDD_LOG (LOG_ERR, "connection_thread_main: realloc(++) failed."); } else { @@ -1986,24 +2046,36 @@ static void *connection_thread_main (void *args) /* {{{ */ out_close: close_connection(sock); - self = pthread_self (); /* Remove this thread from the connection threads list */ pthread_mutex_lock (&connection_threads_lock); - /* Find out own index in the array */ - for (i = 0; i < connection_threads_num; i++) - if (pthread_equal (connection_threads[i], self) != 0) - break; - assert (i < connection_threads_num); - - /* Move the trailing threads forward. */ - if (i < (connection_threads_num - 1)) { - memmove (connection_threads + i, - connection_threads + i + 1, - sizeof (pthread_t) * (connection_threads_num - i - 1)); - } + pthread_t self; + pthread_t *temp; + + /* Find out own index in the array */ + self = pthread_self (); + for (i = 0; i < connection_threads_num; i++) + if (pthread_equal (connection_threads[i], self) != 0) + break; + assert (i < connection_threads_num); + + /* Move the trailing threads forward. */ + if (i < (connection_threads_num - 1)) + { + memmove (connection_threads + i, + connection_threads + i + 1, + sizeof (pthread_t) * (connection_threads_num - i - 1)); + } + + connection_threads_num--; - connection_threads_num--; + temp = rrd_realloc(connection_threads, + sizeof(*connection_threads) * connection_threads_num); + if (connection_threads_num > 0 && temp == NULL) + RRDD_LOG(LOG_ERR, "connection_thread_main: realloc(--) failed."); + else + connection_threads = temp; + } pthread_mutex_unlock (&connection_threads_lock); return (NULL); @@ -2021,7 +2093,7 @@ static int open_listen_socket_unix (const listen_socket_t *sock) /* {{{ */ if (strncmp(path, "unix:", strlen("unix:")) == 0) path += strlen("unix:"); - temp = (listen_socket_t *) realloc (listen_fds, + temp = (listen_socket_t *) rrd_realloc (listen_fds, sizeof (listen_fds[0]) * (listen_fds_num + 1)); if (temp == NULL) { @@ -2150,7 +2222,7 @@ static int open_listen_socket_network(const listen_socket_t *sock) /* {{{ */ listen_socket_t *temp; int one = 1; - temp = (listen_socket_t *) realloc (listen_fds, + temp = (listen_socket_t *) rrd_realloc (listen_fds, sizeof (listen_fds[0]) * (listen_fds_num + 1)); if (temp == NULL) { @@ -2186,6 +2258,7 @@ static int open_listen_socket_network(const listen_socket_t *sock) /* {{{ */ fprintf (stderr, "rrdcached: listen(%s) failed: %s\n.", sock->addr, rrd_strerror(errno)); close (fd); + freeaddrinfo(ai_res); return (-1); } @@ -2194,6 +2267,7 @@ static int open_listen_socket_network(const listen_socket_t *sock) /* {{{ */ listen_fds_num++; } /* for (ai_ptr) */ + freeaddrinfo(ai_res); return (0); } /* }}} static int open_listen_socket_network */ @@ -2254,7 +2328,6 @@ static void *listen_thread_main (void *args __attribute__((unused))) /* {{{ */ while (do_shutdown == 0) { - assert (pollfds_num == ((int) listen_fds_num)); for (i = 0; i < pollfds_num; i++) { pollfds[i].fd = listen_fds[i].fd; @@ -2345,6 +2418,8 @@ static void *listen_thread_main (void *args __attribute__((unused))) /* {{{ */ } pthread_mutex_unlock (&connection_threads_lock); + free(pollfds); + return (NULL); } /* }}} void *listen_thread_main */ @@ -2365,7 +2440,12 @@ static int daemonize (void) /* {{{ */ if (config_listen_address_list_len > 0) { for (int i = 0; i < config_listen_address_list_len; i++) + { open_listen_socket (config_listen_address_list[i]); + free_listen_socket (config_listen_address_list[i]); + } + + free(config_listen_address_list); } else { @@ -2423,7 +2503,8 @@ static int daemonize (void) /* {{{ */ openlog ("rrdcached", LOG_PID, LOG_DAEMON); RRDD_LOG(LOG_INFO, "starting up"); - cache_tree = g_tree_new ((GCompareFunc) strcmp); + cache_tree = g_tree_new_full ((GCompareDataFunc) strcmp, NULL, NULL, + (GDestroyNotify) free_cache_item); if (cache_tree == NULL) { RRDD_LOG (LOG_ERR, "daemonize: g_tree_new failed."); @@ -2441,11 +2522,31 @@ 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); + free(journal_old); + + pthread_mutex_lock(&cache_lock); + g_tree_destroy(cache_tree); + RRDD_LOG(LOG_INFO, "goodbye"); closelog (); @@ -2457,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) { @@ -2479,7 +2580,7 @@ static int read_options (int argc, char **argv) /* {{{ */ } memset(new, 0, sizeof(listen_socket_t)); - temp = (listen_socket_t **) realloc (config_listen_address_list, + temp = (listen_socket_t **) rrd_realloc (config_listen_address_list, sizeof (listen_socket_t *) * (config_listen_address_list_len + 1)); if (temp == NULL) { @@ -2542,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; @@ -2661,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" @@ -2717,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); }