IOCP网络模型

引言：犹豫好久，虽然网上有很多关于完成端口很棒的博文，但是不是自己的，心里难免有点膈应（不知道各位道友是否也如此），于是始终想写一篇来记录自己的学习收获，不论简单与复杂，算做对自己知识的检验。

目录

1、 相关信息

2、 API详解

3、 实现过程

4、 收获

1、 相关信息

本例源码是对Ghost的中IOCP模型的学习，但是也有不同的地方。在原有的基础上凭着粗略的技术将accept改成了AcceptEx，虽然过程曲折，自己动手的过程中的收获总比看得强，同时粗略的领略到Ghost作者的编程思想，感觉也是受益匪浅。希望各位共同进步，Ghost源码https://pan.baidu.com/s/1i4Yidz3

2、 API详解

（1） 完成端口实现的API

CreateIoCompletionPort

HANDLE WINAPI CreateIoCompletionPort(
_In_    HANDLE FileHandle,
_In_opt_  HANDLE ExistingCompletionPort,
_In_    ULONG_PTR CompletionKey,
_In_    DWORD NumberOfConcurrentThreads
);

返回值：如果函数成功，则返回值是I / O完成端口的句柄：如果函数失败，则返回值为NULL。

功能：两个功能，创建完成端口句柄与将新的文件句柄（套接字）绑定到完成端口（我们也可以理解为完成队列，只是这个队列由操作系统自己维护）

FileHandle：文件句柄或INVALID_HANDLE_VALUE。创建完成端口的时候，该值设置为INVALID_HANDLE_VALUE，Ghost里面时候的是一个临时的socket句柄，不过我们不用一定要这样。主要

ExistingCompletionPort：现有I / O完成端口的句柄或NULL。如果此参数为现有I / O完成端口，则该函数将其与FileHandle参数指定的句柄相关联。如果成功则函数返回现有I / O完成端口的句柄。如果此参数为NULL，则该函数将创建一个新的I / O完成端口，如果FileHandle参数有效，则将其与新的I / O完成端口相关联。否则，不会发生文件句柄关联。如果成功，该函数将把句柄返回给新的I / O完成端口。

CompletionKey：该值就是类似线程里面传递的一个参数，我们在GetQueuedCompletionStatus中第三个参数获得的就是这个值。

NumberOfConcurrentThreads：如果此参数为NULL，则系统允许与系统中的处理器一样多的并发运行的线程。如果ExistingCompletionPort参数不是NULL，则忽略此参数。

GetQueuedCompletionStatus

BOOL WINAPI GetQueuedCompletionStatus(
_In_   HANDLE CompletionPort,
_Out_  LPDWORD lpNumberOfBytes,
_Out_  PULONG_PTR lpCompletionKey,
_Out_  LPOVERLAPPED *lpOverlapped,
_In_   DWORD dwMilliseconds
);

返回值：成功返回TRUE，失败返回FALSE，如果设置了超时时间，超时返回FALSE

功能：从完成端口中获取已经完成的消息

CompletionPort：完成端口的句柄。

lpNumberOfBytes：该变量接收已完成的I / O操作期间传输的字节数。

lpCompletionKey：该变量及时我们 CreateIoCompletionPort中传递的第三个参数

lpOverlapped：接收完成的I / O操作启动时指定的OVERLAPPED结构的地址。我们可以通过CONTAINING_RECORD这个宏获取以该重叠结构为首地址的结构体信息，也就是该重叠结构为什么必须放在结构体的首地址的原因。

dwMilliseconds：超时时间（毫秒），如果为INFINITE则一直等待直到有消息到来。

PostQueuedCompletionStatus

BOOL WINAPI PostQueuedCompletionStatus(
_In_      HANDLE CompletionPort,
_In_      DWORD dwNumberOfBytesTransferred,
_In_      ULONG_PTR dwCompletionKey,
_In_opt_  LPOVERLAPPED lpOverlapped
);

