X-Git-Url: https://git.octo.it/?p=rrdtool.git;a=blobdiff_plain;f=src%2Frrd_daemon.c;h=d56cf06da51ddde4126d302557b9a3915f257244;hp=bee21a2d7e6e248cb49f36b9983694892caed058;hb=8df853711a3c776d77fed52b6393b09e6289963a;hpb=5d074b0e4c13e9345e442aec5cf02b9acd198738 diff --git a/src/rrd_daemon.c b/src/rrd_daemon.c index bee21a2..d56cf06 100644 --- a/src/rrd_daemon.c +++ b/src/rrd_daemon.c @@ -62,7 +62,16 @@ /* * 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 @@ -420,7 +429,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"); @@ -580,18 +589,13 @@ static void remove_from_queue(cache_item_t *ci) /* {{{ */ ci->flags &= ~CI_FLAGS_IN_QUEUE; } /* }}} 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++) @@ -605,8 +609,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: @@ -699,7 +703,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) { @@ -743,7 +747,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) @@ -1072,9 +1076,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] = @@ -1116,6 +1122,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 +1181,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 +1203,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 +1365,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 +1381,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 +1392,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,10 +1496,10 @@ static int handle_request_update (listen_socket_t *sock, /* {{{ */ wipe_ci_values(ci, now); ci->flags = CI_FLAGS_IN_TREE; - ci->flushed = PTHREAD_COND_INITIALIZER; + 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); @@ -1479,23 +1525,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) { @@ -1525,21 +1570,7 @@ static int handle_request_update (listen_socket_t *sock, /* {{{ */ pthread_mutex_unlock (&cache_lock); if (values_num < 1) - { - /* journal replay mode */ - if (sock == NULL) return RESP_ERR; - - /* 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); @@ -1644,6 +1675,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) @@ -1652,6 +1685,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); @@ -1771,7 +1806,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; @@ -1880,18 +1915,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; @@ -1913,11 +1959,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 { @@ -1993,24 +2039,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; - connection_threads_num--; + /* 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--; + + 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); @@ -2028,7 +2086,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) { @@ -2157,7 +2215,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) { @@ -2193,6 +2251,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); } @@ -2201,6 +2260,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 */ @@ -2261,7 +2321,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; @@ -2352,6 +2411,8 @@ static void *listen_thread_main (void *args __attribute__((unused))) /* {{{ */ } pthread_mutex_unlock (&connection_threads_lock); + free(pollfds); + return (NULL); } /* }}} void *listen_thread_main */ @@ -2372,7 +2433,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 { @@ -2430,7 +2496,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."); @@ -2453,6 +2520,14 @@ static int cleanup (void) /* {{{ */ remove_pidfile (); + 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 (); @@ -2486,7 +2561,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) {