winsock IO模型 完成端口

2015/05/28 11:42
阅读数 74

Winsock工作模型有下面六种
一:select模型
二:WSAAsyncSelect模型
三:WSAEventSelect模型
四:Overlapped  I/O 事件通知模型
五:Overlapped I/O  完成例程模型
六:IOCP模型

 

重叠I/O模型 Winsock2的发布使得Socket I/O有了和文件I/O统一的接口。我们可以通过使用Win32文件操纵函数ReadFile和WriteFile来进行Socket I/O。伴随而来的,用于普通文件I/O的重叠I/O模型和完成端口模型对Socket I/O也适用了。这些模型的优点是可以达到更佳的系统性能,但是实现较为复杂,里面涉及较多的C语言技巧。例如我们在完成端口模型中会经常用到所谓的“尾 随数据”。

在WINDOWS下进行网络服务端程序开发,Winsock 完成端口模型是最高效的。Winsock的完成端口模型借助Widnows的重叠IO和完成端口来实现,完成端口模型懂了之后是比较简单的,但是要想掌握 Winsock完成端口模型,需要对WINDOWS下的线程、线程同步,Winsock API以及WINDOWS IO机制有一定的了解。如果不了解,推荐几本书:《WINDOWS核心编程》,《WIN32多线程程序设计》、《WINDOWS网络编程技术》。在去年, 我在C语言下用完成端口模型写了一个 WEBSERVER,前些天,我决定用C++重写这个WEBSERVER,给这个WEBSERVER增加了一些功能,并改进完成端口操作方法,比如采用 AcceptEx来代替accept和使用LOOKASIDE LIST来管理内存,使得WEBSERVER的性能有了比较大的提高。

 

一:完成端口模型

首先我们要抽象出一个完成端口大概的处理流程:

1:创建一个完成端口。

2:创建一个线程A。


3:A线程循环调用GetQueuedCompletionStatus()函数来得到IO操作结果,这个函数是个阻塞函数。


4:主线程循环里调用accept等待客户端连接上来。


5:主线程里accept返回新连接建立以后,把这个新的套接字句柄用CreateIoCompletionPort关联到完成端口,然后发出一个异步
的WSASend或者WSARecv调用,因为是异步函数,WSASend/WSARecv会马上返回,实际的发送或者接收数据的操作由WINDOWS系 统去做。


6:主线程继续下一次循环,阻塞在accept这里等待客户端连接。


7:WINDOWS系统完成WSASend或者WSArecv的操作,把结果发到完成端口。

8:A线程里的GetQueuedCompletionStatus()马上返回,并从完成端口取得刚完成的WSASend/WSARecv的结果。


9:在A线程里对这些数据进行处理(如果处理过程很耗时,需要新开线程处理),然后接着发出WSASend/WSARecv,并继续下一次循环阻塞在GetQueuedCompletionStatus()这里。

我 们不停地发出异步的WSASend/WSARecv
IO操作,具体的IO处理过程由WINDOWS系统完成,WINDOWS系统完成实际的IO处理后,把结果送到完成端口上(如果有多个IO都完成了,那么
就在完成端口那里排成一个队列)。我们在另外一个线程里从完成端口不断地取出IO操作结果,然后根据需要再发出WSASend/WSARecv IO操作。

二:提高完成端口效率的几种有效方法

1:使用AcceptEx代替accept。AcceptEx函 数是微软的Winsosk
扩展函数,这个函数和accept的区别就是:accept是阻塞的,一直要到有客户端连接上来后accept才返回,而AcceptEx是异步的,直接
就返回了,所以我们利用AcceptEx可以发出多个AcceptEx调用

等待客户端连接。另外,如果我们可以预见到客户
端一连接上来后就会发送数据(比如WEBSERVER的客户端浏览器),那么可以随着AcceptEx投递一个BUFFER进去,这样连接一建立成功,就
可以接收客户端发出的数据到BUFFER里,这样使用的话,一次AcceptEx调用相当于accpet和recv的一次连续调用。同时,微软的几个扩展
函数针对操作系统优化过,效率优于WINSOCK 的标准API函数。


