http://blog.csdn.net/winlinvip/article/details/7764526

Linux多进程服务器真的很给力，赞一个！

Linux多进程一般是master负责侦听，worker接受和伺服client。

一个使用了以下技术的多进程模型：

1. sigset：安全信号，信号屏蔽和接受。

2. epoll：异步io模型。

master进程使用信号模型，侦听用户信号和程序信号，并和worker交流。它的主循环是sigsuspend。

worker进程使用事件模型，使用epoll_wait等待事件，同时他也接受信号（信号会中断epoll_wait）。

[cpp]  view plain copy
  1. // mpserver —— multiple processes server  
  2. #include <stdio.h>  
  3. #include <stdlib.h>  
  4. #include <iostream>  
  5. using namespace std;  
  6.   
  7. #include <unistd.h>  
  8. #include <signal.h>  
  9. #include <sys/types.h>  
  10. #include <sys/socket.h>  
  11. #include <netinet/in.h>  
  12. #include <sys/wait.h>  
  13. #include <sys/epoll.h>  
  14. #include <string.h>  
  15. #include <errno.h>  
  16.   
  17. #define err_exit(msg) cout << "[error] " << msg << endl; exit(1)  
  18.   
  19. struct UserOptions{  
  20.     int port;  
  21.     int num_processes;  
  22. };  
  23.   
  24. void discovery_user_options(int argc, char** argv, UserOptions& options){  
  25.     if(argc <= 2){  
  26.         cout << "Usage: " << argv[0] << " <port> <num_processes>" << endl  
  27.             << "port: the port to listen" << endl  
  28.             << "num_processes: the num to start processes. if 0, use single process mode" << endl;  
  29.         exit(1);  
  30.     }  
  31.       
  32.     options.port = atoi(argv[1]);  
  33.     options.num_processes = atoi(argv[2]);  
  34. }  
  35.   
  36. int listen_server_socket(UserOptions& options){  
  37.     int serverfd = socket(AF_INET, SOCK_STREAM, 0);  
  38.       
  39.     if(serverfd == -1){  
  40.         err_exit("init socket error!");  
  41.     }  
  42.     cout << "init socket success! #" << serverfd << endl;  
  43.       
  44.     int reuse_socket = 1;  
  45.     if(setsockopt(serverfd, SOL_SOCKET, SO_REUSEADDR, &reuse_socket, sizeof(int)) == -1){  
  46.         err_exit("setsockopt reuse-addr error!");  
  47.     }  
  48.     cout << "setsockopt reuse-addr success!" << endl;  
  49.       
  50.     sockaddr_in addr;  
  51.     addr.sin_family = AF_INET;  
  52.     addr.sin_port = htons(options.port);  
  53.     addr.sin_addr.s_addr = INADDR_ANY;  
  54.     if(bind(serverfd, (const sockaddr*)&addr, sizeof(sockaddr_in)) == -1){  
  55.         err_exit("bind socket error!");  
  56.     }  
  57.     cout << "bind socket success!" << endl;  
  58.       
  59.     if(listen(serverfd, 10) == -1){  
  60.         err_exit("listen socket error!");  
  61.     }  
  62.     cout << "listen socket success! " << options.port << endl;  
  63.       
  64.     return serverfd;  
  65. }  
  66.   
  67. void signal_handler(int signo){  
  68.     cout << "#" << getpid() << " get a signal:";  
  69.       
  70.     bool do_exit = false;  
  71.     switch(signo){  
  72.         case SIGCHLD:  
  73.             cout << "SIGCHLD";  
  74.             int status;  
  75.             while(waitpid(0, &status, WNOHANG) > 0){  
  76.             }  
  77.             break;  
  78.         case SIGALRM:  
  79.             cout << "SIGALRM";  
  80.             break;  
  81.         case SIGIO:  
  82.             cout << "SIGIO";  
  83.             break;  
  84.         case SIGINT:  
  85.             cout << "SIGINT";  
  86.             do_exit = true;  
  87.             break;  
  88.         case SIGHUP:  
  89.             cout << "SIGHUP";  
  90.             do_exit = true;  
  91.             break;  
  92.         case SIGTERM:  
  93.             cout << "SIGTERM";  
  94.             do_exit = true;  
  95.             break;  
  96.         case SIGQUIT:  
  97.             cout << "SIGQUIT";  
  98.             do_exit = true;  
  99.             break;  
  100.         case SIGUSR1:  
  101.             cout << "SIGUSR1";  
  102.             break;  
  103.         case SIGUSR2:  
  104.             cout << "SIGUSR2";  
  105.             break;  
  106.     }  
  107.       
  108.     cout << "!" << endl;  
  109.     if(do_exit){  
  110.         exit(0);  
  111.     }  
  112. }  
  113.   
  114. void register_signal_handler_imp(int signum, void (*handler)(int)){  
  115.     struct sigaction action;  
  116.       
  117.     action.sa_handler = handler;  
  118.     sigemptyset(&action.sa_mask);  
  119.     //action.sa_flags = 0;  
  120.       
  121.     if(sigaction(signum, &action, NULL) == -1){  
  122.         err_exit("register signal handler failed!");  
  123.     }  
  124. }  
  125. void register_signal_handler(){  
  126.     register_signal_handler_imp(SIGCHLD, signal_handler);  
  127.     register_signal_handler_imp(SIGALRM, signal_handler);  
  128.     register_signal_handler_imp(SIGIO, signal_handler);  
  129.     register_signal_handler_imp(SIGINT, signal_handler);  
  130.     register_signal_handler_imp(SIGHUP, signal_handler);  
  131.     register_signal_handler_imp(SIGTERM, signal_handler);  
  132.     register_signal_handler_imp(SIGQUIT, signal_handler);  
  133.     register_signal_handler_imp(SIGUSR1, signal_handler);  
  134.     register_signal_handler_imp(SIGUSR2, signal_handler);  
  135.     cout << "register signal handler success!" << endl;  
  136. }  
  137.   
  138. void block_specified_signals(){  
  139.     sigset_t set;  
  140.       
  141.     sigemptyset(&set);  
  142.       
  143.     sigaddset(&set, SIGCHLD);  
  144.     sigaddset(&set, SIGALRM);  
  145.     sigaddset(&set, SIGIO);  
  146.     sigaddset(&set, SIGINT);  
  147.     sigaddset(&set, SIGHUP);  
  148.     sigaddset(&set, SIGTERM);  
  149.     sigaddset(&set, SIGQUIT);  
  150.     sigaddset(&set, SIGUSR1);  
  151.     sigaddset(&set, SIGUSR2);  
  152.       
  153.     if(sigprocmask(SIG_BLOCK, &set, NULL) == -1){  
  154.         err_exit("sigprocmask block signal failed!");  
  155.     }  
  156.     cout << "sigprocmask block signal success!" << endl;  
  157. }  
  158.   
  159. void unblock_all_signals(){  
  160.     sigset_t set;  
  161.     sigemptyset(&set);  
  162.       
  163.     // the master process block all signals, we unblock all for we use epoll to wait events.  
  164.     if(sigprocmask(SIG_SETMASK, &set, NULL) == -1){  
  165.         err_exit("sigprocmask block signal failed!");  
  166.     }  
  167.     cout << "sigprocmask block signal success!" << endl;  
  168. }  
  169.   
  170. void epoll_add_event(int ep, int fd){  
  171.     epoll_event ee;  
  172.     ee.events = EPOLLIN;  
  173.     ee.data.fd = fd;  
  174.     if(epoll_ctl(ep, EPOLL_CTL_ADD, fd, &ee) == -1){  
  175.         err_exit("epoll add event failed!");  
  176.     }  
  177.     cout << "epoll add fd success: #" << fd << endl;  
  178. }  
  179. void epoll_remove_event(int ep, int fd){  
  180.     if(epoll_ctl(ep, EPOLL_CTL_DEL, fd, NULL) == -1){  
  181.         err_exit("epoll remove event failed! fd=" << fd << ", errno=0x" << hex << errno << dec << " " << strerror(errno));  
  182.     }  
  183.     cout << "epoll remove fd success: #" << fd << endl;  
  184. }  
  185.   
  186. void close_client(int ep, int fd){  
  187.     cout << "the client dead, remove it: #" << fd << endl;  
  188.     epoll_remove_event(ep, fd);  
  189.     close(fd);  
  190. }  
  191.   
  192. void worker_get_event(int ep, int serverfd, epoll_event& active){  
  193.     int fd = active.data.fd;  
  194.       
  195.     // server listening socket event.  
  196.     if(fd == serverfd){  
  197.         int client = accept(serverfd, NULL, 0);  
  198.         if(client == -1){  
  199.             err_exit("accept client socket error!");  
  200.         }  
  201.         cout << "get a client: #" << client << endl;  
  202.         epoll_add_event(ep, client);  
  203.         return;  
  204.     }  
  205.       
  206.     // client event.  
  207.     if((active.events & EPOLLHUP) == EPOLLHUP || (active.events & EPOLLERR) == EPOLLERR){  
  208.         cout << "get a EPOLLHUP or EPOLLERR event from client #" << fd << endl;  
  209.         close_client(ep, fd);  
  210.         return;  
  211.     }  
  212.     if((active.events & EPOLLIN) == EPOLLIN){  
  213.         if(true){  
  214.             cout << "get a EPOLLIN event from client #" << fd << endl;  
  215.             char buf[1024];  
  216.             memset(buf, 0, sizeof(buf));  
  217.             if(recv(fd, buf, sizeof(buf), 0) <= 0){  
  218.                 close_client(ep, fd);  
  219.                 return;  
  220.             }  
  221.             cout << "recv from client: " << buf << endl;  
  222.         }  
  223.         if(true){  
  224.             char msg[] = "hello, server ping!";  
  225.             if(send(fd, msg, sizeof(msg), 0) <= 0){  
  226.                 close_client(ep, fd);  
  227.                 return;  
  228.             }  
  229.         }  
  230.         return;  
  231.     }  
  232. }  
  233.   
  234. void worker_process_cycle(int serverfd){  
  235.     cout << "start worker process cycle" << endl;  
  236.     unblock_all_signals();  
  237.   
  238.     int ep = epoll_create(1024);  
  239.     if(ep == -1){  
  240.         err_exit("epoll_create failed!");  
  241.     }  
  242.     cout << "epoll create success!" << endl;  
  243.     epoll_add_event(ep, serverfd);  
  244.       
  245.     for(;;){  
  246.         epoll_event events[1024];  
  247.         int incoming = epoll_wait(ep, events, 1024, -1);  
  248.         if(incoming == -1){  
  249.             break;  
  250.         }  
  251.         for(int i = 0; i < incoming; i++){  
  252.             worker_get_event(ep, serverfd, events[i]);  
  253.         }  
  254.     }  
  255.       
  256.     cout << "worker process exited" << endl;  
  257.     close(ep);  
  258. }  
  259.   
  260. void start_worker_process(int serverfd){  
  261.     pid_t pid = fork();  
  262.       
  263.     if(pid == -1){  
  264.         err_exit("fork process failed");  
  265.     }  
  266.       
  267.     if(pid == 0){  
  268.         worker_process_cycle(serverfd);  
  269.         exit(0);  
  270.     }  
  271.       
  272.     cout << "fork process success: #" << pid << endl;  
  273. }  
  274.   
  275. int main(int argc, char** argv){  
  276.     register_signal_handler();  
  277.     block_specified_signals();  
  278.       
  279.     //sleep(3);  
  280.     UserOptions options;  
  281.     discovery_user_options(argc, argv, options);  
  282.     int serverfd = listen_server_socket(options);  
  283.       
  284.     for(int i = 0; i < options.num_processes; i++){  
  285.         start_worker_process(serverfd);  
  286.     }  
  287.     if(options.num_processes == 0){  
  288.         worker_process_cycle(serverfd);  
  289.     }  
  290.       
  291.     sigset_t set;  
  292.     sigemptyset(&set);  
  293.     for(;;){  
  294.         sigsuspend(&set);  
  295.     }  
  296.   
  297.     return 0;  
  298. }  

