【网络编程】完成端口 IOCP

10.11 完成端口

10.11.1 基本概念

完成端口的全称是I/O 完成端口，英文为IOCP(I/O Completion Port) 。IOCP是一个异 步I/O 的 API, 可以高效地将I/O 事件通知给应用程序。与使用select() 或是其他异步方法不同 的是，一个套接字与一个完成端口关联了起来，然后就可以继续进行正常的Winsock 操作了。 然而，当一个事件发生的时候，此完成端口就将被操作系统加入一个队列中。然后应用程序可 以对核心层进行查询以得到此完成端口。

这里我要对上面的一些概念略做补充，在解释“完成”两字之前，想再次复习一下同步和 异步这两个概念，从逻辑上来讲做完一件事后再去做另一件事就是同步，而同时一起做两件或 两件以上的事就是异步了。你也可以拿单线程和多线程来做比喻，但是我们一定要将同步和堵 塞、异步和非堵塞区分开来。

所谓的堵塞函数诸如accept(…), 当调用此函数后，线程将挂起， 直到操作系统通知它，“有人连进来了”,那个挂起的线程将继续进行工作，也就是符合“生 产者-消费者”模型。堵塞和同步看上去有两分相似，但却是完全不同的概念。大家都知道I/O 设备是一个相对慢速的设备，不论打印机、调制解调器还是硬盘，与CPU 相比都是奇慢无比 的，坐下来等I/O 的完成是不明智的，有时候数据的流动率非常惊人，把数据从你的文件服务 器中以Ethernet 速度搬走，其速度可能高达每秒一百万字节。如果你尝试从文件服务器中读取 100KB, 在用户的眼光来看几乎是瞬间完成，但是要知道，你的线程执行这个命令，已经浪费 了10个一百万次CPU 周期。

所以说，我们一般使用另一个线程来进行I/O。重叠IO(overlapped I/O) 是 Win32 的一项技术，你可以要求操作系统为你传送数据，并且在传送完毕时通知你。 这也就是“完成”的含义。这项技术使你的程序在I/O 进行的过程中仍然能够继续处理事务。 事实上，操作系统内部正是以线程来完成overlapped I/O。你可以获得线程所有利益，而不需要付出什么痛苦的代价。

完成端口中所谓的“端口”并不是我们在TCP/IP 中所提到的端口，可以说完全没有关系。 笔者其实也困惑一个I/O 设 备(I/O Device) 和端口(IOCP 中 的Port) 到底有什么关系。IOCP 只不过是用来进行读写操作，和文件I/O 倒是有些类似。既然是一个读写设备，我们所能要求 它的只是在处理读与写上的高效。

接着我们再来探究一下“完成”的含义。首先，它之所以叫“完成”端口，因为系统在网 络I/O 操作“完成”之后才会通知我们。也就是说，我们在接到系统通知的时候，其实网络操 作已经完成了(在系统通知我们的时候，并非是有数据从网络上到来，而是来自于网络上的数 据已经接收完毕了；或者是客户端的连入请求已经被系统接入完毕了，等等),我们只需要处 理后面的事情就好了。

各位同志可能会很开心，什么?已经处理完毕了才通知我们，那岂不是很爽?其实也没什 么爽的，那是因为我们在之前给系统分派工作的时候都嘱咐好了，我们会通过代码告诉系统“你 给我做这个做那个，等待做完了再通知我”,只是这些工作是做在之前还是之后的区别而已。

其次，我们需要知道，所谓的完成端口其实和 HANDLE 一样，也是一个内核对象， Windows 大师Jeff Richter曾说，“完成端口可能是最为复杂的内核对象了”,但是我们也不 用去管它复杂，因为具体的内部是如何实现的和我们无关，只要我们能够学会用它相关的API 把这个完成端口的框架搭建起来就可以了。我们暂时只用把它大体理解为一个容纳网络通信操 作的队列就好了，它会把网络操作完成的通知都放在这个队列里面，咱们只用从这个队列里面 取就行了，取走一个就少一个。

10.11.2 完成端口能干什么