2:在套接字上使用SO_RCVBUF和SO_SNDBUF选项来关闭系统缓冲区。详细的介绍可以参考《WINDOWS核心编程》第9章。

3:内存分配方法。因为每次为一个新建立的套接字都要动态分配一个“单IO数据”和“单句柄数据”的数据结构,然后在套接字关闭的时候释放,这样如果有
成千上万个客户频繁连接时候,会使得程序很多开销花费在内存分配和释放上。这里我们可以使用lookaside list。开始在微软的platform
sdk里的SAMPLE里看到lookaside list,我一点不明白,MSDN里有没有。后来还是在DDK的文档中找到了,,

lookaside
list

A system-managed queue from which entries of a fixed size can be
allocated and into which entries can be deallocated dynamically. Callers of the
Ex(ecutive) Support lookaside list routines can use a lookaside list to manage
any dynamically sized set of fixed-size buffers or structures with
caller-determined contents.

For example, the I/O Manager uses a
lookaside for fast allocation and deallocation of IRPs and MDLs. As another
example, some of the system-supplied SCSI class drivers use lookaside lists to
allocate and release memory for SRBs.
lookaside
其实就是一种内存管理方法,和内存池使用方法类似。我个人的理解:就是一个单链表。每次要分配内
存前,先查看这个链表是否为空,如果不为空,就从这个链表中解下一个结点,则不需要新分配。如果为空,再动态分配。使用完成后,把这个数据结构不释放,而
是把它插入到链表中去,以便下一次使用。这样相比效率就高了很多。在我的程序中,我就使用了这种单链表来管理。

在我们使用AcceptEx并随着AcceptEx投递一个BUFFER后会带来一个副作用:比如某个客户端只执行一个connect操作,并不执行
send操作,那么AcceptEx这个请求不会完成,相应的,我们用GetQueuedCompletionStatus在完成端口中得不到操作结果,
这样,如果有很多个这样的连接,对程序性能会造成巨大的影响,我们需要用一种方法来定时检测,当某个连接已经建立并且连接时间超过我们规定的时间而且没有
收发过数据,那么我们就把它关闭。检测连接时间可以用SO_CONNECT_TIME来调用getsockopt得到。

还 有一个值得注意的地方:就是我们不能一下子发出很多AcceptEx调用等待客户连接,这样对程序的性能有影响,同时,在我们发出的AcceptEx调用
耗尽的时候需要新增加AcceptEx调用,我们可以把FD_ACCEPT事件和一个EVENT关联起来,然后用WaitForSingleObject
等待这个Event,当已经发出AccpetEx调用数目耗尽而又有新的客户端需要连接上来,FD_ACCEPT事件将被触发,EVENT变为已传信状 态, WaitForSingleObject返回,我们就重新发出足够的AcceptEx调用。


用完成例程方式实现的重叠I/O模型

#include <WINSOCK2.H>
#include <stdio.h>

#define PORT    5150
#define MSGSIZE 1024
#pragma comment(lib, "ws2_32.lib")
typedef enum 
{ RECV_POSTED
}OPERATION_TYPE;

typedef struct 
{ 
    WSAOVERLAPPED overlap; 
    WSABUF         Buffer;
    char           szMessage[MSGSIZE]; 
    DWORD          NumberOfBytesRecvd;
    DWORD          Flags; 
    OPERATION_TYPE OperationType;
}PER_IO_OPERATION_DATA, *LPPER_IO_OPERATION_DATA;

