【WinSock】网络编程 – I/O模型(三) – 完成端口

接上文，写完最后一个I/O模型 – 完成端口

“完成端口” 模型是迄今为止最为复杂的一种I/O模型

假若一个应用程序同时需要管理为数众多的套接字，那么采用这种模型，往往可以达到最佳的系统性能！

因其设计的复杂性，只有在你的应用程序需要同时管理数百乃至上千个套接字的时候，而且希望随着系统内安装的CPU数量的增多，应用程序的性能也可以线性提升，才应考虑采用“完成端口”模型。

要记住的一个基本准则是，希望为大量套接字I/O请求提供服务（Web服务器便是这方面的典型例子），那么I/O完成端口模型便是最佳选择！

完成端口模型在多连接（成千上万）的情况下，仅仅依靠一两个辅助线程，就可以达到非常高的吞吐量。

还是上代码吧：

#include <Winsock2.h>
#include <stdio.h>
#include <process.h>

#pragma comment(lib,"ws2_32.lib")

#define PORT 5150
#define MSGSIZE 1024

typedef enum
{
	RECV_POSTED,
}OPERATION_TYPE;

typedef struct
{
	WSAOVERLAPPED overlap;
	WSABUF Buffer;
	char szMessage[MSGSIZE];
	DWORD NumberOfBytesRecvd;
	DWORD Flags;
	OPERATION_TYPE OperationType;
}PER_IO_OPERATION_DATA, *LPPER_IO_OPERATION_DATA;

unsigned int WINAPI WorkerThread(void *pParam);

int main()
{
	//初始化环境
	WORD wVersionRequested = MAKEWORD(2, 2);
	WSADATA wsaData;
	WSAStartup(wVersionRequested, &wsaData);

	//创建完成端口
	HANDLE CompletionPort = INVALID_HANDLE_VALUE;
	CompletionPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);

	//创建工作线程
	SYSTEM_INFO systeminfo;
	GetSystemInfo(&systeminfo);
	unsigned int dwThreadId;
	for(int i = 0; i < systeminfo.dwNumberOfProcessors; i++)
	{
		_beginthreadex(NULL, 0, WorkerThread, CompletionPort, 0, &dwThreadId);
	}

	//创建监听套接字
	SOCKET sListen = socket(AF_INET, SOCK_STREAM, 0);

	//绑定
	SOCKADDR_IN local;
	local.sin_addr.S_un.S_addr = htonl(INADDR_ANY);
	local.sin_family = AF_INET;
	local.sin_port = htons(PORT);
	bind(sListen, (struct sockaddr *)&local, sizeof(SOCKADDR_IN));

	//侦听
	listen(sListen, 20);

	while(TRUE)
	{
		//接受连接
		SOCKADDR_IN client;
		int iaddrSize = sizeof(SOCKADDR_IN);
		SOCKET sClient = accept(sListen, (struct sockaddr*)&client, &iaddrSize);
		printf("Accepted client: %s:%d\n", inet_ntoa(client.sin_addr), ntohs(client.sin_port));

		// Associate the newly arrived client socket with completion port
		// 将新来的客户端套接字与完成端口关联起来
		CreateIoCompletionPort((HANDLE)sClient, CompletionPort, (DWORD)sClient, 0);

		//创建一个新的结构体
		LPPER_IO_OPERATION_DATA lpPerIOData = (LPPER_IO_OPERATION_DATA)HeapAlloc(
			GetProcessHeap(),
			HEAP_ZERO_MEMORY,
			sizeof(PER_IO_OPERATION_DATA));

		//初始化结构体
		lpPerIOData->Buffer.len = MSGSIZE;
		lpPerIOData->Buffer.buf = lpPerIOData->szMessage;
		lpPerIOData->OperationType = RECV_POSTED;

		//发送异步消息
		WSARecv(
			sClient,
			&lpPerIOData->Buffer,
			1,
			&lpPerIOData->NumberOfBytesRecvd,
			&lpPerIOData->Flags,
			&lpPerIOData->overlap,
			NULL);
	}

	//发送退出消息
	PostQueuedCompletionStatus(CompletionPort, 0xFFFFFFFF, 0, NULL);
	CloseHandle(CompletionPort);
	closesocket(sListen);
	WSACleanup();
	return 0;
}

