server.c

快速开发

    err_sys_quit(errfd, "ERROR: can't change user id: setuid");

  err_report(errfd, "INFO: changed process user id to '%s'", username);
}


/******************************************************************/

static void open_log_files(void)
{
  int fd;
  char str[32];

  if (interactive_mode)
    return;

  /* Open access log */
  if (log_access)
    logbuf_open();

  /* Open and write pid to pid file */
  if ((fd = open(PID_FILE, O_CREAT | O_WRONLY | O_TRUNC, 0644)) < 0)
    err_sys_quit(errfd, "ERROR: can't open pid file: open");
  sprintf(str, "%d\n", (int)getpid());
  if (write(fd, str, strlen(str)) != strlen(str))
    err_sys_quit(errfd, "ERROR: can't write to pid file: write");
  close(fd);

  /* Open error log file */
  if ((fd = open(ERRORS_FILE, O_CREAT | O_WRONLY | O_APPEND, 0644)) < 0)
    err_sys_quit(errfd, "ERROR: can't open error log file: open");
  errfd = fd;

  err_report(errfd, "INFO: starting the server...");
}


/******************************************************************/

static void start_processes(void)
{
  int i, status;
  pid_t pid;
  sigset_t mask, omask;

  if (interactive_mode) {
    my_index = 0;
    my_pid = getpid();
    return;
  }

  for (i = 0; i < vp_count; i++) {
    if ((pid = fork()) < 0) {
      err_sys_report(errfd, "ERROR: can't create process: fork");
      if (i == 0)
	exit(1);
      err_report(errfd, "WARN: started only %d processes out of %d", i,
		 vp_count);
      vp_count = i;
      break;
    }
    if (pid == 0) {
      my_index = i;
      my_pid = getpid();
      /* Child returns to continue in main() */
      return;
    }
    vp_pids[i] = pid;
  }

  /*
   * Parent process becomes a "watchdog" and never returns to main().
   */

  /* Install signal handlers */
  Signal(SIGTERM, wdog_sighandler);  /* terminate */
  Signal(SIGHUP,  wdog_sighandler);  /* restart   */
  Signal(SIGUSR1, wdog_sighandler);  /* dump info */

  /* Now go to sleep waiting for a child termination or a signal */
  for ( ; ; ) {
    if ((pid = wait(&status)) < 0) {
      if (errno == EINTR)
	continue;
      err_sys_quit(errfd, "ERROR: watchdog: wait");
    }
    /* Find index of the exited child */
    for (i = 0; i < vp_count; i++) {
      if (vp_pids[i] == pid)
	break;
    }

    /* Block signals while printing and forking */
    sigemptyset(&mask);
    sigaddset(&mask, SIGTERM);
    sigaddset(&mask, SIGHUP);
    sigaddset(&mask, SIGUSR1);
    sigprocmask(SIG_BLOCK, &mask, &omask);

    if (WIFEXITED(status))
      err_report(errfd, "WARN: watchdog: process %d (pid %d) exited"
		 " with status %d", i, pid, WEXITSTATUS(status));
    else if (WIFSIGNALED(status))
      err_report(errfd, "WARN: watchdog: process %d (pid %d) terminated"
		 " by signal %d", i, pid, WTERMSIG(status));
    else if (WIFSTOPPED(status))
      err_report(errfd, "WARN: watchdog: process %d (pid %d) stopped"
		 " by signal %d", i, pid, WSTOPSIG(status));
    else
      err_report(errfd, "WARN: watchdog: process %d (pid %d) terminated:"
		 " unknown termination reason", i, pid);

    /* Fork another VP */
    if ((pid = fork()) < 0) {
      err_sys_report(errfd, "ERROR: watchdog: can't create process: fork");
    } else if (pid == 0) {
      my_index = i;
      my_pid = getpid();
      /* Child returns to continue in main() */
      return;
    }
    vp_pids[i] = pid;

    /* Restore the signal mask */
    sigprocmask(SIG_SETMASK, &omask, NULL);
  }
}


