server.c

快速开发

/*
 * Portions created by SGI are Copyright (C) 2000 Silicon Graphics, Inc.
 * All Rights Reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met: 
 *
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer. 
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 * 3. Neither the name of Silicon Graphics, Inc. nor the names of its
 *    contributors may be used to endorse or promote products derived from
 *    this software without specific prior written permission. 
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 * ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
 * HOLDERS AND CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED
 * TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
 * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
 * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <time.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/socket.h>
#include <sys/wait.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <fcntl.h>
#include <signal.h>
#include <pwd.h>
#include "st.h"


/******************************************************************
 * Server configuration parameters
 */

/* Log files */
#define PID_FILE    "pid"
#define ERRORS_FILE "errors"
#define ACCESS_FILE "access"

/* Default server port */
#define SERV_PORT_DEFAULT 8000

/* Socket listen queue size */
#define LISTENQ_SIZE_DEFAULT 256

/* Max number of listening sockets ("hardware virtual servers") */
#define MAX_BIND_ADDRS 16

/* Max number of "spare" threads per process per socket */
#define MAX_WAIT_THREADS_DEFAULT 8

/* Number of file descriptors needed to handle one client session */
#define FD_PER_THREAD 2

/* Access log buffer flushing interval (in seconds) */
#define ACCLOG_FLUSH_INTERVAL 30

/* Request read timeout (in seconds) */
#define REQUEST_TIMEOUT 30


/******************************************************************
 * Global data
 */

struct socket_info {
  st_netfd_t nfd;               /* Listening socket                     */
  char *addr;                   /* Bind address                         */
  int port;                     /* Port                                 */
  int wait_threads;             /* Number of threads waiting to accept  */
  int busy_threads;             /* Number of threads processing request */
  int rqst_count;               /* Total number of processed requests   */
} srv_socket[MAX_BIND_ADDRS];   /* Array of listening sockets           */

static int sk_count = 0;        /* Number of listening sockets          */

static int vp_count = 0;        /* Number of server processes (VPs)     */
static pid_t *vp_pids;          /* Array of VP pids                     */

static int my_index = -1;       /* Current process index */
static pid_t my_pid = -1;       /* Current process pid   */

static st_netfd_t sig_pipe[2];  /* Signal pipe           */

/*
 * Configuration flags/parameters
 */
static int interactive_mode = 0;
static int serialize_accept = 0;
static int log_access       = 0;
static char *logdir     = NULL;
static char *username   = NULL;
static int listenq_size = LISTENQ_SIZE_DEFAULT;
static int errfd        = STDERR_FILENO;

/*
 * Thread throttling parameters (all numbers are per listening socket).
 * Zero values mean use default.
 */
static int max_threads = 0;       /* Max number of threads         */
static int max_wait_threads = 0;  /* Max number of "spare" threads */
static int min_wait_threads = 2;  /* Min number of "spare" threads */


/******************************************************************
 * Useful macros
 */

#ifndef INADDR_NONE
#define INADDR_NONE 0xffffffff
#endif

#define SEC2USEC(s) ((s)*1000000LL)

#define WAIT_THREADS(i)  (srv_socket[i].wait_threads)
#define BUSY_THREADS(i)  (srv_socket[i].busy_threads)
#define TOTAL_THREADS(i) (WAIT_THREADS(i) + BUSY_THREADS(i))
#define RQST_COUNT(i)    (srv_socket[i].rqst_count)


/******************************************************************
 * Forward declarations
 */

static void usage(const char *progname);
static void parse_arguments(int argc, char *argv[]);
static void start_daemon(void);
static void set_thread_throttling(void);
static void create_listeners(void);
static void change_user(void);
static void open_log_files(void);
static void start_processes(void);
static void wdog_sighandler(int signo);
static void child_sighandler(int signo);
static void install_sighandlers(void);
static void start_threads(void);
static void *process_signals(void *arg);
static void *flush_acclog_buffer(void *arg);
static void *handle_connections(void *arg);
static void dump_server_info(void);