unsigned int WINAPI WorkerThread(void * CompletionPortID)
{
	HANDLE CompletionPort = (HANDLE)CompletionPortID;

	while(TRUE)
	{

		DWORD dwBytesTransferred;
		SOCKET sClient;
		LPPER_IO_OPERATION_DATA  lpPerIOData = NULL;
		//I-O的相关信息	
		GetQueuedCompletionStatus(
			CompletionPort,
			&dwBytesTransferred,
			(PULONG_PTR)
			&sClient, 
			(LPOVERLAPPED *)&lpPerIOData,
			INFINITE);

		//判断结果
		if(dwBytesTransferred == 0xFFFFFFFF)
		{
			return 0;
		}

		if(lpPerIOData->OperationType == RECV_POSTED)
		{
			if(dwBytesTransferred == 0)
			{
				//客户端关闭了
				closesocket(sClient);
				HeapFree(GetProcessHeap(), 0, lpPerIOData);
			}
			else
			{
				//回射
				lpPerIOData->szMessage[dwBytesTransferred] = '\0';
				send(sClient, lpPerIOData->szMessage, dwBytesTransferred, 0);

				//重置
				memset(lpPerIOData, 0, sizeof(PER_IO_OPERATION_DATA));
				lpPerIOData->Buffer.len = MSGSIZE;
				lpPerIOData->Buffer.buf = lpPerIOData->szMessage;
				lpPerIOData->OperationType = RECV_POSTED;

				//再次请求
				WSARecv(sClient,
						&lpPerIOData->Buffer,
						1,
						&lpPerIOData->NumberOfBytesRecvd,
						&lpPerIOData->Flags,
						&lpPerIOData->overlap,
						NULL);

			}
		}
	}
	return 0;
}

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

#include <Winsock2.h>

#include <stdio.h>

#include <process.h>

#pragma comment(lib,"ws2_32.lib")

#define PORT 5150

#define MSGSIZE 1024

typedef enum

{

RECV_POSTED,

}OPERATION_TYPE;

typedef struct

{

WSAOVERLAPPED overlap;

WSABUF Buffer;

char szMessage[MSGSIZE];

DWORD NumberOfBytesRecvd;

DWORD Flags;

OPERATION_TYPE OperationType;

}PER_IO_OPERATION_DATA, *LPPER_IO_OPERATION_DATA;

unsigned int WINAPI WorkerThread(void *pParam);

int main()

{

//初始化环境

WORD wVersionRequested = MAKEWORD(2, 2);

WSADATA wsaData;

WSAStartup(wVersionRequested, &wsaData);

//创建完成端口

HANDLE CompletionPort = INVALID_HANDLE_VALUE;

CompletionPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);

//创建工作线程

SYSTEM_INFO systeminfo;

GetSystemInfo(&systeminfo);

unsigned int dwThreadId;

for(int i = 0; i < systeminfo.dwNumberOfProcessors; i++)

{

_beginthreadex(NULL, 0, WorkerThread, CompletionPort, 0, &dwThreadId);

}

//创建监听套接字

SOCKET sListen = socket(AF_INET, SOCK_STREAM, 0);

//绑定

SOCKADDR_IN local;

local.sin_addr.S_un.S_addr = htonl(INADDR_ANY);

local.sin_family = AF_INET;

local.sin_port = htons(PORT);

bind(sListen, (struct sockaddr *)&local, sizeof(SOCKADDR_IN));

//侦听

listen(sListen, 20);

while(TRUE)

{

//接受连接

SOCKADDR_IN client;

int iaddrSize = sizeof(SOCKADDR_IN);

SOCKET sClient = accept(sListen, (struct sockaddr*)&client, &iaddrSize);

printf("Accepted client: %s:%d\n", inet_ntoa(client.sin_addr), ntohs(client.sin_port));

// Associate the newly arrived client socket with completion port

// 将新来的客户端套接字与完成端口关联起来

CreateIoCompletionPort((HANDLE)sClient, CompletionPort, (DWORD)sClient, 0);

//创建一个新的结构体

LPPER_IO_OPERATION_DATA lpPerIOData = (LPPER_IO_OPERATION_DATA)HeapAlloc(

GetProcessHeap(),

HEAP_ZERO_MEMORY,

sizeof(PER_IO_OPERATION_DATA));

//初始化结构体

lpPerIOData->Buffer.len = MSGSIZE;

lpPerIOData->Buffer.buf = lpPerIOData->szMessage;

lpPerIOData->OperationType = RECV_POSTED;

//发送异步消息

WSARecv(

sClient,

&lpPerIOData->Buffer,

&lpPerIOData->NumberOfBytesRecvd,

&lpPerIOData->Flags,

&lpPerIOData->overlap,

NULL);

}