/******************************************************************/

static void wdog_sighandler(int signo)
{
  int i, err;

  /* Save errno */
  err = errno;
  /* Forward the signal to all children */
  for (i = 0; i < vp_count; i++) {
    if (vp_pids[i] > 0)
      kill(vp_pids[i], signo);
  }
  /*
   * It is safe to do pretty much everything here because process is
   * sleeping in wait() which is async-safe.
   */
  switch (signo) {
  case SIGHUP:
    err_report(errfd, "INFO: watchdog: caught SIGHUP");
    /* Reopen log files - needed for log rotation */
    if (log_access) {
      logbuf_close();
      logbuf_open();
    }
    close(errfd);
    if ((errfd = open(ERRORS_FILE, O_CREAT | O_WRONLY | O_APPEND, 0644)) < 0)
      err_sys_quit(STDERR_FILENO, "ERROR: watchdog: open");
    break;
  case SIGTERM:
    /* Non-graceful termination */
    err_report(errfd, "INFO: watchdog: caught SIGTERM, terminating");
    unlink(PID_FILE);
    exit(0);
  case SIGUSR1:
    err_report(errfd, "INFO: watchdog: caught SIGUSR1");
    break;
  default:
    err_report(errfd, "INFO: watchdog: caught signal %d", signo);
  }
  /* Restore errno */
  errno = err;
}


/******************************************************************/

static void install_sighandlers(void)
{
  sigset_t mask;
  int p[2];

  /* Create signal pipe */
  if (pipe(p) < 0)
    err_sys_quit(errfd, "ERROR: process %d (pid %d): can't create"
		 " signal pipe: pipe", my_index, my_pid);
  if ((sig_pipe[0] = st_netfd_open(p[0])) == NULL ||
      (sig_pipe[1] = st_netfd_open(p[1])) == NULL)
    err_sys_quit(errfd, "ERROR: process %d (pid %d): can't create"
		 " signal pipe: st_netfd_open", my_index, my_pid);

  /* Install signal handlers */
  Signal(SIGTERM, child_sighandler);  /* terminate */
  Signal(SIGHUP,  child_sighandler);  /* restart   */
  Signal(SIGUSR1, child_sighandler);  /* dump info */

  /* Unblock signals */
  sigemptyset(&mask);
  sigaddset(&mask, SIGTERM);
  sigaddset(&mask, SIGHUP);
  sigaddset(&mask, SIGUSR1);
  sigprocmask(SIG_UNBLOCK, &mask, NULL);
}


/******************************************************************/

static void child_sighandler(int signo)
{
  int err, fd;

  err = errno;
  fd = st_netfd_fileno(sig_pipe[1]);

  /* write() is async-safe */
  if (write(fd, &signo, sizeof(int)) != sizeof(int))
    err_sys_quit(errfd, "ERROR: process %d (pid %d): child's signal"
		 " handler: write", my_index, my_pid);
  errno = err;
}


/******************************************************************
 * The "main" function of the signal processing thread.
 */

