check permissions before writing to journal -- kevin
[rrdtool.git] / src / rrd_daemon.c
index c438028..5163f34 100644 (file)
 /*
  * Types
  */
+typedef enum
+{
+  PRIV_LOW,
+  PRIV_HIGH
+} socket_privilege;
+
 struct listen_socket_s
 {
   int fd;
-  char path[PATH_MAX + 1];
+  char addr[PATH_MAX + 1];
+  int family;
+  socket_privilege privilege;
 };
 typedef struct listen_socket_s listen_socket_t;
 
@@ -119,7 +127,7 @@ struct cache_item_s
 #define CI_FLAGS_IN_TREE  (1<<0)
 #define CI_FLAGS_IN_QUEUE (1<<1)
   int flags;
-
+  pthread_cond_t  flushed;
   cache_item_t *next;
 };
 
@@ -165,15 +173,14 @@ static cache_item_t   *cache_queue_tail = NULL;
 static pthread_mutex_t cache_lock = PTHREAD_MUTEX_INITIALIZER;
 static pthread_cond_t  cache_cond = PTHREAD_COND_INITIALIZER;
 
-static pthread_cond_t  flush_cond = PTHREAD_COND_INITIALIZER;
-
 static int config_write_interval = 300;
 static int config_write_jitter   = 0;
 static int config_flush_interval = 3600;
+static int config_flush_at_shutdown = 0;
 static char *config_pid_file = NULL;
 static char *config_base_dir = NULL;
 
-static char **config_listen_address_list = NULL;
+static listen_socket_t **config_listen_address_list = NULL;
 static int config_listen_address_list_len = 0;
 
 static uint64_t stats_queue_length = 0;
@@ -197,45 +204,96 @@ static void journal_rotate(void);
 /* 
  * Functions
  */
-static void sig_int_handler (int s __attribute__((unused))) /* {{{ */
+static void sig_common (const char *sig) /* {{{ */
 {
-  RRDD_LOG(LOG_NOTICE, "caught SIGINT");
+  RRDD_LOG(LOG_NOTICE, "caught SIG%s", sig);
   do_shutdown++;
   pthread_cond_broadcast(&cache_cond);
+} /* }}} void sig_common */
+
+static void sig_int_handler (int s __attribute__((unused))) /* {{{ */
+{
+  sig_common("INT");
 } /* }}} void sig_int_handler */
 
 static void sig_term_handler (int s __attribute__((unused))) /* {{{ */
 {
-  RRDD_LOG(LOG_NOTICE, "caught SIGTERM");
-  do_shutdown++;
-  pthread_cond_broadcast(&cache_cond);
+  sig_common("TERM");
 } /* }}} void sig_term_handler */
 
