src/rrd_daemon.c

   1 /**
   2  * RRDTool - src/rrd_daemon.c
   3  * Copyright (C) 2008 Florian octo Forster
   4  *
   5  * This program is free software; you can redistribute it and/or modify it
   6  * under the terms of the GNU General Public License as published by the
   7  * Free Software Foundation; only version 2 of the License is applicable.
   8  *
   9  * This program is distributed in the hope that it will be useful, but
  10  * WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12  * General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU General Public License along
  15  * with this program; if not, write to the Free Software Foundation, Inc.,
  16  * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
  17  *
  18  * Authors:
  19  *   Florian octo Forster <octo at verplant.org>
  20  **/
  21
  22 /*
  23  * First tell the compiler to stick to the C99 and POSIX standards as close as
  24  * possible.
  25  */
  26 #ifndef __STRICT_ANSI__ /* {{{ */
  27 # define __STRICT_ANSI__
  28 #endif
  29
  30 #ifndef _ISOC99_SOURCE
  31 # define _ISOC99_SOURCE
  32 #endif
  33
  34 #ifdef _POSIX_C_SOURCE
  35 # undef _POSIX_C_SOURCE
  36 #endif
  37 #define _POSIX_C_SOURCE 200112L
  38
  39 /* Single UNIX needed for strdup. */
  40 #ifdef _XOPEN_SOURCE
  41 # undef _XOPEN_SOURCE
  42 #endif
  43 #define _XOPEN_SOURCE 500
  44
  45 #ifndef _REENTRANT
  46 # define _REENTRANT
  47 #endif
  48
  49 #ifndef _THREAD_SAFE
  50 # define _THREAD_SAFE
  51 #endif
  52
  53 #ifdef _GNU_SOURCE
  54 # undef _GNU_SOURCE
  55 #endif
  56 /* }}} */
  57
  58 /*
  59  * Now for some includes..
  60  */
  61 #include "rrd.h" /* {{{ */
  62 #include "rrd_client.h"
  63
  64 #include <stdlib.h>
  65 #include <stdint.h>
  66 #include <stdio.h>
  67 #include <unistd.h>
  68 #include <string.h>
  69 #include <strings.h>
  70 #include <stdint.h>
  71 #include <inttypes.h>
  72
  73 #include <sys/types.h>
  74 #include <sys/stat.h>
  75 #include <fcntl.h>
  76 #include <signal.h>
  77 #include <sys/socket.h>
  78 #include <sys/un.h>
  79 #include <netdb.h>
  80 #include <poll.h>
  81 #include <syslog.h>
  82 #include <pthread.h>
  83 #include <errno.h>
  84 #include <assert.h>
  85 #include <sys/time.h>
  86 #include <time.h>
  87
  88 #include <glib-2.0/glib.h>
  89 /* }}} */
  90
  91 #define RRDD_LOG(severity, ...) syslog ((severity), __VA_ARGS__)
  92
  93 #ifndef __GNUC__
  94 # define __attribute__(x) /**/
  95 #endif
  96
  97 /*
  98  * Types
  99  */
 100 struct listen_socket_s
 101 {
 102   int fd;
 103   char path[PATH_MAX + 1];
 104 };
 105 typedef struct listen_socket_s listen_socket_t;
 106
 107 struct cache_item_s;
 108 typedef struct cache_item_s cache_item_t;
 109 struct cache_item_s
 110 {
 111   char *file;
 112   char **values;
 113   int values_num;
 114   time_t last_flush_time;
 115 #define CI_FLAGS_IN_TREE  0x01
 116 #define CI_FLAGS_IN_QUEUE 0x02
 117   int flags;
 118
 119   cache_item_t *next;
 120 };
 121
 122 struct callback_flush_data_s
 123 {
 124   time_t now;
 125   char **keys;
 126   size_t keys_num;
 127 };
 128 typedef struct callback_flush_data_s callback_flush_data_t;
 129
 130 enum queue_side_e
 131 {
 132   HEAD,
 133   TAIL
 134 };
 135 typedef enum queue_side_e queue_side_t;
 136
 137 /*
 138  * Variables
 139  */
 140 static listen_socket_t *listen_fds = NULL;
 141 static size_t listen_fds_num = 0;
 142
 143 static int do_shutdown = 0;
 144
 145 static pthread_t queue_thread;
 146
 147 static pthread_t *connetion_threads = NULL;
 148 static pthread_mutex_t connetion_threads_lock = PTHREAD_MUTEX_INITIALIZER;
 149 static int connetion_threads_num = 0;
 150
 151 /* Cache stuff */
 152 static GTree          *cache_tree = NULL;
 153 static cache_item_t   *cache_queue_head = NULL;
 154 static cache_item_t   *cache_queue_tail = NULL;
 155 static pthread_mutex_t cache_lock = PTHREAD_MUTEX_INITIALIZER;
 156 static pthread_cond_t  cache_cond = PTHREAD_COND_INITIALIZER;
 157
 158 static pthread_cond_t  flush_cond = PTHREAD_COND_INITIALIZER;
 159
 160 static int config_write_interval = 300;
 161 static int config_flush_interval = 3600;
 162 static char *config_pid_file = NULL;
 163 static char *config_base_dir = NULL;
 164
 165 static char **config_listen_address_list = NULL;
 166 static int config_listen_address_list_len = 0;
 167
 168 static uint64_t stats_queue_length = 0;
 169 static uint64_t stats_updates_written = 0;
 170 static uint64_t stats_data_sets_written = 0;
 171 static pthread_mutex_t stats_lock = PTHREAD_MUTEX_INITIALIZER;
 172
 173 /*
 174  * Functions
 175  */
 176 static void sig_int_handler (int s __attribute__((unused))) /* {{{ */
 177 {
 178   do_shutdown++;
 179 } /* }}} void sig_int_handler */
 180
 181 static void sig_term_handler (int s __attribute__((unused))) /* {{{ */
 182 {
 183   do_shutdown++;
 184 } /* }}} void sig_term_handler */
 185
 186 static int write_pidfile (void) /* {{{ */
 187 {
 188   pid_t pid;
 189   char *file;
 190   FILE *fh;
 191
 192   pid = getpid ();
 193
 194   file = (config_pid_file != NULL)
 195     ? config_pid_file
 196     : LOCALSTATEDIR "/run/rrdcached.pid";
 197
 198   fh = fopen (file, "w");
 199   if (fh == NULL)
 200   {
 201     RRDD_LOG (LOG_ERR, "write_pidfile: Opening `%s' failed.", file);
 202     return (-1);
 203   }
 204
 205   fprintf (fh, "%i\n", (int) pid);
 206   fclose (fh);
 207
 208   return (0);
 209 } /* }}} int write_pidfile */
 210
 211 static int remove_pidfile (void) /* {{{ */
 212 {
 213   char *file;
 214   int status;
 215
 216   file = (config_pid_file != NULL)
 217     ? config_pid_file
 218     : LOCALSTATEDIR "/run/rrdcached.pid";
 219
 220   status = unlink (file);
 221   if (status == 0)
 222     return (0);
 223   return (errno);
 224 } /* }}} int remove_pidfile */
 225
 226 static ssize_t sread (int fd, void *buffer_void, size_t buffer_size) /* {{{ */
 227 {
 228   char    *buffer;
 229   size_t   buffer_used;
 230   size_t   buffer_free;
 231   ssize_t  status;
 232
 233   buffer       = (char *) buffer_void;
 234   buffer_used  = 0;
 235   buffer_free  = buffer_size;
 236
 237   while (buffer_free > 0)
 238   {
 239     status = read (fd, buffer + buffer_used, buffer_free);
 240     if ((status < 0) && ((errno == EAGAIN) || (errno == EINTR)))
 241       continue;
 242
 243     if (status < 0)
 244       return (-1);
 245
 246     if (status == 0)
 247       return (0);
 248
 249     assert ((0 > status) || (buffer_free >= (size_t) status));
 250
 251     buffer_free = buffer_free - status;
 252     buffer_used = buffer_used + status;
 253
 254     if (buffer[buffer_used - 1] == '\n')
 255       break;
 256   }
 257
 258   assert (buffer_used > 0);
 259
 260   if (buffer[buffer_used - 1] != '\n')
 261   {
 262     errno = ENOBUFS;
 263     return (-1);
 264   }
 265
 266   buffer[buffer_used - 1] = 0;
 267
 268   /* Fix network line endings. */
 269   if ((buffer_used > 1) && (buffer[buffer_used - 2] == '\r'))
 270   {
 271     buffer_used--;
 272     buffer[buffer_used - 1] = 0;
 273   }
 274
 275   return (buffer_used);
 276 } /* }}} ssize_t sread */
 277
 278 static ssize_t swrite (int fd, const void *buf, size_t count) /* {{{ */
 279 {
 280   const char *ptr;
 281   size_t      nleft;
 282   ssize_t     status;
 283
 284   ptr   = (const char *) buf;
 285   nleft = count;
 286
 287   while (nleft > 0)
 288   {
 289     status = write (fd, (const void *) ptr, nleft);
 290
 291     if ((status < 0) && ((errno == EAGAIN) || (errno == EINTR)))
 292       continue;
 293
 294     if (status < 0)
 295       return (status);
 296
 297     nleft = nleft - status;
 298     ptr   = ptr   + status;
 299   }
 300
 301   return (0);
 302 } /* }}} ssize_t swrite */
 303
 304 /*
 305  * enqueue_cache_item:
 306  * `cache_lock' must be acquired before calling this function!
 307  */
 308 static int enqueue_cache_item (cache_item_t *ci, /* {{{ */
 309     queue_side_t side)
 310 {
 311   int did_insert = 0;
 312
 313   RRDD_LOG (LOG_DEBUG, "enqueue_cache_item: Adding %s to the update queue.",
 314       ci->file);
 315
 316   if (ci == NULL)
 317     return (-1);
 318
 319   if (ci->values_num == 0)
 320     return (0);
 321
 322   if (side == HEAD)
 323   {
 324     if ((ci->flags & CI_FLAGS_IN_QUEUE) == 0)
 325     {
 326       assert (ci->next == NULL);
 327       ci->next = cache_queue_head;
 328       cache_queue_head = ci;
 329
 330       if (cache_queue_tail == NULL)
 331         cache_queue_tail = cache_queue_head;
 332
 333       did_insert = 1;
 334     }
 335     else if (cache_queue_head == ci)
 336     {
 337       /* do nothing */
 338     }
 339     else /* enqueued, but not first entry */
 340     {
 341       cache_item_t *prev;
 342
 343       /* find previous entry */
 344       for (prev = cache_queue_head; prev != NULL; prev = prev->next)
 345         if (prev->next == ci)
 346           break;
 347       assert (prev != NULL);
 348
 349       /* move to the front */
 350       prev->next = ci->next;
 351       ci->next = cache_queue_head;
 352       cache_queue_head = ci;
 353
 354       /* check if we need to adapt the tail */
 355       if (cache_queue_tail == ci)
 356         cache_queue_tail = prev;
 357     }
 358   }
 359   else /* (side == TAIL) */
 360   {
 361     /* We don't move values back in the list.. */
 362     if ((ci->flags & CI_FLAGS_IN_QUEUE) != 0)
 363       return (0);
 364
 365     assert (ci->next == NULL);
 366
 367     if (cache_queue_tail == NULL)
 368       cache_queue_head = ci;
 369     else
 370       cache_queue_tail->next = ci;
 371     cache_queue_tail = ci;
 372
 373     did_insert = 1;
 374   }
 375
 376   ci->flags |= CI_FLAGS_IN_QUEUE;
 377
 378   if (did_insert)
 379   {
 380     pthread_mutex_lock (&stats_lock);
 381     stats_queue_length++;
 382     pthread_mutex_unlock (&stats_lock);
 383   }
 384
 385   return (0);
 386 } /* }}} int enqueue_cache_item */
 387
 388 /*
 389  * tree_callback_flush:
 390  * Called via `g_tree_foreach' in `queue_thread_main'. `cache_lock' is held
 391  * while this is in progress.
 392  */
 393 static gboolean tree_callback_flush (gpointer key, gpointer value, /* {{{ */
 394     gpointer data)
 395 {
 396   cache_item_t *ci;
 397   callback_flush_data_t *cfd;
 398
 399   ci = (cache_item_t *) value;
 400   cfd = (callback_flush_data_t *) data;
 401
 402   if (((cfd->now - ci->last_flush_time) >= config_write_interval)
 403       && ((ci->flags & CI_FLAGS_IN_QUEUE) == 0)
 404       && (ci->values_num > 0))
 405   {
 406     enqueue_cache_item (ci, TAIL);
 407   }
 408   else if (((cfd->now - ci->last_flush_time) >= config_flush_interval)
 409       && ((ci->flags & CI_FLAGS_IN_QUEUE) == 0)
 410       && (ci->values_num <= 0))
 411   {
 412     char **temp;
 413
 414     temp = (char **) realloc (cfd->keys,
 415         sizeof (char *) * (cfd->keys_num + 1));
 416     if (temp == NULL)
 417     {
 418       RRDD_LOG (LOG_ERR, "tree_callback_flush: realloc failed.");
 419       return (FALSE);
 420     }
 421     cfd->keys = temp;
 422     /* Make really sure this points to the _same_ place */
 423     assert ((char *) key == ci->file);
 424     cfd->keys[cfd->keys_num] = (char *) key;
 425     cfd->keys_num++;
 426   }
 427
 428   return (FALSE);
 429 } /* }}} gboolean tree_callback_flush */
 430
 431 static void *queue_thread_main (void *args __attribute__((unused))) /* {{{ */
 432 {
 433   struct timeval now;
 434   struct timespec next_flush;
 435
 436   gettimeofday (&now, NULL);
 437   next_flush.tv_sec = now.tv_sec + config_flush_interval;
 438   next_flush.tv_nsec = 1000 * now.tv_usec;
 439
 440   pthread_mutex_lock (&cache_lock);
 441   while ((do_shutdown == 0) || (cache_queue_head != NULL))
 442   {
 443     cache_item_t *ci;
 444     char *file;
 445     char **values;
 446     int values_num;
 447     int status;
 448     int i;
 449
 450     /* First, check if it's time to do the cache flush. */
 451     gettimeofday (&now, NULL);
 452     if ((now.tv_sec > next_flush.tv_sec)
 453         || ((now.tv_sec == next_flush.tv_sec)
 454           && ((1000 * now.tv_usec) > next_flush.tv_nsec)))
 455     {
 456       callback_flush_data_t cfd;
 457       size_t k;
 458
 459       memset (&cfd, 0, sizeof (cfd));
 460       /* Pass the current time as user data so that we don't need to call
 461        * `time' for each node. */
 462       cfd.now = time (NULL);
 463       cfd.keys = NULL;
 464       cfd.keys_num = 0;
 465
 466       /* `tree_callback_flush' will return the keys of all values that haven't
 467        * been touched in the last `config_flush_interval' seconds in `cfd'.
 468        * The char*'s in this array point to the same memory as ci->file, so we
 469        * don't need to free them separately. */
 470       g_tree_foreach (cache_tree, tree_callback_flush, (gpointer) &cfd);
 471
 472       for (k = 0; k < cfd.keys_num; k++)
 473       {
 474         /* This must not fail. */
 475         ci = (cache_item_t *) g_tree_lookup (cache_tree, cfd.keys[k]);
 476         assert (ci != NULL);
 477
 478         /* If we end up here with values available, something's seriously
 479          * messed up. */
 480         assert (ci->values_num == 0);
 481
 482         /* Remove the node from the tree */
 483         g_tree_remove (cache_tree, cfd.keys[k]);
 484         cfd.keys[k] = NULL;
 485
 486         /* Now free and clean up `ci'. */
 487         free (ci->file);
 488         ci->file = NULL;
 489         free (ci);
 490         ci = NULL;
 491       } /* for (k = 0; k < cfd.keys_num; k++) */
 492
 493       if (cfd.keys != NULL)
 494       {
 495         free (cfd.keys);
 496         cfd.keys = NULL;
 497       }
 498
 499       /* Determine the time of the next cache flush. */
 500       while (next_flush.tv_sec < now.tv_sec)
 501         next_flush.tv_sec += config_flush_interval;
 502     }
 503
 504     /* Now, check if there's something to store away. If not, wait until
 505      * something comes in or it's time to do the cache flush. */
 506     if (cache_queue_head == NULL)
 507     {
 508       status = pthread_cond_timedwait (&cache_cond, &cache_lock, &next_flush);
 509       if ((status != 0) && (status != ETIMEDOUT))
 510       {
 511         RRDD_LOG (LOG_ERR, "queue_thread_main: "
 512             "pthread_cond_timedwait returned %i.", status);
 513       }
 514     }
 515
 516     /* Check if a value has arrived. This may be NULL if we timed out or there
 517      * was an interrupt such as a signal. */
 518     if (cache_queue_head == NULL)
 519       continue;
 520
 521     ci = cache_queue_head;
 522
 523     /* copy the relevant parts */
 524     file = strdup (ci->file);
 525     if (file == NULL)
 526     {
 527       RRDD_LOG (LOG_ERR, "queue_thread_main: strdup failed.");
 528       continue;
 529     }
 530
 531     values = ci->values;
 532     values_num = ci->values_num;
 533
 534     ci->values = NULL;
 535     ci->values_num = 0;
 536
 537     ci->last_flush_time = time (NULL);
 538     ci->flags &= ~(CI_FLAGS_IN_QUEUE);
 539
 540     cache_queue_head = ci->next;
 541     if (cache_queue_head == NULL)
 542       cache_queue_tail = NULL;
 543     ci->next = NULL;
 544
 545     pthread_mutex_lock (&stats_lock);
 546     assert (stats_queue_length > 0);
 547     stats_queue_length--;
 548     pthread_mutex_unlock (&stats_lock);
 549
 550     pthread_mutex_unlock (&cache_lock);
 551
 552     RRDD_LOG (LOG_DEBUG, "queue_thread_main: rrd_update (%s, %i, %p)",
 553         file, values_num, (void *) values);
 554
 555     status = rrd_update_r (file, NULL, values_num, (void *) values);
 556     if (status != 0)
 557     {
 558       RRDD_LOG (LOG_ERR, "queue_thread_main: "
 559           "rrd_update_r failed with status %i.",
 560           status);
 561     }
 562
 563     free (file);
 564     for (i = 0; i < values_num; i++)
 565       free (values[i]);
 566
 567     pthread_mutex_lock (&stats_lock);
 568     stats_updates_written++;
 569     stats_data_sets_written += values_num;
 570     pthread_mutex_unlock (&stats_lock);
 571
 572     pthread_mutex_lock (&cache_lock);
 573     pthread_cond_broadcast (&flush_cond);
 574   } /* while (do_shutdown == 0) */
 575   pthread_mutex_unlock (&cache_lock);
 576
 577   RRDD_LOG (LOG_DEBUG, "queue_thread_main: Exiting.");
 578
 579   return (NULL);
 580 } /* }}} void *queue_thread_main */
 581
 582 static int buffer_get_field (char **buffer_ret, /* {{{ */
 583     size_t *buffer_size_ret, char **field_ret)
 584 {
 585   char *buffer;
 586   size_t buffer_pos;
 587   size_t buffer_size;
 588   char *field;
 589   size_t field_size;
 590   int status;
 591
 592   buffer = *buffer_ret;
 593   buffer_pos = 0;
 594   buffer_size = *buffer_size_ret;
 595   field = *buffer_ret;
 596   field_size = 0;
 597
 598   if (buffer_size <= 0)
 599     return (-1);
 600
 601   /* This is ensured by `handle_request'. */
 602   assert (buffer[buffer_size - 1] == ' ');
 603
 604   status = -1;
 605   while (buffer_pos < buffer_size)
 606   {
 607     /* Check for end-of-field or end-of-buffer */
 608     if (buffer[buffer_pos] == ' ')
 609     {
 610       field[field_size] = 0;
 611       field_size++;
 612       buffer_pos++;
 613       status = 0;
 614       break;
 615     }
 616     /* Handle escaped characters. */
 617     else if (buffer[buffer_pos] == '\\')
 618     {
 619       if (buffer_pos >= (buffer_size - 1))
 620         break;
 621       buffer_pos++;
 622       field[field_size] = buffer[buffer_pos];
 623       field_size++;
 624       buffer_pos++;
 625     }
 626     /* Normal operation */
 627     else
 628     {
 629       field[field_size] = buffer[buffer_pos];
 630       field_size++;
 631       buffer_pos++;
 632     }
 633   } /* while (buffer_pos < buffer_size) */
 634
 635   if (status != 0)
 636     return (status);
 637
 638   *buffer_ret = buffer + buffer_pos;
 639   *buffer_size_ret = buffer_size - buffer_pos;
 640   *field_ret = field;
 641
 642   return (0);
 643 } /* }}} int buffer_get_field */
 644
 645 static int flush_file (const char *filename) /* {{{ */
 646 {
 647   cache_item_t *ci;
 648
 649   pthread_mutex_lock (&cache_lock);
 650
 651   ci = (cache_item_t *) g_tree_lookup (cache_tree, filename);
 652   if (ci == NULL)
 653   {
 654     pthread_mutex_unlock (&cache_lock);
 655     return (ENOENT);
 656   }
 657
 658   /* Enqueue at head */
 659   enqueue_cache_item (ci, HEAD);
 660   pthread_cond_signal (&cache_cond);
 661
 662   while ((ci->flags & CI_FLAGS_IN_QUEUE) != 0)
 663   {
 664     ci = NULL;
 665
 666     pthread_cond_wait (&flush_cond, &cache_lock);
 667
 668     ci = g_tree_lookup (cache_tree, filename);
 669     if (ci == NULL)
 670     {
 671       RRDD_LOG (LOG_ERR, "flush_file: Tree node went away "
 672           "while waiting for flush.");
 673       pthread_mutex_unlock (&cache_lock);
 674       return (-1);
 675     }
 676   }
 677
 678   pthread_mutex_unlock (&cache_lock);
 679   return (0);
 680 } /* }}} int flush_file */
 681
 682 static int handle_request_help (int fd, /* {{{ */
 683     char *buffer, size_t buffer_size)
 684 {
 685   int status;
 686   char **help_text;
 687   size_t help_text_len;
 688   char *command;
 689   size_t i;
 690
 691   char *help_help[] =
 692   {
 693     "4 Command overview\n",
 694     "FLUSH <filename>\n",
 695     "HELP [<command>]\n",
 696     "UPDATE <filename> <values> [<values> ...]\n",
 697     "STATS\n"
 698   };
 699   size_t help_help_len = sizeof (help_help) / sizeof (help_help[0]);
 700
 701   char *help_flush[] =
 702   {
 703     "4 Help for FLUSH\n",
 704     "Usage: FLUSH <filename>\n",
 705     "\n",
 706     "Adds the given filename to the head of the update queue and returns\n",
 707     "after is has been dequeued.\n"
 708   };
 709   size_t help_flush_len = sizeof (help_flush) / sizeof (help_flush[0]);
 710
 711   char *help_update[] =
 712   {
 713     "9 Help for UPDATE\n",
 714     "Usage: UPDATE <filename> <values> [<values> ...]\n"
 715     "\n",
 716     "Adds the given file to the internal cache if it is not yet known and\n",
 717     "appends the given value(s) to the entry. See the rrdcached(1) manpage\n",
 718     "for details.\n",
 719     "\n",
 720     "Each <values> has the following form:\n",
 721     "  <values> = <time>:<value>[:<value>[...]]\n",
 722     "See the rrdupdate(1) manpage for details.\n"
 723   };
 724   size_t help_update_len = sizeof (help_update) / sizeof (help_update[0]);
 725
 726   char *help_stats[] =
 727   {
 728     "4 Help for STATS\n",
 729     "Usage: STATS\n",
 730     "\n",
 731     "Returns some performance counters, see the rrdcached(1) manpage for\n",
 732     "a description of the values.\n"
 733   };
 734   size_t help_stats_len = sizeof (help_stats) / sizeof (help_stats[0]);
 735
 736   status = buffer_get_field (&buffer, &buffer_size, &command);
 737   if (status != 0)
 738   {
 739     help_text = help_help;
 740     help_text_len = help_help_len;
 741   }
 742   else
 743   {
 744     if (strcasecmp (command, "update") == 0)
 745     {
 746       help_text = help_update;
 747       help_text_len = help_update_len;
 748     }
 749     else if (strcasecmp (command, "flush") == 0)
 750     {
 751       help_text = help_flush;
 752       help_text_len = help_flush_len;
 753     }
 754     else if (strcasecmp (command, "stats") == 0)
 755     {
 756       help_text = help_stats;
 757       help_text_len = help_stats_len;
 758     }
 759     else
 760     {
 761       help_text = help_help;
 762       help_text_len = help_help_len;
 763     }
 764   }
 765
 766   for (i = 0; i < help_text_len; i++)
 767   {
 768     status = swrite (fd, help_text[i], strlen (help_text[i]));
 769     if (status < 0)
 770     {
 771       status = errno;
 772       RRDD_LOG (LOG_ERR, "handle_request_help: swrite returned an error.");
 773       return (status);
 774     }
 775   }
 776
 777   return (0);
 778 } /* }}} int handle_request_help */
 779
 780 static int handle_request_stats (int fd, /* {{{ */
 781     char *buffer __attribute__((unused)),
 782     size_t buffer_size __attribute__((unused)))
 783 {
 784   int status;
 785   char outbuf[4096];
 786
 787   uint64_t copy_queue_length;
 788   uint64_t copy_updates_written;
 789   uint64_t copy_data_sets_written;
 790
 791   uint64_t tree_nodes_number;
 792   uint64_t tree_depth;
 793
 794   pthread_mutex_lock (&stats_lock);
 795   copy_queue_length       = stats_queue_length;
 796   copy_updates_written    = stats_updates_written;
 797   copy_data_sets_written  = stats_data_sets_written;
 798   pthread_mutex_unlock (&stats_lock);
 799
 800   pthread_mutex_lock (&cache_lock);
 801   tree_nodes_number = (uint64_t) g_tree_nnodes (cache_tree);
 802   tree_depth        = (uint64_t) g_tree_height (cache_tree);
 803   pthread_mutex_unlock (&cache_lock);
 804
 805 #define RRDD_STATS_SEND \
 806   outbuf[sizeof (outbuf) - 1] = 0; \
 807   status = swrite (fd, outbuf, strlen (outbuf)); \
 808   if (status < 0) \
 809   { \
 810     status = errno; \
 811     RRDD_LOG (LOG_INFO, "handle_request_stats: swrite returned an error."); \
 812     return (status); \
 813   }
 814
 815   strncpy (outbuf, "5 Statistics follow\n", sizeof (outbuf));
 816   RRDD_STATS_SEND;
 817
 818   snprintf (outbuf, sizeof (outbuf),
 819       "QueueLength: %"PRIu64"\n", copy_queue_length);
 820   RRDD_STATS_SEND;
 821
 822   snprintf (outbuf, sizeof (outbuf),
 823       "UpdatesWritten: %"PRIu64"\n", copy_updates_written);
 824   RRDD_STATS_SEND;
 825
 826   snprintf (outbuf, sizeof (outbuf),
 827       "DataSetsWritten: %"PRIu64"\n", copy_data_sets_written);
 828   RRDD_STATS_SEND;
 829
 830   snprintf (outbuf, sizeof (outbuf),
 831       "TreeNodesNumber: %"PRIu64"\n", tree_nodes_number);
 832   RRDD_STATS_SEND;
 833
 834   snprintf (outbuf, sizeof (outbuf),
 835       "TreeDepth: %"PRIu64"\n", tree_depth);
 836   RRDD_STATS_SEND;
 837
 838   return (0);
 839 #undef RRDD_STATS_SEND
 840 } /* }}} int handle_request_stats */
 841
 842 static int handle_request_flush (int fd, /* {{{ */
 843     char *buffer, size_t buffer_size)
 844 {
 845   char *file;
 846   int status;
 847   char result[4096];
 848
 849   status = buffer_get_field (&buffer, &buffer_size, &file);
 850   if (status != 0)
 851   {
 852     strncpy (result, "-1 Usage: flush <filename>\n", sizeof (result));
 853   }
 854   else
 855   {
 856     status = flush_file (file);
 857     if (status == 0)
 858       snprintf (result, sizeof (result), "0 Successfully flushed %s.\n", file);
 859     else if (status == ENOENT)
 860       snprintf (result, sizeof (result), "-1 No such file: %s.\n", file);
 861     else if (status < 0)
 862       strncpy (result, "-1 Internal error.\n", sizeof (result));
 863     else
 864       snprintf (result, sizeof (result), "-1 Failed with status %i.\n", status);
 865   }
 866   result[sizeof (result) - 1] = 0;
 867
 868   status = swrite (fd, result, strlen (result));
 869   if (status < 0)
 870   {
 871     status = errno;
 872     RRDD_LOG (LOG_INFO, "handle_request_flush: swrite returned an error.");
 873     return (status);
 874   }
 875
 876   return (0);
 877 } /* }}} int handle_request_flush */
 878
 879 static int handle_request_update (int fd, /* {{{ */
 880     char *buffer, size_t buffer_size)
 881 {
 882   char *file;
 883   int values_num = 0;
 884   int status;
 885
 886   time_t now;
 887
 888   cache_item_t *ci;
 889   char answer[4096];
 890
 891 #define RRDD_UPDATE_SEND \
 892   answer[sizeof (answer) - 1] = 0; \
 893   status = swrite (fd, answer, strlen (answer)); \
 894   if (status < 0) \
 895   { \
 896     status = errno; \
 897     RRDD_LOG (LOG_INFO, "handle_request_update: swrite returned an error."); \
 898     return (status); \
 899   }
 900
 901   now = time (NULL);
 902
 903   status = buffer_get_field (&buffer, &buffer_size, &file);
 904   if (status != 0)
 905   {
 906     strncpy (answer, "-1 Usage: UPDATE <filename> <values> [<values> ...]\n",
 907         sizeof (answer));
 908     RRDD_UPDATE_SEND;
 909     return (0);
 910   }
 911
 912   pthread_mutex_lock (&cache_lock);
 913
 914   ci = g_tree_lookup (cache_tree, file);
 915   if (ci == NULL) /* {{{ */
 916   {
 917     ci = (cache_item_t *) malloc (sizeof (cache_item_t));
 918     if (ci == NULL)
 919     {
 920       pthread_mutex_unlock (&cache_lock);
 921       RRDD_LOG (LOG_ERR, "handle_request_update: malloc failed.");
 922
 923       strncpy (answer, "-1 malloc failed.\n", sizeof (answer));
 924       RRDD_UPDATE_SEND;
 925       return (0);
 926     }
 927     memset (ci, 0, sizeof (cache_item_t));
 928
 929     ci->file = strdup (file);
 930     if (ci->file == NULL)
 931     {
 932       pthread_mutex_unlock (&cache_lock);
 933       free (ci);
 934       RRDD_LOG (LOG_ERR, "handle_request_update: strdup failed.");
 935
 936       strncpy (answer, "-1 strdup failed.\n", sizeof (answer));
 937       RRDD_UPDATE_SEND;
 938       return (0);
 939     }
 940
 941     ci->values = NULL;
 942     ci->values_num = 0;
 943     ci->last_flush_time = now;
 944     ci->flags = CI_FLAGS_IN_TREE;
 945
 946     g_tree_insert (cache_tree, (void *) ci->file, (void *) ci);
 947
 948     RRDD_LOG (LOG_DEBUG, "handle_request_update: Created new tree node %s.",
 949         ci->file);
 950   } /* }}} */
 951   assert (ci != NULL);
 952
 953   while (buffer_size > 0)
 954   {
 955     char **temp;
 956     char *value;
 957
 958     status = buffer_get_field (&buffer, &buffer_size, &value);
 959     if (status != 0)
 960     {
 961       RRDD_LOG (LOG_INFO, "handle_request_update: Error reading field.");
 962       break;
 963     }
 964
 965     temp = (char **) realloc (ci->values,
 966         sizeof (char *) * (ci->values_num + 1));
 967     if (temp == NULL)
 968     {
 969       RRDD_LOG (LOG_ERR, "handle_request_update: realloc failed.");
 970       continue;
 971     }
 972     ci->values = temp;
 973
 974     ci->values[ci->values_num] = strdup (value);
 975     if (ci->values[ci->values_num] == NULL)
 976     {
 977       RRDD_LOG (LOG_ERR, "handle_request_update: strdup failed.");
 978       continue;
 979     }
 980     ci->values_num++;
 981
 982     values_num++;
 983   }
 984
 985   if (((now - ci->last_flush_time) >= config_write_interval)
 986       && ((ci->flags & CI_FLAGS_IN_QUEUE) == 0)
 987       && (ci->values_num > 0))
 988   {
 989     enqueue_cache_item (ci, TAIL);
 990     pthread_cond_signal (&cache_cond);
 991   }
 992
 993   pthread_mutex_unlock (&cache_lock);
 994
 995   if (values_num < 1)
 996   {
 997     strncpy (answer, "-1 No values updated.\n", sizeof (answer));
 998   }
 999   else
1000   {
1001     snprintf (answer, sizeof (answer), "0 Enqueued %i value%s\n", values_num,
1002         (values_num == 1) ? "" : "s");
1003   }
1004   RRDD_UPDATE_SEND;
1005   return (0);
1006 #undef RRDD_UPDATE_SEND
1007 } /* }}} int handle_request_update */
1008
1009 static int handle_request (int fd) /* {{{ */
1010 {
1011   char buffer[4096];
1012   size_t buffer_size;
1013   char *buffer_ptr;
1014   char *command;
1015   int status;
1016
1017   status = (int) sread (fd, buffer, sizeof (buffer));
1018   if (status == 0)
1019   {
1020     return (1);
1021   }
1022   else if (status < 0)
1023   {
1024     RRDD_LOG (LOG_ERR, "handle_request: sread failed.");
1025     return (-1);
1026   }
1027   buffer_size = (size_t) status;
1028   assert (buffer_size <= sizeof (buffer));
1029   assert (buffer[buffer_size - 1] == 0);
1030
1031   /* Place the normal field separator at the end to simplify
1032    * `buffer_get_field's work. */
1033   buffer[buffer_size - 1] = ' ';
1034
1035   buffer_ptr = buffer;
1036   command = NULL;
1037   status = buffer_get_field (&buffer_ptr, &buffer_size, &command);
1038   if (status != 0)
1039   {
1040     RRDD_LOG (LOG_INFO, "handle_request: Unable parse command.");
1041     return (-1);
1042   }
1043
1044   if (strcasecmp (command, "update") == 0)
1045   {
1046     return (handle_request_update (fd, buffer_ptr, buffer_size));
1047   }
1048   else if (strcasecmp (command, "flush") == 0)
1049   {
1050     return (handle_request_flush (fd, buffer_ptr, buffer_size));
1051   }
1052   else if (strcasecmp (command, "stats") == 0)
1053   {
1054     return (handle_request_stats (fd, buffer_ptr, buffer_size));
1055   }
1056   else if (strcasecmp (command, "help") == 0)
1057   {
1058     return (handle_request_help (fd, buffer_ptr, buffer_size));
1059   }
1060   else
1061   {
1062     char result[4096];
1063
1064     snprintf (result, sizeof (result), "-1 Unknown command: %s\n", command);
1065     result[sizeof (result) - 1] = 0;
1066
1067     status = swrite (fd, result, strlen (result));
1068     if (status < 0)
1069     {
1070       RRDD_LOG (LOG_ERR, "handle_request: swrite failed.");
1071       return (-1);
1072     }
1073   }
1074
1075   return (0);
1076 } /* }}} int handle_request */
1077
1078 static void *connection_thread_main (void *args /* {{{ */
1079     __attribute__((unused)))
1080 {
1081   pthread_t self;
1082   int i;
1083   int fd;
1084
1085   fd = *((int *) args);
1086
1087   pthread_mutex_lock (&connetion_threads_lock);
1088   {
1089     pthread_t *temp;
1090
1091     temp = (pthread_t *) realloc (connetion_threads,
1092         sizeof (pthread_t) * (connetion_threads_num + 1));
1093     if (temp == NULL)
1094     {
1095       RRDD_LOG (LOG_ERR, "connection_thread_main: realloc failed.");
1096     }
1097     else
1098     {
1099       connetion_threads = temp;
1100       connetion_threads[connetion_threads_num] = pthread_self ();
1101       connetion_threads_num++;
1102     }
1103   }
1104   pthread_mutex_unlock (&connetion_threads_lock);
1105
1106   while (do_shutdown == 0)
1107   {
1108     struct pollfd pollfd;
1109     int status;
1110
1111     pollfd.fd = fd;
1112     pollfd.events = POLLIN | POLLPRI;
1113     pollfd.revents = 0;
1114
1115     status = poll (&pollfd, 1, /* timeout = */ 500);
1116     if (status == 0) /* timeout */
1117       continue;
1118     else if (status < 0) /* error */
1119     {
1120       status = errno;
1121       if (status == EINTR)
1122         continue;
1123       RRDD_LOG (LOG_ERR, "connection_thread_main: poll(2) failed.");
1124       continue;
1125     }
1126
1127     if ((pollfd.revents & POLLHUP) != 0) /* normal shutdown */
1128     {
1129       close (fd);
1130       break;
1131     }
1132     else if ((pollfd.revents & (POLLIN | POLLPRI)) == 0)
1133     {
1134       RRDD_LOG (LOG_WARNING, "connection_thread_main: "
1135           "poll(2) returned something unexpected: %#04hx",
1136           pollfd.revents);
1137       close (fd);
1138       break;
1139     }
1140
1141     status = handle_request (fd);
1142     if (status != 0)
1143     {
1144       close (fd);
1145       break;
1146     }
1147   }
1148
1149   self = pthread_self ();
1150   /* Remove this thread from the connection threads list */
1151   pthread_mutex_lock (&connetion_threads_lock);
1152   /* Find out own index in the array */
1153   for (i = 0; i < connetion_threads_num; i++)
1154     if (pthread_equal (connetion_threads[i], self) != 0)
1155       break;
1156   assert (i < connetion_threads_num);
1157
1158   /* Move the trailing threads forward. */
1159   if (i < (connetion_threads_num - 1))
1160   {
1161     memmove (connetion_threads + i,
1162         connetion_threads + i + 1,
1163         sizeof (pthread_t) * (connetion_threads_num - i - 1));
1164   }
1165
1166   connetion_threads_num--;
1167   pthread_mutex_unlock (&connetion_threads_lock);
1168
1169   free (args);
1170   return (NULL);
1171 } /* }}} void *connection_thread_main */
1172
1173 static int open_listen_socket_unix (const char *path) /* {{{ */
1174 {
1175   int fd;
1176   struct sockaddr_un sa;
1177   listen_socket_t *temp;
1178   int status;
1179
1180   temp = (listen_socket_t *) realloc (listen_fds,
1181       sizeof (listen_fds[0]) * (listen_fds_num + 1));
1182   if (temp == NULL)
1183   {
1184     RRDD_LOG (LOG_ERR, "open_listen_socket_unix: realloc failed.");
1185     return (-1);
1186   }
1187   listen_fds = temp;
1188   memset (listen_fds + listen_fds_num, 0, sizeof (listen_fds[0]));
1189
1190   fd = socket (PF_UNIX, SOCK_STREAM, /* protocol = */ 0);
1191   if (fd < 0)
1192   {
1193     RRDD_LOG (LOG_ERR, "open_listen_socket_unix: socket(2) failed.");
1194     return (-1);
1195   }
1196
1197   memset (&sa, 0, sizeof (sa));
1198   sa.sun_family = AF_UNIX;
1199   strncpy (sa.sun_path, path, sizeof (sa.sun_path) - 1);
1200
1201   status = bind (fd, (struct sockaddr *) &sa, sizeof (sa));
1202   if (status != 0)
1203   {
1204     RRDD_LOG (LOG_ERR, "open_listen_socket_unix: bind(2) failed.");
1205     close (fd);
1206     unlink (path);
1207     return (-1);
1208   }
1209
1210   status = listen (fd, /* backlog = */ 10);
1211   if (status != 0)
1212   {
1213     RRDD_LOG (LOG_ERR, "open_listen_socket_unix: listen(2) failed.");
1214     close (fd);
1215     unlink (path);
1216     return (-1);
1217   }
1218
1219   listen_fds[listen_fds_num].fd = fd;
1220   snprintf (listen_fds[listen_fds_num].path,
1221       sizeof (listen_fds[listen_fds_num].path) - 1,
1222       "unix:%s", path);
1223   listen_fds_num++;
1224
1225   return (0);
1226 } /* }}} int open_listen_socket_unix */
1227
1228 static int open_listen_socket (const char *addr) /* {{{ */
1229 {
1230   struct addrinfo ai_hints;
1231   struct addrinfo *ai_res;
1232   struct addrinfo *ai_ptr;
1233   int status;
1234
1235   assert (addr != NULL);
1236
1237   if (strncmp ("unix:", addr, strlen ("unix:")) == 0)
1238     return (open_listen_socket_unix (addr + strlen ("unix:")));
1239   else if (addr[0] == '/')
1240     return (open_listen_socket_unix (addr));
1241
1242   memset (&ai_hints, 0, sizeof (ai_hints));
1243   ai_hints.ai_flags = 0;
1244 #ifdef AI_ADDRCONFIG
1245   ai_hints.ai_flags |= AI_ADDRCONFIG;
1246 #endif
1247   ai_hints.ai_family = AF_UNSPEC;
1248   ai_hints.ai_socktype = SOCK_STREAM;
1249
1250   ai_res = NULL;
1251   status = getaddrinfo (addr, RRDCACHED_DEFAULT_PORT, &ai_hints, &ai_res);
1252   if (status != 0)
1253   {
1254     RRDD_LOG (LOG_ERR, "open_listen_socket: getaddrinfo(%s) failed: "
1255         "%s", addr, gai_strerror (status));
1256     return (-1);
1257   }
1258
1259   for (ai_ptr = ai_res; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next)
1260   {
1261     int fd;
1262     listen_socket_t *temp;
1263
1264     temp = (listen_socket_t *) realloc (listen_fds,
1265         sizeof (listen_fds[0]) * (listen_fds_num + 1));
1266     if (temp == NULL)
1267     {
1268       RRDD_LOG (LOG_ERR, "open_listen_socket: realloc failed.");
1269       continue;
1270     }
1271     listen_fds = temp;
1272     memset (listen_fds + listen_fds_num, 0, sizeof (listen_fds[0]));
1273
1274     fd = socket (ai_ptr->ai_family, ai_ptr->ai_socktype, ai_ptr->ai_protocol);
1275     if (fd < 0)
1276     {
1277       RRDD_LOG (LOG_ERR, "open_listen_socket: socket(2) failed.");
1278       continue;
1279     }
1280
1281     status = bind (fd, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
1282     if (status != 0)
1283     {
1284       RRDD_LOG (LOG_ERR, "open_listen_socket: bind(2) failed.");
1285       close (fd);
1286       continue;
1287     }
1288
1289     status = listen (fd, /* backlog = */ 10);
1290     if (status != 0)
1291     {
1292       RRDD_LOG (LOG_ERR, "open_listen_socket: listen(2) failed.");
1293       close (fd);
1294       return (-1);
1295     }
1296
1297     listen_fds[listen_fds_num].fd = fd;
1298     strncpy (listen_fds[listen_fds_num].path, addr,
1299         sizeof (listen_fds[listen_fds_num].path) - 1);
1300     listen_fds_num++;
1301   } /* for (ai_ptr) */
1302
1303   return (0);
1304 } /* }}} int open_listen_socket */
1305
1306 static int close_listen_sockets (void) /* {{{ */
1307 {
1308   size_t i;
1309
1310   for (i = 0; i < listen_fds_num; i++)
1311   {
1312     close (listen_fds[i].fd);
1313     if (strncmp ("unix:", listen_fds[i].path, strlen ("unix:")) == 0)
1314       unlink (listen_fds[i].path + strlen ("unix:"));
1315   }
1316
1317   free (listen_fds);
1318   listen_fds = NULL;
1319   listen_fds_num = 0;
1320
1321   return (0);
1322 } /* }}} int close_listen_sockets */
1323
1324 static void *listen_thread_main (void *args __attribute__((unused))) /* {{{ */
1325 {
1326   struct pollfd *pollfds;
1327   int pollfds_num;
1328   int status;
1329   int i;
1330
1331   for (i = 0; i < config_listen_address_list_len; i++)
1332   {
1333     RRDD_LOG (LOG_DEBUG, "listen_thread_main: config_listen_address_list[%i] "
1334         "= %s", i, config_listen_address_list[i]);
1335     open_listen_socket (config_listen_address_list[i]);
1336   }
1337
1338   if (config_listen_address_list_len < 1)
1339     open_listen_socket (RRDCACHED_DEFAULT_ADDRESS);
1340
1341   if (listen_fds_num < 1)
1342   {
1343     RRDD_LOG (LOG_ERR, "listen_thread_main: No listen sockets "
1344         "could be opened. Sorry.");
1345     return (NULL);
1346   }
1347
1348   pollfds_num = listen_fds_num;
1349   pollfds = (struct pollfd *) malloc (sizeof (*pollfds) * pollfds_num);
1350   if (pollfds == NULL)
1351   {
1352     RRDD_LOG (LOG_ERR, "listen_thread_main: malloc failed.");
1353     return (NULL);
1354   }
1355   memset (pollfds, 0, sizeof (*pollfds) * pollfds_num);
1356
1357   while (do_shutdown == 0)
1358   {
1359     assert (pollfds_num == ((int) listen_fds_num));
1360     for (i = 0; i < pollfds_num; i++)
1361     {
1362       pollfds[i].fd = listen_fds[i].fd;
1363       pollfds[i].events = POLLIN | POLLPRI;
1364       pollfds[i].revents = 0;
1365     }
1366
1367     status = poll (pollfds, pollfds_num, /* timeout = */ -1);
1368     if (status < 1)
1369     {
1370       status = errno;
1371       if (status != EINTR)
1372       {
1373         RRDD_LOG (LOG_ERR, "listen_thread_main: poll(2) failed.");
1374       }
1375       continue;
1376     }
1377
1378     for (i = 0; i < pollfds_num; i++)
1379     {
1380       int *client_sd;
1381       struct sockaddr_storage client_sa;
1382       socklen_t client_sa_size;
1383       pthread_t tid;
1384
1385       if (pollfds[i].revents == 0)
1386         continue;
1387
1388       if ((pollfds[i].revents & (POLLIN | POLLPRI)) == 0)
1389       {
1390         RRDD_LOG (LOG_ERR, "listen_thread_main: "
1391             "poll(2) returned something unexpected for listen FD #%i.",
1392             pollfds[i].fd);
1393         continue;
1394       }
1395
1396       client_sd = (int *) malloc (sizeof (int));
1397       if (client_sd == NULL)
1398       {
1399         RRDD_LOG (LOG_ERR, "listen_thread_main: malloc failed.");
1400         continue;
1401       }
1402
1403       client_sa_size = sizeof (client_sa);
1404       *client_sd = accept (pollfds[i].fd,
1405           (struct sockaddr *) &client_sa, &client_sa_size);
1406       if (*client_sd < 0)
1407       {
1408         RRDD_LOG (LOG_ERR, "listen_thread_main: accept(2) failed.");
1409         continue;
1410       }
1411
1412       status = pthread_create (&tid, /* attr = */ NULL, connection_thread_main,
1413           /* args = */ (void *) client_sd);
1414       if (status != 0)
1415       {
1416         RRDD_LOG (LOG_ERR, "listen_thread_main: pthread_create failed.");
1417         close (*client_sd);
1418         free (client_sd);
1419         continue;
1420       }
1421     } /* for (pollfds_num) */
1422   } /* while (do_shutdown == 0) */
1423
1424   close_listen_sockets ();
1425
1426   pthread_mutex_lock (&connetion_threads_lock);
1427   while (connetion_threads_num > 0)
1428   {
1429     pthread_t wait_for;
1430
1431     wait_for = connetion_threads[0];
1432
1433     pthread_mutex_unlock (&connetion_threads_lock);
1434     pthread_join (wait_for, /* retval = */ NULL);
1435     pthread_mutex_lock (&connetion_threads_lock);
1436   }
1437   pthread_mutex_unlock (&connetion_threads_lock);
1438
1439   RRDD_LOG (LOG_DEBUG, "listen_thread_main: Exiting.");
1440
1441   return (NULL);
1442 } /* }}} void *listen_thread_main */
1443
1444 static int daemonize (void) /* {{{ */
1445 {
1446   pid_t child;
1447   int status;
1448   char *base_dir;
1449
1450   /* These structures are static, because `sigaction' behaves weird if the are
1451    * overwritten.. */
1452   static struct sigaction sa_int;
1453   static struct sigaction sa_term;
1454   static struct sigaction sa_pipe;
1455
1456   child = fork ();
1457   if (child < 0)
1458   {
1459     fprintf (stderr, "daemonize: fork(2) failed.\n");
1460     return (-1);
1461   }
1462   else if (child > 0)
1463   {
1464     return (1);
1465   }
1466
1467   /* Change into the /tmp directory. */
1468   base_dir = (config_base_dir != NULL)
1469     ? config_base_dir
1470     : "/tmp";
1471   status = chdir (base_dir);
1472   if (status != 0)
1473   {
1474     fprintf (stderr, "daemonize: chdir (%s) failed.\n", base_dir);
1475     return (-1);
1476   }
1477
1478   /* Become session leader */
1479   setsid ();
1480
1481   /* Open the first three file descriptors to /dev/null */
1482   close (2);
1483   close (1);
1484   close (0);
1485
1486   open ("/dev/null", O_RDWR);
1487   dup (0);
1488   dup (0);
1489
1490   /* Install signal handlers */
1491   memset (&sa_int, 0, sizeof (sa_int));
1492   sa_int.sa_handler = sig_int_handler;
1493   sigaction (SIGINT, &sa_int, NULL);
1494
1495   memset (&sa_term, 0, sizeof (sa_term));
1496   sa_term.sa_handler = sig_term_handler;
1497   sigaction (SIGTERM, &sa_term, NULL);
1498
1499   memset (&sa_pipe, 0, sizeof (sa_pipe));
1500   sa_pipe.sa_handler = SIG_IGN;
1501   sigaction (SIGPIPE, &sa_pipe, NULL);
1502
1503   openlog ("rrdcached", LOG_PID, LOG_DAEMON);
1504
1505   cache_tree = g_tree_new ((GCompareFunc) strcmp);
1506   if (cache_tree == NULL)
1507   {
1508     RRDD_LOG (LOG_ERR, "daemonize: g_tree_new failed.");
1509     return (-1);
1510   }
1511
1512   memset (&queue_thread, 0, sizeof (queue_thread));
1513   status = pthread_create (&queue_thread, /* attr = */ NULL,
1514       queue_thread_main, /* args = */ NULL);
1515   if (status != 0)
1516   {
1517     RRDD_LOG (LOG_ERR, "daemonize: pthread_create failed.");
1518     return (-1);
1519   }
1520
1521   write_pidfile ();
1522
1523   return (0);
1524 } /* }}} int daemonize */
1525
1526 static int cleanup (void) /* {{{ */
1527 {
1528   RRDD_LOG (LOG_DEBUG, "cleanup ()");
1529
1530   do_shutdown++;
1531
1532   RRDD_LOG (LOG_DEBUG, "cleanup: Joining queue_thread..");
1533   pthread_cond_signal (&cache_cond);
1534   pthread_join (queue_thread, /* return = */ NULL);
1535   RRDD_LOG (LOG_DEBUG, "cleanup: done");
1536
1537   remove_pidfile ();
1538
1539   closelog ();
1540
1541   return (0);
1542 } /* }}} int cleanup */
1543
1544 static int read_options (int argc, char **argv) /* {{{ */
1545 {
1546   int option;
1547   int status = 0;
1548
1549   while ((option = getopt(argc, argv, "l:f:w:b:p:h?")) != -1)
1550   {
1551     switch (option)
1552     {
1553       case 'l':
1554       {
1555         char **temp;
1556
1557         temp = (char **) realloc (config_listen_address_list,
1558             sizeof (char *) * (config_listen_address_list_len + 1));
1559         if (temp == NULL)
1560         {
1561           fprintf (stderr, "read_options: realloc failed.\n");
1562           return (2);
1563         }
1564         config_listen_address_list = temp;
1565
1566         temp[config_listen_address_list_len] = strdup (optarg);
1567         if (temp[config_listen_address_list_len] == NULL)
1568         {
1569           fprintf (stderr, "read_options: strdup failed.\n");
1570           return (2);
1571         }
1572         config_listen_address_list_len++;
1573       }
1574       break;
1575
1576       case 'f':
1577       {
1578         int temp;
1579
1580         temp = atoi (optarg);
1581         if (temp > 0)
1582           config_flush_interval = temp;
1583         else
1584         {
1585           fprintf (stderr, "Invalid flush interval: %s\n", optarg);
1586           status = 3;
1587         }
1588       }
1589       break;
1590
1591       case 'w':
1592       {
1593         int temp;
1594
1595         temp = atoi (optarg);
1596         if (temp > 0)
1597           config_write_interval = temp;
1598         else
1599         {
1600           fprintf (stderr, "Invalid write interval: %s\n", optarg);
1601           status = 2;
1602         }
1603       }
1604       break;
1605
1606       case 'b':
1607       {
1608         size_t len;
1609
1610         if (config_base_dir != NULL)
1611           free (config_base_dir);
1612         config_base_dir = strdup (optarg);
1613         if (config_base_dir == NULL)
1614         {
1615           fprintf (stderr, "read_options: strdup failed.\n");
1616           return (3);
1617         }
1618
1619         len = strlen (config_base_dir);
1620         while ((len > 0) && (config_base_dir[len - 1] == '/'))
1621         {
1622           config_base_dir[len - 1] = 0;
1623           len--;
1624         }
1625
1626         if (len < 1)
1627         {
1628           fprintf (stderr, "Invalid base directory: %s\n", optarg);
1629           return (4);
1630         }
1631       }
1632       break;
1633
1634       case 'p':
1635       {
1636         if (config_pid_file != NULL)
1637           free (config_pid_file);
1638         config_pid_file = strdup (optarg);
1639         if (config_pid_file == NULL)
1640         {
1641           fprintf (stderr, "read_options: strdup failed.\n");
1642           return (3);
1643         }
1644       }
1645       break;
1646
1647       case 'h':
1648       case '?':
1649         printf ("RRDd %s  Copyright (C) 2008 Florian octo Forster\n"
1650             "\n"
1651             "Usage: rrdcached [options]\n"
1652             "\n"
1653             "Valid options are:\n"
1654             "  -l <address>  Socket address to listen to.\n"
1655             "  -w <seconds>  Interval in which to write data.\n"
1656             "  -f <seconds>  Interval in which to flush dead data.\n"
1657             "  -p <file>     Location of the PID-file.\n"
1658             "  -b <dir>      Base directory to change to.\n"
1659             "\n"
1660             "For more information and a detailed description of all options "
1661             "please refer\n"
1662             "to the rrdcached(1) manual page.\n",
1663             VERSION);
1664         status = -1;
1665         break;
1666     } /* switch (option) */
1667   } /* while (getopt) */
1668
1669   return (status);
1670 } /* }}} int read_options */
1671
1672 int main (int argc, char **argv)
1673 {
1674   int status;
1675
1676   status = read_options (argc, argv);
1677   if (status != 0)
1678   {
1679     if (status < 0)
1680       status = 0;
1681     return (status);
1682   }
1683
1684   status = daemonize ();
1685   if (status == 1)
1686   {
1687     struct sigaction sigchld;
1688
1689     memset (&sigchld, 0, sizeof (sigchld));
1690     sigchld.sa_handler = SIG_IGN;
1691     sigaction (SIGCHLD, &sigchld, NULL);
1692
1693     return (0);
1694   }
1695   else if (status != 0)
1696   {
1697     fprintf (stderr, "daemonize failed, exiting.\n");
1698     return (1);
1699   }
1700
1701   listen_thread_main (NULL);
1702
1703   cleanup ();
1704
1705   return (0);
1706 } /* int main */
1707
1708 /*
1709  * vim: set sw=2 sts=2 ts=8 et fdm=marker :
1710  */