完成端口会主动帮我们完成网络I/O 数据复制。这一点其实也就是他与其他网络模型最直 接的区别了。 一般网络操作包括两个步骤，以recv 来说，如果是一般模型，那么其第一步是 通知等待的线程有数据可以读取，这时线程会调用recv 或 者recvfrom 等函数将数据从读缓冲 区复制到用户空间，然后做下一步的处理，而IOCP 能帮我们的是，它会在内核中帮我们监听 那些我们感兴趣的事件。

例如，我们希望接收客户端数据，那么我们向完成端口投递一个读事 件，完成端口在监测有读事件到来的时候会主动地去帮我们把数据从内存空间复制到用户空 间，然后通知我们过来取数据就可以了，这就是IOCP 提供的方便之处。

另外，IOCP 在内部管理线程，实现负载平衡。上面提到了Windows 的 alertable I/O的 负 载均衡是它的一个弊端，那么IOCP 是如何自己管理线程调度的呢?简单地说就是以栈的方式 进行管理。

10.11.3 完成端口的优势

完成端口会充分利用Windows 内核来进行I/O 的调度，是用于C/S 通信模式中性能最好 的网络通信模型，没有之一；甚至连和它性能接近的通信模型都没有。
微软提出完成端口模型的初衷就是为了解决同步方式那种一个线程处理一个客户端的模式 (one-thread-per-client)缺点的，它充分利用内核对象的调度，只使用少量的几个线程来处 理和客户端的所有通信，消除了无谓的线程上下文切换，最大限度地提高了网络通信的性能。

相比于其他异步模型，对于内存占用都是差不多的，真正的差别就在于CPU 的占用，其他的网络模型都需要更多的CPU 动力来支撑同样的连接数据。

完成端口被广泛地应用于各个高性能服务器程序上，例如著名的Apache 服务器，如果你 想要编写的服务器端需要同时处理的并发客户端连接数量有数百上千个，那不用纠结了，就是 它 了 。

