《TCP/IP网络编程》第21章

• 理解异步通知I/O模型
• • 同步和异步
  • 同步I/O缺点及异步解决方案
  • 异步通知I/O模型
• 实现异步通知I/O模型
• • WSAEventSelect函数
  • manual-reset模式事件对象的创建
  • 验证是否发生了事件
  • 区分事件类型
  • 异步通知I/O模型回声服务器

理解异步通知I/O模型

同步和异步

同步的关键是函数的调用及返回时刻，以及数据传输的开始和完成时刻。

• 调用send函数的瞬间开始传输数据，send函数执行完（返回）的时刻完成数据传输（数据完全传输到输出缓冲，操作系统完成网络传输）。
• 调用recv函数的瞬间开始接收数据，recv函数执行完（返回）的时刻完成数据接收。

异步（Asynchronous），不一致。
异步I/O指I/O函数的返回时刻与数据收发的完成时刻不一致。

同步I/O缺点及异步解决方案

同步I/O的过程中函数无法返回，不能执行其他任务。
异步I/O立即返回函数，可以执行其他任务，更有效地使用CPU。

异步通知I/O模型

异步工作方式“通知I/O”。
通知I/O：通知输入缓冲收到数据并需要读取，以及输出缓冲为空故可以发送数据。

select是同步方式通知I/O，需要I/O或可以进行I/O的时间点（I/O相关事件发生的时间点）与select函数的返回时间点一致。

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

实现异步通知I/O模型

WSAAsyncSelect函数，需指定Windows句柄以获取发生的事件（UI相关内容），本书不会涉及。

WSAEventSelect函数

I/O的状态变化

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

#include <winsock2.h>

//成功0，失败SOCKET_ERROR
int WSAEventSelect(
	SOCKET s, 
	WSAEVENT hEventObject, 	//事件对象句柄
	long lNetworkEvents		//监视的事件类型信息，或运算
);

传入参数s的套接字内只要发生lNetworkEvents中指定事件之一， WSAEventSelect函数就将hEventObject句柄所指内核对象改为signaled状态。该函数是连接事件对象和套接字的函数，以异步通知方式工作，调用后直接返回。

事件类型信息：

• FD_READ：可读
• FD_WRITE：可写
• FD_OOB：收到带外数据
• FD_ACCEPT：新连接
• FD_CLOSE：断开连接

manual-reset模式事件对象的创建

CreateEvent函数可以创建auto-reset或manual-reset模式事件对象。
WSACreateEvent只创建manual-reset模式non-signaled状态的事件对象。

#include <winsock2.h>

//#define WSAEVENT HANDLE

//失败WSA_INVALID_EVENT
WSAEVENT WSACreateEvent(void);

//销毁上述函数创建的事件对象
BOOL WSACloseEvent(WSAEVENT hEvent);

验证是否发生了事件

#include <winsock2.h>

//失败WSA_INVALID_EVENT
DWORD WSAWaitForMultipleEvents(
	DWORD cEvents,//验证是否转为signaled状态的事件对象个数
	const WSAEVENT *lphEvents,//事件对象句柄数组
	BOOL fWaitAll,//TRUE，所有事件对象signaled状态才返回；FALSE，任一事件对象signaled状态就返回
	DWORD dwTimeout,//超时（1/1000秒），WSA_INFINITE一直等待
	BOOL fAlertable//TRUE时进入alertable wait（可警告等待）状态
);
//返回值减去WSA_WAIT_EVENT_0，得到转为signaled状态事件对象句柄对应索引（事件对象句柄数组）。
//多个signaled状态事件对象，得到较小值（调用一次只能获取一个signaled状态事件对象句柄）。
//超时返回WAIT_TIMEOUT

最多可传递64个事件对象，如需监视更多句柄，只能创建线程或扩展保存句柄的数组，并多次调用上述函数。

获取所有signaled状态的事件对象。

int posInfo, startIdx, i;
posInfo = WSAWaitForMultipleEvents(nmuOfSock, hEventArray, FALSE, WSA_INFINITE, FALSE);
startIdx = posInfo-WSA_WAIT_EVENT_0;
for(i=startIdx; i<numOfSock, i++) {
	int sigEventIdx = WSAWaitForMultipleEvents(1, &hEventArray[i], TRUE, 0, FALSE)-WSA_WAIT_EVENT_0;
}

