连接-接入模式中每个对应client能不能新申请一个反应器?

peakzhang

在server端派生自ACE_Svc_Handler写对应Client的处理类ServerClient
有人说如果ServerClient开线程,因为默认ServerClient与ACE_Acceptor是同一个单体的反应器,所以在ServerClient线程中写的话就会出现分别在2个线程里读 写,于是需要同步访问ServerClient的sock_stream,
我的问题就是
确实需要同步吗?
假设需要同步,于是打算把ServerClient的读与写都放在自己的线程里:重载ServerClient的open函数重新申请一个反应器.
该怎么申请呢?
我简单尝试了一下在ServerClient中组合了一个反应器,但是在ServerClient的handler_close中delete this总是报错,似乎是新定义的反映器出错(我保证handler_close只被调用了一次)
完整的代码贴上,大家帮调试一下:
Server端:

1. #include <ace/Reactor.h>
2. #include <ace/Svc_Handler.h>
3. #include <ace/Acceptor.h>
4. #include <ace/sock_stream.h>
5. #include <ace/Acceptor.h>
6. #include <ace/SOCK_Acceptor.h>
7. #include <iostream>
8. using std::cout;
9. using std::endl;
10. class CSvcHandler :public ACE_Svc_Handler <ACE_SOCK_STREAM,ACE_NULL_SYNCH>
11. {
12. public:
13. CSvcHandler(void);
14. public:
15. ~CSvcHandler(void);
16. public:
17.   int handle_input(ACE_HANDLE);
18.   int handle_close (ACE_HANDLE handle ,ACE_Reactor_Mask colse_mask);
19.   int open(void*);
20.   int svc();
21. private:
22. ACE_Reactor* m_pobjReactor;
23.     char data[1024];
24. };
25. CSvcHandler::CSvcHandler(void)
26. {
27. cout<<"Acceptor ClientPoint:I'm coming.\n"<<endl;
28. m_pobjReactor = new ACE_Reactor();
29. }
30. CSvcHandler::~CSvcHandler(void)
31. {
32. cout<<"Acceptor ClientPoint:I'm out.\n";
33. delete m_pobjReactor;
34. }
35. int CSvcHandler::handle_input(ACE_HANDLE)
36. {
37. memset(data,0,sizeof(data));
38. int rev = peer().recv(data,sizeof(data) - 1);
39. if(rev < 1)
40. {
41.   return -1;
42. }
43. else
44. {
45.   cout<<"Acceptor ClientPoint recv:"<<data<<endl<<"and return:"<<data<<endl<<endl;
46.   peer().send(data,sizeof(data) - 1);
47.   return 0;
48. }
49. }
50. int CSvcHandler::handle_close (ACE_HANDLE handle ,ACE_Reactor_Mask colse_mask)
51. {
52. cout<<"Acceptor ClientPoint:handle_close\n";
53. delete this;
54. return 0;
55. }
57. int CSvcHandler:pen(void*)
58. {
59. reactor(m_pobjReactor);
60. m_pobjReactor->register_handler(this,  ACE_Event_Handler::READ_MASK);
61. activate();
62. return 0;
63. }
64. int CSvcHandler::svc()
65. {
66. while(true)
67. {
68.   m_pobjReactor->handle_events();
69. }
70. return 0;
72. }
73. int _tmain(int argc, _TCHAR* argv[])
74. {
75. ACE_INET_Addr l_objLocalAddr(3000,"127.0.0.1");
76. ACE_Acceptor<CSvcHandler,ACE_SOCK_ACCEPTOR> l_objAcceptor;
77. l_objAcceptor.open(l_objLocalAddr);
78. while(1)
79. {
80.   ACE_Reactor::instance()->handle_events();
81. }
82. return 0;
83. }
84. _________________________________
85. Client端:
86. #include <ace/SOCK_Connector.h>
87. #include <ace/Connector.h>
88. #include <ace/Reactor.h>
89. #include <ace/Svc_Handler.h>
90. #include <ace/Acceptor.h"
91. #include <ace/sock_stream.h>
92. #include <iostream>
93. using std::cout;
94. using std::endl;
95. class CSvcHandler :public ACE_Svc_Handler <ACE_SOCK_STREAM,ACE_NULL_SYNCH>
96. {
97. public:
98. CSvcHandler(void);
99. public:
100. ~CSvcHandler(void);
101. public:
102.   int handle_input(ACE_HANDLE);
103.   int handle_close (ACE_HANDLE handle ,ACE_Reactor_Mask colse_mask);
104. private:
105.     char data[1024];
106. };
108. CSvcHandler::CSvcHandler(void)
109. {
110. cout<<"Connector ClientPoint:I'm coming.\n"<<endl;
111. }
112. CSvcHandler::~CSvcHandler(void)
113. {
114. cout<<"Connector ClientPoint: I'm out.\n";
115. }
116. int CSvcHandler::handle_input(ACE_HANDLE)
117. {
118. memset(data,0,sizeof(data));
119. int rev = peer().recv(data,sizeof(data) - 1);
120. if(rev < 1)
121. {
122.   return -1;
123. }
124. else
125. {
126.   cout<<"Connector ClientPoint recv:"<<data<<endl;
127.   return 0;
128. }
129. }
131. int CSvcHandler::handle_close (ACE_HANDLE handle ,ACE_Reactor_Mask colse_mask)
132. {
133. cout<<"Connector ClientPoint:handle_close\n";
134. delete this;
135. return 0;
136. }
138. int _tmain(int argc, _TCHAR* argv[])
139. {
140. ACE_INET_Addr l_objRemoteAddr(3000,"127.0.0.1");
141. CSvcHandler* l_pobjClientPoint = new CSvcHandler();
142. ACE_Connector<CSvcHandler,ACE_SOCK_CONNECTOR> l_objConnector;
143. l_objConnector.connect(l_pobjClientPoint,l_objRemoteAddr);
144. char l_szBuffer[64] = "C-A模式";
145. int l_nSize = l_pobjClientPoint->peer().send(l_szBuffer,sizeof(l_szBuffer));
146. cout<<"Connector ClientPoint send:"<<l_szBuffer<<endl;
147. while(true)               
148. {
149.   ACE_Reactor::instance()->handle_events();
150. }
151. delete l_pobjClientPoint;
152. return 0;
153. }

复制代码

[ 本帖最后由 peakzhang 于 2007-12-11 22:17 编辑 ]

peakzhang

一堆问题：
你怎么确保在删除对象的时候，对象已经停止运行？

CSvcHandler::~CSvcHandler(void)
{
cout<<"Acceptor ClientPoint:I'm out.\n";
delete m_pobjReactor;

}


int CSvcHandler::svc()
{
while(true)
{
  m_pobjReactor->handle_events(); //同步呢？
}

return 0;

}
你退出的时候，要删除m_pobjReactor，你怎么保证删除的时候，它已经停止运行了。

xjh_001

ServerClient开线程的方式为:
1在open方法中调用activate方法启动多个处理线程.
2在handle_input方法中,将从网络接收到的数据调用put方法,加入到消息队列中
3在svc中,不停地getq,以取得接收到的消息,消息内容可以包括具体的接收数据,接收socket的句柄(便于向客户端发送应答数据)
4svc中接收到退出消息时,退出线程.因此在handle_close方法中,要发送n(激活的工作线程的个数)个退出消息,并调用wait方法,等待所有工作线程退出.
在这里,只是消息的处理是多线程的,数据的接收和发送用的是主线程的Reactor.