返回值：成功，返回非零，失败返回零。使用GetLasrError获取最后的错误码

功能：手动向完成端口投递一个异步消息。就类似我们Win32中的PostMessage

CompletionPort：完成端口的句柄。

dwNumberOfBytesTransferred：通过GetQueuedCompletionStatus函数的lpNumberOfBytesTransferred参数返回的值。

dwCompletionKey：通过GetQueuedCompletionStatus函数的lpCompletionKey参数返回的值。

lpOverlapped：通过GetQueuedCompletionStatus函数的lpOverlapped参数返回的值。

CONTAINING_RECORD

PCHAR CONTAINING_RECORD(
[in]  PCHAR Address,
[in]  TYPE Type,
[in]  PCHAR Field
);

功能：返回给定结构类型的结构实例的基地址和包含结构中字段的地址。

返回值：返回包含Field的结构的基地址。

Address：我们通过GetQueuedCompletionStatus获取的重叠结构

Type：以重叠结构为首地址的结构体

Field：Type结构体的重叠结构变量

（2）相关其他函数

AcceptEx

BOOL AcceptEx(
_In_   SOCKET sListenSocket,
_In_   SOCKET sAcceptSocket,
_In_   PVOID lpOutputBuffer,
_In_   DWORD dwReceiveDataLength,
_In_   DWORD dwLocalAddressLength,
_In_   DWORD dwRemoteAddressLength,
_Out_  LPDWORD lpdwBytesReceived,
_In_   LPOVERLAPPED lpOverlapped
);

返回值：成功返回TRUE，失败返回FALSE

功能：投递异步的接收操作，类似于实现了一个网络内存池，这个池中存放的是已经创造好的套接字（由于要进行异步操作，所以该套接字也要使用WSASocket创建），当有用户连接的时候，操作系统会直接从这个网络内存池中拿出一个来给连接的客户端，这个过程我们少去了连接时才创造套接字的过程（创建一个套接字的过程内部是很复杂的），这也是这个函数优异的地方。

该函数的参数很明确，只是有些其余的话还需要提醒，AcceptEx该函数还需要通过函数指针获得，因为该函数不是windows自身的API。具体的获取过程也只是按部就班，MSDN有详细的例子，示例代码中也有详细的过程，笔者就不赘述了。

3、 实现过程

这个过程我就是用代码的流程图呈现给大家，大家在阅读示例代码的时候，也更好的理解

（1）启动完成端口的大概过程

创建工作者线程数目一般为核心数的两倍或者两倍+2，创建成功后，工作者线程就已经开始工作了。我们投递了n个（此处10个）异步接收的套接字，用于接收套接字的连接。



StartIOCP

bool CIOCPServer::StartIOCP(NOTIFYPROC pNotifyProc, const UIN
4000
T& nPort)
{
m_nPort = nPort;
m_pNotifyProc = pNotifyProc;
InitializeCriticalSection(&m_cs);

bool bRet = false;
do
{
if (NULL == (m_hShutDownEvent = CreateEvent(NULL, TRUE, FALSE, NULL)))
break;
if (!InitNetEnvironment())
break;
if (!InitializeIOCP())
break;
if (!InitializeListenSocket())
break;
bRet = true;
} while (FALSE);

if (!bRet)
{
TCHAR szErr[32];
_stprintf(szErr, _T("Error code:%d"), GetLastError());
::MessageBox(GetDesktopWindow(), szErr, L"启动服务器失败", MB_OK | MB_ICONHAND);
}

return bRet;
}

（2）工作者线程

通过从完成端口中取已经完成的消息，通过判断返回值是否有错误，是否为客户端断开连接，然后自然就是对数据进行相应的处理。怎么相应的处理？自然是我们用于接收的结构体中有一个标识操作类型的变量IOType，然后进行接收连接、接收消息、发送消息的相应操作，每个操作的流程往下看。



