ACE Proactor模式的一个问题(使用ACE Proactor开发的朋友来看看吧.)

我现在想使用ACE里的Proactor来做一个通信的程序框架.看书上说就是使用的完成端口的技术.可以提供最为优化的通信.
但是在使用的时候会有这样的问题.我使用了VC写了一个程序.连接一个socket.然后开启一个线程发送数据包.每sleep 10 毫秒发一个大小为50byte的数据包.这样我开了40个之后.就会出显问题了.服务器就会出错误了.我的源代码是这样写的:///接收器类// Accepte.h: interface for the Accepte class.
//
//////////////////////////////////////////////////////////////////////#if !defined(AFX_ACCEPTE_H__DCEC809D_E5D2_48D1_A8A7_C9FD3C4D7C15__INCLUDED_)
#define AFX_ACCEPTE_H__DCEC809D_E5D2_48D1_A8A7_C9FD3C4D7C15__INCLUDED_#if _MSC_VER > 1000
#pragma once
#endif // _MSC_VER > 1000
#include <ace/Asynch_Acceptor.h>
#include "Receive.h"
class Accepte : public ACE_Asynch_Acceptor<Receive>
{
public:
Receive* make_handler (void);
Accepte();
virtual ~Accepte();
virtual int validate_connection (const ACE_Asynch_Accept::Result& result,
                                   const ACE_INET_Addr &remote,
                                   const ACE_INET_Addr& local);
};#endif // !defined(AFX_ACCEPTE_H__DCEC809D_E5D2_48D1_A8A7_C9FD3C4D7C15__INCLUDED_)#include "Accepte.h"//////////////////////////////////////////////////////////////////////
// Construction/Destruction
//////////////////////////////////////////////////////////////////////Accepte::Accepte()
{}Accepte::~Accepte()
{}Receive* Accepte::make_handler(void)
{
 return new Receive();
}int Accepte::validate_connection (const ACE_Asynch_Accept::Result& result,
                                   const ACE_INET_Addr &remote,
                                   const ACE_INET_Addr& local)
{
return 0;
}///客户端类
// Receive.h: interface for the Receive class.
//
//////////////////////////////////////////////////////////////////////#if !defined(AFX_RECEIVE_H__0E7EF8C0_465F_4D9C_8A29_0C2A0F1EAFFE__INCLUDED_)
#define AFX_RECEIVE_H__0E7EF8C0_465F_4D9C_8A29_0C2A0F1EAFFE__INCLUDED_#if _MSC_VER > 1000
#pragma once
#endif // _MSC_VER > 1000
#include <ace/Asynch_io.h>
#include <ace/Message_Block.h>
#include <ace/Log_Msg.h>
#include <ace/OS_Memory.h>
class Receive : public ACE_Service_Handler
{
public:
Receive();
virtual ~Receive();
virtual void open(ACE_HANDLE h,ACE_Message_Block& );
virtual void handle_read_stream(const ACE_Asynch_Read_Stream::Result &result);
virtual void handle_write_stream(const ACE_Asynch_Write_Stream::Result &result);
private:
void pro_read();
ACE_Asynch_Write_Stream write_;
ACE_Asynch_Read_Stream reader_;
ACE_Message_Block* mb_;

};#endif // !defined(AFX_RECEIVE_H__0E7EF8C0_465F_4D9C_8A29_0C2A0F1EAFFE__INCLUDED_)///
// Receive.cpp: implementation of the Receive class.
//
//////////////////////////////////////////////////////////////////////#include "Receive.h"//////////////////////////////////////////////////////////////////////
// Construction/Destruction
//////////////////////////////////////////////////////////////////////Receive::Receive()
{
mb_=0;
ACE_DEBUG((LM_INFO,ACE_TEXT("有一个连接.\n")));
}
Receive::~Receive()
{
{
ACE_DEBUG((LM_INFO,ACE_TEXT("连接退出.\n")));
if (this->handle() != ACE_INVALID_HANDLE )
{
closesocket(SOCKET(this->handle()));
}
}
}
void Receive::pro_read()
{
if (mb_==0)
{
mb_ = new ACE_Message_Block(1024);
}
else
{
mb_->reset();
}
if ( this->reader_.read(*mb_,mb_->space()) != 0)
{
ACE_ERROR((LM_ERROR,ACE_TEXT(" (%t) error information %p.")));
mb_->release();
delete this;
return;
}
}
void Receive::open(ACE_HANDLE h,ACE_Message_Block& )
{
this->handle(h);
if (this->write_.open(*this)!=0 || 
this->reader_.open(*this) != 0 )
{
delete this;
return ;
}
pro_read();
}
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
{
printf("接收到数据包\n");
//printf("内容:%s\n",(char*)mb_.base());
printf("数据包的大小为:%d\n",result.bytes_transferred());
pro_read();
}
return ;
}void Receive::handle_write_stream (const ACE_Asynch_Write_Stream::Result &result)
{
//result.message_block().release();
return ;
}////////////////////////////////////////////////////////////////////////////////////////////////////
//main.cpp#ifdef _DEBUG 
#pragma comment(lib,"aced")
#else
#pragma comment(lib,"ace")
#endif
#include <ace\ace.h>
#include "Accepte.h"
#include "Proactor_Task.h"int ACE_TMAIN(int ,char*[])
{
Accepte accepte;
accepte.open(ACE_INET_Addr (2222), 0, 1,ACE_DEFAULT_BACKLOG,1,0,1);
int nExit=0;
while (nExit==0)
{
ACE_Proactor::instance()->run_event_loop();
}
return 0;
}请教做过ACE前摄式的朋友来看一下这个代码吧.谢谢啦.