父進(jìn)程listen，子進(jìn)程epoll監(jiān)聽(tīng)事件的問(wèn)題，各位高手請(qǐng)進(jìn)。

oscarzhou

設(shè)計(jì)的一個(gè)server，開(kāi)始是單進(jìn)程模式的，2w/s的訪問(wèn)量沒(méi)有問(wèn)題。但為了充分利用多cpu，想把它改成多進(jìn)程方式。

改動(dòng)不大，主要就是父進(jìn)程先創(chuàng)建listen對(duì)象，監(jiān)聽(tīng)端口，然后fork子進(jìn)程，每個(gè)子進(jìn)程創(chuàng)建自己的epoll，監(jiān)聽(tīng)listen_fd、accept_fd，以及跟后臺(tái)各server交互的fd。

問(wèn)題來(lái)了：比如我們起4個(gè)子進(jìn)程（每個(gè)cpu一個(gè)），然后用10個(gè)線程客戶端去壓力測(cè)試，在開(kāi)始測(cè)試的時(shí)候，客戶端總會(huì)有少量的接收超時(shí)，沒(méi)有收到應(yīng)答包。但是過(guò)一段事件之后，就會(huì)跑的比較平穩(wěn)，沒(méi)有超時(shí)，壓力也很固定。

調(diào)試的時(shí)候，關(guān)閉了到后臺(tái)各Server的交互，收到包之后，本地處理就回包，也會(huì)有少量的超時(shí)。

不知道這是怎么回事？靜待高手解答。不要告訴我換多線程，或者父進(jìn)程epoll監(jiān)聽(tīng)listen_fd的實(shí)現(xiàn)，哥懂那個(gè)，只是想明白這種設(shè)計(jì)范式下的問(wèn)題可能是什么原因。

另外，據(jù)說(shuō)linux 2.6內(nèi)核版本已經(jīng)解決了此設(shè)計(jì)模式下的驚群?jiǎn)栴}，但是否會(huì)有消息亂序，即A子進(jìn)程的fd加入到了B子進(jìn)程的epoll事件里去。


簡(jiǎn)化的實(shí)例代碼如下：

1. 
   
2. 
   
3. /* 單進(jìn)程可以監(jiān)聽(tīng)本地多端口，每個(gè)端口創(chuàng)建一個(gè)對(duì)象，用vector保存起來(lái)，用于傳遞到子進(jìn)程里，每個(gè)子進(jìn)程去創(chuàng)建epoll監(jiān)聽(tīng)listen_fd */
   
4. vector<CPollerObject*> g_listen_obj;
   
5. 
   
6. 
   
7. int InitGlobalInstance(){
   
8.         pid_t pid = getpid ();
   
9.         int iRet;
   
10.        
    
11.         /* 啟動(dòng)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.         /* 初始化內(nèi)存池管理 */
    
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());
    
31. 
    
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());
    
39. 
    
40.         /* 初始化統(tǒng)計(jì)對(duì)象。 */
    
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()));
    
49. 
    
50.         return 0;
    
51. }
    
52. 
    
53. int MainThread()
    
54. {
    
55.         InitGlobalInstance();
    
56. 
    
57.         time_t tNow = time(NULL);
    
58.         while(true){
    
59.                 _g_child_flag = 1;
    
60.                 /* 處理連接任務(wù) */
    
61.                 CPollerUnit::Ins()->ProcessAllEvents();
    
62.                        
    
63.                 /* 超時(shí)檢查*/
    
64.                 CSession::Ins()->CheckTimeOut();
    
65. 
    
66.                 /* 清理空閑的tcp連接。 */
    
67.                 CTcpPoll::Ins()->CheckIdle(g_conf.m_session._idle, g_conf.m_session._timeInterval, g_conf.m_session._emptyRatio);
    
68. 
    
69.                 /* 檢查是否到了統(tǒng)計(jì)間隔 */
    
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. };
    
77. 
    
78. 
    
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.         /* 不指定進(jìn)程個(gè)數(shù)，或者指定進(jìn)程數(shù)比cpu個(gè)數(shù)還多，修改為cpu的個(gè)數(shù)。 */
    
86.         if( 0 == process_group_num || process_group_num > cpu_process_num )
    
87.                 process_group_num = cpu_process_num;
    
88. 
    
89.         ERROR_LOG( "cpu num: %d, config process num: %d, real process num: %d", cpu_process_num, g_conf.m_iProcessNum, process_group_num);
    
90. 
    
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;
    
100. 
     
101.                         case 0:
     
102.                                 SetRunCpu( i );
     
103.                                 MainThread();
     
104.                                 return 0;
     
105. 
     
106.                         default:
     
107.                                 g_processcpu.insert(make_pair(pid, i));
     
108.                 };
     
109.         }
     
110. 
     
111.         return 0;
     
112. }
     
113. 
     
114. 
     
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.         }
     
121. 
     
122.         /* 顯示系統(tǒng)說(shuō)明。 */
     
123.         if( 0==strcasecmp(argv[1], "-v") ){
     
124.                 PrintVersion();
     
125.                 return 0;
     
126.         }
     
127.         PrintVersion();
     
128. 
     
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.         }
     
135. 
     
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);
     
138. 
     
139.         /* 設(shè)置信號(hào) */
     
140.         SetUser1Handle();
     
141.         /* 綁定cpu */
     
142.         //if ( g_conf.m_iCpuBind>= 0 )        SetRunCpu(g_conf.m_iCpuBind);       
     
143. 
     
144. 
     
145.         /* 監(jiān)聽(tīng)端口 */
     
146.         if ( CreateClientSocket() < 0 )
     
147.         {
     
148.                 fprintf(stderr, "CreateClientSocket\n");
     
149.                 return E_FAIL;
     
150.         }
     
151.         DEBUG_LOG("CreateClientSocket Succ!");
     
152. 
     
153.         /* fork進(jìn)程，設(shè)置為守護(hù)進(jìn)程。 */
     
154.         daemon_init();
     
155. 
     
156.         fork_process_group();
     
157. 
     
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. }
     
193. 
     

復(fù)制代碼

qianguozheng

CPollerUnit這個(gè)東西是C里面的不？我咋沒(méi)見(jiàn)過(guò)啊，還是你自定義的，請(qǐng)指教

oscarzhou

自定義的類，封裝epoll操作。

zhuxuqing666

cisco,linux,oracle授權(quán)培訓(xùn)中心，高端的IT培訓(xùn)，中心03年成立，本著以客戶為中心的宗旨，有需要的朋友請(qǐng)聯(lián)系:朱老師：13777819351  QQ：365850824   博學(xué)教育