[cpp]  view plain copy
  1. // mpclient —— multiple processes client  
  2. #include <stdio.h>  
  3. #include <stdlib.h>  
  4. #include <iostream>  
  5. using namespace std;  
  6.   
  7. #include <unistd.h>  
  8. #include <signal.h>  
  9. #include <sys/types.h>  
  10. #include <sys/socket.h>  
  11. #include <netinet/in.h>  
  12. #include <arpa/inet.h>  
  13. #include <string.h>  
  14.   
  15. #define err_exit(msg) cout << "[error] " << msg << endl; exit(1)  
  16.   
  17. struct UserOptions{  
  18.     char* server_ip;  
  19.     int port;  
  20.     int num_processes;  
  21. };  
  22.   
  23. void discovery_user_options(int argc, char** argv, UserOptions& options){  
  24.     if(argc <= 2){  
  25.         cout << "Usage: " << argv[0] << " <server_ip> <port> <num_processes>" << endl  
  26.             << "server_ip: the ip address of server" << endl  
  27.             << "port: the port to connect at" << endl  
  28.             << "num_processes: the num to start processes. if 0, use single process mode" << endl;  
  29.         exit(1);  
  30.     }  
  31.       
  32.     options.server_ip = argv[1];  
  33.     options.port = atoi(argv[2]);  
  34.     options.num_processes = atoi(argv[3]);  
  35. }  
  36.   
  37. int connect_server_socket(UserOptions& options){  
  38.     int clientfd = socket(AF_INET, SOCK_STREAM, 0);  
  39.       
  40.     if(clientfd == -1){  
  41.         err_exit("init socket error!");  
  42.     }  
  43.     cout << "init socket success!" << endl;  
  44.       
  45.     sockaddr_in addr;  
  46.     addr.sin_family = AF_INET;  
  47.     addr.sin_port = htons(options.port);  
  48.     addr.sin_addr.s_addr = inet_addr(options.server_ip);  
  49.     if(connect(clientfd, (const sockaddr*)&addr, sizeof(sockaddr_in)) == -1){  
  50.         err_exit("connect socket error!");  
  51.     }  
  52.     cout << "connect socket success!" << endl;  
  53.       
  54.     return clientfd;  
  55. }  
  56.   
  57. int main(int argc, char** argv){  
  58.     UserOptions options;  
  59.     discovery_user_options(argc, argv, options);  
  60.     int* fds = new int[options.num_processes];  
  61.     for(int i = 0; i < options.num_processes; i++){  
  62.         fds[i] = connect_server_socket(options);  
  63.     }  
  64.       
  65.     bool do_loop = true;  
  66.     while(do_loop){  
  67.         for(int i = 0; i < options.num_processes; i++){  
  68.             int fd = fds[i];  
  69.             if(true){  
  70.                 char msg[] = "hello, client ping!";  
  71.                 if(send(fd, msg, sizeof(msg), 0) <= 0){  
  72.                     close(fd);  
  73.                     do_loop = false;  
  74.                     break;  
  75.                 }  
  76.             }  
  77.             if(true){  
  78.                 char buf[1024];  
  79.                 memset(buf, 0, sizeof(buf));  
  80.                 if(recv(fd, buf, sizeof(buf), 0) <= 0){  
  81.                     close(fd);  
  82.                     do_loop = false;  
  83.                     break;  
  84.                 }  
  85.                 cout << "recv from server: " << buf << endl;  
  86.             }  
  87.               
  88.             sleep(3);  
  89.         }  
  90.     }  
  91.       
  92.     delete[] fds;  
  93.   
  94.     return 0;  
  95. }  