ThreadPoolFunc

unsigned __stdcall CIOCPServer::ThreadPoolFunc(LPVOID lpParam)
{
CIOCPServer* pThis = (CIOCPServer*)lpParam;
OVERLAPPED* pOverlapped = NULL;
DWORD dwIoSize = 0;
BOOL bRet = FALSE;
DWORD dwErr = 0;
IOCP_PARAM* pIocpParam = NULL;
PER_IO_CONTEXT* pIoContext = NULL;

// 循环处理，直到 退出事件 有信号到来，就进行退出
while (WAIT_OBJECT_0 != WaitForSingleObject(pThis->m_hShutDownEvent, 10))
{
bRet = GetQueuedCompletionStatus(//  从完成端口中获取消息
pThis->m_hIOCompletionPort,
&dwIoSize,
(PULONG_PTR)&pIocpParam,
&pOverlapped,
INFINITE);

if (EXIT_CODE == pIocpParam)
break;

pIoContext = CONTAINING_RECORD(pOverlapped, PER_IO_CONTEXT, m_ol);

if (!bRet)// 处理错误信息
{
dwErr = GetLastError();
if (WAIT_TIMEOUT == dwErr)// 超时
{
// 超时后，通过发送一个消息，判断是否断线，否则在socket上投递WSARecv会出错
// 因为如果客户端网络异常断开(例如客户端崩溃或者拔掉网线等)的时候，服务器端是无法收到客户端断开的通知的
if (-1 == send(pIocpParam->m_sock, "", 0, 0))
{
pThis->MoveToFreeParamPool(pIocpParam);
pThis->RemoveStaleClient(pIoContext,FALSE);
}
continue;
}
if(ERROR_NETNAME_DELETED == dwErr)// 客户端异常退出
{
pThis->MoveToFreeParamPool(pIocpParam);
pThis->RemoveStaleClient(pIoContext, FALSE);
continue;
}

break;// 完成端口出现错误
}

// 正式处理接收到的数据 读取接收到的数据
// CONTAINING_RECORD宏返回给定结构类型的结构实例的 基地址 和包含结构中字段的地址。
if (bRet && 0 == dwIoSize)
{
// 客户端断开连接，释放资源
pThis->MoveToFreeParamPool(pIocpParam);
pThis->RemoveStaleClient(pIoContext, FALSE);
continue;
}

if (bRet && NULL != pIoContext && NULL != pIocpParam)
{
try
{
pThis->ProcessIOMessage(pIoContext->m_ioType, pIoContext, dwIoSize);
}
catch (...) {}
}
}
return 0;
}

（3）接收客户端连接的操作

OnClientAccept就是映射的接受客户连接的操作，m_lpfnGetAcceptExSockAddrs就是我们获得的用来获取连接的套接字的第一条信息、本地地址信息和套接字的地址信息的函数指针。为什么我们要分配一个新的结构体呢（结构体的详细信息见示例代码）？因为用于接收连接的套接字结构体是由全局链表m_listAcceptExSock管理的（清空的时候利于清理），该套接字稍后我们还要清空并再次投递异步接收（可以理解为，我们专门为接收客户端连接创建了一个内存池，里面存放了十个创建好的结构体，我们用后还要放回去）。

设置心跳包，该心跳包设置的的为底层帮助我们检测是否掉线，不会将检测 的信息反馈到我们的应用层。这里额外几句，常见的心跳包其实在UDP中常见，因为UDP是无连接的，无法知道是否还在线，通常还可以专门设置一个线程用于向连接的套接字发送消息，通过一个变量统计是否10次（多少次实际而定）都没有反应（有反应就减一，无反应就加一），统计值为10时，此时就是掉线了。为什么我们TCP是面向连接的还需要心跳包呢？因为网线突然断了我们是无法接收到的，该套接字就会在我们的服务器中遗留，长期运行的服务器不用说这也是非常危险的，所以我们需要心跳包的设置。

