《TCP/IP网络编程》学习笔记 | Chapter 23:IOCP
- 《TCP/IP网络编程》学习笔记 | Chapter 23:IOCP
- 通过重叠 I/O 理解 IOCP
- epoll 和 IOCP 的性能比较
- 实现非阻塞模式的套接字
- 以纯重叠 I/O 方式实现回声服务器端
- 重新实现客户端
- 测试
- 从重叠 I/O 模型到 IOCP 模型
- 分阶段实现 IOCP 程序
- 创建“完成端口"
- 连接完成端口对象和套接字
- 确认完成端口已完成的 I/O 和线程的 I/O 处理
- 实现基于 IOCP 的回声服务器端
- IOCP 性能更优的原因
- 习题
- (1)完成端口对象将分配多个线程用于处理 I/O。如何创建这些线程?如何分配?请从源代码级别进行说明。
- (2)CreateIoCompletionPort 函数与其他函数不同,提供 2 种功能。请问是哪 2 种?
- (3)完成端口对象和套接字之间的连接意味着什么?如何连接?
- (4)下列关于 IOCP 的说法错误的是?
- (5)判断下列关于 IOCP 中选择合理线程数的方法是否合适。
- (6)利用本章的 IOCP 模型实现聊天服务器端,该聊天服务器端应当结合第 20 章的聊天客户端 chat_clnt_win.c 正常运行。编写程序时不必刻意套用本章 IOCP 示例中的框架,那样反而会加大实现难度。
《TCP/IP网络编程》学习笔记 | Chapter 23:IOCP
通过重叠 I/O 理解 IOCP
epoll 和 IOCP 的性能比较
两种模型都很优秀,它们的差异主要在于操作系统内部的工作机制。
服务器端的响应时间和并发服务数是衡量服务器端好坏的重要因素。硬件性能和分配带宽充足情况下,若响应时间和并发服务数出了问题,查看以下两点:
- 低效的 I/O 结构或低效的 CPU 使用
- 数据库设计和查询语句(Query)的结构
实现非阻塞模式的套接字
前一章中只介绍了执行重叠 I/O 的 Sender 和 Receiver,但还未利用该模型实现过服务器端。因此,我们先利用重叠 I/O 模型实现回声服务器端。
首先介绍创建非阻塞模式套接字的方法。我们曾在第 17 章创建过非阻塞模式的套接字,与之类似,在 Windows 中通过如下函数调用将套接字属性改为非阻塞模式。
SOCKET hLisnSock;
int mode = 1;
.....
hListSock = WSASocket(PF_INET, SOCK_STREAM, 0, NULL, 0, WSA_FLAG_OVERLAPPED); //创建重叠 IO
ioctlsocket(hLisnSock, FIONBIO, &mode); // 设定套接字为非阻塞属性
.....
ioctlsocket 函数负责控制套接字 I/O 方式,其调用具有如下含义:将 hLisnSock 句柄引用的套接字 I/O 模式(FIONBIO)改为变量 mode 中指定的形式。也就是说,FIONBIO 是用于更改套接字 I/O 模式的选项,该函数的第三个参数中传入的变量中若存有 0,则说明套接字是阻塞模式的;如果存有非 0 值,则说明已将套接字模式改为非阻塞模式。改为非阻塞模式后,除了以非阻塞模式进行 I/O 外,还具有如下特点:
- 如果在没有客户端连接请求的状态下调用 accept 函数,将直接返回 INVALID_SOCKET。调用 WSAGetLastError 函数时返回 WSAEWOULDBLOCK。
- 调用 accept 函数时创建的套接字同样具有非阻塞属性。
因此,针对非阻塞套接字调用 accept 函数并返回 INVALID_SOCKET 时,应该通过 WSAGetLastError 函数确认返回 INVALID_SOCKET 的理由,再进行适当处理。
以纯重叠 I/O 方式实现回声服务器端
由于代码量较大,我们分3个部分学习。
片段 1:main 函数之前
#include <stdio.h>
#include <stdlib.h>
#include <winsock2.h>
#define BUF_SIZE 1024
void CALLBACK ReadCompRoutine(DWORD, DWORD, LPWSAOVERLAPPED, DWORD);
void CALLBACK WriteCompRoutine(DWORD, DWORD, LPWSAOVERLAPPED, DWORD);
void ErrorHandling(char *message);
typedef struct
{
SOCKET hClntSock; // 套接字句柄
char buf[BUF_SIZE]; // 读写数据放置地址
WSABUF wsaBuf; // 指向放置读写的数据的地址和大小
} PER_IO_DATA, *LPPER_IO_DATA;
PER_IO_DATA 结构体中的信息足够进行数据交换。
片段 2:main 函数
int main(int argc, char *argv[])
{
WSADATA wsaData;
SOCKET hLisnSock, hRecvSock;
SOCKADDR_IN lisnAdr, recvAdr;
LPWSAOVERLAPPED lpOvLp;
DWORD recvBytes, flagInfo = 0;
LPPER_IO_DATA hbInfo;
u_long mode = 1;
int recvAdrsz;
if (argc != 2)
{
printf("Usage: %s <port>\n", argv[0]);
exit(1);
}
if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0)
ErrorHandling("WSAStartup() error!");
hLisnSock = WSASocket(PF_INET, SOCK_STREAM, 0, NULL, 0, WSA_FLAG_OVERLAPPED);
ioctlsocket(hLisnSock, FIONBIO, &mode);
memset(&lisnAdr, 0, sizeof(lisnAdr));
lisnAdr.sin_family = AF_INET;
lisnAdr.sin_addr.s_addr = htonl(INADDR_ANY);
lisnAdr.sin_port = htons(atoi(argv[1]));
if (bind(hLisnSock, (SOCKADDR *)&lisnAdr, sizeof(lisnAdr)) == SOCKET_ERROR)
ErrorHandling("bind() error");
if (listen(hLisnSock, 5) == SOCKET_ERROR)
ErrorHandling("listen() error");
recvAdrsz = sizeof(recvAdr);
while (1)
{
SleepEx(100, TRUE);
hRecvSock = accept(hLisnSock, (SOCKADDR *)&recvAdr, &recvAdrsz);
if (hRecvSock == INVALID_SOCKET)
{
if (WSAGetLastError() == WSAEWOULDBLOCK)
continue;
else
ErrorHandling("accept() error");
}
puts("Client connected.....");
lpOvLp = (LPWSAOVERLAPPED)malloc(sizeof(WSAOVERLAPPED));
memset(lpOvLp, 0, sizeof(WSAOVERLAPPED));
hbInfo = (LPPER_IO_DATA)malloc(sizeof(PER_IO_DATA));
hbInfo->hClntSock = (DWORD)hRecvSock;
(hbInfo->wsaBuf).buf = hbInfo->buf;
(hbInfo->wsaBuf).len = BUF_SIZE;
lpOvLp->hEvent = (HANDLE)hbInfo;
WSARecv(hRecvSock, &(hbInfo->wsaBuf), 1, &recvBytes, &flagInfo, lpOvLp, ReadCompRoutine);
}
closesocket(hRecvSock);
closesocket(hLisnSock);
WSACleanup();
return 0;
}
注意几点:
- 之所以在循环内部申请 WSAOVERLAPPED 结构体空间,是因为每个客户端都需要独立的 WSAOVERLAPPED 结构体变量。
- 基于 Completion Routine 函数的重叠 I/O 中不需要事件对象,因此,hEvent 中可以写入自定义结构体信息。
- 调用 WSARecv 函数时将 ReadCompRoutine 函数指定为 Completion Routine。其
中第六个参数 WSAOVERLAPPED 结构体变量地址值将传递到 Completion Routine 的第三个参数,因此,Completion Routine 函数内可以访问完成 I/O 的套接字句柄和缓冲。 - 为了运行 Completion Routine 函数,循环调用 SleepEx 函数。
片段 3:2 个 Completion Routine 函数
void CALLBACK ReadCompRoutine(DWORD dwError, DWORD szRecvBytes, LPWSAOVERLAPPED lpOverlapped, DWORD flags)
{
LPPER_IO_DATA hbInfo = (LPPER_IO_DATA)(lpOverlapped->hEvent);
SOCKET hSock = hbInfo->hClntSock;
LPWSABUF bufInfo = &(hbInfo->wsaBuf);
DWORD sentBytes;
if (szRecvBytes == 0)
{ // 如果接收到了 EOF 那么则关闭套接字以及释放对应的空间
closesocket(hSock);
free(lpOverlapped->hEvent);
free(lpOverlapped);
puts("Client disconnected.....");
}
else
{ // 如果不为零那么一定有需要回声的内容
bufInfo->len = szRecvBytes;
WSASend(hSock, bufInfo, 1, &sentBytes, 0, lpOverlapped, WriteCompRoutine);
}
}
void CALLBACK WriteCompRoutine(DWORD dwError, DWORD szSendBytes, LPWSAOVERLAPPED lpOverlapped, DWORD flags)
{
LPPER_IO_DATA hbInfo = (LPPER_IO_DATA)(lpOverlapped->hEvent);
SOCKET hSock = hbInfo->hClntSock;
LPWSABUF bufInfo = &(hbInfo->wsaBuf);
DWORD recvBytes, flagInfo = 0;
// 发送后默认等待再次回声
WSARecv(hSock, bufInfo, 1, &recvBytes, &flagInfo, lpOverlapped, ReadCompRoutine);
}
void ErrorHandling(char *message)
{
fputs(message, stderr);
fputc('\n', stderr);
exit(1);
}
实际的回声服务是通过这两个函数完成的。
上述示例的工作原理整理如下:
- 有新的客户端连接时调用 WSARecv 函数,并以非阻塞模式接收数据,接收完成后调用 ReadCompRoutine 函数。
- 调用 ReadCompRoutine 函数后调用 WSASend 函数,并以非阻塞模式发送数据,发送完成后
调用 WriteCompRoutine 函数。 - 此时调用的 WriteCompRoutine 函数将再次调用 WSARecv 函数,并以非阻塞模式等待接收数据。
通过交替调用 ReadCompRoutine 函数和 WriteCompRoutine 函数,反复执行数据的接收和发送操作。另外,每次增加 1 个客户端都会定义 PER_IO_DATA 结构体,以便将新创建的套接字句柄和缓冲信息传递给 ReadCompRoutine 函数和 WriteCompRoutine 函数。同时将该结构体地址值写入 WSAOVERLAPPED 结构体成员 hEvent,并传递给 Completion Routine 函数。这非常重要,可概括如下:“使用 WSAOVERLAPPED 结构体成员 hEvent 向完成 I/O 时自动调用的 Completion Routine 函数内部传递客户端信息(套接字和缓冲)。”
接下来需要验证运行结果,先要编写回声客户端,因为使用第 4 章的回声客户端会无法得到预想的结果。
重新实现客户端
#include <stdio.h>
#include <stdlib.h>
#include <winsock2.h>
#define BUF_SIZE 1024
void ErrorHanding(const char *message);
int main(int argc, char *argv[])
{
WSADATA wsaData;
if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0)
ErrorHanding("WSAStartup() error!");
SOCKET hSocket = socket(PF_INET, SOCK_STREAM, 0);
if (hSocket == INVALID_SOCKET)
ErrorHanding("socket() error!");
int szAddr = sizeof(SOCKADDR_IN);
SOCKADDR_IN servAddr;
memset(&servAddr, 0, szAddr);
servAddr.sin_family = AF_INET;
servAddr.sin_addr.s_addr = inet_addr(argv[1]);
servAddr.sin_port = htons(atoi(argv[2]));
if (connect(hSocket, (SOCKADDR *)&servAddr, szAddr) == SOCKET_ERROR)
ErrorHanding("connect() error!");
else
puts("Connected.......");
while (1)
{
fputs("Input message(Q to quit): ", stdout);
char message[BUF_SIZE] = {0};
fgets(message, BUF_SIZE - 1, stdin);
if (!strcmp(message, "q\n") || !strcmp(message, "Q\n"))
break;
int strLen = strlen(message);
send(hSocket, message, strLen, 0);
int readLen = 0;
while (1)
{
readLen += recv(hSocket, &message[readLen], BUF_SIZE - 1, 0);
if (readLen >= strLen)
break;
}
message[strLen] = '\0';
printf("Message from server: %s\n", message);
}
closesocket(hSocket);
WSACleanup();
return 0;
}
void ErrorHanding(const char *message)
{
fputs(message, stderr);
fputc('\n', stderr);
exit(1);
}
考虑到 TCP 的传输特性而重复调用了 recv 函数,直至接收完所有数据。
测试
编译:
gcc CmplRouEchoServ_win.c -lwsock32 -o CmplRouEchoServ
gcc StableEchoClnt_win.c -lws2_32 -o StableEchoClnt
运行结果:
从重叠 I/O 模型到 IOCP 模型
重叠 I/O 模型回声服务器端的缺点:重复调用非阻塞模式的 accept 函数和以进入 alertable wait 状态为目的的 SleepEx 函数将影响性能。
如果正确理解了之前的示例,应该不难发现这一点。既不能为了处理连接请求而只调用 accept 函数,也不能为了 Completion Routine 而只调用 SleepEx 函数,因此轮流调用了非阻塞模式的 accept 函数和 SleepEx 函数(设置较短的超时时间)。这个恰恰是影响性能的代码结构。
这属于重叠I/O结构固有的缺陷。
可以考虑如下方法:让 main 线程(在 main 函数内部)调用 accept 函数,再单独创建 1 个线程负责客户端 I/O。
其实这就是 IOCP 中采用的服务器端模型。换言之,IOCP 将创建专用的 I/O 线程,该线程负责与所有客户端进行I/O。
分阶段实现 IOCP 程序
本节我们编写最后一种服务器模型 IOCP,比阅读代码更重要的是理解 IOCP 本身。
IOCP 关注焦点:
- I/O 是否以非阻塞模式工作?
- 如何确定非阻塞模式的 I/O 是否完成?
创建“完成端口"
IOCP 中已完成的 I/O 信息将注册到完成端口对象(Completion Port,简称 CP 对象),但这个过
程并非单纯的注册,首先需要经过如下请求过程:“该套接字的 I/O 完成时,请把状态信息注册到指定 CP 对象。"
该过程称为“套接字和CP对象之间的连接请求”。因此,为了实现基于 IOCP 模型的服务器端,需要做如下 2 项工作。
- 创建完成端口对象。
- 建立完成端口对象和套接字之间的联系。
此时的套接字必须被赋予重叠属性。上述 2 项工作可以通过 1 个函数完成,但为了创建 CP 对象,先介绍如下函数。
#include <windows.h>
HANDLE CreateIoCompletionport(
HANDLE FileHandle,
HANDLE ExistingCompletionPort,
ULONG_PTR Completionkey,
DWORD NumberofConcurrentThreads
);
参数:
- FileHandle:创建 CP 对象时传递 INVALID_HANDLE_VALUE。
- ExistingCompletionPort:创建 CP 对象时传递 NULL。
- CompletionKey:创建 CP 对象时传递 0。
- NumberOfConcurrentThreads:分配给 CP 对象的用于处理 I/O 的线程数。例如,该参数为 2 时,说明分配给 CP 对象的可以同时运行的线程数最多为 2 个;如果该参数为 0,系统中 CPU 个数(内核数)就是可同时运行的最大线程数。
成功时返回 CP 对象句柄,失败时返回 NULL。
以创建 CP 对象为目的调用上述函数时,只有最后一个参数才真正具有含义。可以用如下代码段将分配给 CP 对象的用于处理 I/O 的线程数指定为 2。
HANDLE hCpObject;
.....
hCpObject = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 2);
连接完成端口对象和套接字
既然有了 CP 对象,接下来就要将该对象连接到套接字,只有这样才能使已完成的套接字 I/O 信息注册到 CP 对象。下面以建立连接为目的再次介绍 CreateIoCompletionPort 函数。
#include <windows.h>
HANDLE CreateIoCompletionport(
HANDLE FileHandle,
HANDLE ExistingCompletionPort,
ULONG_PTR Completionkey,
DWORD NumberofConcurrentThreads
);
参数:
- FileHandle:要连接到 CP 对象的套接字句柄。
- ExistingCompletionPort:要连接套接字的 CP 对象句柄。
- CompletionKey:传递已完成 I/O 相关信息,关于该参数将在稍后介绍的 GetQueuedCompletionStatus 函数中共同讨论。
- NumberOfConcurrentThreads:无论传递何值,只要该函数的第二个参数非 NULL 就会忽略。
成功时返回 CP 对象句柄,失败时返回 NULL。
上述函数的第二种功能就是将 FileHandle 句柄指向的套接字和 ExistingCompletionPort 指向的 CP 对象相连。该函数的调用方式如下:
HANDLE hCpObject;
SOCKET hSock;
......
CreateIoCompletionPort((HANDLE)hSock, hCpObject, (DWORD)ioInfo, 0);
调用 CreateIoCompletionPort 函数后,只要针对 hSock 的 I/O 完成,相关信息就将注册到 hCpObject 指向的 CP 对象。
确认完成端口已完成的 I/O 和线程的 I/O 处理
我们已经掌握了 CP 对象的创建及其与套接字建立连接的方法,接下来就要学习如何确认 CP 中注册的已完成的 I/O。完成该功能的函数如下。
#include <windows.h>
BOOL GetQueuedCompletionStatus(
HANDLE CompletionPort,
LPDWORD lpNumberOfBytes,
PULONG_PTR lpCompletionKey,
LPOVERLAPPED * lpoverlapped,
DWORD dwMilliseconds
);
参数:
- CompletionPort:注册有已完成 I/O 信息的CP对象句柄。
- IpNumberOfBytes:用于保存 I/O 过程中传输的数据大小的变量地址值。
- IpCompletionKey:用于保存 CreateIoCompletionPort 函数的第三个参数值的变量地址值。
- IpOverlapped:用于保存调用 WSASend、WSARecv 函数时传递的 OVERLAPPED 结构体地址的变量地址值。
- dwMilliseconds:超时信息超过该指定时间后将返回FALSE并跳出函数。传递 INFINITE 时,程序将阻塞,直到已完成 I/O 信息写入 CP 对象。
成功时返回 TRUE,失败时返回 FALSE。
虽然只介绍了 2 个 IOCP 相关函数,但依然有些复杂,特别是上述函数的第三个和第四个参数更是如此。其实这 2 个参数主要是为了获取需要的信息而设置的,下面介绍这 2 种信息的含义。
- 通过 GetQueuedCompletionStatus 函数的第三个参数得到的是以连接套接字和 CP 对象为目的而调用的 CreateloCompletionPort 函数的第三个参数值。
- 通过 GetQueueCompletionStatus 函数的第四个参数得到的是调用 WSASend、WSARecv 函数时传入的 WSAOVERLAPPED 结构体变量地址值。
IOCP 中将创建全职 I/O 线程,由该线程针对所有客户端进行 I/O。程序员自行创建调用 WSASend、WSARecv 等 I/O 函数的线程,该线程为了确认 I/O 的完成会调用 GetQueuedCompletionStatus 函数。
虽然任何线程都能调用 GetQueuedCompletionStatus 函数,但实际得到 I/O 完成信息的线程数不会超过调用 CreateIoCompletionPort 函数时指定的最大线程数。
实现基于 IOCP 的回声服务器端
同样分段进行讲解。
片段 1:main 函数之前
#include <stdio.h>
#include <stdlib.h>
#include <process.h>
#include <winsock2.h>
#include <windows.h>
#define BUF_SIZE 100
#define READ 3
#define WRITE 5
typedef struct // socket info
{
SOCKET hClntSock;
SOCKADDR_IN clntAdr;
} PER_HANDLE_DATA, *LPPER_HANDLE_DATA;
typedef struct // buffer info
{
OVERLAPPED overlapped;
WSABUF wsaBuf;
char buffer[BUF_SIZE];
int rwMode; // READ or WRITE
} PER_IO_DATA, *LPPER_IO_DATA;
DWORD WINAPI EchoThreadMain(LPVOID CompletionPortIo);
void ErrorHandling(char *message);
其中,PER_HANDLE_DATA 是保存与客户端相连套接字的结构体。将 I/O 中使用的缓冲和重叠 I/O 中需要的 OVERLAPPED 结构体变量封装到同一结构体中进行定义。
片段 2:main 函数
int main(int argc, char *argv[])
{
WSADATA wsaData;
HANDLE hComPort;
SYSTEM_INFO sysInfo;
LPPER_IO_DATA ioInfo;
LPPER_HANDLE_DATA handleInfo;
SOCKET hServSock;
SOCKADDR_IN servAdr;
DWORD recvBytes, flags = 0;
if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0)
ErrorHandling("WSAStartup() error!");
hComPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0); // 创建 CP 对象
GetSystemInfo(&sysInfo); // 获得当前系统的信息
// 创建线程数等于 CPU 数量
for (int i = 0; i < sysInfo.dwNumberOfProcessors; i++)
_beginthreadex(NULL, 0, (LPVOID)EchoThreadMain, (LPVOID)hComPort, 0, NULL);
hServSock = WSASocket(AF_INET, SOCK_STREAM, 0, NULL, 0, WSA_FLAG_OVERLAPPED);
memset(&servAdr, 0, sizeof(servAdr));
servAdr.sin_family = AF_INET;
servAdr.sin_addr.s_addr = htonl(INADDR_ANY);
servAdr.sin_port = htons(atoi(argv[1]));
if (bind(hServSock, (SOCKADDR *)&servAdr, sizeof(servAdr)) == SOCKET_ERROR)
ErrorHandling("bind() error");
if (listen(hServSock, 5) == SOCKET_ERROR)
ErrorHandling("listen() error");
while (1)
{
SOCKET hClntSock;
SOCKADDR_IN clntAdr;
int addrLen = sizeof(clntAdr);
hClntSock = accept(hServSock, (SOCKADDR *)&clntAdr, &addrLen);
if (hClntSock == INVALID_SOCKET)
{
if (WSAGetLastError() == WSAEWOULDBLOCK)
continue;
else
ErrorHandling("accept() error");
}
handleInfo = (LPPER_HANDLE_DATA)malloc(sizeof(PER_HANDLE_DATA));
handleInfo->hClntSock = hClntSock;
memcpy(&(handleInfo->clntAdr), &clntAdr, addrLen);
CreateIoCompletionPort((HANDLE)hClntSock, hComPort, (ULONG_PTR)handleInfo, 0); // 建立连接
ioInfo = (LPPER_IO_DATA)malloc(sizeof(PER_IO_DATA));
memset(&(ioInfo->overlapped), 0, sizeof(OVERLAPPED));
ioInfo->wsaBuf.len = BUF_SIZE;
ioInfo->wsaBuf.buf = ioInfo->buffer;
ioInfo->rwMode = READ;
WSARecv(handleInfo->hClntSock, &(ioInfo->wsaBuf), 1, &recvBytes, &flags, &(ioInfo->overlapped), NULL);
}
closesocket(hServSock);
WSACleanup();
return 0;
}
套接字的重叠 I/O 完成时,已完成信息将写入连接的 CP 对象,这会引起 GetQueuedCompletionStatus 函数的返回。请注意观察 handleInfo,它同样是在 GetQueuedCompletionStatus 函数返回时得到的。
IOCP 本身不会帮我们区分输入完成和输出完成的状态。无论输入还是输出,只通知完成 I/O 的状态,因此需要通过额外的变量区分这 2 种 I/O。PER_IO_DATA 结构体中的 rwMode 就用于完成该功能。
WSARecv 函数的第七个参数为 OVERLAPPED 结构体变量地址值,该值也可以在 GetQueuedCompletionStatus 函数返回时得到。
片段 3:线程的 main 函数
DWORD WINAPI EchoThreadMain(LPVOID pComPort)
{
HANDLE hComPort = (HANDLE)pComPort;
SOCKET sock;
DWORD bytesTrans;
LPPER_HANDLE_DATA handleInfo;
LPPER_IO_DATA ioInfo;
DWORD flags = 0;
while (1)
{
GetQueuedCompletionStatus(hComPort, &bytesTrans, (PULONG_PTR)&handleInfo, (LPOVERLAPPED *)&ioInfo, INFINITE);
sock = handleInfo->hClntSock;
if (ioInfo->rwMode == READ)
{
puts("message received!");
if (bytesTrans == 0)
{ // 传输 EOF 时
closesocket(sock);
free(handleInfo);
free(ioInfo);
continue;
}
memset(&(ioInfo->overlapped), 0, sizeof(OVERLAPPED));
ioInfo->wsaBuf.len = bytesTrans;
ioInfo->rwMode = WRITE;
WSASend(sock, &(ioInfo->wsaBuf), 1, NULL, 0, &(ioInfo->overlapped), NULL);
ioInfo = (LPPER_IO_DATA)malloc(sizeof(PER_IO_DATA));
memset(&(ioInfo->overlapped), 0, sizeof(OVERLAPPED));
ioInfo->wsaBuf.len = BUF_SIZE;
ioInfo->wsaBuf.buf = ioInfo->buffer;
ioInfo->rwMode = READ;
WSARecv(sock, &(ioInfo->wsaBuf), 1, NULL, &flags, &(ioInfo->overlapped), NULL);
}
else
{
puts("message sent!");
free(ioInfo);
}
}
return 0;
}
void ErrorHandling(char *message)
{
fputs(message, stderr);
fputc('\n', stderr);
exit(1);
}
指针 iolnfo 中保存的既是 OVERLAPPED 结构体变量地址值,也是 PER_IO_DATA 结构体变量地址值。因此,可以通过检查 rwMode 成员中的值判断是输入完成还是输出完成。
完整程序:
#include <stdio.h>
#include <stdlib.h>
#include <process.h>
#include <winsock2.h>
#include <windows.h>
#define BUF_SIZE 100
#define READ 3
#define WRITE 5
typedef struct // socket info
{
SOCKET hClntSock;
SOCKADDR_IN clntAdr;
} PER_HANDLE_DATA, *LPPER_HANDLE_DATA;
typedef struct // buffer info
{
OVERLAPPED overlapped;
WSABUF wsaBuf;
char buffer[BUF_SIZE];
int rwMode; // READ or WRITE
} PER_IO_DATA, *LPPER_IO_DATA;
DWORD WINAPI EchoThreadMain(LPVOID CompletionPortIo);
void ErrorHandling(char *message);
int main(int argc, char *argv[])
{
WSADATA wsaData;
HANDLE hComPort;
SYSTEM_INFO sysInfo;
LPPER_IO_DATA ioInfo;
LPPER_HANDLE_DATA handleInfo;
SOCKET hServSock;
SOCKADDR_IN servAdr;
DWORD recvBytes, flags = 0;
if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0)
ErrorHandling("WSAStartup() error!");
hComPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0); // 创建 CP 对象
GetSystemInfo(&sysInfo); // 获得当前系统的信息
// 创建线程数等于 CPU 数量
for (int i = 0; i < sysInfo.dwNumberOfProcessors; i++)
_beginthreadex(NULL, 0, (LPVOID)EchoThreadMain, (LPVOID)hComPort, 0, NULL);
hServSock = WSASocket(AF_INET, SOCK_STREAM, 0, NULL, 0, WSA_FLAG_OVERLAPPED);
memset(&servAdr, 0, sizeof(servAdr));
servAdr.sin_family = AF_INET;
servAdr.sin_addr.s_addr = htonl(INADDR_ANY);
servAdr.sin_port = htons(atoi(argv[1]));
if (bind(hServSock, (SOCKADDR *)&servAdr, sizeof(servAdr)) == SOCKET_ERROR)
ErrorHandling("bind() error");
if (listen(hServSock, 5) == SOCKET_ERROR)
ErrorHandling("listen() error");
while (1)
{
SOCKET hClntSock;
SOCKADDR_IN clntAdr;
int addrLen = sizeof(clntAdr);
hClntSock = accept(hServSock, (SOCKADDR *)&clntAdr, &addrLen);
if (hClntSock == INVALID_SOCKET)
{
if (WSAGetLastError() == WSAEWOULDBLOCK)
continue;
else
ErrorHandling("accept() error");
}
handleInfo = (LPPER_HANDLE_DATA)malloc(sizeof(PER_HANDLE_DATA));
handleInfo->hClntSock = hClntSock;
memcpy(&(handleInfo->clntAdr), &clntAdr, addrLen);
CreateIoCompletionPort((HANDLE)hClntSock, hComPort, (ULONG_PTR)handleInfo, 0); // 建立连接
ioInfo = (LPPER_IO_DATA)malloc(sizeof(PER_IO_DATA));
memset(&(ioInfo->overlapped), 0, sizeof(OVERLAPPED));
ioInfo->wsaBuf.len = BUF_SIZE;
ioInfo->wsaBuf.buf = ioInfo->buffer;
ioInfo->rwMode = READ;
WSARecv(handleInfo->hClntSock, &(ioInfo->wsaBuf), 1, &recvBytes, &flags, &(ioInfo->overlapped), NULL);
}
closesocket(hServSock);
WSACleanup();
return 0;
}
DWORD WINAPI EchoThreadMain(LPVOID pComPort)
{
HANDLE hComPort = (HANDLE)pComPort;
SOCKET sock;
DWORD bytesTrans;
LPPER_HANDLE_DATA handleInfo;
LPPER_IO_DATA ioInfo;
DWORD flags = 0;
while (1)
{
GetQueuedCompletionStatus(hComPort, &bytesTrans, (PULONG_PTR)&handleInfo, (LPOVERLAPPED *)&ioInfo, INFINITE);
sock = handleInfo->hClntSock;
if (ioInfo->rwMode == READ)
{
puts("message received!");
if (bytesTrans == 0)
{ // 传输 EOF 时
closesocket(sock);
free(handleInfo);
free(ioInfo);
continue;
}
memset(&(ioInfo->overlapped), 0, sizeof(OVERLAPPED));
ioInfo->wsaBuf.len = bytesTrans;
ioInfo->rwMode = WRITE;
WSASend(sock, &(ioInfo->wsaBuf), 1, NULL, 0, &(ioInfo->overlapped), NULL);
ioInfo = (LPPER_IO_DATA)malloc(sizeof(PER_IO_DATA));
memset(&(ioInfo->overlapped), 0, sizeof(OVERLAPPED));
ioInfo->wsaBuf.len = BUF_SIZE;
ioInfo->wsaBuf.buf = ioInfo->buffer;
ioInfo->rwMode = READ;
WSARecv(sock, &(ioInfo->wsaBuf), 1, NULL, &flags, &(ioInfo->overlapped), NULL);
}
else
{
puts("message sent!");
free(ioInfo);
}
}
return 0;
}
void ErrorHandling(char *message)
{
fputs(message, stderr);
fputc('\n', stderr);
exit(1);
}
编译:
gcc IOCPEchoServ_win.c -lws2_32 -o IOCPEchoServ
运行结果:
IOCP 性能更优的原因
- 因为是非阻塞模式的 I/O,所以不会由 I/O 引发延迟。
- 查找已完成I/O时无需添加循环。
- 无需将作为 I/O 对象的套接字句柄保存到数组进行管理。
- 可以调整处理 I/O 的线程数,所以可在实验数据的基础上选用合适的线程数。
IOCP 是 Windows 特有的功能,所以很大程度上要归功于操作系统。
习题
(1)完成端口对象将分配多个线程用于处理 I/O。如何创建这些线程?如何分配?请从源代码级别进行说明。
首先使用 CreateIoCompletionPort 函数创建 IOCP 对象:
HANDLE hCompletionPort = CreateIoCompletionPort(
INVALID_HANDLE_VALUE, // 首次创建时使用此参数
NULL,
0, // 初始线程数(实际忽略)
0 // 并发线程数(通常为 CPU 核心数)
);
然后将 FileHandle 句柄指向的套接字和 CP 对象相连:
HANDLE hDevice = CreateFile(...); // 或 socket()
CreateIoCompletionPort(
hDevice, // 要关联的句柄
hCompletionPort, // 目标 IOCP
(ULONG_PTR)key, // 自定义标识(如会话上下文)
0 // 并发线程数(继承 IOCP 的设置)
);
调用 CreateIoCompletionPort 函数后,只要针对 hSock 的 I/O 完成,相关信息就将注册到 hCpObject 指向的 CP 对象。
每个线程调用 GetQueuedCompletionStatus 等待 I/O 事件:
DWORD WINAPI WorkerThreadFunction(LPVOID lpParam) {
HANDLE hCompletionPort = (HANDLE)lpParam;
DWORD dwBytesTransferred;
ULONG_PTR completionKey;
LPOVERLAPPED pOverlapped;
while (true) {
BOOL status = GetQueuedCompletionStatus(
hCompletionPort,
&dwBytesTransferred,
&completionKey,
&pOverlapped,
INFINITE // 无限等待事件
);
if (!status) {
// 处理错误或连接关闭
if (pOverlapped == NULL) break; // 收到退出信号
DWORD err = GetLastError();
HandleError(err);
continue;
}
// 处理完成事件
HandleIOCompletion(completionKey, dwBytesTransferred, pOverlapped);
}
return 0;
}
(2)CreateIoCompletionPort 函数与其他函数不同,提供 2 种功能。请问是哪 2 种?
- 创建“完成端口”对象。
- 建立“完成端口”对象和套接字之间的联系。
(3)完成端口对象和套接字之间的连接意味着什么?如何连接?
意味着将已完成的套接字 I/O 信息注册到 CP 对象,使得该套接字上发生的异步操作(如读写操作)的完成事件能够通过完成端口机制进行通知和调度。通过CreateIoCompletionPort 函数实现。
下面以建立连接为目的再次介绍 CreateIoCompletionPort 函数。
#include <windows.h>
HANDLE CreateIoCompletionport(
HANDLE FileHandle,
HANDLE ExistingCompletionPort,
ULONG_PTR Completionkey,
DWORD NumberofConcurrentThreads
);
参数:
- FileHandle:要连接到 CP 对象的套接字句柄。
- ExistingCompletionPort:要连接套接字的 CP 对象句柄。
- CompletionKey:传递已完成 I/O 相关信息,关于该参数将在稍后介绍的 GetQueuedCompletionStatus 函数中共同讨论。
- NumberOfConcurrentThreads:无论传递何值,只要该函数的第二个参数非 NULL 就会忽略。
成功时返回 CP 对象句柄,失败时返回 NULL。
该函数的调用方式如下:
HANDLE hCpObject;
SOCKET hSock;
......
CreateIoCompletionPort((HANDLE)hSock, hCpObject, (DWORD)ioInfo, 0);
调用 CreateIoCompletionPort 函数后,只要针对 hSock 的 I/O 完成,相关信息就将注册到 hCpObject 指向的 CP 对象。
(4)下列关于 IOCP 的说法错误的是?
a. 以最少的线程处理多数 I/O 的结构,因此可以减少上下文切换引起的性能低下。
b. 执行 I/O 的过程中,服务器端无需等待I/O完成,可以执行其他任务,故能提高 CPU 效率。
c. I/O 完成时会自动调用相关 Completion Routine 函数,因此没必要调用特定函数以等待 I/O 完成。
d. 除 Windows 外,其他操作系统同样支持 IOCP,所以这种模型具有良好的移植性。
答:
c、d。
(5)判断下列关于 IOCP 中选择合理线程数的方法是否合适。
- 通常会选择与 CPU 数同样数量的线程。(√)
- 最好在条件允许的范围内通过实验决定线程数。(√)
- 分配的线程数越多越好。例如,1 个线程就足够多的情况下应该多分配几个,比如创建 3 个线程分配给 IOCP。(×)
(6)利用本章的 IOCP 模型实现聊天服务器端,该聊天服务器端应当结合第 20 章的聊天客户端 chat_clnt_win.c 正常运行。编写程序时不必刻意套用本章 IOCP 示例中的框架,那样反而会加大实现难度。
服务器端代码:
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <process.h>
#include <winsock2.h>
#include <windows.h>
#define BUF_SIZE 120
#define MAX_CLNT 256
#define READ 3
#define WRITE 5
typedef struct // socket info
{
SOCKET hClntSock;
SOCKADDR_IN clntAdr;
char name[20];
} PER_CLIENT_INFO, *LPPER_CLIENT_INFO;
typedef struct // buffer info
{
OVERLAPPED overlapped;
WSABUF wsaBuf;
char buffer[BUF_SIZE];
int rwMode; // READ or WRITE
} PER_IO_DATA, *LPPER_IO_DATA;
DWORD WINAPI WorkerThread(LPVOID pComPort);
void SendMsg(char *msg, int len);
void ErrorHandling(char *msg);
int clntCnt = 0;
PER_CLIENT_INFO clntSocks[MAX_CLNT];
HANDLE hMutex;
int main(int argc, char *argv[])
{
WSADATA wsaData;
HANDLE hComPort;
SYSTEM_INFO sysInfo;
LPPER_IO_DATA ioInfo;
LPPER_CLIENT_INFO clntInfo;
SOCKET hServSock;
SOCKADDR_IN servAdr;
DWORD recvBytes, flags = 0;
if (argc != 2)
{
printf("Usage : %s <port>\n", argv[0]);
exit(1);
}
if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0)
ErrorHandling("WSAStartup() error!");
hMutex = CreateMutex(NULL, FALSE, NULL);
// 创建 CP 对象
hComPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);
// 获得当前系统的信息
GetSystemInfo(&sysInfo);
// 创建线程数等于 CPU 数量
for (int i = 0; i < sysInfo.dwNumberOfProcessors; i++)
_beginthreadex(NULL, 0, (LPVOID)WorkerThread, (LPVOID)hComPort, 0, NULL);
hServSock = WSASocket(AF_INET, SOCK_STREAM, 0, NULL, 0, WSA_FLAG_OVERLAPPED);
memset(&servAdr, 0, sizeof(servAdr));
servAdr.sin_family = AF_INET;
servAdr.sin_addr.s_addr = htonl(INADDR_ANY);
servAdr.sin_port = htons(atoi(argv[1]));
if (bind(hServSock, (SOCKADDR *)&servAdr, sizeof(servAdr)) == SOCKET_ERROR)
ErrorHandling("bind() error");
if (listen(hServSock, 5) == SOCKET_ERROR)
ErrorHandling("listen() error");
// 接受客户端连接
while (1)
{
SOCKET hClntSock;
SOCKADDR_IN clntAdr;
int addrLen = sizeof(clntAdr);
hClntSock = accept(hServSock, (SOCKADDR *)&clntAdr, &addrLen);
if (hClntSock == INVALID_SOCKET)
{
if (WSAGetLastError() == WSAEWOULDBLOCK)
continue;
else
ErrorHandling("accept() error");
}
WaitForSingleObject(hMutex, INFINITE);
clntSocks[clntCnt++].hClntSock = hClntSock;
ReleaseMutex(hMutex);
clntInfo = (LPPER_CLIENT_INFO)malloc(sizeof(PER_CLIENT_INFO));
clntInfo->hClntSock = hClntSock;
memcpy(&(clntInfo->clntAdr), &clntAdr, addrLen);
CreateIoCompletionPort((HANDLE)hClntSock, hComPort, (ULONG_PTR)clntInfo, 0); // 建立连接
ioInfo = (LPPER_IO_DATA)malloc(sizeof(PER_IO_DATA));
memset(&(ioInfo->overlapped), 0, sizeof(OVERLAPPED));
ioInfo->wsaBuf.len = BUF_SIZE;
ioInfo->wsaBuf.buf = ioInfo->buffer;
ioInfo->rwMode = READ;
WSARecv(clntInfo->hClntSock, &(ioInfo->wsaBuf), 1, &recvBytes, &flags, &(ioInfo->overlapped), NULL);
}
closesocket(hServSock);
WSACleanup();
return 0;
}
DWORD WINAPI WorkerThread(LPVOID pComPort)
{
HANDLE hComPort = (HANDLE)pComPort;
SOCKET sock;
DWORD bytesTrans;
LPPER_CLIENT_INFO clntInfo;
LPPER_IO_DATA ioInfo;
DWORD flags = 0;
while (1)
{
GetQueuedCompletionStatus(hComPort, &bytesTrans, (PULONG_PTR)&clntInfo, (LPOVERLAPPED *)&ioInfo, INFINITE);
sock = clntInfo->hClntSock;
if (ioInfo->rwMode == READ)
{
if (bytesTrans == 0)
{ // 传输 EOF 时
closesocket(sock);
free(clntInfo);
free(ioInfo);
continue;
}
memset(&(ioInfo->overlapped), 0, sizeof(OVERLAPPED));
ioInfo->wsaBuf.len = bytesTrans;
ioInfo->rwMode = WRITE;
SendMsg(ioInfo->buffer, bytesTrans); // broadcast
ioInfo = (LPPER_IO_DATA)malloc(sizeof(PER_IO_DATA));
memset(&(ioInfo->overlapped), 0, sizeof(OVERLAPPED));
ioInfo->wsaBuf.len = BUF_SIZE;
ioInfo->wsaBuf.buf = ioInfo->buffer;
ioInfo->rwMode = READ;
WSARecv(sock, &(ioInfo->wsaBuf), 1, NULL, &flags, &(ioInfo->overlapped), NULL);
}
else
{
free(ioInfo);
}
}
return 0;
}
void SendMsg(char *msg, int len)
{ // 发送给全部人
int i;
WaitForSingleObject(hMutex, INFINITE);
for (i = 0; i < clntCnt; i++)
send(clntSocks[i].hClntSock, msg, len, 0);
ReleaseMutex(hMutex);
}
void ErrorHandling(char *msg)
{
fputs(msg, stderr);
fputc('\n', stderr);
exit(1);
}
编译:
gcc IOCPChatServ.c -lws2_32 -o IOCPChatServ
运行结果: