《TCP/IP网络编程》学习笔记 | Chapter 21：异步通知 I/O 模型

《TCP/IP网络编程》学习笔记 | Chapter 21：异步通知 I/O 模型

同步与异步

同步

同步指的是任务按照顺序执行，一个任务必须等待前一个任务完成后才能继续。换句话说，在进行同步操作时，程序会阻塞在某个操作上，直到该操作完成后才会继续往下执行。

同步方式的数据 I/O：

可以通过下图解析上述两句话的含义：

异步

异步是指任务并行执行，程序发起一个操作后不等待其完成，而是继续执行其他任务。程序可以在等待操作完成的同时执行其他操作，最终通过回调、事件、信号等方式获取操作结果。

异步 I/O 是指 I/O 函数的返回时刻与数据接收的完成时刻不一致。

对比

同步 I/O 的缺点

进行 I/O 的过程中函数无法返回，所以不能执行其他任务。

异步 I/O 的优点

无论数据是否完成交换都返回函数，这就意味着可以执行其他任务。

异步方式能够比同步方式更有效使用 CPU。

理解异步通知 I/O 模型

通知 I/O 的含义：

顾名思义，通知 I/O 是指发生了I/O相关的特定情况。典型的通知 I/O 模型是 select 方式，但这种通知是以同步方式进行的，原因在于，需要 I/O 或可以进行 I/O 的时间点（简言之就是 I/O 相关事件发生的时间点）与 select 函数的返回时间点一致。

异步通知 I/O 模型意为通知 I/O 是以异步方式工作的。与“select 函数只在需要或可以进行 I/O 的情况下返回”不同，异步通知 I/O 模型中函数的返回与 I/O 状态无关。

本章的 WSAEventSelect 函数就是 select 函数的差异版本。

可能有人疑问：“既然函数的返回与I/O状态无关，那是否需要监视 I/O 状态变化？”

当然需要！异步通知 I/O 中，指定 I/O 监视对象的函数和实际验证状态变化的函数是相互分离的。因此，指定监视对象后可以离开执行其他任务，最后再回来验证状态变化。

实现异步通知 I/O 模型

异步通知 I/O 模型的实现方法有 2 种：一种是稍后介绍的 WSAEventSelect 函数，第二种是使用 WSAAsyncSelect 函数，第二种方法是 UI 相关内容，不进行介绍，需要了解可自行查阅资料。

WSAEventSelect 函数和通知

如前所述，告知 I/O 状态变化的操作就是“通知”。I/O的状态变化可以分为不同情况：

• 套接字的状态变化：套接字的I/O状态变化。
• 发生套接字相关事件：发生套接字I/O相关事件。

这 2 种情况都意味着发生了需要或可以进行 I/O 的事件，我将根据上下文适当混用这些概念。

WSAEventSelect 是 Windows Sockets API（Winsock）中的一个函数，用于将一个 Windows 事件对象与一个套接字（socket）关联，以便在套接字状态发生变化时获得通知。这个函数常用于异步网络编程，特别是在处理多个套接字时。

#include<winsock2.h>int WSAEventSelect(SOCKET s,HANDLE hEvent,long lNetworkEvents
);

参数：

• s：监视对象的套接字句柄。
• hEventObject：传递事件对象句柄以验证事件发生与否。
• INetworkEvents：希望监视的事件类型信息。

成功时返回 0，失败时返回 SOCKET_ERROR。

传入参数 s 的套接字内只要发生 INetworkEvents 中指定的事件之一，WSAEventSelect 函数就将 hEventObject 句柄所指内核对象改为 signaled 状态。因此，该函数又称“连接事件对象和套接字的函数”。该函数以异步通知方式工作。无论事件发生与否，WSAEventSelect 函数调用后都会直接返回。

下面介绍作为该函数第三个参数的事件类型信息，可以通过位或运算同时指定多个信息。

• FD_READ：是否存在需要接收的数据？
• FD_WRITE：能否以非阻塞方式传输数据？
• FD_OOB：是否收到带外数据？
• FD_ACCEPT：是否有新的连接请求？
• FD_CLOSE：是否有断开连接的请求？

以上就是 WSAEventSelect 函数的调用方法。

仅从概念上看，WSAEventSelect函数 的功能偏弱。但使用该函数时，没必要针对多个套接字进行调用。从 select 函数返回时，为了验证事件的发生需要再次针对所有句柄调用函数，但通过调用 WSAEventSelect 函数传递的套接字信息已注册到操作系统，所以无需再次调用。这反而是 WSAEventSelect 函数比 select 函数的优势所在。

从函数说明中可以看出，我们还需要知道以下内容：

