父进程listen，子进程epoll监听事件，客户端会超时，请高手指教。

Question

设计的一个server，开始是单进程模式的，2w/s的访问量没有问题。但为了充分利用多cpu，想把它改成多进程方式。

改动不大，主要就是父进程先创建listen对象，监听端口，然后fork子进程，每个子进程创建自己的epoll，监听listen_fd、accept_fd，以及跟后台各server交互的fd。

问题来了：比如我们起4个子进程（每个cpu一个），然后用10个线程客户端去压力测试，在开始测试的时候，客户端总会有少量的接收超时，没有收到应答包。但是过一段事件之后，就会跑的比较平稳，没有超时，压力也很固定。

调试的时候，关闭了到后台各Server的交互，收到包之后，本地处理就回包，也会有少量的超时。不知道这是怎么回事？

静待高手解答。不要告诉我换多线程，或者父进程epoll监听listen_fd的实现，哥懂那个，只是想明白这种设计范式下的问题可能是什么原因。

另外，据说linux 2.6内核版本已经解决了此设计模式下的惊群问题，但是否会有消息乱序，即A子进程的fd加入到了B子进程的epoll事件里去。

简化的实例代码如下：

3. /* 单进程可以监听本地多端口，每个端口创建一个对象，用vector保存起来，用于传递到子进程里，每个子进程去创建epoll监听listen_fd */
4. vector<CPollerObject*> g_listen_obj;
7. int InitGlobalInstance(){
8.         pid_t pid = getpid ();
9.         int iRet;
10.        
11.         /* 启动epoll */
12.         iRet = CPollerUnit::Ins()->Create(g_conf.m_epoll._maxfd, g_conf.m_epoll._timeout);
13.         if( E_OK!=iRet ){
14.                 fprintf(stderr, "Create Epoll Error:%d\n", iRet);
15.                 return E_FAIL;
16.         }
17.         DEBUG_LOG("CPollerUnit Create Succ: %p", CPollerUnit::Ins());
18.        
19.         for( vector<CPollerObject*>::iterator it=g_listen_obj.begin(); it!=g_listen_obj.end(); it++){
20.                 (*it)->EnableInput();
21.                 (*it)->AttachPoller();
22.         }
23.        
24.         /* 初始化内存池管理 */
25.         iRet = CMemPool::Ins()->Create(g_conf.m_session._bufSize);
26.         if( E_OK!=iRet ){
27.                 fprintf(stderr, "Create Epoll Error:%d\n", iRet);
28.                 return E_FAIL;
29.         }
30.         DEBUG_LOG("CMemPool Create Succ: %p", CMemPool::Ins());
32.         /* 初始化连接池 */
33.         iRet = CTcpPoll::Ins()->Init(g_conf.m_session._max, g_conf.m_errReport._failCount, g_conf.m_errReport._timeInterval);
34.         if( E_OK!=iRet ){
35.                 fprintf(stderr, "Create CTcpPoll Error:%d\n", iRet);
36.                 return E_FAIL;
37.         }
38.         DEBUG_LOG("CTcpPoll Create Succ: %p", CTcpPoll::Ins());
40.         /* 初始化统计对象。 */
41.         char pathWithPid[512] = {'\0'};
42.         snprintf(pathWithPid, sizeof(pathWithPid)-1, "%s_%d", g_conf.m_stat._prefix, pid);
43.         iRet = CProxyStat::Instance().Init(g_conf.m_stat._path, pathWithPid);
44.         if( E_OK!=iRet ){
45.                 fprintf(stderr, "Create CProxyStat Error:%d\n", iRet);
46.                 return E_FAIL;               
47.         }
48.         DEBUG_LOG("CProxyStat Create Succ: %p", &(CProxyStat::Instance()));
50.         return 0;
51. }
53. int MainThread()
54. {
55.         InitGlobalInstance();
57.         time_t tNow = time(NULL);
58.         while(true){
59.                 _g_child_flag = 1;
60.                 /* 处理连接任务 */
61.                 CPollerUnit::Ins()->ProcessAllEvents();
62.                        
63.                 /* 超时检查*/
64.                 CSession::Ins()->CheckTimeOut();
66.                 /* 清理空闲的tcp连接。 */
67.                 CTcpPoll::Ins()->CheckIdle(g_conf.m_session._idle, g_conf.m_session._timeInterval, g_conf.m_session._emptyRatio);
69.                 /* 检查是否到了统计间隔 */
70.                 if( int(time(NULL) - tNow) < g_conf.m_stat._interval) continue;
71.                 CProxyStat::Instance().ProcessStat();
72.                 tNow = time(NULL);
73.         }
74.        
75.         return -1;
76. };
79. int fork_process_group()
80. {
81.         //process group配置
82.         int cpu_process_num = sysconf(_SC_NPROCESSORS_CONF);
83.         int process_group_num        = g_conf.m_iProcessNum;       
84.        
85.         /* 不指定进程个数，或者指定进程数比cpu个数还多，修改为cpu的个数。 */
86.         if( 0 == process_group_num || process_group_num > cpu_process_num )
87.                 process_group_num = cpu_process_num;
89.         ERROR_LOG( "cpu num: %d, config process num: %d, real process num: %d", cpu_process_num, g_conf.m_iProcessNum, process_group_num);
91.         int pid;
92.         for(int i = 0; i < process_group_num ; i++)
93.         {
94.                 pid = fork();
95.                 switch(pid)
96.                 {
97.                         case -1:
98.                                 ERROR_LOG("fork error!");
99.                                 return -1;
101.                         case 0:
102.                                 SetRunCpu( i );
103.                                 MainThread();
104.                                 return 0;
106.                         default:
107.                                 g_processcpu.insert(make_pair(pid, i));
108.                 };
109.         }
111.         return 0;
112. }
115. int main(int argc,char** argv)
116. {
117.         if( argc <2 ){
118.                 fprintf(stderr, "usage:%s etc\n", argv[0]);
119.                 return -1;
120.         }
122.         /* 显示系统说明。 */
123.         if( 0==strcasecmp(argv[1], "-v") ){
124.                 PrintVersion();
125.                 return 0;
126.         }
127.         PrintVersion();
129.         /* 初始化，读取配置文件。 */
130.         int iRet = g_conf.Init(argv[1]);
131.         if( E_OK != iRet ){
132.                 fprintf(stderr, "Init Conf Error!:%d\n", iRet);
133.                 return E_FAIL;
134.         }
136.         /* 初始化log */
137.         log_init(g_conf.m_log._path, g_conf.m_log._level, g_conf.m_log._size, g_conf.m_log._prefix);
139.         /* 设置信号 */
140.         SetUser1Handle();
141.         /* 绑定cpu */
142.         //if ( g_conf.m_iCpuBind>= 0 )        SetRunCpu(g_conf.m_iCpuBind);       
145.         /* 监听端口 */
146.         if ( CreateClientSocket() < 0 )
147.         {
148.                 fprintf(stderr, "CreateClientSocket\n");
149.                 return E_FAIL;
150.         }
151.         DEBUG_LOG("CreateClientSocket Succ!");
153.         /* fork进程，设置为守护进程。 */
154.         daemon_init();
156.         fork_process_group();
158. #ifndef DEBUG       
159.         pid_t pid;
160.         for(;;)
161.         {
162.                 pid = waitpid(-1,NULL,0);
163.                 if ( pid <= 0 )
164.                 {
165.                         ERROR_LOG("waitpid error:%s\n", strerror(errno));
166.                         sleep(1);
167.                         continue;
168.                 }
169.                 fflush(NULL);
170.                
171.                 ERROR_LOG("WorkMain[%d] restart...................................\n", pid);
172.                 usleep(100000);
173.                
174.                 int cpunum = g_processcpu.find(pid) == g_processcpu.end()?  0 : g_processcpu[pid];
175.                 int newpid = fork();
176.                 switch(newpid){
177.                         case -1:
178.                                 ERROR_LOG("fork error\n");
179.                                 break;
180.                         case 0:
181.                                 SetRunCpu(cpunum);
182.                                 MainThread();
183.                                 break;
184.                         default:
185.                                 g_processcpu[newpid] = cpunum;
186.                                 break;
187.                 }
188.         }
189. #endif
190.        
191.         return 0;
192. }

