大家好，又见面了，我是你们的朋友全栈君。
最基础的IOCP例子, 没有使用扩展函数AcceptEx: IOCP模型
* 关于iocp的核心就一点:
GetQueuedCompletionStatus 将携带返回2个重要的参数, 一个lpCompletionKey, 一个lpOverlapped.
lpCompletionKey : 是 CreateIoCompletionPort((HANDLE)clientSocket , hIOCP,(ULONG_PTR)自定义的结构,0); 跟
iocp绑定的一个自定义参数;
lpOverlapped : 是传递给 WSASend / WSARecv 的参数;
这2个参数最终会被GetQueuedCompletionStatus 携带回来.
同样的 , AcceptEx 也要传递一个Overlapped结构,现在问题来了,如果只调用了AcceptEx ,
GetQueuedCompletionStatus 是不会返回的, 因为只有跟 iocp 关联（CreateIoCompletionPort）的HANDLE / SOCKET 才会
被触发, 因此只需要把 监听套接字 跟iocp 关联即可;
下面代码使用了AccpetEx 和一个用于获取地址的扩展函数[此函数可以先忽略].
总体来说就是预先分配一些socket , 以及相关的内存块[到时有客户进来后,直接使用此内存块接受数据];
不再让accept系统调用来创建socket了.
所有需要注意的点都写在注释里了.
下面代码里没有使用 CancelIo 之类的函数,如果实际需要直接用 CancelIoEx 来取消无关线程的Overlapped操作,
另:在发送数据[WSASend] 完成后 , 需要检查是否发送完成, 如果没有发完需要继续发送.
代码语言：javascript复制#include <mswsock.h>
#include <deque>
#include <vector>
#define IO_ACCEPT 1
#define IO_READ 2
#define IO_WRITE 3
#define BUFFSIZE 4096
#define SPINCOUNT 4000
struct Per_IO_Data;
std::deque<Per_IO_Data*> io_pool;
CRITICAL_SECTION io_pool_cs;
LPFN_ACCEPTEX FuncAcceptEx = NULL;
LPFN_GETACCEPTEXSOCKADDRS FuncGetAddr = NULL;
struct Per_Sock_Data{
SOCKET sock;
SOCKADDR_IN addr;
HANDLE iocp;
Per_Sock_Data():sock(INVALID_SOCKET), iocp(INVALID_HANDLE_VALUE){}
};
struct Per_IO_Data{
OVERLAPPED ol;
WSABUF wsabuf;
char *buf;
int ioMode; //读 写 接受
SOCKET sAcceptSock; //acceptex 预先创建的socket
int nTotalBytes;
int nSendBytes;
Per_IO_Data(): buf(NULL) ,ioMode(-1) , sAcceptSock(INVALID_SOCKET)
,nSendBytes(0), nTotalBytes(0)
{
memset(&ol, 0 , sizeof(ol));
}
~Per_IO_Data(){
if(buf) delete buf;
}
};
void resetPerIOData(Per_IO_Data * pdata , int mode = -1){
if(!pdata) return;
memset(&pdata->ol, 0 ,sizeof(OVERLAPPED));
if(pdata->buf)
memset(&pdata->buf,0,sizeof (char) * BUFFSIZE);
pdata->wsabuf.buf=pdata->buf;
pdata->ioMode = mode;
pdata->wsabuf.len = BUFFSIZE;
pdata->nSendBytes = 0 ;
pdata->nTotalBytes = 0 ;
pdata->sAcceptSock = INVALID_SOCKET;
}
Per_IO_Data * getNewPerIOData(int mode = -1){
Per_IO_Data * pdata = new Per_IO_Data;
if(!pdata) return NULL;
pdata->buf = new char[BUFFSIZE];
pdata->ioMode = mode;
pdata->wsabuf.buf = pdata->buf;
pdata->wsabuf.len = BUFFSIZE;
return pdata;
}
bool doAccept(SOCKET listenfd){
if(listenfd == INVALID_SOCKET) return false;
SOCKET sock = WSASocket(AF_INET,SOCK_STREAM,0,0,0,WSA_FLAG_OVERLAPPED);
if(sock == INVALID_SOCKET)
return false;
Per_IO_Data *pData = getNewPerIOData(IO_ACCEPT);
if(!pData) { closesocket(sock); return false;}
pData->sAcceptSock = sock;
DWORD nBytes = 0;
BOOL ret = FuncAcceptEx(listenfd,
pData->sAcceptSock,
pData->buf,
0, //唯一这个参数注意一下, 自己找msdn
sizeof(SOCKADDR_IN) 16,
sizeof(SOCKADDR_IN) 16,
&nBytes,
&pData->ol);
if(!ret && WSAGetLastError() != ERROR_IO_PENDING){
closesocket(sock);
EnterCriticalSection(&io_pool_cs);
io_pool.push_back(pData);
LeaveCriticalSection(&io_pool_cs);
return false;
}
return true;
}
void handleIO( Per_IO_Data * pData, Per_Sock_Data * pSock , DWORD nBytesTransfered){
if(!pData || !pSock) return;
DWORD flag = 0;
if(IO_READ == pData->ioMode ){
memset(&pData->ol, 0 , sizeof(OVERLAPPED)); //清一下OVERLAPPED,虽然对socket没啥用,还是习惯问题
pData->ioMode = IO_WRITE; //把这块数据直接发出去
pData->wsabuf.len = nBytesTransfered;
pData->nSendBytes = 0;
pData->nTotalBytes = nBytesTransfered;
WSASend(pSock->sock,&pData->wsabuf,1,NULL,flag,&pData->ol,NULL); //复制到发送缓冲区
//每次创建Per_IO_Data 太浪费了, 因此把之前用过的内存块存在全局deque中
//如果太多了deque中太多了 自己想办法解决去删除一些,方法很多.
//如果觉得麻烦,直接 new Per_IO_Data即可
Per_IO_Data * pNewIOData = NULL;
EnterCriticalSection(&io_pool_cs);
if(!io_pool.empty()){
pNewIOData = io_pool.front();
io_pool.pop_front();
}
LeaveCriticalSection(&io_pool_cs);
if(NULL == pNewIOData){
pNewIOData = getNewPerIOData(IO_READ);
}
else{
pNewIOData->ioMode = IO_READ;
}
//继续接受
flag = 0;
WSARecv(pSock->sock , &pData->wsabuf,1,NULL,&flag,&pData->ol,NULL);
}
else if(IO_WRITE == pData->ioMode){
// 检查是否发送完毕, 如果没有发送完毕 , 继续发送剩余的数据,
// 如果发完了就进入 io_pool
pData->ioMode = IO_WRITE;
pData->nSendBytes = nBytesTransfered;
if(pData->nSendBytes < pData->nTotalBytes){
flag = 0;
pData->wsabuf.len = pData->nTotalBytes - pData->nSendBytes;
pData->wsabuf.buf = pData->buf pData->nSendBytes;
WSASend(pSock->sock,&pData->wsabuf,1,NULL,flag,&pData->ol,NULL); //复制到发送缓冲区
}
else{
//复制到发送缓存区后, 将回到这里,此时为了节省开销, 把pData重置一下仍进io_pool中
resetPerIOData(pData,IO_READ);
EnterCriticalSection(&io_pool_cs);
io_pool.push_back(pData);
LeaveCriticalSection(&io_pool_cs);
}
}
else if(IO_ACCEPT == pData->ioMode){ //核心点
//会到这里说明有连接进来了 , 由于listenfd 关联到了IOCP,
//因此对应的 key 也将一起传递进来 ,也就是这个pSock, pSock->sock就是监听套接字
//而这个pData就是在doAccept 中创建的那个内存块,内部携带了预先创建的sAcceptSocket
//但是这个sAcceptSocket 并没有继承监听socket的属性
//因此继承属性,比如缓冲区大小
setsockopt(pData->sAcceptSock, SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT,
(char *)&pSock->sock, sizeof(pSock->sock) );
//为此新的socket创建一个key , 用于之后的操作
Per_Sock_Data * pNewSock = new Per_Sock_Data;
pNewSock->sock = pData->sAcceptSock;
//用扩展函数来获取客户端的地址 GetAcceptExSockaddrs
//里面很多参数都是和 AcceptEx中相同的
SOCKADDR_IN * localAddr = NULL, *remoteAddr = NULL;
int localLen = 0 , remoteLen = 0;
FuncGetAddr(
pData->buf,
0, //此参数注意一下,翻一下msdn , 与AcceptEx那个要注意的参数相同
sizeof(SOCKADDR_IN) 16,
sizeof(SOCKADDR_IN) 16,
(SOCKADDR **)&localAddr,
&localLen,
(SOCKADDR **)&remoteAddr,
&remoteLen
);
//把客户端的地址复制进去
memcpy(&pNewSock->addr,remoteAddr,remoteLen);
//把新的key与iocp关联;
//为了方便,我把iocp放到了与监听套接字关联的结构体中, 可以直接做为全局变量更方便
CreateIoCompletionPort((HANDLE)pNewSock->sock,pSock->iocp,(ULONG_PTR)pNewSock,0);
//重置一下,把ioMode修改一下
resetPerIOData(pData,IO_READ);
pData->sAcceptSock = INVALID_SOCKET; //防止歧义
//开始接受数据
flag = 0;
WSARecv(pNewSock->sock,&pData->wsabuf,1,NULL,&flag,&pData->ol,NULL);
//继续接受一个连接,我在一开始的时候只调用了一次,因此只能接受一个连接
doAccept(pSock->sock);
}
else{
cout << "???????????? 走错地方了把" << endl;
}
}
unsigned iocp_thread(void *arg){
HANDLE iocp = *(HANDLE*)arg;
Per_IO_Data * pData = NULL;
Per_Sock_Data *pSock = NULL;
DWORD nBytesTrans = 0;
BOOL ret = FALSE;
while(1){
pSock = NULL;
pData = NULL;
ret = GetQueuedCompletionStatus(iocp,&nBytesTrans,(PULONG_PTR)&pSock,
(LPOVERLAPPED *)&pData,INFINITE);
//自己做 ret ==FALSE 的错误处理
if(0 == nBytesTrans && ( IO_READ == pData->ioMode || IO_WRITE == pData->ioMode)){
cout << "peer closed :" << inet_ntoa(pSock->addr.sin_addr) << endl;
closesocket(pSock->sock);
delete pSock;
delete pData;
}
else
handleIO(pData,pSock, nBytesTrans);
}
return 0;
}
int main(int argc, char *argv[])
{
//起手式都一样
WSADATA wsadata;
WSAStartup(MAKEWORD(2,2) , &wsadata);
SOCKET listenfd = socket(AF_INET, SOCK_STREAM, 0); //自带overlapped属性 == wsasock(... , overlapped)
int openReuse = 1;
int openReuseLen = sizeof(openReuse);
setsockopt(listenfd,SOL_SOCKET,SO_REUSEADDR,(char*)&openReuse,openReuseLen);
struct sockaddr_in serv_addr;
serv_addr.sin_family = AF_INET;
serv_addr.sin_port = htons(PORT);
serv_addr.sin_addr.S_un.S_addr = INADDR_ANY;
bind(listenfd,(SOCKADDR*)&serv_addr,sizeof(struct sockaddr_in));
listen(listenfd, SOMAXCONN);
GUID GuidAcceptEx = WSAID_ACCEPTEX;
GUID GuidGetAddr = WSAID_GETACCEPTEXSOCKADDRS;
DWORD dwbytes = 0;
//2个扩展函数, 没什么好说的, 固定模式, 看一边msdn,照抄即可
WSAIoctl(listenfd,SIO_GET_EXTENSION_FUNCTION_POINTER,
&GuidAcceptEx,sizeof(GuidAcceptEx),
&FuncAcceptEx,sizeof(FuncAcceptEx),
&dwbytes,0,0);
WSAIoctl(listenfd,SIO_GET_EXTENSION_FUNCTION_POINTER,
&GuidGetAddr,sizeof(GuidGetAddr),
&FuncGetAddr,sizeof(FuncGetAddr),
&dwbytes,0,0);
InitializeCriticalSectionAndSpinCount(&io_pool_cs, SPINCOUNT);
SYSTEM_INFO sysinfo = {0};
GetSystemInfo(&sysinfo);
int thread_num = sysinfo.dwNumberOfProcessors * 2 2;//线程数
HANDLE iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE,NULL,0,thread_num);
//把监听sock 关联到iocp , 至此任何客户端的connect, 都由iocp来完成,就像其他socket一样
Per_Sock_Data * pSock = new Per_Sock_Data; //给监听套接字绑定一个key
pSock->iocp = iocp;
pSock->sock = listenfd;
memcpy(&pSock->addr,&serv_addr,sizeof(serv_addr));
CreateIoCompletionPort((HANDLE)listenfd, iocp,(ULONG_PTR)pSock,0);
//接下来起线程
vector<HANDLE> thread_handles;
vector<unsigned int> tids;
for(int i = 0; i < thread_num ; i){
unsigned tid = 0;
HANDLE t = (HANDLE)_beginthreadex(0,0,iocp_thread,(void*)&iocp,0 , &tid);
thread_handles.push_back(t);
tids.push_back( tid);
}
doAccept(listenfd); //只接受一个客户端 , for(int i = 0 ; i < 10000; i) doAccept ...
//接下来主线程没事了 , 该创建的都创建了,可以干其他事了
while(1){
Sleep(10000);
cout << "main 睡觉 : " << io_pool.size() << endl;
}
return 0;
}版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 举报，一经查实，本站将立刻删除。
发布者：全栈程序员栈长，转载请注明出处：https://javaforall.cn/187774.html原文链接：https://javaforall.cn