1. WSAEventSelect 函数的第二个参数中用到的事件对象的创建方法。
2. 调用 WSAEventSelect 函数后发生事件的验证方法。
3. 验证事件发生后事件类型的查看方法。

manual-reset 模式事件的其他创建方法

之前创建事件对象是利用 CreateEvent 函数。 CreateEvent 函数在创建事件对象时，可以在 auto-reset 模式和 manual-reset 模式中任选其一。但是我们只需要 manual-reset 模式 non-signaled 状态的事件对象，所以利用下面的函数创建比较方便。

#include <winsock2.h>WSAEVENT WSACreateEvent(void);

成功时返回事件对象句柄，失败时返回 WSA_INVALID_EVENT。

上述声明中返回类型 WSAEVENT 的定义如下：

#define WSAEVENT HANDLE

实际上就是我们熟悉的内核对象句柄。

另外，可使用如下函数销毁上述函数创建的事件对象：

#include <winsock2.h>BOOL WSACloseEvent(WSAEVENT hEvent);

成功时返回 TRUE， 失败时返回 FALSE。

验证是否发生事件

为了验证是否发生事件，需要查看事件对象。完成该任务的函数如下，除了多 1 个参数外，其余部分与 WaitForMultipleObjects 函数完全相同。

#include <winsock2.h>DWORD WSAWaitForMultipleEvents(DWORD cEvents,                // 要等待的事件对象数量const WSAEVENT* lphEvents,    // 指向事件对象句柄数组的指针BOOL fWaitAll,                // 指定是等待所有事件还是任何一个事件DWORD dwTimeout,              // 超时时间（毫秒）BOOL fAlertable               // 指定等待期间是否允许进入警报状态（APC）
);

参数：

• cEvents：需要验证是否转为 signaled 状态的事件对象的个数。
• IphEvents：存有事件对象句柄数组地址值。
• fWaitAll：传递 TRUE 时，所有事件对象在 signaled 状态时返回；传递 FALSE 时，只要其中 1 个变为 signaled 状态就返回。
• dwTimeout：以 1ms 为单位指定超时，传递 WSA_INFINITE 时，直到变为 signaled 状态时才会返回。
• fAlertable：传递 TRUE 时进入 alertable_wait（可警告等待）状态（第 22 章）。

返回值减去常量 WSA_WAIT_EVENT_0 时，可以得到转变为 signaled 状态的事件对象句柄对应的索引，可以通过该索引在第二个参数指定的数组中查找句柄。如果有多个事件对象变为 signaled 状态，则会得到其中较小的值。发生超时将返回 WSA_WAIT_TIMEOUT。

可以通过以宏的方式声明的 WSA_MAXIMUM_WAIT_EVENTS 常量得知 WSAWaitForMultipleEvents 函数可以同时监视的最大事件对象数。该常量值为 64。

只通过 1 次函数调用无法得到转为 signaled 状态的所有事件对象句柄的信息。通过该函数可以得到转为 signaled 状态的事件对象中的第一个（按数组中的保存顺序）索引值。但可以利用 “事件对象为 manual-reset 模式” 的特点，通过如下方式获得所有 signaled 状态的事件对象。

int posInfo, startIdx, i;
...
// 等待 hEventArray 中的任意一个事件对象变为 signaled 状态
posInfo = WSAWaitForMultipleEvents(numOfSock, hEventarray, FALSE, WSA_INFINITE, FALSE);
// startIdx 是当前进入 signaled 状态的事件在 hEventArray 数组中的索引
startIdx = posInfo - WSA_WAIT_EVENT_0;
...
// 从第一个 signaled 的事件开始，遍历剩余的事件，检查它们是否已经被触发
for(i = startIdx; i < numOfSock; i++)
{// 仅检查 hEventArray[i] 这个单个事件对象是否被触发。由于超时时间为 0，它不会等待，只会立即返回结果int sigEventIdx = WSAWaitForMultipleEvents(1, &hEventArray[i], TRUE, 0, FALSE);...// 如果这个事件已经 signaled，则 sigEventIdx 返回 WSA_WAIT_EVENT_0// 否则返回 WSA_WAIT_TIMEOUT 或 WSA_WAIT_FAILED
}

之所以能做到这一点，完全是因为事件对象为 manual-reset 模式，这也解释了为何在异步通知 I/O 模型中事件对象必须为 manual-reset 模式。

区分事件类型

既然已经通过WSAWaitForMultipleEvents函数得到了转为signaled状态的事件对象，最后就要确定相应对象进入signaled状态的原因。为完成该任务，我们引入如下函数。调用此函数时，不仅需要signaled状态的事件对象句柄，还需要与之连接的（由WSAEventSelect函数调用引发的）发生的套接字句柄。

