src/lwes-journaller.c

   1 /*======================================================================*
   2  * Copyright (C) 2008 Light Weight Event System                         *
   3  * All rights reserved.                                                 *
   4  *                                                                      *
   5  * This program is free software; you can redistribute it and/or modify *
   6  * it under the terms of the GNU General Public License as published by *
   7  * the Free Software Foundation; either version 2 of the License, or    *
   8  * (at your option) any later version.                                  *
   9  *                                                                      *
  10  * This program is distributed in the hope that it will be useful,      *
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of       *
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the        *
  13  * GNU General Public License for more details.                         *
  14  *                                                                      *
  15  * You should have received a copy of the GNU General Public License    *
  16  * along with this program; if not, write to the Free Software          *
  17  * Foundation, Inc., 51 Franklin Street, Fifth Floor,                   *
  18  * Boston, MA 02110-1301 USA.                                           *
  19  *======================================================================*/
  20
  21 #include "config.h"
  22
  23 #include "log.h"
  24 #include "opt.h"
  25 #include "perror.h"
  26 #include "queue_to_journal.h"
  27 #include "sig.h"
  28 #include "xport_to_queue.h"
  29
  30 #if HAVE_PTHREAD_H
  31 #include <pthread.h>
  32 #endif
  33
  34 #include <assert.h>
  35 #include <fcntl.h>
  36 #include <limits.h>
  37 #include <signal.h>
  38 #include <stdio.h>
  39 #include <stdlib.h>
  40 #include <strings.h>
  41 #include <unistd.h>
  42
  43 #include <netinet/in.h>
  44
  45 #include <sys/stat.h>
  46 #include <sys/types.h>
  47 #include <sys/wait.h>
  48
  49 #define NREADERS_MAX 5
  50
  51 #ifndef NAME_MAX
  52 #define NAME_MAX 255
  53 #endif
  54
  55 #ifndef INADDR_NONE
  56 #define INADDR_NONE 0
  57 #endif
  58
  59 /* Start a new process via fork() and exec() Return the process ID of
  60    the new child process. */
  61
  62 static int start_program(const char* program, const char* argv[])
  63 {
  64   int     i;
  65   int     pid;
  66   char*   fn;
  67   char    prg[PATH_MAX + NAME_MAX + 1];
  68   char    exec_path[PATH_MAX + 1];
  69
  70   /* Make sure the path to our executable is within reasonable
  71      limits. */
  72
  73   if ( strlen(argv[0]) > PATH_MAX )
  74     {
  75       LOG_ER("Program path too long: %s.\n", argv[0]);
  76       exit(EXIT_FAILURE);
  77     }
  78   if ( strlen(program) > NAME_MAX )
  79     {
  80       LOG_ER("Program name too long: %s.\n", program);
  81       exit(EXIT_FAILURE);
  82     }
  83
  84   /* Create a child process. */
  85   if ( (pid = fork()) < 0 )
  86     {
  87       PERROR("fork()");
  88       exit(EXIT_FAILURE);
  89     }
  90
  91   /* If we're the parent process, we're done -- the child process
  92      continues on. */
  93   if ( pid )
  94     {
  95       return pid;
  96     }
  97
  98   /* Get the path to our executable, use it to exec() other programs
  99      from the same directory. */
 100
 101   if ( 0 != (fn = rindex(argv[0], '/')) )
 102     {
 103       strncpy(exec_path, argv[0], fn - argv[0] + 1);
 104       exec_path[fn - argv[0] + 1] = '\0';
 105     }
 106   else
 107     {
 108       exec_path[0] = '\0';
 109     }
 110   snprintf(prg, sizeof(prg), "%s%s", exec_path, program);
 111
 112   /* Use new program name.  Make sure to mess with argv[0] after the
 113      fork() call, so as not to bugger up the calling processes
 114      argv[0]. */
 115
 116   argv[0] = prg;
 117
 118   LOG_PROG("About to execvp(\"%s\",\n", prg);
 119   for ( i=0; argv[i]; ++i )
 120     {
 121       LOG_PROG("  argv[%d] == \"%s\"\n", i, argv[i]);
 122     }
 123   LOG_PROG("  argv[%d] == NULL\n\n", i);
 124
 125   /* Replace our current process image with the new one. */
 126   execvp(prg, (char** const)argv);
 127
 128   /* Any return is an error. */
 129   PERROR("execvp()");
 130   LOG_ER("The execvp() function returned, exiting.\n");
 131   exit(EXIT_FAILURE);      /* Should never get to this anyway. */
 132 }
 133
 134
 135 /* Start the external processes, then loop to restart any program
 136    that dies. */
 137
 138 static void process_model(const char* argv[])
 139 {
 140   int i;
 141   int waiting_pid = -1;
 142   int queue_to_journal_pid = -1;
 143   int xport_to_queue_pid[NREADERS_MAX];
 144
 145   for ( i=0; i<NREADERS_MAX; ++i )
 146     {
 147       xport_to_queue_pid[i] = -1;
 148     }
 149
 150   while ( ! gbl_done )
 151     {
 152       int status;
 153
 154       if ((-1 == queue_to_journal_pid) || (waiting_pid == queue_to_journal_pid))
 155         {
 156           queue_to_journal_pid = start_program("queue_to_journal", argv);
 157         }
 158
 159       for ( i=0; i<arg_nreaders; ++i )
 160         {
 161           if ((-1 == xport_to_queue_pid[i]) ||
 162               (waiting_pid == xport_to_queue_pid[i]))
 163             {
 164               xport_to_queue_pid[i] = start_program("xport_to_queue", argv);
 165             }
 166         }
 167
 168       /* Wait and restart any program that exits. */
 169       waiting_pid = wait(&status);
 170
 171       /* If a signal interrupts the wait, we loop. */
 172       if ( (waiting_pid < 0) && (errno == EINTR) )
 173         {
 174           /* If we get a rotate, send it along to the process
 175            * that writes journals. */
 176           if ( gbl_rotate )
 177             {
 178               if ( -1 != queue_to_journal_pid )
 179                 {
 180                   LOG_PROG("Sending SIGHUP to queue_to_journal[%d] to "
 181                            "trigger log rotate.\n", queue_to_journal_pid);
 182                   kill(queue_to_journal_pid, SIGHUP);
 183                 }
 184               gbl_rotate = 0;
 185             }
 186           continue;
 187         }
 188
 189       if ( WIFEXITED(status) )
 190         {
 191           break;
 192         }
 193
 194       if ( WIFSIGNALED(status) )
 195         {
 196           /* If any child process exits abnormally, log it and restart. */
 197           const char* program = "unknown";
 198
 199           if ( waiting_pid == queue_to_journal_pid )
 200             {
 201               program = "queue_to_journal";
 202             }
 203
 204           for ( i=0; i<arg_nreaders; ++i )
 205             {
 206               if ( waiting_pid == xport_to_queue_pid[i] )
 207                 {
 208                   program = "xport_to_queue";
 209                 }
 210             }
 211
 212           log_msg(LOG_WARNING, __FILE__, __LINE__,
 213                   "Our %s process exited (pid=%d) with "
 214                   "signal %d, restarting.\n",
 215                   program, waiting_pid, WTERMSIG(status));
 216         }
 217     }
 218
 219   LOG_INF("Shutting down.");
 220
 221   /* We give the children a chance to run before we kill them. */
 222   sleep(1);
 223
 224   /* We are done, send a quit signal to our children. */
 225   kill(0, SIGQUIT);
 226   while ( wait(0) > 0 )   /* Wait for children. */
 227     ;
 228   if ( errno != ECHILD && errno != EINTR )
 229     {
 230       PERROR("wait");
 231     }
 232 }
 233
 234
 235 #if HAVE_PTHREAD_H
 236 static void thread_model()
 237 {
 238   int i;
 239   pthread_t queue_to_journal_tid;
 240   pthread_t xport_to_queue_tid[NREADERS_MAX];
 241
 242   static pthread_attr_t m_pthread_attr ;
 243   // Start thread attributes with pthread FIFO policy (default would be OTHER)
 244   pthread_attr_init(&m_pthread_attr) ;
 245   pthread_attr_setdetachstate(&m_pthread_attr, PTHREAD_CREATE_JOINABLE);
 246   if ( arg_rt )
 247     {
 248       pthread_attr_setschedpolicy(&m_pthread_attr, SCHED_FIFO) ;
 249     }
 250
 251   if ( pthread_create(&queue_to_journal_tid,
 252                       &m_pthread_attr, queue_to_journal, NULL) )
 253     {
 254       PERROR("pthread_create(&queue_to_journal_tid, NULL, queue_to_journal, NULL)");
 255     }
 256   LOG_PROG("queue_to_journal_tid == %d\n", queue_to_journal_tid);
 257
 258   for ( i=0; i<arg_nreaders; ++i )
 259     {
 260       if ( pthread_create(&xport_to_queue_tid[i], &m_pthread_attr,
 261                           xport_to_queue, NULL) )
 262         {
 263           PERROR("pthread_create(&xport_to_queue_tid[i], NULL, xport_to_queue, NULL)");
 264         }
 265         {
 266           int schedpolicy ;
 267           pthread_attr_getschedpolicy((pthread_attr_t*)&m_pthread_attr,
 268                                       &schedpolicy) ;
 269           switch ( schedpolicy )
 270             {
 271             case SCHED_OTHER: LOG_ER("pthread_policy=SCHED_OTHER\n") ; break ;
 272             case SCHED_FIFO: LOG_ER("pthread_policy=SCHED_FIFO") ; break ;
 273             case SCHED_RR: LOG_ER("pthread_policy=SCHED_RR") ; break ;
 274                             //case  SCHED_MIN   = SCHED_OTHER,
 275                             //case  SCHED_MAX   = SCHED_RR
 276             default: LOG_ER("pthread_policy=unknown") ; break ;
 277             }
 278
 279           LOG_PROG("xport_to_queue_tid[%d] == %d\n",
 280                    i, xport_to_queue_tid[i]);
 281         }
 282     }
 283
 284   while ( ! gbl_done )
 285     {
 286       sleep(99);
 287     }
 288
 289   LOG_INF("Shutting down.");
 290
 291   /* All threads share a single gbl_done, so just interrupt them and
 292      wait for them to finish on their own. */
 293
 294   for ( i=0; i<arg_nreaders; ++i )
 295     {
 296       LOG_PROG("Sending SIGQUIT to xport_to_queue thread %d (%d).\n",
 297                i, xport_to_queue_tid);
 298       if ( pthread_kill(xport_to_queue_tid[i], SIGQUIT) )
 299         {
 300           PERROR("pthread_kill");
 301         }
 302     }
 303
 304   LOG_PROG("Sending SIGQUIT to queue_to_journal thread (%d).\n",
 305            queue_to_journal_tid);
 306   if ( pthread_kill(queue_to_journal_tid, SIGQUIT) )
 307     {
 308       switch ( errno )
 309         {
 310         default:
 311           PERROR("pthread_kill");
 312
 313         case ESRCH:   /* If the thread has already terminated. */
 314         case ECHILD:
 315           break;
 316         }
 317     }
 318
 319   for (i=0; i<arg_nreaders; ++i)
 320     {
 321       LOG_PROG("Joining xport_to_queue thread %d (%d).\n",
 322                i, xport_to_queue_tid[i]);
 323       if ( pthread_join(xport_to_queue_tid[i], NULL) )
 324         {
 325           PERROR("pthread_join");
 326         }
 327     }
 328
 329   LOG_PROG("Joining queue_to_journal thread (%d).\n", queue_to_journal_tid);
 330   if ( pthread_join(queue_to_journal_tid, NULL) )
 331     {
 332       PERROR("pthread_join");
 333     }
 334
 335   pthread_attr_destroy(&m_pthread_attr);
 336 }
 337 #else
 338 static void thread_model()
 339 {
 340   LOG_ER("No POSIX thread support on this platform.\n");
 341 }
 342 #endif
 343
 344 /* Return non-zero if addrstr is a multicast address. */
 345
 346 #include <netinet/in.h>
 347 #include <arpa/inet.h>
 348
 349 static int is_multicast(const char* addrstr)
 350 {
 351   struct in_addr addr;
 352
 353   if ( 0 == inet_aton(addrstr, &addr) )
 354     {
 355     return 0;
 356     }
 357
 358   return IN_MULTICAST(ntohl(addr.s_addr));
 359 }
 360
 361 static void do_fork()
 362 {
 363   switch ( fork() )
 364     {
 365     case 0:
 366       break;
 367
 368     case -1:
 369       PERROR("fork()");
 370       exit(EXIT_FAILURE);
 371
 372     default:
 373       exit(EXIT_SUCCESS);
 374   }
 375 }
 376
 377 static void daemonize()
 378 {
 379   int fd, fdlimit;
 380   const char *chdir_root = "/";
 381
 382   do_fork();
 383   if ( setsid() < 0 )
 384     {
 385     PERROR("setsid()");
 386     exit(EXIT_FAILURE);
 387   }
 388   do_fork();
 389
 390   umask(0);
 391   if ( chdir(chdir_root) < 0)
 392     {
 393       LOG_ER("Unable to chdir(\"%s\"): %s\n", chdir_root, strerror(errno));
 394       exit(EXIT_FAILURE);
 395     }
 396
 397   /* Close all open files. */
 398   fdlimit = sysconf (_SC_OPEN_MAX);
 399   for ( fd=0; fd<fdlimit; ++fd )
 400     {
 401     close(fd);
 402     }
 403
 404   /* Open 0, 1 and 2. */
 405   open("/dev/null", O_RDWR);
 406   if ( dup(0) != 1 )
 407     {
 408       PERROR("Unable to dup() to replace stdout: %s\n");
 409       exit(EXIT_FAILURE);
 410     }
 411   if ( dup(0) != 2 )
 412     {
 413       PERROR("Unable to dup() to replace stderr: %s\n");
 414       exit(EXIT_FAILURE);
 415     }
 416 }
 417
 418
 419 /* Write the process ID of this process into a file so we may be
 420    easily signaled. */
 421
 422 static void write_pid_file()
 423 {
 424   /* Write out the PID into the PID file. */
 425   if ( arg_pid_file )
 426     {
 427       FILE* pidfp = fopen(arg_pid_file, "w");
 428       if ( ! pidfp )
 429         {
 430           LOG_ER("Can't open PID file \"%s\"\n", arg_pid_file);
 431         }
 432       else
 433         {
 434           /* This obnoxiousness is to avoid compiler warnings that we'd get
 435              compiling on systems with long or int pid_t. */
 436           pid_t pid = getpid();
 437           if ( sizeof(pid_t) == sizeof(long) )
 438             {
 439               long x = pid;
 440               fprintf(pidfp, "%ld\n", x);
 441             }
 442           else
 443             {
 444               int x = pid;
 445               fprintf(pidfp, "%d\n", x);
 446             }
 447           fclose(pidfp);
 448         }
 449     }
 450 }
 451
 452
 453 /* Delete the PID file, if one was configured. */
 454
 455 static void delete_pid_file()
 456 {
 457   if ( arg_pid_file )
 458     {
 459       if ( unlink ( arg_pid_file ) < 0 )
 460         {
 461           PERROR("Unable to delete pid file");
 462         }
 463     }
 464 }
 465
 466
 467 int main(int argc, const char* argv[])
 468 {
 469   char _buf[100] ; // for log messages
 470   char _progver[30] ;
 471
 472   process_options(argc, argv);
 473
 474   if ( ! arg_nodaemonize )
 475     {
 476       daemonize();
 477     }
 478
 479   strcpy(_progver, "lwes-journaller-") ;
 480   strcat(_progver, ""VERSION"") ;
 481
 482   strcpy(_buf, "Initializing - ") ;
 483   strcat(_buf, _progver) ;
 484   strcat(_buf, "\n") ;
 485   LOG_INF(_buf);
 486
 487   install_signal_handlers();
 488
 489   write_pid_file();
 490
 491   /* Check options. */
 492   if ( is_multicast(arg_ip) )
 493     {
 494       if ( ! arg_join_group )
 495         {
 496           LOG_WARN("Using multi-cast address without joining group.\n");
 497         }
 498       if ( arg_nreaders > 1 )
 499         {
 500           LOG_WARN("Multiple reader threads can't be used with multi-cast "
 501                    "addresses, will use a single thread instead.\n");
 502           arg_nreaders = 1;
 503         }
 504     }
 505   else
 506     {
 507       if ( arg_join_group )
 508         {
 509           LOG_WARN("Can't join group with uni-cast address.\n");
 510         }
 511     }
 512   if ( arg_nreaders > NREADERS_MAX )
 513     {
 514       LOG_WARN("Too many reader threads specified (%d) reducing to %d.\n",
 515                arg_nreaders, NREADERS_MAX);
 516       arg_nreaders = NREADERS_MAX;
 517     }
 518   if ( arg_njournalls > 30 )
 519     {
 520       LOG_WARN("suspiciously large (%d) number of journals.",
 521                arg_njournalls);
 522     }
 523
 524   strcpy(_buf, "Starting up - ") ;
 525   strcat(_buf, _progver) ;
 526   strcat(_buf, "\n") ;
 527   LOG_INF(_buf);
 528
 529   if ( strcmp(arg_proc_type, ARG_PROCESS) == 0 )
 530     {
 531       process_model(argv);
 532     }
 533
 534   if ( strcmp(arg_proc_type, ARG_THREAD) == 0 )
 535     {
 536       thread_model();
 537     }
 538
 539   strcpy(_buf, "Normal shutdown complete - ") ;
 540   strcat(_buf, _progver) ;
 541   strcat(_buf, "\n") ;
 542   LOG_INF(_buf);
 543
 544   delete_pid_file();
 545
 546   return 0;
 547 }