static void Signal(int sig, void (*handler)(int));
static int cpu_count(void);

extern void handle_session(long srv_socket_index, st_netfd_t cli_nfd);
extern void load_configs(void);
extern void logbuf_open(void);
extern void logbuf_flush(void);
extern void logbuf_close(void);

/* Error reporting functions defined in the error.c file */
extern void err_sys_report(int fd, const char *fmt, ...);
extern void err_sys_quit(int fd, const char *fmt, ...);
extern void err_sys_dump(int fd, const char *fmt, ...);
extern void err_report(int fd, const char *fmt, ...);
extern void err_quit(int fd, const char *fmt, ...);


/*
 * General server example: accept a client connection and do something.
 * This program just outputs a short HTML page, but can be easily adapted
 * to do other things.
 *
 * This server creates a constant number of processes ("virtual processors"
 * or VPs) and replaces them when they die. Each virtual processor manages
 * its own independent set of state threads (STs), the number of which varies
 * with load against the server. Each state thread listens to exactly one
 * listening socket. The initial process becomes the watchdog, waiting for
 * children (VPs) to die or for a signal requesting termination or restart.
 * Upon receiving a restart signal (SIGHUP), all VPs close and then reopen
 * log files and reload configuration. All currently active connections remain
 * active. It is assumed that new configuration affects only request
 * processing and not the general server parameters such as number of VPs,
 * thread limits, bind addresses, etc. Those are specified as command line
 * arguments, so the server has to be stopped and then started again in order
 * to change them.
 *
 * Each state thread loops processing connections from a single listening
 * socket. Only one ST runs on a VP at a time, and VPs do not share memory,
 * so no mutual exclusion locking is necessary on any data, and the entire
 * server is free to use all the static variables and non-reentrant library
 * functions it wants, greatly simplifying programming and debugging and
 * increasing performance (for example, it is safe to ++ and -- all global
 * counters or call inet_ntoa(3) without any mutexes). The current thread on
 * each VP maintains equilibrium on that VP, starting a new thread or
 * terminating itself if the number of spare threads exceeds the lower or
 * upper limit.
 *
 * All I/O operations on sockets must use the State Thread library's I/O
 * functions because only those functions prevent blocking of the entire VP
 * process and perform state thread scheduling.
 */
int main(int argc, char *argv[])
{
  /* Parse command-line options */
  parse_arguments(argc, argv);

  /* Allocate array of server pids */
  if ((vp_pids = calloc(vp_count, sizeof(pid_t))) == NULL)
    err_sys_quit(errfd, "ERROR: calloc failed");

  /* Start the daemon */
  if (!interactive_mode)
    start_daemon();

  /* Initialize the ST library */
  if (st_init() < 0)
    err_sys_quit(errfd, "ERROR: initialization failed: st_init");

  /* Set thread throttling parameters */
  set_thread_throttling();

  /* Create listening sockets */
  create_listeners();

  /* Change the user */
  if (username)
    change_user();

  /* Open log files */
  open_log_files();

  /* Start server processes (VPs) */
  start_processes();

  /* Turn time caching on */
  st_timecache_set(1);

  /* Install signal handlers */
  install_sighandlers();

  /* Load configuration from config files */
  load_configs();

  /* Start all threads */
  start_threads();

  /* Become a signal processing thread */
  process_signals(NULL);

  /* NOTREACHED */
  return 1;
}


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

static void usage(const char *progname)
{
  fprintf(stderr, "Usage: %s -l <log_directory> [<options>]\n\n"
	  "Possible options:\n\n"
	  "\t-b <host>:<port>        Bind to specified address. Multiple"
	  " addresses\n"
	  "\t                        are permitted.\n"
	  "\t-p <num_processes>      Create specified number of processes.\n"
	  "\t-t <min_thr>:<max_thr>  Specify thread limits per listening"
	  " socket\n"
	  "\t                        across all processes.\n"
	  "\t-u <user>               Change server's user id to specified"
	  " value.\n"
	  "\t-q <backlog>            Set max length of pending connections"
	  " queue.\n"
	  "\t-a                      Enable access logging.\n"
	  "\t-i                      Run in interactive mode.\n"
	  "\t-S                      Serialize all accept() calls.\n"
	  "\t-h                      Print this message.\n",
	  progname);
  exit(1);
}


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