master启动的主要过程：

1. 注册信号处理函数：sigaction。

2. 屏蔽所有关心的信号，让内核暂时保留信号，不要发给程序，因为这时候程序不系统被打断。

3. 侦听端口。

4. 启动worker进程。

5. 等待信号的到来：sigsuspend。


worker启动的主要过程：

1. 取消信号的屏蔽。因为这时候worker拷贝的是master的内存结构，所以信号屏蔽也拷贝过来了。若不取消，则无法收到退出消息，只能重启服务器或它自己退出才能杀这个进程了。

2. 创建epoll。

3. 将侦听的server_socket_fd添加到epoll。

4. 等待epoll事件的到来：epoll_wait。或者信号也能中断这个epoll_wait：返回-1.


典型的多进程服务器模型如下图所示：


考虑一个RTMP服务器，假设和FMS类似以vhost为边界启动worker，但是client应该连接到指定的worker。所以这个典型的模型就需要修改了。

有一种方式是master接受连接，handshake，然后接收到clieint的connect请求后，知道连接的vhost是哪个，再把这个client fd转发给worker。如下图所示：


但这种方式master过于复杂，它实际上需要做worker要做的事情。多进程稳定的一个原因是进程的自然边界，worker进程崩溃不会影响其他worker和master，如果中心master崩溃，那么系统就完蛋了。所以master要尽量简单不能有过多的逻辑。另外，单进程和多进程切换的一个条件，就是worker能用一个函数表达，单进程时直接调用这个函数，而多进程时fork后调用这个函数；若master要混合worker的功能，单进程时也需要过多交互。

改进如下图：


master变得简单，worker功能也单一了。worker发现不是自己的client时，将client发给给master，由master分发给特定的worker。单进程时没有分发这一步了；多进程时master也只需要分发fd，而不需要处理协议的逻辑。

在master和worker之间传递fd不是一个好办法，因为过多频繁的SIGIO可能导致“I/O possible”；google上找了一下说是由于信号SIGIO没有处理引起，但明显已经侦听了handler；有人说是signal queue overflow，恩，这个很像！

socketpaire有时还会引起“[write_channel] sendmsg error, ret=-1, errno=0xb Resource temporarily unavailable”错误。管道错误。一般发生在worker进程中，如果多个woker进程都向master发送数据，而master处理一个数据后sleep或者做其他的事情去了，就会造成这种情况（master的socketpaire缓冲区满了）。

master若需要和worker使用socketpair通信，master使用信号来实现异步，当worker的socketpair有SIGIO信号时，说明worker在发数据，如何判断是哪个worker发的呢？可以用select，当然不太好；可以使用F_SETSIG来更改默认的SIGIO，改为SIGRTMIN+fd，当worker有数据时，收到的信号就包含了fd了。

获取SIGIO的fd的另外一个方法，是使用siginfo_t取得fd。使用F_SETSIG可以设置的SIG较少，SIGRTMAX为64，所以最多可用32个自定义实时信号，一般用来对SIGIO分类。必须将SIGIO定义为实时信号，然后在sigaction时设置SA_SIGINFO和使用void (*handler)(int signo, siginfo_t* info, void* context)，info->si_fd就是SIGIO的fd。

以下是一个糅合了多进程、信号、socketpair、passing fd的一个原型，较稳定的一个原型：

