创建一个可正常结束的Proactor服务器

peakzhang

作者：Qinglan
这是APG上关于Proactor使用的例子，去掉了trace信息，做了一点小修改：

2. #include <ace/Os_main.h>
3. #include <ace/Asynch_Acceptor.h>
4. #include <ace/Proactor.h>
5. #define LISTEN_PORT 5222
7. class HA_Proactive_Service : public ACE_Service_Handler
8. {
9. public:
10.        ~HA_Proactive_Service ()
11.        {
12.               if (this->handle () != ACE_INVALID_HANDLE)
13.                      ACE_OS::closesocket (this->handle ());
14.        }
16.        virtual void open (ACE_HANDLE h, ACE_Message_Block&)
17.        {
18.               this->handle (h);
19.               if (this->reader_.open (*this) != 0 || this->writer_.open (*this) != 0   )
20.               {
21.                      delete this;
22.                      return;
23.               }
25.               ACE_Message_Block *mb;
26.               ACE_NEW_NORETURN (mb, ACE_Message_Block (1024));
27.               if (this->reader_.read (*mb, mb->space ()) != 0)
28.               {
29.                      mb->release ();
30.                      delete this;
31.                      return;
32.               }
33.        }
35.        virtual void handle_read_stream (const ACE_Asynch_Read_Stream::Result &result)
36.        {
37.               ACE_Message_Block &mb = result.message_block ();
38.               if (!result.success () || result.bytes_transferred () == 0)
39.               {
40.                      mb.release ();
41.                      delete this;
42.               }
43.               else
44.               {
45.                      ACE_DEBUG ((LM_DEBUG, ACE_TEXT("Received Data : %c\n"), *mb.rd_ptr()));
46.                      mb.release();
48.                      ACE_Message_Block *new_mb;
49.                      ACE_NEW_NORETURN (new_mb, ACE_Message_Block (1024));
50.                      this->reader_.read (*new_mb, new_mb->space ());
51.               }
52.        }
54.        virtual void handle_write_stream (const ACE_Asynch_Write_Stream::Result &result)
55.        { result.message_block ().release (); }
57. private:
58.        ACE_Asynch_Read_Stream reader_;
59.        ACE_Asynch_Write_Stream writer_;
60. };
62. class HA_Proactive_Acceptor : public ACE_Asynch_Acceptor<HA_Proactive_Service>
63. {
64. };
66. int ACE_TMAIN (int, ACE_TCHAR *[])
67. {
68.        ACE_INET_Addr listen_addr( LISTEN_PORT );
69.        HA_Proactive_Acceptor aio_acceptor;
70.        if (0 != aio_acceptor.open (listen_addr, 0, 0, 5, 1, 0, 0))
71.               ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("acceptor open")), 1);
73.        ACE_Proactor::instance ()->proactor_run_event_loop ();
75.        return 0;
76. }

复制代码

程序在5222端口上监听，对于每个已建立的连接，服务器将所有收到的数据打印到控制台上。这就是整个程序所实现的功能。
但是这个程序存在一个很大的问题：没有办法正常结束。当运行这个程序后，只能强制结束这个进程，因为程序没有提供从proactor事件循环中退出来的方法。
让程序能够正常退出是很有必要的，比如在进行内存泄漏的检测时，如果程序没有办法正常退出，那么内存泄漏的检测工作将很难进行。

为了实现能够正常关闭程序的目的，需要稍微修改一下上面的程序实现。
这次程序将会有两个线程在运行，一个是Proactor事件循环线程，另一个则为控制线程。在控制线程中，要能够关闭Proactor线程，然后将程序干净地退出。
得益于ACE强大的封装，所做的改动并不大：

2. class ProactorThread : public ACE_Task_Base
3. {
4. public:
5.        int open()
6.        {
7.               return this->activate();
8.        }
10.        int close()
11.        {
12.               ACE_Proactor::instance()->proactor_end_event_loop();
13.               this->wait();
14.               return 0;
15.        }
17.        virtual int svc()
18.        {
19.               ACE_INET_Addr listen_addr( LISTEN_PORT );
20.               HA_Proactive_Acceptor aio_acceptor;
21.               if (0 != aio_acceptor.open (listen_addr, 0, 0, 5, 1, 0, 0))
22.                      ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("acceptor open")), 1);
24.               ACE_Proactor::instance ()->proactor_run_event_loop ();
25.               ACE_DEBUG ((LM_DEBUG, ACE_TEXT("Thread ended\n")));
27.               return 0;
28.        }
29. };
31. int ACE_TMAIN (int, ACE_TCHAR *[])
32. {
33.        ProactorThread thread;
34.        thread.open();
36.        int x;
37.        std::cin >> x;
39.        thread.close();
41.        return 0;
42. }

复制代码

仅仅只是添加了一个新的类，用于封装Proactor事件循环线程，另外修改了程序的主函数。这次程序将首先创建一个Proactor事件循环线程，然后主线程等待用户输入任意数后结束Proactor线程，在Proactor线程正常退出后主线程退出。

注意这里有两个要点：
1．当执行proactor_end_event_loop()之后，Proactor线程是从proactor_run_event_loop()处退出，所以会接着执行下面的代码，打印出” Thread ended”信息。
2．主线程必须使用wait()等待Proactor线程正常结束后再退出，否则如果主线程退出后，Proactor线程仍有一些资源释放工作未完成，那么就会存在资源未正确释放的问题。

peakzhang

更深入的处理可以是这样的:
把
       int x;
       std::cin >> x;
改为注册成信号,用户按ctrl+c,向系统插入一个特殊的ACE_Message_Block的实例,各线程收到后,均正常结束,在ACE_Stream框架中,常用这种办法.
这处结束过程还可以把一些未处理的消息写入磁盘,下次启动后载入后接着处理.

psycheqiqi

我把这个程序移植到MFC里面的时候，在ProactorThread中aio_acceptor.open的时候跳转不到HA_Proactive_Service中，为什么？

winston

错误是什么？