深圳商城网站制作,北京小程序app开发,wordpress底部版权代码,泗阳网站设计10.11 完成端口
10.11.1 基本概念
完成端口的全称是I/O 完成端口#xff0c;英文为IOCP(I/O Completion Port) 。IOCP是一个异 步I/O 的 API, 可以高效地将I/O 事件通知给应用程序。与使用select() 或是其他异步方法不同 的是#xff0c;一个套接字与一个完成端口关联了起来…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;
}