static void parse_arguments(int argc, char *argv[])
{
  extern char *optarg;
  int opt;
  char *c;

  while ((opt = getopt(argc, argv, "b:p:l:t:u:q:aiSh")) != EOF) {
    switch (opt) {
    case 'b':
      if (sk_count >= MAX_BIND_ADDRS)
	err_quit(errfd, "ERROR: max number of bind addresses (%d) exceeded",
		 MAX_BIND_ADDRS);
      if ((c = strdup(optarg)) == NULL)
	err_sys_quit(errfd, "ERROR: strdup");
      srv_socket[sk_count++].addr = c;
      break;
    case 'p':
      vp_count = atoi(optarg);
      if (vp_count < 1)
	err_quit(errfd, "ERROR: invalid number of processes: %s", optarg);
      break;
    case 'l':
      logdir = optarg;
      break;
    case 't':
      max_wait_threads = (int) strtol(optarg, &c, 10);
      if (*c++ == ':')
	max_threads = atoi(c);
      if (max_wait_threads < 0 || max_threads < 0)
	err_quit(errfd, "ERROR: invalid number of threads: %s", optarg);
      break;
    case 'u':
      username = optarg;
      break;
    case 'q':
      listenq_size = atoi(optarg);
      if (listenq_size < 1)
	err_quit(errfd, "ERROR: invalid listen queue size: %s", optarg);
      break;
    case 'a':
      log_access = 1;
      break;
    case 'i':
      interactive_mode = 1;
      break;
    case 'S':
      /*
       * Serialization decision is tricky on some platforms. For example,
       * Solaris 2.6 and above has kernel sockets implementation, so supposedly
       * there is no need for serialization. The ST library may be compiled
       * on one OS version, but used on another, so the need for serialization
       * should be determined at run time by the application. Since it's just
       * an example, the serialization decision is left up to user.
       * Only on platforms where the serialization is never needed on any OS
       * version st_netfd_serialize_accept() is a no-op.
       */
      serialize_accept = 1;
      break;
    case 'h':
    case '?':
      usage(argv[0]);
    }
  }

  if (logdir == NULL && !interactive_mode) {
    err_report(errfd, "ERROR: logging directory is required\n");
    usage(argv[0]);
  }

  if (getuid() == 0 && username == NULL)
    err_report(errfd, "WARNING: running as super-user!");

  if (vp_count == 0 && (vp_count = cpu_count()) < 1)
    vp_count = 1;

  if (sk_count == 0) {
    sk_count = 1;
    srv_socket[0].addr = "0.0.0.0";
  }
}


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

static void start_daemon(void)
{
  pid_t pid;

  /* Start forking */
  if ((pid = fork()) < 0)
    err_sys_quit(errfd, "ERROR: fork");
  if (pid > 0)
    exit(0);                  /* parent */

  /* First child process */
  setsid();                   /* become session leader */

  if ((pid = fork()) < 0)
    err_sys_quit(errfd, "ERROR: fork");
  if (pid > 0)                /* first child */
    exit(0);

  umask(022);

  if (chdir(logdir) < 0)
    err_sys_quit(errfd, "ERROR: can't change directory to %s: chdir", logdir);
}


/******************************************************************
 * For simplicity, the minimal size of thread pool is considered
 * as a maximum number of spare threads (max_wait_threads) that
 * will be created upon server startup. The pool size can grow up
 * to the max_threads value. Note that this is a per listening
 * socket limit. It is also possible to limit the total number of
 * threads for all sockets rather than impose a per socket limit.
 */

