ACE中使用完成端口－Proactor框架

peakzhang

作者：阿彪
地址：阿彪的Blog

使用ACE中的Proactor的话，会要比我们使用我们直接写的要来得简单。
在说Proactor之前我们需要了解Windows里的完成端口的工作原理。
      完成端口是WinNT内核里的一个框架。我们可以为我们的一些异步的操作
新建一个完成端口，然后这个完成端口会有几个工作线程来处理。我们
可以将socket，或是一个文件读写，或是一个串口的收发数据的句柄，
梆定到这个完成端口之上，当一个读或是写的事件完成之后，完成端口
机制将会自动将一个完成消息放到完成队列中，完成端口的工作线程池
将会被触发调用。回调的时候我们梆定时将一些基础的信息也梆在其中，
当工作线程也会通过一种叫做完成项的指针返回给你。就是说，你可能
梆定了多个socket或是文件都是没有问题的。按微软人写的文档里说的
可以面对百万个这样的异步对象。
      这里我就不再使用WinAPI写完成端口了。
      现在是使用ACE框架来写一个。
      使用他来做一个完成端口步骤也是一样的。
开始的时候需要一个完成端口，还有完成端口的工作线程池。在ACE框架
里提供了一种叫ACE_Task的线程池模块类
和一般的线程类一样，它的工作时调用的函数是
virtual int svc (void);
只是如何使用呢。无非是开启线程与关闭线程两个操作。
在此类中定义一个ACE_Thread_Semaphore sem_;变量
然后开户n个线程的过程就是这样的:
int Proactor_Task::star(int nMax)
{
...
this->activate (THR_NEW_LWP, nMax);
for (;nMax>0;nMax--)
{
  sem_.acquire();
}
return 0;
}
一个是创建，二个是一个一个的触发。让这一些线程都工作.
当然工作线程都要释放自己:
int Proactor_Task::svc()
{
ACE_DEBUG((LM_INFO,ACE_TEXT("svc函数调用!\n")));
sem_.release(1);
...
return 0;
}
好了。这个线程池开始工作了。接下来，我们要做将完成端口对象给创建出来：
在这个线程池里定义一个完成端口对象指针:
ACE_Proactor * proactor_;
创建的过程是这样的。
//是在Win32下，就使用这个Proactor的实现
ACE_WIN32_Proactor *proactor_impl = new ACE_WIN32_Proactor();   //新建proactor的实现
proactor_=new ACE_Proactor(proactor_impl,1);     //与proactor关联
ACE_Proactor::instance (this->proactor_, 1);     //将新建出来的proactor保存在静态框架里
如何删除呢。
ACE_Proactor::end_event_loop();
this->wait();
之后来写线程池里的函数
ACE_Proactor::run_event_loop();
只要写一句就OK了。
这就完成了一个完成端口对象的创建过程。我们只要做一下封装就OK了。
给它一个工作线程的大小。之后它就会自动的新建一个完成端口在ACE_Proactor::instance里。
接下来我们要做Acceptor与recv。
实计上ACE里已经为我们写好了。它们就是：
ACE_Asynch_Acceptor类ACE_Service_Handler类
class Accepte : public ACE_Asynch_Acceptor<Receive>
class Receive : public ACE_Service_Handler
这样一继承，工作就已经完成了。
如果我们想得到这一些网络事件的话，可以做一些继承就OK了。
他们内部调用的过程是这样的：
当有一个新的用户连接上来之后。
Accepte会有一个函数回调。
virtual HANDLER *make_handler (void);
这个函数里，我们必需写一个new Receive对象。
new完成之后Receive的open函数将会回调
open函数调用的时候，此接收对象的socket句柄就得到了。我们就在这个时候需要将一个
读、写的流梆定在其中。还有就是做一步异步的接收数据。这个如果你写过重叠方式的话就会
比较的了解，这也叫做异步的I/O的投递。等这个recv完成之后就会回调。
好了。这就算完成了。现在把代码贴出来。
我不知道如何做一个下载点。不好意思只有大家自己复制下来
我使用的VC6.0,ACE5.4.1的编程环境

