IOCP完成端口模型

//iocp.cpp
//VS2012
//主要参考windows网络编程第二版第五章及各个关于完成端口的博客解释
#include"iocp.h"

CIocp::CIocp(){
m_nWorkerThread = new HANDLE[16];
m_pkey = new COMPLETIONKEY;
}
void CIocp::main_process(){
if(0!=wrap_CreateNewIoCompletionPort(&m_hIOCompletionPort))
return;
int m_nThreads = 2 * wrap_GetProcessorsNumber();//处理器数目2倍的线程
for(int i=0;i<m_nThreads;i++){
/*int* ID = new int;
*ID = i;*/
m_nWorkerThread[i]=(HANDLE)_beginthreadex(NULL,0,_WorkerThreads,this,0,0);
}
if(wrap_WSAStartup()!=0)
return;
m_sockListen = WSASocket(AF_INET,SOCK_STREAM,0,NULL,0,WSA_FLAG_OVERLAPPED);
m_pkey->s = m_sockListen;
//绑定完成端口
CreateIoCompletionPort((HANDLE)m_sockListen,m_hIOCompletionPort,(ULONG_PTR)m_pkey,0);

//addrinfo获取本机地址信息并绑定
addrinfo *serverinfo,*p;
if(0!=wrap_getaddrinfo(NULL,PORT,&serverinfo))
return;
for(p=serverinfo;p!=NULL;p=p->ai_next)
if(SOCKET_ERROR!=bind(m_sockListen,serverinfo->ai_addr,serverinfo->ai_addrlen))
break;
else err_no("bind");
freeaddrinfo(serverinfo);
if(p==NULL) err_no("all bind");

listen(m_sockListen,5);
//这部分主要用于获得函数AcceptExnmm，避免直接调用AcceptEx时的WSAIoctl开销
GUID guidAcceptEx = WSAID_ACCEPTEX;
DWORD dwBytes = 0;
if(WSAIoctl(m_sockListen,SIO_GET_EXTENSION_FUNCTION_POINTER,
&guidAcceptEx,sizeof(guidAcceptEx),&m_lpfnAcceptEx,sizeof(m_lpfnAcceptEx),
&dwBytes,NULL,NULL)!=0){
err_no("wsaioctl");
}

//投递Accept,并且将标识此次投递的IO_OPERATION_DATA传入,取出时可凭此获得与该次投递有关的信息
int i=200;
while(i--){
IO_OPERATION_DATA* pcompletionkey = new IO_OPERATION_DATA;
ASY_accept(pcompletionkey);

}
//工作时间
Sleep(100000);
//对每个线程发送关闭信号
for(int i =0;i<m_nThreads;i++){
IO_OPERATION_DATA* pIOData = new IO_OPERATION_DATA;
pIOData->t_info=0x00;
PostQueuedCompletionStatus(m_hIOCompletionPort,-1,(ULONG_PTR)m_pkey,&pIOData->overlapped);
}
//线程数通常比较少，一个wait足够
WaitForMultipleObjects(m_nThreads,m_nWorkerThread,true,INFINITE);
closesocket(m_sockListen);
CloseHandle(m_hIOCompletionPort);
}

unsigned int __stdcall CIocp::_WorkerThreads(PVOID pm){
/*int* tmp = (int*)pm;
int ID = *tmp;
delete tmp;*/
CIocp* CI  =(CIocp*)pm;
PCOMPLETIONKEY  lpContext = NULL;

LPOVERLAPPED    pOverlapped = NULL;
DWORD       dwBytesTransfered = -1;
IO_OPERATION_DATA *iod = NULL;
//进入睡眠，等待请求完成时被唤醒，INFINITE将等待时间设定为无限长
for(;;){
BOOL bReturn  =  GetQueuedCompletionStatus(CI->GetIoCompletionPort(),&dwBytesTransfered,(PULONG_PTR)&lpContext,&pOverlapped,INFINITE); //key 用地址的指针
//取出的结构里包含了投递这个请求时传入的信息，利用t_info知道是哪一个请求已经完成了
iod = (IO_OPERATION_DATA*)pOverlapped;
//这里的标志都是表示 已完成 的操作
switch(iod->t_info){
case ACCEPT:{//已收到新连接
//printf("%d:accept complete\n",ID);
//接着投递接收请求
CI->ASY_recv(iod->s,iod);
}
case READ://已完成读请求
//          printf("%d:read complete\n",ID);
//          printf("%d:%s\n",iod->s,iod->buffer);
//          //投递一个回复
//          CI->ASY_send(iod->s,iod);
//          break;
//
//case WRITE://已完成写请求
//          printf("%d:write complete\n\n",ID);
closesocket(iod->s);
if(!WSACloseEvent(iod->overlapped.hEvent))
err_no("close event");
CI->ASY_accept(iod);
break;

default:
printf("break\n");
delete iod;
return 0;
break;
}
}
return 0;

}

