IOCP（I/OCompletionPort，I/O完成端口）是性能最好的一種I/O模型。它是應用程序使用線(xiàn)程池處理異步I/O請求的一種機制。在處理多個(gè)并發(fā)的異步I/O請求時(shí)，以往的模型都是在接收請求是創(chuàng )建一個(gè)線(xiàn)程來(lái)應答請求。這樣就有很多的線(xiàn)程并行地運行在系統中。而這些線(xiàn)程都是可運行的，Windows內核花費大量的時(shí)間在進(jìn)行線(xiàn)程的上下文切換，并沒(méi)有多少時(shí)間花在線(xiàn)程運行上。再加上創(chuàng )建新線(xiàn)程的開(kāi)銷(xiāo)比較大，所以造成了效率的低下。
調用的步驟如下:
抽象出一個(gè)完成端口大概的處理流程：
1：創(chuàng )建一個(gè)完成端口。
2：創(chuàng )建一個(gè)線(xiàn)程A。
3：A線(xiàn)程循環(huán)調用GetQueuedCompletionStatus()函數來(lái)得到IO操作結果，這個(gè)函數是個(gè)阻塞函數。
4：主線(xiàn)程循環(huán)里調用accept等待客戶(hù)端連接上來(lái)。
5：主線(xiàn)程里accept返回新連接建立以后，把這個(gè)新的套接字句柄用CreateIoCompletionPort關(guān)聯(lián)到完成端口，然后發(fā)出一個(gè)異步的WSASend或者WSARecv調用，因為是異步函數，WSASend/WSARecv會(huì )馬上返回，實(shí)際的發(fā)送或者接收數據的操作由WINDOWS系統去做。
6：主線(xiàn)程繼續下一次循環(huán)，阻塞在accept這里等待客戶(hù)端連接。
7：WINDOWS系統完成WSASend或者WSArecv的操作，把結果發(fā)到完成端口。
8：A線(xiàn)程里的GetQueuedCompletionStatus()馬上返回，并從完成端口取得剛完成的WSASend/WSARecv的結果。
9：在A(yíng)線(xiàn)程里對這些數據進(jìn)行處理(如果處理過(guò)程很耗時(shí)，需要新開(kāi)線(xiàn)程處理)，然后接著(zhù)發(fā)出WSASend/WSARecv，并繼續下一次循環(huán)阻塞在GetQueuedCompletionStatus()這里。
歸根到底概括完成端口模型一句話(huà)：
我們不停地發(fā)出異步的WSASend/WSARecvIO操作，具體的IO處理過(guò)程由WINDOWS系統完成，WINDOWS系統完成實(shí)際的IO處理后，把結果送到完成端口上（如果有多個(gè)IO都完成了，那么就在完成端口那里排成一個(gè)隊列）。我們在另外一個(gè)線(xiàn)程里從完成端口不斷地取出IO操作結果，然后根據需要再發(fā)出WSASend/WSARecvIO操作。
而IOCP模型是事先開(kāi)好了N個(gè)線(xiàn)程，存儲在線(xiàn)程池中，讓他們hold。然后將所有用戶(hù)的請求都投遞到一個(gè)完成端口上，然后N個(gè)工作線(xiàn)程逐一地從完成端口中取得用戶(hù)消息并加以處理。這樣就避免了為每個(gè)用戶(hù)開(kāi)一個(gè)線(xiàn)程。既減少了線(xiàn)程資源，又提高了線(xiàn)程的利用率。
完成端口模型是怎樣實(shí)現的呢？我們先創(chuàng )建一個(gè)完成端口（::CreateIoCompletioPort()）。然后再創(chuàng )建一個(gè)或多個(gè)工作線(xiàn)程，并指定他們到這個(gè)完成端口上去讀取數據。我們再將遠程連接的套接字句柄關(guān)聯(lián)到這個(gè)完成端口（還是用::CreateIoCompletionPort()）。一切就OK了。
工作線(xiàn)程都干些什么呢？首先是調用::GetQueuedCompletionStatus()函數在關(guān)聯(lián)到這個(gè)完成端口上的所有套接字上等待I/O的完成。再判斷完成了什么類(lèi)型的I/O。一般來(lái)說(shuō)，有三種類(lèi)型的I/O，OP_ACCEPT，OP_READ和OP_WIRTE。我們到數據緩沖區內讀取數據后，再投遞一個(gè)或是多個(gè)同類(lèi)型的I/O即可（::AcceptEx()、::WSARecv()、::WSASend()）。對讀取到的數據，我們可以按照自己的需要來(lái)進(jìn)行相應的處理。
為此，我們需要一個(gè)以OVERLAPPED（重疊I/O）結構為第一個(gè)字段的per-I/O數據自定義結構。
typedef struct _PER_IO_DATA
{
OVERLAPPED ol; // 重疊I/O結構
char buf[BUFFER_SIZE]; // 數據緩沖區
int nOperationType; //I/O操作類(lèi)型
#define OP_READ 1
#define OP_WRITE 2
#define OP_ACCEPT 3
} PER_IO_DATA, *PPER_IO_DATA;
將一個(gè)PER_IO_DATA結構強制轉化成一個(gè)OVERLAPPED結構傳給::GetQueuedCompletionStatus()函數，返回的這個(gè)PER_IO_DATA結構的的nOperationType就是I/O操作的類(lèi)型。當然，這些類(lèi)型都是在投遞I/O請求時(shí)自己設置的。
這樣一個(gè)IOCP服務(wù)器的框架就出來(lái)了。當然，要做一個(gè)好的IOCP服務(wù)器，還有考慮很多問(wèn)題，如內存資源管理、接受連接的方法、惡意的客戶(hù)連接、包的重排序等等。以上是個(gè)人對于IOCP模型的一些理解與看法，還有待完善。另外各Winsock API的用法參見(jiàn)MSDN。
補充IOCP模型的實(shí)現：
//創(chuàng )建一個(gè)完成端口
HANDLE FCompletPort = CreateIoCompletionPort( INVALID_HANDLE_VALUE, 0,0,0 );
//接受遠程連接，并把這個(gè)連接的socket句柄綁定到剛才創(chuàng )建的IOCP上
AConnect = accept( FListenSock, addr, len);
CreateIoCompletionPort( AConnect, FCompletPort, NULL, 0 );
//創(chuàng )建CPU數*2 + 2個(gè)線(xiàn)程
SYSTEM_INFO si;
GetSystemInfo(&si);
for (int i=1;si.dwNumberOfProcessors*2+2;i++)
{
AThread = TRecvSendThread.Create( false );
AThread.CompletPort = FCompletPort;//告訴這個(gè)線(xiàn)程，你要去這個(gè)IOCP去訪(fǎng)問(wèn)數據
}
OK，就這么簡(jiǎn)單，我們要做的就是建立一個(gè)IOCP，把遠程連接的socket句柄綁定到剛才創(chuàng )建的IOCP上，最后創(chuàng )建n個(gè)線(xiàn)程，并告訴這n個(gè)線(xiàn)程到這個(gè)IOCP上去訪(fǎng)問(wèn)數據就可以了。
再看一下TRecvSendThread線(xiàn)程都干些什么：
void TRecvSendThread.Execute(...)
{
while (!self.Terminated)
{
//查詢(xún)IOCP狀態(tài)（數據讀寫(xiě)操作是否完成）
GetQueuedCompletionStatus( CompletPort, BytesTransd, CompletKey, POVERLAPPED(pPerIoDat), TIME_OUT );
if (BytesTransd !=0) .......
....;//數據讀寫(xiě)操作完成
//再投遞一個(gè)讀數據請求
WSARecv( CompletKey, &(pPerIoDat->BufData), 1, BytesRecv, Flags, &(pPerIoDat->Overlap), NULL );
}
}
讀寫(xiě)線(xiàn)程只是簡(jiǎn)單地檢查IOCP是否完成了我們投遞的讀寫(xiě)操作，如果完成了則再投遞一個(gè)新的讀寫(xiě)請求。
應該注意到，我們創(chuàng )建的所有TRecvSendThread都在訪(fǎng)問(wèn)同一個(gè)IOCP（因為我們只創(chuàng )建了一個(gè)IOCP），并且我們沒(méi)有使用臨界區！難道不會(huì )產(chǎn)生沖突嗎？不用考慮同步問(wèn)題嗎？
呵呵，這正是IOCP的奧妙所在。IOCP不是一個(gè)普通的對象，不需要考慮線(xiàn)程安全問(wèn)題。它會(huì )自動(dòng)調配訪(fǎng)問(wèn)它的線(xiàn)程：如果某個(gè)socket上有一個(gè)線(xiàn)程A正在訪(fǎng)問(wèn)，那么線(xiàn)程B的訪(fǎng)問(wèn)請求會(huì )被分配到另外一個(gè)socket。這一切都是由系統自動(dòng)調配的，我們無(wú)需過(guò)問(wèn)。
實(shí)例:
簡(jiǎn)單實(shí)現，適合IOCP入門(mén)
參考：《WINDOWS網(wǎng)絡(luò )與通信程序設計》
/******************************************************************
*
* Copyright (c) 2008, xxxxx有限公司
* All rights reserved.
*
* 文件名稱(chēng)：IOCPHeader.h
* 摘 要： IOCP定義文件
*
* 當前版本：1.0
* 作 者：吳會(huì )然
* 完成日期：2008-9-16
*
* 取代版本：
* 原 作者：
* 完成日期：
*
******************************************************************/
#ifndef _IOCPHEADER_H_20080916_
#define _IOCPHEADER_H_20080916_
#include <WINSOCK2.H>
#include <windows.h>
#define BUFFER_SIZE 1024
/******************************************************************
* per_handle 數據
*******************************************************************/
typedef struct _PER_HANDLE_DATA
{
SOCKET s; // 對應的套接字句柄
sockaddr_in addr; // 對方的地址
}PER_HANDLE_DATA, *PPER_HANDLE_DATA;
/******************************************************************
* per_io 數據
*******************************************************************/
typedef struct _PER_IO_DATA
{
OVERLAPPED ol; // 重疊結構
char buf[BUFFER_SIZE]; // 數據緩沖區
int nOperationType; // 操作類(lèi)型
#define OP_READ 1
#define OP_WRITE 2
#define OP_ACCEPT 3
}PER_IO_DATA, *PPER_IO_DATA;
#endif
/******************************************************************
*
* Copyright (c) 2008, xxxxx有限公司
* All rights reserved.
*
* 文件名稱(chēng)：main.cpp
* 摘 要： iocp demo
*
* 當前版本：1.0
* 作 者：吳會(huì )然
* 完成日期：2008-9-16
*
* 取代版本：
* 原 作者：
* 完成日期：
*
******************************************************************/
#include <iostream>
#include <string>
#include "IOCPHeader.h"
using namespace std;
DWORD WINAPI ServerThread( LPVOID lpParam );
int main( int argc, char *argv[] )
{
//////////////////////////////////////////////////////////////////////////
WSADATA wsaData;
if( 0 != WSAStartup( MAKEWORD( 2, 2 ), &wsaData ) )
{
printf( "Using %s (Status:%s)\n", wsaData.szDescription, wsaData.szSystemStatus );
printf( "with API versions: %d.%d to %d.%d",
LOBYTE( wsaData.wVersion), HIBYTE( wsaData.wVersion ),
LOBYTE( wsaData.wHighVersion), HIBYTE( wsaData.wHighVersion) );
return -1;
}
else
{
printf("Windows sockets 2.2 startup\n");
}
//////////////////////////////////////////////////////////////////////////
int nPort = 20055;
// 創(chuàng )建完成端口對象
// 創(chuàng )建工作線(xiàn)程處理完成端口對象的事件
HANDLE hIocp = ::CreateIoCompletionPort( INVALID_HANDLE_VALUE, 0, 0, 0 );
::CreateThread( NULL, 0, ServerThread, (LPVOID)hIocp, 0, 0 );
// 創(chuàng )建監聽(tīng)套接字，綁定本地端口，開(kāi)始監聽(tīng)
SOCKET sListen = ::socket( AF_INET,-SOCK_STREAM, 0 );
SOCKADDR_IN addr;
addr.sin_family = AF_INET;
addr.sin_port = ::htons( nPort );
addr.sin_addr.S_un.S_addr = INADDR_ANY;
::bind( sListen, (sockaddr *)&addr, sizeof( addr ) );
::listen( sListen, 5 );
printf( "iocp demo start......\n" );
// 循環(huán)處理到來(lái)的請求
while ( TRUE )
{
// 等待接受未決的連接請求
SOCKADDR_IN saRemote;
int nRemoteLen = sizeof( saRemote );
SOCKET sRemote = ::accept( sListen, (sockaddr *)&saRemote, &nRemoteLen );
// 接受到新連接之后，為它創(chuàng )建一個(gè)per_handle數據，并將他們關(guān)聯(lián)到完成端口對象
PPER_HANDLE_DATA pPerHandle = ( PPER_HANDLE_DATA )::GlobalAlloc( GPTR, sizeof( PPER_HANDLE_DATA ) );
if( pPerHandle == NULL )
{
break;
}
pPerHandle->s = sRemote;
memcpy( &pPerHandle->addr, &saRemote, nRemoteLen );
::CreateIoCompletionPort( ( HANDLE)pPerHandle->s, hIocp, (DWORD)pPerHandle, 0 );
// 投遞一個(gè)接受請求
PPER_IO_DATA pIoData = ( PPER_IO_DATA )::GlobalAlloc( GPTR, sizeof( PPER_IO_DATA ) );
if( pIoData == NULL )
{
break;
}
pIoData->nOperationType = OP_READ;
WSABUF buf;
buf.buf = pIoData->buf;
buf.len = BUFFER_SIZE;
DWORD dwRecv = 0;
DWORD dwFlags = 0;
::WSARecv( pPerHandle->s, &buf, 1, &dwRecv, &dwFlags, &pIoData->ol, NULL );
}
//////////////////////////////////////////////////////////////////////////
ERROR_PROC:
WSACleanup();
//////////////////////////////////////////////////////////////////////////
return 0;
}
/******************************************************************
* 函數介紹：處理完成端口對象事件的線(xiàn)程
* 輸入參數：
* 輸出參數：
* 返回值 ：
*******************************************************************/
DWORD WINAPI ServerThread( LPVOID lpParam )
{
HANDLE hIocp = ( HANDLE )lpParam;
if( hIocp == NULL )
{
return -1;
}
DWORD dwTrans = 0;
PPER_HANDLE_DATA pPerHandle;
PPER_IO_DATA pPerIo;
while( TRUE )
{
// 在關(guān)聯(lián)到此完成端口的所有套接字上等待I/O完成
BOOL bRet = ::GetQueuedCompletionStatus( hIocp, &dwTrans,(LPDWORD)&pPerHandle, (LPOVERLAPPED*)&pPerIo, WSA_INFINITE );
if( !bRet ) // 發(fā)生錯誤
{
::closesocket( pPerHandle->s );
::GlobalFree( pPerHandle );
::GlobalFree( pPerIo );
cout << "error" << endl;
continue;
}
// 套接字被對方關(guān)閉
if( dwTrans == 0 && ( pPerIo->nOperationType ==OP_READ || pPerIo->nOperationType&nb-sp;== OP_WRITE ) )
{
::closesocket( pPerHandle->s );
::GlobalFree( pPerHandle );
::GlobalFree( pPerIo );
cout << "client closed" << endl;
continue;
}
switch ( pPerIo->nOperationType )
{
case OP_READ: // 完成一個(gè)接收請求
{
pPerIo->buf[dwTrans] = '\0';
printf( "%s\n", pPerIo->buf );
// 繼續投遞接受操作
WSABUF buf;
buf.buf = pPerIo->buf;
buf.len = BUFFER_SIZE;
pPerIo->nOperationType = OP_READ;
DWORD dwRecv = 0;
DWORD dwFlags = 0;
::WSARecv( pPerHandle->s, &buf, 1, &dwRecv, &dwFlags, &pPerIo->ol, NULL );
}
break;
case OP_WRITE:
case OP_ACCEPT:
break;
}
}
return 0;
}
聯(lián)系客服