2. // Accepte.h: interface for the Accepte class.
3. //
4. //////////////////////////////////////////////////////////////////////
5. #if !defined(AFX_ACCEPTE_H__DCEC809D_E5D2_48D1_A8A7_C9FD3C4D7C15__INCLUDED_)
6. #define AFX_ACCEPTE_H__DCEC809D_E5D2_48D1_A8A7_C9FD3C4D7C15__INCLUDED_
7. #if _MSC_VER > 1000
8. #pragma once
9. #endif // _MSC_VER > 1000
10. #include <ace/Asynch_Acceptor.h>
11. #include "Receive.h"
12. class Accepte : public ACE_Asynch_Acceptor<Receive>
13. {
14. public:
15. Receive* make_handler (void);
16. Accepte();
17. virtual ~Accepte();
18. };
19. #endif // !defined(AFX_ACCEPTE_H__DCEC809D_E5D2_48D1_A8A7_C9FD3C4D7C15__INCLUDED_)
21. // Accepte.cpp: implementation of the Accepte class.
22. //
23. //////////////////////////////////////////////////////////////////////
24. #include "Accepte.h"
25. //////////////////////////////////////////////////////////////////////
26. // Construction/Destruction
27. //////////////////////////////////////////////////////////////////////
28. Accepte::Accepte()
29. {
30. }
31. Accepte::~Accepte()
32. {
33. }
34. Receive* Accepte::make_handler(void)
35. {
36. return new Receive();
37. }
38. // Proactor_Task.h: interface for the Proactor_Task class.
39. //
40. //////////////////////////////////////////////////////////////////////
41. #if !defined(AFX_PROACTOR_TASK_H__12F37C95_9872_4923_89A2_5A59AE7AC1FD__INCLUDED_)
42. #define AFX_PROACTOR_TASK_H__12F37C95_9872_4923_89A2_5A59AE7AC1FD__INCLUDED_
43. #if _MSC_VER > 1000
44. #pragma once
45. #endif // _MSC_VER > 1000
46. #include "ace\Task_T.h"
47. #include "ace\Thread_Semaphore.h"
48. #include "ace\Proactor.h"
49. #include "ace\WIN32_Proactor.h"
50. class Proactor_Task : public ACE_Task<ACE_MT_SYNCH>
51. {
52. public:
53. Proactor_Task();
54. virtual ~Proactor_Task();
55. int star(int nMax);
56. int stop();
57. virtual int svc (void);
58. int create_proactor();
59. int release_proactor();
60. ACE_Thread_Semaphore sem_;
61. ACE_Proactor * proactor_;
62. };
63. #endif // !defined(AFX_PROACTOR_TASK_H__12F37C95_9872_4923_89A2_5A59AE7AC1FD__INCLUDED_)
65. // Proactor_Task.cpp: implementation of the Proactor_Task class.
66. //
67. //////////////////////////////////////////////////////////////////////
68. #include "Proactor_Task.h"
69. //////////////////////////////////////////////////////////////////////
70. // Construction/Destruction
71. //////////////////////////////////////////////////////////////////////
72. Proactor_Task::Proactor_Task()
73. {
74. }
75. Proactor_Task::~Proactor_Task()
76. {
77. }
78. int Proactor_Task::star(int nMax)
79. {
80. create_proactor();
81. this->activate (THR_NEW_LWP, nMax);
82. for (;nMax>0;nMax--)
83. {
84.   sem_.acquire();
85. }
86. return 0;
87. }
88. int Proactor_Task::stop()
89. {
90. ACE_Proactor::end_event_loop();
91. this->wait();
92. return 0;
93. }
94. int Proactor_Task::release_proactor()
95. {
96. ACE_Proactor::close_singleton ();
97. proactor_ = 0;
98. return 0;
99. }
100. int Proactor_Task::create_proactor()
101. {
102. ACE_WIN32_Proactor *proactor_impl = 0;
104. ACE_NEW_RETURN (proactor_impl,
105.   ACE_WIN32_Proactor,
106.   -1);
107.   // always delete implementation  1 , not  !(proactor_impl == 0)
108. ACE_NEW_RETURN (this->proactor_,
109.   ACE_Proactor (proactor_impl, 1 ),
110.   -1);
111. // Set new singleton and delete it in close_singleton()
112. ACE_Proactor::instance (this->proactor_, 1);
113. return 0;
114. }
115. int Proactor_Task::svc()
116. {
117. ACE_DEBUG((LM_INFO,ACE_TEXT("svc函数调用!\n")));
118. sem_.release(1);
119. ACE_Proactor::run_event_loop();
120. return 0;
121. }
122. // Receive.h: interface for the Receive class.
123. //
124. //////////////////////////////////////////////////////////////////////
125. #if !defined(AFX_RECEIVE_H__0E7EF8C0_465F_4D9C_8A29_0C2A0F1EAFFE__INCLUDED_)
126. #define AFX_RECEIVE_H__0E7EF8C0_465F_4D9C_8A29_0C2A0F1EAFFE__INCLUDED_
127. #if _MSC_VER > 1000
128. #pragma once
129. #endif // _MSC_VER > 1000
130. #include <ace/Asynch_io.h>
131. #include <ace/Message_Block.h>
132. #include <ace/Log_Msg.h>
133. #include <ace/OS_Memory.h>
134. class Receive : public ACE_Service_Handler
135. {
136. public:
137. Receive();
138. virtual ~Receive()
139. {
140.   if (this->handle() != ACE_INVALID_HANDLE )
141.   {
142.    closesocket(SOCKET(this->handle()));
143.   }
144. }
145. virtual void open(ACE_HANDLE h,ACE_Message_Block& );
146. virtual void handle_read_stream(const ACE_Asynch_Read_Stream::Result &result);
147. virtual void handle_write_stream(const ACE_Asynch_Write_Stream::Result &result);
148. private:
149. ACE_Asynch_Write_Stream write_;
150. ACE_Asynch_Read_Stream reader_;
151. };
152. #endif // !defined(AFX_RECEIVE_H__0E7EF8C0_465F_4D9C_8A29_0C2A0F1EAFFE__INCLUDED_)
153. // Receive.cpp: implementation of the Receive class.
154. //
155. //////////////////////////////////////////////////////////////////////
156. #include "Receive.h"
157. //////////////////////////////////////////////////////////////////////
158. // Construction/Destruction
159. //////////////////////////////////////////////////////////////////////
160. Receive::Receive()
161. {
162. }
163. void Receive::open(ACE_HANDLE h,ACE_Message_Block& )
164. {
165. this->handle(h);
166. if (this->write_.open(*this)!=0 ||
167.   this->reader_.open(*this) != 0 )
168. {
169.   delete this;
170.   return ;
171. }
172. ACE_Message_Block *mb;
173. ACE_NEW_NORETURN(mb,ACE_Message_Block(1024));
174. if ( this->reader_.read(*mb,mb->space()) != 0)
175. {
176.   ACE_ERROR((LM_ERROR,ACE_TEXT(" (%t) error information %p.")));
177.   mb->release();
178.   delete this;
179.   return;
180. }
181. }
182. void Receive::handle_read_stream (const ACE_Asynch_Read_Stream::Result &result)
183. {
184. ACE_Message_Block &mb = result.message_block();
185. if ( !result.success() || result.bytes_transferred() == 0)
186. {
187.   mb.release();
188.   delete this;
189. }
190. else
191. {
192.   ACE_Message_Block* new_mb;
193.   ACE_NEW_NORETURN(new_mb,ACE_Message_Block(1024));
194.   this->reader_.read(*new_mb,new_mb->space());
195. }
196. return ;
197. }
198. void Receive::handle_write_stream (const ACE_Asynch_Write_Stream::Result &result)
199. {
200. result.message_block().release();
201. return ;
202. }
203. ////////////////////////////////////////////////////////////////////////////////////////////////////
204. //main.cpp
205. #ifdef _DEBUG
206. #pragma comment(lib,"aced")
207. #else
208. #pragma comment(lib,"ace")
209. #endif
210. #include <ace\ace.h>
211. #include "Accepte.h"
212. #include "Proactor_Task.h"
213. int ACE_TMAIN(int ,char*[])
214. {
215. Proactor_Task task;
216. task.star(3);
218. Accepte accepte;
219. accepte.open(ACE_INET_Addr (2222), 0, 1,ACE_DEFAULT_BACKLOG,1,ACE_Proactor::instance());
220. int nExit=0;
221. while (nExit==0)
222.   scanf("%d",&nExit);
223. return 0;
224. }