投递一个异步操作，当接受到客户端的数据时，我们就能在工作者线程中通过获取完成端口的数据得到客户端发送的数据。



OnAccept

// 接收到客户端的连接
bool CIOCPServer::OnAccept(PER_IO_CONTEXT* pIoContext)
{
SOCKADDR_IN* RemoteSockAddr = NULL;
SOCKADDR_IN* LocalSockAddr = NULL;
int nLen = sizeof(SOCKADDR_IN);

///////////////////////////////////////////////////////////////////////////
// 1. m_lpfnGetAcceptExSockAddrs 取得客户端和本地端的地址信息 与 客户端发来的第一组数据
// 如果客户端只是连接了而不发消息，是不会接收到的
this->m_lpfnGetAcceptExSockAddrs(
pIoContext->m_wsaBuf.buf,                       // 第一条信息
pIoContext->m_wsaBuf.len - ((nLen + 16) * 2),
nLen + 16, nLen + 16,
(sockaddr**)&LocalSockAddr, &nLen,              // 本地信息
(sockaddr**)&RemoteSockAddr, &nLen);            // 客户端信息

PER_IO_CONTEXT* pNewIoContext = AllocateClientIOContext();
pNewIoContext->m_sock = pIoContext->m_sock;
pNewIoContext->m_addr = *RemoteSockAddr;

// 处理消息，此处为连接上，第一次接受到客户端的数据
m_pNotifyProc(NULL, pIoContext, NC_CLIENT_CONNECT);

IOCP_PARAM* pIocpParam = AllocateIocpParam();
pIocpParam->m_sock = pNewIoContext->m_sock;
// 将新连接的客户端的socket，绑定到完成端口
if (!AssociateSocketWithCompletionPort(pNewIoContext->m_sock,(DWORD)pIocpParam))
{
closesocket(m_socListen);
closesocket(pNewIoContext->m_sock);

delete pNewIoContext;
delete pIocpParam;
pNewIoContext = NULL;
pIocpParam = NULL;
return false;
}

// Set KeepAlive 设置心跳包，开启保活机制，用于保证TCP的长连接（它会在底层发一些数据，不会传到应用层）
unsigned long chOpt = 1;
if (SOCKET_ERROR == setsockopt(pNewIoContext->m_sock,SOL_SOCKET,SO_KEEPALIVE,(char*)&chOpt,sizeof(char)))
{
// 心跳激活失败
MoveToFreeParamPool(pIocpParam);
RemoveStaleClient(pNewIoContext,TRUE);
return false;
}

// 设置超时详细信息
tcp_keepalive   klive;
klive.onoff = 1; // 启用保活
klive.keepalivetime = m_nKeepLiveTime;
klive.keepaliveinterval = 1000 * 10; // 重试间隔为10秒 Resend if No-Reply
WSAIoctl
(
pNewIoContext->m_sock,
SIO_KEEPALIVE_VALS,
&klive,
sizeof(tcp_keepalive),
NULL,
0,
(unsigned long *)&chOpt,
0,
NULL
);

// 给新连接的套接字投递接收操作
if (!PostRecv(pNewIoContext))
{
MoveToFreeParamPool(pIocpParam);
}

CLock cs(m_cs, "OnAccept");

pIoContext->Clear();        // 再次初始化，便于再次利用
return PostAcceptEx(pIoContext);
}

PostAcceptEx

bool CIOCPServer::PostAcceptEx(PER_IO_CONTEXT* pAcceptIoContext)
{
pAcceptIoContext->m_ioType = IOAccept;// 初始化 IO类型 为接收套接字

// 为以后新连入的客户端准备好Socket（这是与传统Accept最大的区别）
// 实际上市创建一个 网络连接池 ，类似于 内存池，我们先创建一定数量的socket，然后直接使用就是了
pAcceptIoContext->m_sock = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_IP, NULL, 0, WSA_FLAG_OVERLAPPED);
if (INVALID_SOCKET == pAcceptIoContext->m_sock)
return false;

