X-Git-Url: https://git.octo.it/?p=rrdtool.git;a=blobdiff_plain;f=src%2Frrd_daemon.c;h=6445e55d6bad10ad207ea4d64ff73e4a413d4cee;hp=ea607d83df80810ddecd660e4bd5552cb05f5a22;hb=d4110e29da41ce702bcc3327e86768c6f266915e;hpb=814c75ac076932467801c10ff48f8cd6534fde5a diff --git a/src/rrd_daemon.c b/src/rrd_daemon.c index ea607d8..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 @@ -117,7 +131,7 @@ struct listen_socket_s socket_privilege privilege; /* state for BATCH processing */ - int batch_mode; + time_t batch_start; int batch_cmd; /* buffered IO */ @@ -138,6 +152,7 @@ struct cache_item_s char **values; int values_num; time_t last_flush_time; + time_t last_update_stamp; #define CI_FLAGS_IN_TREE (1<<0) #define CI_FLAGS_IN_QUEUE (1<<1) int flags; @@ -177,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; @@ -188,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; @@ -227,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))) /* {{{ */ @@ -285,7 +305,7 @@ static void install_signal_handlers(void) /* {{{ */ } /* }}} void install_signal_handlers */ -static int open_pidfile(void) /* {{{ */ +static int open_pidfile(char *action, int oflag) /* {{{ */ { int fd; char *file; @@ -294,14 +314,52 @@ static int open_pidfile(void) /* {{{ */ ? config_pid_file : LOCALSTATEDIR "/run/rrdcached.pid"; - fd = open(file, O_CREAT|O_EXCL|O_WRONLY, S_IRUSR|S_IRGRP|S_IROTH); + fd = open(file, oflag, S_IWUSR|S_IRUSR|S_IRGRP|S_IROTH); if (fd < 0) - fprintf(stderr, "FATAL: cannot create '%s' (%s)\n", - file, rrd_strerror(errno)); + fprintf(stderr, "rrdcached: can't %s pid file '%s' (%s)\n", + action, file, rrd_strerror(errno)); return(fd); } /* }}} static int open_pidfile */ +/* check existing pid file to see whether a daemon is running */ +static int check_pidfile(void) +{ + int pid_fd; + pid_t pid; + char pid_str[16]; + + pid_fd = open_pidfile("open", O_RDWR); + if (pid_fd < 0) + return pid_fd; + + if (read(pid_fd, pid_str, sizeof(pid_str)) <= 0) + return -1; + + pid = atoi(pid_str); + if (pid <= 0) + return -1; + + /* another running process that we can signal COULD be + * a competing rrdcached */ + if (pid != getpid() && kill(pid, 0) == 0) + { + fprintf(stderr, + "FATAL: Another rrdcached daemon is running?? (pid %d)\n", pid); + close(pid_fd); + return -1; + } + + lseek(pid_fd, 0, SEEK_SET); + ftruncate(pid_fd, 0); + + fprintf(stderr, + "rrdcached: removed stale PID file (no rrdcached on pid %d)\n" + "rrdcached: starting normally.\n", pid); + + return pid_fd; +} /* }}} static int check_pidfile */ + static int write_pidfile (int fd) /* {{{ */ { pid_t pid; @@ -381,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"); @@ -404,7 +462,7 @@ static int add_response_info(listen_socket_t *sock, char *fmt, ...) /* {{{ */ int len; if (sock == NULL) return 0; /* journal replay mode */ - if (sock->batch_mode) return 0; /* no extra info returned when in BATCH */ + if (sock->batch_start) return 0; /* no extra info returned when in BATCH */ va_start(argp, fmt); #ifdef HAVE_VSNPRINTF @@ -452,7 +510,7 @@ static int send_response (listen_socket_t *sock, response_code rc, if (sock == NULL) return rc; /* journal replay mode */ - if (sock->batch_mode) + if (sock->batch_start) { if (rc == RESP_OK) return rc; /* no response on success during BATCH */ @@ -477,7 +535,7 @@ static int send_response (listen_socket_t *sock, response_code rc, len += rclen; /* append the result to the wbuf, don't write to the user */ - if (sock->batch_mode) + if (sock->batch_start) return add_to_wbuf(sock, buffer, len); /* first write must be complete */ @@ -487,7 +545,7 @@ static int send_response (listen_socket_t *sock, response_code rc, return -1; } - if (sock->wbuf != NULL) + if (sock->wbuf != NULL && rc == RESP_OK) { wrote = 0; while (wrote < sock->wbuf_len) @@ -525,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 */ @@ -538,8 +597,37 @@ 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 + * must hold cache_lock when calling this function + */ +static void *free_cache_item(cache_item_t *ci) /* {{{ */ +{ + if (ci == NULL) return NULL; + + remove_from_queue(ci); + + for (int i=0; i < ci->values_num; i++) + free(ci->values[i]); + + free (ci->values); + free (ci->file); + + /* in case anyone is waiting */ + pthread_cond_broadcast(&ci->flushed); + + free (ci); + + return NULL; +} /* }}} static void *free_cache_item */ + /* * enqueue_cache_item: * `cache_lock' must be acquired before calling this function! @@ -558,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; @@ -592,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); @@ -602,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, /* {{{ */ @@ -614,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) { @@ -674,26 +761,10 @@ static int flush_old_values (int max_age) for (k = 0; k < cfd.keys_num; k++) { - cache_item_t *ci; - - /* This must not fail. */ - ci = (cache_item_t *) g_tree_lookup (cache_tree, cfd.keys[k]); - assert (ci != NULL); - - /* If we end up here with values available, something's seriously - * messed up. */ - assert (ci->values_num == 0); - - /* Remove the node from the tree */ - g_tree_remove (cache_tree, cfd.keys[k]); - cfd.keys[k] = NULL; - - /* Now free and clean up `ci'. */ - free (ci->file); - ci->file = NULL; - free (ci); - ci = NULL; - } /* for (k = 0; k < cfd.keys_num; k++) */ + /* should never fail, since we have held the cache_lock + * the entire time */ + assert( g_tree_remove(cache_tree, cfd.keys[k]) == TRUE ); + } if (cfd.keys != NULL) { @@ -704,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) @@ -735,8 +799,8 @@ static void *queue_thread_main (void *args __attribute__((unused))) /* {{{ */ flush_old_values (config_write_interval); /* Determine the time of the next cache flush. */ - while (next_flush.tv_sec <= now.tv_sec) - next_flush.tv_sec += config_flush_interval; + next_flush.tv_sec = + now.tv_sec + next_flush.tv_sec % config_flush_interval; /* unlock the cache while we rotate so we don't block incoming * updates if the fsync() blocks on disk I/O */ @@ -745,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) @@ -791,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 (); @@ -825,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 */ @@ -943,6 +1005,40 @@ err: return 0; } /* }}} static int check_file_access */ +/* when using a base dir, convert relative paths to absolute paths. + * if necessary, modifies the "filename" pointer to point + * to the new path created in "tmp". "tmp" is provided + * by the caller and sizeof(tmp) must be >= PATH_MAX. + * + * this allows us to optimize for the expected case (absolute path) + * with a no-op. + */ +static void get_abs_path(char **filename, char *tmp) +{ + assert(tmp != NULL); + assert(filename != NULL && *filename != NULL); + + if (config_base_dir == NULL || **filename == '/') + return; + + snprintf(tmp, PATH_MAX, "%s/%s", config_base_dir, *filename); + *filename = tmp; +} /* }}} static int get_abs_path */ + +/* returns 1 if we have the required privilege level, + * otherwise issue an error to the user on sock */ +static int has_privilege (listen_socket_t *sock, /* {{{ */ + socket_privilege priv) +{ + if (sock == NULL) /* journal replay */ + return 1; + + if (sock->privilege >= priv) + return 1; + + return send_response(sock, RESP_ERR, "%s\n", rrd_strerror(EACCES)); +} /* }}} static int has_privilege */ + static int flush_file (const char *filename) /* {{{ */ { cache_item_t *ci; @@ -963,6 +1059,9 @@ static int flush_file (const char *filename) /* {{{ */ pthread_cond_wait(&ci->flushed, &cache_lock); } + /* DO NOT DO ANYTHING WITH ci HERE!! The entry + * may have been purged during our cond_wait() */ + pthread_mutex_unlock(&cache_lock); return (0); @@ -979,12 +1078,16 @@ static int handle_request_help (listen_socket_t *sock, /* {{{ */ { "Command overview\n" , + "HELP []\n" "FLUSH \n" "FLUSHALL\n" - "HELP []\n" + "PENDING \n" + "FORGET \n" + "QUEUE\n" "UPDATE [ ...]\n" "BATCH\n" "STATS\n" + "QUIT\n" }; char *help_flush[2] = @@ -994,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] = @@ -1006,6 +1109,38 @@ static int handle_request_help (listen_socket_t *sock, /* {{{ */ "Triggers writing of all pending updates. Returns immediately.\n" }; + char *help_pending[2] = + { + "Help for PENDING\n" + , + "Usage: PENDING \n" + "\n" + "Shows any 'pending' updates for a file, in order.\n" + "The updates shown have not yet been written to the underlying RRD file.\n" + }; + + char *help_forget[2] = + { + "Help for FORGET\n" + , + "Usage: FORGET \n" + "\n" + "Removes the file completely from the cache.\n" + "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" @@ -1053,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; @@ -1064,10 +1206,18 @@ static int handle_request_help (listen_socket_t *sock, /* {{{ */ help_text = help_flush; else if (strcasecmp (command, "flushall") == 0) help_text = help_flushall; + else if (strcasecmp (command, "pending") == 0) + 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; } @@ -1127,7 +1277,7 @@ static int handle_request_stats (listen_socket_t *sock) /* {{{ */ static int handle_request_flush (listen_socket_t *sock, /* {{{ */ char *buffer, size_t buffer_size) { - char *file; + char *file, file_tmp[PATH_MAX]; int status; status = buffer_get_field (&buffer, &buffer_size, &file); @@ -1141,6 +1291,7 @@ static int handle_request_flush (listen_socket_t *sock, /* {{{ */ stats_flush_received++; pthread_mutex_unlock(&stats_lock); + get_abs_path(&file, file_tmp); if (!check_file_access(file, sock)) return 0; status = flush_file (file); @@ -1165,10 +1316,15 @@ static int handle_request_flush (listen_socket_t *sock, /* {{{ */ /* NOTREACHED */ assert(1==0); -} /* }}} int handle_request_slurp */ +} /* }}} int handle_request_flush */ static int handle_request_flushall(listen_socket_t *sock) /* {{{ */ { + int status; + + status = has_privilege(sock, PRIV_HIGH); + if (status <= 0) + return status; RRDD_LOG(LOG_DEBUG, "Received FLUSHALL"); @@ -1179,17 +1335,111 @@ static int handle_request_flushall(listen_socket_t *sock) /* {{{ */ return send_response(sock, RESP_OK, "Started flush.\n"); } /* }}} static int handle_request_flushall */ +static int handle_request_pending(listen_socket_t *sock, /* {{{ */ + char *buffer, size_t buffer_size) +{ + int status; + char *file, file_tmp[PATH_MAX]; + cache_item_t *ci; + + status = buffer_get_field(&buffer, &buffer_size, &file); + if (status != 0) + return send_response(sock, RESP_ERR, + "Usage: PENDING \n"); + + status = has_privilege(sock, PRIV_HIGH); + if (status <= 0) + return status; + + get_abs_path(&file, file_tmp); + + pthread_mutex_lock(&cache_lock); + ci = g_tree_lookup(cache_tree, file); + if (ci == NULL) + { + pthread_mutex_unlock(&cache_lock); + return send_response(sock, RESP_ERR, "%s\n", rrd_strerror(ENOENT)); + } + + for (int i=0; i < ci->values_num; i++) + add_response_info(sock, "%s\n", ci->values[i]); + + pthread_mutex_unlock(&cache_lock); + return send_response(sock, RESP_OK, "updates pending\n"); +} /* }}} static int handle_request_pending */ + +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); + if (status != 0) + return send_response(sock, RESP_ERR, + "Usage: FORGET \n"); + + status = has_privilege(sock, PRIV_HIGH); + if (status <= 0) + return status; + + get_abs_path(&file, file_tmp); + if (!check_file_access(file, sock)) return 0; + + pthread_mutex_lock(&cache_lock); + found = g_tree_remove(cache_tree, file); + pthread_mutex_unlock(&cache_lock); + + if (found == TRUE) + { + if (sock != NULL) + journal_write("forget", file); + + return send_response(sock, RESP_OK, "Gone!\n"); + } + else + 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, /* {{{ */ - char *buffer, size_t buffer_size) + time_t now, + char *buffer, size_t buffer_size) { - char *file; + char *file, file_tmp[PATH_MAX]; int values_num = 0; int status; + char orig_buf[CMD_MAX]; - time_t now; cache_item_t *ci; - now = time (NULL); + status = has_privilege(sock, PRIV_HIGH); + if (status <= 0) + return status; + + /* save it for the journal later */ + strncpy(orig_buf, buffer, sizeof(orig_buf)-1); status = buffer_get_field (&buffer, &buffer_size, &file); if (status != 0) @@ -1200,6 +1450,7 @@ static int handle_request_update (listen_socket_t *sock, /* {{{ */ stats_updates_received++; pthread_mutex_unlock(&stats_lock); + get_abs_path(&file, file_tmp); if (!check_file_access(file, sock)) return 0; pthread_mutex_lock (&cache_lock); @@ -1252,16 +1503,23 @@ 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); + /* don't re-write updates in replay mode */ + if (sock != NULL) + journal_write("update", orig_buf); + while (buffer_size > 0) { char **temp; char *value; + time_t stamp; + char *eostamp; status = buffer_get_field (&buffer, &buffer_size, &value); if (status != 0) @@ -1270,7 +1528,26 @@ static int handle_request_update (listen_socket_t *sock, /* {{{ */ break; } - temp = (char **) realloc (ci->values, + /* make sure update time is always moving forward */ + stamp = strtol(value, &eostamp, 10); + if (eostamp == value || eostamp == NULL || *eostamp != ':') + { + 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) + { + 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 **) rrd_realloc (ci->values, sizeof (char *) * (ci->values_num + 1)); if (temp == NULL) { @@ -1302,7 +1579,8 @@ static int handle_request_update (listen_socket_t *sock, /* {{{ */ if (values_num < 1) return send_response(sock, RESP_ERR, "No values updated.\n"); else - return send_response(sock, RESP_OK, "Enqueued %i value(s).\n", values_num); + return send_response(sock, RESP_OK, + "errors, enqueued %i value(s).\n", values_num); /* NOTREACHED */ assert(1==0); @@ -1312,7 +1590,7 @@ static int handle_request_update (listen_socket_t *sock, /* {{{ */ /* we came across a "WROTE" entry during journal replay. * throw away any values that we have accumulated for this file */ -static int handle_request_wrote (const char *buffer) /* {{{ */ +static int handle_request_wrote (const char *buffer, time_t now) /* {{{ */ { int i; cache_item_t *ci; @@ -1335,7 +1613,7 @@ static int handle_request_wrote (const char *buffer) /* {{{ */ free(ci->values); } - wipe_ci_values(ci, time(NULL)); + wipe_ci_values(ci, now); remove_from_queue(ci); pthread_mutex_unlock(&cache_lock); @@ -1346,12 +1624,12 @@ static int handle_request_wrote (const char *buffer) /* {{{ */ static int batch_start (listen_socket_t *sock) /* {{{ */ { int status; - if (sock->batch_mode) + if (sock->batch_start) return send_response(sock, RESP_ERR, "Already in BATCH\n"); status = send_response(sock, RESP_OK, "Go ahead. End with dot '.' on its own line.\n"); - sock->batch_mode = 1; + sock->batch_start = time(NULL); sock->batch_cmd = 0; return status; @@ -1360,27 +1638,15 @@ static int batch_start (listen_socket_t *sock) /* {{{ */ /* finish "BATCH" processing and return results to the client */ static int batch_done (listen_socket_t *sock) /* {{{ */ { - assert(sock->batch_mode); - sock->batch_mode = 0; + assert(sock->batch_start); + sock->batch_start = 0; sock->batch_cmd = 0; return send_response(sock, RESP_OK, "errors\n"); } /* }}} static int batch_done */ -/* returns 1 if we have the required privilege level */ -static int has_privilege (listen_socket_t *sock, /* {{{ */ - socket_privilege priv) -{ - if (sock == NULL) /* journal replay */ - return 1; - - if (sock->privilege >= priv) - return 1; - - return send_response(sock, RESP_ERR, "%s\n", rrd_strerror(EACCES)); -} /* }}} static int has_privilege */ - /* if sock==NULL, we are in journal replay mode */ static int handle_request (listen_socket_t *sock, /* {{{ */ + time_t now, char *buffer, size_t buffer_size) { char *buffer_ptr; @@ -1398,44 +1664,36 @@ static int handle_request (listen_socket_t *sock, /* {{{ */ return (-1); } - if (sock != NULL && sock->batch_mode) + if (sock != NULL && sock->batch_start) sock->batch_cmd++; if (strcasecmp (command, "update") == 0) - { - status = has_privilege(sock, PRIV_HIGH); - if (status <= 0) - return status; - - /* don't re-write updates in replay mode */ - if (sock != NULL) - journal_write(command, buffer_ptr); - - return (handle_request_update (sock, buffer_ptr, buffer_size)); - } + return (handle_request_update (sock, now, buffer_ptr, buffer_size)); else if (strcasecmp (command, "wrote") == 0 && sock == NULL) { /* this is only valid in replay mode */ - return (handle_request_wrote (buffer_ptr)); + return (handle_request_wrote (buffer_ptr, now)); } else if (strcasecmp (command, "flush") == 0) return (handle_request_flush (sock, buffer_ptr, buffer_size)); else if (strcasecmp (command, "flushall") == 0) - { - status = has_privilege(sock, PRIV_HIGH); - if (status <= 0) - return status; - return (handle_request_flushall(sock)); - } + else if (strcasecmp (command, "pending") == 0) + 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) return (handle_request_help (sock, buffer_ptr, buffer_size)); else if (strcasecmp (command, "batch") == 0 && sock != NULL) return batch_start(sock); - else if (strcasecmp (command, ".") == 0 && sock != NULL && sock->batch_mode) + 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); @@ -1550,11 +1808,12 @@ static int journal_replay (const char *file) /* {{{ */ int fail_cnt = 0; uint64_t line = 0; char entry[CMD_MAX]; + time_t now; if (file == NULL) return 0; { - char *reason; + char *reason = "unknown error"; int status = 0; struct stat statbuf; @@ -1602,6 +1861,8 @@ static int journal_replay (const char *file) /* {{{ */ else RRDD_LOG(LOG_NOTICE, "replaying from journal: %s", file); + now = time(NULL); + while(!feof(fh)) { size_t entry_len; @@ -1623,7 +1884,7 @@ static int journal_replay (const char *file) /* {{{ */ entry[entry_len - 1] = '\0'; - if (handle_request(NULL, entry, entry_len) == 0) + if (handle_request(NULL, now, entry, entry_len) == 0) ++entry_cnt; else ++fail_cnt; @@ -1661,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; @@ -1694,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 { @@ -1714,6 +1986,7 @@ static void *connection_thread_main (void *args) /* {{{ */ char *cmd; ssize_t cmd_len; ssize_t rbytes; + time_t now; struct pollfd pollfd; int status; @@ -1730,23 +2003,18 @@ static void *connection_thread_main (void *args) /* {{{ */ else if (status < 0) /* error */ { status = errno; - if (status == EINTR) - continue; - RRDD_LOG (LOG_ERR, "connection_thread_main: poll(2) failed."); + if (status != EINTR) + RRDD_LOG (LOG_ERR, "connection_thread_main: poll(2) failed."); continue; } if ((pollfd.revents & POLLHUP) != 0) /* normal shutdown */ - { - close_connection(sock); break; - } else if ((pollfd.revents & (POLLIN | POLLPRI)) == 0) { RRDD_LOG (LOG_WARNING, "connection_thread_main: " "poll(2) returned something unexpected: %#04hx", pollfd.revents); - close_connection(sock); break; } @@ -1762,9 +2030,14 @@ static void *connection_thread_main (void *args) /* {{{ */ sock->next_read += rbytes; + if (sock->batch_start) + now = sock->batch_start; + else + now = time(NULL); + while ((cmd = next_cmd(sock, &cmd_len)) != NULL) { - status = handle_request (sock, cmd, cmd_len+1); + status = handle_request (sock, now, cmd, cmd_len+1); if (status != 0) goto out_close; } @@ -1773,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); @@ -1808,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; @@ -1821,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); } @@ -1829,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); @@ -1887,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; @@ -1900,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 = ']') */ @@ -1920,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); } @@ -1931,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; @@ -1944,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; } @@ -1953,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; } @@ -1961,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); } @@ -1971,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 */ @@ -2012,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); } @@ -2043,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; @@ -2134,19 +2418,48 @@ 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 fd; + int pid_fd; char *base_dir; daemon_uid = geteuid(); - fd = open_pidfile(); - if (fd < 0) return fd; + pid_fd = open_pidfile("create", O_CREAT|O_EXCL|O_WRONLY); + if (pid_fd < 0) + pid_fd = check_pidfile(); + 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) { @@ -2156,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 (); @@ -2180,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(); @@ -2192,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 (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 (); @@ -2223,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) { @@ -2245,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) { @@ -2308,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; @@ -2315,6 +2664,7 @@ static int read_options (int argc, char **argv) /* {{{ */ case 'b': { size_t len; + char base_realpath[PATH_MAX]; if (config_base_dir != NULL) free (config_base_dir); @@ -2325,6 +2675,27 @@ static int read_options (int argc, char **argv) /* {{{ */ return (3); } + /* make sure that the base directory is not resolved via + * symbolic links. this makes some performance-enhancing + * assumptions possible (we don't have to resolve paths + * that start with a "/") + */ + if (realpath(config_base_dir, base_realpath) == NULL) + { + fprintf (stderr, "Invalid base directory '%s'.\n", config_base_dir); + return 5; + } + else if (strncmp(config_base_dir, + base_realpath, sizeof(base_realpath)) != 0) + { + fprintf(stderr, + "Base directory (-b) resolved via file system links!\n" + "Please consult rrdcached '-b' documentation!\n" + "Consider specifying the real directory (%s)\n", + base_realpath); + return 5; + } + len = strlen (config_base_dir); while ((len > 0) && (config_base_dir[len - 1] == '/')) { @@ -2405,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" @@ -2453,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); }