2008-04-02 23:24
1、WIN32下面用proactor可以達(dá)到幾乎RAW IOCP的效率，由于封裝關(guān)系，應(yīng)該是差那么一點(diǎn)。
客戶(hù)端處理類(lèi)的常規(guī)寫(xiě)法：
//處理客戶(hù)端連接消息
class ClientHandler : public ACE_Service_Handler
{
public:
/**構(gòu)造函數(shù)
*
*
*/
ClientHandler(unsigned int client_recv_buf_size=SERVER_CLIENT_RECEIVE_BUF_SIZE)
:_read_msg_block(client_recv_buf_size),_io_count(0)
{
}
~ClientHandler(){}
/**
*初始化，因?yàn)榭赡芤玫紺lientHandler內(nèi)存池，而這個(gè)池又不一定要用NEW
*/
void init();
/**清理函數(shù)，因?yàn)榭赡芤玫絻?nèi)存池
*
*/
void fini();
//檢查是否超時(shí)的函數(shù)
void check_time_out(time_t cur_time);
public:
/**客戶(hù)端連接服務(wù)器成功后調(diào)用
*
* \param handle 套接字句柄
* \param &message_block 第一次讀到的數(shù)據(jù)（未用）
*/
//由Acceptor來(lái)調(diào)用！！！
virtual void open (ACE_HANDLE handle,ACE_Message_Block &message_block);
/**處理網(wǎng)絡(luò)讀操作結(jié)束消息
*
* \param &result 讀操作結(jié)果
*/
virtual void handle_read_stream (const ACE_Asynch_Read_Stream::Result &result);
/**處理網(wǎng)絡(luò)寫(xiě)操作結(jié)束消息
*
* \param &result 寫(xiě)操作結(jié)果
*/
virtual void handle_write_stream (const ACE_Asynch_Write_Stream::Result &result);
private:
//**生成一個(gè)網(wǎng)絡(luò)讀請(qǐng)求
*
* \param void
* \return 0-成功，-1失敗
*/
int initiate_read_stream (void);
/**生成一個(gè)寫(xiě)請(qǐng)求
*
* \param mb 待發(fā)送的數(shù)據(jù)
* \param nBytes 待發(fā)送數(shù)據(jù)大小
* \return 0－成功，－1失敗
*/
int initiate_write_stream (ACE_Message_Block & mb, size_t nBytes );
/**
*
* \return 檢查是否可以刪除，用的是一個(gè)引用計(jì)數(shù)。每一個(gè)外出IO的時(shí)候＋1，每一個(gè)IO成功后－1
*/
int check_destroy();
//異步讀
ACE_Asynch_Read_Stream _rs;
//異步寫(xiě)
ACE_Asynch_Write_Stream _ws;
//接收緩沖區(qū)只要一個(gè)就夠了，因?yàn)閴焊蜎](méi)想過(guò)要多讀，我直到現(xiàn)在也不是很清楚為什么要多讀，多讀的話要考慮很多問(wèn)題
ACE_Message_Block _read_msg_block;
//套接字句柄,這個(gè)可以不要了，因?yàn)榛?lèi)就有個(gè)HANDLER在里面的。
//ACE_HANDLE _handle;
//一個(gè)鎖，客戶(hù)端反正有東東要鎖的，注意，要用ACE_Recursive_Thread_Mutex而不是用ACE_Thread_Mutex，這里面是可以重入的，而且在WIN32下是直接的EnterCriticalSection，可以達(dá)到很高的效率
ACE_Recursive_Thread_Mutex _lock;
//在外IO數(shù)量,其實(shí)就是引用計(jì)數(shù)啦，沒(méi)啥的。為0的時(shí)候就把這個(gè)東東關(guān)掉啦。
long _io_count;
//檢查超時(shí)用的，過(guò)段時(shí)間沒(méi)東東就CLOSE他了。
time_t _last_net_io;
private:
//本來(lái)想用另外一種模型的，只用1個(gè)或者2個(gè)外出讀，后來(lái)想想，反正一般內(nèi)存都是足夠的，就不管了。
//ACE_Message_Block _send_msg_blocks[2];
//ACE_Message_Block &_sending_msg_block;
//ACE_Message_Block &_idle_msg_block;
private:
public:
//TODO:move to prriva and use friend class!!!
//只是為了效率更高，不用STL的LIST是因?yàn)榈浆F(xiàn)在我沒(méi)有可用的Node_Allocator，所以效率上會(huì)有問(wèn)題。
ClientHandler *_next;
ClientHandler *next(){return _next;}
void next(ClientHandler *obj){_next=obj;}
};
//這是具體實(shí)現(xiàn)，有些地方比較亂，懶得管了，鎖的有些地方不對(duì)。懶得改了，反正在出錯(cuò)或者有瓶頸的時(shí)候再做也不遲。
void ClientHandler::handle_read_stream (const ACE_Asynch_Read_Stream::Result &result)
{
_last_net_io=ACE_OS::time(NULL);
int byterecved=result.bytes_transferred ();
if ( (result.success ()) && (byterecved != 0))
{
//ACE_DEBUG ((LM_DEBUG, "Receiver completed:%d\n",byterecved));
//處理完數(shù)據(jù)
if(handle_received_data()==true)
{
//ACE_DEBUG ((LM_DEBUG, "go on reading...\n"));
//把東東推到頭部，處理粘包
_read_msg_block.crunch();
initiate_read_stream();
}
}
//這個(gè)地方不想用ACE_Atom_op，因?yàn)榉凑幸粋€(gè)鎖，而且一般都會(huì)用鎖，不管了。假如不在意的話，應(yīng)該直接用ACE_Atom_Op以達(dá)到最好的效率
{
ACE_Guard<ACE_Recursive_Thread_Mutex> locker (_lock);
_io_count--;
}
check_destroy ();
}
void ClientHandler::init()
{
//初始化數(shù)據(jù)，并不在構(gòu)造函數(shù)里做。
_last_net_io=ACE_OS::time(NULL);
_read_msg_block.rd_ptr(_read_msg_block.base());
_read_msg_block.wr_ptr(_read_msg_block.base());
this->handle(ACE_INVALID_HANDLE);
}
bool ClientHandler::handle_received_data()
{
...........自己處理
return true;
}
//==================================================================
void ClientHandler::handle_write_stream (const ACE_Asynch_Write_Stream::Result &result)
{
//發(fā)送成功，RELEASE掉
//這個(gè)不可能有多個(gè)RELEASE，直接X(jué)X掉
//result.message_block ().release ();
MsgBlockManager::get_instance().release_msg_block(&result.message_block());
{
ACE_Guard<ACE_Recursive_Thread_Mutex> locker (_lock);
_io_count--;
}
check_destroy ();
}
//bool ClientHandler::destroy ()
//{
// FUNC_ENTER;
// ClientManager::get_instance().release_client_handle(this);
// FUNC_LEAVE;
// return false ;
//}
int ClientHandler::initiate_read_stream (void)
{
ACE_Guard<ACE_Recursive_Thread_Mutex> locker (_lock);
//考慮到粘包的呀
if (_rs.read (_read_msg_block, _read_msg_block.space()) == -1)
{
ACE_ERROR_RETURN ((LM_ERROR,"%p\n","ACE_Asynch_Read_Stream::read"),-1);
}
_io_count++;
return 0;
}
/**生成一個(gè)寫(xiě)請(qǐng)求
*
* \param mb 待發(fā)送的數(shù)據(jù)
* \param nBytes 待發(fā)送數(shù)據(jù)大小
* \return 0－成功，－1失敗
*/
int ClientHandler::initiate_write_stream (ACE_Message_Block & mb, size_t nBytes )
{
ACE_Guard<ACE_Recursive_Thread_Mutex> locker (_lock);
if (_ws.write (mb , nBytes ) == -1)
{
mb.release ();
ACE_ERROR_RETURN((LM_ERROR,"%p\n","ACE_Asynch_Write_File::write"),-1);
}
_io_count++;
return 0;
}
void ClientHandler::open (ACE_HANDLE handle,ACE_Message_Block &message_block)
{
//FUNC_ENTER;
_last_net_io=ACE_OS::time(NULL);
_io_count=0;
if(_ws.open(*this,this->handle())==-1)
{
ACE_ERROR ((LM_ERROR,"%p\n","ACE_Asynch_Write_Stream::open"));
}
else if (_rs.open (*this, this->handle()) == -1)
{
ACE_ERROR ((LM_ERROR,"%p\n","ACE_Asynch_Read_Stream::open"));
}
else
{
initiate_read_stream ();
}
check_destroy();
//FUNC_LEAVE;
}
void ClientHandler::fini()
{
}
void ClientHandler::check_time_out(time_t cur_time)
{
//ACE_Guard<ACE_Recursive_Thread_Mutex> locker (_lock);
//ACE_DEBUG((LM_DEBUG,"cur_time is %u,last io is %u\n",cur_time,_last_net_io));
//檢測(cè)是否已經(jīng)為0了
if(this->handle()==ACE_INVALID_HANDLE)
return;
if(cur_time-_last_net_io>CLIENT_TIME_OUT_SECONDS)
{
ACE_OS::shutdown(this->handle(),SD_BOTH);
ACE_OS::closesocket(this->handle());
this->handle(ACE_INVALID_HANDLE);
}
}
int ClientHandler::check_destroy()
{
{
ACE_Guard<ACE_Recursive_Thread_Mutex> locker (_lock);
if (_io_count> 0)
return 1;
}
ACE_OS::shutdown(this->handle(),SD_BOTH);
ACE_OS::closesocket(this->handle());
this->handle(ACE_INVALID_HANDLE);
//這個(gè)地方給內(nèi)存池吧。
ClientManager::get_instance().release_client_handle(this);
//delete this;
return 0;
}