// 投递异步 AcceptEx
if (FALSE == m_lpfnAcceptEx(
m_socListen,
pAcceptIoContext->m_sock,
pAcceptIoContext->m_wsaBuf.buf,
pAcceptIoContext->m_wsaBuf.len - ((sizeof(SOCKADDR_IN) + 16) * 2),
sizeof(SOCKADDR_IN) + 16,
sizeof(SOCKADDR_IN) + 16,
&(pAcceptIoContext->m_dwBytesRecv),
&(pAcceptIoContext->m_ol))
)
{
if (WSA_IO_PENDING != WSAGetLastError())
return false;
}

return true;
}

（4）接收到客户端数据的操作

回调函数指针是我们初始化的时候就赋值的，具体你怎么处理数据就看你的实际应用了。处理之后，自动投递一个异步接收，方便我们下次再接收客户端的数据。



OnClientReading

bool CIOCPServer::OnClientReading(PER_IO_CONTEXT* pIOContext, DWORD dwSize /*= 0*/)
{
CLock cs(m_cs, "OnClientReading");
bool bRet = false;
try
{
// 处理接收到的数据
m_pNotifyProc(NULL, pIOContext, NC_RECEIVE);

// 再投递一个异步接收消息
bRet = PostRecv(pIOContext);
}
catch (...){}

return bRet;
}

PostRecv

bool CIOCPServer::PostRecv(PER_IO_CONTEXT* pIoContext)
{
// 清空缓冲区，再次投递
ZeroMemory(pIoContext->m_szBuf, MAX_BUFFER_LEN);
ZeroMemory(&pIoContext->m_ol, sizeof(OVERLAPPED));
pIoContext->m_ioType = IORecv;
DWORD dwNumBytesOfRecvd;

ULONG ulFlags = MSG_PARTIAL;
UINT nRet = WSARecv(
pIoContext->m_sock,
&(pIoContext->m_wsaBuf),
1,
&dwNumBytesOfRecvd,// 接收的字节数，异步操作的返回结果一般为0，具体接收到的字节数在完成端口获得
&(ulFlags),
&(pIoContext->m_ol),
NULL);
if (SOCKET_ERROR == nRet && WSA_IO_PENDING != WSAGetLastError())
{
RemoveStaleClient(pIoContext, FALSE);
return false;
}
return true;
}

（5）处理发送给客户端的操作

这样可能有点没说清楚，发送给客户端自然是服务端做的事，至于我们为什么要处理自己发送的消息。

因为我们不知道我们发送的数据是否成功并完整发送，如果没有完整发送，我们就需要再次投递异步发送，重新发送。

关于内存池，也没有什么神秘，后面会简单讲解下。



OnClientWriting

bool CIOCPServer::OnClientWriting(PER_IO_CONTEXT* pIOContext, DWORD dwSize /*= 0*/)
{
bool bRet = false;
try
{
// 异步发送的返回的传输成功的结果是否 少于 要求发送的数据大小（未发送完成），此时重发
// 最好使用CRC校验之类的，更加严谨性（可以在结构体中放一个计算的CRC值），但是计算会更消耗性能
if (dwSize != pIOContext->m_dwBytesSend)
{
bRet = PostSend(pIOContext);
}
else// 已经发送成功，将结构体放回内存池
{
m_pNotifyProc(NULL, pIOContext, NC_TRANSMIT);
MoveToFreePool(pIOContext);
}
}
catch (...) {}

return bRet;
}

PostSend