void CIocp::ASY_send(SOCKET s,IO_OPERATION_DATA* p){
//重用已分配的内存，不要浪费
p->t_info = WRITE;
sprintf(p->buffer,"hallo");
p->dataBuf.buf = p->buffer;
p->dataBuf.len = sizeof(p->buffer);
//重新设置为未传信状态
WSAResetEvent(p->overlapped.hEvent);
DWORD flags=0;
DWORD size = sizeof(p->dataBuf);
//再次投递recv请求，并且将信息传入
//printf("send client:%d\n",p->s);
if(0!=WSASend(p->s,&p->dataBuf,1,&size,flags,&p->overlapped,NULL)){
int err;
if((err=WSAGetLastError())==WSA_IO_PENDING)
printf("send pending\n");
else printf("send:%d\n",err);
}
else
{
printf("sended\n\n");
closesocket(p->s);
ASY_accept(p);
}
}

void CIocp::ASY_recv(SOCKET s,IO_OPERATION_DATA* p){
//基本和send一样
p->t_info = READ;
p->dataBuf.buf = p->buffer;
p->dataBuf.len = sizeof(p->buffer);
if(!WSAResetEvent(p->overlapped.hEvent))//重设事件状态
err_no("reset event");
DWORD flags=0;
WSARecv(p->s
ebf8
,&p->dataBuf,1,NULL,&flags,&p->overlapped,NULL);
if(WSAGetLastError()!=WSA_IO_PENDING)
err_no("recv");
}

void CIocp::ASY_accept(IO_OPERATION_DATA* p){
p->s = WSASocket(AF_INET,SOCK_STREAM,0,NULL,0,WSA_FLAG_OVERLAPPED);
p->t_info = ACCEPT;
DWORD dwBytes = 0;
memset(&p->overlapped, 0, sizeof (p->overlapped));
//设置事件
p->overlapped.hEvent = WSACreateEvent();
if(p->overlapped.hEvent == WSA_INVALID_EVENT)
err_no("create event");
//通过指针调用AcceptEx
if(0==m_lpfnAcceptEx(m_sockListen,p->s,p->buffer,BUFFER_SIZE-(sizeof(SOCKADDR_IN)+16)*2,sizeof(SOCKADDR_IN)+16,sizeof(SOCKADDR_IN)+16,&dwBytes,&p->overlapped)){
if(WSAGetLastError()==WSA_IO_PENDING)
printf("pending……\n");
printf_time();
}else
printf("notpending***\n");
}

//iocp.h
#pragma once
#pragma warning(disable:4996)
#include"wrap_sock.h"
#include"err_no.h"
#include<iostream>
#include<process.h>
#include <mswsock.h>
#define PORT "3490"
//#define WSAID_ACCEPTEX \
//        {0xb5367df1,0xcbac,0x11cf,{0x95,0xca,0x00,0x80,0x5f,0x48,0xa1,0x92}}
#define OPERATION_TYPE int
#define WRITE 0x1
#define READ 0x2
#define ACCEPT 0x3
#define BUFFER_SIZE 512

//typedef BOOL (WINAPI *LPFN_ACCEPTEX)(SOCKET, SOCKET, PVOID, DWORD, DWORD, DWORD, LPDWORD, LPOVERLAPPED);

typedef struct _completionKey

{

SOCKET s;

SOCKADDR_IN clientAddr;

}COMPLETIONKEY,*PCOMPLETIONKEY;

