X-Git-Url: https://git.octo.it/?p=rrdtool.git;a=blobdiff_plain;f=src%2Frrd_daemon.c;h=2a35ec2abe59cfc2f87a316a0e097c051879a98d;hp=1e5317532dabf5a9784098c546fef3b587cdb07c;hb=ff4657f155a7acb054aaa702a7c2f610991bd0e9;hpb=2f2e580d4570417d95b8a3b4ef09a3c1e79808bb diff --git a/src/rrd_daemon.c b/src/rrd_daemon.c index 1e53175..2a35ec2 100644 --- a/src/rrd_daemon.c +++ b/src/rrd_daemon.c @@ -144,13 +144,38 @@ struct listen_socket_s }; typedef struct listen_socket_s listen_socket_t; +struct command; +/* note: guard against "unused" warnings in the handlers */ +#define DISPATCH_PROTO listen_socket_t *sock __attribute__((unused)),\ + time_t now __attribute__((unused)),\ + char *buffer __attribute__((unused)),\ + size_t buffer_size __attribute__((unused)) + +#define HANDLER_PROTO struct command *cmd __attribute__((unused)),\ + DISPATCH_PROTO + +struct command { + char *cmd; + int (*handler)(HANDLER_PROTO); + socket_privilege min_priv; + + char context; /* where we expect to see it */ +#define CMD_CONTEXT_CLIENT (1<<0) +#define CMD_CONTEXT_BATCH (1<<1) +#define CMD_CONTEXT_JOURNAL (1<<2) +#define CMD_CONTEXT_ANY (0x7f) + + char *syntax; + char *help; +}; + struct cache_item_s; typedef struct cache_item_s cache_item_t; struct cache_item_s { char *file; char **values; - int values_num; + size_t values_num; time_t last_flush_time; time_t last_update_stamp; #define CI_FLAGS_IN_TREE (1<<0) @@ -190,7 +215,11 @@ static uid_t daemon_uid; static listen_socket_t *listen_fds = NULL; static size_t listen_fds_num = 0; -static int do_shutdown = 0; +enum { + RUNNING, /* normal operation */ + FLUSHING, /* flushing remaining values */ + SHUTDOWN /* shutting down */ +} state = RUNNING; static pthread_t *queue_threads; static pthread_cond_t queue_cond = PTHREAD_COND_INITIALIZER; @@ -219,7 +248,7 @@ static size_t _config_base_dir_len = 0; static int config_write_base_only = 0; static listen_socket_t **config_listen_address_list = NULL; -static int config_listen_address_list_len = 0; +static size_t config_listen_address_list_len = 0; static uint64_t stats_queue_length = 0; static uint64_t stats_updates_received = 0; @@ -239,13 +268,16 @@ static int journal_write(char *cmd, char *args); static void journal_done(void); static void journal_rotate(void); +/* prototypes for forward refernces */ +static int handle_request_help (HANDLER_PROTO); + /* * Functions */ static void sig_common (const char *sig) /* {{{ */ { RRDD_LOG(LOG_NOTICE, "caught SIG%s", sig); - do_shutdown++; + state = FLUSHING; pthread_cond_broadcast(&flush_cond); pthread_cond_broadcast(&queue_cond); } /* }}} void sig_common */ @@ -351,7 +383,13 @@ static int check_pidfile(void) } lseek(pid_fd, 0, SEEK_SET); - ftruncate(pid_fd, 0); + if (ftruncate(pid_fd, 0) == -1) + { + fprintf(stderr, + "FATAL: Faild to truncate stale PID file. (pid %d)\n", pid); + close(pid_fd); + return -1; + } fprintf(stderr, "rrdcached: removed stale PID file (no rrdcached on pid %d)\n" @@ -573,7 +611,7 @@ static void wipe_ci_values(cache_item_t *ci, time_t when) ci->last_flush_time = when; if (config_write_jitter > 0) - ci->last_flush_time += (random() % config_write_jitter); + ci->last_flush_time += (rrd_random() % config_write_jitter); } /* remove_from_queue @@ -614,7 +652,7 @@ static void *free_cache_item(cache_item_t *ci) /* {{{ */ remove_from_queue(ci); - for (int i=0; i < ci->values_num; i++) + for (size_t i=0; i < ci->values_num; i++) free(ci->values[i]); free (ci->values); @@ -622,6 +660,7 @@ static void *free_cache_item(cache_item_t *ci) /* {{{ */ /* in case anyone is waiting */ pthread_cond_broadcast(&ci->flushed); + pthread_cond_destroy(&ci->flushed); free (ci); @@ -704,33 +743,20 @@ static gboolean tree_callback_flush (gpointer key, gpointer value, /* {{{ */ if (ci->flags & CI_FLAGS_IN_QUEUE) return FALSE; - if ((ci->last_flush_time <= cfd->abs_timeout) - && (ci->values_num > 0)) - { - enqueue_cache_item (ci, TAIL); - } - else if ((do_shutdown != 0) - && (ci->values_num > 0)) + if (ci->values_num > 0 + && (ci->last_flush_time <= cfd->abs_timeout || state != RUNNING)) { enqueue_cache_item (ci, TAIL); } else if (((cfd->now - ci->last_flush_time) >= config_flush_interval) && (ci->values_num <= 0)) { - char **temp; - - temp = (char **) rrd_realloc (cfd->keys, - sizeof (char *) * (cfd->keys_num + 1)); - if (temp == NULL) + assert ((char *) key == ci->file); + if (!rrd_add_ptr((void ***)&cfd->keys, &cfd->keys_num, (void *)key)) { - RRDD_LOG (LOG_ERR, "tree_callback_flush: realloc failed."); + RRDD_LOG (LOG_ERR, "tree_callback_flush: rrd_add_ptrs failed."); return (FALSE); } - cfd->keys = temp; - /* Make really sure this points to the _same_ place */ - assert ((char *) key == ci->file); - cfd->keys[cfd->keys_num] = (char *) key; - cfd->keys_num++; } return (FALSE); @@ -787,21 +813,22 @@ static void *flush_thread_main (void *args __attribute__((unused))) /* {{{ */ pthread_mutex_lock(&cache_lock); - while (!do_shutdown) + while (state == RUNNING) { gettimeofday (&now, NULL); if ((now.tv_sec > next_flush.tv_sec) || ((now.tv_sec == next_flush.tv_sec) && ((1000 * now.tv_usec) > next_flush.tv_nsec))) { + RRDD_LOG(LOG_DEBUG, "flushing old values"); + + /* Determine the time of the next cache flush. */ + next_flush.tv_sec = now.tv_sec + config_flush_interval; + /* Flush all values that haven't been written in the last * `config_write_interval' seconds. */ flush_old_values (config_write_interval); - /* Determine the time of the next cache flush. */ - 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 */ pthread_mutex_unlock(&cache_lock); @@ -820,6 +847,8 @@ static void *flush_thread_main (void *args __attribute__((unused))) /* {{{ */ if (config_flush_at_shutdown) flush_old_values (-1); /* flush everything */ + state = SHUTDOWN; + pthread_mutex_unlock(&cache_lock); return NULL; @@ -829,19 +858,18 @@ static void *queue_thread_main (void *args __attribute__((unused))) /* {{{ */ { pthread_mutex_lock (&cache_lock); - while (!do_shutdown + while (state != SHUTDOWN || (cache_queue_head != NULL && config_flush_at_shutdown)) { cache_item_t *ci; char *file; char **values; - int values_num; + size_t values_num; int status; - int i; /* Now, check if there's something to store away. If not, wait until - * something comes in. if we are shutting down, do not wait around. */ - if (cache_queue_head == NULL && !do_shutdown) + * something comes in. */ + if (cache_queue_head == NULL) { status = pthread_cond_wait (&queue_cond, &cache_lock); if ((status != 0) && (status != ETIMEDOUT)) @@ -878,7 +906,7 @@ static void *queue_thread_main (void *args __attribute__((unused))) /* {{{ */ pthread_mutex_unlock (&cache_lock); rrd_clear_error (); - status = rrd_update_r (file, NULL, values_num, (void *) values); + status = rrd_update_r (file, NULL, (int) values_num, (void *) values); if (status != 0) { RRDD_LOG (LOG_NOTICE, "queue_thread_main: " @@ -887,12 +915,16 @@ static void *queue_thread_main (void *args __attribute__((unused))) /* {{{ */ } journal_write("wrote", file); - pthread_cond_broadcast(&ci->flushed); - for (i = 0; i < values_num; i++) - free (values[i]); + /* Search again in the tree. It's possible someone issued a "FORGET" + * while we were writing the update values. */ + pthread_mutex_lock(&cache_lock); + ci = (cache_item_t *) g_tree_lookup(cache_tree, file); + if (ci) + pthread_cond_broadcast(&ci->flushed); + pthread_mutex_unlock(&cache_lock); - free(values); + rrd_free_ptrs((void ***) &values, &values_num); free(file); if (status == 0) @@ -1067,166 +1099,17 @@ static int flush_file (const char *filename) /* {{{ */ return (0); } /* }}} int flush_file */ -static int handle_request_help (listen_socket_t *sock, /* {{{ */ - char *buffer, size_t buffer_size) +static int syntax_error(listen_socket_t *sock, struct command *cmd) /* {{{ */ { - int status; - char **help_text; - char *command; + char *err = "Syntax error.\n"; - char *help_help[2] = - { - "Command overview\n" - , - "HELP []\n" - "FLUSH \n" - "FLUSHALL\n" - "PENDING \n" - "FORGET \n" - "QUEUE\n" - "UPDATE [ ...]\n" - "BATCH\n" - "STATS\n" - "QUIT\n" - }; + if (cmd && cmd->syntax) + err = cmd->syntax; - char *help_flush[2] = - { - "Help for FLUSH\n" - , - "Usage: FLUSH \n" - "\n" - "Adds the given filename to the head of the update queue and returns\n" - "after it has been dequeued.\n" - }; + return send_response(sock, RESP_ERR, "Usage: %s", err); +} /* }}} static int syntax_error() */ - char *help_flushall[2] = - { - "Help for FLUSHALL\n" - , - "Usage: FLUSHALL\n" - "\n" - "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" - , - "Usage: UPDATE [ ...]\n" - "\n" - "Adds the given file to the internal cache if it is not yet known and\n" - "appends the given value(s) to the entry. See the rrdcached(1) manpage\n" - "for details.\n" - "\n" - "Each has the following form:\n" - " =