#include <winsock2.h>// 
int WSAEnumNetworkEvents(SOCKET s,                           // 套接字句柄WSAEVENT hEventObject,              // 可选的事件对象句柄LPWSANETWORKEVENTS lpNetworkEvents  // 指向 WSANETWORKEVENTS 结构的指针
);

参数：

• s：发生事件的套接字句柄。该套接字应该与某个事件对象通过 WSAEventSelect 函数关联。
• hEventObject：一个可选的事件对象句柄。如果这个参数非空，WSAEnumNetworkEvents 函数会自动将这个事件对象的状态重置为 non-signaled 状态。 如果传递的是 NULL，那么该函数仅仅返回网络事件信息，而不重置任何事件。
• lpNetworkEvents：指向 WSANETWORKEVENTS 结构的指针，用于接收套接字的事件信息。此结构包含了发生的网络事件及其对应的错误码。

WSANETWORKEVENTS 结构体：

typedef struct _WSANETWORKEVENTS
{long lNetworkEvents;  // 事件的掩码（可以是多个事件的组合）int iErrorCode[FD_MAX_EVENTS];  // 每个事件对应的错误码
} WSANETWORKEVENTS, *LPWSANETWORKEVENTS;

上述结构体的 lNetworkEvents 成员将保存发生的事件信息，与 WSAEventSelect 函数的第三个参数相同，需要接收数据时，该成员为 FD_READ；有连接请求时，该成员为 FD_ACCEPT。因此，可通过如下方式查看发生的事件类型：

WSANETWORKEVENTS netEvents;
...
WSAEnumNetworkEvents(hSock, hEvent, &netEvents);
if(netEvents.lNetworkEvents & FD_ACCEPT)
{// FD_ACCEPT 事件的处理
}if(netEvents.lNetworkEvents & FD_READ)
{// FD_READ 事件的处理
}if(netEvents.lNetworkEvents & FD_CLOSE)
{// FD_CLOSE 事件的处理
}

另外，错误信息将保存到声明为成员的 iErrorCode 数组(发生错误的原因可能很多，因此用数组声明)。验证方法如下：

• 如果发生 FD_READ 相关错误，则在 iErrorCode[FD_READ_BIT]中保存除 0 以外的其他值。
• 如果发生 FD_WRITE 相关错误，则在 iErrorCode[FD_WRITE_BIT] 中保存除 0 以外的其他值。

可以通过以下的描述来理解上述内容：“如果发生 FD_XXX 相关错误，则在 iErrorCode[FD_XXX_BIT] 中保存除 0 以外的其他值”。

因此可以用如下方式检查错误：