复制代码

peakzhang

这个用在MFC的工程中会有问题，开始一下，CPU占用很高，开了一个Listen的端口。过几秒就关闭那个Socket了。
把Main函数放在一个类中了

2. BOOL CServerMsg::StartNetWork(WORD wPort)
3. {
4. //m_network.Stop();
5. ////这里应该修正异步事件Socket的启动函数,当启动线程失败应该返回-1
6. //m_wListenPort = m_network.Start(m_wListenPort,m_wListenPort?FALSE:TRUE);
7. //if(m_wListenPort!=0xFFFF)
8. // return TRUE;
9. if (wPort > 0)
10. {
11.   m_wListenPort = wPort;
12. }
13. //ACE_Proactor * pAP = ACE_Proactor::instance();
14. m_task.star(3);
15. m_accepte.open(ACE_INET_Addr (m_wListenPort), 0, 1,
16.   ACE_DEFAULT_BACKLOG,1, ACE_Proactor::instance());
17. int nExit=0;
18. while (nExit==0)
19. {
20.   Sleep(100);
21.   //scanf("%d",&nExit);
22. }
23. return TRUE;
24. }

复制代码

然后在对话框的初始化中调用这个函数。
其它的类一样的处理

peakzhang

问题解决了，这个例子是正确可以用了。
我查宏发现 ACE_DEFAULT_BACKLOG为5，可是调试时不知道为什么这个变成了一个很大的数字几百万的随机数.
use 5 directly,
m_accepte.open(ACE_INET_Addr (m_wListenPort), 0, 1,
  5 ,1, ACE_Proactor::instance()); //ACE_DEFAULT_BACKLOG