static void set_thread_throttling(void)
{
  /*
   * Calculate total values across all processes.
   * All numbers are per listening socket.
   */
  if (max_wait_threads == 0)
    max_wait_threads = MAX_WAIT_THREADS_DEFAULT * vp_count;
  /* Assuming that each client session needs FD_PER_THREAD file descriptors */
  if (max_threads == 0)
    max_threads = (st_getfdlimit() * vp_count) / FD_PER_THREAD / sk_count;
  if (max_wait_threads > max_threads)
    max_wait_threads = max_threads;

  /*
   * Now calculate per-process values.
   */
  if (max_wait_threads % vp_count)
    max_wait_threads = max_wait_threads / vp_count + 1;
  else
    max_wait_threads = max_wait_threads / vp_count;
  if (max_threads % vp_count)
    max_threads = max_threads / vp_count + 1;
  else
    max_threads = max_threads / vp_count;

  if (min_wait_threads > max_wait_threads)
    min_wait_threads = max_wait_threads;
}


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

static void create_listeners(void)
{
  int i, n, sock;
  char *c;
  struct sockaddr_in serv_addr;
  struct hostent *hp;
  short port;

  for (i = 0; i < sk_count; i++) {
    port = 0;
    if ((c = strchr(srv_socket[i].addr, ':')) != NULL) {
      *c++ = '\0';
      port = (short) atoi(c);
    }
    if (srv_socket[i].addr[0] == '\0')
      srv_socket[i].addr = "0.0.0.0";
    if (port == 0)
      port = SERV_PORT_DEFAULT;

    /* Create server socket */
    if ((sock = socket(PF_INET, SOCK_STREAM, 0)) < 0)
      err_sys_quit(errfd, "ERROR: can't create socket: socket");
    n = 1;
    if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, (char *)&n, sizeof(n)) < 0)
      err_sys_quit(errfd, "ERROR: can't set SO_REUSEADDR: setsockopt");
    memset(&serv_addr, 0, sizeof(serv_addr));
    serv_addr.sin_family = AF_INET;
    serv_addr.sin_port = htons(port);
    serv_addr.sin_addr.s_addr = inet_addr(srv_socket[i].addr);
    if (serv_addr.sin_addr.s_addr == INADDR_NONE) {
      /* not dotted-decimal */
      if ((hp = gethostbyname(srv_socket[i].addr)) == NULL)
	err_quit(errfd, "ERROR: can't resolve address: %s",
		 srv_socket[i].addr);
      memcpy(&serv_addr.sin_addr, hp->h_addr, hp->h_length);
    }
    srv_socket[i].port = port;

    /* Do bind and listen */
    if (bind(sock, (struct sockaddr *)&serv_addr, sizeof(serv_addr)) < 0)
      err_sys_quit(errfd, "ERROR: can't bind to address %s, port %d",
		   srv_socket[i].addr, port);
    if (listen(sock, listenq_size) < 0)
      err_sys_quit(errfd, "ERROR: listen");

    /* Create file descriptor object from OS socket */
    if ((srv_socket[i].nfd = st_netfd_open_socket(sock)) == NULL)
      err_sys_quit(errfd, "ERROR: st_netfd_open_socket");
    /*
     * On some platforms (e.g. IRIX, Linux) accept() serialization is never
     * needed for any OS version.  In that case st_netfd_serialize_accept()
     * is just a no-op. Also see the comment above.
     */
    if (serialize_accept && st_netfd_serialize_accept(srv_socket[i].nfd) < 0)
      err_sys_quit(errfd, "ERROR: st_netfd_serialize_accept");
  }
}


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

static void change_user(void)
{
  struct passwd *pw;

  if ((pw = getpwnam(username)) == NULL)
    err_quit(errfd, "ERROR: can't find user '%s': getpwnam failed", username);

  if (setgid(pw->pw_gid) < 0)
    err_sys_quit(errfd, "ERROR: can't change group id: setgid");
  if (setuid(pw->pw_uid) < 0)