|
|
Posted on 2009-01-31 19:47 S.l.e!ep.￠% 閱讀(3285) 評論(8) 編輯 收藏 引用 所屬分類: C++
1. 不使用線程池 2.? 暫時只封裝了 win32 的 icop 3.? 暫時還未寫測試用例，因為不知道怎么寫 -_-!!!  //?networksocket.h:?interface?for?the?CNetworkSocket?class.
//
  /**///////////////////////////////////////////////////////////////////////
#if?!defined(NETWORKSOCKET_H__A63ED8A5_7ED1_463D_A0F0_41F6E9C79441__INCLUDED_)
#define?NETWORKSOCKET_H__A63ED8A5_7ED1_463D_A0F0_41F6E9C79441__INCLUDED_
#if?_MSC_VER?>?1000
#pragma?once
#endif?//?_MSC_VER?>?1000
#include?<vector>
#include?<winsock2.h>
#include?<process.h>
#pragma??comment(lib,?"ws2_32.lib")
#define?BUFFER_SIZE?4096
enum?OPERATION_TYPE
 {
 ????OPERATION_TYPE_RECV?=?0,
????OPERATION_TYPE_SEND?=?1
 };
//IO操作數據
typedef?struct?_PER_IO_OPERATION_DATA
{
????OVERLAPPED?OverLapped;
????WSABUF?DataBuf;
????char?szBuf[BUFFER_SIZE];
????OPERATION_TYPE?OperationType;??????//?操作類型表示
}PER_IO_OPERATION_DATA,*PPER_IO_OPERATION_DATA;
class?CNetworkSocket??
{
public:
????CNetworkSocket();
????virtual?~CNetworkSocket();
????bool?init();
????bool?uninit();
????bool?initNetWork(unsigned?short?nPort);
????bool?unitNetWork();
????bool?send(SOCKET?s,?const?char*?pBuf,?int?nLen);
private:
????SOCKET?m_listensock;
????HANDLE?m_hIocpPort;
????std::vector<HANDLE>?m_vectorThreadHandle;
????static?unsigned?__stdcall?_WorkerThreadProc(void*?pVoid);????
????static?unsigned?__stdcall?_AcceptThreadProc(void*?pVoid);
};
#endif?//?!defined(NETWORKSOCKET_H__A63ED8A5_7ED1_463D_A0F0_41F6E9C79441__INCLUDED_)
//?networksocket.cpp:?implementation?of?the?CNetworkSocket?class.
//
/**///////////////////////////////////////////////////////////////////////
#include?"networksocket.h"
/**/////////////////////////////////////////////////////////////////////////?Construction/Destruction
/**///////////////////////////////////////////////////////////////////////
CNetworkSocket::CNetworkSocket()
{
????m_listensock?=?INVALID_SOCKET;
????m_hIocpPort??=?INVALID_HANDLE_VALUE;
}
CNetworkSocket::~CNetworkSocket()
{
}
bool?CNetworkSocket::init()
{
????WORD?wVersionRequested;
????WSADATA?wsaData;
????int?nErrCode?=?0;
????
????wVersionRequested?=?MAKEWORD(2,?2);
????
????nErrCode?=?::WSAStartup(wVersionRequested,?&wsaData);
????if?(?nErrCode?!=?0?)?
  ????{
????????return?false;
 ????}
????if?(?LOBYTE(wsaData.wVersion)?!=?2?||?HIBYTE(wsaData.wVersion)?!=?2?)?
????{
????????bool?b?=?uninit();
????????return?false;
????}
????else
????{
????????return?true;
????}
}
bool?CNetworkSocket::uninit()
{
????int?nErrCode?=?0;
????nErrCode?=?::WSACleanup();
????return?true;
}
bool?CNetworkSocket::initNetWork(unsigned?short?nPort)
{
????int?nErrCode?=?0;
????//?創建?socket
????m_listensock?=?::WSASocket(AF_INET,?SOCK_STREAM,?0,?NULL,?0,?WSA_FLAG_OVERLAPPED);
????
????if(?INVALID_SOCKET?==?m_listensock?)
????{
????????nErrCode?=?::WSAGetLastError();
????????return?false;
????}
????//?綁定端口
????sockaddr_in?InternetAddr;
????InternetAddr.sin_family?=?AF_INET;
????InternetAddr.sin_addr.S_un.S_addr?=?::htonl(INADDR_ANY);
????InternetAddr.sin_port?=?htons(nPort);
????if(?::bind(?m_listensock,?(PSOCKADDR)&InternetAddr,?sizeof(InternetAddr))?==?SOCKET_ERROR?)
????{
????????nErrCode?=?::WSAGetLastError();
????????return?false;
????}
????//?開始監聽
????if(????::listen(m_listensock,?5)?==?SOCKET_ERROR?)
????{
????????nErrCode?=?::WSAGetLastError();
????????return?false;
????}
????//?創建完成端口
????m_hIocpPort?=?::CreateIoCompletionPort(INVALID_HANDLE_VALUE,?NULL,?0,?0);
????????
????if(?m_hIocpPort?==?INVALID_HANDLE_VALUE?)
????{
????????nErrCode?=?::WSAGetLastError();
????????return?false;
????}
????//?啟動工作線程，線程數為CPU處理器數量*2+2
????SYSTEM_INFO?sys_Info;
????::GetSystemInfo(&sys_Info);
????for(int?i?=?0;?i?<?(int(sys_Info.dwNumberOfProcessors)?*?2?+?2);?i++)
????{
????????HANDLE?ThreadHandle?=?INVALID_HANDLE_VALUE;
????????DWORD?ThreadID??????=?0;
????????
????????unsigned?int?nThreadId??=???0?;
????????ThreadHandle?=?(HANDLE)_beginthreadex(NULL,?0,?_WorkerThreadProc,?(void?*)this,??0,?&nThreadId);
????????
????????if?(?ThreadHandle?==?0?)
????????????return?false;
????????m_vectorThreadHandle.push_back(ThreadHandle);
????}
????//?啟動偵聽線程
????HANDLE?ThreadHandle?=?INVALID_HANDLE_VALUE;
????unsigned?int?nThreadId?=?0?;
????ThreadHandle?=?(HANDLE)_beginthreadex(NULL,?0,?_AcceptThreadProc,?(void?*)this,??0,?&nThreadId);
????
????if?(?ThreadHandle?==?0?)
????????????return?false;
????m_vectorThreadHandle.push_back(ThreadHandle);
????return?true;
}
bool?CNetworkSocket::unitNetWork()
{
????//?啟動工作線程，線程數為CPU處理器數量*2+2
????SYSTEM_INFO?sys_Info;
????::GetSystemInfo(&sys_Info);
????
????for(int?i?=?0;?i?<?(int(sys_Info.dwNumberOfProcessors)?*?2?+?2);?i++)
????{
????????//寄出退出消息
????????::PostQueuedCompletionStatus(m_hIocpPort,????-1,????-1,?NULL);
????}
????std::vector<HANDLE>::iterator?iter_t;
????for?(?iter_t?=?m_vectorThreadHandle.begin();?iter_t?!=?m_vectorThreadHandle.end();?iter_t++?)
????{?
????????DWORD?dwRet?=?::WaitForSingleObject(*iter_t,?INFINITE);
????}
????//關閉網絡的偵聽
????::shutdown(m_listensock,?0);
????::closesocket(m_listensock);
????
????return?true;
}
unsigned?__stdcall?CNetworkSocket::_WorkerThreadProc(void*?pVoid)
{
????CNetworkSocket*?pThis?=?(CNetworkSocket*)pVoid;
????DWORD?dwByteTransferred?=?0;
????unsigned?long?nFlag?=?0;
????DWORD?RecvByte?=?0;
????SOCKET?ClientSock?=?INVALID_SOCKET;
?????PPER_IO_OPERATION_DATA?PerIoData;
?????while?(?true?)
?????{
????????BOOL?bSuccess?=?::GetQueuedCompletionStatus(pThis->m_hIocpPort,
???????????????????????????????????????????????????????&dwByteTransferred,
?????????????????????????????????????????????????????(LPDWORD)&ClientSock,
?????????????????????????????????????????????????????(LPOVERLAPPED*)&PerIoData,
?????????????????????????????????????????????????????INFINITE);
????????//退出信號到達，退出線程
????????if(?dwByteTransferred?==?-1?&&?PerIoData?==?NULL?)
????????????return?1;
????????//客戶機已經斷開連接或者連接出現錯誤
????????if(?dwByteTransferred?==?0?&&?(PerIoData->OperationType?==?OPERATION_TYPE_RECV?||?PerIoData->OperationType?==?OPERATION_TYPE_SEND?)?)
????????{
????????????::closesocket(ClientSock);
????????????::GlobalFree(PerIoData);
????????????continue;
????????}
????????//?接收完成
????????if?(?PerIoData->OperationType?==?OPERATION_TYPE_RECV?)
????????{
????????????//?處理接收數據
????????????//?pThis->OnRecv(ClientSock,?PerIoData->szBuf,?dwByteTransferred);
????????????//將源數據置空
????????????::memset(PerIoData->szBuf,?0,?BUFFER_SIZE);
????????????dwByteTransferred=0;
????????????//重置IO操作數據
????????????unsigned?long?Flag=0;
????????????::ZeroMemory(&(PerIoData->OverLapped),?sizeof(OVERLAPPED));
????????????
????????????PerIoData->DataBuf.buf????=?PerIoData->szBuf;
????????????PerIoData->DataBuf.len????=?BUFFER_SIZE;
????????????PerIoData->OperationType?=?OPERATION_TYPE_RECV;
????????????//?再投遞另一個Recv請求
????????????::WSARecv(ClientSock,?&(PerIoData->DataBuf),?1,?&RecvByte,?&Flag,?&(PerIoData->OverLapped),????NULL);
????????}
????????//?發送完成，置空緩沖區，釋放緩沖區
????????if(?PerIoData->OperationType?==?OPERATION_TYPE_SEND?)
????????{
????????????::memset(PerIoData,?0,?sizeof(PER_IO_OPERATION_DATA));
????????????::GlobalFree(PerIoData);
????????????dwByteTransferred?=?0;
????????}
?
?????}//?while?(?true?)
????return?0;
}
unsigned?__stdcall?CNetworkSocket::_AcceptThreadProc(void*?pVoid)
{
????CNetworkSocket*?pThis?=?(CNetworkSocket*)pVoid;
????SOCKET?AcceptSock?=?INVALID_SOCKET;
????while?(?true?)
????{
????????AcceptSock?=?::WSAAccept(pThis->m_listensock,?NULL,?NULL,?NULL,?0);
????????//關聯客戶端口到完成端口，句柄數據在此時被綁定到完成端口
????????::CreateIoCompletionPort((HANDLE)AcceptSock,?pThis->m_hIocpPort,?(DWORD)AcceptSock,?0);
????????PPER_IO_OPERATION_DATA?PerIoData?=?(PPER_IO_OPERATION_DATA)::GlobalAlloc(GPTR,?sizeof(PER_IO_OPERATION_DATA));
????????unsigned?long?nFlag?=?0;
????????DWORD?RecvByte?=?0;
????????::ZeroMemory(&(PerIoData->OverLapped),sizeof(OVERLAPPED));
????????
????????PerIoData->DataBuf.buf?????=?PerIoData->szBuf;
????????PerIoData->DataBuf.len?????=?BUFFER_SIZE;
????????PerIoData->OperationType??=?OPERATION_TYPE_RECV;
????????//提交首個接收數據請求
????????//這時
????????//如果客戶端斷開連接
????????//則也可以以接收數據時得到通知????
????????::WSARecv(AcceptSock,?&(PerIoData->DataBuf),?1,?&RecvByte,?&nFlag,?&(PerIoData->OverLapped),?NULL);
????}//?while?(?true?)
????return?0;
}
bool?CNetworkSocket::send(SOCKET?s,?const?char*?pBuf,?int?nLen)
{
????if(?s?==?INVALID_SOCKET?||?pBuf?==?NULL?||?nLen?==?0?)
????????return?false;
????
????PPER_IO_OPERATION_DATA?PerIoData?=?(PPER_IO_OPERATION_DATA)::GlobalAlloc(GPTR,?sizeof(PER_IO_OPERATION_DATA));
????
????unsigned?long?nFlag?=?0;
????DWORD?dwSendByte?=?0;
????::ZeroMemory(&(PerIoData->OverLapped),?sizeof(OVERLAPPED));
????::memcpy(PerIoData->szBuf,?pBuf,?nLen);
????PerIoData->DataBuf.buf?=?PerIoData->szBuf;
????PerIoData->DataBuf.len?=?nLen;
????PerIoData->OperationType??=?OPERATION_TYPE_SEND;
????int?nRet?=?::WSASend(s,?&(PerIoData->DataBuf),?1,?&dwSendByte,?nFlag,?&(PerIoData->OverLapped),?NULL);
????if(?nRet==SOCKET_ERROR?&&?GetLastError()?!=?WSA_IO_PENDING?)
????{
????????return?false;
????}
????else
????{
????????return?true;
????}
}
Feedback
在send時，重新分配了一個IO對象，這個對象也許不會在你想象中那樣被釋放了。
在GET中的釋放的地方，好像沒問題，但如果達到上千個連接，而且每個連接的數據非量密的時候，會發現GET時總是在判斷Recv的情況而沒有判斷Send的情況。
現在我一直在懷疑WSASend后，GET一直沒有取得完成的事件。
雖然都說肯定會有完成事件的。
總這，你這個類可以應付小型的網絡應用，但無法應付上規模的應用。
這個類還有一個問題，
就是當連接斷開后，沒有清理資源。
還有沒有主動檢測無效連接的功能。
在WinSock上使用IOCP
本文章假設你已經理解WindowsNT的I/O模型以及I/O完成端口(IOCP)，并且比較熟悉將要用到的API，如果你打算學習IOCP，請參考Jeffery Richter的Advanced Windows(第三版)，第15章I/O設備，里面有極好的關于完成端口的討論以及對即將使用API的說明。
IOCP提供了一個用于開發高效率和易擴展程序的模型。Winsock2提供了對IOCP的支持，并在WindowsNT平臺得到了完整的實現。然而IOCP是所有WindowsNT I/O模型中最難理解和實現的，為了幫助你使用IOCP設計一個更好的Socket服務，本文提供了一些訣竅。
Tip 1:使用Winsock2 IOCP函數例如WSASend和WSARecv，如同Win32文件I/O函數，例如WriteFile和ReadFile。
微軟提供的Socket句柄是一個可安裝文件系統(IFS)句柄，因此你可以使用Win32的文件I/O函數調用這個句柄，然而，將Socket句柄和文件系統聯系起來，你不得不陷入很多的Kernal/User模式轉換的問題中，例如線程的上下文轉換，花費的代價還包括參數的重新排列導致的性能降低。
因此你應該使用只被Winsock2中IOCP允許的函數來使用IOCP。在ReadFile和WriteFile中會發生的額外的參數重整以及模式轉換只會發生在一種情況下，那就是如果句柄的提供者并沒有將自己的WSAPROTOCOL_INFO結構中的DwServiceFlags1設置為XP1_IFS_HANDLES。
注解：即使使用WSASend和WSARecv，這些提供者仍然具有不可避免的額外的模式轉換，當然ReadFile和WriteFile需要更多的轉換。
TIP 2: 確定并發工作線程數量和產生的工作線程總量。
并發工作線程的數量和工作線程的數量并不是同一概念。你可以決定IOCP使用最多2個的并發線程以及包括10個工作線程的線程池。工作線程池擁有的線程多于或者等于并發線程的數量時，工作線程處理隊列中一個封包的時候可以調用win32的Wait函數，這樣可以無延遲的處理隊列中另外的封包。
如果隊列中有正在等待被處理的封包，系統將會喚醒一個工作線程處理他，最后，第一個線程確認正在休眠并且可以被再次調用，此時，可調用線程數量會多于IOCP允許的并發線程數量(例如,NumberOFConcurrentThreads)。然而，當下一個線程調用GetQueueCompletionStatus并且進入等待狀態，系統不會喚醒他。一般來說，系統會試圖保持你設定的并發工作線程數量。
一般來講，每擁有一個CPU，在IOCP中你可以使用一個并發工作線程，要做到這點，當你第一次初始化IOCP的時候，可以在調用CreateIOCompletionPort的時候將NumberOfConcurrentThreads設置為0。
TIP 3：將一個提交的I/O操作和完成封包的出列聯系起來。
當對一個封包進行出列，可以調用GetQueuedCompletionStatus返回一個完成Key和一個復合的結構體給I/O。你可以分別的使用這兩個結構體來返回一個句柄和一個I/O操作信息，當你將IOCP提供的句柄信息注冊給Socket，那么你可以將注冊的Socket句柄當做一個完成Key來使用。為每一個I/O的"extend"操作提供一個包含你的應用程序IO狀態信息的復合結構體。當然，必須確定你為每個的I/O提供的是唯一的復合結構體。當I/O完成的時候，會返回一個指向結構體的指針。
TIP 4:I/O完成封包隊列的行為
IOCP中完成封包隊列的等待次序并不決定于Winsock2 I/O調用產生的順序。如果一個Winsock2的I/O調用返回了SUCCESS或者IO_PENDING，那么他保證當I/O操作完成后，完成封包會進入IOCP的等待隊列，而不管Socket句柄是否已經關閉。如果你關閉了socket句柄，那么將來調用WSASend,WSASendTo,WSARecv和WSARecvFrom會失敗并返回一個不同于SUCCES或者IO_PENDING的代碼，這時將不會產生一個完成封包。而在這種情況下，前一次使用GetQueuedCompletionStatus提交的I/O操作所得到的完成封包，會顯示一個失敗的信息。
如果你刪除了IOCP本身，那么不會有任何I/O請求發送給IOCP，因為IOCP的句柄已經不可用，盡管系統底層的IOCP核心結構并不會在所有已提交I/O請求完成之前被移除。
TIP5:IOCP的清除
很重要的一件事是使用復合I/O時候的IOCP清除：如果一個I/O操作尚未完成，那么千萬不要釋放該操作創建的復合結構體。HasOverlappedIoCompleted函數可以幫助你檢查一個I/O操作是否已經完成。
關閉服務一般有兩種情況，第一種你并不關心尚未結束的I/O操作的完成狀態，你只希望盡可能快的關閉他。第二種，你打算關閉服務，但是你需要獲知未結束I/O操作的完成狀態。
第一種情況你可以調用PostQueueCompletionStatus(N次，N等于你的工作線程數量)來提交一個特殊的完成封包，他通知所有的工作線程立即退出，關閉所有socket句柄和他們關聯的復合結構體，然后關閉完成端口(IOCP)。在關閉復合結構體之前使用HasOverlappedIOCompleted檢查他的完成狀態。如果一個socket關閉了，所有基于他的未結束的I/O操作會很快的完成。
在第二種情況，你可以延遲工作線程的退出來保證所有的完成封包可以被適當的出列。你可以首先關閉所有的socket句柄和IOCP。可是，你需要維護一個未完成I/O的數字，以便你的線程可以知道可以安全退出的時間。盡管當隊列中有很多完成封包在等待的時候，活動的工作線程不能立即退出，但是在IOCP服務中使用全局I/O計數器并且使用臨界區保護他的代價并不會象你想象的那樣昂貴。
因為我發現當大理recv后，就沒機會處理send的完成事件了。
如果在recv有同步或有復雜的邏輯，比如recv后需要send幾個，此時可能產生死鎖，以致永遠沒機會get到send的完成事件。
而如果IOCP本身就能將這兩個事件用兩個不同的GET分開，那業務邏輯就也好處理一些。
@苦丁茶
我也在寫iocp，對于你的評論，我認為非常正確，但是到現在為止，還沒有找到一個好的辦法處理send的完成事件。你找到了嗎
|