复制代码

Answer 1

我没仔细看你的代码, 能说下主进程listen accept后怎么将socket fd传给子进程的?
我估计是这块引起的超时.

假设一个客户端被主进程accept后, 子进程没去epoll进行读写的监听，可能会引起超时.

Answer 2

这个我之前也是一直不懂 等人指教

Answer 3

嗯。。没有人来指教一下么。。

Answer 4

Answer 5

加日志查看了一下，消息并没有乱掉，但是epoll已经通知不上来事件。一个发送，或者接收，3s以上都没有触发epoll。

还是不明白这个问题到底是怎么产生的。

ps：修改了代码框架，主进程里增加一个线程进行listen，accept到的fd发给子进程，终于没有问题了，性能达到6w/s

Answer 6

期待内核级人物指点，呵呵。

Answer 7

看了你的描述和代码，说两点：
1 消息乱序理论上是不会的，每个fd只在一个epollfd中管理
2 超时现象比较奇怪，看了你的代码，没看出什么问题，照理说应该不会有这个情况，lighthttp用的进程模型和你是一样的。

Answer 8

是epoll的同一个listenFD吗
会不会取等待队列里的连接的时候出现竞争
好像不对  listenFD的引用计数增加而已

Answer 9

mark
如楼主所说，我们公司框架就是在主进程中进行监听，收到后发子进程。不过性能还远远不及楼主的。