[cpp]  view plain copy
  1. // mpserver2 —— multiple processes server  
  2. #include <stdio.h>  
  3. #include <string.h>  
  4. #include <stdlib.h>  
  5. #include <iostream>  
  6. #include <vector>  
  7. using namespace std;  
  8.   
  9. #include <unistd.h>  
  10. #include <signal.h>  
  11. #include <sys/types.h>  
  12. #include <sys/socket.h>  
  13. #include <netinet/in.h>  
  14. #include <sys/wait.h>  
  15. #include <sys/epoll.h>  
  16. #include <string.h>  
  17. #include <errno.h>  
  18. #include <fcntl.h>  
  19. #include <sys/ioctl.h>  
  20.   
  21. #define info(msg) cout << "#" << getpid() << " " << msg  
  22. #define err_exit(msg) info("[error] " << msg << ", errno=0x" << hex << errno << dec << " " << strerror(errno) << endl); exit(-1)  
  23.   
  24. // define the SIGIO to RTSig, the SIGIO now means rt-sig queue full.  
  25. #define SIG_SOCKET_IO SIGRTMIN+1  
  26.   
  27. bool global_terminate = false;  
  28. bool global_channel_msg = false;  
  29. // when global_channel_msg is true, the current_active_io_fd set to the fd.  
  30. int current_active_io_fd = -1;  
  31.   
  32. int current_client_num = 0;  
  33. bool accepting = true;  
  34.   
  35. // the num of client for a proecess to serve.  
  36. // 2+N  
  37. // 2: the serverfd, the channelfd  
  38. // N: clients.  
  39. #define max_clients_per_process 2+1000  
  40. // the physical max clients, if exceed this value, error!  
  41. #define physical_max_clients 1024  
  42.   
  43. #define max_workers 100  
  44.   
  45. struct UserOptions{  
  46.     int port;  
  47.     int num_processes;  
  48. };  
  49.   
  50. void discovery_user_options(int argc, char** argv, UserOptions& options){  
  51.     if(argc <= 2){  
  52.         cout << "Usage: " << argv[0] << " <port> <num_processes>" << endl  
  53.             << "port: the port to listen" << endl  
  54.             << "num_processes: the num to start processes. if 0, use single process mode" << endl  
  55.             << "for example: " << argv[0] << " 1990 20" << endl;  
  56.         exit(1);  
  57.     }  
  58.       
  59.     options.port = atoi(argv[1]);  
  60.     options.num_processes = atoi(argv[2]);  
  61. }  
  62.   
  63. #define CMD_FD 100  
  64. #define CMD_INFO 200  
  65. struct channel_msg{  
  66.     int command;  
  67.     int fd; // the fd, set to -1 if no fd.  
  68.     int total_client; // the total clients number of specified process.  
  69.     int param; // the param.  
  70. };  
  71. void write_channel(int sock, channel_msg* data, int size)  
  72. {  
  73.     msghdr msg;  
  74.       
  75.     // init msg_control  
  76.     if(data->fd == -1){  
  77.         msg.msg_control = NULL;  
  78.         msg.msg_controllen = 0;  
  79.     }  
  80.     else{  
  81.         union {  
  82.             struct cmsghdr cm;  
  83.             char space[CMSG_SPACE(sizeof(int))];  
  84.         } cmsg;  
  85.         memset(&cmsg, 0, sizeof(cmsg));  
  86.           
  87.         cmsg.cm.cmsg_level = SOL_SOCKET;  
  88.         cmsg.cm.cmsg_type = SCM_RIGHTS; // we are sending fd.  
  89.         cmsg.cm.cmsg_len = CMSG_LEN(sizeof(int));  
  90.           
  91.         msg.msg_control = (cmsghdr*)&cmsg;  
  92.         msg.msg_controllen = sizeof(cmsg);  
  93.         *(int *)CMSG_DATA(&cmsg.cm) = data->fd;  
  94.     }  
  95.       
  96.     // init msg_iov  
  97.     iovec iov[1];  
  98.     iov[0].iov_base = data;  
  99.     iov[0].iov_len  = size;  
  100.       
  101.     msg.msg_iov = iov;  
  102.     msg.msg_iovlen = 1;  
  103.       
  104.     // init msg_name  
  105.     msg.msg_name = NULL;  
  106.     msg.msg_namelen = 0;  
  107.       
  108.     info("[write_channel][info] data fd=" << data->fd << ", command="   
  109.         << data->command << ", total_client=" << data->total_client   
  110.         << ", param=" << data->param << endl);  
  111.           
  112.     while(true){  
  113.         ssize_t ret;  
  114.         if ((ret = sendmsg(sock, &msg, 0)) <= 0){  
  115.             // errno = 0xb Resource temporarily unavailable  
  116.             // donot retry, for it's no use, error infinite loop.  
  117.             err_exit("[write_channel] sendmsg error, ret=" << ret);  
  118.         }  
  119.         break;  
  120.     }  
  121. }  
  122. void read_channel(int sock, channel_msg* data, int size)  
  123. {  
  124.     msghdr msg;  
  125.       
  126.     // msg_iov  
  127.     iovec iov[1];  
  128.     iov[0].iov_base = data;  
  129.     iov[0].iov_len = size;  
  130.       
  131.     msg.msg_iov = iov;  
  132.     msg.msg_iovlen  = 1;  
  133.       
  134.     // msg_name  
  135.     msg.msg_name = NULL;  
  136.     msg.msg_namelen = 0;  
  137.       
  138.     // msg_control  
  139.     union { // union to create a 8B aligned memory.  
  140.         struct cmsghdr cm; // 16B = 8+4+4  
  141.         char space[CMSG_SPACE(sizeof(int))]; // 24B = 16+4+4  
  142.     } cmsg;  
  143.     memset(&cmsg, 0, sizeof(cmsg));  
  144.       
  145.     msg.msg_control = (cmsghdr*)&cmsg;  
  146.     msg.msg_controllen = sizeof(cmsg);  
  147.       
  148.     ssize_t ret;  
  149.     if ((ret = recvmsg(sock, &msg, 0)) == -1) {  
  150.         err_exit("[read_channel] recvmsg error");  
  151.     }  
  152.     if(ret == 0){  
  153.         info("[read_channel] connection closed" << endl);  
  154.         exit(0);  
  155.     }  
  156.       
  157.     info("[read_channel][info] data fd=" << data->fd << ", command="   
  158.         << data->command << ", total_client=" << data->total_client   
  159.         << ", param=" << data->param << endl);  
  160.     if(data->command == CMD_FD){  
  161.         int fd = *(int *)CMSG_DATA(&cmsg.cm);  
  162.         if(fd == 0 || fd == 1){  
  163.             err_exit("[read_channel] recvmsg invalid fd=" << fd   
  164.                 << ", data->fd=" << data->fd << ", command=" << data->command  
  165.                 << ", ret=" << ret);  
  166.         }  
  167.         data->fd = fd;  
  168.     }  
  169.     else{  
  170.         data->fd = -1;  
  171.     }  
  172. }  
  173.   
  174. void set_noblocking(int fd){  
  175.     int flag = 1;  
  176.     if(ioctl(fd, FIONBIO, &flag) == -1){  
  177.         err_exit("[master] ioctl failed!");  
  178.     }  
  179.     info("[master] server socket no-block flag set success" << endl);  
  180. }  
  181.   
  182. int listen_server_socket(UserOptions& options){  
  183.     int serverfd = socket(AF_INET, SOCK_STREAM, 0);  
  184.       
  185.     if(serverfd == -1){  
  186.         err_exit("[master] init socket error!");  
  187.     }  
  188.     info("[master] init socket success! #" << serverfd << endl);  
  189.       
  190.     int reuse_socket = 1;  
  191.     if(setsockopt(serverfd, SOL_SOCKET, SO_REUSEADDR, &reuse_socket, sizeof(int)) == -1){  
  192.         err_exit("[master] setsockopt reuse-addr error!");  
  193.     }  
  194.     info("[master] setsockopt reuse-addr success!" << endl);  
  195.       
  196.     set_noblocking(serverfd);  
  197.       
  198.     sockaddr_in addr;  
  199.     addr.sin_family = AF_INET;  
  200.     addr.sin_port = htons(options.port);  
  201.     addr.sin_addr.s_addr = INADDR_ANY;  
  202.     if(bind(serverfd, (const sockaddr*)&addr, sizeof(sockaddr_in)) == -1){  
  203.         err_exit("[master] bind socket error!");  
  204.     }  
  205.     info("[master] bind socket success!" << endl);  
  206.       
  207.     if(listen(serverfd, 10) == -1){  
  208.         err_exit("[master] listen socket error!");  
  209.     }  
  210.     info("[master] listen socket success! " << options.port << endl);  
  211.       
  212.     return serverfd;  
  213. }  
  214.   
  215. // both master and worker will invoke this handler.  
  216. void signal_handler(int signo, siginfo_t* info, void* context){  
  217.     info("#" << getpid() << " get a signal:");  
  218.       
  219.     if(signo == SIG_SOCKET_IO){  
  220.         cout << "SIG_SOCKET_IO!" << endl;  
  221.         global_channel_msg = true;  
  222.         current_active_io_fd = info->si_fd;  
  223.         info("[master] get SIGIO: fd=" << info->si_fd << endl);  
  224.         return;  
  225.     }  
  226.       
  227.     switch(signo){  
  228.         case SIGCHLD:  
  229.             cout << "SIGCHLD";  
  230.             break;  
  231.         case SIGALRM:  
  232.             cout << "SIGALRM";  
  233.             break;  
  234.         case SIGIO:  
  235.             cout << "SIGIO";  
  236.             info("[master][warning] ignore the SIGIO: rt-sig queue is full!" << endl);  
  237.             break;  
  238.         case SIGINT:  
  239.             cout << "SIGINT";  
  240.             global_terminate = true;  
  241.             break;  
  242.         case SIGHUP:  
  243.             cout << "SIGHUP";  
  244.             global_terminate = true;  
  245.             break;  
  246.         case SIGTERM:  
  247.             cout << "SIGTERM";  
  248.             global_terminate = true;  
  249.             break;  
  250.         case SIGQUIT:  
  251.             cout << "SIGQUIT";  
  252.             global_terminate = true;  
  253.             break;  
  254.         case SIGUSR1:  
  255.             cout << "SIGUSR1";  
  256.             break;  
  257.         case SIGUSR2:  
  258.             cout << "SIGUSR2";  
  259.             break;  
  260.     }  
  261.       
  262.     cout << "!" << endl;  
  263. }  
  264.   
  265. void register_signal_handler_imp(int signum, void (*handler)(int, siginfo_t*, void*)){  
  266.     struct sigaction action;  
  267.       
  268.     action.sa_flags = SA_SIGINFO;  
  269.     action.sa_sigaction = handler;  
  270.     sigemptyset(&action.sa_mask);  
  271.     //action.sa_flags = 0;  
  272.       
  273.     if(sigaction(signum, &action, NULL) == -1){  
  274.         err_exit("[master] register signal handler failed!");  
  275.     }  
  276. }  
  277. void register_signal_handler(){  
  278.     register_signal_handler_imp(SIGCHLD, signal_handler);  
  279.     register_signal_handler_imp(SIGALRM, signal_handler);  
  280.     register_signal_handler_imp(SIGIO, signal_handler);  
  281.     register_signal_handler_imp(SIG_SOCKET_IO, signal_handler);  
  282.     register_signal_handler_imp(SIGINT, signal_handler);  
  283.     register_signal_handler_imp(SIGHUP, signal_handler);  
  284.     register_signal_handler_imp(SIGTERM, signal_handler);  
  285.     register_signal_handler_imp(SIGQUIT, signal_handler);  
  286.     register_signal_handler_imp(SIGUSR1, signal_handler);  
  287.     register_signal_handler_imp(SIGUSR2, signal_handler);  
  288.     info("[master] register signal handler success!" << endl);  
  289. }  
  290.   
  291. void block_specified_signals(){  
  292.     sigset_t set;  
  293.       
  294.     sigemptyset(&set);  
  295.       
  296.     sigaddset(&set, SIGCHLD);  
  297.     sigaddset(&set, SIGALRM);  
  298.     sigaddset(&set, SIGIO);  
  299.     sigaddset(&set, SIG_SOCKET_IO);  
  300.     sigaddset(&set, SIGINT);  
  301.     sigaddset(&set, SIGHUP);  
  302.     sigaddset(&set, SIGTERM);  
  303.     sigaddset(&set, SIGQUIT);  
  304.     sigaddset(&set, SIGUSR1);  
  305.     sigaddset(&set, SIGUSR2);  
  306.       
  307.     if(sigprocmask(SIG_BLOCK, &set, NULL) == -1){  
  308.         err_exit("[master] sigprocmask block signal failed!");  
  309.     }  
  310.     info("[master] sigprocmask block signal success!" << endl);  
  311. }  
  312.   
  313. void unblock_all_signals(){  
  314.     sigset_t set;  
  315.     sigemptyset(&set);  
  316.       
  317.     // the master process block all signals, we unblock all for we use epoll to wait events.  
  318.     if(sigprocmask(SIG_SETMASK, &set, NULL) == -1){  
  319.         err_exit("[worker] sigprocmask unblock signal failed!");  
  320.     }  
  321.     info("[worker] sigprocmask unblock signal success!" << endl);  
  322. }  
  323.   
  324. void epoll_add_event(int ep, int fd){  
  325.     epoll_event ee;  
  326.     ee.events = EPOLLIN;  
  327.     ee.data.fd = fd;  
  328.     if(epoll_ctl(ep, EPOLL_CTL_ADD, fd, &ee) == -1){  
  329.         err_exit("[worker] epoll add event failed!");  
  330.     }  
  331.     current_client_num ++;  
  332.     info("[worker] epoll add fd success: #" << fd << ", current_client_num=" << current_client_num << endl);  
  333. }  
  334. void epoll_remove_event(int ep, int fd){  
  335.     if(epoll_ctl(ep, EPOLL_CTL_DEL, fd, NULL) == -1){  
  336.         err_exit("[worker] epoll remove event failed! fd=" << fd << ", errno=0x" << hex << errno << dec << " " << strerror(errno));  
  337.     }  
  338.     current_client_num --;  
  339.     info("[worker] epoll remove fd success: #" << fd << ", current_client_num=" << current_client_num << endl);  
  340. }  
  341.   
  342. void close_client(int ep, int fd, int serverfd, int worker_channel, bool do_report = true){  
  343.     info("[worker] the client dead, remove it: #" << fd << endl);  
  344.     epoll_remove_event(ep, fd);  
  345.     close(fd);  
  346.       
  347.     if(serverfd == fd){  
  348.         info("[worker] warning to close the server fd" << endl);  
  349.     }  
  350.       
  351.     // start accept again.  
  352.     if(!accepting && current_client_num < max_clients_per_process){  
  353.         epoll_add_event(ep, serverfd);  
  354.         accepting = true;  
  355.     }  
  356.       
  357.     // notice the master: worker can take more.  
  358.     if(do_report){  
  359.         channel_msg msg;  
  360.         msg.command = CMD_INFO;  
  361.         msg.total_client = current_client_num;  
  362.         msg.fd = -1;  
  363.         msg.param = 0;  
  364.         write_channel(worker_channel, &msg, sizeof(channel_msg));  
  365.     }  
  366. }  
  367.   
  368. void worker_get_event(int ep, int serverfd, epoll_event& active, int worker_channel){  
  369.     info("[worker] process #" << getpid() << " current_client_num=" << current_client_num << ", worker_channel=" << worker_channel << endl);  
  370.     int fd = active.data.fd;  
  371.       
  372.     // server listening socket event.  
  373.     if(fd == serverfd){  
  374.         if(current_client_num < physical_max_clients){  
  375.             int client = accept(serverfd, NULL, 0);  
  376.             if(client == -1){  
  377.                 // thundering herd  
  378.                 // http://www.citi.umich.edu/projects/linux-scalability/reports/accept.html  
  379.                 if(errno == EAGAIN || errno == EWOULDBLOCK){  
  380.                     info("[worker][warning] get a thundering herd at #" << getpid() << endl);  
  381.                     return;  
  382.                 }  
  383.                 err_exit("[worker] accept client socket error!");  
  384.             }  
  385.               
  386.             int reuse_socket = 1;  
  387.             if(setsockopt(client, SOL_SOCKET, SO_REUSEADDR, &reuse_socket, sizeof(int)) == -1){  
  388.                 err_exit("[worker] setsockopt reuse-addr error!");  
  389.             }  
  390.             info("[worker] setsockopt reuse-addr success!" << endl);  
  391.               
  392.             info("[worker] process #" << getpid() << " get a client: #" << client << endl);  
  393.             epoll_add_event(ep, client);  
  394.         }  
  395.         else{  
  396.             epoll_remove_event(ep, fd); // donot accept.  
  397.             accepting = false;  
  398.             info("[worker] warning, the worker #" << getpid()   
  399.                 << " exceed the max clients, current_client_num=" << current_client_num << endl);  
  400.         }  
  401.         return;  
  402.     }  
  403.       
  404.     // channel event.  
  405.     if(fd == worker_channel){  
  406.         channel_msg msg;  
  407.         read_channel(worker_channel, &msg, sizeof(channel_msg));  
  408.         info("[worker] get a channel message. fd=" << msg.fd << ", command=" << msg.command  
  409.             << ", current_client_num=" << current_client_num << ", max_clients_per_process=" << max_clients_per_process  
  410.             << ", physical_max_clients=" << physical_max_clients << endl);  
  411.         if(msg.command == CMD_FD && msg.fd != -1){  
  412.             if(msg.fd == 0 || msg.fd == 1){  
  413.                 info("[worker][warning] ignore the invalid passing fd=" << msg.fd << endl);  
  414.                 return;  
  415.             }  
  416.             // accept all client from master, util the read overloaded.  
  417.             if(current_client_num < physical_max_clients){  
  418.                 epoll_add_event(ep, msg.fd);  
  419.                 info("[worker] get a dispatched client. #" << msg.fd << endl);  
  420.             }  
  421.             else{  
  422.                 info("[worker][warning] worker overloaded, close the fd: " << msg.fd << endl);  
  423.                 close(msg.fd);  
  424.             }  
  425.         }  
  426.         else{  
  427.             err_exit("[worker] get a invalid channel");  
  428.         }  
  429.         return;  
  430.     }  
  431.       
  432.     // client event.  
  433.     if((active.events & EPOLLHUP) == EPOLLHUP || (active.events & EPOLLERR) == EPOLLERR){  
  434.         info("[worker] get a EPOLLHUP or EPOLLERR event from client #" << fd << endl);  
  435.         close_client(ep, fd, serverfd, worker_channel);  
  436.         return;  
  437.     }  
  438.     if((active.events & EPOLLIN) == EPOLLIN){  
  439.         if(true){  
  440.             info("[worker] get a EPOLLIN event from client #" << fd << endl);  
  441.             char ch_control; // we MUST not read more bytes, for the fd maybe passing to other process!  
  442.             if(recv(fd, &ch_control, sizeof(char), 0) <= 0){  
  443.                 close_client(ep, fd, serverfd, worker_channel);  
  444.                 return;  
  445.             }  
  446.             // check the first byte, if 'M' it's message, if 'C' it's control message.  
  447.             if(ch_control == 'C'){  
  448.                 int client_required_id; //worker_channel  
  449.                 if(recv(fd, &client_required_id, sizeof(int), 0) <= 0){  
  450.                     close_client(ep, fd, serverfd, worker_channel);  
  451.                     return;  
  452.                 }  
  453.               
  454.                 pid_t pid = getpid();  
  455.                 info("[worker] get client control message: client_required_id=" << client_required_id << endl);  
  456.                 if((client_required_id % pid) == 0){  
  457.                     info("[worker][verified] client exactly required this server" << endl);  
  458.                 }  
  459.                 else{  
  460.                     // never send the follows:  
  461.                     if(fd == 0 || fd == 1 || fd == serverfd || fd == worker_channel){  
  462.                         info("[worker][warning] ignore invalid fd to passing: " << fd << endl);  
  463.                         close_client(ep, fd, serverfd, worker_channel);  
  464.                         return;  
  465.                     }  
  466.                     info("[worker] send to master to dispatch it" << endl);  
  467.                     channel_msg msg;  
  468.                     msg.fd = fd;  
  469.                     msg.command = CMD_FD;  
  470.                     msg.total_client = current_client_num;  
  471.                     msg.param = client_required_id;  
  472.                     write_channel(worker_channel, &msg, sizeof(channel_msg));  
  473.                     // donot report to master, for the CMD_FD message has reported.  
  474.                     close_client(ep, fd, serverfd, worker_channel, false);  
  475.                     return;  
  476.                 }  
  477.             }  
  478.             else{  
  479.                 char buf[1024];  
  480.                  if(recv(fd, buf, sizeof(buf), 0) <= 0){  
  481.                     close_client(ep, fd, serverfd, worker_channel);  
  482.                     return;  
  483.                 }  
  484.                info("[worker] get client data message: " << buf << endl);  
  485.             }  
  486.         }  
  487.         if(true){  
  488.             char msg[] = "hello, server ping!";  
  489.             if(send(fd, msg, sizeof(msg), 0) <= 0){  
  490.                 close_client(ep, fd, serverfd, worker_channel);  
  491.                 return;  
  492.             }  
  493.         }  
  494.         return;  
  495.     }  
  496. }  
  497.   
  498. void worker_process_cycle(int serverfd, int worker_channel){  
  499.     cout << "[worker] start worker process cycle. serverfd=" << serverfd   
  500.         << ", worker_channel=" << worker_channel   
  501.         << ((worker_channel == -1)? "(single process)" : "(multiple processes)")<< endl;  
  502.     unblock_all_signals();  
  503.   
  504.     int ep = epoll_create(1024);  
  505.     if(ep == -1){  
  506.         err_exit("[worker] epoll_create failed!");  
  507.     }  
  508.     info("[worker] epoll create success!" << endl);  
  509.     epoll_add_event(ep, serverfd);  
  510.     epoll_add_event(ep, worker_channel);  
  511.       
  512.     // worker use epoll event, the signal handler will break the epoll_wait to return -1.  
  513.     for(;;){  
  514.         epoll_event events[1024];  
  515.         int incoming = epoll_wait(ep, events, 1024, -1);  
  516.           
  517.         // get a event or error.  
  518.         if(incoming == -1){  
  519.             if(global_terminate){  
  520.                 close(worker_channel);  
  521.                 close(serverfd);  
  522.                 info("[worker] worker process exit" << endl);  
  523.                 exit(0);  
  524.             }  
  525.             break;  
  526.         }  
  527.           
  528.         for(int i = 0; i < incoming; i++){  
  529.             worker_get_event(ep, serverfd, events[i], worker_channel);  
  530.         }  
  531.     }  
  532.       
  533.     info("[worker] worker process exited" << endl);  
  534.     close(ep);  
  535. }  
  536.   
  537. struct WorkerProcess{  
  538.     pid_t pid;  
  539.     int channel;  
  540.     int num_clients;  
  541. };  
  542.   
  543. void start_worker_process(int serverfd, WorkerProcess& worker_process){  
  544.     // master/worker use domain socket to communicate.  
  545.     int fds[2];  
  546.     if(socketpair(AF_UNIX, SOCK_STREAM, 0, fds) == -1){  
  547.         err_exit("[master] sockpair create domain socket failed");  
  548.     }  
  549.     info("[master] sockpair create domain socket success! [" << fds[0] << ", " << fds[1] << "]" << endl);  
  550.       
  551.     // master process, use the fds[0].  
  552.     worker_process.num_clients = 0;  
  553.     worker_process.channel = fds[0];  
  554.       
  555.     /** 
  556.     * SIGIO (http://www.lindevdoc.org/wiki/SIGIO) 
  557.     * The SIGIO signal is sent to a process to notify it about some event on a file descriptor, namely: 
  558.     *   the file descriptor is ready to receive data (previous write has finished). 
  559.     *   the file descriptor has new data to read 
  560.     *   an error has occurred 
  561.     * This signal is not sent by default; it must be enabled by specifying the O_ASYNC flag to the descriptor 
  562.     * 
  563.     * O_ASYNC (http://www.lindevdoc.org/wiki/O_ASYNC) 
  564.     * The O_ASYNC flag specifies that the process that owns a data stream (by default the process that opened it,  
  565.     * but can be changed with fcntl() with the F_SETOWN command) will receive a signal (SIGIO by default, but can  
  566.     * be changed with fcntl() with the F_SETSIG command) when the file is ready for reading or writing. 
  567.     */  
  568.     // set the socket owner to master, to get a SIGIO signal.  
  569.     if(fcntl(worker_process.channel, F_SETOWN, getpid()) == -1){  
  570.         err_exit("[master] fcntl F_SETOWN to master failed");  
  571.     }  
  572.     info("[master] fcntl F_SETOWN to master success" << endl);  
  573.     // set socket to async to singal SIGIO when data coming.  
  574.     int on = 1;  
  575.     if(ioctl(worker_process.channel, FIOASYNC, &on) == -1){  
  576.         err_exit("[master] ioctl set FIOASYNC failed");  
  577.     }  
  578.     info("[master] ioctl set FIOASYNC success" << endl);  
  579.     /** F_SETSIG 
  580.     * if donot use set_sig, we must use select to find which fd is read. 
  581.     * http://ajaxxx.livejournal.com/62378.html 
  582.     * Linux has the additional feature of fcntl(F_SETSIG), which lets you get the file descriptor that generated the  
  583.     * signal back as part of the signal information. So I implemented this, with the obvious semantics: receiving a  
  584.     * SIGIO would call the handler for just that file descriptor and then return. 
  585.     * remark: the SIGRTMIN=34, SIGRTMAX=62, so only 28 fd cannbe discoveried. so it's used for category signal,  
  586.     *       and we use SA_SIGINFO for sigaction to get the fd of SIGIO.  
  587.     */  
  588.     if(fcntl(worker_process.channel, F_SETSIG, SIG_SOCKET_IO) == -1){  
  589.         err_exit("[master] fcntl set rtsig failed: F_SETSIG.");  
  590.     }  
  591.     info("[master] fcntl set rtsig success" << endl);  
  592.     // set to unblocking.  
  593.     set_noblocking(worker_process.channel);  
  594.     set_noblocking(fds[1]);  
  595.       
  596.     // start process by fork.  
  597.     pid_t pid = fork();  
  598.       
  599.     if(pid == -1){  
  600.         err_exit("[master] fork process failed");  
  601.     }  
  602.       
  603.     // worker process, use the fds[1].  
  604.     if(pid == 0){  
  605.         close(fds[0]);  
  606.         worker_process_cycle(serverfd, fds[1]);  
  607.         exit(0);  
  608.     }  
  609.     // master process.  
  610.     worker_process.pid = pid;  
  611.     close(fds[1]);  
  612.       
  613.     info("[master] fork process success: #" << pid << ", channel=" << worker_process.channel << endl);  
  614. }  
  615.   
  616. int find_the_required_worker(int serverfd, vector<WorkerProcess>& workers, int client_required_id, int pre_pid){  
  617.     // find the exactly worker.  
  618.     WorkerProcess* target = NULL;  
  619.     int channel = -1;  
  620.     for(vector<WorkerProcess>::iterator ite = workers.begin(); ite != workers.end(); ++ite){  
  621.         WorkerProcess& worker = *ite;  
  622.         if((client_required_id % worker.pid) == 0){  
  623.             target = &worker;  
  624.             break;  
  625.         }  
  626.     }  
  627.     // find the most idling worker.  
  628.     int pre_num = 0;  
  629.     for(vector<WorkerProcess>::iterator ite = workers.begin(); ite != workers.end(); ++ite){  
  630.         WorkerProcess& worker = *ite;  
  631.         if(worker.num_clients < max_clients_per_process && worker.pid != pre_pid){  
  632.             if(pre_num == 0 || worker.num_clients < pre_num){  
  633.                 target = &worker;  
  634.                 pre_num = worker.num_clients;  
  635.             }  
  636.         }  
  637.     }  
  638.       
  639.     if(target == NULL){  
  640.         if(workers.size() < max_workers){  
  641.             // start a new worker  
  642.             WorkerProcess worker;  
  643.             start_worker_process(serverfd, worker);  
  644.             workers.push_back(worker);  
  645.               
  646.             return find_the_required_worker(serverfd, workers, client_required_id, pre_pid);  
  647.         }  
  648.         else{  
  649.             err_exit("[master] all " << workers.size() << " workers overloaded!");  
  650.         }  
  651.     }  
  652.     info("[master] find a target to passing fd, pid=" << target->pid << ", num_clients=" << target->num_clients << endl);  
  653.     target->num_clients ++;  
  654.       
  655.     return target->channel;  
  656. }  
  657. void on_channel_message(int serverfd, WorkerProcess& host, vector<WorkerProcess>& workers){  
  658.     int sock = host.channel;  
  659.       
  660.     channel_msg msg;  
  661.     read_channel(sock, &msg, sizeof(channel_msg));  
  662.     host.num_clients = msg.total_client;  
  663.       
  664.     cout << "[master] get a channel msg, pid=#" << host.pid << ", fd=" << msg.fd   
  665.         << ", total_client=" << msg.total_client  
  666.         << ", param=" << msg.param << ", command=" << msg.command << endl;  
  667.     if(msg.command == CMD_FD){  
  668.         // dispatch and re-dispatch fd  
  669.         int channel = find_the_required_worker(serverfd, workers, msg.param, host.pid);  
  670.         msg.command = CMD_FD;  
  671.         if(msg.fd == 0 || msg.fd == 1){  
  672.             info("[master][warning] ignore invalid fd to passing: " << msg.fd << endl);  
  673.             return;  
  674.         }  
  675.         write_channel(channel, &msg, sizeof(channel_msg));  
  676.         // direct close the fd, the target worker MUST take it or close it.  
  677.         close(msg.fd);  
  678.         host.num_clients--;  
  679.     }  
  680.     else if(msg.command == CMD_INFO){  
  681.         return;  
  682.     }  
  683.     else{  
  684.         err_exit("[master] invalid response!");  
  685.     }  
  686. }  
  687.   
  688. int main(int argc, char** argv){  
  689.     info("[master] SIGRTMIN=0x" << hex << SIGRTMIN   
  690.         << ", SIGRTMAX=0x" << hex << SIGRTMAX << dec   
  691.         << ", SIGIO=0x" << hex << SIGIO << dec   
  692.         << ", SIGPOLL=0x" << hex << SIGPOLL << dec   
  693.         << endl);  
  694.     register_signal_handler();  
  695.     block_specified_signals();  
  696.       
  697.     //sleep(3);  
  698.     UserOptions options;  
  699.     discovery_user_options(argc, argv, options);  
  700.     int serverfd = listen_server_socket(options);  
  701.       
  702.     // single process mode  
  703.     if(options.num_processes == 0){  
  704.         worker_process_cycle(serverfd, -1);  
  705.     }  
  706.     // multiple processes mode  
  707.     else{  
  708.         vector<WorkerProcess> workers;  
  709.         for(int i = 0; i < options.num_processes; i++){  
  710.             WorkerProcess worker;  
  711.             start_worker_process(serverfd, worker);  
  712.             workers.push_back(worker);  
  713.         }  
  714.           
  715.         // the empty set for sigsuspend  
  716.         sigset_t set;  
  717.         sigemptyset(&set);  
  718.         // master process use signal only.  
  719.         for(;;){  
  720.             sigsuspend(&set);  
  721.               
  722.             if(global_terminate){  
  723.                 // kill all workers.  
  724.                 for(vector<WorkerProcess>::iterator ite = workers.begin(); ite != workers.end(); ++ite){  
  725.                     WorkerProcess& worker = *ite;  
  726.                     kill(worker.pid, SIGTERM);  
  727.                 }  
  728.                 // wait for workers to exit.  
  729.                 for(vector<WorkerProcess>::iterator ite = workers.begin(); ite != workers.end(); ++ite){  
  730.                     WorkerProcess& worker = *ite;  
  731.                     int status;  
  732.                     waitpid(worker.pid, &status, 0);  
  733.                 }  
  734.                 info("[master] all worker terminated, master exit" << endl);  
  735.                 exit(0);  
  736.             }  
  737.             if(global_channel_msg){  
  738.                 int active_fd = current_active_io_fd;  
  739.                 current_active_io_fd = -1;  
  740.                 global_channel_msg = false;  
  741.                   
  742.                 // read each channel.  
  743.                 for(vector<WorkerProcess>::iterator ite = workers.begin(); ite != workers.end(); ++ite){  
  744.                     WorkerProcess& worker = *ite;  
  745.                     if(worker.channel == active_fd){  
  746.                         on_channel_message(serverfd, worker, workers);  
  747.                         break;  
  748.                     }  
  749.                 }  
  750.             }  
  751.         }  
  752.     }  
  753.   
  754.     return 0;  
  755. }  