typedef struct _io_operation_data

{

OVERLAPPED overlapped;

WSABUF dataBuf;

CHAR   buffer[BUFFER_SIZE];
int    t_info;
SOCKET s;
}IO_OPERATION_DATA;

class CIocp{
public:
CIocp();
void main_process();
void ASY_send(SOCKET s,IO_OPERATION_DATA* p);
void ASY_recv(SOCKET s,IO_OPERATION_DATA* p);
void ASY_accept(IO_OPERATION_DATA* pcompletionkey);
LPFN_ACCEPTEX GetAcceptFunc()const{return m_lpfnAcceptEx;}
HANDLE GetIoCompletionPort()const{return m_hIOCompletionPort;}
private:
PHANDLE m_nWorkerThread;
HANDLE m_hIOCompletionPort;
LPFN_ACCEPTEX m_lpfnAcceptEx;
PCOMPLETIONKEY m_pkey;
SOCKET m_sockListen;

static unsigned int __stdcall _WorkerThreads(PVOID pm);
};

//wrap_sock.h
//主要包裹了错误报告
#include<WinSock2.h>
#include<WS2tcpip.h>
#include"err_no.h"
#pragma comment(lib,"ws2_32.lib")

int wrap_CreateNewIoCompletionPort(HANDLE* iocp);

int wrap_GetProcessorsNumber();

int wrap_WSAStartup();

//set hints and get info
int wrap_getaddrinfo(const char* servername,const char* port,PADDRINFOA* m_serverinfo);

int wrap_socket(SOCKET* sock,int af,int type,int protocol);

//if failed,print and closesocket
int wrap_connect(SOCKET serverfd,const sockaddr* addr,int namelen);

//if failed,print and closesocket
int wrap_recv(SOCKET serverfd,char* recvbuf,int bufsize,int flag);

//if failed,print and closesocket
int wrap_send(SOCKET serverfd,const char* recvbuf,int bufsize,int flag);

//wrap_sock.h
#include"wrap_sock.h"

int wrap_CreateNewIoCompletionPort(HANDLE* iocp){
*iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE,NULL,0,0);
if(*iocp==0){
err_no("create IOCP");
return -1;
}
else return 0;
}

int wrap_GetProcessorsNumber(){
SYSTEM_INFO si;
GetSystemInfo(&si);
return si.dwNumberOfProcessors;
}

int wrap_WSAStartup(){
WSAData m_wd;
if(0!=WSAStartup(MAKEWORD(1,1),&m_wd)){
err_no("WSAStartup");
return -1;
}
return 0;
}

int wrap_getaddrinfo(const char* servername,const char* port,PADDRINFOA* serverinfo){
addrinfo hints;
//set hints
memset(&hints,0,sizeof(addrinfo));
hints.ai_family = AF_INET;
hints.ai_socktype = SOCK_STREAM;
hints.ai_flags = AI_PASSIVE;
if(0!=getaddrinfo(servername,port,&hints,serverinfo)){
err_no("getaddrinfo failed");
return -1;
}
return 0;
}

int wrap_recv(SOCKET serverfd,char* recvbuf,int bufsize,int flag){
if(-1==recv(serverfd,recvbuf,bufsize,flag)){
//printf("no:%d\n",tID);
err_no("recv failed");
closesocket(serverfd);
return -1;
}
return 0;
}

int wrap_send(SOCKET serverfd,const char* recvbuf,int bufsize,int flag){
if(-1==send(serverfd,recvbuf,bufsize,flag)){
//printf("no:%d\n",tID);
err_no("send failed");
closesocket(serverfd);
return -1;
}
return 0;
}

int wrap_connect(SOCKET serverfd,const sockaddr* addr,int namelen){
if(0!=connect(serverfd,addr,namelen)){
err_no("connect failed");
closesocket(serverfd);
return -1;
}
return 0;
}
int wrap_socket(SOCKET* sock,int af,int type,int protocol){
SOCKET tmp;
if((tmp = socket(af,type,protocol))==-1){
err_no("socket failed");
return -1;
}
*sock = tmp;
return 0;
}