peakzhang

这个函数时有问题的
void Receive::handle_read_stream (const ACE_Asynch_Read_Stream::Result &result)
{
ACE_Message_Block &mb = result.message_block();
if ( !result.success() || result.bytes_transferred() == 0)
{
  mb.release();
  delete this;
}
else
{
  ACE_Message_Block* new_mb;
  ACE_NEW_NORETURN(new_mb,ACE_Message_Block(1024));
  this->reader_.read(*new_mb,new_mb->space());
}
这个已经是接受完成后 的消息处理，为什么还要从Stream中去Read呢，直接处理 Result的mb 就行啊？
void Receive::handle_read_stream (const ACE_Asynch_Read_Stream::Result &result)
{
size_t nMsgLen = result.bytes_transferred() ;
ACE_Message_Block &mb = result.message_block();
if ( !result.success() || nMsgLen == 0)
{
  mb.release();
  delete this;
}
else
{
//ACE_Message_Block* new_mb;
  //ACE_NEW_NORETURN(new_mb,ACE_Message_Block(nMsgLen + 1));
  //this->reader_.read(*new_mb, nMsgLen);
  //消息头分析处理发到给处理中处理
  RecvMsgHander(&mb);//自定义消息处理函数，直接处理接收完成的数据
}
mb.release();
return ;
}