//发送退出消息

PostQueuedCompletionStatus(CompletionPort, 0xFFFFFFFF, 0, NULL);

CloseHandle(CompletionPort);

closesocket(sListen);

WSACleanup();

return 0;

}

unsigned int WINAPI WorkerThread(void * CompletionPortID)

{

HANDLE CompletionPort = (HANDLE)CompletionPortID;

while(TRUE)

{

DWORD dwBytesTransferred;

SOCKET sClient;

LPPER_IO_OPERATION_DATA lpPerIOData = NULL;

//I-O的相关信息

GetQueuedCompletionStatus(

CompletionPort,

&dwBytesTransferred,

(PULONG_PTR)

&sClient,

(LPOVERLAPPED *)&lpPerIOData,

INFINITE);

//判断结果

if(dwBytesTransferred == 0xFFFFFFFF)

{

return 0;

}

if(lpPerIOData->OperationType == RECV_POSTED)

{

if(dwBytesTransferred == 0)

{

//客户端关闭了

closesocket(sClient);

HeapFree(GetProcessHeap(), 0, lpPerIOData);

}

else

{

//回射

lpPerIOData->szMessage[dwBytesTransferred] = '\0';

send(sClient, lpPerIOData->szMessage, dwBytesTransferred, 0);

//重置

memset(lpPerIOData, 0, sizeof(PER_IO_OPERATION_DATA));

lpPerIOData->Buffer.len = MSGSIZE;

lpPerIOData->Buffer.buf = lpPerIOData->szMessage;

lpPerIOData->OperationType = RECV_POSTED;

//再次请求

WSARecv(sClient,

&lpPerIOData->Buffer,

&lpPerIOData->NumberOfBytesRecvd,

&lpPerIOData->Flags,

&lpPerIOData->overlap,

NULL);

}

return 0;

}

完成端口的主要流程大致如下：

创建完成端口对象
创建工作者线程（这里工作者线程的数量是按照CPU的个数来决定的，这样可以达到最佳性能）
创建监听套接字，绑定，监听，然后程序进入循环

在循环中，我做了以下几件事情：

1. 接受一个客户端连接
2. 将该客户端套接字与完成端口绑定到一起(还是调用CreateIoCompletionPort，但这次的作用不同)，注意，按道理来讲，此时传递给 CreateIoCompletionPort的第三个参数应该是一个完成键，一般来讲，程序都是传递一个单句柄数据结构的地址，该单句柄数据包含了和该客户端连接有关的信息，由于我们只关心套接字句柄，所以直接将套接字句柄作为完成键传递；
3. 触发一个WSARecv异步调用，这次又用到了“尾随数据”，使接收数据所用的缓冲区紧跟在WSAOVERLAPPED对象之后，此外，还有操作类型等重要信息。

在工作者线程的循环中，我们步骤如下：
1.调用GetQueuedCompletionStatus取得本次I/O的相关信息（例如套接字句柄、传送的字节数、单I/O数据结构的地址等等）
2.通过单I/O数据结构找到接收数据缓冲区，然后将数据原封不动的发送到客户端(回射)，当然我们也可以处理，然后将处理结果发送过去。
3.再次触发一个WSARecv异步操作

这样就是一个完成端口的大致流程。

PS.完成端口模型又叫 IOCP … 🙂

最后附上前面两篇的链接：

【WinSock】网络编程 – I/O模型(一) – 选择模型

【WinSock】网络编程 – I/O模型(二) – 重叠I/O

…有问题可以留言。

【WinSock】网络编程 – I/O模型(三) – 完成端口

Tagged on: Network