首页 > 代码库 > winsock编程IOCP模型实现代码
winsock编程IOCP模型实现代码
winsock编程IOCP模型实现代码
话不多说，上代码。借鉴《windows核心编程》部分源码和CSDN小猪部分代码。
stdafx.h依赖头文件：
1 #include <iostream> 2 #include <WinSock2.h> 3 #include <MSWSock.h> 4 #include <vector> 5 #include "Singleton.h" 6 #include "IOCPWrapper.h" 7 #include "OverlappedIOInfo.h" 8 #include "TaskSvc.h" 9 10 using namespace std;
其中，TaskSvc.h、Singleton.h源码可以在我的blog里面找到。
IOCPWrapper.h源码：
1 /****************************************************************************** 2 Module: IOCP.h 3 Notices: Copyright (c) 2007 Jeffrey Richter & Christophe Nasarre 4 Purpose: This class wraps an I/O Completion Port. 5 Revise: IOCP封装类，由《windows核心编程》第10章示例程序源码改编所得 6 ******************************************************************************/ 7 #pragma once 8 9 class CIOCP 10 { 11 public: 12 CIOCP(int nMaxConcurrency = -1) 13 { 14 m_hIOCP = NULL; 15 if (nMaxConcurrency != -1) 16 Create(nMaxConcurrency); 17 } 18 ~CIOCP() 19 { 20 if (m_hIOCP != NULL) 21 VERIFY(CloseHandle(m_hIOCP)); 22 } 23 24 //关闭IOCP 25 BOOL Close() 26 { 27 BOOL bResult = CloseHandle(m_hIOCP); 28 m_hIOCP = NULL; 29 return(bResult); 30 } 31 32 //创建IOCP，nMaxConcurrency指定最大线程并发数量，0默认为cpu数量 33 BOOL Create(int nMaxConcurrency = 0) 34 { 35 m_hIOCP = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, nMaxConcurrency); 36 ASSERT(m_hIOCP != NULL); 37 return(m_hIOCP != NULL); 38 } 39 40 //为设备（文件、socket、邮件槽、管道等）关联一个IOCP 41 BOOL AssociateDevice(HANDLE hDevice, ULONG_PTR CompKey) 42 { 43 BOOL fOk = (CreateIoCompletionPort(hDevice, m_hIOCP, CompKey, 0) == m_hIOCP); 44 ASSERT(fOk); 45 return(fOk); 46 } 47 48 //为socket关联一个IOCP 49 BOOL AssociateSocket(SOCKET hSocket, ULONG_PTR CompKey) 50 { 51 return(AssociateDevice((HANDLE) hSocket, CompKey)); 52 } 53 54 //为iocp传递事件通知 55 BOOL PostStatus(ULONG_PTR CompKey, DWORD dwNumBytes = 0,OVERLAPPED* po = NULL) 56 { 57 BOOL fOk = PostQueuedCompletionStatus(m_hIOCP, dwNumBytes, CompKey, po); 58 ASSERT(fOk); 59 return(fOk); 60 } 61 62 //从IO完成队列中获取事件通知。IO完成队列无事件时，该函数将阻塞 63 BOOL GetStatus(ULONG_PTR* pCompKey, PDWORD pdwNumBytes,OVERLAPPED** ppo, DWORD dwMilliseconds = INFINITE) 64 { 65 return(GetQueuedCompletionStatus(m_hIOCP, pdwNumBytes,pCompKey, ppo, dwMilliseconds)); 66 } 67 68 //获取IOCP对象 69 const HANDLE GetIOCP() 70 { 71 return m_hIOCP; 72 } 73 private: 74 //IOCP句柄 75 HANDLE m_hIOCP; 76 }; 77 78 ///////////////////////////////// End of File /////////////////////////////////
OverlappedIOInfo.h源码
1 /****************************************************************************** 2 Module: OverlappedIOInfo.h 3 Notices: Copyright (c) 20161201 whg 4 Purpose: 5 IOCP网络编程模型中，需要用到GetQueuedCompletionStatus函数获取已完成事件。 6 但该函数的返回参数无socket或buffer的描述信息。 7 8 一个简单的解决办法，创建一个新的结构，该结构第一个参数是OVERLAPPED。 9 由于AcceptEx、WSASend等重叠IO操作传入的是Overlapped结构体的地址，调用AcceptEx等重叠IO操作， 10 在Overlapped结构体后面开辟新的空间，写入socket或buffer的信息，即可将socket或buffer的信息由 11 GetQueuedCompletionStatus带回。 12 13 参考《windows核心编程》和CSDN PiggyXP 14 ******************************************************************************/ 15 16 #pragma once 17 18 #define MAXBUF 8*1024 19 20 enum IOOperType{ 21 TYPE_ACP, //accept事件到达，有新连接请求 22 TYPE_RECV, //数据接收事件 23 TYPE_SEND, //数据发送事件 24 TYPE_CLOSE, //关闭事件 25 TYPE_NO_OPER 26 }; 27 28 class COverlappedIOInfo : public OVERLAPPED 29 { 30 public: 31 COverlappedIOInfo(void) 32 { 33 m_sSock = INVALID_SOCKET; 34 ResetOverlapped(); 35 ResetRecvBuffer(); 36 ResetSendBuffer(); 37 } 38 ~COverlappedIOInfo(void) 39 { 40 if (m_sSock != INVALID_SOCKET) 41 { 42 closesocket(m_sSock); 43 m_sSock = INVALID_SOCKET; 44 } 45 } 46 void ResetOverlapped() 47 { 48 Internal = InternalHigh = 0; 49 Offset = OffsetHigh = 0; 50 hEvent = NULL; 51 } 52 void ResetRecvBuffer() 53 { 54 ZeroMemory(m_cRecvBuf,MAXBUF); 55 m_recvBuf.buf = m_cRecvBuf; 56 m_recvBuf.len = MAXBUF; 57 } 58 void ResetSendBuffer() 59 { 60 ZeroMemory(m_cSendBuf,MAXBUF); 61 m_sendBuf.buf = m_cSendBuf; 62 m_sendBuf.len = MAXBUF; 63 } 64 public: 65 //套接字 66 SOCKET m_sSock; 67 //接收缓冲区，用于AcceptEx、WSARecv操作 68 WSABUF m_recvBuf; 69 char m_cRecvBuf[MAXBUF]; 70 //发送缓冲区，用于WSASend操作 71 WSABUF m_sendBuf; 72 char m_cSendBuf[MAXBUF]; 73 //对端地址 74 sockaddr_in m_addr; 75 };
server.h
1 #pragma once 2 3 4 class CServer:public CTaskSvc 5 { 6 #define ACCEPT_SOCKET_NUM 10 7 8 public: 9 CServer(void); 10 ~CServer(void); 11 bool StartListen(unsigned short port,std::string ip); 12 13 protected: 14 virtual void svc(); 15 16 private: 17 //启动CPU*2个线程，返回已启动线程个数 18 UINT StartThreadPull(); 19 //获取AcceptEx和GetAcceptExSockaddrs函数指针 20 bool GetLPFNAcceptEXAndGetAcceptSockAddrs(); 21 //利用AcceptEx监听accept请求 22 bool PostAccept(COverlappedIOInfo* ol); 23 //处理accept请求,NumberOfBytes=0表示没有收到第一帧数据，>0表示收到第一帧数据 24 bool DoAccept(COverlappedIOInfo* ol,DWORD NumberOfBytes=0); 25 //投递recv请求 26 bool PostRecv(COverlappedIOInfo* ol); 27 //处理recv请求 28 bool DoRecv(COverlappedIOInfo* ol); 29 //从已连接socket列表中移除socket及释放空间 30 bool DeleteLink(SOCKET s); 31 //释放3个部分步骤： 32 //1：清空IOCP线程队列，退出线程 33 //2: 清空等待accept的套接字m_vecAcps 34 //3: 清空已连接的套接字m_vecContInfo并清空缓存 35 void CloseServer(); 36 private: 37 //winsock版本类型 38 WSAData m_wsaData; 39 //端口监听套接字 40 SOCKET m_sListen; 41 //等待accept的套接字，这些套接字是没有使用过的，数量为ACCEPT_SOCKET_NUM。同时会有10个套接字等待accept 42 std::vector<SOCKET> m_vecAcps; 43 //已建立连接的信息，每个结构含有一个套接字、发送缓冲和接收缓冲，以及对端地址 44 std::vector<COverlappedIOInfo*> m_vecContInfo; 45 //操作vector的互斥访问锁 46 CThreadLockCs m_lsc; 47 //IOCP封装类 48 CIOCP m_iocp; 49 //AcceptEx函数指针 50 LPFN_ACCEPTEX m_lpfnAcceptEx; 51 //GetAcceptSockAddrs函数指针 52 LPFN_GETACCEPTEXSOCKADDRS m_lpfnGetAcceptSockAddrs; 53 }; 54 55 typedef CSingleton<CServer> SERVER;
server.cpp
1 #include "StdAfx.h" 2 #include "Server.h" 3 4 CServer::CServer(void) 5 { 6 m_lpfnAcceptEx = NULL; 7 m_lpfnGetAcceptSockAddrs = NULL; 8 WSAStartup(MAKEWORD(2,2),&m_wsaData); 9 } 10 11 CServer::~CServer(void) 12 { 13 CloseServer(); 14 WSACleanup(); 15 } 16 17 bool CServer::StartListen(unsigned short port,std::string ip) 18 { 19 //listen socket需要将accept操作投递到完成端口，因此，listen socket属性必须有重叠IO 20 m_sListen = WSASocket(AF_INET,SOCK_STREAM,IPPROTO_TCP,NULL,0,WSA_FLAG_OVERLAPPED); 21 if(m_sListen == INVALID_SOCKET) 22 { 23 cout<<"WSASocket create socket error"<<endl; 24 return false; 25 } 26 //创建并设置IOCP并发线程数量 27 if (m_iocp.Create() == FALSE) 28 { 29 cout<<"IOCP create error,error code "<<WSAGetLastError()<<endl; 30 return false; 31 } 32 //将listen socket绑定至iocp 33 if (!m_iocp.AssociateSocket(m_sListen,TYPE_ACP)) 34 { 35 cout<<"iocp Associate listen Socket error"<<endl; 36 return false; 37 } 38 sockaddr_in service; 39 service.sin_family = AF_INET; 40 service.sin_port = htons(port); 41 if (ip.empty()) 42 { 43 service.sin_addr.s_addr = INADDR_ANY; 44 } 45 else 46 { 47 service.sin_addr.s_addr = inet_addr(ip.c_str()); 48 } 49 50 if (bind(m_sListen,(sockaddr*)&service,sizeof(service)) == SOCKET_ERROR) 51 { 52 cout<<"bind() error,error code "<<WSAGetLastError()<<endl; 53 return false; 54 } 55 cout<<"bind ok!"<<endl; 56 57 if (listen(m_sListen,SOMAXCONN) == SOCKET_ERROR) 58 { 59 cout<<"listen() error,error code "<<WSAGetLastError()<<endl; 60 return false; 61 } 62 cout<<"listen ok!"<<endl; 63 //启动工作者线程 64 int threadnum = StartThreadPull(); 65 cout<<"启动工作者线程,num="<<threadnum<<endl; 66 //获取AcceptEx和GetAcceptSockAddrs函数指针 67 if (!GetLPFNAcceptEXAndGetAcceptSockAddrs()) 68 { 69 return false; 70 } 71 //创建10个acceptex 72 for (int i=0;i<ACCEPT_SOCKET_NUM;i++) 73 { 74 //用accept 75 COverlappedIOInfo* ol = new COverlappedIOInfo; 76 if (!PostAccept(ol)) 77 { 78 delete ol; 79 return false; 80 } 81 } 82 } 83 84 bool CServer::GetLPFNAcceptEXAndGetAcceptSockAddrs() 85 { 86 DWORD BytesReturned = 0; 87 //获取AcceptEx函数指针 88 GUID GuidAcceptEx = WSAID_ACCEPTEX; 89 if (SOCKET_ERROR == WSAIoctl( 90 m_sListen, 91 SIO_GET_EXTENSION_FUNCTION_POINTER, 92 &GuidAcceptEx, 93 sizeof(GuidAcceptEx), 94 &m_lpfnAcceptEx, 95 sizeof(m_lpfnAcceptEx), 96 &BytesReturned, 97 NULL,NULL)) 98 { 99 cout<<"WSAIoctl get AcceptEx function error,error code "<<WSAGetLastError()<<endl; 100 return false; 101 } 102 103 //获取GetAcceptexSockAddrs函数指针 104 GUID GuidGetAcceptexSockAddrs = WSAID_GETACCEPTEXSOCKADDRS; 105 if (SOCKET_ERROR == WSAIoctl( 106 m_sListen, 107 SIO_GET_EXTENSION_FUNCTION_POINTER, 108 &GuidGetAcceptexSockAddrs, 109 sizeof(GuidGetAcceptexSockAddrs), 110 &m_lpfnGetAcceptSockAddrs, 111 sizeof(m_lpfnGetAcceptSockAddrs), 112 &BytesReturned, 113 NULL,NULL)) 114 { 115 cout<<"WSAIoctl get GetAcceptexSockAddrs function error,error code "<<WSAGetLastError()<<endl; 116 return false; 117 } 118 return true; 119 } 120 121 bool CServer::PostAccept(COverlappedIOInfo* ol) 122 { 123 if (m_lpfnAcceptEx == NULL) 124 { 125 cout << "m_lpfnAcceptEx is NULL"<<endl; 126 return false; 127 } 128 SOCKET s = ol->m_sSock; 129 ol->ResetRecvBuffer(); 130 ol->ResetOverlapped(); 131 ol->ResetSendBuffer(); 132 ol->m_sSock = WSASocket(AF_INET,SOCK_STREAM,IPPROTO_TCP,NULL,0,WSA_FLAG_OVERLAPPED); 133 if (ol->m_sSock == INVALID_SOCKET) 134 { 135 cout<<"WSASocket error ,error code "<<WSAGetLastError()<<endl; 136 return false; 137 } 138 //这里建立的socket用来和对端建立连接，终会加入m_vecContInfo列表 139 //调用acceptex将accept socket绑定至完成端口，并开始进行事件监听 140 //这里需要传递Overlapped，new一个COverlappedIOInfo 141 //AcceptEx是m_listen的监听事件，m_listen已经绑定了完成端口；虽然ol->m_sSock已经创建， 142 //但未使用，现在不必为ol->m_sSock绑定完成端口。在AcceptEx事件发生后，再为ol->m_sSock绑定IOCP 143 DWORD byteReceived = 0; 144 if (FALSE == m_lpfnAcceptEx( 145 m_sListen, 146 ol->m_sSock, 147 ol->m_recvBuf.buf, 148 ol->m_recvBuf.len - (sizeof(SOCKADDR_IN)+16)*2, 149 sizeof(SOCKADDR_IN)+16, 150 sizeof(SOCKADDR_IN)+16, 151 &byteReceived, 152 ol)) 153 { 154 DWORD res = WSAGetLastError(); 155 if (ERROR_IO_PENDING != res) 156 { 157 cout<<"AcceptEx error , error code "<<res<<endl; 158 return false; 159 } 160 } 161 std::vector<SOCKET>::iterator iter = m_vecAcps.begin(); 162 for (;iter != m_vecAcps.end(); iter++) 163 { 164 if (*iter == s) 165 { 166 *iter = ol->m_sSock; 167 } 168 } 169 if (iter == m_vecAcps.end()) 170 { 171 m_vecAcps.push_back(ol->m_sSock); 172 } 173 return true; 174 } 175 176 bool CServer::DoAccept(COverlappedIOInfo* ol,DWORD NumberOfBytes) 177 { 178 //分支用于获取远端地址。 179 //如果接收TYPE_ACP同时收到第一帧数据，则第一帧数据内包含远端地址。 180 //如果没有收到第一帧数据，则通过getpeername获取远端地址 181 if (NumberOfBytes > 0) 182 { 183 //接受的数据分成3部分，第1部分是客户端发来的数据，第2部分是本地地址，第3部分是远端地址。 184 if (m_lpfnGetAcceptSockAddrs) 185 { 186 SOCKADDR_IN* ClientAddr = NULL; 187 SOCKADDR_IN* LocalAddr = NULL; 188 int remoteLen = sizeof(SOCKADDR_IN), localLen = sizeof(SOCKADDR_IN); 189 m_lpfnGetAcceptSockAddrs( 190 ol->m_recvBuf.buf, 191 ol->m_recvBuf.len - (sizeof(SOCKADDR_IN)+16)*2, 192 sizeof(SOCKADDR_IN)+16, 193 sizeof(SOCKADDR_IN)+16, 194 (LPSOCKADDR*)&LocalAddr, 195 &localLen, 196 (LPSOCKADDR*)&ClientAddr, 197 &remoteLen); 198 cout<<"收到新的连接请求,ip="<<inet_ntoa(ClientAddr->sin_addr)<<",port="<<ClientAddr->sin_port<<endl; 199 COverlappedIOInfo* pol = new COverlappedIOInfo; 200 pol->m_sSock = ol->m_sSock; 201 pol->m_addr = *ClientAddr; 202 //服务端只收取recv，同时监听recv和send可用设计位偏移，用或运算实现 203 if (m_iocp.AssociateSocket(pol->m_sSock,TYPE_RECV)) 204 { 205 PostRecv(pol); 206 207 m_vecContInfo.push_back(pol); 208 } 209 else 210 { 211 delete pol; 212 return false; 213 } 214 } 215 } 216 else if (NumberOfBytes==0) 217 { 218 //未收到第一帧数据 219 } 220 return true; 221 } 222 223 bool CServer::DoRecv(COverlappedIOInfo* ol) 224 { 225 cout<<"收到客户端数据：ip="<<inet_ntoa(ol->m_addr.sin_addr)<<",port="<<ol->m_addr.sin_port<< 226 "；内容="<<ol->m_recvBuf.buf<<endl; 227 return true; 228 } 229 230 bool CServer::PostRecv(COverlappedIOInfo* ol) 231 { 232 DWORD BytesRecvd = 0; 233 DWORD dwFlags = 0; 234 ol->ResetOverlapped(); 235 ol->ResetRecvBuffer(); 236 int recvnum = WSARecv(ol->m_sSock,&ol->m_recvBuf,1,&BytesRecvd,&dwFlags,(OVERLAPPED*)ol,NULL); 237 if (recvnum != 0) 238 { 239 int res = WSAGetLastError(); 240 if (WSA_IO_PENDING != res) 241 { 242 cout<<"WSARecv error,error code "<<res<<endl; 243 } 244 } 245 return true; 246 } 247 248 UINT CServer::StartThreadPull() 249 { 250 //获取系统cpu个数启动线程 251 SYSTEM_INFO si; 252 GetSystemInfo(&si); 253 //启动cpu数量*2个线程 254 return Activate(si.dwNumberOfProcessors * 2); 255 } 256 257 bool CServer::DeleteLink(SOCKET s) 258 { 259 m_lsc.lock(); 260 std::vector<COverlappedIOInfo*>::iterator iter = m_vecContInfo.begin(); 261 for (;iter!=m_vecContInfo.end();iter++) 262 { 263 if (s == (*iter)->m_sSock) 264 { 265 COverlappedIOInfo* ol = *iter; 266 closesocket(s); 267 m_vecContInfo.erase(iter); 268 delete ol; 269 break; 270 } 271 } 272 m_lsc.unlock(); 273 return true; 274 } 275 276 void CServer::svc() 277 { 278 while (true) 279 { 280 DWORD NumberOfBytes = 0; 281 unsigned long CompletionKey = 0; 282 OVERLAPPED* ol = NULL; 283 if (FALSE != GetQueuedCompletionStatus(m_iocp.GetIOCP(),&NumberOfBytes,&CompletionKey,&ol,WSA_INFINITE)) 284 { 285 if (CompletionKey == TYPE_CLOSE) 286 { 287 break; 288 } 289 COverlappedIOInfo* olinfo = (COverlappedIOInfo*)ol; 290 if (CompletionKey == TYPE_ACP) 291 { 292 DoAccept(olinfo,NumberOfBytes); 293 PostAccept(olinfo); 294 } 295 else if(CompletionKey == TYPE_RECV) 296 { 297 if (NumberOfBytes == 0) 298 { 299 //客户端断开连接 300 cout<<"客户端断开连接,ip="<<inet_ntoa(olinfo->m_addr.sin_addr)<<",port="<<olinfo->m_addr.sin_port<<endl; 301 DeleteLink(olinfo->m_sSock); 302 continue; 303 } 304 DoRecv(olinfo); 305 PostRecv(olinfo); 306 } 307 } 308 else 309 { 310 cout<<"workthread GetQueuedCompletionStatus error,error code "<<WSAGetLastError()<<endl; 311 break; 312 } 313 } 314 cout<<"workthread stop"<<endl; 315 } 316 317 void CServer::CloseServer() 318 { 319 //1：清空IOCP线程队列，退出线程,有多少个线程发送多少个PostQueuedCompletionStatus信息 320 int threadnum = GetThreadsNum(); 321 for (int i=0;i<threadnum;i++) 322 { 323 if (FALSE == m_iocp.PostStatus(TYPE_CLOSE)) 324 { 325 cout<<"PostQueuedCompletionStatus error,error code "<<WSAGetLastError()<<endl; 326 } 327 } 328 //2:清空等待accept的套接字m_vecAcps 329 std::vector<SOCKET>::iterator iter = m_vecAcps.begin(); 330 for (;iter != m_vecAcps.end();iter++) 331 { 332 SOCKET s = *iter; 333 closesocket(s); 334 } 335 m_vecAcps.clear(); 336 //3:清空已连接的套接字m_vecContInfo并清空缓存 337 std::vector<COverlappedIOInfo*>::iterator iter2 = m_vecContInfo.begin(); 338 for (;iter2 != m_vecContInfo.end();iter2++) 339 { 340 COverlappedIOInfo* ol = *iter2; 341 closesocket(ol->m_sSock); 342 iter2 = m_vecContInfo.erase(iter2); 343 delete ol; 344 } 345 m_vecContInfo.clear(); 346 }
调用方法，控制台程序main函数内加入
1 SERVER::Instance()->StartListen(8828,"127.0.0.1"); 2 int outId; 3 cin>>outId; 4 if (outId == 0) 5 { 6 SERVER::Close(); 7 }
输入0结束服务程序。
测试结果1，接收数据
测试结果2，客户端断开连接
机子比较垃圾，开1000个线程占用20多的CPU。有条件的可以多测试一些。
winsock编程IOCP模型实现代码
声明：以上内容来自用户投稿及互联网公开渠道收集整理发布，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任，若内容有误或涉及侵权可进行投诉： 投诉/举报 工作人员会在5个工作日内联系你，一经查实，本站将立刻删除涉嫌侵权内容。