/* ARGSUSED */
static void *process_signals(void *arg)
{
  int signo;

  for ( ; ; ) {
    /* Read the next signal from the signal pipe */
    if (st_read(sig_pipe[0], &signo, sizeof(int),
     ST_UTIME_NO_TIMEOUT) != sizeof(int))
      err_sys_quit(errfd, "ERROR: process %d (pid %d): signal processor:"
		   " st_read", my_index, my_pid);

    switch (signo) {
    case SIGHUP:
      err_report(errfd, "INFO: process %d (pid %d): caught SIGHUP,"
		 " reloading configuration", my_index, my_pid);
      if (interactive_mode) {
	load_configs();
	break;
      }
      /* Reopen log files - needed for log rotation */
      if (log_access) {
	logbuf_flush();
	logbuf_close();
	logbuf_open();
      }
      close(errfd);
      if ((errfd = open(ERRORS_FILE, O_CREAT | O_WRONLY | O_APPEND, 0644)) < 0)
	err_sys_quit(STDERR_FILENO, "ERROR: process %d (pid %d): signal"
		     " processor: open", my_index, my_pid);
      /* Reload configuration */
      load_configs();
      break;
    case SIGTERM:
      /*
       * Terminate ungracefully since it is generally not known how long
       * it will take to gracefully complete all client sessions.
       */
      err_report(errfd, "INFO: process %d (pid %d): caught SIGTERM,"
		 " terminating", my_index, my_pid);
      if (log_access)
	logbuf_flush();
      exit(0);
    case SIGUSR1:
      err_report(errfd, "INFO: process %d (pid %d): caught SIGUSR1",
		 my_index, my_pid);
      /* Print server info to stderr */
      dump_server_info();
      break;
    default:
      err_report(errfd, "INFO: process %d (pid %d): caught signal %d",
		 my_index, my_pid, signo);
    }
  }

  /* NOTREACHED */
  return NULL;
}


/******************************************************************
 * The "main" function of the access log flushing thread.
 */

/* ARGSUSED */
static void *flush_acclog_buffer(void *arg)
{
  for ( ; ; ) {
    st_sleep(ACCLOG_FLUSH_INTERVAL);
    logbuf_flush();
  }

  /* NOTREACHED */
  return NULL;
}


/******************************************************************/

static void start_threads(void)
{
  long i, n;

  /* Create access log flushing thread */
  if (log_access && st_thread_create(flush_acclog_buffer, NULL, 0, 0) == NULL)
    err_sys_quit(errfd, "ERROR: process %d (pid %d): can't create"
		 " log flushing thread", my_index, my_pid);

  /* Create connections handling threads */
  for (i = 0; i < sk_count; i++) {
    err_report(errfd, "INFO: process %d (pid %d): starting %d threads"
	       " on %s:%d", my_index, my_pid, max_wait_threads,
	       srv_socket[i].addr, srv_socket[i].port);
    WAIT_THREADS(i) = 0;
    BUSY_THREADS(i) = 0;
    RQST_COUNT(i) = 0;
    for (n = 0; n < max_wait_threads; n++) {
      if (st_thread_create(handle_connections, (void *)i, 0, 0) != NULL)
	WAIT_THREADS(i)++;
      else
	err_sys_report(errfd, "ERROR: process %d (pid %d): can't create"
		       " thread", my_index, my_pid);
    }
    if (WAIT_THREADS(i) == 0)
      exit(1);
  }
}


/******************************************************************/

static void *handle_connections(void *arg)
{
  st_netfd_t srv_nfd, cli_nfd;
  struct sockaddr_in from;
  int fromlen;
  long i = (long) arg;

  srv_nfd = srv_socket[i].nfd;
  fromlen = sizeof(from);

  while (WAIT_THREADS(i) <= max_wait_threads) {
    cli_nfd = st_accept(srv_nfd, (struct sockaddr *)&from, &fromlen,
     ST_UTIME_NO_TIMEOUT);
    if (cli_nfd == NULL) {
      err_sys_report(errfd, "ERROR: can't accept connection: st_accept");
      continue;
    }
    /* Save peer address, so we can retrieve it later */
    st_netfd_setspecific(cli_nfd, &from.sin_addr, NULL);

    WAIT_THREADS(i)--;
    BUSY_THREADS(i)++;
    if (WAIT_THREADS(i) < min_wait_threads && TOTAL_THREADS(i) < max_threads) {
      /* Create another spare thread */
      if (st_thread_create(handle_connections, (void *)i, 0, 0) != NULL)
	WAIT_THREADS(i)++;
      else
	err_sys_report(errfd, "ERROR: process %d (pid %d): can't create"
		       " thread", my_index, my_pid);
    }

    handle_session(i, cli_nfd);

    st_netfd_close(cli_nfd);
    WAIT_THREADS(i)++;
    BUSY_THREADS(i)--;
  }

  WAIT_THREADS(i)--;
  return NULL;
}