bool CIOCPServer::PostSend(PER_IO_CONTEXT* pIoContext)
{
pIoContext->m_wsaBuf.buf = pIoContext->m_szBuf;
pIoContext->m_wsaBuf.len = strlen(pIoContext->m_szBuf);
pIoContext->m_ioType = IOSend;
ULONG ulFlags = MSG_PARTIAL;

INT nRet = WSASend(
pIoContext->m_sock,
&pIoContext->m_wsaBuf,
1,
&(pIoContext->m_wsaBuf.len),
ulFlags,
&(pIoContext->m_ol),
NULL);
if (SOCKET_ERROR == nRet && WSA_IO_PENDING != WSAGetLastError())
{
RemoveStaleClient(pIoContext, FALSE);
return false;
}
return true;
}

具体的重要过程也就这么些了，想来也是挺简单的。不过要弄懂还是要花一些时间。

关于内存池：示例代码中，我们共用到两个内存池，都是使用链表来处理的（其实就是链表…），但是我们只要理解到内存池的原理后无论是链表还是红黑树都不是阻碍。示例代码中，Allocate开头的方法就是提取内存池中的一个结构体，MoveToFree开头的方法就是将不使用的结构体存放到内存池中。这样我们不用花过多的时间片用于创建结构体与释放结构体，我们在关闭服务器的时候统一处理就行了。至于我们在内存池中取的数据超过内存池时，这没事，我们可以扩展内存池。

4、收获

（1）经典的用于线程线程同步的类：

class CLock
{
public:
CLock(CRITICAL_SECTION& cs, const std::string& strFunc)
{
m_strFunc = strFunc;
m_pcs = &cs;
Lock();
}
~CLock()
{
Unlock();
}

protected:
CRITICAL_SECTION*   m_pcs;
std::string         m_strFunc;

void Unlock()
{
LeaveCriticalSection(m_pcs);
}

void Lock()
{
EnterCriticalSection(m_pcs);
}
};

（2）虽然MFC中常见窗口消息映射，但是实际开发中并没有使用到，没想到用起来还挺不错的。记录在此，方便各位道友学习

消息映射的声明：

#ifndef __IO_MAPPER__
#define __IO_MAPPER__

#define net_msg

/*
__declspec(novtable) 在C++中接口中广泛应用. 不容易看到它是因为在很多地方它都被定义成
为了宏. 比如说ATL活动模板库中的ATL_NO_VTABLE, 其实就是__declspec(novtable).

__declspec(novtable) 就是让类不要有虚函数表以及对虚函数表的初始化代码, 这样可以节省运
行时间和空间.但是这个类一定不允许生成实例, 因为没有虚函数表, 就无法对虚函数进行调用.

因此, __declspec(novtable)一般是应用于接口(其实就是包含纯虚函数的类), 因为接口包含的都
是纯虚函数,不可能生成实例. 我们把 __declspec(novtable)应用到接口类中, 这些接口类就不用
包含虚函数表和初始化虚函数表的代码了. 它的派生类会自己包含自己的虚函数表和初始化代码.

*/

class __declspec(novtable) CIOMessageMap
{
public:
virtual bool ProcessIOMessage(IOType clientIO, PER_IO_CONTEXT* pIOContext, DWORD dwSize) = 0;
};

#define BEGIN_IO_MSG_MAP() \
public: \
bool ProcessIOMessage(IOType clientIO, PER_IO_CONTEXT* pIOContext, DWORD dwSize) \
{ \
bool bRet = false;

#define IO_MESSAGE_HANDLER(msg, func) \
if(msg == clientIO) \
bRet = func(pIOContext, dwSize);

#define END_IO_MSG_MAP() \
return bRet; \
}

#endif // !__IO_MAPPER__

消息映射的使用：

// 消息映射
BEGIN_IO_MSG_MAP()
IO_MESSAGE_HANDLER(IORecv, OnClientReading)
IO_MESSAGE_HANDLER(IOSend, OnClientWriting)
IO_MESSAGE_HANDLER(IOAccept, OnClientAccept)
END_IO_MSG_MAP()

本文示例源码：https://pan.baidu.com/s/1geOIXwv

希望对各位道友有所帮助，本文难免有所错误，如有问题欢迎留言