DWORD WINAPI WorkerThread(LPVOID);
int main() 
{ 
    WSADATA                 wsaData;  
    SOCKET                  sListen, sClient; 
    SOCKADDR_IN             local, client; 
    DWORD                   i, dwThreadId; 
    int                     iaddrSize = sizeof(SOCKADDR_IN);
    HANDLE                  CompletionPort = INVALID_HANDLE_VALUE;
    SYSTEM_INFO             systeminfo;  
    LPPER_IO_OPERATION_DATA lpPerIOData = NULL;
    WSAStartup(0x0202, &wsaData);
    CompletionPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);

    // Create worker thread
    GetSystemInfo(&systeminfo);
    for (i = 0; i < systeminfo.dwNumberOfProcessors; i++) 
    {   
        CreateThread(NULL, 0, WorkerThread, CompletionPort, 0, &dwThreadId);
    }
    // Create listening socket
    sListen = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);

    // Bind
    local.sin_addr.S_un.S_addr = htonl(INADDR_ANY);
    local.sin_family = AF_INET; 
    local.sin_port = htons(PORT);
    bind(sListen, (struct sockaddr *)&local, sizeof(SOCKADDR_IN));

    // Listen
    listen(sListen, 3);

    while (TRUE) 
    {     // Accept a connection 
        sClient = accept(sListen, (struct sockaddr *)&client, &iaddrSize);  
        printf("Accepted client:%s:%d\n", inet_ntoa(client.sin_addr), ntohs(client.sin_port));
        // Associate the newly arrived client socket with completion port   
        CreateIoCompletionPort((HANDLE)sClient, CompletionPort, (DWORD)sClient, 0);     
        // Launch an asynchronous operation for new arrived connection   
        lpPerIOData = (LPPER_IO_OPERATION_DATA)HeapAlloc( GetProcessHeap(), HEAP_ZERO_MEMORY, sizeof(PER_IO_OPERATION_DATA));
        lpPerIOData->Buffer.len = MSGSIZE;   
        lpPerIOData->Buffer.buf = lpPerIOData->szMessage;   
        lpPerIOData->OperationType = RECV_POSTED;   
        WSARecv(sClient, &lpPerIOData->Buffer, 1, &lpPerIOData->NumberOfBytesRecvd, &lpPerIOData->Flags, 
            &lpPerIOData->overlap, NULL); 
    }
    PostQueuedCompletionStatus(CompletionPort, 0xFFFFFFFF, 0, NULL);
    CloseHandle(CompletionPort); 
    closesocket(sListen); 
    WSACleanup(); 
    return 0;
}

DWORD WINAPI WorkerThread(LPVOID CompletionPortID) 
{ 
    HANDLE                  CompletionPort=(HANDLE)CompletionPortID; 
    DWORD                   dwBytesTransferred; 
    SOCKET                  sClient;
    LPPER_IO_OPERATION_DATA lpPerIOData = NULL;
    while (TRUE) 
    {   
        GetQueuedCompletionStatus(CompletionPort, &dwBytesTransferred,  &sClient,
            (LPOVERLAPPED *)&lpPerIOData,       INFINITE);   

        if (dwBytesTransferred == 0xFFFFFFFF)    
        {      
            return 0;    
        }        
        if (lpPerIOData->OperationType == RECV_POSTED)  
        {   
            if (dwBytesTransferred == 0)   
            {  
                // Connection was closed by client     
                closesocket(sClient);   
                HeapFree(GetProcessHeap(), 0, lpPerIOData);   
            }     
            else       
            {        
                lpPerIOData->szMessage[dwBytesTransferred] = '\0';   
                send(sClient, lpPerIOData->szMessage, dwBytesTransferred, 0);    
                // Launch another asynchronous operation for sClient    
                memset(lpPerIOData, 0, sizeof(PER_IO_OPERATION_DATA));     
                lpPerIOData->Buffer.len = MSGSIZE;       
                lpPerIOData->Buffer.buf = lpPerIOData->szMessage;         
                lpPerIOData->OperationType = RECV_POSTED;      
                WSARecv(sClient, &lpPerIOData->Buffer,1, &lpPerIOData->NumberOfBytesRecvd, 
                    &lpPerIOData->Flags, &lpPerIOData->overlap,  NULL);
            }    
        } 
    } 
    return 0;
}



展开阅读全文
打赏
0
0 收藏
分享
加载中
更多评论
打赏
0 评论
0 收藏
0
分享
返回顶部
顶部