/******************************************************************/

static void dump_server_info(void)
{
  char *buf;
  int i, len;

  if ((buf = malloc(sk_count * 512)) == NULL) {
    err_sys_report(errfd, "ERROR: malloc failed");
    return;
  }

  len = sprintf(buf, "\n\nProcess #%d (pid %d):\n", my_index, (int)my_pid);
  for (i = 0; i < sk_count; i++) {
    len += sprintf(buf + len, "\nListening Socket #%d:\n"
		   "-------------------------\n"
		   "Address                    %s:%d\n"
		   "Thread limits (min/max)    %d/%d\n"
		   "Waiting threads            %d\n"
		   "Busy threads               %d\n"
		   "Requests served            %d\n",
		   i, srv_socket[i].addr, srv_socket[i].port,
		   max_wait_threads, max_threads,
		   WAIT_THREADS(i), BUSY_THREADS(i), RQST_COUNT(i));
  }

  write(STDERR_FILENO, buf, len);
  free(buf);
}


/******************************************************************
 * Stubs
 */

/*
 * Session handling function stub. Just dumps small HTML page.
 */
void handle_session(long srv_socket_index, st_netfd_t cli_nfd)
{
  static char resp[] = "HTTP/1.0 200 OK\r\nContent-type: text/html\r\n"
                       "Connection: close\r\n\r\n<H2>It worked!</H2>\n";
  char buf[512];
  int n = sizeof(resp) - 1;
  struct in_addr *from = st_netfd_getspecific(cli_nfd);

  if (st_read(cli_nfd, buf, sizeof(buf), SEC2USEC(REQUEST_TIMEOUT)) < 0) {
    err_sys_report(errfd, "WARN: can't read request from %s: st_read",
		   inet_ntoa(*from));
    return;
  }
  if (st_write(cli_nfd, resp, n, ST_UTIME_NO_TIMEOUT) != n) {
    err_sys_report(errfd, "WARN: can't write response to %s: st_write",
		   inet_ntoa(*from));
    return;
  }

  RQST_COUNT(srv_socket_index)++;
}


/*
 * Configuration loading function stub.
 */
void load_configs(void)
{
  err_report(errfd, "INFO: process %d (pid %d): configuration loaded",
	     my_index, my_pid);
}


/*
 * Buffered access logging methods.
 * Note that stdio functions (fopen(3), fprintf(3), fflush(3), etc.) cannot
 * be used if multiple VPs are created since these functions can flush buffer
 * at any point and thus write only partial log record to disk.
 * Also, it is completely safe for all threads of the same VP to write to
 * the same log buffer without any mutex protection (one buffer per VP, of
 * course).
 */
void logbuf_open(void)
{

}


void logbuf_flush(void)
{

}


void logbuf_close(void)
{

}


/******************************************************************
 * Small utility functions
 */

static void Signal(int sig, void (*handler)(int))
{
  struct sigaction sa;

  sa.sa_handler = handler;
  sigemptyset(&sa.sa_mask);
  sa.sa_flags = 0;
  sigaction(sig, &sa, NULL);
}

static int cpu_count(void)
{
  int n;

#if defined (_SC_NPROCESSORS_ONLN)
  n = (int) sysconf(_SC_NPROCESSORS_ONLN);
#elif defined (_SC_NPROC_ONLN)
  n = (int) sysconf(_SC_NPROC_ONLN);
#elif defined (HPUX)
#include <sys/mpctl.h>
  n = mpctl(MPC_GETNUMSPUS, 0, 0);
#else
  n = -1;
  errno = ENOSYS;
#endif

  return n;
}

/******************************************************************/