src/rrd_daemon.c

   1 /**
   2  * RRDTool - src/rrd_daemon.c
   3  * Copyright (C) 2008-2010 Florian octo Forster
   4  * Copyright (C) 2008,2009 Kevin Brintnall
   5  *
   6  * This program is free software; you can redistribute it and/or modify it
   7  * under the terms of the GNU General Public License as published by the
   8  * Free Software Foundation; only version 2 of the License is applicable.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU General Public License along
  16  * with this program; if not, write to the Free Software Foundation, Inc.,
  17  * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
  18  *
  19  * Authors:
  20  *   Florian octo Forster <octo at verplant.org>
  21  *   kevin brintnall <kbrint@rufus.net>
  22  **/
  23
  24 #if 0
  25 /*
  26  * First tell the compiler to stick to the C99 and POSIX standards as close as
  27  * possible.
  28  */
  29 #ifndef __STRICT_ANSI__ /* {{{ */
  30 # define __STRICT_ANSI__
  31 #endif
  32
  33 #ifndef _ISOC99_SOURCE
  34 # define _ISOC99_SOURCE
  35 #endif
  36
  37 #ifdef _POSIX_C_SOURCE
  38 # undef _POSIX_C_SOURCE
  39 #endif
  40 #define _POSIX_C_SOURCE 200112L
  41
  42 /* Single UNIX needed for strdup. */
  43 #ifdef _XOPEN_SOURCE
  44 # undef _XOPEN_SOURCE
  45 #endif
  46 #define _XOPEN_SOURCE 500
  47
  48 #ifndef _REENTRANT
  49 # define _REENTRANT
  50 #endif
  51
  52 #ifndef _THREAD_SAFE
  53 # define _THREAD_SAFE
  54 #endif
  55
  56 #ifdef _GNU_SOURCE
  57 # undef _GNU_SOURCE
  58 #endif
  59 /* }}} */
  60 #endif /* 0 */
  61
  62 /*
  63  * Now for some includes..
  64  */
  65 /* {{{ */
  66 #if defined(_WIN32) && !defined(__CYGWIN__) && !defined(__CYGWIN32__) && !defined(HAVE_CONFIG_H)
  67 #include "../win32/config.h"
  68 #else
  69 #ifdef HAVE_CONFIG_H
  70 #include "../rrd_config.h"
  71 #endif
  72 #endif
  73
  74 #include "rrd.h"
  75 #include "rrd_client.h"
  76 #include "unused.h"
  77
  78 #include <stdlib.h>
  79
  80 #ifndef WIN32
  81 #ifdef HAVE_STDINT_H
  82 #  include <stdint.h>
  83 #endif
  84 #include <unistd.h>
  85 #include <strings.h>
  86 #include <inttypes.h>
  87 #include <sys/socket.h>
  88
  89 #else
  90
  91 #endif
  92 #include <stdio.h>
  93 #include <string.h>
  94
  95 #include <sys/types.h>
  96 #include <sys/stat.h>
  97 #include <dirent.h>
  98 #include <fcntl.h>
  99 #include <signal.h>
 100 #include <sys/un.h>
 101 #include <netdb.h>
 102 #include <poll.h>
 103 #include <syslog.h>
 104 #include <pthread.h>
 105 #include <errno.h>
 106 #include <assert.h>
 107 #include <sys/time.h>
 108 #include <time.h>
 109 #include <libgen.h>
 110 #include <grp.h>
 111
 112 #include <glib-2.0/glib.h>
 113 /* }}} */
 114
 115 #define RRDD_LOG(severity, ...) \
 116   do { \
 117     if (stay_foreground) \
 118       fprintf(stderr, __VA_ARGS__); \
 119     syslog ((severity), __VA_ARGS__); \
 120   } while (0)
 121
 122 /*
 123  * Types
 124  */
 125 typedef enum { RESP_ERR = -1, RESP_OK = 0 } response_code;
 126
 127 struct listen_socket_s
 128 {
 129   int fd;
 130   char addr[PATH_MAX + 1];
 131   int family;
 132
 133   /* state for BATCH processing */
 134   time_t batch_start;
 135   int batch_cmd;
 136
 137   /* buffered IO */
 138   char *rbuf;
 139   off_t next_cmd;
 140   off_t next_read;
 141
 142   char *wbuf;
 143   ssize_t wbuf_len;
 144
 145   uint32_t permissions;
 146
 147   gid_t  socket_group;
 148   mode_t socket_permissions;
 149 };
 150 typedef struct listen_socket_s listen_socket_t;
 151
 152 struct command_s;
 153 typedef struct command_s command_t;
 154 /* note: guard against "unused" warnings in the handlers */
 155 #define DISPATCH_PROTO  listen_socket_t UNUSED(*sock),\
 156                         time_t UNUSED(now),\
 157                         char  UNUSED(*buffer),\
 158                         size_t UNUSED(buffer_size)
 159
 160 #define HANDLER_PROTO   command_t UNUSED(*cmd),\
 161                         DISPATCH_PROTO
 162
 163 struct command_s {
 164   char   *cmd;
 165   int (*handler)(HANDLER_PROTO);
 166
 167   char  context;                /* where we expect to see it */
 168 #define CMD_CONTEXT_CLIENT      (1<<0)
 169 #define CMD_CONTEXT_BATCH       (1<<1)
 170 #define CMD_CONTEXT_JOURNAL     (1<<2)
 171 #define CMD_CONTEXT_ANY         (0x7f)
 172
 173   char *syntax;
 174   char *help;
 175 };
 176
 177 struct cache_item_s;
 178 typedef struct cache_item_s cache_item_t;
 179 struct cache_item_s
 180 {
 181   char *file;
 182   char **values;
 183   size_t values_num;            /* number of valid pointers */
 184   size_t values_alloc;          /* number of allocated pointers */
 185   time_t last_flush_time;
 186   time_t last_update_stamp;
 187 #define CI_FLAGS_IN_TREE  (1<<0)
 188 #define CI_FLAGS_IN_QUEUE (1<<1)
 189   int flags;
 190   pthread_cond_t  flushed;
 191   cache_item_t *prev;
 192   cache_item_t *next;
 193 };
 194
 195 struct callback_flush_data_s
 196 {
 197   time_t now;
 198   time_t abs_timeout;
 199   char **keys;
 200   size_t keys_num;
 201 };
 202 typedef struct callback_flush_data_s callback_flush_data_t;
 203
 204 enum queue_side_e
 205 {
 206   HEAD,
 207   TAIL
 208 };
 209 typedef enum queue_side_e queue_side_t;
 210
 211 /* describe a set of journal files */
 212 typedef struct {
 213   char **files;
 214   size_t files_num;
 215 } journal_set;
 216
 217 /* max length of socket command or response */
 218 #define CMD_MAX 4096
 219 #define RBUF_SIZE (CMD_MAX*2)
 220
 221 /*
 222  * Variables
 223  */
 224 static int stay_foreground = 0;
 225 static uid_t daemon_uid;
 226
 227 static listen_socket_t *listen_fds = NULL;
 228 static size_t listen_fds_num = 0;
 229
 230 enum {
 231   RUNNING,              /* normal operation */
 232   FLUSHING,             /* flushing remaining values */
 233   SHUTDOWN              /* shutting down */
 234 } state = RUNNING;
 235
 236 static pthread_t *queue_threads;
 237 static pthread_cond_t queue_cond = PTHREAD_COND_INITIALIZER;
 238 static int config_queue_threads = 4;
 239
 240 static pthread_t flush_thread;
 241 static pthread_cond_t flush_cond = PTHREAD_COND_INITIALIZER;
 242
 243 static pthread_mutex_t connection_threads_lock = PTHREAD_MUTEX_INITIALIZER;
 244 static pthread_cond_t  connection_threads_done = PTHREAD_COND_INITIALIZER;
 245 static int connection_threads_num = 0;
 246
 247 /* Cache stuff */
 248 static GTree          *cache_tree = NULL;
 249 static cache_item_t   *cache_queue_head = NULL;
 250 static cache_item_t   *cache_queue_tail = NULL;
 251 static pthread_mutex_t cache_lock = PTHREAD_MUTEX_INITIALIZER;
 252
 253 static int config_write_interval = 300;
 254 static int config_write_jitter   = 0;
 255 static int config_flush_interval = 3600;
 256 static int config_flush_at_shutdown = 0;
 257 static char *config_pid_file = NULL;
 258 static char *config_base_dir = NULL;
 259 static size_t _config_base_dir_len = 0;
 260 static int config_write_base_only = 0;
 261 static size_t config_alloc_chunk = 1;
 262
 263 static listen_socket_t **config_listen_address_list = NULL;
 264 static size_t config_listen_address_list_len = 0;
 265
 266 static uint64_t stats_queue_length = 0;
 267 static uint64_t stats_updates_received = 0;
 268 static uint64_t stats_flush_received = 0;
 269 static uint64_t stats_updates_written = 0;
 270 static uint64_t stats_data_sets_written = 0;
 271 static uint64_t stats_journal_bytes = 0;
 272 static uint64_t stats_journal_rotate = 0;
 273 static pthread_mutex_t stats_lock = PTHREAD_MUTEX_INITIALIZER;
 274
 275 /* Journaled updates */
 276 #define JOURNAL_REPLAY(s) ((s) == NULL)
 277 #define JOURNAL_BASE "rrd.journal"
 278 static journal_set *journal_cur = NULL;
 279 static journal_set *journal_old = NULL;
 280 static char *journal_dir = NULL;
 281 static FILE *journal_fh = NULL;         /* current journal file handle */
 282 static long  journal_size = 0;          /* current journal size */
 283 #define JOURNAL_MAX (1 * 1024 * 1024 * 1024)
 284 static pthread_mutex_t journal_lock = PTHREAD_MUTEX_INITIALIZER;
 285 static int journal_write(char *cmd, char *args);
 286 static void journal_done(void);
 287 static void journal_rotate(void);
 288
 289 /* prototypes for forward refernces */
 290 static int handle_request_help (HANDLER_PROTO);
 291
 292 /*
 293  * Functions
 294  */
 295 static void sig_common (const char *sig) /* {{{ */
 296 {
 297   RRDD_LOG(LOG_NOTICE, "caught SIG%s", sig);
 298   state = FLUSHING;
 299   pthread_cond_broadcast(&flush_cond);
 300   pthread_cond_broadcast(&queue_cond);
 301 } /* }}} void sig_common */
 302
 303 static void sig_int_handler (int UNUSED(s)) /* {{{ */
 304 {
 305   sig_common("INT");
 306 } /* }}} void sig_int_handler */
 307
 308 static void sig_term_handler (int UNUSED(s)) /* {{{ */
 309 {
 310   sig_common("TERM");
 311 } /* }}} void sig_term_handler */
 312
 313 static void sig_usr1_handler (int UNUSED(s)) /* {{{ */
 314 {
 315   config_flush_at_shutdown = 1;
 316   sig_common("USR1");
 317 } /* }}} void sig_usr1_handler */
 318
 319 static void sig_usr2_handler (int UNUSED(s)) /* {{{ */
 320 {
 321   config_flush_at_shutdown = 0;
 322   sig_common("USR2");
 323 } /* }}} void sig_usr2_handler */
 324
 325 static void install_signal_handlers(void) /* {{{ */
 326 {
 327   /* These structures are static, because `sigaction' behaves weird if the are
 328    * overwritten.. */
 329   static struct sigaction sa_int;
 330   static struct sigaction sa_term;
 331   static struct sigaction sa_pipe;
 332   static struct sigaction sa_usr1;
 333   static struct sigaction sa_usr2;
 334
 335   /* Install signal handlers */
 336   memset (&sa_int, 0, sizeof (sa_int));
 337   sa_int.sa_handler = sig_int_handler;
 338   sigaction (SIGINT, &sa_int, NULL);
 339
 340   memset (&sa_term, 0, sizeof (sa_term));
 341   sa_term.sa_handler = sig_term_handler;
 342   sigaction (SIGTERM, &sa_term, NULL);
 343
 344   memset (&sa_pipe, 0, sizeof (sa_pipe));
 345   sa_pipe.sa_handler = SIG_IGN;
 346   sigaction (SIGPIPE, &sa_pipe, NULL);
 347
 348   memset (&sa_pipe, 0, sizeof (sa_usr1));
 349   sa_usr1.sa_handler = sig_usr1_handler;
 350   sigaction (SIGUSR1, &sa_usr1, NULL);
 351
 352   memset (&sa_usr2, 0, sizeof (sa_usr2));
 353   sa_usr2.sa_handler = sig_usr2_handler;
 354   sigaction (SIGUSR2, &sa_usr2, NULL);
 355
 356 } /* }}} void install_signal_handlers */
 357
 358 static int open_pidfile(char *action, int oflag) /* {{{ */
 359 {
 360   int fd;
 361   const char *file;
 362   char *file_copy, *dir;
 363
 364   file = (config_pid_file != NULL)
 365     ? config_pid_file
 366     : LOCALSTATEDIR "/run/rrdcached.pid";
 367
 368   /* dirname may modify its argument */
 369   file_copy = strdup(file);
 370   if (file_copy == NULL)
 371   {
 372     fprintf(stderr, "rrdcached: strdup(): %s\n",
 373         rrd_strerror(errno));
 374     return -1;
 375   }
 376
 377   dir = dirname(file_copy);
 378   if (rrd_mkdir_p(dir, 0777) != 0)
 379   {
 380     fprintf(stderr, "Failed to create pidfile directory '%s': %s\n",
 381         dir, rrd_strerror(errno));
 382     return -1;
 383   }
 384
 385   free(file_copy);
 386
 387   fd = open(file, oflag, S_IWUSR|S_IRUSR|S_IRGRP|S_IROTH);
 388   if (fd < 0)
 389     fprintf(stderr, "rrdcached: can't %s pid file '%s' (%s)\n",
 390             action, file, rrd_strerror(errno));
 391
 392   return(fd);
 393 } /* }}} static int open_pidfile */
 394
 395 /* check existing pid file to see whether a daemon is running */
 396 static int check_pidfile(void)
 397 {
 398   int pid_fd;
 399   pid_t pid;
 400   char pid_str[16];
 401
 402   pid_fd = open_pidfile("open", O_RDWR);
 403   if (pid_fd < 0)
 404     return pid_fd;
 405
 406   if (read(pid_fd, pid_str, sizeof(pid_str)) <= 0)
 407     return -1;
 408
 409   pid = atoi(pid_str);
 410   if (pid <= 0)
 411     return -1;
 412
 413   /* another running process that we can signal COULD be
 414    * a competing rrdcached */
 415   if (pid != getpid() && kill(pid, 0) == 0)
 416   {
 417     fprintf(stderr,
 418             "FATAL: Another rrdcached daemon is running?? (pid %d)\n", pid);
 419     close(pid_fd);
 420     return -1;
 421   }
 422
 423   lseek(pid_fd, 0, SEEK_SET);
 424   if (ftruncate(pid_fd, 0) == -1)
 425   {
 426     fprintf(stderr,
 427             "FATAL: Faild to truncate stale PID file. (pid %d)\n", pid);
 428     close(pid_fd);
 429     return -1;
 430   }
 431
 432   fprintf(stderr,
 433           "rrdcached: removed stale PID file (no rrdcached on pid %d)\n"
 434           "rrdcached: starting normally.\n", pid);
 435
 436   return pid_fd;
 437 } /* }}} static int check_pidfile */
 438
 439 static int write_pidfile (int fd) /* {{{ */
 440 {
 441   pid_t pid;
 442   FILE *fh;
 443
 444   pid = getpid ();
 445
 446   fh = fdopen (fd, "w");
 447   if (fh == NULL)
 448   {
 449     RRDD_LOG (LOG_ERR, "write_pidfile: fdopen() failed.");
 450     close(fd);
 451     return (-1);
 452   }
 453
 454   fprintf (fh, "%i\n", (int) pid);
 455   fclose (fh);
 456
 457   return (0);
 458 } /* }}} int write_pidfile */
 459
 460 static int remove_pidfile (void) /* {{{ */
 461 {
 462   char *file;
 463   int status;
 464
 465   file = (config_pid_file != NULL)
 466     ? config_pid_file
 467     : LOCALSTATEDIR "/run/rrdcached.pid";
 468
 469   status = unlink (file);
 470   if (status == 0)
 471     return (0);
 472   return (errno);
 473 } /* }}} int remove_pidfile */
 474
 475 static char *next_cmd (listen_socket_t *sock, ssize_t *len) /* {{{ */
 476 {
 477   char *eol;
 478
 479   eol = memchr(sock->rbuf + sock->next_cmd, '\n',
 480                sock->next_read - sock->next_cmd);
 481
 482   if (eol == NULL)
 483   {
 484     /* no commands left, move remainder back to front of rbuf */
 485     memmove(sock->rbuf, sock->rbuf + sock->next_cmd,
 486             sock->next_read - sock->next_cmd);
 487     sock->next_read -= sock->next_cmd;
 488     sock->next_cmd = 0;
 489     *len = 0;
 490     return NULL;
 491   }
 492   else
 493   {
 494     char *cmd = sock->rbuf + sock->next_cmd;
 495     *eol = '\0';
 496
 497     sock->next_cmd = eol - sock->rbuf + 1;
 498
 499     if (eol > sock->rbuf && *(eol-1) == '\r')
 500       *(--eol) = '\0'; /* handle "\r\n" EOL */
 501
 502     *len = eol - cmd;
 503
 504     return cmd;
 505   }
 506
 507   /* NOTREACHED */
 508   assert(1==0);
 509 } /* }}} char *next_cmd */
 510
 511 /* add the characters directly to the write buffer */
 512 static int add_to_wbuf(listen_socket_t *sock, char *str, size_t len) /* {{{ */
 513 {
 514   char *new_buf;
 515
 516   assert(sock != NULL);
 517
 518   new_buf = rrd_realloc(sock->wbuf, sock->wbuf_len + len + 1);
 519   if (new_buf == NULL)
 520   {
 521     RRDD_LOG(LOG_ERR, "add_to_wbuf: realloc failed");
 522     return -1;
 523   }
 524
 525   strncpy(new_buf + sock->wbuf_len, str, len + 1);
 526
 527   sock->wbuf = new_buf;
 528   sock->wbuf_len += len;
 529
 530   return 0;
 531 } /* }}} static int add_to_wbuf */
 532
 533 /* add the text to the "extra" info that's sent after the status line */
 534 static int add_response_info(listen_socket_t *sock, char *fmt, ...) /* {{{ */
 535 {
 536   va_list argp;
 537   char buffer[CMD_MAX];
 538   int len;
 539
 540   if (JOURNAL_REPLAY(sock)) return 0;
 541   if (sock->batch_start) return 0; /* no extra info returned when in BATCH */
 542
 543   va_start(argp, fmt);
 544 #ifdef HAVE_VSNPRINTF
 545   len = vsnprintf(buffer, sizeof(buffer), fmt, argp);
 546 #else
 547   len = vsprintf(buffer, fmt, argp);
 548 #endif
 549   va_end(argp);
 550   if (len < 0)
 551   {
 552     RRDD_LOG(LOG_ERR, "add_response_info: vnsprintf failed");
 553     return -1;
 554   }
 555
 556   return add_to_wbuf(sock, buffer, len);
 557 } /* }}} static int add_response_info */
 558
 559 static int count_lines(char *str) /* {{{ */
 560 {
 561   int lines = 0;
 562
 563   if (str != NULL)
 564   {
 565     while ((str = strchr(str, '\n')) != NULL)
 566     {
 567       ++lines;
 568       ++str;
 569     }
 570   }
 571
 572   return lines;
 573 } /* }}} static int count_lines */
 574
 575 /* send the response back to the user.
 576  * returns 0 on success, -1 on error
 577  * write buffer is always zeroed after this call */
 578 static int send_response (listen_socket_t *sock, response_code rc,
 579                           char *fmt, ...) /* {{{ */
 580 {
 581   va_list argp;
 582   char buffer[CMD_MAX];
 583   int lines;
 584   ssize_t wrote;
 585   int rclen, len;
 586
 587   if (JOURNAL_REPLAY(sock)) return rc;
 588
 589   if (sock->batch_start)
 590   {
 591     if (rc == RESP_OK)
 592       return rc; /* no response on success during BATCH */
 593     lines = sock->batch_cmd;
 594   }
 595   else if (rc == RESP_OK)
 596     lines = count_lines(sock->wbuf);
 597   else
 598     lines = -1;
 599
 600   rclen = sprintf(buffer, "%d ", lines);
 601   va_start(argp, fmt);
 602 #ifdef HAVE_VSNPRINTF
 603   len = vsnprintf(buffer+rclen, sizeof(buffer)-rclen, fmt, argp);
 604 #else
 605   len = vsprintf(buffer+rclen, fmt, argp);
 606 #endif
 607   va_end(argp);
 608   if (len < 0)
 609     return -1;
 610
 611   len += rclen;
 612
 613   /* append the result to the wbuf, don't write to the user */
 614   if (sock->batch_start)
 615     return add_to_wbuf(sock, buffer, len);
 616
 617   /* first write must be complete */
 618   if (len != write(sock->fd, buffer, len))
 619   {
 620     RRDD_LOG(LOG_INFO, "send_response: could not write status message");
 621     return -1;
 622   }
 623
 624   if (sock->wbuf != NULL && rc == RESP_OK)
 625   {
 626     wrote = 0;
 627     while (wrote < sock->wbuf_len)
 628     {
 629       ssize_t wb = write(sock->fd, sock->wbuf + wrote, sock->wbuf_len - wrote);
 630       if (wb <= 0)
 631       {
 632         RRDD_LOG(LOG_INFO, "send_response: could not write results");
 633         return -1;
 634       }
 635       wrote += wb;
 636     }
 637   }
 638
 639   free(sock->wbuf); sock->wbuf = NULL;
 640   sock->wbuf_len = 0;
 641
 642   return 0;
 643 } /* }}} */
 644
 645 static void wipe_ci_values(cache_item_t *ci, time_t when)
 646 {
 647   ci->values = NULL;
 648   ci->values_num = 0;
 649   ci->values_alloc = 0;
 650
 651   ci->last_flush_time = when;
 652   if (config_write_jitter > 0)
 653     ci->last_flush_time += (rrd_random() % config_write_jitter);
 654 }
 655
 656 /* remove_from_queue
 657  * remove a "cache_item_t" item from the queue.
 658  * must hold 'cache_lock' when calling this
 659  */
 660 static void remove_from_queue(cache_item_t *ci) /* {{{ */
 661 {
 662   if (ci == NULL) return;
 663   if ((ci->flags & CI_FLAGS_IN_QUEUE) == 0) return; /* not queued */
 664
 665   if (ci->prev == NULL)
 666     cache_queue_head = ci->next; /* reset head */
 667   else
 668     ci->prev->next = ci->next;
 669
 670   if (ci->next == NULL)
 671     cache_queue_tail = ci->prev; /* reset the tail */
 672   else
 673     ci->next->prev = ci->prev;
 674
 675   ci->next = ci->prev = NULL;
 676   ci->flags &= ~CI_FLAGS_IN_QUEUE;
 677
 678   pthread_mutex_lock (&stats_lock);
 679   assert (stats_queue_length > 0);
 680   stats_queue_length--;
 681   pthread_mutex_unlock (&stats_lock);
 682
 683 } /* }}} static void remove_from_queue */
 684
 685 /* free the resources associated with the cache_item_t
 686  * must hold cache_lock when calling this function
 687  */
 688 static void *free_cache_item(cache_item_t *ci) /* {{{ */
 689 {
 690   if (ci == NULL) return NULL;
 691
 692   remove_from_queue(ci);
 693
 694   for (size_t i=0; i < ci->values_num; i++)
 695     free(ci->values[i]);
 696
 697   free (ci->values);
 698   free (ci->file);
 699
 700   /* in case anyone is waiting */
 701   pthread_cond_broadcast(&ci->flushed);
 702   pthread_cond_destroy(&ci->flushed);
 703
 704   free (ci);
 705
 706   return NULL;
 707 } /* }}} static void *free_cache_item */
 708
 709 /*
 710  * enqueue_cache_item:
 711  * `cache_lock' must be acquired before calling this function!
 712  */
 713 static int enqueue_cache_item (cache_item_t *ci, /* {{{ */
 714     queue_side_t side)
 715 {
 716   if (ci == NULL)
 717     return (-1);
 718
 719   if (ci->values_num == 0)
 720     return (0);
 721
 722   if (side == HEAD)
 723   {
 724     if (cache_queue_head == ci)
 725       return 0;
 726
 727     /* remove if further down in queue */
 728     remove_from_queue(ci);
 729
 730     ci->prev = NULL;
 731     ci->next = cache_queue_head;
 732     if (ci->next != NULL)
 733       ci->next->prev = ci;
 734     cache_queue_head = ci;
 735
 736     if (cache_queue_tail == NULL)
 737       cache_queue_tail = cache_queue_head;
 738   }
 739   else /* (side == TAIL) */
 740   {
 741     /* We don't move values back in the list.. */
 742     if (ci->flags & CI_FLAGS_IN_QUEUE)
 743       return (0);
 744
 745     assert (ci->next == NULL);
 746     assert (ci->prev == NULL);
 747
 748     ci->prev = cache_queue_tail;
 749
 750     if (cache_queue_tail == NULL)
 751       cache_queue_head = ci;
 752     else
 753       cache_queue_tail->next = ci;
 754
 755     cache_queue_tail = ci;
 756   }
 757
 758   ci->flags |= CI_FLAGS_IN_QUEUE;
 759
 760   pthread_cond_signal(&queue_cond);
 761   pthread_mutex_lock (&stats_lock);
 762   stats_queue_length++;
 763   pthread_mutex_unlock (&stats_lock);
 764
 765   return (0);
 766 } /* }}} int enqueue_cache_item */
 767
 768 /*
 769  * tree_callback_flush:
 770  * Called via `g_tree_foreach' in `flush_thread_main'. `cache_lock' is held
 771  * while this is in progress.
 772  */
 773 static gboolean tree_callback_flush (gpointer key, gpointer value, /* {{{ */
 774     gpointer data)
 775 {
 776   cache_item_t *ci;
 777   callback_flush_data_t *cfd;
 778
 779   ci = (cache_item_t *) value;
 780   cfd = (callback_flush_data_t *) data;
 781
 782   if (ci->flags & CI_FLAGS_IN_QUEUE)
 783     return FALSE;
 784
 785   if (ci->values_num > 0
 786       && (ci->last_flush_time <= cfd->abs_timeout || state != RUNNING))
 787   {
 788     enqueue_cache_item (ci, TAIL);
 789   }
 790   else if (((cfd->now - ci->last_flush_time) >= config_flush_interval)
 791       && (ci->values_num <= 0))
 792   {
 793     assert ((char *) key == ci->file);
 794     if (!rrd_add_ptr((void ***)&cfd->keys, &cfd->keys_num, (void *)key))
 795     {
 796       RRDD_LOG (LOG_ERR, "tree_callback_flush: rrd_add_ptrs failed.");
 797       return (FALSE);
 798     }
 799   }
 800
 801   return (FALSE);
 802 } /* }}} gboolean tree_callback_flush */
 803
 804 static int flush_old_values (int max_age)
 805 {
 806   callback_flush_data_t cfd;
 807   size_t k;
 808
 809   memset (&cfd, 0, sizeof (cfd));
 810   /* Pass the current time as user data so that we don't need to call
 811    * `time' for each node. */
 812   cfd.now = time (NULL);
 813   cfd.keys = NULL;
 814   cfd.keys_num = 0;
 815
 816   if (max_age > 0)
 817     cfd.abs_timeout = cfd.now - max_age;
 818   else
 819     cfd.abs_timeout = cfd.now + 2*config_write_jitter + 1;
 820
 821   /* `tree_callback_flush' will return the keys of all values that haven't
 822    * been touched in the last `config_flush_interval' seconds in `cfd'.
 823    * The char*'s in this array point to the same memory as ci->file, so we
 824    * don't need to free them separately. */
 825   g_tree_foreach (cache_tree, tree_callback_flush, (gpointer) &cfd);
 826
 827   for (k = 0; k < cfd.keys_num; k++)
 828   {
 829     gboolean status = g_tree_remove(cache_tree, cfd.keys[k]);
 830     /* should never fail, since we have held the cache_lock
 831      * the entire time */
 832     assert(status == TRUE);
 833   }
 834
 835   if (cfd.keys != NULL)
 836   {
 837     free (cfd.keys);
 838     cfd.keys = NULL;
 839   }
 840
 841   return (0);
 842 } /* int flush_old_values */
 843
 844 static void *flush_thread_main (void UNUSED(*args)) /* {{{ */
 845 {
 846   struct timeval now;
 847   struct timespec next_flush;
 848   int status;
 849
 850   gettimeofday (&now, NULL);
 851   next_flush.tv_sec = now.tv_sec + config_flush_interval;
 852   next_flush.tv_nsec = 1000 * now.tv_usec;
 853
 854   pthread_mutex_lock(&cache_lock);
 855
 856   while (state == RUNNING)
 857   {
 858     gettimeofday (&now, NULL);
 859     if ((now.tv_sec > next_flush.tv_sec)
 860         || ((now.tv_sec == next_flush.tv_sec)
 861           && ((1000 * now.tv_usec) > next_flush.tv_nsec)))
 862     {
 863       RRDD_LOG(LOG_DEBUG, "flushing old values");
 864
 865       /* Determine the time of the next cache flush. */
 866       next_flush.tv_sec = now.tv_sec + config_flush_interval;
 867
 868       /* Flush all values that haven't been written in the last
 869        * `config_write_interval' seconds. */
 870       flush_old_values (config_write_interval);
 871
 872       /* unlock the cache while we rotate so we don't block incoming
 873        * updates if the fsync() blocks on disk I/O */
 874       pthread_mutex_unlock(&cache_lock);
 875       journal_rotate();
 876       pthread_mutex_lock(&cache_lock);
 877     }
 878
 879     status = pthread_cond_timedwait(&flush_cond, &cache_lock, &next_flush);
 880     if (status != 0 && status != ETIMEDOUT)
 881     {
 882       RRDD_LOG (LOG_ERR, "flush_thread_main: "
 883                 "pthread_cond_timedwait returned %i.", status);
 884     }
 885   }
 886
 887   if (config_flush_at_shutdown)
 888     flush_old_values (-1); /* flush everything */
 889
 890   state = SHUTDOWN;
 891
 892   pthread_mutex_unlock(&cache_lock);
 893
 894   return NULL;
 895 } /* void *flush_thread_main */
 896
 897 static void *queue_thread_main (void UNUSED(*args)) /* {{{ */
 898 {
 899   pthread_mutex_lock (&cache_lock);
 900
 901   while (state != SHUTDOWN
 902          || (cache_queue_head != NULL && config_flush_at_shutdown))
 903   {
 904     cache_item_t *ci;
 905     char *file;
 906     char **values;
 907     size_t values_num;
 908     int status;
 909
 910     /* Now, check if there's something to store away. If not, wait until
 911      * something comes in. */
 912     if (cache_queue_head == NULL)
 913     {
 914       status = pthread_cond_wait (&queue_cond, &cache_lock);
 915       if ((status != 0) && (status != ETIMEDOUT))
 916       {
 917         RRDD_LOG (LOG_ERR, "queue_thread_main: "
 918             "pthread_cond_wait returned %i.", status);
 919       }
 920     }
 921
 922     /* Check if a value has arrived. This may be NULL if we timed out or there
 923      * was an interrupt such as a signal. */
 924     if (cache_queue_head == NULL)
 925       continue;
 926
 927     ci = cache_queue_head;
 928
 929     /* copy the relevant parts */
 930     file = strdup (ci->file);
 931     if (file == NULL)
 932     {
 933       RRDD_LOG (LOG_ERR, "queue_thread_main: strdup failed.");
 934       continue;
 935     }
 936
 937     assert(ci->values != NULL);
 938     assert(ci->values_num > 0);
 939
 940     values = ci->values;
 941     values_num = ci->values_num;
 942
 943     wipe_ci_values(ci, time(NULL));
 944     remove_from_queue(ci);
 945
 946     pthread_mutex_unlock (&cache_lock);
 947
 948     rrd_clear_error ();
 949     status = rrd_update_r (file, NULL, (int) values_num, (void *) values);
 950     if (status != 0)
 951     {
 952       RRDD_LOG (LOG_NOTICE, "queue_thread_main: "
 953           "rrd_update_r (%s) failed with status %i. (%s)",
 954           file, status, rrd_get_error());
 955     }
 956
 957     journal_write("wrote", file);
 958
 959     /* Search again in the tree.  It's possible someone issued a "FORGET"
 960      * while we were writing the update values. */
 961     pthread_mutex_lock(&cache_lock);
 962     ci = (cache_item_t *) g_tree_lookup(cache_tree, file);
 963     if (ci)
 964       pthread_cond_broadcast(&ci->flushed);
 965     pthread_mutex_unlock(&cache_lock);
 966
 967     if (status == 0)
 968     {
 969       pthread_mutex_lock (&stats_lock);
 970       stats_updates_written++;
 971       stats_data_sets_written += values_num;
 972       pthread_mutex_unlock (&stats_lock);
 973     }
 974
 975     rrd_free_ptrs((void ***) &values, &values_num);
 976     free(file);
 977
 978     pthread_mutex_lock (&cache_lock);
 979   }
 980   pthread_mutex_unlock (&cache_lock);
 981
 982   return (NULL);
 983 } /* }}} void *queue_thread_main */
 984
 985 static int buffer_get_field (char **buffer_ret, /* {{{ */
 986     size_t *buffer_size_ret, char **field_ret)
 987 {
 988   char *buffer;
 989   size_t buffer_pos;
 990   size_t buffer_size;
 991   char *field;
 992   size_t field_size;
 993   int status;
 994
 995   buffer = *buffer_ret;
 996   buffer_pos = 0;
 997   buffer_size = *buffer_size_ret;
 998   field = *buffer_ret;
 999   field_size = 0;
1000
1001   if (buffer_size <= 0)
1002     return (-1);
1003
1004   /* This is ensured by `handle_request'. */
1005   assert (buffer[buffer_size - 1] == '\0');
1006
1007   status = -1;
1008   while (buffer_pos < buffer_size)
1009   {
1010     /* Check for end-of-field or end-of-buffer */
1011     if (buffer[buffer_pos] == ' ' || buffer[buffer_pos] == '\0')
1012     {
1013       field[field_size] = 0;
1014       field_size++;
1015       buffer_pos++;
1016       status = 0;
1017       break;
1018     }
1019     /* Handle escaped characters. */
1020     else if (buffer[buffer_pos] == '\\')
1021     {
1022       if (buffer_pos >= (buffer_size - 1))
1023         break;
1024       buffer_pos++;
1025       field[field_size] = buffer[buffer_pos];
1026       field_size++;
1027       buffer_pos++;
1028     }
1029     /* Normal operation */
1030     else
1031     {
1032       field[field_size] = buffer[buffer_pos];
1033       field_size++;
1034       buffer_pos++;
1035     }
1036   } /* while (buffer_pos < buffer_size) */
1037
1038   if (status != 0)
1039     return (status);
1040
1041   *buffer_ret = buffer + buffer_pos;
1042   *buffer_size_ret = buffer_size - buffer_pos;
1043   *field_ret = field;
1044
1045   return (0);
1046 } /* }}} int buffer_get_field */
1047
1048 /* if we're restricting writes to the base directory,
1049  * check whether the file falls within the dir
1050  * returns 1 if OK, otherwise 0
1051  */
1052 static int check_file_access (const char *file, listen_socket_t *sock) /* {{{ */
1053 {
1054   assert(file != NULL);
1055
1056   if (!config_write_base_only
1057       || JOURNAL_REPLAY(sock)
1058       || config_base_dir == NULL)
1059     return 1;
1060
1061   if (strstr(file, "../") != NULL) goto err;
1062
1063   /* relative paths without "../" are ok */
1064   if (*file != '/') return 1;
1065
1066   /* file must be of the format base + "/" + <1+ char filename> */
1067   if (strlen(file) < _config_base_dir_len + 2) goto err;
1068   if (strncmp(file, config_base_dir, _config_base_dir_len) != 0) goto err;
1069   if (*(file + _config_base_dir_len) != '/') goto err;
1070
1071   return 1;
1072
1073 err:
1074   if (sock != NULL && sock->fd >= 0)
1075     send_response(sock, RESP_ERR, "%s\n", rrd_strerror(EACCES));
1076
1077   return 0;
1078 } /* }}} static int check_file_access */
1079
1080 /* when using a base dir, convert relative paths to absolute paths.
1081  * if necessary, modifies the "filename" pointer to point
1082  * to the new path created in "tmp".  "tmp" is provided
1083  * by the caller and sizeof(tmp) must be >= PATH_MAX.
1084  *
1085  * this allows us to optimize for the expected case (absolute path)
1086  * with a no-op.
1087  */
1088 static void get_abs_path(char **filename, char *tmp)
1089 {
1090   assert(tmp != NULL);
1091   assert(filename != NULL && *filename != NULL);
1092
1093   if (config_base_dir == NULL || **filename == '/')
1094     return;
1095
1096   snprintf(tmp, PATH_MAX, "%s/%s", config_base_dir, *filename);
1097   *filename = tmp;
1098 } /* }}} static int get_abs_path */
1099
1100 static int flush_file (const char *filename) /* {{{ */
1101 {
1102   cache_item_t *ci;
1103
1104   pthread_mutex_lock (&cache_lock);
1105
1106   ci = (cache_item_t *) g_tree_lookup (cache_tree, filename);
1107   if (ci == NULL)
1108   {
1109     pthread_mutex_unlock (&cache_lock);
1110     return (ENOENT);
1111   }
1112
1113   if (ci->values_num > 0)
1114   {
1115     /* Enqueue at head */
1116     enqueue_cache_item (ci, HEAD);
1117     pthread_cond_wait(&ci->flushed, &cache_lock);
1118   }
1119
1120   /* DO NOT DO ANYTHING WITH ci HERE!!  The entry
1121    * may have been purged during our cond_wait() */
1122
1123   pthread_mutex_unlock(&cache_lock);
1124
1125   return (0);
1126 } /* }}} int flush_file */
1127
1128 static int syntax_error(listen_socket_t *sock, command_t *cmd) /* {{{ */
1129 {
1130   char *err = "Syntax error.\n";
1131
1132   if (cmd && cmd->syntax)
1133     err = cmd->syntax;
1134
1135   return send_response(sock, RESP_ERR, "Usage: %s", err);
1136 } /* }}} static int syntax_error() */
1137
1138 static int handle_request_stats (HANDLER_PROTO) /* {{{ */
1139 {
1140   uint64_t copy_queue_length;
1141   uint64_t copy_updates_received;
1142   uint64_t copy_flush_received;
1143   uint64_t copy_updates_written;
1144   uint64_t copy_data_sets_written;
1145   uint64_t copy_journal_bytes;
1146   uint64_t copy_journal_rotate;
1147
1148   uint64_t tree_nodes_number;
1149   uint64_t tree_depth;
1150
1151   pthread_mutex_lock (&stats_lock);
1152   copy_queue_length       = stats_queue_length;
1153   copy_updates_received   = stats_updates_received;
1154   copy_flush_received     = stats_flush_received;
1155   copy_updates_written    = stats_updates_written;
1156   copy_data_sets_written  = stats_data_sets_written;
1157   copy_journal_bytes      = stats_journal_bytes;
1158   copy_journal_rotate     = stats_journal_rotate;
1159   pthread_mutex_unlock (&stats_lock);
1160
1161   pthread_mutex_lock (&cache_lock);
1162   tree_nodes_number = (uint64_t) g_tree_nnodes (cache_tree);
1163   tree_depth        = (uint64_t) g_tree_height (cache_tree);
1164   pthread_mutex_unlock (&cache_lock);
1165
1166   add_response_info(sock,
1167                     "QueueLength: %"PRIu64"\n", copy_queue_length);
1168   add_response_info(sock,
1169                     "UpdatesReceived: %"PRIu64"\n", copy_updates_received);
1170   add_response_info(sock,
1171                     "FlushesReceived: %"PRIu64"\n", copy_flush_received);
1172   add_response_info(sock,
1173                     "UpdatesWritten: %"PRIu64"\n", copy_updates_written);
1174   add_response_info(sock,
1175                     "DataSetsWritten: %"PRIu64"\n", copy_data_sets_written);
1176   add_response_info(sock, "TreeNodesNumber: %"PRIu64"\n", tree_nodes_number);
1177   add_response_info(sock, "TreeDepth: %"PRIu64"\n", tree_depth);
1178   add_response_info(sock, "JournalBytes: %"PRIu64"\n", copy_journal_bytes);
1179   add_response_info(sock, "JournalRotate: %"PRIu64"\n", copy_journal_rotate);
1180
1181   send_response(sock, RESP_OK, "Statistics follow\n");
1182
1183   return (0);
1184 } /* }}} int handle_request_stats */
1185
1186 static int handle_request_flush (HANDLER_PROTO) /* {{{ */
1187 {
1188   char *file, file_tmp[PATH_MAX];
1189   int status;
1190
1191   status = buffer_get_field (&buffer, &buffer_size, &file);
1192   if (status != 0)
1193   {
1194     return syntax_error(sock,cmd);
1195   }
1196   else
1197   {
1198     pthread_mutex_lock(&stats_lock);
1199     stats_flush_received++;
1200     pthread_mutex_unlock(&stats_lock);
1201
1202     get_abs_path(&file, file_tmp);
1203     if (!check_file_access(file, sock)) return 0;
1204
1205     status = flush_file (file);
1206     if (status == 0)
1207       return send_response(sock, RESP_OK, "Successfully flushed %s.\n", file);
1208     else if (status == ENOENT)
1209     {
1210       /* no file in our tree; see whether it exists at all */
1211       struct stat statbuf;
1212
1213       memset(&statbuf, 0, sizeof(statbuf));
1214       if (stat(file, &statbuf) == 0 && S_ISREG(statbuf.st_mode))
1215         return send_response(sock, RESP_OK, "Nothing to flush: %s.\n", file);
1216       else
1217         return send_response(sock, RESP_ERR, "No such file: %s.\n", file);
1218     }
1219     else if (status < 0)
1220       return send_response(sock, RESP_ERR, "Internal error.\n");
1221     else
1222       return send_response(sock, RESP_ERR, "Failed with status %i.\n", status);
1223   }
1224
1225   /* NOTREACHED */
1226   assert(1==0);
1227 } /* }}} int handle_request_flush */
1228
1229 static int handle_request_flushall(HANDLER_PROTO) /* {{{ */
1230 {
1231   RRDD_LOG(LOG_DEBUG, "Received FLUSHALL");
1232
1233   pthread_mutex_lock(&cache_lock);
1234   flush_old_values(-1);
1235   pthread_mutex_unlock(&cache_lock);
1236
1237   return send_response(sock, RESP_OK, "Started flush.\n");
1238 } /* }}} static int handle_request_flushall */
1239
1240 static int handle_request_pending(HANDLER_PROTO) /* {{{ */
1241 {
1242   int status;
1243   char *file, file_tmp[PATH_MAX];
1244   cache_item_t *ci;
1245
1246   status = buffer_get_field(&buffer, &buffer_size, &file);
1247   if (status != 0)
1248     return syntax_error(sock,cmd);
1249
1250   get_abs_path(&file, file_tmp);
1251
1252   pthread_mutex_lock(&cache_lock);
1253   ci = g_tree_lookup(cache_tree, file);
1254   if (ci == NULL)
1255   {
1256     pthread_mutex_unlock(&cache_lock);
1257     return send_response(sock, RESP_ERR, "%s\n", rrd_strerror(ENOENT));
1258   }
1259
1260   for (size_t i=0; i < ci->values_num; i++)
1261     add_response_info(sock, "%s\n", ci->values[i]);
1262
1263   pthread_mutex_unlock(&cache_lock);
1264   return send_response(sock, RESP_OK, "updates pending\n");
1265 } /* }}} static int handle_request_pending */
1266
1267 static int handle_request_forget(HANDLER_PROTO) /* {{{ */
1268 {
1269   int status;
1270   gboolean found;
1271   char *file, file_tmp[PATH_MAX];
1272
1273   status = buffer_get_field(&buffer, &buffer_size, &file);
1274   if (status != 0)
1275     return syntax_error(sock,cmd);
1276
1277   get_abs_path(&file, file_tmp);
1278   if (!check_file_access(file, sock)) return 0;
1279
1280   pthread_mutex_lock(&cache_lock);
1281   found = g_tree_remove(cache_tree, file);
1282   pthread_mutex_unlock(&cache_lock);
1283
1284   if (found == TRUE)
1285   {
1286     if (!JOURNAL_REPLAY(sock))
1287       journal_write("forget", file);
1288
1289     return send_response(sock, RESP_OK, "Gone!\n");
1290   }
1291   else
1292     return send_response(sock, RESP_ERR, "%s\n", rrd_strerror(ENOENT));
1293
1294   /* NOTREACHED */
1295   assert(1==0);
1296 } /* }}} static int handle_request_forget */
1297
1298 static int handle_request_queue (HANDLER_PROTO) /* {{{ */
1299 {
1300   cache_item_t *ci;
1301
1302   pthread_mutex_lock(&cache_lock);
1303
1304   ci = cache_queue_head;
1305   while (ci != NULL)
1306   {
1307     add_response_info(sock, "%d %s\n", ci->values_num, ci->file);
1308     ci = ci->next;
1309   }
1310
1311   pthread_mutex_unlock(&cache_lock);
1312
1313   return send_response(sock, RESP_OK, "in queue.\n");
1314 } /* }}} int handle_request_queue */
1315
1316 static int handle_request_update (HANDLER_PROTO) /* {{{ */
1317 {
1318   char *file, file_tmp[PATH_MAX];
1319   int values_num = 0;
1320   int status;
1321   char orig_buf[CMD_MAX];
1322
1323   cache_item_t *ci;
1324
1325   /* save it for the journal later */
1326   if (!JOURNAL_REPLAY(sock))
1327     strncpy(orig_buf, buffer, buffer_size);
1328
1329   status = buffer_get_field (&buffer, &buffer_size, &file);
1330   if (status != 0)
1331     return syntax_error(sock,cmd);
1332
1333   pthread_mutex_lock(&stats_lock);
1334   stats_updates_received++;
1335   pthread_mutex_unlock(&stats_lock);
1336
1337   get_abs_path(&file, file_tmp);
1338   if (!check_file_access(file, sock)) return 0;
1339
1340   pthread_mutex_lock (&cache_lock);
1341   ci = g_tree_lookup (cache_tree, file);
1342
1343   if (ci == NULL) /* {{{ */
1344   {
1345     struct stat statbuf;
1346     cache_item_t *tmp;
1347
1348     /* don't hold the lock while we setup; stat(2) might block */
1349     pthread_mutex_unlock(&cache_lock);
1350
1351     memset (&statbuf, 0, sizeof (statbuf));
1352     status = stat (file, &statbuf);
1353     if (status != 0)
1354     {
1355       RRDD_LOG (LOG_NOTICE, "handle_request_update: stat (%s) failed.", file);
1356
1357       status = errno;
1358       if (status == ENOENT)
1359         return send_response(sock, RESP_ERR, "No such file: %s\n", file);
1360       else
1361         return send_response(sock, RESP_ERR,
1362                              "stat failed with error %i.\n", status);
1363     }
1364     if (!S_ISREG (statbuf.st_mode))
1365       return send_response(sock, RESP_ERR, "Not a regular file: %s\n", file);
1366
1367     if (access(file, R_OK|W_OK) != 0)
1368       return send_response(sock, RESP_ERR, "Cannot read/write %s: %s\n",
1369                            file, rrd_strerror(errno));
1370
1371     ci = (cache_item_t *) malloc (sizeof (cache_item_t));
1372     if (ci == NULL)
1373     {
1374       RRDD_LOG (LOG_ERR, "handle_request_update: malloc failed.");
1375
1376       return send_response(sock, RESP_ERR, "malloc failed.\n");
1377     }
1378     memset (ci, 0, sizeof (cache_item_t));
1379
1380     ci->file = strdup (file);
1381     if (ci->file == NULL)
1382     {
1383       free (ci);
1384       RRDD_LOG (LOG_ERR, "handle_request_update: strdup failed.");
1385
1386       return send_response(sock, RESP_ERR, "strdup failed.\n");
1387     }
1388
1389     wipe_ci_values(ci, now);
1390     ci->flags = CI_FLAGS_IN_TREE;
1391     pthread_cond_init(&ci->flushed, NULL);
1392
1393     pthread_mutex_lock(&cache_lock);
1394
1395     /* another UPDATE might have added this entry in the meantime */
1396     tmp = g_tree_lookup (cache_tree, file);
1397     if (tmp == NULL)
1398       g_tree_replace (cache_tree, (void *) ci->file, (void *) ci);
1399     else
1400     {
1401       free_cache_item (ci);
1402       ci = tmp;
1403     }
1404
1405     /* state may have changed while we were unlocked */
1406     if (state == SHUTDOWN)
1407       return -1;
1408   } /* }}} */
1409   assert (ci != NULL);
1410
1411   /* don't re-write updates in replay mode */
1412   if (!JOURNAL_REPLAY(sock))
1413     journal_write("update", orig_buf);
1414
1415   while (buffer_size > 0)
1416   {
1417     char *value;
1418     time_t stamp;
1419     char *eostamp;
1420
1421     status = buffer_get_field (&buffer, &buffer_size, &value);
1422     if (status != 0)
1423     {
1424       RRDD_LOG (LOG_INFO, "handle_request_update: Error reading field.");
1425       break;
1426     }
1427
1428     /* make sure update time is always moving forward */
1429     stamp = strtol(value, &eostamp, 10);
1430     if (eostamp == value || eostamp == NULL || *eostamp != ':')
1431     {
1432       pthread_mutex_unlock(&cache_lock);
1433       return send_response(sock, RESP_ERR,
1434                            "Cannot find timestamp in '%s'!\n", value);
1435     }
1436     else if (stamp <= ci->last_update_stamp)
1437     {
1438       pthread_mutex_unlock(&cache_lock);
1439       return send_response(sock, RESP_ERR,
1440                            "illegal attempt to update using time %ld when last"
1441                            " update time is %ld (minimum one second step)\n",
1442                            stamp, ci->last_update_stamp);
1443     }
1444     else
1445       ci->last_update_stamp = stamp;
1446
1447     if (!rrd_add_strdup_chunk(&ci->values, &ci->values_num, value,
1448                               &ci->values_alloc, config_alloc_chunk))
1449     {
1450       RRDD_LOG (LOG_ERR, "handle_request_update: rrd_add_strdup failed.");
1451       continue;
1452     }
1453
1454     values_num++;
1455   }
1456
1457   if (((now - ci->last_flush_time) >= config_write_interval)
1458       && ((ci->flags & CI_FLAGS_IN_QUEUE) == 0)
1459       && (ci->values_num > 0))
1460   {
1461     enqueue_cache_item (ci, TAIL);
1462   }
1463
1464   pthread_mutex_unlock (&cache_lock);
1465
1466   if (values_num < 1)
1467     return send_response(sock, RESP_ERR, "No values updated.\n");
1468   else
1469     return send_response(sock, RESP_OK,
1470                          "errors, enqueued %i value(s).\n", values_num);
1471
1472   /* NOTREACHED */
1473   assert(1==0);
1474
1475 } /* }}} int handle_request_update */
1476
1477 static int handle_request_fetch (HANDLER_PROTO) /* {{{ */
1478 {
1479   char *file;
1480   char *cf;
1481
1482   char *start_str;
1483   char *end_str;
1484   rrd_time_value_t start_tv;
1485   rrd_time_value_t end_tv;
1486   time_t start_tm;
1487   time_t end_tm;
1488
1489   unsigned long step;
1490   unsigned long ds_cnt;
1491   char **ds_namv;
1492   rrd_value_t *data;
1493
1494   int status;
1495   unsigned long i;
1496   time_t t;
1497   rrd_value_t *data_ptr;
1498
1499   file = NULL;
1500   cf = NULL;
1501   start_str = NULL;
1502   end_str = NULL;
1503
1504   /* Read the arguments */
1505   do /* while (0) */
1506   {
1507     status = buffer_get_field (&buffer, &buffer_size, &file);
1508     if (status != 0)
1509       break;
1510
1511     status = buffer_get_field (&buffer, &buffer_size, &cf);
1512     if (status != 0)
1513       break;
1514
1515     status = buffer_get_field (&buffer, &buffer_size, &start_str);
1516     if (status != 0)
1517     {
1518       start_str = NULL;
1519       status = 0;
1520       break;
1521     }
1522
1523     status = buffer_get_field (&buffer, &buffer_size, &end_str);
1524     if (status != 0)
1525     {
1526       end_str = NULL;
1527       status = 0;
1528       break;
1529     }
1530   } while (0);
1531
1532   if (status != 0)
1533     return (syntax_error(sock,cmd));
1534
1535   status = flush_file (file);
1536   if ((status != 0) && (status != ENOENT))
1537     return (send_response (sock, RESP_ERR,
1538           "flush_file (%s) failed with status %i.\n", file, status));
1539
1540   /* Parse start time */
1541   if (start_str != NULL)
1542   {
1543     const char *errmsg;
1544
1545     errmsg = rrd_parsetime (start_str, &start_tv);
1546     if (errmsg != NULL)
1547       return (send_response(sock, RESP_ERR,
1548             "Cannot parse start time `%s': %s\n", start_str, errmsg));
1549   }
1550   else
1551     rrd_parsetime ("-86400", &start_tv);
1552
1553   /* Parse end time */
1554   if (end_str != NULL)
1555   {
1556     const char *errmsg;
1557
1558     errmsg = rrd_parsetime (end_str, &end_tv);
1559     if (errmsg != NULL)
1560       return (send_response(sock, RESP_ERR,
1561             "Cannot parse end time `%s': %s\n", end_str, errmsg));
1562   }
1563   else
1564     rrd_parsetime ("now", &end_tv);
1565
1566   start_tm = 0;
1567   end_tm = 0;
1568   status = rrd_proc_start_end (&start_tv, &end_tv, &start_tm, &end_tm);
1569   if (status != 0)
1570     return (send_response(sock, RESP_ERR,
1571           "rrd_proc_start_end failed: %s\n", rrd_get_error ()));
1572
1573   step = -1;
1574   ds_cnt = 0;
1575   ds_namv = NULL;
1576   data = NULL;
1577
1578   status = rrd_fetch_r (file, cf, &start_tm, &end_tm, &step,
1579       &ds_cnt, &ds_namv, &data);
1580   if (status != 0)
1581     return (send_response(sock, RESP_ERR,
1582           "rrd_fetch_r failed: %s\n", rrd_get_error ()));
1583
1584   add_response_info (sock, "FlushVersion: %lu\n", 1);
1585   add_response_info (sock, "Start: %lu\n", (unsigned long) start_tm);
1586   add_response_info (sock, "End: %lu\n", (unsigned long) end_tm);
1587   add_response_info (sock, "Step: %lu\n", step);
1588   add_response_info (sock, "DSCount: %lu\n", ds_cnt);
1589
1590 #define SSTRCAT(buffer,str,buffer_fill) do { \
1591     size_t str_len = strlen (str); \
1592     if ((buffer_fill + str_len) > sizeof (buffer)) \
1593       str_len = sizeof (buffer) - buffer_fill; \
1594     if (str_len > 0) { \
1595       strncpy (buffer + buffer_fill, str, str_len); \
1596       buffer_fill += str_len; \
1597       assert (buffer_fill <= sizeof (buffer)); \
1598       if (buffer_fill == sizeof (buffer)) \
1599         buffer[buffer_fill - 1] = 0; \
1600       else \
1601         buffer[buffer_fill] = 0; \
1602     } \
1603   } while (0)
1604
1605   { /* Add list of DS names */
1606     char linebuf[1024];
1607     size_t linebuf_fill;
1608
1609     memset (linebuf, 0, sizeof (linebuf));
1610     linebuf_fill = 0;
1611     for (i = 0; i < ds_cnt; i++)
1612     {
1613       if (i > 0)
1614         SSTRCAT (linebuf, " ", linebuf_fill);
1615       SSTRCAT (linebuf, ds_namv[i], linebuf_fill);
1616     }
1617     add_response_info (sock, "DSName: %s\n", linebuf);
1618   }
1619
1620   /* Add the actual data */
1621   assert (step > 0);
1622   data_ptr = data;
1623   for (t = start_tm + step; t <= end_tm; t += step)
1624   {
1625     char linebuf[1024];
1626     size_t linebuf_fill;
1627     char tmp[128];
1628
1629     memset (linebuf, 0, sizeof (linebuf));
1630     linebuf_fill = 0;
1631     for (i = 0; i < ds_cnt; i++)
1632     {
1633       snprintf (tmp, sizeof (tmp), " %0.10e", *data_ptr);
1634       tmp[sizeof (tmp) - 1] = 0;
1635       SSTRCAT (linebuf, tmp, linebuf_fill);
1636
1637       data_ptr++;
1638     }
1639
1640     add_response_info (sock, "%10lu:%s\n", (unsigned long) t, linebuf);
1641   } /* for (t) */
1642
1643   return (send_response (sock, RESP_OK, "Success\n"));
1644 #undef SSTRCAT
1645 } /* }}} int handle_request_fetch */
1646
1647 /* we came across a "WROTE" entry during journal replay.
1648  * throw away any values that we have accumulated for this file
1649  */
1650 static int handle_request_wrote (HANDLER_PROTO) /* {{{ */
1651 {
1652   cache_item_t *ci;
1653   const char *file = buffer;
1654
1655   pthread_mutex_lock(&cache_lock);
1656
1657   ci = g_tree_lookup(cache_tree, file);
1658   if (ci == NULL)
1659   {
1660     pthread_mutex_unlock(&cache_lock);
1661     return (0);
1662   }
1663
1664   if (ci->values)
1665     rrd_free_ptrs((void ***) &ci->values, &ci->values_num);
1666
1667   wipe_ci_values(ci, now);
1668   remove_from_queue(ci);
1669
1670   pthread_mutex_unlock(&cache_lock);
1671   return (0);
1672 } /* }}} int handle_request_wrote */
1673
1674 /* start "BATCH" processing */
1675 static int batch_start (HANDLER_PROTO) /* {{{ */
1676 {
1677   int status;
1678   if (sock->batch_start)
1679     return send_response(sock, RESP_ERR, "Already in BATCH\n");
1680
1681   status = send_response(sock, RESP_OK,
1682                          "Go ahead.  End with dot '.' on its own line.\n");
1683   sock->batch_start = time(NULL);
1684   sock->batch_cmd = 0;
1685
1686   return status;
1687 } /* }}} static int batch_start */
1688
1689 /* finish "BATCH" processing and return results to the client */
1690 static int batch_done (HANDLER_PROTO) /* {{{ */
1691 {
1692   assert(sock->batch_start);
1693   sock->batch_start = 0;
1694   sock->batch_cmd  = 0;
1695   return send_response(sock, RESP_OK, "errors\n");
1696 } /* }}} static int batch_done */
1697
1698 static int handle_request_quit (HANDLER_PROTO) /* {{{ */
1699 {
1700   return -1;
1701 } /* }}} static int handle_request_quit */
1702
1703 static command_t list_of_commands[] = { /* {{{ */
1704   {
1705     "UPDATE",
1706     handle_request_update,
1707     CMD_CONTEXT_ANY,
1708     "UPDATE <filename> <values> [<values> ...]\n"
1709     ,
1710     "Adds the given file to the internal cache if it is not yet known and\n"
1711     "appends the given value(s) to the entry. See the rrdcached(1) manpage\n"
1712     "for details.\n"
1713     "\n"
1714     "Each <values> has the following form:\n"
1715     "  <values> = <time>:<value>[:<value>[...]]\n"
1716     "See the rrdupdate(1) manpage for details.\n"
1717   },
1718   {
1719     "WROTE",
1720     handle_request_wrote,
1721     CMD_CONTEXT_JOURNAL,
1722     NULL,
1723     NULL
1724   },
1725   {
1726     "FLUSH",
1727     handle_request_flush,
1728     CMD_CONTEXT_CLIENT | CMD_CONTEXT_BATCH,
1729     "FLUSH <filename>\n"
1730     ,
1731     "Adds the given filename to the head of the update queue and returns\n"
1732     "after it has been dequeued.\n"
1733   },
1734   {
1735     "FLUSHALL",
1736     handle_request_flushall,
1737     CMD_CONTEXT_CLIENT,
1738     "FLUSHALL\n"
1739     ,
1740     "Triggers writing of all pending updates.  Returns immediately.\n"
1741   },
1742   {
1743     "PENDING",
1744     handle_request_pending,
1745     CMD_CONTEXT_CLIENT,
1746     "PENDING <filename>\n"
1747     ,
1748     "Shows any 'pending' updates for a file, in order.\n"
1749     "The updates shown have not yet been written to the underlying RRD file.\n"
1750   },
1751   {
1752     "FORGET",
1753     handle_request_forget,
1754     CMD_CONTEXT_ANY,
1755     "FORGET <filename>\n"
1756     ,
1757     "Removes the file completely from the cache.\n"
1758     "Any pending updates for the file will be lost.\n"
1759   },
1760   {
1761     "QUEUE",
1762     handle_request_queue,
1763     CMD_CONTEXT_CLIENT,
1764     "QUEUE\n"
1765     ,
1766         "Shows all files in the output queue.\n"
1767     "The output is zero or more lines in the following format:\n"
1768     "(where <num_vals> is the number of values to be written)\n"
1769     "\n"
1770     "<num_vals> <filename>\n"
1771   },
1772   {
1773     "STATS",
1774     handle_request_stats,
1775     CMD_CONTEXT_CLIENT,
1776     "STATS\n"
1777     ,
1778     "Returns some performance counters, see the rrdcached(1) manpage for\n"
1779     "a description of the values.\n"
1780   },
1781   {
1782     "HELP",
1783     handle_request_help,
1784     CMD_CONTEXT_CLIENT,
1785     "HELP [<command>]\n",
1786     NULL, /* special! */
1787   },
1788   {
1789     "BATCH",
1790     batch_start,
1791     CMD_CONTEXT_CLIENT,
1792     "BATCH\n"
1793     ,
1794     "The 'BATCH' command permits the client to initiate a bulk load\n"
1795     "   of commands to rrdcached.\n"
1796     "\n"
1797     "Usage:\n"
1798     "\n"
1799     "    client: BATCH\n"
1800     "    server: 0 Go ahead.  End with dot '.' on its own line.\n"
1801     "    client: command #1\n"
1802     "    client: command #2\n"
1803     "    client: ... and so on\n"
1804     "    client: .\n"
1805     "    server: 2 errors\n"
1806     "    server: 7 message for command #7\n"
1807     "    server: 9 message for command #9\n"
1808     "\n"
1809     "For more information, consult the rrdcached(1) documentation.\n"
1810   },
1811   {
1812     ".",   /* BATCH terminator */
1813     batch_done,
1814     CMD_CONTEXT_BATCH,
1815     NULL,
1816     NULL
1817   },
1818   {
1819     "FETCH",
1820     handle_request_fetch,
1821     CMD_CONTEXT_CLIENT,
1822     "FETCH <file> <CF> [<start> [<end>]]\n"
1823     ,
1824     "The 'FETCH' can be used by the client to retrieve values from an RRD file.\n"
1825   },
1826   {
1827     "QUIT",
1828     handle_request_quit,
1829     CMD_CONTEXT_CLIENT | CMD_CONTEXT_BATCH,
1830     "QUIT\n"
1831     ,
1832     "Disconnect from rrdcached.\n"
1833   }
1834 }; /* }}} command_t list_of_commands[] */
1835 static size_t list_of_commands_len = sizeof (list_of_commands)
1836   / sizeof (list_of_commands[0]);
1837
1838 static command_t *find_command(char *cmd)
1839 {
1840   size_t i;
1841
1842   for (i = 0; i < list_of_commands_len; i++)
1843     if (strcasecmp(cmd, list_of_commands[i].cmd) == 0)
1844       return (&list_of_commands[i]);
1845   return NULL;
1846 }
1847
1848 /* We currently use the index in the `list_of_commands' array as a bit position
1849  * in `listen_socket_t.permissions'. This member schould NEVER be accessed from
1850  * outside these functions so that switching to a more elegant storage method
1851  * is easily possible. */
1852 static ssize_t find_command_index (const char *cmd) /* {{{ */
1853 {
1854   size_t i;
1855
1856   for (i = 0; i < list_of_commands_len; i++)
1857     if (strcasecmp(cmd, list_of_commands[i].cmd) == 0)
1858       return ((ssize_t) i);
1859   return (-1);
1860 } /* }}} ssize_t find_command_index */
1861
1862 static int socket_permission_check (listen_socket_t *sock, /* {{{ */
1863     const char *cmd)
1864 {
1865   ssize_t i;
1866
1867   if (JOURNAL_REPLAY(sock))
1868     return (1);
1869
1870   if (cmd == NULL)
1871     return (-1);
1872
1873   if ((strcasecmp ("QUIT", cmd) == 0)
1874       || (strcasecmp ("HELP", cmd) == 0))
1875     return (1);
1876   else if (strcmp (".", cmd) == 0)
1877     cmd = "BATCH";
1878
1879   i = find_command_index (cmd);
1880   if (i < 0)
1881     return (-1);
1882   assert (i < 32);
1883
1884   if ((sock->permissions & (1 << i)) != 0)
1885     return (1);
1886   return (0);
1887 } /* }}} int socket_permission_check */
1888
1889 static int socket_permission_add (listen_socket_t *sock, /* {{{ */
1890     const char *cmd)
1891 {
1892   ssize_t i;
1893
1894   i = find_command_index (cmd);
1895   if (i < 0)
1896     return (-1);
1897   assert (i < 32);
1898
1899   sock->permissions |= (1 << i);
1900   return (0);
1901 } /* }}} int socket_permission_add */
1902
1903 /* check whether commands are received in the expected context */
1904 static int command_check_context(listen_socket_t *sock, command_t *cmd)
1905 {
1906   if (JOURNAL_REPLAY(sock))
1907     return (cmd->context & CMD_CONTEXT_JOURNAL);
1908   else if (sock->batch_start)
1909     return (cmd->context & CMD_CONTEXT_BATCH);
1910   else
1911     return (cmd->context & CMD_CONTEXT_CLIENT);
1912
1913   /* NOTREACHED */
1914   assert(1==0);
1915 }
1916
1917 static int handle_request_help (HANDLER_PROTO) /* {{{ */
1918 {
1919   int status;
1920   char *cmd_str;
1921   char *resp_txt;
1922   command_t *help = NULL;
1923
1924   status = buffer_get_field (&buffer, &buffer_size, &cmd_str);
1925   if (status == 0)
1926     help = find_command(cmd_str);
1927
1928   if (help && (help->syntax || help->help))
1929   {
1930     char tmp[CMD_MAX];
1931
1932     snprintf(tmp, sizeof(tmp)-1, "Help for %s\n", help->cmd);
1933     resp_txt = tmp;
1934
1935     if (help->syntax)
1936       add_response_info(sock, "Usage: %s\n", help->syntax);
1937
1938     if (help->help)
1939       add_response_info(sock, "%s\n", help->help);
1940   }
1941   else
1942   {
1943     size_t i;
1944
1945     resp_txt = "Command overview\n";
1946
1947     for (i = 0; i < list_of_commands_len; i++)
1948     {
1949       if (list_of_commands[i].syntax == NULL)
1950         continue;
1951       add_response_info (sock, "%s", list_of_commands[i].syntax);
1952     }
1953   }
1954
1955   return send_response(sock, RESP_OK, resp_txt);
1956 } /* }}} int handle_request_help */
1957
1958 static int handle_request (DISPATCH_PROTO) /* {{{ */
1959 {
1960   char *buffer_ptr = buffer;
1961   char *cmd_str = NULL;
1962   command_t *cmd = NULL;
1963   int status;
1964
1965   assert (buffer[buffer_size - 1] == '\0');
1966
1967   status = buffer_get_field (&buffer_ptr, &buffer_size, &cmd_str);
1968   if (status != 0)
1969   {
1970     RRDD_LOG (LOG_INFO, "handle_request: Unable parse command.");
1971     return (-1);
1972   }
1973
1974   if (sock != NULL && sock->batch_start)
1975     sock->batch_cmd++;
1976
1977   cmd = find_command(cmd_str);
1978   if (!cmd)
1979     return send_response(sock, RESP_ERR, "Unknown command: %s\n", cmd_str);
1980
1981   if (!socket_permission_check (sock, cmd->cmd))
1982     return send_response(sock, RESP_ERR, "Permission denied.\n");
1983
1984   if (!command_check_context(sock, cmd))
1985     return send_response(sock, RESP_ERR, "Can't use '%s' here.\n", cmd_str);
1986
1987   return cmd->handler(cmd, sock, now, buffer_ptr, buffer_size);
1988 } /* }}} int handle_request */
1989
1990 static void journal_set_free (journal_set *js) /* {{{ */
1991 {
1992   if (js == NULL)
1993     return;
1994
1995   rrd_free_ptrs((void ***) &js->files, &js->files_num);
1996
1997   free(js);
1998 } /* }}} journal_set_free */
1999
2000 static void journal_set_remove (journal_set *js) /* {{{ */
2001 {
2002   if (js == NULL)
2003     return;
2004
2005   for (uint i=0; i < js->files_num; i++)
2006   {
2007     RRDD_LOG(LOG_DEBUG, "removing old journal %s", js->files[i]);
2008     unlink(js->files[i]);
2009   }
2010 } /* }}} journal_set_remove */
2011
2012 /* close current journal file handle.
2013  * MUST hold journal_lock before calling */
2014 static void journal_close(void) /* {{{ */
2015 {
2016   if (journal_fh != NULL)
2017   {
2018     if (fclose(journal_fh) != 0)
2019       RRDD_LOG(LOG_ERR, "cannot close journal: %s", rrd_strerror(errno));
2020   }
2021
2022   journal_fh = NULL;
2023   journal_size = 0;
2024 } /* }}} journal_close */
2025
2026 /* MUST hold journal_lock before calling */
2027 static void journal_new_file(void) /* {{{ */
2028 {
2029   struct timeval now;
2030   int  new_fd;
2031   char new_file[PATH_MAX + 1];
2032
2033   assert(journal_dir != NULL);
2034   assert(journal_cur != NULL);
2035
2036   journal_close();
2037
2038   gettimeofday(&now, NULL);
2039   /* this format assures that the files sort in strcmp() order */
2040   snprintf(new_file, PATH_MAX, "%s/%s.%010d.%06d",
2041            journal_dir, JOURNAL_BASE, (int)now.tv_sec, (int)now.tv_usec);
2042
2043   new_fd = open(new_file, O_WRONLY|O_CREAT|O_APPEND,
2044                 S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
2045   if (new_fd < 0)
2046     goto error;
2047
2048   journal_fh = fdopen(new_fd, "a");
2049   if (journal_fh == NULL)
2050     goto error;
2051
2052   journal_size = ftell(journal_fh);
2053   RRDD_LOG(LOG_DEBUG, "started new journal %s", new_file);
2054
2055   /* record the file in the journal set */
2056   rrd_add_strdup(&journal_cur->files, &journal_cur->files_num, new_file);
2057
2058   return;
2059
2060 error:
2061   RRDD_LOG(LOG_CRIT,
2062            "JOURNALING DISABLED: Error while trying to create %s : %s",
2063            new_file, rrd_strerror(errno));
2064   RRDD_LOG(LOG_CRIT,
2065            "JOURNALING DISABLED: All values will be flushed at shutdown");
2066
2067   close(new_fd);
2068   config_flush_at_shutdown = 1;
2069
2070 } /* }}} journal_new_file */
2071
2072 /* MUST NOT hold journal_lock before calling this */
2073 static void journal_rotate(void) /* {{{ */
2074 {
2075   journal_set *old_js = NULL;
2076
2077   if (journal_dir == NULL)
2078     return;
2079
2080   RRDD_LOG(LOG_DEBUG, "rotating journals");
2081
2082   pthread_mutex_lock(&stats_lock);
2083   ++stats_journal_rotate;
2084   pthread_mutex_unlock(&stats_lock);
2085
2086   pthread_mutex_lock(&journal_lock);
2087
2088   journal_close();
2089
2090   /* rotate the journal sets */
2091   old_js = journal_old;
2092   journal_old = journal_cur;
2093   journal_cur = calloc(1, sizeof(journal_set));
2094
2095   if (journal_cur != NULL)
2096     journal_new_file();
2097   else
2098     RRDD_LOG(LOG_CRIT, "journal_rotate: malloc(journal_set) failed\n");
2099
2100   pthread_mutex_unlock(&journal_lock);
2101
2102   journal_set_remove(old_js);
2103   journal_set_free  (old_js);
2104
2105 } /* }}} static void journal_rotate */
2106
2107 /* MUST hold journal_lock when calling */
2108 static void journal_done(void) /* {{{ */
2109 {
2110   if (journal_cur == NULL)
2111     return;
2112
2113   journal_close();
2114
2115   if (config_flush_at_shutdown)
2116   {
2117     RRDD_LOG(LOG_INFO, "removing journals");
2118     journal_set_remove(journal_old);
2119     journal_set_remove(journal_cur);
2120   }
2121   else
2122   {
2123     RRDD_LOG(LOG_INFO, "expedited shutdown; "
2124              "journals will be used at next startup");
2125   }
2126
2127   journal_set_free(journal_cur);
2128   journal_set_free(journal_old);
2129   free(journal_dir);
2130
2131 } /* }}} static void journal_done */
2132
2133 static int journal_write(char *cmd, char *args) /* {{{ */
2134 {
2135   int chars;
2136
2137   if (journal_fh == NULL)
2138     return 0;
2139
2140   pthread_mutex_lock(&journal_lock);
2141   chars = fprintf(journal_fh, "%s %s\n", cmd, args);
2142   journal_size += chars;
2143
2144   if (journal_size > JOURNAL_MAX)
2145     journal_new_file();
2146
2147   pthread_mutex_unlock(&journal_lock);
2148
2149   if (chars > 0)
2150   {
2151     pthread_mutex_lock(&stats_lock);
2152     stats_journal_bytes += chars;
2153     pthread_mutex_unlock(&stats_lock);
2154   }
2155
2156   return chars;
2157 } /* }}} static int journal_write */
2158
2159 static int journal_replay (const char *file) /* {{{ */
2160 {
2161   FILE *fh;
2162   int entry_cnt = 0;
2163   int fail_cnt = 0;
2164   uint64_t line = 0;
2165   char entry[CMD_MAX];
2166   time_t now;
2167
2168   if (file == NULL) return 0;
2169
2170   {
2171     char *reason = "unknown error";
2172     int status = 0;
2173     struct stat statbuf;
2174
2175     memset(&statbuf, 0, sizeof(statbuf));
2176     if (stat(file, &statbuf) != 0)
2177     {
2178       reason = "stat error";
2179       status = errno;
2180     }
2181     else if (!S_ISREG(statbuf.st_mode))
2182     {
2183       reason = "not a regular file";
2184       status = EPERM;
2185     }
2186     if (statbuf.st_uid != daemon_uid)
2187     {
2188       reason = "not owned by daemon user";
2189       status = EACCES;
2190     }
2191     if (statbuf.st_mode & (S_IWGRP|S_IWOTH))
2192     {
2193       reason = "must not be user/group writable";
2194       status = EACCES;
2195     }
2196
2197     if (status != 0)
2198     {
2199       RRDD_LOG(LOG_ERR, "journal_replay: %s : %s (%s)",
2200                file, rrd_strerror(status), reason);
2201       return 0;
2202     }
2203   }
2204
2205   fh = fopen(file, "r");
2206   if (fh == NULL)
2207   {
2208     if (errno != ENOENT)
2209       RRDD_LOG(LOG_ERR, "journal_replay: cannot open journal file: '%s' (%s)",
2210                file, rrd_strerror(errno));
2211     return 0;
2212   }
2213   else
2214     RRDD_LOG(LOG_NOTICE, "replaying from journal: %s", file);
2215
2216   now = time(NULL);
2217
2218   while(!feof(fh))
2219   {
2220     size_t entry_len;
2221
2222     ++line;
2223     if (fgets(entry, sizeof(entry), fh) == NULL)
2224       break;
2225     entry_len = strlen(entry);
2226
2227     /* check \n termination in case journal writing crashed mid-line */
2228     if (entry_len == 0)
2229       continue;
2230     else if (entry[entry_len - 1] != '\n')
2231     {
2232       RRDD_LOG(LOG_NOTICE, "Malformed journal entry at line %"PRIu64, line);
2233       ++fail_cnt;
2234       continue;
2235     }
2236
2237     entry[entry_len - 1] = '\0';
2238
2239     if (handle_request(NULL, now, entry, entry_len) == 0)
2240       ++entry_cnt;
2241     else
2242       ++fail_cnt;
2243   }
2244
2245   fclose(fh);
2246
2247   RRDD_LOG(LOG_INFO, "Replayed %d entries (%d failures)",
2248            entry_cnt, fail_cnt);
2249
2250   return entry_cnt > 0 ? 1 : 0;
2251 } /* }}} static int journal_replay */
2252
2253 static int journal_sort(const void *v1, const void *v2)
2254 {
2255   char **jn1 = (char **) v1;
2256   char **jn2 = (char **) v2;
2257
2258   return strcmp(*jn1,*jn2);
2259 }
2260
2261 static void journal_init(void) /* {{{ */
2262 {
2263   int had_journal = 0;
2264   DIR *dir;
2265   struct dirent *dent;
2266   char path[PATH_MAX+1];
2267
2268   if (journal_dir == NULL) return;
2269
2270   pthread_mutex_lock(&journal_lock);
2271
2272   journal_cur = calloc(1, sizeof(journal_set));
2273   if (journal_cur == NULL)
2274   {
2275     RRDD_LOG(LOG_CRIT, "journal_rotate: malloc(journal_set) failed\n");
2276     return;
2277   }
2278
2279   RRDD_LOG(LOG_INFO, "checking for journal files");
2280
2281   /* Handle old journal files during transition.  This gives them the
2282    * correct sort order.  TODO: remove after first release
2283    */
2284   {
2285     char old_path[PATH_MAX+1];
2286     snprintf(old_path, PATH_MAX, "%s/%s", journal_dir, JOURNAL_BASE ".old" );
2287     snprintf(path,     PATH_MAX, "%s/%s", journal_dir, JOURNAL_BASE ".0000");
2288     rename(old_path, path);
2289
2290     snprintf(old_path, PATH_MAX, "%s/%s", journal_dir, JOURNAL_BASE        );
2291     snprintf(path,     PATH_MAX, "%s/%s", journal_dir, JOURNAL_BASE ".0001");
2292     rename(old_path, path);
2293   }
2294
2295   dir = opendir(journal_dir);
2296   if (!dir) {
2297     RRDD_LOG(LOG_CRIT, "journal_init: opendir(%s) failed\n", journal_dir);
2298     return;
2299   }
2300   while ((dent = readdir(dir)) != NULL)
2301   {
2302     /* looks like a journal file? */
2303     if (strncmp(dent->d_name, JOURNAL_BASE, strlen(JOURNAL_BASE)))
2304       continue;
2305
2306     snprintf(path, PATH_MAX, "%s/%s", journal_dir, dent->d_name);
2307
2308     if (!rrd_add_strdup(&journal_cur->files, &journal_cur->files_num, path))
2309     {
2310       RRDD_LOG(LOG_CRIT, "journal_init: cannot add journal file %s!",
2311                dent->d_name);
2312       break;
2313     }
2314   }
2315   closedir(dir);
2316
2317   qsort(journal_cur->files, journal_cur->files_num,
2318         sizeof(journal_cur->files[0]), journal_sort);
2319
2320   for (uint i=0; i < journal_cur->files_num; i++)
2321     had_journal += journal_replay(journal_cur->files[i]);
2322
2323   journal_new_file();
2324
2325   /* it must have been a crash.  start a flush */
2326   if (had_journal && config_flush_at_shutdown)
2327     flush_old_values(-1);
2328
2329   pthread_mutex_unlock(&journal_lock);
2330
2331   RRDD_LOG(LOG_INFO, "journal processing complete");
2332
2333 } /* }}} static void journal_init */
2334
2335 static void free_listen_socket(listen_socket_t *sock) /* {{{ */
2336 {
2337   assert(sock != NULL);
2338
2339   free(sock->rbuf);  sock->rbuf = NULL;
2340   free(sock->wbuf);  sock->wbuf = NULL;
2341   free(sock);
2342 } /* }}} void free_listen_socket */
2343
2344 static void close_connection(listen_socket_t *sock) /* {{{ */
2345 {
2346   if (sock->fd >= 0)
2347   {
2348     close(sock->fd);
2349     sock->fd = -1;
2350   }
2351
2352   free_listen_socket(sock);
2353
2354 } /* }}} void close_connection */
2355
2356 static void *connection_thread_main (void *args) /* {{{ */
2357 {
2358   listen_socket_t *sock;
2359   int fd;
2360
2361   sock = (listen_socket_t *) args;
2362   fd = sock->fd;
2363
2364   /* init read buffers */
2365   sock->next_read = sock->next_cmd = 0;
2366   sock->rbuf = malloc(RBUF_SIZE);
2367   if (sock->rbuf == NULL)
2368   {
2369     RRDD_LOG(LOG_ERR, "connection_thread_main: cannot malloc read buffer");
2370     close_connection(sock);
2371     return NULL;
2372   }
2373
2374   pthread_mutex_lock (&connection_threads_lock);
2375   connection_threads_num++;
2376   pthread_mutex_unlock (&connection_threads_lock);
2377
2378   while (state == RUNNING)
2379   {
2380     char *cmd;
2381     ssize_t cmd_len;
2382     ssize_t rbytes;
2383     time_t now;
2384
2385     struct pollfd pollfd;
2386     int status;
2387
2388     pollfd.fd = fd;
2389     pollfd.events = POLLIN | POLLPRI;
2390     pollfd.revents = 0;
2391
2392     status = poll (&pollfd, 1, /* timeout = */ 500);
2393     if (state != RUNNING)
2394       break;
2395     else if (status == 0) /* timeout */
2396       continue;
2397     else if (status < 0) /* error */
2398     {
2399       status = errno;
2400       if (status != EINTR)
2401         RRDD_LOG (LOG_ERR, "connection_thread_main: poll(2) failed.");
2402       continue;
2403     }
2404
2405     if ((pollfd.revents & POLLHUP) != 0) /* normal shutdown */
2406       break;
2407     else if ((pollfd.revents & (POLLIN | POLLPRI)) == 0)
2408     {
2409       RRDD_LOG (LOG_WARNING, "connection_thread_main: "
2410           "poll(2) returned something unexpected: %#04hx",
2411           pollfd.revents);
2412       break;
2413     }
2414
2415     rbytes = read(fd, sock->rbuf + sock->next_read,
2416                   RBUF_SIZE - sock->next_read);
2417     if (rbytes < 0)
2418     {
2419       RRDD_LOG(LOG_ERR, "connection_thread_main: read() failed.");
2420       break;
2421     }
2422     else if (rbytes == 0)
2423       break; /* eof */
2424
2425     sock->next_read += rbytes;
2426
2427     if (sock->batch_start)
2428       now = sock->batch_start;
2429     else
2430       now = time(NULL);
2431
2432     while ((cmd = next_cmd(sock, &cmd_len)) != NULL)
2433     {
2434       status = handle_request (sock, now, cmd, cmd_len+1);
2435       if (status != 0)
2436         goto out_close;
2437     }
2438   }
2439
2440 out_close:
2441   close_connection(sock);
2442
2443   /* Remove this thread from the connection threads list */
2444   pthread_mutex_lock (&connection_threads_lock);
2445   connection_threads_num--;
2446   if (connection_threads_num <= 0)
2447     pthread_cond_broadcast(&connection_threads_done);
2448   pthread_mutex_unlock (&connection_threads_lock);
2449
2450   return (NULL);
2451 } /* }}} void *connection_thread_main */
2452
2453 static int open_listen_socket_unix (const listen_socket_t *sock) /* {{{ */
2454 {
2455   int fd;
2456   struct sockaddr_un sa;
2457   listen_socket_t *temp;
2458   int status;
2459   const char *path;
2460   char *path_copy, *dir;
2461
2462   path = sock->addr;
2463   if (strncmp(path, "unix:", strlen("unix:")) == 0)
2464     path += strlen("unix:");
2465
2466   /* dirname may modify its argument */
2467   path_copy = strdup(path);
2468   if (path_copy == NULL)
2469   {
2470     fprintf(stderr, "rrdcached: strdup(): %s\n",
2471         rrd_strerror(errno));
2472     return (-1);
2473   }
2474
2475   dir = dirname(path_copy);
2476   if (rrd_mkdir_p(dir, 0777) != 0)
2477   {
2478     fprintf(stderr, "Failed to create socket directory '%s': %s\n",
2479         dir, rrd_strerror(errno));
2480     return (-1);
2481   }
2482
2483   free(path_copy);
2484
2485   temp = (listen_socket_t *) rrd_realloc (listen_fds,
2486       sizeof (listen_fds[0]) * (listen_fds_num + 1));
2487   if (temp == NULL)
2488   {
2489     fprintf (stderr, "rrdcached: open_listen_socket_unix: realloc failed.\n");
2490     return (-1);
2491   }
2492   listen_fds = temp;
2493   memcpy (listen_fds + listen_fds_num, sock, sizeof (listen_fds[0]));
2494
2495   fd = socket (PF_UNIX, SOCK_STREAM, /* protocol = */ 0);
2496   if (fd < 0)
2497   {
2498     fprintf (stderr, "rrdcached: unix socket(2) failed: %s\n",
2499              rrd_strerror(errno));
2500     return (-1);
2501   }
2502
2503   memset (&sa, 0, sizeof (sa));
2504   sa.sun_family = AF_UNIX;
2505   strncpy (sa.sun_path, path, sizeof (sa.sun_path) - 1);
2506
2507   /* if we've gotten this far, we own the pid file.  any daemon started
2508    * with the same args must not be alive.  therefore, ensure that we can
2509    * create the socket...
2510    */
2511   unlink(path);
2512
2513   status = bind (fd, (struct sockaddr *) &sa, sizeof (sa));
2514   if (status != 0)
2515   {
2516     fprintf (stderr, "rrdcached: bind(%s) failed: %s.\n",
2517              path, rrd_strerror(errno));
2518     close (fd);
2519     return (-1);
2520   }
2521
2522   /* tweak the sockets group ownership */
2523   if (sock->socket_group != (gid_t)-1)
2524   {
2525     if ( (chown(path, getuid(), sock->socket_group) != 0) ||
2526          (chmod(path, (S_IRUSR|S_IWUSR|S_IXUSR | S_IRGRP|S_IWGRP)) != 0) )
2527     {
2528       fprintf(stderr, "rrdcached: failed to set socket group permissions (%s)\n", strerror(errno));
2529     }
2530   }
2531
2532   if (sock->socket_permissions != (mode_t)-1)
2533   {
2534     if (chmod(path, sock->socket_permissions) != 0)
2535       fprintf(stderr, "rrdcached: failed to set socket file permissions (%o): %s\n",
2536           (unsigned int)sock->socket_permissions, strerror(errno));
2537   }
2538
2539   status = listen (fd, /* backlog = */ 10);
2540   if (status != 0)
2541   {
2542     fprintf (stderr, "rrdcached: listen(%s) failed: %s.\n",
2543              path, rrd_strerror(errno));
2544     close (fd);
2545     unlink (path);
2546     return (-1);
2547   }
2548
2549   listen_fds[listen_fds_num].fd = fd;
2550   listen_fds[listen_fds_num].family = PF_UNIX;
2551   strncpy(listen_fds[listen_fds_num].addr, path,
2552           sizeof (listen_fds[listen_fds_num].addr) - 1);
2553   listen_fds_num++;
2554
2555   return (0);
2556 } /* }}} int open_listen_socket_unix */
2557
2558 static int open_listen_socket_network(const listen_socket_t *sock) /* {{{ */
2559 {
2560   struct addrinfo ai_hints;
2561   struct addrinfo *ai_res;
2562   struct addrinfo *ai_ptr;
2563   char addr_copy[NI_MAXHOST];
2564   char *addr;
2565   char *port;
2566   int status;
2567
2568   strncpy (addr_copy, sock->addr, sizeof(addr_copy)-1);
2569   addr_copy[sizeof (addr_copy) - 1] = 0;
2570   addr = addr_copy;
2571
2572   memset (&ai_hints, 0, sizeof (ai_hints));
2573   ai_hints.ai_flags = 0;
2574 #ifdef AI_ADDRCONFIG
2575   ai_hints.ai_flags |= AI_ADDRCONFIG;
2576 #endif
2577   ai_hints.ai_family = AF_UNSPEC;
2578   ai_hints.ai_socktype = SOCK_STREAM;
2579
2580   port = NULL;
2581   if (*addr == '[') /* IPv6+port format */
2582   {
2583     /* `addr' is something like "[2001:780:104:2:211:24ff:feab:26f8]:12345" */
2584     addr++;
2585
2586     port = strchr (addr, ']');
2587     if (port == NULL)
2588     {
2589       fprintf (stderr, "rrdcached: Malformed address: %s\n", sock->addr);
2590       return (-1);
2591     }
2592     *port = 0;
2593     port++;
2594
2595     if (*port == ':')
2596       port++;
2597     else if (*port == 0)
2598       port = NULL;
2599     else
2600     {
2601       fprintf (stderr, "rrdcached: Garbage after address: %s\n", port);
2602       return (-1);
2603     }
2604   } /* if (*addr == '[') */
2605   else
2606   {
2607     port = rindex(addr, ':');
2608     if (port != NULL)
2609     {
2610       *port = 0;
2611       port++;
2612     }
2613   }
2614   ai_res = NULL;
2615   status = getaddrinfo (addr,
2616                         port == NULL ? RRDCACHED_DEFAULT_PORT : port,
2617                         &ai_hints, &ai_res);
2618   if (status != 0)
2619   {
2620     fprintf (stderr, "rrdcached: getaddrinfo(%s) failed: %s\n",
2621              addr, gai_strerror (status));
2622     return (-1);
2623   }
2624
2625   for (ai_ptr = ai_res; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next)
2626   {
2627     int fd;
2628     listen_socket_t *temp;
2629     int one = 1;
2630
2631     temp = (listen_socket_t *) rrd_realloc (listen_fds,
2632         sizeof (listen_fds[0]) * (listen_fds_num + 1));
2633     if (temp == NULL)
2634     {
2635       fprintf (stderr,
2636                "rrdcached: open_listen_socket_network: realloc failed.\n");
2637       continue;
2638     }
2639     listen_fds = temp;
2640     memcpy (listen_fds + listen_fds_num, sock, sizeof (listen_fds[0]));
2641
2642     fd = socket (ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol);
2643     if (fd < 0)
2644     {
2645       fprintf (stderr, "rrdcached: network socket(2) failed: %s.\n",
2646                rrd_strerror(errno));
2647       continue;
2648     }
2649
2650     setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
2651
2652     status = bind (fd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
2653     if (status != 0)
2654     {
2655       fprintf (stderr, "rrdcached: bind(%s) failed: %s.\n",
2656                sock->addr, rrd_strerror(errno));
2657       close (fd);
2658       continue;
2659     }
2660
2661     status = listen (fd, /* backlog = */ 10);
2662     if (status != 0)
2663     {
2664       fprintf (stderr, "rrdcached: listen(%s) failed: %s\n.",
2665                sock->addr, rrd_strerror(errno));
2666       close (fd);
2667       freeaddrinfo(ai_res);
2668       return (-1);
2669     }
2670
2671     listen_fds[listen_fds_num].fd = fd;
2672     listen_fds[listen_fds_num].family = ai_ptr->ai_family;
2673     listen_fds_num++;
2674   } /* for (ai_ptr) */
2675
2676   freeaddrinfo(ai_res);
2677   return (0);
2678 } /* }}} static int open_listen_socket_network */
2679
2680 static int open_listen_socket (const listen_socket_t *sock) /* {{{ */
2681 {
2682   assert(sock != NULL);
2683   assert(sock->addr != NULL);
2684
2685   if (strncmp ("unix:", sock->addr, strlen ("unix:")) == 0
2686       || sock->addr[0] == '/')
2687     return (open_listen_socket_unix(sock));
2688   else
2689     return (open_listen_socket_network(sock));
2690 } /* }}} int open_listen_socket */
2691
2692 static int close_listen_sockets (void) /* {{{ */
2693 {
2694   size_t i;
2695
2696   for (i = 0; i < listen_fds_num; i++)
2697   {
2698     close (listen_fds[i].fd);
2699
2700     if (listen_fds[i].family == PF_UNIX)
2701       unlink(listen_fds[i].addr);
2702   }
2703
2704   free (listen_fds);
2705   listen_fds = NULL;
2706   listen_fds_num = 0;
2707
2708   return (0);
2709 } /* }}} int close_listen_sockets */
2710
2711 static void *listen_thread_main (void UNUSED(*args)) /* {{{ */
2712 {
2713   struct pollfd *pollfds;
2714   int pollfds_num;
2715   int status;
2716   int i;
2717
2718   if (listen_fds_num < 1)
2719   {
2720     RRDD_LOG(LOG_ERR, "listen_thread_main: no listen_fds !");
2721     return (NULL);
2722   }
2723
2724   pollfds_num = listen_fds_num;
2725   pollfds = (struct pollfd *) malloc (sizeof (*pollfds) * pollfds_num);
2726   if (pollfds == NULL)
2727   {
2728     RRDD_LOG (LOG_ERR, "listen_thread_main: malloc failed.");
2729     return (NULL);
2730   }
2731   memset (pollfds, 0, sizeof (*pollfds) * pollfds_num);
2732
2733   RRDD_LOG(LOG_INFO, "listening for connections");
2734
2735   while (state == RUNNING)
2736   {
2737     for (i = 0; i < pollfds_num; i++)
2738     {
2739       pollfds[i].fd = listen_fds[i].fd;
2740       pollfds[i].events = POLLIN | POLLPRI;
2741       pollfds[i].revents = 0;
2742     }
2743
2744     status = poll (pollfds, pollfds_num, /* timeout = */ 1000);
2745     if (state != RUNNING)
2746       break;
2747     else if (status == 0) /* timeout */
2748       continue;
2749     else if (status < 0) /* error */
2750     {
2751       status = errno;
2752       if (status != EINTR)
2753       {
2754         RRDD_LOG (LOG_ERR, "listen_thread_main: poll(2) failed.");
2755       }
2756       continue;
2757     }
2758
2759     for (i = 0; i < pollfds_num; i++)
2760     {
2761       listen_socket_t *client_sock;
2762       struct sockaddr_storage client_sa;
2763       socklen_t client_sa_size;
2764       pthread_t tid;
2765       pthread_attr_t attr;
2766
2767       if (pollfds[i].revents == 0)
2768         continue;
2769
2770       if ((pollfds[i].revents & (POLLIN | POLLPRI)) == 0)
2771       {
2772         RRDD_LOG (LOG_ERR, "listen_thread_main: "
2773             "poll(2) returned something unexpected for listen FD #%i.",
2774             pollfds[i].fd);
2775         continue;
2776       }
2777
2778       client_sock = (listen_socket_t *) malloc (sizeof (listen_socket_t));
2779       if (client_sock == NULL)
2780       {
2781         RRDD_LOG (LOG_ERR, "listen_thread_main: malloc failed.");
2782         continue;
2783       }
2784       memcpy(client_sock, &listen_fds[i], sizeof(listen_fds[0]));
2785
2786       client_sa_size = sizeof (client_sa);
2787       client_sock->fd = accept (pollfds[i].fd,
2788           (struct sockaddr *) &client_sa, &client_sa_size);
2789       if (client_sock->fd < 0)
2790       {
2791         RRDD_LOG (LOG_ERR, "listen_thread_main: accept(2) failed.");
2792         free(client_sock);
2793         continue;
2794       }
2795
2796       pthread_attr_init (&attr);
2797       pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
2798
2799       status = pthread_create (&tid, &attr, connection_thread_main,
2800                                client_sock);
2801       if (status != 0)
2802       {
2803         RRDD_LOG (LOG_ERR, "listen_thread_main: pthread_create failed.");
2804         close_connection(client_sock);
2805         continue;
2806       }
2807     } /* for (pollfds_num) */
2808   } /* while (state == RUNNING) */
2809
2810   RRDD_LOG(LOG_INFO, "starting shutdown");
2811
2812   close_listen_sockets ();
2813
2814   pthread_mutex_lock (&connection_threads_lock);
2815   while (connection_threads_num > 0)
2816     pthread_cond_wait(&connection_threads_done, &connection_threads_lock);
2817   pthread_mutex_unlock (&connection_threads_lock);
2818
2819   free(pollfds);
2820
2821   return (NULL);
2822 } /* }}} void *listen_thread_main */
2823
2824 static int daemonize (void) /* {{{ */
2825 {
2826   int pid_fd;
2827   char *base_dir;
2828
2829   daemon_uid = geteuid();
2830
2831   pid_fd = open_pidfile("create", O_CREAT|O_EXCL|O_WRONLY);
2832   if (pid_fd < 0)
2833     pid_fd = check_pidfile();
2834   if (pid_fd < 0)
2835     return pid_fd;
2836
2837   /* open all the listen sockets */
2838   if (config_listen_address_list_len > 0)
2839   {
2840     for (size_t i = 0; i < config_listen_address_list_len; i++)
2841       open_listen_socket (config_listen_address_list[i]);
2842
2843     rrd_free_ptrs((void ***) &config_listen_address_list,
2844                   &config_listen_address_list_len);
2845   }
2846   else
2847   {
2848     listen_socket_t sock;
2849     memset(&sock, 0, sizeof(sock));
2850     strncpy(sock.addr, RRDCACHED_DEFAULT_ADDRESS, sizeof(sock.addr)-1);
2851     open_listen_socket (&sock);
2852   }
2853
2854   if (listen_fds_num < 1)
2855   {
2856     fprintf (stderr, "rrdcached: FATAL: cannot open any listen sockets\n");
2857     goto error;
2858   }
2859
2860   if (!stay_foreground)
2861   {
2862     pid_t child;
2863
2864     child = fork ();
2865     if (child < 0)
2866     {
2867       fprintf (stderr, "daemonize: fork(2) failed.\n");
2868       goto error;
2869     }
2870     else if (child > 0)
2871       exit(0);
2872
2873     /* Become session leader */
2874     setsid ();
2875
2876     /* Open the first three file descriptors to /dev/null */
2877     close (2);
2878     close (1);
2879     close (0);
2880
2881     open ("/dev/null", O_RDWR);
2882     if (dup(0) == -1 || dup(0) == -1){
2883         RRDD_LOG (LOG_ERR, "faild to run dup.\n");
2884     }
2885   } /* if (!stay_foreground) */
2886
2887   /* Change into the /tmp directory. */
2888   base_dir = (config_base_dir != NULL)
2889     ? config_base_dir
2890     : "/tmp";
2891
2892   if (chdir (base_dir) != 0)
2893   {
2894     fprintf (stderr, "daemonize: chdir (%s) failed.\n", base_dir);
2895     goto error;
2896   }
2897
2898   install_signal_handlers();
2899
2900   openlog ("rrdcached", LOG_PID, LOG_DAEMON);
2901   RRDD_LOG(LOG_INFO, "starting up");
2902
2903   cache_tree = g_tree_new_full ((GCompareDataFunc) strcmp, NULL, NULL,
2904                                 (GDestroyNotify) free_cache_item);
2905   if (cache_tree == NULL)
2906   {
2907     RRDD_LOG (LOG_ERR, "daemonize: g_tree_new failed.");
2908     goto error;
2909   }
2910
2911   return write_pidfile (pid_fd);
2912
2913 error:
2914   remove_pidfile();
2915   return -1;
2916 } /* }}} int daemonize */
2917
2918 static int cleanup (void) /* {{{ */
2919 {
2920   pthread_cond_broadcast (&flush_cond);
2921   pthread_join (flush_thread, NULL);
2922
2923   pthread_cond_broadcast (&queue_cond);
2924   for (int i = 0; i < config_queue_threads; i++)
2925     pthread_join (queue_threads[i], NULL);
2926
2927   if (config_flush_at_shutdown)
2928   {
2929     assert(cache_queue_head == NULL);
2930     RRDD_LOG(LOG_INFO, "clean shutdown; all RRDs flushed");
2931   }
2932
2933   free(queue_threads);
2934   free(config_base_dir);
2935
2936   pthread_mutex_lock(&cache_lock);
2937   g_tree_destroy(cache_tree);
2938
2939   pthread_mutex_lock(&journal_lock);
2940   journal_done();
2941
2942   RRDD_LOG(LOG_INFO, "goodbye");
2943   closelog ();
2944
2945   remove_pidfile ();
2946   free(config_pid_file);
2947
2948   return (0);
2949 } /* }}} int cleanup */
2950
2951 static int read_options (int argc, char **argv) /* {{{ */
2952 {
2953   int option;
2954   int status = 0;
2955
2956   char **permissions = NULL;
2957   size_t permissions_len = 0;
2958
2959   gid_t  socket_group = (gid_t)-1;
2960   mode_t socket_permissions = (mode_t)-1;
2961
2962   while ((option = getopt(argc, argv, "gl:s:m:P:f:w:z:t:Bb:p:Fj:a:h?")) != -1)
2963   {
2964     switch (option)
2965     {
2966       case 'g':
2967         stay_foreground=1;
2968         break;
2969
2970       case 'l':
2971       {
2972         listen_socket_t *new;
2973
2974         new = malloc(sizeof(listen_socket_t));
2975         if (new == NULL)
2976         {
2977           fprintf(stderr, "read_options: malloc failed.\n");
2978           return(2);
2979         }
2980         memset(new, 0, sizeof(listen_socket_t));
2981
2982         strncpy(new->addr, optarg, sizeof(new->addr)-1);
2983
2984         /* Add permissions to the socket {{{ */
2985         if (permissions_len != 0)
2986         {
2987           size_t i;
2988           for (i = 0; i < permissions_len; i++)
2989           {
2990             status = socket_permission_add (new, permissions[i]);
2991             if (status != 0)
2992             {
2993               fprintf (stderr, "read_options: Adding permission \"%s\" to "
2994                   "socket failed. Most likely, this permission doesn't "
2995                   "exist. Check your command line.\n", permissions[i]);
2996               status = 4;
2997             }
2998           }
2999         }
3000         else /* if (permissions_len == 0) */
3001         {
3002           /* Add permission for ALL commands to the socket. */
3003           size_t i;
3004           for (i = 0; i < list_of_commands_len; i++)
3005           {
3006             status = socket_permission_add (new, list_of_commands[i].cmd);
3007             if (status != 0)
3008             {
3009               fprintf (stderr, "read_options: Adding permission \"%s\" to "
3010                   "socket failed. This should never happen, ever! Sorry.\n",
3011                   permissions[i]);
3012               status = 4;
3013             }
3014           }
3015         }
3016         /* }}} Done adding permissions. */
3017
3018         new->socket_group = socket_group;
3019         new->socket_permissions = socket_permissions;
3020
3021         if (!rrd_add_ptr((void ***)&config_listen_address_list,
3022                          &config_listen_address_list_len, new))
3023         {
3024           fprintf(stderr, "read_options: rrd_add_ptr failed.\n");
3025           return (2);
3026         }
3027       }
3028       break;
3029
3030       /* set socket group permissions */
3031       case 's':
3032       {
3033         gid_t group_gid;
3034         struct group *grp;
3035
3036         group_gid = strtoul(optarg, NULL, 10);
3037         if (errno != EINVAL && group_gid>0)
3038         {
3039           /* we were passed a number */
3040           grp = getgrgid(group_gid);
3041         }
3042         else
3043         {
3044           grp = getgrnam(optarg);
3045         }
3046
3047         if (grp)
3048         {
3049           socket_group = grp->gr_gid;
3050         }
3051         else
3052         {
3053           /* no idea what the user wanted... */
3054           fprintf (stderr, "read_options: couldn't map \"%s\" to a group, Sorry\n", optarg);
3055           return (5);
3056         }
3057       }
3058       break;
3059
3060       /* set socket file permissions */
3061       case 'm':
3062       {
3063         long  tmp;
3064         char *endptr = NULL;
3065
3066         tmp = strtol (optarg, &endptr, 8);
3067         if ((endptr == optarg) || (! endptr) || (*endptr != '\0')
3068             || (tmp > 07777) || (tmp < 0)) {
3069           fprintf (stderr, "read_options: Invalid file mode \"%s\".\n",
3070               optarg);
3071           return (5);
3072         }
3073
3074         socket_permissions = (mode_t)tmp;
3075       }
3076       break;
3077
3078       case 'P':
3079       {
3080         char *optcopy;
3081         char *saveptr;
3082         char *dummy;
3083         char *ptr;
3084
3085         rrd_free_ptrs ((void *) &permissions, &permissions_len);
3086
3087         optcopy = strdup (optarg);
3088         dummy = optcopy;
3089         saveptr = NULL;
3090         while ((ptr = strtok_r (dummy, ", ", &saveptr)) != NULL)
3091         {
3092           dummy = NULL;
3093           rrd_add_strdup ((void *) &permissions, &permissions_len, ptr);
3094         }
3095
3096         free (optcopy);
3097       }
3098       break;
3099
3100       case 'f':
3101       {
3102         int temp;
3103
3104         temp = atoi (optarg);
3105         if (temp > 0)
3106           config_flush_interval = temp;
3107         else
3108         {
3109           fprintf (stderr, "Invalid flush interval: %s\n", optarg);
3110           status = 3;
3111         }
3112       }
3113       break;
3114
3115       case 'w':
3116       {
3117         int temp;
3118
3119         temp = atoi (optarg);
3120         if (temp > 0)
3121           config_write_interval = temp;
3122         else
3123         {
3124           fprintf (stderr, "Invalid write interval: %s\n", optarg);
3125           status = 2;
3126         }
3127       }
3128       break;
3129
3130       case 'z':
3131       {
3132         int temp;
3133
3134         temp = atoi(optarg);
3135         if (temp > 0)
3136           config_write_jitter = temp;
3137         else
3138         {
3139           fprintf (stderr, "Invalid write jitter: -z %s\n", optarg);
3140           status = 2;
3141         }
3142
3143         break;
3144       }
3145
3146       case 't':
3147       {
3148         int threads;
3149         threads = atoi(optarg);
3150         if (threads >= 1)
3151           config_queue_threads = threads;
3152         else
3153         {
3154           fprintf (stderr, "Invalid thread count: -t %s\n", optarg);
3155           return 1;
3156         }
3157       }
3158       break;
3159
3160       case 'B':
3161         config_write_base_only = 1;
3162         break;
3163
3164       case 'b':
3165       {
3166         size_t len;
3167         char base_realpath[PATH_MAX];
3168
3169         if (config_base_dir != NULL)
3170           free (config_base_dir);
3171         config_base_dir = strdup (optarg);
3172         if (config_base_dir == NULL)
3173         {
3174           fprintf (stderr, "read_options: strdup failed.\n");
3175           return (3);
3176         }
3177
3178         if (rrd_mkdir_p (config_base_dir, 0777) != 0)
3179         {
3180           fprintf (stderr, "Failed to create base directory '%s': %s\n",
3181               config_base_dir, rrd_strerror (errno));
3182           return (3);
3183         }
3184
3185         /* make sure that the base directory is not resolved via
3186          * symbolic links.  this makes some performance-enhancing
3187          * assumptions possible (we don't have to resolve paths
3188          * that start with a "/")
3189          */
3190         if (realpath(config_base_dir, base_realpath) == NULL)
3191         {
3192           fprintf (stderr, "Failed to canonicalize the base directory '%s': "
3193               "%s\n", config_base_dir, rrd_strerror(errno));
3194           return 5;
3195         }
3196
3197         len = strlen (config_base_dir);
3198         while ((len > 0) && (config_base_dir[len - 1] == '/'))
3199         {
3200           config_base_dir[len - 1] = 0;
3201           len--;
3202         }
3203
3204         if (len < 1)
3205         {
3206           fprintf (stderr, "Invalid base directory: %s\n", optarg);
3207           return (4);
3208         }
3209
3210         _config_base_dir_len = len;
3211
3212         len = strlen (base_realpath);
3213         while ((len > 0) && (base_realpath[len - 1] == '/'))
3214         {
3215           base_realpath[len - 1] = '\0';
3216           len--;
3217         }
3218
3219         if (strncmp(config_base_dir,
3220                          base_realpath, sizeof(base_realpath)) != 0)
3221         {
3222           fprintf(stderr,
3223                   "Base directory (-b) resolved via file system links!\n"
3224                   "Please consult rrdcached '-b' documentation!\n"
3225                   "Consider specifying the real directory (%s)\n",
3226                   base_realpath);
3227           return 5;
3228         }
3229       }
3230       break;
3231
3232       case 'p':
3233       {
3234         if (config_pid_file != NULL)
3235           free (config_pid_file);
3236         config_pid_file = strdup (optarg);
3237         if (config_pid_file == NULL)
3238         {
3239           fprintf (stderr, "read_options: strdup failed.\n");
3240           return (3);
3241         }
3242       }
3243       break;
3244
3245       case 'F':
3246         config_flush_at_shutdown = 1;
3247         break;
3248
3249       case 'j':
3250       {
3251         char journal_dir_actual[PATH_MAX];
3252         const char *dir;
3253         dir = journal_dir = strdup(realpath((const char *)optarg, journal_dir_actual));
3254
3255         status = rrd_mkdir_p(dir, 0777);
3256         if (status != 0)
3257         {
3258           fprintf(stderr, "Failed to create journal directory '%s': %s\n",
3259               dir, rrd_strerror(errno));
3260           return 6;
3261         }
3262
3263         if (access(dir, R_OK|W_OK|X_OK) != 0)
3264         {
3265           fprintf(stderr, "Must specify a writable directory with -j! (%s)\n",
3266                   errno ? rrd_strerror(errno) : "");
3267           return 6;
3268         }
3269       }
3270       break;
3271
3272       case 'a':
3273       {
3274         int temp = atoi(optarg);
3275         if (temp > 0)
3276           config_alloc_chunk = temp;
3277         else
3278         {
3279           fprintf(stderr, "Invalid allocation size: %s\n", optarg);
3280           return 10;
3281         }
3282       }
3283       break;
3284
3285       case 'h':
3286       case '?':
3287         printf ("RRDCacheD %s\n"
3288             "Copyright (C) 2008,2009 Florian octo Forster and Kevin Brintnall\n"
3289             "\n"
3290             "Usage: rrdcached [options]\n"
3291             "\n"
3292             "Valid options are:\n"
3293             "  -l <address>  Socket address to listen to.\n"
3294             "  -P <perms>    Sets the permissions to assign to all following "
3295                             "sockets\n"
3296             "  -w <seconds>  Interval in which to write data.\n"
3297             "  -z <delay>    Delay writes up to <delay> seconds to spread load\n"
3298             "  -t <threads>  Number of write threads.\n"
3299             "  -f <seconds>  Interval in which to flush dead data.\n"
3300             "  -p <file>     Location of the PID-file.\n"
3301             "  -b <dir>      Base directory to change to.\n"
3302             "  -B            Restrict file access to paths within -b <dir>\n"
3303             "  -g            Do not fork and run in the foreground.\n"
3304             "  -j <dir>      Directory in which to create the journal files.\n"
3305             "  -F            Always flush all updates at shutdown\n"
3306             "  -s <id|name>  Group owner of all following UNIX sockets\n"
3307             "                (the socket will also have read/write permissions "
3308                             "for that group)\n"
3309             "  -m <mode>     File permissions (octal) of all following UNIX "
3310                             "sockets\n"
3311             "  -a <size>     Memory allocation chunk size. Default is 1."
3312             "\n"
3313             "For more information and a detailed description of all options "
3314             "please refer\n"
3315             "to the rrdcached(1) manual page.\n",
3316             VERSION);
3317         if (option == 'h')
3318           status = -1;
3319         else
3320           status = 1;
3321         break;
3322     } /* switch (option) */
3323   } /* while (getopt) */
3324
3325   /* advise the user when values are not sane */
3326   if (config_flush_interval < 2 * config_write_interval)
3327     fprintf(stderr, "WARNING: flush interval (-f) should be at least"
3328             " 2x write interval (-w) !\n");
3329   if (config_write_jitter > config_write_interval)
3330     fprintf(stderr, "WARNING: write delay (-z) should NOT be larger than"
3331             " write interval (-w) !\n");
3332
3333   if (config_write_base_only && config_base_dir == NULL)
3334     fprintf(stderr, "WARNING: -B does not make sense without -b!\n"
3335             "  Consult the rrdcached documentation\n");
3336
3337   if (journal_dir == NULL)
3338     config_flush_at_shutdown = 1;
3339
3340   rrd_free_ptrs ((void *) &permissions, &permissions_len);
3341
3342   return (status);
3343 } /* }}} int read_options */
3344
3345 int main (int argc, char **argv)
3346 {
3347   int status;
3348
3349   status = read_options (argc, argv);
3350   if (status != 0)
3351   {
3352     if (status < 0)
3353       status = 0;
3354     return (status);
3355   }
3356
3357   status = daemonize ();
3358   if (status != 0)
3359   {
3360     fprintf (stderr, "rrdcached: daemonize failed, exiting.\n");
3361     return (1);
3362   }
3363
3364   journal_init();
3365
3366   /* start the queue threads */
3367   queue_threads = calloc(config_queue_threads, sizeof(*queue_threads));
3368   if (queue_threads == NULL)
3369   {
3370     RRDD_LOG (LOG_ERR, "FATAL: cannot calloc queue threads");
3371     cleanup();
3372     return (1);
3373   }
3374   for (int i = 0; i < config_queue_threads; i++)
3375   {
3376     memset (&queue_threads[i], 0, sizeof (*queue_threads));
3377     status = pthread_create (&queue_threads[i], NULL, queue_thread_main, NULL);
3378     if (status != 0)
3379     {
3380       RRDD_LOG (LOG_ERR, "FATAL: cannot create queue thread");
3381       cleanup();
3382       return (1);
3383     }
3384   }
3385
3386   /* start the flush thread */
3387   memset(&flush_thread, 0, sizeof(flush_thread));
3388   status = pthread_create (&flush_thread, NULL, flush_thread_main, NULL);
3389   if (status != 0)
3390   {
3391     RRDD_LOG (LOG_ERR, "FATAL: cannot create flush thread");
3392     cleanup();
3393     return (1);
3394   }
3395
3396   listen_thread_main (NULL);
3397   cleanup ();
3398
3399   return (0);
3400 } /* int main */
3401
3402 /*
3403  * vim: set sw=2 sts=2 ts=8 et fdm=marker :
3404  */