[cpp]  view plain copy
  1. // mpclient2 —— multiple processes client  
  2. #include <stdio.h>  
  3. #include <stdlib.h>  
  4. #include <iostream>  
  5. using namespace std;  
  6.   
  7. #include <unistd.h>  
  8. #include <signal.h>  
  9. #include <sys/types.h>  
  10. #include <sys/socket.h>  
  11. #include <netinet/in.h>  
  12. #include <arpa/inet.h>  
  13. #include <string.h>  
  14.   
  15. #define err_exit(msg) cout << "[error] " << msg << endl; exit(1)  
  16.   
  17. struct UserOptions{  
  18.     char* server_ip;  
  19.     int port;  
  20.     int num_clients;  
  21.     int sleep_ms;  
  22. };  
  23.   
  24. void discovery_user_options(int argc, char** argv, UserOptions& options){  
  25.     if(argc <= 4){  
  26.         cout << "Usage: " << argv[0] << " <server_ip> <port> <num_clients> <sleep_ms>" << endl  
  27.             << "server_ip: the ip address of server" << endl  
  28.             << "port: the port to connect at" << endl  
  29.             << "num_clients: the num to start client." << endl  
  30.             << "sleep_ms: the time to sleep in ms." << endl  
  31.             << "for example: " << argv[0] << " 127.0.0.1 1990 1000 500" << endl;  
  32.         exit(1);  
  33.     }  
  34.       
  35.     options.server_ip = argv[1];  
  36.     options.port = atoi(argv[2]);  
  37.     options.num_clients = atoi(argv[3]);  
  38.     options.sleep_ms = atoi(argv[4]);  
  39.     if(options.num_clients <= 0){  
  40.         err_exit("num_clients must > 0.");  
  41.     }  
  42. }  
  43.   
  44. int connect_server_socket(UserOptions& options){  
  45.     int clientfd = socket(AF_INET, SOCK_STREAM, 0);  
  46.       
  47.     if(clientfd == -1){  
  48.         err_exit("init socket error!");  
  49.     }  
  50.     cout << "init socket success!" << endl;  
  51.       
  52.     sockaddr_in addr;  
  53.     addr.sin_family = AF_INET;  
  54.     addr.sin_port = htons(options.port);  
  55.     addr.sin_addr.s_addr = inet_addr(options.server_ip);  
  56.     if(connect(clientfd, (const sockaddr*)&addr, sizeof(sockaddr_in)) == -1){  
  57.         err_exit("connect socket error!");  
  58.     }  
  59.     cout << "connect socket success!" << endl;  
  60.       
  61.     return clientfd;  
  62. }  
  63.   
  64. int main(int argc, char** argv){  
  65.     srand(0);  
  66.     UserOptions options;  
  67.     discovery_user_options(argc, argv, options);  
  68.     int* fds = new int[options.num_clients];  
  69.     for(int i = 0; i < options.num_clients; i++){  
  70.         fds[i] = connect_server_socket(options);  
  71.         usleep(10 * 1000); // to prevent the SIGIO error: maybe signal queue overflow!  
  72.     }  
  73.       
  74.     for(int i = 0; i < options.num_clients; i++){  
  75.         int fd = fds[i];  
  76.         if(true){  
  77.             char c = 'C'; // control message  
  78.             if(send(fd, &c, sizeof(char), 0) <= 0){  
  79.                 close(fd);  
  80.                 err_exit("send message failed!");  
  81.             }  
  82.             int required_id = rand();  
  83.             if(send(fd, &required_id, sizeof(int), 0) <= 0){  
  84.                 close(fd);  
  85.                 err_exit("send message failed!");  
  86.             }  
  87.         }  
  88.     }  
  89.       
  90.     bool do_loop = true;  
  91.     while(do_loop){  
  92.         int ret = 0;  
  93.         for(int i = 0; i < options.num_clients; i++){  
  94.             int fd = fds[i];  
  95.             if(true){  
  96.                 char c = 'M'; // data message  
  97.                 ret = send(fd, &c, sizeof(char), 0);  
  98.                 if(ret <= 0){  
  99.                     cout << "send control message to server error" << endl;  
  100.                     close(fd);  
  101.                     do_loop = false;  
  102.                     break;  
  103.                 }  
  104.                 char msg[] = "client, ping message!";  
  105.                 ret = send(fd, msg, sizeof(msg), 0);  
  106.                 if(ret <= 0){  
  107.                     cout << "send control value to server error" << endl;  
  108.                     close(fd);  
  109.                     do_loop = false;  
  110.                     break;  
  111.                 }  
  112.                 memset(&msg, 0, sizeof(msg));  
  113.             }  
  114.             if(true){  
  115.                 char buf[1024];  
  116.                 memset(buf, 0, sizeof(buf));  
  117.                 ret = recv(fd, buf, sizeof(buf), 0);  
  118.                 if(ret <= 0){  
  119.                     cout << "recv from server error" << endl;  
  120.                     close(fd);  
  121.                     do_loop = false;  
  122.                     break;  
  123.                 }  
  124.                 cout << "recv from server: " << buf << endl;  
  125.             }  
  126.               
  127.             usleep(options.sleep_ms * 1000);  
  128.         }  
  129.     }  
  130.       
  131.     delete[] fds;  
  132.   
  133.     return 0;  
  134. }  

Logo
Linux

更多推荐

网卡速率和双工模式的配置

http://linux.chinaitlab.com/system/792187.html1、mii-tool 配置网络设备协商方式的工具；    1.1 mii-tool 介绍；    mii-tool - view, manipulate media-independent interface status （mii-tool 是查看，管理介质的网络接口的状态）

avatar Linux

Linux虚拟文件系统之文件系统卸载（sys_umount())

Linux中卸载文件系统由umount系统调用实现，入口函数为sys_umount()。较于文件系统的安装较为简单，下面是具体的实现。1. /*sys_umont系统调用*/2. SYSCALL_DEFINE2(umount, char __user *, name, int, flags)3. {4.struct path path;

avatar Linux

Linux系统下超级终端Minicom的使用方法（例如：连接交换机，路由器等）转http://baike.baidu.com/view/2911642.htm?fr=ala0_1

<br />　　Linux系统下超级终端Minicom的使用方法 <br />　　Linux下的Minicom的功能与下的超级终端功能相似，适于在通过超级终端对设备的管理以及对嵌入操作系统的升级，现写出Minicom的使用手册： <br />　　1． 启动minicom <br />　　以root权限登录系统 <br />　　使用命令 <br />　　minicom –s 则minicom启动，屏

avatar Linux