WSANETWORKEVENTS netEvents;
...
WSAEnumNetworkEvents(hSock,hEvent,&netEvents);
...
if(netEvents.iErrorCode[FD_READ_BIT] != 0
{// 发生 FD_READ 事件相关错误
}

利用异步通知 I/O 模型实现回声服务器端

由于代码较长，所以将分成多个部分进行介绍。

#include <stdio.h>
#include <string.h>
#include <winsock2.h>#define BUF_SIZE 100void CompressSockets(SOCKET hSockArr[], int idx, int total);
void CompressEvents(WSAEVENT hEventArr[], int idx, int total);
void ErrorHandling(char *msg);int main(int argc, char *argv[])
{WSADATA wsaData;SOCKET hServSock, hClntSock;SOCKADDR_IN servAdr, clntAdr;SOCKET hSockArr[WSA_MAXIMUM_WAIT_EVENTS];WSAEVENT hEventArr[WSA_MAXIMUM_WAIT_EVENTS];WSAEVENT newEvent;WSANETWORKEVENTS netEvents;int numOfClntSock = 0;int strLen, i;int posInfo, startIdx;int clntAdrLen;char msg[BUF_SIZE];if (argc != 2){printf("Usage: %s <port>\n", argv[0]);exit(1);}if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0)ErrorHandling("WSAStartup() error!");

以上是初始化和声明代码，没有什么要特别说明的。

    hServSock = socket(PF_INET, SOCK_STREAM, 0);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");newEvent = WSACreateEvent();if (WSAEventSelect(hServSock, newEvent, FD_ACCEPT) == SOCKET_ERROR)ErrorHandling("WSAEventSelect() error");hSockArr[numOfClntSock] = hServSock;hEventArr[numOfClntSock] = newEvent;numOfClntSock++;

上述代码创建了用于接收客户端连接请求的服务器端套接字。为了完成监听任务，针对 FD_ACCEPT 事件调用了 WSAEventSelect 函数。此处需要注意如下 2 条语句。

    hSockArr[numOfClntSock] = hServSock;hEventArr[numOfClntSock] = newEvent;

这段代码把通过 WSAEventSelect 函数连接的套接字和事件对象的句柄分别存入 hSockArr 和 hEventArr 数组。也就是说，应该可以通过 hSockArr[idx] 找到连接到套接字的事件对象，反之，也可以通过 hEventArr[idx] 找到连接到事件对象的套接字。因此，该示例将套接字和事件对象句柄保存到数组时统一了保存位置。也就有了下列公式：

• 与 hSockArr[n] 中的套接字相连的事件对象应保存到 hEventArr[n]。
• 与 hEventArr[n] 中的事件对象相连的套接字应保存到 hSockArr[n]。

接下来是 while 循环部分，之前学习的大部分内容都在此。

    while (1){posInfo = WSAWaitForMultipleEvents(numOfClntSock, hEventArr, FALSE, WSA_INFINITE, FALSE);startIdx = posInfo - WSA_WAIT_EVENT_0;for (i = startIdx; i < numOfClntSock; i++){int sigEventIdx =WSAWaitForMultipleEvents(1, &hEventArr[i], TRUE, 0, FALSE);if ((sigEventIdx == WSA_WAIT_FAILED || sigEventIdx == WSA_WAIT_TIMEOUT)){continue;}else{sigEventIdx = i;WSAEnumNetworkEvents(hSockArr[sigEventIdx], hEventArr[sigEventIdx], &netEvents);if (netEvents.lNetworkEvents & FD_ACCEPT) // 请求连接时{if (netEvents.iErrorCode[FD_ACCEPT_BIT] != 0){puts("Accept Error");break;}clntAdrLen = sizeof(clntAdr);hClntSock = accept(hSockArr[sigEventIdx], (SOCKADDR *)&clntAdr, &clntAdrLen);newEvent = WSACreateEvent();WSAEventSelect(hClntSock, newEvent, FD_READ | FD_CLOSE);hEventArr[numOfClntSock] = newEvent;hSockArr[numOfClntSock] = hClntSock;numOfClntSock++;puts("connected new client...");}if (netEvents.lNetworkEvents & FD_READ) // 接收数据时{if (netEvents.iErrorCode[FD_READ_BIT] != 0){puts("Read Error");break;}strLen = recv(hSockArr[sigEventIdx], msg, sizeof(msg), 0);send(hSockArr[sigEventIdx], msg, strLen, 0);}if (netEvents.lNetworkEvents & FD_CLOSE) // 断开连接时{if (netEvents.iErrorCode[FD_CLOSE_BIT] != 0){puts("Close Error");break;}WSACloseEvent(hEventArr[sigEventIdx]);closesocket(hSockArr[sigEventIdx]);numOfClntSock--;CompressSockets(hSockArr, sigEventIdx, numOfClntSock);CompressEvents(hEventArr, sigEventIdx, numOfClntSock);}}}}WSACleanup();return 0;
}

最后给出上述代码调用的两个函数 CompressSockets 和 CompressEvents 的函数声明。

void CompressSockets(SOCKET hSockArr[], int idx, int total)
{int i;for (i = idx; i < total; i++)hSockArr[i] = hSockArr[i + 1];
}void CompressEvents(WSAEVENT hEventArr[], int idx, int total)
{int i;for (i = idx; i < total; i++)hEventArr[i] = hEventArr[i + 1];
}void ErrorHandling(char *msg)
{fputs(msg, stderr);fputc('\n', stderr);exit(1);
}

完整的服务器端代码：

#include <stdio.h>
#include <string.h>
#include <winsock2.h>#define BUF_SIZE 100void CompressSockets(SOCKET hSockArr[], int idx, int total);
void CompressEvents(WSAEVENT hEventArr[], int idx, int total);
void ErrorHandling(char *msg);int main(int argc, char *argv[])
{WSADATA wsaData;SOCKET hServSock, hClntSock;SOCKADDR_IN servAdr, clntAdr;SOCKET hSockArr[WSA_MAXIMUM_WAIT_EVENTS];WSAEVENT hEventArr[WSA_MAXIMUM_WAIT_EVENTS];WSAEVENT newEvent;WSANETWORKEVENTS netEvents;int numOfClntSock = 0;int strLen, i;int posInfo, startIdx;int clntAdrLen;char msg[BUF_SIZE];if (argc != 2){printf("Usage: %s <port>\n", argv[0]);exit(1);}if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0)ErrorHandling("WSAStartup() error!");hServSock = socket(PF_INET, SOCK_STREAM, 0);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");newEvent = WSACreateEvent();if (WSAEventSelect(hServSock, newEvent, FD_ACCEPT) == SOCKET_ERROR)ErrorHandling("WSAEventSelect() error");hSockArr[numOfClntSock] = hServSock;hEventArr[numOfClntSock] = newEvent;numOfClntSock++;while (1){posInfo = WSAWaitForMultipleEvents(numOfClntSock, hEventArr, FALSE, WSA_INFINITE, FALSE);startIdx = posInfo - WSA_WAIT_EVENT_0;for (i = startIdx; i < numOfClntSock; i++){int sigEventIdx =WSAWaitForMultipleEvents(1, &hEventArr[i], TRUE, 0, FALSE);if ((sigEventIdx == WSA_WAIT_FAILED || sigEventIdx == WSA_WAIT_TIMEOUT)){continue;}else{sigEventIdx = i;WSAEnumNetworkEvents(hSockArr[sigEventIdx], hEventArr[sigEventIdx], &netEvents);if (netEvents.lNetworkEvents & FD_ACCEPT) // 请求连接时{if (netEvents.iErrorCode[FD_ACCEPT_BIT] != 0){puts("Accept Error");break;}clntAdrLen = sizeof(clntAdr);hClntSock = accept(hSockArr[sigEventIdx], (SOCKADDR *)&clntAdr, &clntAdrLen);newEvent = WSACreateEvent();WSAEventSelect(hClntSock, newEvent, FD_READ | FD_CLOSE);hEventArr[numOfClntSock] = newEvent;hSockArr[numOfClntSock] = hClntSock;numOfClntSock++;puts("connected new client...");}if (netEvents.lNetworkEvents & FD_READ) // 接收数据时{if (netEvents.iErrorCode[FD_READ_BIT] != 0){puts("Read Error");break;}strLen = recv(hSockArr[sigEventIdx], msg, sizeof(msg), 0);send(hSockArr[sigEventIdx], msg, strLen, 0);}if (netEvents.lNetworkEvents & FD_CLOSE) // 断开连接时{if (netEvents.iErrorCode[FD_CLOSE_BIT] != 0){puts("Close Error");break;}WSACloseEvent(hEventArr[sigEventIdx]);closesocket(hSockArr[sigEventIdx]);numOfClntSock--;CompressSockets(hSockArr, sigEventIdx, numOfClntSock);CompressEvents(hEventArr, sigEventIdx, numOfClntSock);}}}}WSACleanup();return 0;
}void CompressSockets(SOCKET hSockArr[], int idx, int total)
{int i;for (i = idx; i < total; i++)hSockArr[i] = hSockArr[i + 1];
}void CompressEvents(WSAEVENT hEventArr[], int idx, int total)
{int i;for (i = idx; i < total; i++)hEventArr[i] = hEventArr[i + 1];
}void ErrorHandling(char *msg)
{fputs(msg, stderr);fputc('\n', stderr);exit(1);
}

客户端：

在这里插入代码片

编译：

在这里插入代码片

运行结果：

习题

（1）结合 send & recv 函数解释同步和异步方式的 I/O。并请说明同步 I/O 的缺点，以及怎样通过异步 I/O 进行解决。

（2）异步 I/O 并不是所有情况下的最佳选择。它具有哪些缺点？何种情况下同步 I/O 更优？可以参考异步 I/O 相关源代码，亦可结合线程进行说明。

（3）判断下列关于 select 模型描述的正误。

• select 模型通过函数的返回值通知 I/O 相关事件，故可视为通知 I/O 模型。（√）
• select 模型中 I/O 相关事件的发生时间点和函数返回的时间点一致，故不属于异步模型。（√）
• WSAEventSelect 函数可视为 select 方式的异步模型，因为该函数的 I/O 相关事件的通知方式。（√）

（4）请从源代码的角度说明 select 函数和 WSAEventSelect 函数在使用上的差异

（5）第 17 章的 epoll 可以在条件触发和边缘触发这 2 种方式下工作。哪种方式更适合异步 I/O 模型？为什么？请概括说明。

（6）Linux 中的 epoll 同样属于异步 I/O 模型。请说明原因。

（7）如何获取 WSAWaitForMultipleEvents 函数可以监视的最大句柄数？请编写代码读取该值。

（8）为何异步通知 I/O 模型中的事件对象必须是 manual-reset 模式？

（9）请在本章的通知 I/O 模型的基础上编写聊天服务器端。要求该服务器端能够结合第 20 章的聊天客户端 chat_clnt_win.c 运行。