State_Threads实现多协程服务器

1. /*
2. * Portions created by SGI are Copyright (C) 2000 Silicon Graphics, Inc.
3. * All Rights Reserved.
4. *
5. * Redistribution and use in source and binary forms, with or without
6. * modification, are permitted provided that the following conditions
7. * are met:
8. *
9. * 1. Redistributions of source code must retain the above copyright
10. *    notice, this list of conditions and the following disclaimer.
11. * 2. Redistributions in binary form must reproduce the above copyright
12. *    notice, this list of conditions and the following disclaimer in the
13. *    documentation and/or other materials provided with the distribution.
14. * 3. Neither the name of Silicon Graphics, Inc. nor the names of its
15. *    contributors may be used to endorse or promote products derived from
16. *    this software without specific prior written permission.
17. *
18. * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
19. * ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
20. * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
21. * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
22. * HOLDERS AND CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
23. * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED
24. * TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
25. * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
26. * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
27. * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
28. * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
29. */
31. 系统socket编成，listen后在accept阻塞监听：[连接](https://blog.csdn.net/qq_36784975/article/details/88768320?ops_request_misc=%257B%2522request%255Fid%2522%253A%25220BE0E468-2E50-442E-B409-280726303185%2522%252C%2522scm%2522%253A%252220140713.130102334..%2522%257D&request_id=0BE0E468-2E50-442E-B409-280726303185&biz_id=0&utm_medium=distribute.pc_search_result.none-task-blog-2~all~sobaiduend~default-1-88768320-null-null.142%5Ev100%5Epc_search_result_base4&utm_term=socket%28PF_INET,%20SOCK_STREAM,%200%29%29&spm=1018.2226.3001.4187)
33. 源码：[连接](https://download.csdn.net/download/weixin_43466192/89891912)
34. 罗老师：[https://blog.csdn.net/u012117034/article/details/135124626](https://blog.csdn.net/u012117034/article/details/135124626)
35. 源码目录介绍：[https://blog.csdn.net/caoshangpa/article/details/79582873](https://blog.csdn.net/caoshangpa/article/details/79582873)
37. #include <stdio.h>
38. #include <stdlib.h>
39. #include <string.h>
40. #include <unistd.h>
41. #include <time.h>
42. #include <sys/types.h>
43. #include <sys/stat.h>
44. #include <sys/socket.h>
45. #include <sys/wait.h>
46. #include <netinet/in.h>
47. #include <arpa/inet.h>
48. #include <netdb.h>
49. #include <fcntl.h>
50. #include <signal.h>
51. #include <pwd.h>
52. #include "st.h"
55. /******************************************************************
56. * Server configuration parameters
57. */
59. /* Log files */
60. #define PID_FILE    "pid"
61. #define ERRORS_FILE "errors"
62. #define ACCESS_FILE "access"
64. /* Default server port */
65. #define SERV_PORT_DEFAULT 8000
67. /* Socket listen queue size */
68. #define LISTENQ_SIZE_DEFAULT 256
70. /* Max number of listening sockets ("hardware virtual servers") */
71. #define MAX_BIND_ADDRS 16
73. /* Max number of "spare" threads per process per socket */
74. #define MAX_WAIT_THREADS_DEFAULT 8
76. /* Number of file descriptors needed to handle one client session */
77. #define FD_PER_THREAD 2
79. /* Access log buffer flushing interval (in seconds) */
80. #define ACCLOG_FLUSH_INTERVAL 30
82. /* Request read timeout (in seconds) */
83. #define REQUEST_TIMEOUT 30
86. /******************************************************************
87. * Global data
88. */
90. struct socket_info {
91.   st_netfd_t nfd;               /* Listening socket                     */
92.   char *addr;                   /* Bind address                         */
93.   unsigned int port;            /* Port                                 */
94.   int wait_threads;             /* Number of threads waiting to accept  */
95.   int busy_threads;             /* Number of threads processing request */
96.   int rqst_count;               /* Total number of processed requests   */
97. } srv_socket[MAX_BIND_ADDRS];   /* Array of listening sockets           */
99. static int sk_count = 0;        /* Number of listening sockets          */
101. static int vp_count = 0;        /* Number of server processes (VPs)     */
102. static pid_t *vp_pids;          /* Array of VP pids                     */
104. static int my_index = -1;       /* Current process index */
105. static pid_t my_pid = -1;       /* Current process pid   */
107. static st_netfd_t sig_pipe[2];  /* Signal pipe           */
109. /*
110. * Configuration flags/parameters
111. */
112. static int interactive_mode = 0;
113. static int serialize_accept = 0;
114. static int log_access       = 0;
115. static char *logdir     = NULL;
116. static char *username   = NULL;
117. static int listenq_size = LISTENQ_SIZE_DEFAULT;
118. static int errfd        = STDERR_FILENO;
120. /*
121. * Thread throttling parameters (all numbers are per listening socket).
122. * Zero values mean use default.
123. */
124. static int max_threads = 0;       /* Max number of threads         */
125. static int max_wait_threads = 0;  /* Max number of "spare" threads */
126. static int min_wait_threads = 2;  /* Min number of "spare" threads */
129. /******************************************************************
130. * Useful macros
131. */
133. #ifndef INADDR_NONE
134. #define INADDR_NONE 0xffffffff
135. #endif
137. #define SEC2USEC(s) ((s)*1000000LL)
139. #define WAIT_THREADS(i)  (srv_socket[i].wait_threads)
140. #define BUSY_THREADS(i)  (srv_socket[i].busy_threads)
141. #define TOTAL_THREADS(i) (WAIT_THREADS(i) + BUSY_THREADS(i))
142. #define RQST_COUNT(i)    (srv_socket[i].rqst_count)
145. /******************************************************************
146. * Forward declarations
147. */
149. static void usage(const char *progname);
150. static void parse_arguments(int argc, char *argv[]);
151. static void start_daemon(void);
152. static void set_thread_throttling(void);
153. static void create_listeners(void);
154. static void change_user(void);
155. static void open_log_files(void);
156. static void start_processes(void);
157. static void wdog_sighandler(int signo);
158. static void child_sighandler(int signo);
159. static void install_sighandlers(void);
160. static void start_threads(void);
161. static void *process_signals(void *arg);
162. static void *flush_acclog_buffer(void *arg);
163. static void *handle_connections(void *arg);
164. static void dump_server_info(void);
166. static void Signal(int sig, void (*handler)(int));
167. static int cpu_count(void);
169. extern void handle_session(long srv_socket_index, st_netfd_t cli_nfd);
170. extern void load_configs(void);
171. extern void logbuf_open(void);
172. extern void logbuf_flush(void);
173. extern void logbuf_close(void);
175. /* Error reporting functions defined in the error.c file */
176. extern void err_sys_report(int fd, const char *fmt, ...);
177. extern void err_sys_quit(int fd, const char *fmt, ...);
178. extern void err_sys_dump(int fd, const char *fmt, ...);
179. extern void err_report(int fd, const char *fmt, ...);
180. extern void err_quit(int fd, const char *fmt, ...);
183. /*
184. * General server example: accept a client connection and do something.
185. * This program just outputs a short HTML page, but can be easily adapted
186. * to do other things.
187. *
188. * This server creates a constant number of processes ("virtual processors"
189. * or VPs) and replaces them when they die. Each virtual processor manages
190. * its own independent set of state threads (STs), the number of which varies
191. * with load against the server. Each state thread listens to exactly one
192. * listening socket. The initial process becomes the watchdog, waiting for
193. * children (VPs) to die or for a signal requesting termination or restart.
194. * Upon receiving a restart signal (SIGHUP), all VPs close and then reopen
195. * log files and reload configuration. All currently active connections remain
196. * active. It is assumed that new configuration affects only request
197. * processing and not the general server parameters such as number of VPs,
198. * thread limits, bind addresses, etc. Those are specified as command line
199. * arguments, so the server has to be stopped and then started again in order
200. * to change them.
201. *
202. * Each state thread loops processing connections from a single listening
203. * socket. Only one ST runs on a VP at a time, and VPs do not share memory,
204. * so no mutual exclusion locking is necessary on any data, and the entire
205. * server is free to use all the static variables and non-reentrant library
206. * functions it wants, greatly simplifying programming and debugging and
207. * increasing performance (for example, it is safe to ++ and -- all global
208. * counters or call inet_ntoa(3) without any mutexes). The current thread on
209. * each VP maintains equilibrium on that VP, starting a new thread or
210. * terminating itself if the number of spare threads exceeds the lower or
211. * upper limit.
212. *
213. * All I/O operations on sockets must use the State Thread library's I/O
214. * functions because only those functions prevent blocking of the entire VP
215. * process and perform state thread scheduling.
216. */
217. int main(int argc, char *argv[])
218. {
219.   /* Parse command-line options */
220.   parse_arguments(argc, argv);
222.   /* Allocate array of server pids */
223.   if ((vp_pids = calloc(vp_count, sizeof(pid_t))) == NULL)
224.     err_sys_quit(errfd, "ERROR: calloc failed");
226.   /* Start the daemon */
227.   if (!interactive_mode)
228.     start_daemon();
230.   /* Initialize the ST library */
231.   if (st_init() < 0)
232.     err_sys_quit(errfd, "ERROR: initialization failed: st_init");
234.   /* Set thread throttling parameters */
235.   set_thread_throttling();
237.   /* Create listening sockets */
238.   create_listeners();
240.   /* Change the user */
241.   if (username)
242.     change_user();
244.   /* Open log files */
245.   open_log_files();
247.   /* Start server processes (VPs) */
248.   start_processes();
250.   /* Turn time caching on */
251.   st_timecache_set(1);
253.   /* Install signal handlers */
254.   install_sighandlers();
256.   /* Load configuration from config files */
257.   load_configs();
259.   /* Start all threads */
260.   start_threads();
262.   /* Become a signal processing thread */
263.   process_signals(NULL);
265.   /* NOTREACHED */
266.   return 1;
267. }
270. /******************************************************************/
272. static void usage(const char *progname)
273. {
274.   fprintf(stderr, "Usage: %s -l <log_directory> [<options>]\n\n"
275.           "Possible options:\n\n"
276.           "\t-b <host>:<port>        Bind to specified address. Multiple"
277.           " addresses\n"
278.           "\t                        are permitted.\n"
279.           "\t-p <num_processes>      Create specified number of processes.\n"
280.           "\t-t <min_thr>:<max_thr>  Specify thread limits per listening"
281.           " socket\n"
282.           "\t                        across all processes.\n"
283.           "\t-u <user>               Change server's user id to specified"
284.           " value.\n"
285.           "\t-q <backlog>            Set max length of pending connections"
286.           " queue.\n"
287.           "\t-a                      Enable access logging.\n"
288.           "\t-i                      Run in interactive mode.\n"
289.           "\t-S                      Serialize all accept() calls.\n"
290.           "\t-h                      Print this message.\n",
291.           progname);
292.   exit(1);
293. }
296. /******************************************************************/
298. static void parse_arguments(int argc, char *argv[])
299. {
300.   extern char *optarg;
301.   int opt;
302.   char *c;
304.   while ((opt = getopt(argc, argv, "b:p:l:t:u:q:aiSh")) != EOF) {
305.     switch (opt) {
306.     case 'b':
307.       if (sk_count >= MAX_BIND_ADDRS)
308.         err_quit(errfd, "ERROR: max number of bind addresses (%d) exceeded",
309.                  MAX_BIND_ADDRS);
310.       if ((c = strdup(optarg)) == NULL)
311.         err_sys_quit(errfd, "ERROR: strdup");
312.       srv_socket[sk_count++].addr = c;
313.       break;
314.     case 'p':
315.       vp_count = atoi(optarg);
316.       if (vp_count < 1)
317.         err_quit(errfd, "ERROR: invalid number of processes: %s", optarg);
318.       break;
319.     case 'l':
320.       logdir = optarg;
321.       break;
322.     case 't':
323.       max_wait_threads = (int) strtol(optarg, &c, 10);
324.       if (*c++ == ':')
325.         max_threads = atoi(c);
326.       if (max_wait_threads < 0 || max_threads < 0)
327.         err_quit(errfd, "ERROR: invalid number of threads: %s", optarg);
328.       break;
329.     case 'u':
330.       username = optarg;
331.       break;
332.     case 'q':
333.       listenq_size = atoi(optarg);
334.       if (listenq_size < 1)
335.         err_quit(errfd, "ERROR: invalid listen queue size: %s", optarg);
336.       break;
337.     case 'a':
338.       log_access = 1;
339.       break;
340.     case 'i':
341.       interactive_mode = 1;
342.       break;
343.     case 'S':
344.       /*
345.        * Serialization decision is tricky on some platforms. For example,
346.        * Solaris 2.6 and above has kernel sockets implementation, so supposedly
347.        * there is no need for serialization. The ST library may be compiled
348.        * on one OS version, but used on another, so the need for serialization
349.        * should be determined at run time by the application. Since it's just
350.        * an example, the serialization decision is left up to user.
351.        * Only on platforms where the serialization is never needed on any OS
352.        * version st_netfd_serialize_accept() is a no-op.
353.        */
354.       serialize_accept = 1;
355.       break;
356.     case 'h':
357.     case '?':
358.       usage(argv[0]);
359.     }
360.   }
362.   if (logdir == NULL && !interactive_mode) {
363.     err_report(errfd, "ERROR: logging directory is required\n");
364.     usage(argv[0]);
365.   }
367.   if (getuid() == 0 && username == NULL)
368.     err_report(errfd, "WARNING: running as super-user!");
370.   if (vp_count == 0 && (vp_count = cpu_count()) < 1)
371.     vp_count = 1;
373.   if (sk_count == 0) {
374.     sk_count = 1;
375.     srv_socket[0].addr = "0.0.0.0";
376.   }
377. }
380. /******************************************************************/
382. static void start_daemon(void)
383. {
384.   pid_t pid;
386.   /* Start forking */
387.   if ((pid = fork()) < 0)
388.     err_sys_quit(errfd, "ERROR: fork");
389.   if (pid > 0)
390.     exit(0);                  /* parent */
392.   /* First child process */
393.   setsid();                   /* become session leader */
395.   if ((pid = fork()) < 0)
396.     err_sys_quit(errfd, "ERROR: fork");
397.   if (pid > 0)                /* first child */
398.     exit(0);
400.   umask(022);
402.   if (chdir(logdir) < 0)
403.     err_sys_quit(errfd, "ERROR: can't change directory to %s: chdir", logdir);
404. }
407. /******************************************************************
408. * For simplicity, the minimal size of thread pool is considered
409. * as a maximum number of spare threads (max_wait_threads) that
410. * will be created upon server startup. The pool size can grow up
411. * to the max_threads value. Note that this is a per listening
412. * socket limit. It is also possible to limit the total number of
413. * threads for all sockets rather than impose a per socket limit.
414. */
416. static void set_thread_throttling(void)
417. {
418.   /*
419.    * Calculate total values across all processes.
420.    * All numbers are per listening socket.
421.    */
422.   if (max_wait_threads == 0)
423.     max_wait_threads = MAX_WAIT_THREADS_DEFAULT * vp_count;
424.   /* Assuming that each client session needs FD_PER_THREAD file descriptors */
425.   if (max_threads == 0)
426.     max_threads = (st_getfdlimit() * vp_count) / FD_PER_THREAD / sk_count;
427.   if (max_wait_threads > max_threads)
428.     max_wait_threads = max_threads;
430.   /*
431.    * Now calculate per-process values.
432.    */
433.   if (max_wait_threads % vp_count)
434.     max_wait_threads = max_wait_threads / vp_count + 1;
435.   else
436.     max_wait_threads = max_wait_threads / vp_count;
437.   if (max_threads % vp_count)
438.     max_threads = max_threads / vp_count + 1;
439.   else
440.     max_threads = max_threads / vp_count;
442.   if (min_wait_threads > max_wait_threads)
443.     min_wait_threads = max_wait_threads;
444. }
447. /******************************************************************/
449. static void create_listeners(void)
450. {
451.   int i, n, sock;
452.   char *c;
453.   struct sockaddr_in serv_addr;
454.   struct hostent *hp;
455.   unsigned short port;
457.   for (i = 0; i < sk_count; i++) {
458.     port = 0;
459.     if ((c = strchr(srv_socket[i].addr, ':')) != NULL) {
460.       *c++ = '\0';
461.       port = (unsigned short) atoi(c);
462.     }
463.     if (srv_socket[i].addr[0] == '\0')
464.       srv_socket[i].addr = "0.0.0.0";
465.     if (port == 0)
466.       port = SERV_PORT_DEFAULT;
468.     /* Create server socket */
469.     if ((sock = socket(PF_INET, SOCK_STREAM, 0)) < 0)
470.       err_sys_quit(errfd, "ERROR: can't create socket: socket");
471.     n = 1;
472.     if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, (char *)&n, sizeof(n)) < 0)
473.       err_sys_quit(errfd, "ERROR: can't set SO_REUSEADDR: setsockopt");
474.     memset(&serv_addr, 0, sizeof(serv_addr));
475.     serv_addr.sin_family = AF_INET;
476.     serv_addr.sin_port = htons(port);
477.     serv_addr.sin_addr.s_addr = inet_addr(srv_socket[i].addr);
478.     if (serv_addr.sin_addr.s_addr == INADDR_NONE) {
479.       /* not dotted-decimal */
480.       if ((hp = gethostbyname(srv_socket[i].addr)) == NULL)
481.         err_quit(errfd, "ERROR: can't resolve address: %s",
482.                  srv_socket[i].addr);
483.       memcpy(&serv_addr.sin_addr, hp->h_addr, hp->h_length);
484.     }
485.     srv_socket[i].port = port;
487.     /* Do bind and listen */
488.     if (bind(sock, (struct sockaddr *)&serv_addr, sizeof(serv_addr)) < 0)
489.       err_sys_quit(errfd, "ERROR: can't bind to address %s, port %hu",
490.                    srv_socket[i].addr, port);
491.     if (listen(sock, listenq_size) < 0)
492.       err_sys_quit(errfd, "ERROR: listen");
494.     /* Create file descriptor object from OS socket */
495.     if ((srv_socket[i].nfd = st_netfd_open_socket(sock)) == NULL)
496.       err_sys_quit(errfd, "ERROR: st_netfd_open_socket");
497.     /*
498.      * On some platforms (e.g. IRIX, Linux) accept() serialization is never
499.      * needed for any OS version.  In that case st_netfd_serialize_accept()
500.      * is just a no-op. Also see the comment above.
501.      */
502.     if (serialize_accept && st_netfd_serialize_accept(srv_socket[i].nfd) < 0)
503.       err_sys_quit(errfd, "ERROR: st_netfd_serialize_accept");
504.   }
505. }
508. /******************************************************************/
510. static void change_user(void)
511. {
512.   struct passwd *pw;
514.   if ((pw = getpwnam(username)) == NULL)
515.     err_quit(errfd, "ERROR: can't find user '%s': getpwnam failed", username);
517.   if (setgid(pw->pw_gid) < 0)
518.     err_sys_quit(errfd, "ERROR: can't change group id: setgid");
519.   if (setuid(pw->pw_uid) < 0)
520.     err_sys_quit(errfd, "ERROR: can't change user id: setuid");
522.   err_report(errfd, "INFO: changed process user id to '%s'", username);
523. }
526. /******************************************************************/
528. static void open_log_files(void)
529. {
530.   int fd;
531.   char str[32];
533.   if (interactive_mode)
534.     return;
536.   /* Open access log */
537.   if (log_access)
538.     logbuf_open();
540.   /* Open and write pid to pid file */
541.   if ((fd = open(PID_FILE, O_CREAT | O_WRONLY | O_TRUNC, 0644)) < 0)
542.     err_sys_quit(errfd, "ERROR: can't open pid file: open");
543.   sprintf(str, "%d\n", (int)getpid());
544.   if (write(fd, str, strlen(str)) != strlen(str))
545.     err_sys_quit(errfd, "ERROR: can't write to pid file: write");
546.   close(fd);
548.   /* Open error log file */
549.   if ((fd = open(ERRORS_FILE, O_CREAT | O_WRONLY | O_APPEND, 0644)) < 0)
550.     err_sys_quit(errfd, "ERROR: can't open error log file: open");
551.   errfd = fd;
553.   err_report(errfd, "INFO: starting the server...");
554. }
557. /******************************************************************/
559. static void start_processes(void)
560. {
561.   int i, status;
562.   pid_t pid;
563.   sigset_t mask, omask;
565.   if (interactive_mode) {
566.     my_index = 0;
567.     my_pid = getpid();
568.     return;
569.   }
571.   for (i = 0; i < vp_count; i++) {
572.     if ((pid = fork()) < 0) {
573.       err_sys_report(errfd, "ERROR: can't create process: fork");
574.       if (i == 0)
575.         exit(1);
576.       err_report(errfd, "WARN: started only %d processes out of %d", i,
577.                  vp_count);
578.       vp_count = i;
579.       break;
580.     }
581.     if (pid == 0) {
582.       my_index = i;
583.       my_pid = getpid();
584.       /* Child returns to continue in main() */
585.       return;
586.     }
587.     vp_pids[i] = pid;
588.   }
590.   /*
591.    * Parent process becomes a "watchdog" and never returns to main().
592.    */
594.   /* Install signal handlers */
595.   Signal(SIGTERM, wdog_sighandler);  /* terminate */
596.   Signal(SIGHUP,  wdog_sighandler);  /* restart   */
597.   Signal(SIGUSR1, wdog_sighandler);  /* dump info */
599.   /* Now go to sleep waiting for a child termination or a signal */
600.   for ( ; ; ) {
601.     if ((pid = wait(&status)) < 0) {
602.       if (errno == EINTR)
603.         continue;
604.       err_sys_quit(errfd, "ERROR: watchdog: wait");
605.     }
606.     /* Find index of the exited child */
607.     for (i = 0; i < vp_count; i++) {
608.       if (vp_pids[i] == pid)
609.         break;
610.     }
612.     /* Block signals while printing and forking */
613.     sigemptyset(&mask);
614.     sigaddset(&mask, SIGTERM);
615.     sigaddset(&mask, SIGHUP);
616.     sigaddset(&mask, SIGUSR1);
617.     sigprocmask(SIG_BLOCK, &mask, &omask);
619.     if (WIFEXITED(status))
620.       err_report(errfd, "WARN: watchdog: process %d (pid %d) exited"
621.                  " with status %d", i, pid, WEXITSTATUS(status));
622.     else if (WIFSIGNALED(status))
623.       err_report(errfd, "WARN: watchdog: process %d (pid %d) terminated"
624.                  " by signal %d", i, pid, WTERMSIG(status));
625.     else if (WIFSTOPPED(status))
626.       err_report(errfd, "WARN: watchdog: process %d (pid %d) stopped"
627.                  " by signal %d", i, pid, WSTOPSIG(status));
628.     else
629.       err_report(errfd, "WARN: watchdog: process %d (pid %d) terminated:"
630.                  " unknown termination reason", i, pid);
632.     /* Fork another VP */
633.     if ((pid = fork()) < 0) {
634.       err_sys_report(errfd, "ERROR: watchdog: can't create process: fork");
635.     } else if (pid == 0) {
636.       my_index = i;
637.       my_pid = getpid();
638.       /* Child returns to continue in main() */
639.       return;
640.     }
641.     vp_pids[i] = pid;
643.     /* Restore the signal mask */
644.     sigprocmask(SIG_SETMASK, &omask, NULL);
645.   }
646. }
649. /******************************************************************/
651. static void wdog_sighandler(int signo)
652. {
653.   int i, err;
655.   /* Save errno */
656.   err = errno;
657.   /* Forward the signal to all children */
658.   for (i = 0; i < vp_count; i++) {
659.     if (vp_pids[i] > 0)
660.       kill(vp_pids[i], signo);
661.   }
662.   /*
663.    * It is safe to do pretty much everything here because process is
664.    * sleeping in wait() which is async-safe.
665.    */
666.   switch (signo) {
667.   case SIGHUP:
668.     err_report(errfd, "INFO: watchdog: caught SIGHUP");
669.     /* Reopen log files - needed for log rotation */
670.     if (log_access) {
671.       logbuf_close();
672.       logbuf_open();
673.     }
674.     close(errfd);
675.     if ((errfd = open(ERRORS_FILE, O_CREAT | O_WRONLY | O_APPEND, 0644)) < 0)
676.       err_sys_quit(STDERR_FILENO, "ERROR: watchdog: open");
677.     break;
678.   case SIGTERM:
679.     /* Non-graceful termination */
680.     err_report(errfd, "INFO: watchdog: caught SIGTERM, terminating");
681.     unlink(PID_FILE);
682.     exit(0);
683.   case SIGUSR1:
684.     err_report(errfd, "INFO: watchdog: caught SIGUSR1");
685.     break;
686.   default:
687.     err_report(errfd, "INFO: watchdog: caught signal %d", signo);
688.   }
689.   /* Restore errno */
690.   errno = err;
691. }
694. /******************************************************************/
696. static void install_sighandlers(void)
697. {
698.   sigset_t mask;
699.   int p[2];
701.   /* Create signal pipe */
702.   if (pipe(p) < 0)
703.     err_sys_quit(errfd, "ERROR: process %d (pid %d): can't create"
704.                  " signal pipe: pipe", my_index, my_pid);
705.   if ((sig_pipe[0] = st_netfd_open(p[0])) == NULL ||
706.       (sig_pipe[1] = st_netfd_open(p[1])) == NULL)
707.     err_sys_quit(errfd, "ERROR: process %d (pid %d): can't create"
708.                  " signal pipe: st_netfd_open", my_index, my_pid);
710.   /* Install signal handlers */
711.   Signal(SIGTERM, child_sighandler);  /* terminate */
712.   Signal(SIGHUP,  child_sighandler);  /* restart   */
713.   Signal(SIGUSR1, child_sighandler);  /* dump info */
715.   /* Unblock signals */
716.   sigemptyset(&mask);
717.   sigaddset(&mask, SIGTERM);
718.   sigaddset(&mask, SIGHUP);
719.   sigaddset(&mask, SIGUSR1);
720.   sigprocmask(SIG_UNBLOCK, &mask, NULL);
721. }
724. /******************************************************************/
726. static void child_sighandler(int signo)
727. {
728.   int err, fd;
730.   err = errno;
731.   fd = st_netfd_fileno(sig_pipe[1]);
733.   /* write() is async-safe */
734.   if (write(fd, &signo, sizeof(int)) != sizeof(int))
735.     err_sys_quit(errfd, "ERROR: process %d (pid %d): child's signal"
736.                  " handler: write", my_index, my_pid);
737.   errno = err;
738. }
741. /******************************************************************
742. * The "main" function of the signal processing thread.
743. */
745. /* ARGSUSED */
746. static void *process_signals(void *arg)
747. {
748.   int signo;
750.   for ( ; ; ) {
751.     /* Read the next signal from the signal pipe */
752.     if (st_read(sig_pipe[0], &signo, sizeof(int),
753.      ST_UTIME_NO_TIMEOUT) != sizeof(int))
754.       err_sys_quit(errfd, "ERROR: process %d (pid %d): signal processor:"
755.                    " st_read", my_index, my_pid);
757.     switch (signo) {
758.     case SIGHUP:
759.       err_report(errfd, "INFO: process %d (pid %d): caught SIGHUP,"
760.                  " reloading configuration", my_index, my_pid);
761.       if (interactive_mode) {
762.         load_configs();
763.         break;
764.       }
765.       /* Reopen log files - needed for log rotation */
766.       if (log_access) {
767.         logbuf_flush();
768.         logbuf_close();
769.         logbuf_open();
770.       }
771.       close(errfd);
772.       if ((errfd = open(ERRORS_FILE, O_CREAT | O_WRONLY | O_APPEND, 0644)) < 0)
773.         err_sys_quit(STDERR_FILENO, "ERROR: process %d (pid %d): signal"
774.                      " processor: open", my_index, my_pid);
775.       /* Reload configuration */
776.       load_configs();
777.       break;
778.     case SIGTERM:
779.       /*
780.        * Terminate ungracefully since it is generally not known how long
781.        * it will take to gracefully complete all client sessions.
782.        */
783.       err_report(errfd, "INFO: process %d (pid %d): caught SIGTERM,"
784.                  " terminating", my_index, my_pid);
785.       if (log_access)
786.         logbuf_flush();
787.       exit(0);
788.     case SIGUSR1:
789.       err_report(errfd, "INFO: process %d (pid %d): caught SIGUSR1",
790.                  my_index, my_pid);
791.       /* Print server info to stderr */
792.       dump_server_info();
793.       break;
794.     default:
795.       err_report(errfd, "INFO: process %d (pid %d): caught signal %d",
796.                  my_index, my_pid, signo);
797.     }
798.   }
800.   /* NOTREACHED */
801.   return NULL;
802. }
805. /******************************************************************
806. * The "main" function of the access log flushing thread.
807. */
809. /* ARGSUSED */
810. static void *flush_acclog_buffer(void *arg)
811. {
812.   for ( ; ; ) {
813.     st_sleep(ACCLOG_FLUSH_INTERVAL);
814.     logbuf_flush();
815.   }
817.   /* NOTREACHED */
818.   return NULL;
819. }
822. /******************************************************************/
824. static void start_threads(void)
825. {
826.   long i, n;
828.   /* Create access log flushing thread */
829.   if (log_access && st_thread_create(flush_acclog_buffer, NULL, 0, 0) == NULL)
830.     err_sys_quit(errfd, "ERROR: process %d (pid %d): can't create"
831.                  " log flushing thread", my_index, my_pid);
833.   /* Create connections handling threads */
834.   for (i = 0; i < sk_count; i++) {
835.     err_report(errfd, "INFO: process %d (pid %d): starting %d threads"
836.                " on %s:%u", my_index, my_pid, max_wait_threads,
837.                srv_socket[i].addr, srv_socket[i].port);
838.     WAIT_THREADS(i) = 0;
839.     BUSY_THREADS(i) = 0;
840.     RQST_COUNT(i) = 0;
841.     for (n = 0; n < max_wait_threads; n++) {
842.       if (st_thread_create(handle_connections, (void *)i, 0, 0) != NULL)
843.         WAIT_THREADS(i)++;
844.       else
845.         err_sys_report(errfd, "ERROR: process %d (pid %d): can't create"
846.                        " thread", my_index, my_pid);
847.     }
848.     if (WAIT_THREADS(i) == 0)
849.       exit(1);
850.   }
851. }
854. /******************************************************************/
856. static void *handle_connections(void *arg)
857. {
858.   st_netfd_t srv_nfd, cli_nfd;
859.   struct sockaddr_in from;
860.   int fromlen;
861.   long i = (long) arg;
863.   srv_nfd = srv_socket[i].nfd;
864.   fromlen = sizeof(from);
866.   while (WAIT_THREADS(i) <= max_wait_threads) {
867.     cli_nfd = st_accept(srv_nfd, (struct sockaddr *)&from, &fromlen,
868.      ST_UTIME_NO_TIMEOUT);
869.     if (cli_nfd == NULL) {
870.       err_sys_report(errfd, "ERROR: can't accept connection: st_accept");
871.       continue;
872.     }
873.     /* Save peer address, so we can retrieve it later */
874.     st_netfd_setspecific(cli_nfd, &from.sin_addr, NULL);
876.     WAIT_THREADS(i)--;
877.     BUSY_THREADS(i)++;
878.     if (WAIT_THREADS(i) < min_wait_threads && TOTAL_THREADS(i) < max_threads) {
879.       /* Create another spare thread */
880.       if (st_thread_create(handle_connections, (void *)i, 0, 0) != NULL)
881.         WAIT_THREADS(i)++;
882.       else
883.         err_sys_report(errfd, "ERROR: process %d (pid %d): can't create"
884.                        " thread", my_index, my_pid);
885.     }
887.     handle_session(i, cli_nfd);
889.     st_netfd_close(cli_nfd);
890.     WAIT_THREADS(i)++;
891.     BUSY_THREADS(i)--;
892.   }
894.   WAIT_THREADS(i)--;
895.   return NULL;
896. }
899. /******************************************************************/
901. static void dump_server_info(void)
902. {
903.   char *buf;
904.   int i, len;
906.   if ((buf = malloc(sk_count * 512)) == NULL) {
907.     err_sys_report(errfd, "ERROR: malloc failed");
908.     return;
909.   }
911.   len = sprintf(buf, "\n\nProcess #%d (pid %d):\n", my_index, (int)my_pid);
912.   for (i = 0; i < sk_count; i++) {
913.     len += sprintf(buf + len, "\nListening Socket #%d:\n"
914.                    "-------------------------\n"
915.                    "Address                    %s:%u\n"
916.                    "Thread limits (min/max)    %d/%d\n"
917.                    "Waiting threads            %d\n"
918.                    "Busy threads               %d\n"
919.                    "Requests served            %d\n",
920.                    i, srv_socket[i].addr, srv_socket[i].port,
921.                    max_wait_threads, max_threads,
922.                    WAIT_THREADS(i), BUSY_THREADS(i), RQST_COUNT(i));
923.   }
925.   write(STDERR_FILENO, buf, len);
926.   free(buf);
927. }
930. /******************************************************************
931. * Stubs
932. */
934. /*
935. * Session handling function stub. Just dumps small HTML page.
936. */
937. void handle_session(long srv_socket_index, st_netfd_t cli_nfd)
938. {
939.   static char resp[] = "HTTP/1.0 200 OK\r\nContent-type: text/html\r\n"
940.                        "Connection: close\r\n\r\n<H2>It worked!</H2>\n";
941.   char buf[512];
942.   int n = sizeof(resp) - 1;
943.   struct in_addr *from = st_netfd_getspecific(cli_nfd);
945.   if (st_read(cli_nfd, buf, sizeof(buf), SEC2USEC(REQUEST_TIMEOUT)) < 0) {
946.     err_sys_report(errfd, "WARN: can't read request from %s: st_read",
947.                    inet_ntoa(*from));
948.     return;
949.   }
950.   if (st_write(cli_nfd, resp, n, ST_UTIME_NO_TIMEOUT) != n) {
951.     err_sys_report(errfd, "WARN: can't write response to %s: st_write",
952.                    inet_ntoa(*from));
953.     return;
954.   }
956.   RQST_COUNT(srv_socket_index)++;
957. }
960. /*
961. * Configuration loading function stub.
962. */
963. void load_configs(void)
964. {
965.   err_report(errfd, "INFO: process %d (pid %d): configuration loaded",
966.              my_index, my_pid);
967. }
970. /*
971. * Buffered access logging methods.
972. * Note that stdio functions (fopen(3), fprintf(3), fflush(3), etc.) cannot
973. * be used if multiple VPs are created since these functions can flush buffer
974. * at any point and thus write only partial log record to disk.
975. * Also, it is completely safe for all threads of the same VP to write to
976. * the same log buffer without any mutex protection (one buffer per VP, of
977. * course).
978. */
979. void logbuf_open(void)
980. {
982. }
985. void logbuf_flush(void)
986. {
988. }
991. void logbuf_close(void)
992. {
994. }
997. /******************************************************************
998. * Small utility functions
999. */
1001. static void Signal(int sig, void (*handler)(int))
1002. {
1003.   struct sigaction sa;
1005.   sa.sa_handler = handler;
1006.   sigemptyset(&sa.sa_mask);
1007.   sa.sa_flags = 0;
1008.   sigaction(sig, &sa, NULL);
1009. }
1011. static int cpu_count(void)
1012. {
1013.   int n;
1015. #if defined (_SC_NPROCESSORS_ONLN)
1016.   n = (int) sysconf(_SC_NPROCESSORS_ONLN);
1017. #elif defined (_SC_NPROC_ONLN)
1018.   n = (int) sysconf(_SC_NPROC_ONLN);
1019. #elif defined (HPUX)
1020. #include <sys/mpctl.h>
1021.   n = mpctl(MPC_GETNUMSPUS, 0, 0);
1022. #else
1023.   n = -1;
1024.   errno = ENOSYS;
1025. #endif
1027.   return n;
1028. }
1030. /******************************************************************/

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。