区分事件类型

#include <winsock2.h>

//成功0，失败SOCKET_ERROR
int WSAEnumNetworkEvents(
	SOCKET s,
	WSAEVENT hEventObject, //与套接字关联的signaled状态的事件对象句柄
	LPWSANETWORKEVENTS lpNetworkEvents//保存发生的事件类型信息和错误信息
);

上述函数将manual-reset模式事件对象改为non-signaled状态，无需单独调用ResetEvent函数。

type struct _WSANETWORKEVENTS {
	long lNetworkEvents;
	int iErrorCode[FD_MAX_EVENTS];
} WSANETWORKEVENTS, *LPWSANETWORKEVENTS;

查看发生的事件类型

WSANETWORKEVENTS netEvents;

WSAEnumNetworkEvents(hSock, hEvent, &netEvents);
if(netEvents.lNetworkEvents & FD_ACCEPT) {
	//
}
if(netEvents.lNetworkEvents & FD_READ) {
	//
}
if(netEvents.lNetworkEvents & FD_WRITE) {
	//
}
if(netEvents.lNetworkEvents & FD_CLOSE) {
	//
}

iErrorCode数组保存错误信息。

• FD_READ错误，查看iErrorCode[FD_READ_BIT]
• FD_WRITE错误，查看iErrorCode[FD_WRITE_BIT]
• FD_XXX错误，查看iErrorCode[FD_XXX_BIT]

WSANETWORKEVENTS netEvents;

WSAEnumNetworkEvents(hSock, hEvent, &netEvents);
if(netEvents.iErrorCode[FD_READ_BIT] != 0) {
	//
}

异步通知I/O模型回声服务器

AsyncNotiEchoServ_win.c

#include <stdio.h>
#include <string.h>
#include <windock2.h>

#define BUF_SIZE 100
void CompressSockets(SOCKET hSockArr[], int idx, int total);
void CompressEvents(WSAEVENT hEventArr[], int ind, int total);
void error_handling(char *message);

int main(int argc, char *argv[]) {
	WSADATA wsaData;
	SOCKET serv_sock;
	SOCKET clnt_sock;

	SOCKADDR_IN serv_addr;
	SOCKADDR_IN clnt_addr;
	int addr_size;

	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)
		error_handling("WSAStartup() error!");

	serv_sock = socket(PF_INET, SOCK_STREAM, 0);
	if(serv_sock==INVALID_SOCKET)
		error_handling("socket() error");

	addr_size = sizeof(SOCKADDR_IN);
	
	memset(&serv_addr, 0, addr_size);
	serv_addr.sin_family = AF_INET;
	serv_addr.sin_addr.s_addr = htonl(INADDR_ANY);
	serv_addr.sin_port = htons(atoi(argv[1]));
	
	if(bind(serv_sock, (SOCKADDR*)&serv_addr, addr_size)==SOCKET_ERROR)
		error_handling("bind() error");
		
	if(listen(serv_sock, 5)==SOCKET_ERROR)
		error_handling("listen() error");
		
	newEvent = WSACreateEvent();
	if(WSAEventSelect(serv_sock, newEvent, FD_ACCEPT)==SOCKET_ERROR)
		error_handling("WSAEventSelect() error");
	
	hSockArr[numOfClntSock] = serv_sock;
	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;
				WASEnumNetworkEvents(hSockArr[sigEventIdx], hEventArr[sigEventIdx], &netEvents);
				if(netEvents.lNetworkEvents & FD_ACCCEPT) {
					if(netEvents.iErrorCode[FD_ACCEPT_BIT]!=0) {
						puts("ACCEPT Error");
						break;
					}
					clnt_sock = accept(hSockArr[sigEventIdx], (SOCKADDR*)&clnt_addr, &addr_size);
					newEvent = WSACreateEvent();
					WSAEventSelect(clnt_sock, newEvent, FD_READ|FD_CLOSE);
					
					hSockArr[numOfClntSock] = clnt_sock;
					hEventArr[numOfClntSock] = newEvent;
					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 error_handling(char *message) {
	fputs(message, stderr);
	fputc('\n', stderr);
	exit(1);
}

cl /EHsc AsyncNotiEchoServ_win.c /Fe:AsyncNotiEchoServ_win.exe
AsyncNotiEchoServ_win 9190