Reactor框架写的一个简单udp服务器框架的问题

peakzhang

我用ACE的Reactor框架写的一个简单udp服务器框架如下：我对其进行压力测试，结果发现问题很大。
当每秒接受的数据包到11000个（每个包1400字节，总共42K字节）左右时，服务器耗尽了所有的cpu，但内存占用非常少不到5％。pc是赛阳2.6，512M内存，系统为redhead9.0，2.6内核。各位看看问题出在什么地方！
以下内容为程序代码:

2. #include "ace/OS_main.h"
3. #include "ace/OS_NS_string.h"
4. #include "ace/OS_NS_unistd.h"
5. #include "ace/Reactor.h"
6. #include "ace/Process.h"
7. #include "ace/SOCK_Dgram.h"
8. #include "ace/INET_Addr.h"
9. #include "ace/Log_Msg.h"
10. #include "ace/Thread_Manager.h"
11. #include "ace/Task_T.h"
12. #include "ace/Shared_Memory_Pool.h"
13. //#include "ace/Process_Mutex.h"
14. #include "ace/Malloc.h"
15. #include "ace/Malloc_T.h"
16. #include "ace/Shared_Memory_MM.h"
18. #include "client.h"
19. #define SERVER_PORT 10101
20. #define POOLNAME "mypoll"
21. typedef ACE_Malloc<ACE_SHARED_MEMORY_POOL, ACE_SYNCH_MUTEX> myalloc;
22. static const size_t TASK_THREAD_POOL_SIZE = 5;
23. myalloc shm_malloc(POOLNAME);
24. static int packages = 0;
25. static int total_bytes = 0;
26. class Task_Worker: public ACE_Task<ACE_MT_SYNCH>
27. {
28.     public:
29.         virtual int svc(void)
30.         {
31.             while(1)
32.             {
33.                 ACE_Message_Block *mb = NULL;
34.                 if(this->getq(mb) == -1)
35.                 {
36.                     continue;
37.                 }
38.                 process_task(mb);
39.             }
40.             return 0;
41.         }
42.     private:
43.         void process_task(ACE_Message_Block *mb)
44.         {
45.             PKG_HEADER header;
46.             memcpy(&header, mb->rd_ptr(), sizeof(PKG_HEADER));
47.             header.magic_num = ntohl(header.magic_num);
48.             header.index = ntohl(header.index);
49.             header.check_sum = ntohl(header.check_sum);
50.             //printf("magic[%d]index[%d]checksum[%d]\n", header.magic_num, header.index, header.check_sum);
51.             mb->release();
52.         }
53. };
54. class Task_Manager: public ACE_Task<ACE_MT_SYNCH>
55. {
56.     public:
57.         virtual int svc(void)
58.         {
59.             Task_Worker task_tp;
60.             task_tp.activate(THR_NEW_LWP | THR_JOINABLE, TASK_THREAD_POOL_SIZE);
61.             while(1)
62.             {
63.                 ACE_Message_Block *mb = NULL;
64.                 if(this->getq(mb) < 0)
65.                 {
66.                     task_tp.msg_queue()->deactivate();
67.                     task_tp.wait();
68.                 }
69.                 task_tp.putq(mb);
70.             }
71.             return 0;
72.         }
73. };
74. class Dgram_Endpoint : public ACE_Event_Handler
75. {
76. public:
77.   Task_Manager task_mgr;
78.   Dgram_Endpoint (const ACE_INET_Addr &local_addr);
79.   virtual ACE_HANDLE get_handle (void) const;
80.   virtual int handle_input (ACE_HANDLE handle);
81.   virtual int handle_timeout (const ACE_Time_Value & tv,const void *arg = 0);
82.   virtual int handle_close (ACE_HANDLE handle,ACE_Reactor_Mask close_mask);
83.   virtual int handle_signal (int signum, siginfo_t*, ucontext_t*);
84.   int send (const char *buf, size_t len, const ACE_INET_Addr &);
85. private:
86.   ACE_SOCK_Dgram endpoint_;
87. };
89. int Dgram_Endpoint::send (const char *buf,size_t len,const ACE_INET_Addr &addr)
90. {
91.   return this->endpoint_.send (buf, len, addr);
92. }
93. Dgram_Endpoint::Dgram_Endpoint (const ACE_INET_Addr &local_addr)
94.   : endpoint_(local_addr)
95. {
96.     task_mgr.activate();
97. }
99. ACE_HANDLE Dgram_Endpoint::get_handle (void) const
100. {
101.   return this->endpoint_.get_handle();
102. }
104. int Dgram_Endpoint::handle_close (ACE_HANDLE handle,ACE_Reactor_Mask)
105. {
106.     ACE_DEBUG((LM_DEBUG, "************handle_close***********\n"));
107.     ACE_UNUSED_ARG (handle);
108.     this->endpoint_.close();
109.     delete this;
110.     return 0;
111. }
112. static int is_heart_package(char *buf, size_t size)
113. {
114.     unsigned long data;
115.     if (size<4)
116.         return -1;
117.     memcpy(&data, buf, sizeof(data));
118.     data = ntohl(data);
119.     if (data == 0x12345678)
120.     {
121.         //printf("yes it's heart beat package\n");
122.         return 0;
123.     }
124.     else
125.         return -1;
126. }
127. int Dgram_Endpoint::handle_input (ACE_HANDLE)
128. {
129.   char buf[BUFSIZ];
130.   ACE_INET_Addr from_addr;
131.   char address[32];
132.   ssize_t nbytes = this->endpoint_.recv (buf, sizeof(buf), from_addr);
133.   packages++;
134.   total_bytes += nbytes;
135.   
136. #if 0
137.   if (nbytes == -1)
138.     ACE_ERROR ((LM_ERROR,"%p","handle_input error\n"));
139.   else
140.     ACE_DEBUG ((LM_DEBUG, "[%d]bytes from[%s] received:%s\n", nbytes, address, buf));
141. #endif
142.   if (0 == is_heart_package(buf, nbytes))
143.   {
144.       return 0;
145.   }
146.   ACE_Message_Block *mb = NULL;
147.   //mb = shm_malloc.malloc(sizeof(ACE_Message_Block));
148.   ACE_NEW_RETURN(mb, ACE_Message_Block(nbytes, ACE_Message_Block::MB_DATA, 0, buf), -1);
149.   mb->wr_ptr(nbytes);
150.   this->task_mgr.putq(mb);
151.   return 0;
152. }
153. int Dgram_Endpoint::handle_timeout (const ACE_Time_Value &,const void *)
154. {
155.   ACE_DEBUG ((LM_DEBUG,"[%d] packages [%d]K bytes are received per secondt\n", packages, total_bytes/1024 ));
156.   packages = 0;
157.   total_bytes = 0;
158.   return 0;
159. }
160. int Dgram_Endpoint::handle_signal (int signum, siginfo_t* siginfo, ucontext_t* context)
161. {
162.     return ACE_Event_Handler::handle_signal (signum, siginfo, context);
163. }
164. int ACE_TMAIN (int argc, ACE_TCHAR *argv[])
165. {
166.     ACE_INET_Addr local_addr(SERVER_PORT);
167.     Dgram_Endpoint *endpoint;
168.     ACE_NEW_RETURN (endpoint,Dgram_Endpoint (local_addr),-1);
169.     if (ACE_Reactor::instance ()->register_handler(endpoint,ACE_Event_Handler::READ_MASK) == -1)
170.     {
171.         ACE_ERROR_RETURN ((LM_ERROR,"ACE_Reactor::register_handler"),-1);
172.     }
173.     ACE_Time_Value time_out(1);
174.     ACE_Reactor::instance()->schedule_timer(endpoint, (void *)"time out",  ACE_Time_Value::zero, time_out);
175. #if 0
176.     if (-1 == ACE_Reactor::instance()->register_handler(SIGINT, endpoint))
177.     {
178.         ACE_ERROR_RETURN((LM_ERROR, "fail to register SIGINT handler"), -1);
179.     }
180. #endif
181.     ACE_Reactor::instance()->run_event_loop();
182.   return 0;
183. }

复制代码

peakzhang

除非用ACE_Dev_Poll_Reactor，各个平台的服务器端，都不推荐用Reactor框架。效率不高，不是ACE效率不高，而是各个操作系统实现中，这种模式下(select)效率很差。除非不用，否则没办法。

如果你要用Reactor，推荐你用ACE_Dev_Poll_Reactor，很简单，初始化的时候，替换默认的select_reactor即可。你可以试试EPoll的效率。或者如果能用AIO，试试Proactor(兼容性不够好)。

至于如何编译使用ACE_Dev_Poll_Reactor,本站有说明，照着做就可以了。