X-Git-Url: https://git.octo.it/?p=rrdtool.git;a=blobdiff_plain;f=src%2Frrd_daemon.c;h=6445e55d6bad10ad207ea4d64ff73e4a413d4cee;hp=9fbbc1133af5a79869a53e84657086b91d4f0b60;hb=d4110e29da41ce702bcc3327e86768c6f266915e;hpb=607eea59c2ba85c63e2de1b665b37a6093e1575c diff --git a/src/rrd_daemon.c b/src/rrd_daemon.c index 9fbbc11..6445e55 100644 --- a/src/rrd_daemon.c +++ b/src/rrd_daemon.c @@ -62,23 +62,37 @@ /* * Now for some includes.. */ -#include "rrd.h" /* {{{ */ +/* {{{ */ +#if defined(_WIN32) && !defined(__CYGWIN__) && !defined(__CYGWIN32__) && !defined(HAVE_CONFIG_H) +#include "../win32/config.h" +#else +#ifdef HAVE_CONFIG_H +#include "../rrd_config.h" +#endif +#endif + +#include "rrd.h" #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 @@ -178,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; @@ -189,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; @@ -228,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))) /* {{{ */ @@ -420,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"); @@ -564,6 +583,7 @@ static void wipe_ci_values(cache_item_t *ci, time_t when) static void remove_from_queue(cache_item_t *ci) /* {{{ */ { if (ci == NULL) return; + if ((ci->flags & CI_FLAGS_IN_QUEUE) == 0) return; /* not queued */ if (ci->prev == NULL) cache_queue_head = ci->next; /* reset head */ @@ -577,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; + if (ci == NULL) return NULL; - ci = g_tree_lookup(cache_tree, file); - if (ci == NULL) - return ENOENT; - - g_tree_remove (cache_tree, file); remove_from_queue(ci); for (int i=0; i < ci->values_num; i++) @@ -604,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: @@ -625,9 +646,8 @@ static int enqueue_cache_item (cache_item_t *ci, /* {{{ */ if (cache_queue_head == ci) return 0; - /* remove from the double linked list */ - if (ci->flags & CI_FLAGS_IN_QUEUE) - remove_from_queue(ci); + /* remove if further down in queue */ + remove_from_queue(ci); ci->prev = NULL; ci->next = cache_queue_head; @@ -659,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); @@ -669,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, /* {{{ */ @@ -681,25 +701,25 @@ static gboolean tree_callback_flush (gpointer key, gpointer value, /* {{{ */ ci = (cache_item_t *) value; cfd = (callback_flush_data_t *) data; + if (ci->flags & CI_FLAGS_IN_QUEUE) + return FALSE; + if ((ci->last_flush_time <= cfd->abs_timeout) - && ((ci->flags & CI_FLAGS_IN_QUEUE) == 0) && (ci->values_num > 0)) { enqueue_cache_item (ci, TAIL); } else if ((do_shutdown != 0) - && ((ci->flags & CI_FLAGS_IN_QUEUE) == 0) && (ci->values_num > 0)) { enqueue_cache_item (ci, TAIL); } else if (((cfd->now - ci->last_flush_time) >= config_flush_interval) - && ((ci->flags & CI_FLAGS_IN_QUEUE) == 0) && (ci->values_num <= 0)) { char **temp; - temp = (char **) realloc (cfd->keys, + temp = (char **) rrd_realloc (cfd->keys, sizeof (char *) * (cfd->keys_num + 1)); if (temp == NULL) { @@ -743,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) @@ -755,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) @@ -796,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) @@ -842,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 (); @@ -876,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 */ @@ -1072,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] = @@ -1084,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] = @@ -1116,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" @@ -1163,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; @@ -1178,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; } @@ -1336,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); @@ -1351,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); @@ -1362,20 +1399,36 @@ 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) { char *file, file_tmp[PATH_MAX]; int values_num = 0; - int bad_timestamps = 0; int status; char orig_buf[CMD_MAX]; @@ -1450,9 +1503,10 @@ static int handle_request_update (listen_socket_t *sock, /* {{{ */ wipe_ci_values(ci, now); ci->flags = CI_FLAGS_IN_TREE; + 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); @@ -1478,23 +1532,22 @@ static int handle_request_update (listen_socket_t *sock, /* {{{ */ stamp = strtol(value, &eostamp, 10); if (eostamp == value || eostamp == NULL || *eostamp != ':') { - ++bad_timestamps; - add_response_info(sock, "Cannot find timestamp in '%s'!\n", value); - continue; + pthread_mutex_unlock(&cache_lock); + return send_response(sock, RESP_ERR, + "Cannot find timestamp in '%s'!\n", value); } else if (stamp <= ci->last_update_stamp) { - ++bad_timestamps; - add_response_info(sock, - "illegal attempt to update using time %ld when" - " last update time is %ld (minimum one second step)\n", - stamp, ci->last_update_stamp); - continue; + pthread_mutex_unlock(&cache_lock); + return send_response(sock, RESP_ERR, + "illegal attempt to update using time %ld when last" + " update time is %ld (minimum one second step)\n", + stamp, ci->last_update_stamp); } 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) { @@ -1524,18 +1577,7 @@ static int handle_request_update (listen_socket_t *sock, /* {{{ */ pthread_mutex_unlock (&cache_lock); if (values_num < 1) - { - /* if we had only one update attempt, then return the full - error message... try to get the most information out - of the limited error space allowed by the protocol - */ - if (bad_timestamps == 1) - return send_response(sock, RESP_ERR, "%s", sock->wbuf); - else - return send_response(sock, RESP_ERR, - "No values updated (%d bad timestamps).\n", - bad_timestamps); - } + return send_response(sock, RESP_ERR, "No values updated.\n"); else return send_response(sock, RESP_OK, "errors, enqueued %i value(s).\n", values_num); @@ -1640,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) @@ -1648,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); @@ -1767,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; @@ -1876,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; @@ -1909,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 { @@ -1989,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); - 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--; + + 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); @@ -2024,11 +2093,11 @@ 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) { - RRDD_LOG (LOG_ERR, "open_listen_socket_unix: realloc failed."); + fprintf (stderr, "rrdcached: open_listen_socket_unix: realloc failed.\n"); return (-1); } listen_fds = temp; @@ -2037,7 +2106,8 @@ static int open_listen_socket_unix (const listen_socket_t *sock) /* {{{ */ fd = socket (PF_UNIX, SOCK_STREAM, /* protocol = */ 0); if (fd < 0) { - RRDD_LOG (LOG_ERR, "open_listen_socket_unix: socket(2) failed."); + fprintf (stderr, "rrdcached: unix socket(2) failed: %s\n", + rrd_strerror(errno)); return (-1); } @@ -2045,19 +2115,26 @@ static int open_listen_socket_unix (const listen_socket_t *sock) /* {{{ */ sa.sun_family = AF_UNIX; strncpy (sa.sun_path, path, sizeof (sa.sun_path) - 1); + /* if we've gotten this far, we own the pid file. any daemon started + * with the same args must not be alive. therefore, ensure that we can + * create the socket... + */ + unlink(path); + status = bind (fd, (struct sockaddr *) &sa, sizeof (sa)); if (status != 0) { - RRDD_LOG (LOG_ERR, "open_listen_socket_unix: bind(2) failed."); + fprintf (stderr, "rrdcached: bind(%s) failed: %s.\n", + path, rrd_strerror(errno)); close (fd); - unlink (path); return (-1); } status = listen (fd, /* backlog = */ 10); if (status != 0) { - RRDD_LOG (LOG_ERR, "open_listen_socket_unix: listen(2) failed."); + fprintf (stderr, "rrdcached: listen(%s) failed: %s.\n", + path, rrd_strerror(errno)); close (fd); unlink (path); return (-1); @@ -2103,8 +2180,7 @@ static int open_listen_socket_network(const listen_socket_t *sock) /* {{{ */ port = strchr (addr, ']'); if (port == NULL) { - RRDD_LOG (LOG_ERR, "open_listen_socket_network: Malformed address: %s", - sock->addr); + fprintf (stderr, "rrdcached: Malformed address: %s\n", sock->addr); return (-1); } *port = 0; @@ -2116,8 +2192,7 @@ static int open_listen_socket_network(const listen_socket_t *sock) /* {{{ */ port = NULL; else { - RRDD_LOG (LOG_ERR, "open_listen_socket_network: Garbage after address: %s", - port); + fprintf (stderr, "rrdcached: Garbage after address: %s\n", port); return (-1); } } /* if (*addr = ']') */ @@ -2136,8 +2211,8 @@ static int open_listen_socket_network(const listen_socket_t *sock) /* {{{ */ &ai_hints, &ai_res); if (status != 0) { - RRDD_LOG (LOG_ERR, "open_listen_socket_network: getaddrinfo(%s) failed: " - "%s", addr, gai_strerror (status)); + fprintf (stderr, "rrdcached: getaddrinfo(%s) failed: %s\n", + addr, gai_strerror (status)); return (-1); } @@ -2147,11 +2222,12 @@ 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) { - RRDD_LOG (LOG_ERR, "open_listen_socket_network: realloc failed."); + fprintf (stderr, + "rrdcached: open_listen_socket_network: realloc failed.\n"); continue; } listen_fds = temp; @@ -2160,7 +2236,8 @@ static int open_listen_socket_network(const listen_socket_t *sock) /* {{{ */ fd = socket (ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol); if (fd < 0) { - RRDD_LOG (LOG_ERR, "open_listen_socket_network: socket(2) failed."); + fprintf (stderr, "rrdcached: network socket(2) failed: %s.\n", + rrd_strerror(errno)); continue; } @@ -2169,7 +2246,8 @@ static int open_listen_socket_network(const listen_socket_t *sock) /* {{{ */ status = bind (fd, ai_ptr->ai_addr, ai_ptr->ai_addrlen); if (status != 0) { - RRDD_LOG (LOG_ERR, "open_listen_socket_network: bind(2) failed."); + fprintf (stderr, "rrdcached: bind(%s) failed: %s.\n", + sock->addr, rrd_strerror(errno)); close (fd); continue; } @@ -2177,8 +2255,10 @@ static int open_listen_socket_network(const listen_socket_t *sock) /* {{{ */ status = listen (fd, /* backlog = */ 10); if (status != 0) { - RRDD_LOG (LOG_ERR, "open_listen_socket_network: listen(2) failed."); + fprintf (stderr, "rrdcached: listen(%s) failed: %s\n.", + sock->addr, rrd_strerror(errno)); close (fd); + freeaddrinfo(ai_res); return (-1); } @@ -2187,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 */ @@ -2228,21 +2309,9 @@ static void *listen_thread_main (void *args __attribute__((unused))) /* {{{ */ int status; int i; - for (i = 0; i < config_listen_address_list_len; i++) - open_listen_socket (config_listen_address_list[i]); - - if (config_listen_address_list_len < 1) - { - listen_socket_t sock; - memset(&sock, 0, sizeof(sock)); - strncpy(sock.addr, RRDCACHED_DEFAULT_ADDRESS, sizeof(sock.addr)); - open_listen_socket (&sock); - } - if (listen_fds_num < 1) { - RRDD_LOG (LOG_ERR, "listen_thread_main: No listen sockets " - "could be opened. Sorry."); + RRDD_LOG(LOG_ERR, "listen_thread_main: no listen_fds !"); return (NULL); } @@ -2259,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; @@ -2350,12 +2418,13 @@ static void *listen_thread_main (void *args __attribute__((unused))) /* {{{ */ } pthread_mutex_unlock (&connection_threads_lock); + free(pollfds); + return (NULL); } /* }}} void *listen_thread_main */ static int daemonize (void) /* {{{ */ { - int status; int pid_fd; char *base_dir; @@ -2367,6 +2436,31 @@ static int daemonize (void) /* {{{ */ if (pid_fd < 0) return pid_fd; + /* open all the listen sockets */ + 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 + { + listen_socket_t sock; + memset(&sock, 0, sizeof(sock)); + strncpy(sock.addr, RRDCACHED_DEFAULT_ADDRESS, sizeof(sock.addr)); + open_listen_socket (&sock); + } + + if (listen_fds_num < 1) + { + fprintf (stderr, "rrdcached: FATAL: cannot open any listen sockets\n"); + goto error; + } + if (!stay_foreground) { pid_t child; @@ -2375,12 +2469,10 @@ static int daemonize (void) /* {{{ */ if (child < 0) { fprintf (stderr, "daemonize: fork(2) failed.\n"); - return (-1); + goto error; } else if (child > 0) - { - return (1); - } + exit(0); /* Become session leader */ setsid (); @@ -2399,11 +2491,11 @@ static int daemonize (void) /* {{{ */ base_dir = (config_base_dir != NULL) ? config_base_dir : "/tmp"; - status = chdir (base_dir); - if (status != 0) + + if (chdir (base_dir) != 0) { fprintf (stderr, "daemonize: chdir (%s) failed.\n", base_dir); - return (-1); + goto error; } install_signal_handlers(); @@ -2411,26 +2503,50 @@ 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."); - return (-1); + goto error; } - status = write_pidfile (pid_fd); - return status; + return write_pidfile (pid_fd); + +error: + remove_pidfile(); + return -1; } /* }}} int daemonize */ 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 (); @@ -2442,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) { @@ -2464,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) { @@ -2527,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; @@ -2646,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" @@ -2694,33 +2825,40 @@ int main (int argc, char **argv) } status = daemonize (); - if (status == 1) + if (status != 0) { - struct sigaction sigchld; + fprintf (stderr, "rrdcached: daemonize failed, exiting.\n"); + return (1); + } - memset (&sigchld, 0, sizeof (sigchld)); - sigchld.sa_handler = SIG_IGN; - sigaction (SIGCHLD, &sigchld, NULL); + journal_init(); - return (0); - } - else if (status != 0) + /* start the queue threads */ + queue_threads = calloc(config_queue_threads, sizeof(*queue_threads)); + if (queue_threads == NULL) { - fprintf (stderr, "daemonize failed, exiting.\n"); + 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); + } + } - 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 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); }