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