开发工程师网站开发工程师招聘,wordpress继续阅读,医院网站建设模板,没有备案网站可以做优化么【完成端口模型导读】完成端口模型#xff0c;算得上是真正的异步网络IO模型吧#xff0c;相对于其它网络IO模型#xff0c;操作系统通知我们的时候#xff0c;要么就是连接已经帮我建立好#xff0c;客户端套接字帮我们准备好#xff1b;要么就是数据已经接收完成#…
【完成端口模型导读】完成端口模型算得上是真正的异步网络IO模型吧相对于其它网络IO模型操作系统通知我们的时候要么就是连接已经帮我建立好客户端套接字帮我们准备好要么就是数据已经接收完成要么就是本端的数据已经发送出去了。我们只需准备接收数据的容器以及客户端的套接字即可。 1、重难点分析 使用完成端口模型的几个主要步骤
1、创建完成端口模型2、创建监听socket并将监听socket绑定到指定的IP地址和端口3、将监听socket绑定到完成端口模型上去 以上步骤可以使用如下的示例代码进行概述
HANDLE hCompletePort CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);//监听socket必须使用WSASocket接口创建而且要加上WSA_FLAG_OVERLAPPED标记 HANDLE listenSocket WSASocket(AF_INET, SOCK_STREAM, 0, NULL, 0, WSA_FLAG_OVERLAPPED);// 将监听Socket绑定到完成端口struct per_socket_context listenContext;//per_socket_context下文会介绍。if( NULL CreateIoCompletionPort((HANDLE)listenSocket, hCompletePort,(ULONG_PTR)listenContext,0)) { return;} //开启监听struct sockaddr_in serverAddress;// 填充地址信息ZeroMemory((char *)serverAddress, sizeof(serverAddress));serverAddress.sin_family AF_INET;// 这里可以绑定任何可用的IP地址或者绑定一个指定的IP地址 serverAddress.sin_addr.s_addr htonl(INADDR_ANY); serverAddress.sin_port htons(8000);if (SOCKET_ERROR bind(listenSocket, (struct sockaddr_in*)serverAddress,sizeof(serverAddress))){return;}if (listen(listenSocket, SOMAXCONN) SOCKET_ERROR){return;} 正如前文所说完成端口模型帮我把连接建立我们只需要准备好客户端套接字即可可以这么理解吧应用层只需要投递个接收连接的请求等到操作系统通知我们的时候连接已经建立好客户端套接字已经生效它表示一个合法的客户端对象此时可以通过这个客户端套接字和对端建立连接进行数据通信。 那投递接收连接请求的接口是哪个 就是AcceptEx。
1、接口声明
BOOL AcceptEx(_In_ SOCKET sListenSocket, //监听套接字_In_ SOCKET sAcceptSocket, //客户端套接字使用完下次投递请求的时候需再补充一个_In_ PVOID lpOutputBuffer, //接收在新连接上发送的第一个数据块_In_ DWORD dwReceiveDataLength, //实际接收数据的字节数_In_ DWORD dwLocalAddressLength, //本地地址信息保留的字节数_In_ DWORD dwRemoteAddressLength, //远程地址信息保留的字节数_Out_ LPDWORD lpdwBytesReceived, //指向接收接收字节计数的DWORD指针_In_ LPOVERLAPPED lpOverlapped //用于处理请求的 OVERLAPPED 结构。必须指定此参数;它不能为 NULL
);2、接口返回值AcceptEx函数成功完成返回值为TRUE 。函数失败AcceptEx返回FALSE。可调用WSAGetLastError返回扩展错误信息。如果 WSAGetLastError 返回 ERROR_IO_PENDING则表示操作已成功启动仍在进行中。 如果是其它错误说明真的出错了。 虽然微软提供了接口的声明根据msdn的官方声明我们还是使用其它的方式去获取AcceptEx函数指针具体的方式如下 DWORD dwBytes 0;
if(SOCKET_ERROR WSAIoctl(listenSocket, //任意一个有效的socket即可 SIO_GET_EXTENSION_FUNCTION_POINTER, GuidAcceptEx, sizeof(GuidAcceptEx), m_lpfnAcceptEx, sizeof(m_lpfnAcceptEx), dwBytes, NULL, NULL))
{return;
} m_lpfnAcceptEx就是我们拿到的AcceptEx函数指针我们可以使用它来投递一个接收连接的请求。 好介绍完投递接收连接请求的接口我们再看看投递接收数据的接口又是怎样的 int WSAAPI WSARecv(SOCKET s, // 客户套接字LPWSABUF lpBuffers,//指向 WSABUF 结构数组的指针DWORD dwBufferCount, //lpBuffers数组中的WSABUF结构数。LPDWORD lpNumberOfBytesRecvd,//接收的数据数以字节为单位的指针LPDWORD lpFlags,//默认为0LPWSAOVERLAPPED lpOverlapped, //overlapped结构指针LPWSAOVERLAPPED_COMPLETION_ROUTINE lpCompletionRoutine
); 那如何投递接收数据的请求
// 初始化变量
DWORD dwFlags 0;
DWORD dwBytes 0;
WSABUF *Wbuf;
OVERLAPPED *pol pIoContext-m_Overlapped;
pIoContext-ResetBuffer();
pIoContext-m_OpType RECV_POSTED;
int nBytesRecv WSARecv(clientSock, Wbuf, 1, dwBytes, dwFlags, pol, NULL );if ((SOCKET_ERROR nBytesRecv) (WSA_IO_PENDING ! WSAGetLastError()))return; 投递发送数据的请求也是类似
// 初始化变量
DWORD dwFlags 0;
DWORD dwBytes 0;
WSABUF *p_wbuf;
OVERLAPPED *pol pIoContext-m_Overlapped;
pIoContext-ResetBuffer();
pIoContext-m_OpType SEND_POSTED;
// 初始化完成后投递WSARecv请求
int nBytesSend WSASend(ClientSock, p_wbuf, 1, dwBytes, dwFlags, p_ol, NULL);
// 如果返回值错误并且错误的代码并非是Pending的话那就说明这个重叠请求失败了
if ((SOCKET_ERROR nBytesSend) (WSA_IO_PENDING ! WSAGetLastError()))
{return ;
} 投递成功后如果对端有数据发送过来那么完成端口的工作线程便会被唤醒我们将从工作线程拿到对端发送过来的数据亦或数据成功发送给对端那么工作线程也会被唤醒这个时候是继续投递发送请求还是投递一个接收数据的请求就看自己的业务需要了。 正如上文所说投递完请求工作线程被唤醒这个时候如何区分是发送成功还是接收成功是监听套接字的事件还是客户端套接字的事件。介绍完如何区分这些差别之前我们先看看工作线程是被什么API给挂起的以至于有事件来时能被唤醒。
BOOL WINAPI GetQueuedCompletionStatus(_In_ HANDLE CompletionPort, //完成端口句柄_Out_ LPDWORD lpNumberOfBytes, //指向接收已完成 I/O 操作中传输的字节数的变量的指针。/*指向和完成端口相关附加数据和CreateIoCompletionPort第三个参数是同一个指针指向同一份内存。因此在将socket绑定到完成端口时可以在lpCompletionKey中指定socket的类型那么工作线程就知道当前的事件是连接套接字产生的还是客户端套接字产生的*/_Out_ PULONG_PTR lpCompletionKey, _Out_ LPOVERLAPPED *lpOverlapped,_In_ DWORD dwMilliseconds
); 像lpCompletionKey和socket相关的结构体指针我们称之为单socket数据具体的用法如下
typedef struct _PER_SOCKET_CONTEXT
{//每一个客户端连接的SocketSOCKET m_Socket; //客户端的地址SOCKADDR_IN m_ClientAddr;//客户端网络操作的上下文数据CArray_PER_IO_CONTEXT* m_arrayIoContext;
};//对于侦听socket
_PER_SOCKET_CONTEXT listenSocketContext;
listenSocketContext.s ListenSocket;
CreateIoCompletionPort(ListenSocket, m_hIOCompletionPort, (ULONG_PTR)listenSocketContext,0);//对于普通客户端连接socket
_PER_SOCKET_CONTEXT clientSocketContext;
clientSocketContext.s acceptSocket;
CreateIoCompletionPort(acceptSocket, m_hIOCompletionPort,(ULONG_PTR)clientSocketContext,0); 那么工作线程可以从对应的socketContext中解析出socket的值具体是监听socket还是客户端socket那可以做区分处理。
DWORD ThreadFunction()
{OVERLAPPED *pOverlapped NULL;_PER_SOCKET_CONTEXT *pSocketContext NULL;DWORD dwBytesTransfered 0;BOOL bReturn GetQueuedCompletionStatus(m_hIOCompletionPort,dwBytesTransfered, (PULONG_PTR)pSocketContext, pOverlapped, INFINITE);if (((SOME_STRUCT*)pSocketContext)-s 监听socket){//新连接接收成功,做一些操作}else //客户端套接字{if (事件类型 收到了一份数据){//解析数据}else if (事件类型 数据发送成功了){//继续投递发送数据请求}}
} 上述伪码中假如是客户端套接字的事件那么如何去区分事件类型到底是数据发送成功还是数据接收成功PER_SOCKET_DATA中是没有表明事件类型的。具体是发送数据还是接收数据我们在投递发送请求或者接收请求的时候是可以标注的我们再回过头看看WSARecv接口的声明。 int WSAAPI WSARecv(SOCKET s, // 客户套接字......LPWSAOVERLAPPED lpOverlapped, //overlapped结构指针LPWSAOVERLAPPED_COMPLETION_ROUTINE lpCompletionRoutine
); OVERLAPPED结构体指针lpOverlapped基于指针的伸缩性我们可以通过偏移访问指定范围的内存数据假如我们在OVERLAPPED结构后面附加一些额外的数据用于标识当前的IO行为是接收还是发送那么工作线程被唤醒时就可以通过lpOverlapped指针取出附加在OVERLAPPED后面的数据了那么具体的事件类型就能很轻松获取到了。针对这种标识IO行为的结构体我们称为PER_IO_DATA。 typedef struct _PER_IO_CONTEXT
{// 每一个重叠网络操作的重叠结构(针对每一个Socket的每一个操作都要有一个) OVERLAPPED m_Overlapped;// 这个网络操作所使用的SocketSOCKET m_sockAccept; // WSA类型的缓冲区用于给重叠操作传参数的WSABUF m_wsaBuf; // 这个是WSABUF里具体存字符的缓冲区char m_szBuffer[MAX_BUFFER_LEN]; // 标识网络操作的类型(对应上面的枚举), 发送或者接收事件OPERATION_TYPE m_OpType;
}; 那上文提到的工作线程就可以这样做改造了
DWORD ThreadFunction()
{OVERLAPPED *pOverlapped NULL;_PER_SOCKET_CONTEXT *pSocketContext NULL;DWORD dwBytesTransfered 0;BOOL bReturn GetQueuedCompletionStatus(m_hIOCompletionPort,dwBytesTransfered, (PULONG_PTR)pSocketContext, pOverlapped, INFINITE);if (((SOME_STRUCT*)pSocketContext)-s 监听socket){//新连接接收成功,做一些操作}else //客户端套接字{//推荐使用微软封装的接口进行转换PER_IO_CONTEXT* pIoContext CONTAINING_RECORD(pOverlapped, PER_IO_CONTEXT, m_Overlapped);if (pIOContext-Type 接收){//解析数据}else if (pIOContext-Type 发送){}}
}
有事件产生时工作线程会被唤醒假如我们想主动退出工作线程那这个时候该如何唤醒挂在GetQueuedCompletionStatus接口的线程既然有Get接口那肯定就有Post接口微软提供如下接口用于唤醒挂在GetQueuedCompletionStatus接口的线程。 BOOL WINAPI PostQueuedCompletionStatus(_In_ HANDLE CompletionPort,_In_ DWORD dwNumberOfBytesTransferred,_In_ ULONG_PTR dwCompletionKey,_In_opt_ LPOVERLAPPED lpOverlapped
); 假设我们将第三个参数dwCompletionKey指向一个特殊的标识符(EXIT_CODE)那么工作线程被唤醒时是不是可以从dwCompletionKey解析出这个特殊的标识符(EXIT_CODE)随后退出工作线程。
PostQueuedCompletionStatus(m_hIOCompletionPort, 0, (DWORD)EXIT_CODE, NULL);DWORD ThreadFunction()
{OVERLAPPED *pOverlapped NULL;PER_SOCKET_CONTEXT *pSocketContext NULL;DWORD dwBytesTransfered 0;BOOL bReturn GetQueuedCompletionStatus(m_hIOCompletionPort, dwBytesTransfered, (PULONG_PTR)pSocketContext, pOverlapped, INFINITE);//收到退出标志退出线程if ( EXIT_CODE(DWORD)pSocketContext ){return 0;}......
}
2、完整示例代码 IOCPModel.cpp
#include CIOCPModel.h#define DEFAULT_NUM_OF_PROCESSOR 2#define RELEASE_HANDLE(x) if(x ! NULL x ! INVALID_HANDLE_VALUE ) if(x ! NULL){ CloseHandle(x); x NULL;}
#define RELEASE_SOCKET(x) if(x! NULL x! INVALID_SOCKET){closesocket(x); x NULL;}using namespace std;DWORD __stdcall CIOCPModel::_WorkThread(LPVOID lpParam)
{THREADPARAMS_WORKER *pThreadParam (PTHREADPARAMS_WORKER)lpParam;CIOCPModel *pIoModel pThreadParam-pIOCPModel;size_t ThreadNo pThreadParam-dwThreadNo;//线程等到系统的IO通知需要准备三个数据结构OVERLAPPED *pOverLapped NULL;PER_SOCKET_CONTEXT *perSocket NULL;DWORD dwByteTransfered 0;while (WAIT_OBJECT_0 ! WaitForSingleObject(pIoModel-m_hShutDownEvent, 0)){bool bFlag GetQueuedCompletionStatus(pIoModel-m_hIOCompletionPort,dwByteTransfered, (PULONG_PTR)perSocket, pOverLapped, INFINITE);OVERLAPPED m_Overlapped;//先处理主动退出的逻辑4if(EXIT_CODE (DWORD)perSocket){break;}if (!bFlag){continue;}else{PER_IO_CONTEXT *perIoContext CONTAINING_RECORD(pOverLapped, PER_IO_CONTEXT, m_Overlapped);if ((dwByteTransfered 0) (perIoContext-m_OperatorType RECV_POSTED|| perIoContext-m_OperatorType SEND_POSTED)){cout Client is closed. endl;pIoModel-_RemoveContextList(perSocket);continue;}switch (perIoContext-m_OperatorType){case RECV_POSTED://pIoModel-_PostRecv(perIoContext);pIoModel-_DoRecv(perSocket, perIoContext);break;case SEND_POSTED:pIoModel-_PostSend(perIoContext);break;case ACCEPT_POSTED:pIoModel-_DoAccept(perSocket, perIoContext);break;default:break;}}}if (pThreadParam ! NULL){delete pThreadParam;}return 0;
}CIOCPModel::CIOCPModel():m_nThreads(0),m_hShutDownEvent(NULL),m_hIOCompletionPort(NULL),m_lpfnAcceptEx(NULL),m_lpfnGetAcceptExSockAddrs(NULL),m_WorkThread(NULL),m_strIP(DEFAULT_IP),m_port(DEFAULT_PORT),pListenSocketContext(NULL){}CIOCPModel::~CIOCPModel()
{this-stop();
}bool CIOCPModel::start()
{InitializeCriticalSection(m_cs);m_hShutDownEvent CreateEvent(NULL, TRUE, FALSE, NULL);if (this-_InitIOCPModel()){std::cout initialize IOCP Model success. std::endl;}else{std::cout initialize IOCP Model failed. std::endl;return false;}//初始化监听socketif (this-_InitializeListenSocket()){std::cout initialize Listen socket success std::endl;}else{std::cout initialize Listen socket error std::endl;return false;}return true;
}
void CIOCPModel::stop()
{//主动退出OVERLAPPED *pOverlapped NULL;PER_SOCKET_CONTEXT *perSocketContext NULL;DWORD dwByteTransfered 0;if (pListenSocketContext ! NULL pListenSocketContext-m_Socket ! INVALID_SOCKET){//先将m_hShutDownEvent设置为有信号的状态SetEvent(m_hShutDownEvent);for (int i 0; i m_nThreads; i){PostQueuedCompletionStatus(m_hIOCompletionPort, dwByteTransfered, (DWORD)EXIT_CODE,pOverlapped);}WaitForMultipleObjects(m_nThreads, m_WorkThread, true, INFINITE);}
}
bool CIOCPModel::LoadSocketLib()
{return true;
}
void CIOCPModel::unLoadSocketLib()
{}
std::string CIOCPModel::GetLocalIP()
{char hostName[256] { 0 };gethostname(hostName, 256);struct hostent FAR* lpHostEnt gethostbyname(hostName);if (lpHostEnt NULL)return 127.0.0.1;LPSTR lpAddr lpHostEnt-h_addr_list[0];struct in_addr inAddr;memmove(inAddr, lpAddr, 4);m_strIP inet_ntoa(inAddr);return m_strIP;
}
void CIOCPModel::setPort(const int m_port)
{}bool CIOCPModel::_InitIOCPModel()
{//创建完成端口模型m_hIOCompletionPort CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, NULL, 0);if (m_hIOCompletionPort NULL){std::cout 创建完成端口模型失败 std::endl;return false;}//初始化一个线程池经验值cpu的数量乘以2m_nThreads 2 * _GetNoOfProcessor();m_WorkThread new HANDLE[m_nThreads];DWORD nThreadId;for (int i 0; i m_nThreads; i){THREADPARAMS_WORKER *pWorkThreadParam new THREADPARAMS_WORKER;pWorkThreadParam-pIOCPModel this;pWorkThreadParam-dwThreadNo i 1;m_WorkThread[i] ::CreateThread(0, 0, _WorkThread, (LPVOID)pWorkThreadParam, 0, nThreadId);}std::cout 创建了 m_nThreads 个线程 std::endl;return true;
}bool CIOCPModel::_InitializeListenSocket()
{GUID GuidAcceptEx WSAID_ACCEPTEX;GUID GuidGetAcceptExSockAddrs WSAID_GETACCEPTEXSOCKADDRS;struct sockaddr_in serverAddr;pListenSocketContext new PER_SOCKET_CONTEXT;pListenSocketContext-m_Socket WSASocket(AF_INET, SOCK_STREAM, 0, NULL,0, WSA_FLAG_OVERLAPPED);if (pListenSocketContext-m_Socket INVALID_SOCKET){std::cout create listen socket error std::endl;return false;}//将监听套接字绑定到完成端口上if (NULL CreateIoCompletionPort((HANDLE)pListenSocketContext-m_Socket,m_hIOCompletionPort, (DWORD)pListenSocketContext, 0)){std::cout Associate completion port to listen socket error std::endl;return false;}//绑定地址ZeroMemory(serverAddr, sizeof(serverAddr));serverAddr.sin_addr.S_un.S_addr htonl(INADDR_ANY);serverAddr.sin_port htons(DEFAULT_PORT);serverAddr.sin_family AF_INET;if (bind(pListenSocketContext-m_Socket, (struct sockaddr *)serverAddr, sizeof(serverAddr)) -1){std::cout bind error std::endl;return false;}if (-1 listen(pListenSocketContext-m_Socket, SOMAXCONN)){std::cout listen error std::endl;return false;}DWORD dwByets 0;if (WSAIoctl(pListenSocketContext-m_Socket,SIO_GET_EXTENSION_FUNCTION_POINTER,GuidAcceptEx,sizeof(GuidAcceptEx),m_lpfnAcceptEx,sizeof(m_lpfnAcceptEx),dwByets,0,0) -1){std::cout get acceptex error std::endl;return false;}if (WSAIoctl(pListenSocketContext-m_Socket,SIO_GET_EXTENSION_FUNCTION_POINTER,GuidGetAcceptExSockAddrs,sizeof(GuidGetAcceptExSockAddrs),m_lpfnGetAcceptExSockAddrs,sizeof(m_lpfnGetAcceptExSockAddrs),dwByets,0,0) -1){std::cout get GuidGetAcceptExSockAddrs error std::endl;return false;}//事先准备五个客户端套接字for (int i 0; i 10; i){PER_IO_CONTEXT *PerIoContext pListenSocketContext-GetNewIOContext();if (this-_PostAccept(PerIoContext) false){std::cout post accept error std::endl;return false;}}return true;
}void CIOCPModel::_DeInitializeIOCPModel()
{DeleteCriticalSection(m_cs);RELEASE_HANDLE(m_hShutDownEvent);for (int i 0; i m_nThreads; i){RELEASE_HANDLE(this-m_WorkThread[i]);}delete []this-m_WorkThread;RELEASE_HANDLE(m_hIOCompletionPort);delete pListenSocketContext;return;
}bool CIOCPModel::_PostAccept(PER_IO_CONTEXT *pAcceptIoContext)
{//投递Accept请求需要事先准备好客户端套接字assert(pListenSocketContext-m_Socket ! INVALID_SOCKET);DWORD dwBytes 0;pAcceptIoContext-m_OperatorType ACCEPT_POSTED;WSABUF *wsaBuf pAcceptIoContext-m_wsaBuf;OVERLAPPED *p_ol pAcceptIoContext-m_Overlapped;pAcceptIoContext-m_AcceptSocket WSASocket(AF_INET, SOCK_STREAM, 0, NULL,0, WSA_FLAG_OVERLAPPED);if (pAcceptIoContext-m_AcceptSocket INVALID_SOCKET){std::cout create accept socket error std::endl;return false;}if (false m_lpfnAcceptEx(pListenSocketContext-m_Socket, pAcceptIoContext-m_AcceptSocket,wsaBuf-buf, wsaBuf-len - (sizeof(SOCKADDR_IN) 16) *2 ,sizeof(SOCKADDR_IN) 16, sizeof(SOCKADDR_IN) 16,dwBytes, p_ol)){DWORD dwError WSAGetLastError();if (dwError ! WSA_IO_PENDING){std::cout Post Accept error std::endl;return false;}}return true;
}
bool CIOCPModel::_PostRecv(PER_IO_CONTEXT *PerContext)
{//投递Recv请求DWORD dwBytes 0;DWORD dwFlag 0;WSABUF *m_wsaBuf PerContext-m_wsaBuf;OVERLAPPED *o_lp PerContext-m_Overlapped;PerContext-ResetBuffer();PerContext-m_OperatorType RECV_POSTED;int dwBytesReturn WSARecv(PerContext-m_AcceptSocket, m_wsaBuf, 1, dwBytes, dwFlag, o_lp, NULL);if (dwBytesReturn -1){if (WSAGetLastError() ! WSA_IO_PENDING){std::cout post recv error WSAGetLastError() std::endl;return false;}}//cout ThreadId: GetCurrentThreadId() recv from client: m_wsaBuf-buf Len: m_wsaBuf-len endl;return true;
}bool CIOCPModel::_PostSend(PER_IO_CONTEXT *PerContext)
{DWORD dwBytes 0;DWORD dwFlag 0;WSABUF *m_wsaBuf PerContext-m_wsaBuf;OVERLAPPED *o_l PerContext-m_Overlapped;PerContext-m_OperatorType SEND_POSTED;int dwBytesReturn WSASend(PerContext-m_AcceptSocket, m_wsaBuf, 1, dwBytes, dwFlag, o_l, NULL);if (dwBytesReturn -1){if (WSAGetLastError() WSA_IO_PENDING){cout Post send error endl;return false;}}return true;
}bool CIOCPModel::_DoAccept(PER_SOCKET_CONTEXT *PerSocket, PER_IO_CONTEXT *PerContext)
{SOCKADDR_IN *ClientAddr NULL;SOCKADDR_IN *ServerAddr NULL;int dwRemoteAddrLen sizeof(SOCKADDR_IN);int dwLocalAddrLen sizeof(SOCKADDR_IN);//把客户端和服务端的IP地址拿到顺带把服务端发过来的第一组数据给拿到this-m_lpfnGetAcceptExSockAddrs(PerContext-m_wsaBuf.buf,PerContext-m_wsaBuf.len - (sizeof(SOCKADDR_IN) 16) * 2,sizeof(SOCKADDR_IN) 16, sizeof(SOCKADDR_IN) 16,(LPSOCKADDR *)ServerAddr, dwLocalAddrLen, (LPSOCKADDR *)ClientAddr, dwRemoteAddrLen);std::cout 客户端地址 inet_ntoa(ClientAddr-sin_addr) 端口: ClientAddr-sin_port std::endl;std::cout 服务端地址: inet_ntoa(ServerAddr-sin_addr) 端口: ServerAddr-sin_port std::endl;std::cout 客户端的发来的数据: PerContext-m_wsaBuf.buf endl;PER_SOCKET_CONTEXT *PerSocketContext new PER_SOCKET_CONTEXT;PerSocketContext-m_Socket PerContext-m_AcceptSocket;memcpy(PerSocketContext-m_ClientAddr, ClientAddr, sizeof(SOCKADDR_IN));//将Accept Socket和完成端口绑定在一起if (false this-_AssociateWithIOCP(PerSocketContext)){std::cout Associate accept scket error std::endl;return false;}PER_IO_CONTEXT *PerIoContext PerSocketContext-GetNewIOContext();PerIoContext-ResetBuffer();PerIoContext-m_OperatorType RECV_POSTED;PerIoContext-m_AcceptSocket PerSocketContext-m_Socket;//开始投递一个Recv请求if (false this-_PostRecv(PerIoContext)){if (WSAGetLastError() ! WSA_IO_PENDING){cout Do Accept Post Recv failed. WSAGetLastError() endl;PerSocket-RemoveIOContext(PerIoContext);return false;}}//投递成功将该请求添加到PerSocketContext队列this-_AddToContextList(PerSocket);//清空缓冲区PerIoContext-ResetBuffer();return this-_PostRecv(PerIoContext);
}bool CIOCPModel::_DoRecv(PER_SOCKET_CONTEXT *PerSocketContext, PER_IO_CONTEXT *PerIoContext)
{SOCKADDR_IN *ClientAddr PerSocketContext-m_ClientAddr;cout Client Addr: inet_ntoa(ClientAddr-sin_addr) Port: ClientAddr-sin_port PerIoContext-m_wsaBuf.buf len: PerIoContext-m_wsaBuf.len endl;//开始投递下一个请求return this-_PostRecv(PerIoContext);
}void CIOCPModel::_AddToContextList(PER_SOCKET_CONTEXT *pSocketContext)
{EnterCriticalSection(m_cs);m_ClientContextDeque.push_back(pSocketContext);LeaveCriticalSection(m_cs);return;
}void CIOCPModel::_RemoveContextList(PER_SOCKET_CONTEXT *pSocketContext)
{EnterCriticalSection(m_cs);PER_SOCKET_CONTEXT *pTempSocket m_ClientContextDeque.front();m_ClientContextDeque.pop_front();delete pTempSocket;LeaveCriticalSection(m_cs);
}void CIOCPModel::_ClearContextList()
{EnterCriticalSection(m_cs);m_ClientContextDeque.clear();LeaveCriticalSection(m_cs);
}bool CIOCPModel::_AssociateWithIOCP(PER_SOCKET_CONTEXT *pSocketContext)
{//将对应的句柄绑定到完成端口中区HANDLE hCompletePort CreateIoCompletionPort((HANDLE)pSocketContext-m_Socket,m_hIOCompletionPort, (DWORD)pSocketContext, 0);if (hCompletePort NULL){cout Associate Io Complettion Port failed. endl;return false;}return true;
}bool CIOCPModel::HandleError(PER_SOCKET_CONTEXT *pSocketContext, const DWORD dwError)
{return true;
}int CIOCPModel::_GetNoOfProcessor()
{SYSTEM_INFO si;GetSystemInfo(si);return si.dwNumberOfProcessors;
}bool CIOCPModel::_IsSocketAlive(SOCKET s)
{int nBytesSend send(s, a, 1, 0);if (nBytesSend -1)return false;return true;
}IOCPModel.h
#pragma once
#ifndef _C_IOCP_MODEL_H_
#define _C_IOCP_MODEL_H_
#include iostream
#include WinSock2.h
#include algorithm
#include deque
#include vector
#include assert.h
#include MSWSock.h
#pragma comment(lib,ws2_32.lib)#define MAX_BUFFER_LEN 8192
#define DEFAULT_PORT 12345
#define DEFAULT_IP 127.0.0.1#define EXIT_CODE 0x111typedef enum _OPERATION_TYPE
{ACCEPT_POSTED,SEND_POSTED,RECV_POSTED,NULL_POSTED
}OPERATION_TYPE;typedef struct _PER_IO_CONTEXT
{OVERLAPPED m_Overlapped;SOCKET m_AcceptSocket;WSABUF m_wsaBuf;char m_szBuf[MAX_BUFFER_LEN];OPERATION_TYPE m_OperatorType;_PER_IO_CONTEXT(){ZeroMemory(m_Overlapped, sizeof(m_Overlapped));ZeroMemory(m_szBuf, MAX_BUFFER_LEN);m_AcceptSocket INVALID_SOCKET;m_wsaBuf.buf m_szBuf;m_wsaBuf.len MAX_BUFFER_LEN;m_OperatorType NULL_POSTED;}~_PER_IO_CONTEXT(){if (m_AcceptSocket ! INVALID_SOCKET){closesocket(m_AcceptSocket);m_AcceptSocket INVALID_SOCKET;}}void ResetBuffer(){ZeroMemory(m_szBuf, MAX_BUFFER_LEN);}}PER_IO_CONTEXT,*PPER_IO_CONTEXT;typedef struct _PER_SOCKET_CONTEXT
{SOCKET m_Socket;SOCKADDR_IN m_ClientAddr;std::dequePER_IO_CONTEXT * m_PerIoContextArr;_PER_SOCKET_CONTEXT(){m_Socket INVALID_SOCKET;memset(m_ClientAddr, 0, sizeof(m_ClientAddr));}~_PER_SOCKET_CONTEXT(){if (m_Socket ! INVALID_SOCKET){closesocket(m_Socket);m_Socket INVALID_SOCKET;}for (int i 0; i m_PerIoContextArr.size();i){delete m_PerIoContextArr.at(i);}m_PerIoContextArr.clear();}_PER_IO_CONTEXT *GetNewIOContext(){_PER_IO_CONTEXT *p new _PER_IO_CONTEXT;m_PerIoContextArr.push_back(p);return p;}void RemoveIOContext(_PER_IO_CONTEXT *pIoContext){assert(pIoContext ! NULL);std::dequePER_IO_CONTEXT *::iterator iter m_PerIoContextArr.begin();for (; iter ! m_PerIoContextArr.end(); iter){if (*iter pIoContext){iter m_PerIoContextArr.erase(iter);break;}}}}PER_SOCKET_CONTEXT,*PPER_SOCKET_CONTEXT;class CIOCPModel;typedef struct _tagThreadParameter_WORKER
{CIOCPModel *pIOCPModel;int dwThreadNo;
}THREADPARAMS_WORKER, *PTHREADPARAMS_WORKER;class CIOCPModel
{
public:CIOCPModel();~CIOCPModel();public:bool start();void stop();bool LoadSocketLib();void unLoadSocketLib();std::string GetLocalIP();void setPort(const int m_port);protected:bool _InitIOCPModel();bool _InitializeListenSocket();void _DeInitializeIOCPModel();bool _PostAccept(PER_IO_CONTEXT *PerContext);bool _PostRecv(PER_IO_CONTEXT *PerContext);bool _PostSend(PER_IO_CONTEXT *PerContext);bool _DoAccept(PER_SOCKET_CONTEXT *PerSocket, PER_IO_CONTEXT *PerContext);bool _DoRecv(PER_SOCKET_CONTEXT *PerSocket, PER_IO_CONTEXT *PerContext);void _AddToContextList(PER_SOCKET_CONTEXT *pSocketContext);void _RemoveContextList(PER_SOCKET_CONTEXT *pSocketContext);void _ClearContextList();bool _AssociateWithIOCP(PER_SOCKET_CONTEXT *pSocketContext);bool HandleError(PER_SOCKET_CONTEXT *pSocketContext, const DWORD dwError);static DWORD __stdcall _WorkThread(LPVOID lpParam);int _GetNoOfProcessor();bool _IsSocketAlive(SOCKET s);
private:HANDLE m_hShutDownEvent;HANDLE m_hIOCompletionPort;HANDLE *m_WorkThread;int m_nThreads;std::string m_strIP;int m_port;CRITICAL_SECTION m_cs;std::dequePER_SOCKET_CONTEXT * m_ClientContextDeque;PER_SOCKET_CONTEXT *pListenSocketContext;LPFN_ACCEPTEX m_lpfnAcceptEx;LPFN_GETACCEPTEXSOCKADDRS m_lpfnGetAcceptExSockAddrs;};#endif
