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