-static int write_pidfile (void) /* {{{ */
+static void sig_usr1_handler (int s __attribute__((unused))) /* {{{ */
+{
+  config_flush_at_shutdown = 1;
+  sig_common("USR1");
+} /* }}} void sig_usr1_handler */
+
+static void sig_usr2_handler (int s __attribute__((unused))) /* {{{ */
+{
+  config_flush_at_shutdown = 0;
+  sig_common("USR2");
+} /* }}} void sig_usr2_handler */
+
+static void install_signal_handlers(void) /* {{{ */
+{
+  /* These structures are static, because `sigaction' behaves weird if the are
+   * overwritten.. */
+  static struct sigaction sa_int;
+  static struct sigaction sa_term;
+  static struct sigaction sa_pipe;
+  static struct sigaction sa_usr1;
+  static struct sigaction sa_usr2;
+
+  /* Install signal handlers */
+  memset (&sa_int, 0, sizeof (sa_int));
+  sa_int.sa_handler = sig_int_handler;
+  sigaction (SIGINT, &sa_int, NULL);
+
+  memset (&sa_term, 0, sizeof (sa_term));
+  sa_term.sa_handler = sig_term_handler;
+  sigaction (SIGTERM, &sa_term, NULL);
+
+  memset (&sa_pipe, 0, sizeof (sa_pipe));
+  sa_pipe.sa_handler = SIG_IGN;
+  sigaction (SIGPIPE, &sa_pipe, NULL);
+
+  memset (&sa_pipe, 0, sizeof (sa_usr1));
+  sa_usr1.sa_handler = sig_usr1_handler;
+  sigaction (SIGUSR1, &sa_usr1, NULL);
+
+  memset (&sa_usr2, 0, sizeof (sa_usr2));
+  sa_usr2.sa_handler = sig_usr2_handler;
+  sigaction (SIGUSR2, &sa_usr2, NULL);
+
+} /* }}} void install_signal_handlers */
+
+static int open_pidfile(void) /* {{{ */
 {
-  pid_t pid;
-  char *file;
   int fd;
-  FILE *fh;
+  char *file;
 
-  pid = getpid ();
-  
   file = (config_pid_file != NULL)
     ? config_pid_file
     : LOCALSTATEDIR "/run/rrdcached.pid";
 
   fd = open(file, O_CREAT|O_EXCL|O_WRONLY, S_IRUSR|S_IRGRP|S_IROTH);
   if (fd < 0)
-  {
-    RRDD_LOG(LOG_ERR, "FATAL: cannot create '%s' (%s)",
-             file, rrd_strerror(errno));
-    return (-1);
-  }
+    fprintf(stderr, "FATAL: cannot create '%s' (%s)\n",
+            file, rrd_strerror(errno));
+
+  return(fd);
+} /* }}} static int open_pidfile */
+
+static int write_pidfile (int fd) /* {{{ */
+{
+  pid_t pid;
+  FILE *fh;
+
+  pid = getpid ();
 
   fh = fdopen (fd, "w");
   if (fh == NULL)
   {
-    RRDD_LOG (LOG_ERR, "write_pidfile: Opening `%s' failed.", file);
+    RRDD_LOG (LOG_ERR, "write_pidfile: fdopen() failed.");
     close(fd);
     return (-1);
   }
@@ -427,6 +485,7 @@ static int enqueue_cache_item (cache_item_t *ci, /* {{{ */
 
   if (did_insert)
   {
+    pthread_cond_broadcast(&cache_cond);
     pthread_mutex_lock (&stats_lock);
     stats_queue_length++;
     pthread_mutex_unlock (&stats_lock);
@@ -499,7 +558,7 @@ static int flush_old_values (int max_age)
   if (max_age > 0)
     cfd.abs_timeout = cfd.now - max_age;
   else
-    cfd.abs_timeout = cfd.now + 1;
+    cfd.abs_timeout = cfd.now + 2*config_write_jitter + 1;
 
   /* `tree_callback_flush' will return the keys of all values that haven't
    * been touched in the last `config_flush_interval' seconds in `cfd'.
@@ -543,6 +602,7 @@ static void *queue_thread_main (void *args __attribute__((unused))) /* {{{ */
 {
   struct timeval now;
   struct timespec next_flush;
+  int final_flush = 0; /* make sure we only flush once on shutdown */
 
   gettimeofday (&now, NULL);
   next_flush.tv_sec = now.tv_sec + config_flush_interval;
@@ -580,8 +640,9 @@ static void *queue_thread_main (void *args __attribute__((unused))) /* {{{ */
     }
 
     /* Now, check if there's something to store away. If not, wait until
-     * something comes in or it's time to do the cache flush. */
-    if (cache_queue_head == NULL)
+     * something comes in or it's time to do the cache flush.  if we are
+     * shutting down, do not wait around.  */
+    if (cache_queue_head == NULL && !do_shutdown)
     {
       status = pthread_cond_timedwait (&cache_cond, &cache_lock, &next_flush);
       if ((status != 0) && (status != ETIMEDOUT))
@@ -591,9 +652,14 @@ static void *queue_thread_main (void *args __attribute__((unused))) /* {{{ */
       }
     }
 
-    /* We're about to shut down, so lets flush the entire tree. */
-    if ((do_shutdown != 0) && (cache_queue_head == NULL))
-      flush_old_values (/* max age = */ -1);
+    /* We're about to shut down */
+    if (do_shutdown != 0 && !final_flush++)
+    {
+      if (config_flush_at_shutdown)
+        flush_old_values (-1); /* flush everything */
+      else
+        break;
+    }
 
     /* Check if a value has arrived. This may be NULL if we timed out or there
      * was an interrupt such as a signal. */
@@ -640,6 +706,7 @@ static void *queue_thread_main (void *args __attribute__((unused))) /* {{{ */
     }
 
     journal_write("wrote", file);
+    pthread_cond_broadcast(&ci->flushed);
 
     for (i = 0; i < values_num; i++)
       free (values[i]);
@@ -656,16 +723,24 @@ static void *queue_thread_main (void *args __attribute__((unused))) /* {{{ */
     }
 
     pthread_mutex_lock (&cache_lock);
-    pthread_cond_broadcast (&flush_cond);
 
-    /* We're about to shut down, so lets flush the entire tree. */
-    if ((do_shutdown != 0) && (cache_queue_head == NULL))
-      flush_old_values (/* max age = */ -1);
+    /* We're about to shut down */
+    if (do_shutdown != 0 && !final_flush++)
+    {
+      if (config_flush_at_shutdown)
+          flush_old_values (-1); /* flush everything */
+      else
+        break;
+    }
   } /* while ((do_shutdown == 0) || (cache_queue_head != NULL)) */
   pthread_mutex_unlock (&cache_lock);
 
-  assert(cache_queue_head == NULL);
-  RRDD_LOG(LOG_INFO, "clean shutdown; all RRDs flushed");
+  if (config_flush_at_shutdown)
+  {
+    assert(cache_queue_head == NULL);
+    RRDD_LOG(LOG_INFO, "clean shutdown; all RRDs flushed");
+  }
+
   journal_done();
 
   return (NULL);
