external/srs
1
0
Fork 0
mirror of https://github.com/ossrs/srs.git synced 2025-03-09 15:49:59 +00:00
Code Issues Releases Wiki Activity
srs/trunk/3rdparty/st-srs/examples/server.c

1025 lines
30 KiB
C
Raw Blame History

/*
* 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 */
unsigned 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;
unsigned short port;
for (i = 0; i < sk_count; i++) {
port = 0;
if ((c = strchr(srv_socket[i].addr, ':')) != NULL) {
*c++ = '\0';
port = (unsigned 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 %hu",
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)
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:%u", 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:%u\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;
}
/******************************************************************/
Reference in a new issue View git blame Copy permalink