总而言之，完成端口的优势就是效率高。在完成端口模型中，我们会实现开好几个线程， 一般是有多少个CPU 就开多少个线程(其实一般是CPU*2 个 ) 。 建 立CPU*2 个线程的好处 是，在一个工作线程被Sleep( 或 者WaitForSingleObject() 被停止的情况下，IOCP 能唤醒同在 一 个CPU 上的另一个线程代替这个Sleep的线程继续执行，这样完成端口就实现了CPU 的 满 负荷工作，效率也就高了。

这样做的好处是可以避免线程的上下文切换。然后让这几个线程等 待，当有用户请求来到的时候，就把这些请求添加到一个公共的消息队列中去。这个时候我们 刚刚开好的那几个线程就有用了，他们会排队逐个去消息队列中提取消息，并加以处理。(其 实这就是一个线程池处理消息的过程， 一个线程队列，一个消息队列，线程队列不断获取消息 队列中的消息。)这种方式很优雅地实现了异步通信和负载均衡的问题，并且线程在没事干的 时候会被系统挂起来，不会占用CPU 周期。

举个例子：假设有100万个用户同时与一个进程保持着TCP 连接，而每一个时刻只有几十或几百个 TCP 连接，所以我们只需要处理100万连接中的一小部分连接，在使用别的模型时只能通过 select的方式对所有的连接都遍历一遍，查询出其中有事件的连接。可想而知，这种查询方式 效率是多么的低下!这时我们的完成端口就闪亮登场了。完成端口是这么干的： 一旦一个连接 上有事件发生，它就会立即将事件组成一个完成包放入到完成端口中(其实就是放入到一个队 列里面),这时我们事先开启的等待线程就可以直接从该队列中取出该事件了，就避免了select 的查询，效率也就提高了很多，同一时间的用户量越大，效率越明显!

10.11.4 完成端口编程的基本流程

总体上讲，使用完成端口只用遵循如下几个编程步骤：

• (1)调用 CreateloCompletionPort() 函数创建一个完成端口，而且在一般情况下，我们需 要且只需要建立这一个完成端口。把它的句柄保存好，我们今后会经常用到它。
• (2)创建一个工作者线程A, 实际上会根据系统中有多少个处理器就建立多少个工作者 线程，这几个线程是专门用来和客户端进行通信的，目前暂时没有什么工作。这里为了说明原 理，我们就说创建一个工作者线程A。
• (3)A 线程循环调用GetQueuedCompletionStatus (函数来得到I/O 操作结果，这个函数是 一个阻塞函数。
• (4)主线程循环里调用accept 等待客户端连接上来。
• (5)主线程里accept 返回新连接建立以后，把这个新的套接字句柄用CreateloCompletionPortO关联到完成端口，然后发出一个异步的WSASend 或者WSARecv 调用以提交I/O 操作，因为 是异步函数，WSASend/WSARecv 会马上返回，实际的发送或者接收数据的操作由WINDOWS 系统去做。
• (6)主线程继续下一次循环，阻塞在accept这里等待客户端连接。
• (7)Windows 系统完成WSASend 或者WSArecv 的操作，把结果发到完成端口。
• (8)A 线程里的 GetQueuedCompletionStatus) 马上返回，并从完成端口取得刚完成的 WSASend/WSARecv的结果。
• ( 9 ) 在A 线程里对这些数据进行处理(如果处理过程很耗时，需要新开线程处理),然 后接着发出WSASend/WSARecv, 并继续下一次循环阻塞在GetQueuedCompletionStatus()。

10.11.5 相 关API

10.11.5.1 函 数CreateloCompletionPort

该函数创建一个输入/输出(I/O) 完成端口并将其与指定的文件句柄关联，或者创建尚未 与文件句柄关联的I/O 完成端口，允许以后进行关联。该函数声明如下：

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

• FileHandle: 完成端口用来关联的一个文件句柄，当使用 CreateFile 函数创建文件句 柄的时候你必须指定该句柄包含 FILE_FLAG_OVERLAPPED 标志位。如果 FileHandle 的值是INVALID_HANDLE_VALUE,CreateloCompletionPort 将会创建 一个不和任何文件关联的完成端口。在这种情况下，参数ExistingCompletionPort 必 须是NULL 并且参数CompletionKey 的内容将会被无视。
• ExistingCompletionPort: 现 有I/O 完成端口的句柄或NULL, 如果此参数为现有I/O 完成端口，那么该函数将其与FileHandle 参数指定的句柄相关联。如果成功，函数就 返回现有I/O 完成端口的句柄。如果此参数为NULL, 则该函数将创建一个新的I/O 完成端口。如果 FileHandle 参数有效，则将其与新的I/0 完成端口相关联。否则， 不会发生文件句柄关联。如果成功，那么该函数将把句柄返回给新的I/O 完成端口。
• CompletionKey: 该值就是类似线程里面传递的 一个参数，我们在 GetQueuedCompletionStatus 中第三个参数获得的就是这个值。
• NumberOfConcurrentThreads:如果此参数为NULL, 那么系统允许与系统中的处理器 一样多的并发运行的线程。如果 ExistingCompletionPort参数不是NULL, 则忽略此参数 。

如果函数执行成功，返回值一定是一个完成端口的地址；如果函数执行失败，就返回NULL。可以调用GetLastError 函数去获得详细的错误信息。

I/O系统可以指示发送I/O完成通知到完成端口，它们在那里排队，CreateloCompletionPort 函数提供了这个功能。完成端口的句柄是一个智能指针，没有人调用的话就会被释放。如果想 要释放完成端口的句柄，那么每个与它关联的文件句柄都必须被释放，然后调用CloseHandle 函数去释放完成端口的句柄。与完成端口关联的文件句柄不能够再被 ReadFileEx 或者 WriteFileEx 函数调用。最好是不要分享这种关联的文件或者继承或调用DuplicateHandle 函数。 这种复制句柄的操作将会产生完成消息通知。

执行一个文件的I/O 操作处理具有关联的I/O 完成端口，在I/O 操作完成时I/O 系统发送 完成通知包到完成端口。该 I/O 完成端口的完成包在 一 个先入先出队列中。使用 GetQueuedCompletionStatus函数来检索这些排队的I/O完成数据包。在同一进程中线程可以使 用 PostQueuedCompletionStatus函数放置在一个完成端口的队列中的I/O 完成通知包。通过这 样做，你可以使用完成端口去接收从进程的其他线程通信，除了接受来自I/O 系统的I/O 完成 通知包。

10.11.5.2 函 数GetQueuedCompletionStatus

该函数尝试从指定的I/O完成端口将I/O完成数据包出列，通俗点说，就是从完成端口中 获取已经完成的消息。如果没有完成数据包排队，那么函数等待与完成端口关联的挂起I/O 操 作完成。函数声明如下：

BOOL WINAPI GetQueuedCompletionStatus(In_ HANDLE CompletionPort,Out_LPDWORD lpNumberOfBytes, _Out_PULONG_PTR lpCompletionKey, _Out_LPOVERLAPPED *lpOverlapped,In DWORD dwMilliseconds);

• CompletionPort: 完成端口的句柄。
• lpNumberOfBytes: 该变量接收已完成的I/O 操作期间传输的字节数。
• lpCompletionKey: 该变量接收CreateloCompletionPort 中传递的第三个参数。
• lpOverlapped: 接收完成的I/O 操作启动时指定的OVERLAPPED 结构的地址。我们 可以通过 CONTAINING_RECORD 这个宏获取以该重叠结构为首地址的结构体信 息，也就是该重叠结构为什么必须放在结构体首地址的原因。
• dwMilliseconds: 超时时间(毫秒),如果为 INFINITE 就一直等待，直到有消息到
  来。

如果函数成功就返回TRUE, 失败则返回 FALSE 。如果设置了超时时间，超时将返回FALSE。

10.11.5.3 宏 CONTAINING_RECORD

该宏返回给定结构类型的结构实例的基地址和包含结构中字段的地址。该宏定义如下：

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

• Address: 通 过GetQueuedCompletionStatus 获取的重叠结构。
• Type: 以重叠结构为首地址的结构体。
• Field:Type 结构体的重叠结构变量。

返回包含Field 域(成员)的结构体的基地址。
为了更好地理解原理，下面看一个简单的例子。服务器端使用完成端口接收来自客户端发 送过来的TCP 消息，进行显示，并发送确认消息(ack) 给客户端，客户端再把收到的消息显示出 来 。

【例10.8】一个简单的端口实例

服务端

// serv.cpp : 定义控制台应用程序的入口点。
//#include <WinSock2.h>#pragma comment(lib, "Ws2_32.lib")		// Socket编程需用的动态链接库
#pragma comment(lib, "Kernel32.lib")	// IOCP需要用到的动态链接库#define BUFFER_SIZE 1024
#define OP_READ    18
#define OP_WRITE   28
#define OP_ACCEPT  38
#define CHECK_CODE 0x010110BOOL bStopThread = false;typedef struct _PER_HANDLE_DATA
{SOCKET s;sockaddr_in addr;      // 客户端地址char buf[BUFFER_SIZE];int nOperationType;
}PER_HANDLE_DATA, *PPER_HANDLE_DATA;#pragma pack(1)
typedef struct MsgAsk
{int iCode;int iBodySize;char szBuffer[32];
}MSG_ASK, *PMSG_ASK;typedef struct MsgBody
{int iBodySize;int iOpType;char szBuffer[64];
}MSG_BODY, *PMSG_BODY;typedef struct MsgAck
{int  iCheckCode;char szBuffer[32];
}MSG_ACK, *PMSG_ACK;
#pragma pack()DWORD WINAPI ServerWorkThread(LPVOID lpParam)
{// 得到完成端口句柄HANDLE           hCompletion = (HANDLE)lpParam;DWORD            dwTrans;PPER_HANDLE_DATA pPerHandle;OVERLAPPED*      pOverLapped;while (!bStopThread){// 在关联到此完成端口的所有套接字上等待I/O完成BOOL bOK = ::GetQueuedCompletionStatus(hCompletion,&dwTrans, (PULONG_PTR)&pPerHandle, &pOverLapped, WSA_INFINITE);if (!bOK){::closesocket(pPerHandle->s);::GlobalFree(pPerHandle);::GlobalFree(pOverLapped);continue;}switch(pPerHandle->nOperationType){case OP_READ:{MSG_ASK msgAsk = {0};memcpy(&msgAsk, pPerHandle->buf, sizeof(msgAsk));if (msgAsk.iCode != CHECK_CODE|| msgAsk.iBodySize != sizeof(msgAsk)){printf("error\n");}else{msgAsk.szBuffer[strlen(msgAsk.szBuffer) + 1] = '\n';printf(msgAsk.szBuffer);printf("Recv bytes = %d, msgAsk.size = %d\n", dwTrans, msgAsk.iBodySize);}MSG_BODY msgBody = {0};memcpy(&msgBody, pPerHandle->buf + msgAsk.iBodySize, sizeof(MSG_BODY));if (msgBody.iOpType == OP_READ && msgBody.iBodySize == sizeof(MSG_BODY)){printf("msgBody.szBuffer = %s\n", msgBody.szBuffer);}MSG_ACK msgAck = {0};msgAck.iCheckCode = CHECK_CODE;memcpy(msgAck.szBuffer, "This is the ack package",strlen("This is the ack package"));// 继续投递发送I/O请求pPerHandle->nOperationType = OP_WRITE;WSABUF buf;buf.buf = (char*)&msgAck;buf.len = sizeof(MSG_ACK);OVERLAPPED *pol = (OVERLAPPED *)::GlobalAlloc(GPTR, sizeof(OVERLAPPED));DWORD dwFlags = 0, dwSend = 0;::WSASend(pPerHandle->s, &buf, 1, &dwSend, dwFlags, pol, NULL); }break;case OP_WRITE:{if (dwTrans == sizeof(MSG_ACK)){printf("Transfer successfully\n");}// 然后投递接收I/O请求}break;case OP_ACCEPT:break;}}return 0;
}DWORD InitWinsock()
{DWORD dwRet = 0;WSADATA wsaData;  dwRet = WSAStartup(MAKEWORD(2,2), &wsaData);  if (dwRet != NO_ERROR)  {  printf("error code = %d\n", GetLastError());  dwRet = GetLastError();  }  return dwRet;
}void UnInitWinsock()
{WSACleanup();
}int main(int argc, _TCHAR* argv[])
{int nPort = 6000;InitWinsock();// 创建完成端口对象HANDLE hCompletion = ::CreateIoCompletionPort(INVALID_HANDLE_VALUE, 0, 0, 0);if (hCompletion == NULL){DWORD dwRet = GetLastError();return dwRet;}// 确定处理器的核心数量SYSTEM_INFO mySysInfo;GetSystemInfo(&mySysInfo);#if 1// 基于处理器的核心数量创建线程for(DWORD i = 0; i < (mySysInfo.dwNumberOfProcessors * 2); ++i){// 创建服务器工作器线程，并将完成端口传递到该线程HANDLE ThreadHandle = CreateThread(NULL, 0, ServerWorkThread, hCompletion, 0, NULL);if(NULL == ThreadHandle){printf("Create Thread Handle failed. Error:%d",GetLastError());//system("pause");return -1;}CloseHandle(ThreadHandle);}
#else::CreateThread(NULL, 0, ServerWorkThread, (LPVOID)hCompletion, 0, 0);
#endif// 创建监听套接字SOCKET sListen = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);SOCKADDR_IN si;si.sin_family = AF_INET;si.sin_port = ::htons(nPort);si.sin_addr.s_addr = INADDR_ANY;::bind(sListen, (sockaddr*)&si, sizeof(si));::listen(sListen, 10);while (TRUE){SOCKADDR_IN saRemote;int nRemoteLen = sizeof(saRemote);printf("Accepting...\n");SOCKET sNew = ::accept(sListen, (sockaddr*)&saRemote, &nRemoteLen);//SOCKET sNew = ::accept(sListen, NULL, NULL);if (sNew == INVALID_SOCKET){continue;}printf("Accept one!\n");// 接受新连接后，创建一个per-handle数据，并关联到完成端口对象PPER_HANDLE_DATA pPerHandle = (PPER_HANDLE_DATA)::GlobalAlloc(GPTR, sizeof(PER_HANDLE_DATA));pPerHandle->s = sNew;memcpy(&pPerHandle->addr, &saRemote, nRemoteLen);pPerHandle->nOperationType = OP_READ;::CreateIoCompletionPort((HANDLE) pPerHandle->s, hCompletion, (ULONG_PTR)pPerHandle, 0);// 投递一个接收请求OVERLAPPED *pol = (OVERLAPPED *)::GlobalAlloc(GPTR, sizeof(OVERLAPPED));WSABUF buf;buf.buf = pPerHandle->buf;buf.len = BUFFER_SIZE;DWORD dwRecv = 0;DWORD dwFlags = 0;::WSARecv(pPerHandle->s, &buf, 1, &dwRecv, &dwFlags, pol, NULL);}return 0;
}

客户端

// client.cpp : 定义控制台应用程序的入口点。
//#include <WinSock2.h>#pragma comment(lib, "Ws2_32.lib")		// Socket编程需用的动态链接库
#pragma comment(lib, "Kernel32.lib")	// IOCP需要用到的动态链接库#define CHECK_CODE 0x010110
#define OP_READ   18
#define OP_WRITE  28
#define OP_ACCEPT 38#pragma pack(1)typedef struct MsgAsk
{int iCode;int iBodySize;char szBuffer[32];
}MSG_ASK, *PMSG_ASK;typedef struct MsgBody
{int iBodySize;int iOpType;char szBuffer[64];
}MSG_BODY, *PMSG_BODY;typedef struct MsgAck
{int  iCheckCode;char szBuffer[32];
}MSG_ACK, *PMSG_ACK;#pragma pack()DWORD SendAll(SOCKET &clientSock, char* buffer, int size)
{DWORD dwStatus = 0;char  *pTemp   = buffer;int   total    = 0, count = 0;while(total < size){count = send(clientSock, pTemp, size - total, 0);if(count < 0){dwStatus = WSAGetLastError();break;}total += count;pTemp += count;}return dwStatus ;
}DWORD RecvAll(SOCKET &sock, char* buffer, int size) 
{ DWORD dwStatus = 0; char *pTemp    = buffer; int total      = 0, count = 0; while (total < size) { count = recv(sock, pTemp, size-total, 0); if (count < 0) { dwStatus = WSAGetLastError(); break; } total += count; pTemp += count; } return dwStatus; 
} int _tmain(int argc, _TCHAR* argv[])
{WSADATA wsaData; int iResult = WSAStartup(MAKEWORD(2,2), &wsaData); if (iResult != NO_ERROR) { printf("error code = %d\n", GetLastError()); return -1; } sockaddr_in clientAddr; clientAddr.sin_addr.s_addr = inet_addr("127.0.0.1"); clientAddr.sin_family      = AF_INET; clientAddr.sin_port        = htons(6000); SOCKET clientSock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); if (clientSock == INVALID_SOCKET) { printf("Create socket failed, error code = %d\n", WSAGetLastError()); return -1; } //connect while (connect(clientSock, (SOCKADDR *)&clientAddr, sizeof(SOCKADDR_IN)) == SOCKET_ERROR) { printf("Connecting...\n"); Sleep(1000); } MSG_ASK msgAsk = {0};msgAsk.iBodySize = sizeof(MSG_ASK);msgAsk.iCode = CHECK_CODE;memcpy(msgAsk.szBuffer, "This is a header", strlen("This is a header"));// 发送头部SendAll(clientSock, (char*)&msgAsk, msgAsk.iBodySize);MSG_BODY msgBody = {0};msgBody.iBodySize = sizeof(MSG_BODY);msgBody.iOpType = OP_READ;memcpy(msgBody.szBuffer, "This is the body", strlen("This is the body"));// 发送bodySendAll(clientSock, (char*)&msgBody, msgBody.iBodySize);MSG_ACK msgAck = {0};RecvAll(clientSock, (char*)&msgAck, sizeof(msgAck));if (msgAck.iCheckCode == CHECK_CODE){printf("The process is successful,\nmsgAck.szBuffer = %s \n", msgAck.szBuffer);}else{printf("failed\n");}closesocket(clientSock);WSACleanup();return 0;
}