@@ -747,27 +822,15 @@ static int flush_file (const char *filename) /* {{{ */
     return (ENOENT);
   }
 
-  /* Enqueue at head */
-  enqueue_cache_item (ci, HEAD);
-  pthread_cond_signal (&cache_cond);
-
-  while ((ci->flags & CI_FLAGS_IN_QUEUE) != 0)
+  if (ci->values_num > 0)
   {
-    ci = NULL;
-
-    pthread_cond_wait (&flush_cond, &cache_lock);
-
-    ci = g_tree_lookup (cache_tree, filename);
-    if (ci == NULL)
-    {
-      RRDD_LOG (LOG_ERR, "flush_file: Tree node went away "
-          "while waiting for flush.");
-      pthread_mutex_unlock (&cache_lock);
-      return (-1);
-    }
+    /* Enqueue at head */
+    enqueue_cache_item (ci, HEAD);
+    pthread_cond_wait(&ci->flushed, &cache_lock);
   }
 
-  pthread_mutex_unlock (&cache_lock);
+  pthread_mutex_unlock(&cache_lock);
+
   return (0);
 } /* }}} int flush_file */
 
@@ -782,8 +845,9 @@ static int handle_request_help (int fd, /* {{{ */
 
   char *help_help[] =
   {
-    "4 Command overview\n",
+    "5 Command overview\n",
     "FLUSH <filename>\n",
+    "FLUSHALL\n",
     "HELP [<command>]\n",
     "UPDATE <filename> <values> [<values> ...]\n",
     "STATS\n"
@@ -800,6 +864,15 @@ static int handle_request_help (int fd, /* {{{ */
   };
   size_t help_flush_len = sizeof (help_flush) / sizeof (help_flush[0]);
 
+  char *help_flushall[] =
+  {
+    "3 Help for FLUSHALL\n",
+    "Usage: FLUSHALL\n",
+    "\n",
+    "Triggers writing of all pending updates.  Returns immediately.\n"
+  };
+  size_t help_flushall_len = sizeof(help_flushall) / sizeof(help_flushall[0]);
+
   char *help_update[] =
   {
     "9 Help for UPDATE\n",
@@ -843,6 +916,11 @@ static int handle_request_help (int fd, /* {{{ */
       help_text = help_flush;
       help_text_len = help_flush_len;
     }
+    else if (strcasecmp (command, "flushall") == 0)
+    {
+      help_text = help_flushall;
+      help_text_len = help_flushall_len;
+    }
     else if (strcasecmp (command, "stats") == 0)
     {
       help_text = help_stats;
@@ -1005,6 +1083,27 @@ static int handle_request_flush (int fd, /* {{{ */
   return (0);
 } /* }}} int handle_request_flush */
 
+static int handle_request_flushall(int fd) /* {{{ */
+{
+  int status;
+  char answer[] ="0 Started flush.\n";
+
+  RRDD_LOG(LOG_DEBUG, "Received FLUSHALL");
+
+  pthread_mutex_lock(&cache_lock);
+  flush_old_values(-1);
+  pthread_mutex_unlock(&cache_lock);
+
+  status = swrite(fd, answer, strlen(answer));
+  if (status < 0)
+  {
+    status = errno;
+    RRDD_LOG(LOG_INFO, "handle_request_flushall: swrite returned an error.");
+  }
+
+  return (status);
+} /* }}} static int handle_request_flushall */
+
 static int handle_request_update (int fd, /* {{{ */
     char *buffer, size_t buffer_size)
 {
@@ -1043,17 +1142,19 @@ static int handle_request_update (int fd, /* {{{ */
   pthread_mutex_unlock(&stats_lock);
 
   pthread_mutex_lock (&cache_lock);
-
   ci = g_tree_lookup (cache_tree, file);
+
   if (ci == NULL) /* {{{ */
   {
     struct stat statbuf;
 
+    /* don't hold the lock while we setup; stat(2) might block */
+    pthread_mutex_unlock(&cache_lock);
+
     memset (&statbuf, 0, sizeof (statbuf));
     status = stat (file, &statbuf);
     if (status != 0)
     {
-      pthread_mutex_unlock (&cache_lock);
       RRDD_LOG (LOG_NOTICE, "handle_request_update: stat (%s) failed.", file);
 
       status = errno;
@@ -1067,16 +1168,12 @@ static int handle_request_update (int fd, /* {{{ */
     }
     if (!S_ISREG (statbuf.st_mode))
     {
-      pthread_mutex_unlock (&cache_lock);
-
       snprintf (answer, sizeof (answer), "-1 Not a regular file: %s\n", file);
       RRDD_UPDATE_SEND;
       return (0);
     }
     if (access(file, R_OK|W_OK) != 0)
     {
-      pthread_mutex_unlock (&cache_lock);
-
       snprintf (answer, sizeof (answer), "-1 Cannot read/write %s: %s\n",
                 file, rrd_strerror(errno));
       RRDD_UPDATE_SEND;
@@ -1086,7 +1183,6 @@ static int handle_request_update (int fd, /* {{{ */
     ci = (cache_item_t *) malloc (sizeof (cache_item_t));
     if (ci == NULL)
     {
-      pthread_mutex_unlock (&cache_lock);
       RRDD_LOG (LOG_ERR, "handle_request_update: malloc failed.");
 
       strncpy (answer, "-1 malloc failed.\n", sizeof (answer));
@@ -1098,7 +1194,6 @@ static int handle_request_update (int fd, /* {{{ */
     ci->file = strdup (file);
     if (ci->file == NULL)
     {
-      pthread_mutex_unlock (&cache_lock);
       free (ci);
       RRDD_LOG (LOG_ERR, "handle_request_update: strdup failed.");
 
@@ -1110,6 +1205,7 @@ static int handle_request_update (int fd, /* {{{ */
     _wipe_ci_values(ci, now);
     ci->flags = CI_FLAGS_IN_TREE;
 
+    pthread_mutex_lock(&cache_lock);
     g_tree_insert (cache_tree, (void *) ci->file, (void *) ci);
   } /* }}} */
   assert (ci != NULL);
@@ -1151,7 +1247,6 @@ static int handle_request_update (int fd, /* {{{ */
       && (ci->values_num > 0))
   {
     enqueue_cache_item (ci, TAIL);
-    pthread_cond_signal (&cache_cond);
   }
 
   pthread_mutex_unlock (&cache_lock);
@@ -1204,8 +1299,28 @@ static int handle_request_wrote (int fd __attribute__((unused)), /* {{{ */
   return (0);
 } /* }}} int handle_request_wrote */
 
+/* returns 1 if we have the required privilege level */
+static int has_privilege (socket_privilege priv, /* {{{ */
+                          socket_privilege required, int fd)
+{
+  int status;
+  char error[CMD_MAX];
+
+  if (priv >= required)
+    return 1;
+
+  sprintf(error, "-1 %s\n", rrd_strerror(EACCES));
+  status = swrite(fd, error, strlen(error));
+
+  if (status < 0)
+    return status;
+  else
+    return 0;
+} /* }}} static int has_privilege */
+
 /* if fd < 0, we are in journal replay mode */
-static int handle_request (int fd, char *buffer, size_t buffer_size) /* {{{ */
+static int handle_request (int fd, socket_privilege privilege, /* {{{ */
+                           char *buffer, size_t buffer_size)
 {
   char *buffer_ptr;
   char *command;
@@ -1224,6 +1339,10 @@ static int handle_request (int fd, char *buffer, size_t buffer_size) /* {{{ */
 
   if (strcasecmp (command, "update") == 0)
   {
+    status = has_privilege(privilege, PRIV_HIGH, fd);
+    if (status <= 0)
+      return status;
+
     /* don't re-write updates in replay mode */
     if (fd >= 0)
       journal_write(command, buffer_ptr);
@@ -1239,6 +1358,14 @@ static int handle_request (int fd, char *buffer, size_t buffer_size) /* {{{ */
   {
     return (handle_request_flush (fd, buffer_ptr, buffer_size));
   }
+  else if (strcasecmp (command, "flushall") == 0)
+  {
+    status = has_privilege(privilege, PRIV_HIGH, fd);
+    if (status <= 0)
+      return status;
+
+    return (handle_request_flushall(fd));
+  }
   else if (strcasecmp (command, "stats") == 0)
   {
     return (handle_request_stats (fd, buffer_ptr, buffer_size));
@@ -1294,10 +1421,16 @@ static void journal_rotate(void) /* {{{ */
     fclose(old_fh);
 
   if (journal_fh == NULL)
+  {
     RRDD_LOG(LOG_CRIT,
              "JOURNALING DISABLED: Cannot open journal file '%s' : (%s)",
              journal_cur, rrd_strerror(errno));
 
+    RRDD_LOG(LOG_ERR,
+             "JOURNALING DISABLED: All values will be flushed at shutdown");
+    config_flush_at_shutdown = 1;
+  }
+
 } /* }}} static void journal_rotate */
 
 static void journal_done(void) /* {{{ */
@@ -1312,10 +1445,18 @@ static void journal_done(void) /* {{{ */
     journal_fh = NULL;
   }
 
-  RRDD_LOG(LOG_INFO, "removing journals");
+  if (config_flush_at_shutdown)
+  {
+    RRDD_LOG(LOG_INFO, "removing journals");
+    unlink(journal_old);
+    unlink(journal_cur);
+  }
+  else
+  {
+    RRDD_LOG(LOG_INFO, "expedited shutdown; "
+             "journals will be used at next startup");
+  }
 
-  unlink(journal_old);
-  unlink(journal_cur);
   pthread_mutex_unlock(&journal_lock);
 
 } /* }}} static void journal_done */
@@ -1367,7 +1508,8 @@ static int journal_replay (const char *file) /* {{{ */
     size_t entry_len;
 
     ++line;
-    fgets(entry, sizeof(entry), fh);
+    if (fgets(entry, sizeof(entry), fh) == NULL)
+      break;
     entry_len = strlen(entry);
 
     /* check \n termination in case journal writing crashed mid-line */
@@ -1382,7 +1524,7 @@ static int journal_replay (const char *file) /* {{{ */
 
     entry[entry_len - 1] = '\0';
 
-    if (handle_request(-1, entry, entry_len) == 0)
+    if (handle_request(-1, PRIV_HIGH, entry, entry_len) == 0)
       ++entry_cnt;
     else
       ++fail_cnt;
@@ -1404,11 +1546,12 @@ static int journal_replay (const char *file) /* {{{ */
 static void *connection_thread_main (void *args) /* {{{ */
 {
   pthread_t self;
+  listen_socket_t *sock;
   int i;
   int fd;
-  
-  fd = *((int *) args);
-  free (args);
+
+  sock = (listen_socket_t *) args;
+  fd = sock->fd;
 
   pthread_mutex_lock (&connection_threads_lock);
   {
@@ -1441,7 +1584,9 @@ static void *connection_thread_main (void *args) /* {{{ */
     pollfd.revents = 0;
 
     status = poll (&pollfd, 1, /* timeout = */ 500);
-    if (status == 0) /* timeout */
+    if (do_shutdown)
+      break;
+    else if (status == 0) /* timeout */
       continue;
     else if (status < 0) /* error */
     {
@@ -1477,12 +1622,13 @@ static void *connection_thread_main (void *args) /* {{{ */
       break;
     }
 
-    status = handle_request (fd, buffer, /*buffer_size=*/ status);
+    status = handle_request (fd, sock->privilege, buffer, status);
     if (status != 0)
       break;
   }
 
   close(fd);
+  free(args);
 
   self = pthread_self ();
   /* Remove this thread from the connection threads list */
@@ -1507,12 +1653,17 @@ static void *connection_thread_main (void *args) /* {{{ */
   return (NULL);
 } /* }}} void *connection_thread_main */
 
-static int open_listen_socket_unix (const char *path) /* {{{ */
+static int open_listen_socket_unix (const listen_socket_t *sock) /* {{{ */
 {
   int fd;
   struct sockaddr_un sa;
   listen_socket_t *temp;
   int status;
+  const char *path;
+
+  path = sock->addr;
+  if (strncmp(path, "unix:", strlen("unix:")) == 0)
+    path += strlen("unix:");
 
   temp = (listen_socket_t *) realloc (listen_fds,
       sizeof (listen_fds[0]) * (listen_fds_num + 1));
@@ -1522,7 +1673,7 @@ static int open_listen_socket_unix (const char *path) /* {{{ */
     return (-1);
   }
   listen_fds = temp;
-  memset (listen_fds + listen_fds_num, 0, sizeof (listen_fds[0]));
+  memcpy (listen_fds + listen_fds_num, sock, sizeof (listen_fds[0]));
 
   fd = socket (PF_UNIX, SOCK_STREAM, /* protocol = */ 0);
   if (fd < 0)
@@ -1552,17 +1703,17 @@ static int open_listen_socket_unix (const char *path) /* {{{ */
     unlink (path);
     return (-1);
   }
-  
+
   listen_fds[listen_fds_num].fd = fd;
-  snprintf (listen_fds[listen_fds_num].path,
-      sizeof (listen_fds[listen_fds_num].path) - 1,
-      "unix:%s", path);
+  listen_fds[listen_fds_num].family = PF_UNIX;
+  strncpy(listen_fds[listen_fds_num].addr, path,
+          sizeof (listen_fds[listen_fds_num].addr) - 1);
   listen_fds_num++;
 
   return (0);
 } /* }}} int open_listen_socket_unix */
 
-static int open_listen_socket (const char *addr_orig) /* {{{ */
+static int open_listen_socket_network(const listen_socket_t *sock) /* {{{ */
 {
   struct addrinfo ai_hints;
   struct addrinfo *ai_res;
@@ -1572,17 +1723,10 @@ static int open_listen_socket (const char *addr_orig) /* {{{ */
   char *port;
   int status;
 
-  assert (addr_orig != NULL);
-
-  strncpy (addr_copy, addr_orig, sizeof (addr_copy));
+  strncpy (addr_copy, sock->addr, sizeof (addr_copy));
   addr_copy[sizeof (addr_copy) - 1] = 0;
   addr = addr_copy;
 
-  if (strncmp ("unix:", addr, strlen ("unix:")) == 0)
-    return (open_listen_socket_unix (addr + strlen ("unix:")));
-  else if (addr[0] == '/')
-    return (open_listen_socket_unix (addr));
-
   memset (&ai_hints, 0, sizeof (ai_hints));
   ai_hints.ai_flags = 0;
 #ifdef AI_ADDRCONFIG
@@ -1592,7 +1736,7 @@ static int open_listen_socket (const char *addr_orig) /* {{{ */
   ai_hints.ai_socktype = SOCK_STREAM;
 
   port = NULL;
- if (*addr == '[') /* IPv6+port format */
 if (*addr == '[') /* IPv6+port format */
   {
     /* `addr' is something like "[2001:780:104:2:211:24ff:feab:26f8]:12345" */
     addr++;
@@ -1600,8 +1744,8 @@ static int open_listen_socket (const char *addr_orig) /* {{{ */
     port = strchr (addr, ']');
     if (port == NULL)
     {
-      RRDD_LOG (LOG_ERR, "open_listen_socket: Malformed address: %s",
-          addr_orig);
+      RRDD_LOG (LOG_ERR, "open_listen_socket_network: Malformed address: %s",
+          sock->addr);
       return (-1);
     }
     *port = 0;
@@ -1613,7 +1757,7 @@ static int open_listen_socket (const char *addr_orig) /* {{{ */
       port = NULL;
     else
     {
-      RRDD_LOG (LOG_ERR, "open_listen_socket: Garbage after address: %s",
+      RRDD_LOG (LOG_ERR, "open_listen_socket_network: Garbage after address: %s",
           port);
       return (-1);
     }
@@ -1633,7 +1777,7 @@ static int open_listen_socket (const char *addr_orig) /* {{{ */
                         &ai_hints, &ai_res);
   if (status != 0)
   {
-    RRDD_LOG (LOG_ERR, "open_listen_socket: getaddrinfo(%s) failed: "
+    RRDD_LOG (LOG_ERR, "open_listen_socket_network: getaddrinfo(%s) failed: "
         "%s", addr, gai_strerror (status));
     return (-1);
   }
@@ -1648,16 +1792,16 @@ static int open_listen_socket (const char *addr_orig) /* {{{ */
         sizeof (listen_fds[0]) * (listen_fds_num + 1));
     if (temp == NULL)
     {
-      RRDD_LOG (LOG_ERR, "open_listen_socket: realloc failed.");
+      RRDD_LOG (LOG_ERR, "open_listen_socket_network: realloc failed.");
       continue;
     }
     listen_fds = temp;
-    memset (listen_fds + listen_fds_num, 0, sizeof (listen_fds[0]));
+    memcpy (listen_fds + listen_fds_num, sock, sizeof (listen_fds[0]));
 
     fd = socket (ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol);
     if (fd < 0)
     {
-      RRDD_LOG (LOG_ERR, "open_listen_socket: socket(2) failed.");
+      RRDD_LOG (LOG_ERR, "open_listen_socket_network: socket(2) failed.");
       continue;
     }
 
@@ -1666,7 +1810,7 @@ static int open_listen_socket (const char *addr_orig) /* {{{ */
     status = bind (fd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
     if (status != 0)
     {
-      RRDD_LOG (LOG_ERR, "open_listen_socket: bind(2) failed.");
+      RRDD_LOG (LOG_ERR, "open_listen_socket_network: bind(2) failed.");
       close (fd);
       continue;
     }
@@ -1674,18 +1818,29 @@ static int open_listen_socket (const char *addr_orig) /* {{{ */
     status = listen (fd, /* backlog = */ 10);
     if (status != 0)
     {
-      RRDD_LOG (LOG_ERR, "open_listen_socket: listen(2) failed.");
+      RRDD_LOG (LOG_ERR, "open_listen_socket_network: listen(2) failed.");
       close (fd);
       return (-1);
     }
 
     listen_fds[listen_fds_num].fd = fd;
-    strncpy (listen_fds[listen_fds_num].path, addr,
-        sizeof (listen_fds[listen_fds_num].path) - 1);
+    listen_fds[listen_fds_num].family = ai_ptr->ai_family;
     listen_fds_num++;
   } /* for (ai_ptr) */
 
   return (0);
+} /* }}} static int open_listen_socket_network */
+
+static int open_listen_socket (const listen_socket_t *sock) /* {{{ */
+{
+  assert(sock != NULL);
+  assert(sock->addr != NULL);
+
+  if (strncmp ("unix:", sock->addr, strlen ("unix:")) == 0
+      || sock->addr[0] == '/')
+    return (open_listen_socket_unix(sock));
+  else
+    return (open_listen_socket_network(sock));
 } /* }}} int open_listen_socket */
 
 static int close_listen_sockets (void) /* {{{ */
@@ -1695,8 +1850,9 @@ static int close_listen_sockets (void) /* {{{ */
   for (i = 0; i < listen_fds_num; i++)
   {
     close (listen_fds[i].fd);
-    if (strncmp ("unix:", listen_fds[i].path, strlen ("unix:")) == 0)
-      unlink (listen_fds[i].path + strlen ("unix:"));
+
+    if (listen_fds[i].family == PF_UNIX)
+      unlink(listen_fds[i].addr);
   }
 
   free (listen_fds);
@@ -1717,7 +1873,12 @@ static void *listen_thread_main (void *args __attribute__((unused))) /* {{{ */
     open_listen_socket (config_listen_address_list[i]);
 
   if (config_listen_address_list_len < 1)
-    open_listen_socket (RRDCACHED_DEFAULT_ADDRESS);
+  {
+    listen_socket_t sock;
+    memset(&sock, 0, sizeof(sock));
+    strncpy(sock.addr, RRDCACHED_DEFAULT_ADDRESS, sizeof(sock.addr));
+    open_listen_socket (&sock);
+  }
 
   if (listen_fds_num < 1)
   {
@@ -1748,11 +1909,11 @@ static void *listen_thread_main (void *args __attribute__((unused))) /* {{{ */
     }
 
     status = poll (pollfds, pollfds_num, /* timeout = */ 1000);
-    if (status == 0)
-    {
-      continue; /* timeout */
-    }
-    else if (status < 0)
+    if (do_shutdown)
+      break;
+    else if (status == 0) /* timeout */
+      continue;
+    else if (status < 0) /* error */
     {
       status = errno;
       if (status != EINTR)
@@ -1764,7 +1925,7 @@ static void *listen_thread_main (void *args __attribute__((unused))) /* {{{ */
 
     for (i = 0; i < pollfds_num; i++)
     {
-      int *client_sd;
+      listen_socket_t *client_sock;
       struct sockaddr_storage client_sa;
       socklen_t client_sa_size;
       pthread_t tid;
@@ -1781,19 +1942,21 @@ static void *listen_thread_main (void *args __attribute__((unused))) /* {{{ */
         continue;
       }
 
-      client_sd = (int *) malloc (sizeof (int));
-      if (client_sd == NULL)
+      client_sock = (listen_socket_t *) malloc (sizeof (listen_socket_t));
+      if (client_sock == NULL)
       {
         RRDD_LOG (LOG_ERR, "listen_thread_main: malloc failed.");
         continue;
       }
+      memcpy(client_sock, &listen_fds[i], sizeof(listen_fds[0]));
 
       client_sa_size = sizeof (client_sa);
-      *client_sd = accept (pollfds[i].fd,
+      client_sock->fd = accept (pollfds[i].fd,
           (struct sockaddr *) &client_sa, &client_sa_size);
-      if (*client_sd < 0)
+      if (client_sock->fd < 0)
       {
         RRDD_LOG (LOG_ERR, "listen_thread_main: accept(2) failed.");
+        free(client_sock);
         continue;
       }
 
@@ -1801,12 +1964,12 @@ static void *listen_thread_main (void *args __attribute__((unused))) /* {{{ */
       pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
 
       status = pthread_create (&tid, &attr, connection_thread_main,
-          /* args = */ (void *) client_sd);
+                               client_sock);
       if (status != 0)
       {
         RRDD_LOG (LOG_ERR, "listen_thread_main: pthread_create failed.");
-        close (*client_sd);
-        free (client_sd);
+        close (client_sock->fd);
+        free (client_sock);
         continue;
       }
     } /* for (pollfds_num) */
@@ -1835,17 +1998,15 @@ static void *listen_thread_main (void *args __attribute__((unused))) /* {{{ */
 static int daemonize (void) /* {{{ */
 {
   int status;
+  int fd;
+  char *base_dir;
 
-  /* These structures are static, because `sigaction' behaves weird if the are
-   * overwritten.. */
-  static struct sigaction sa_int;
-  static struct sigaction sa_term;
-  static struct sigaction sa_pipe;
+  fd = open_pidfile();
+  if (fd < 0) return fd;
 
   if (!stay_foreground)
   {
     pid_t child;
-    char *base_dir;
 
     child = fork ();
     if (child < 0)
@@ -1858,17 +2019,6 @@ static int daemonize (void) /* {{{ */
       return (1);
     }
 
-    /* Change into the /tmp directory. */
-    base_dir = (config_base_dir != NULL)
-      ? config_base_dir
-      : "/tmp";
-    status = chdir (base_dir);
-    if (status != 0)
-    {
-      fprintf (stderr, "daemonize: chdir (%s) failed.\n", base_dir);
-      return (-1);
-    }
-
     /* Become session leader */
     setsid ();
 
@@ -1882,18 +2032,18 @@ static int daemonize (void) /* {{{ */
     dup (0);
   } /* if (!stay_foreground) */
 
-  /* Install signal handlers */
-  memset (&sa_int, 0, sizeof (sa_int));
-  sa_int.sa_handler = sig_int_handler;
-  sigaction (SIGINT, &sa_int, NULL);
-
-  memset (&sa_term, 0, sizeof (sa_term));
-  sa_term.sa_handler = sig_term_handler;
-  sigaction (SIGTERM, &sa_term, NULL);
+  /* Change into the /tmp directory. */
+  base_dir = (config_base_dir != NULL)
+    ? config_base_dir
+    : "/tmp";
+  status = chdir (base_dir);
+  if (status != 0)
+  {
+    fprintf (stderr, "daemonize: chdir (%s) failed.\n", base_dir);
+    return (-1);
+  }
 
-  memset (&sa_pipe, 0, sizeof (sa_pipe));
-  sa_pipe.sa_handler = SIG_IGN;
-  sigaction (SIGPIPE, &sa_pipe, NULL);
+  install_signal_handlers();
 
   openlog ("rrdcached", LOG_PID, LOG_DAEMON);
   RRDD_LOG(LOG_INFO, "starting up");
@@ -1905,7 +2055,7 @@ static int daemonize (void) /* {{{ */
     return (-1);
   }
 
-  status = write_pidfile ();
+  status = write_pidfile (fd);
   return status;
 } /* }}} int daemonize */
 
@@ -1929,7 +2079,7 @@ static int read_options (int argc, char **argv) /* {{{ */
   int option;
   int status = 0;
 
-  while ((option = getopt(argc, argv, "gl:f:w:b:z:p:j:h?")) != -1)
+  while ((option = getopt(argc, argv, "gl:L:f:w:b:z:p:j:h?F")) != -1)
   {
     switch (option)
     {
@@ -1937,12 +2087,22 @@ static int read_options (int argc, char **argv) /* {{{ */
         stay_foreground=1;
         break;
 
+      case 'L':
       case 'l':
       {
-        char **temp;
+        listen_socket_t **temp;
+        listen_socket_t *new;
 
-        temp = (char **) realloc (config_listen_address_list,
-            sizeof (char *) * (config_listen_address_list_len + 1));
+        new = malloc(sizeof(listen_socket_t));
+        if (new == NULL)
+        {
+          fprintf(stderr, "read_options: malloc failed.\n");
+          return(2);
+        }
+        memset(new, 0, sizeof(listen_socket_t));
+
+        temp = (listen_socket_t **) realloc (config_listen_address_list,
+            sizeof (listen_socket_t *) * (config_listen_address_list_len + 1));
         if (temp == NULL)
         {
           fprintf (stderr, "read_options: realloc failed.\n");
@@ -1950,12 +2110,10 @@ static int read_options (int argc, char **argv) /* {{{ */
         }
         config_listen_address_list = temp;
 
-        temp[config_listen_address_list_len] = strdup (optarg);
-        if (temp[config_listen_address_list_len] == NULL)
-        {
-          fprintf (stderr, "read_options: strdup failed.\n");
-          return (2);
-        }
+        strncpy(new->addr, optarg, sizeof(new->addr)-1);
+        new->privilege = (option == 'l') ? PRIV_HIGH : PRIV_LOW;
+
+        temp[config_listen_address_list_len] = new;
         config_listen_address_list_len++;
       }
       break;
@@ -2047,6 +2205,10 @@ static int read_options (int argc, char **argv) /* {{{ */
       }
       break;
 
+      case 'F':
+        config_flush_at_shutdown = 1;
+        break;
+
       case 'j':
       {
         struct stat statbuf;
@@ -2090,6 +2252,7 @@ static int read_options (int argc, char **argv) /* {{{ */
             "\n"
             "Valid options are:\n"
             "  -l <address>  Socket address to listen to.\n"
+            "  -L <address>  Socket address to listen to ('FLUSH' only).\n"
             "  -w <seconds>  Interval in which to write data.\n"
             "  -z <delay>    Delay writes up to <delay> seconds to spread load\n"
             "  -f <seconds>  Interval in which to flush dead data.\n"
@@ -2097,6 +2260,7 @@ static int read_options (int argc, char **argv) /* {{{ */
             "  -b <dir>      Base directory to change to.\n"
             "  -g            Do not fork and run in the foreground.\n"
             "  -j <dir>      Directory in which to create the journal files.\n"
+            "  -F            Always flush all updates at shutdown\n"
             "\n"
             "For more information and a detailed description of all options "
             "please refer\n"
@@ -2115,6 +2279,9 @@ static int read_options (int argc, char **argv) /* {{{ */
     fprintf(stderr, "WARNING: write delay (-z) should NOT be larger than"
             " write interval (-w) !\n");
 
+  if (journal_cur == NULL)
+    config_flush_at_shutdown = 1;
+
   return (status);
 } /* }}} int read_options */