服务器开发入门——理解异步I/O

#include <stdio.h>
#include <Winsock2.h>
#include <windows.h>
#include <process.h>

#define DATA_BUFSIZE 4096

typedef struct
{
    WSAOVERLAPPED  overlap;
    WSABUF         DataBuf;
    char           buffer[DATA_BUFSIZE];
    DWORD          NumberOfBytesRecvd;
    DWORD          Flags;
}PER_IO_OPERATION_DATA, *LPPER_IO_OPERATION_DATA;

void err_sys(
    char* err
)
{
    fprintf(stderr, err);
    WSACleanup();
    exit(-1);
}

unsigned __stdcall WorkThread(
    void* param
);

int main(int argc, char* argv[])
{
    HANDLE        CompletionPort;
    WSADATA        WsaData;
    int            clilen = sizeof(sockaddr_in);
    sockaddr_in seraddr,cliaddr;
    SOCKET        listenfd, connfd;
    int            opt = 1;

    SYSTEM_INFO systeminfo;
    unsigned    i;

    int            ret;
    LPPER_IO_OPERATION_DATA lpperiodata = NULL;

    HANDLE        hWorkThread;
    unsigned    threadID;

    if (WSAStartup (0x0101, &WsaData) == SOCKET_error)
        err_sys("WSAStartup Failed\n");
    
    CompletionPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);
    if(CompletionPort == NULL)
        err_sys("create port error!\n");
    
    seraddr.sin_family = AF_INET;
    seraddr.sin_addr.s_addr = INADDR_ANY;
    seraddr.sin_port = htons(9877);

    listenfd = WSASocket(AF_INET, SOCK_STREAM, 0, NULL, 0, WSA_FLAG_OVERLAPPED);
    if(listenfd == INVALID_SOCKET)
        err_sys("socket error!\n");

    if(setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, (char*)&opt, sizeof(opt)) < 0)
        err_sys("setsockopt error!\n");

    if(bind(listenfd, (SOCKADDR*) &seraddr, sizeof(seraddr)) == SOCKET_ERROR)
        err_sys("bind error\n");

    if(listen(listenfd, 5) == SOCKET_ERROR)
        err_sys("listen error!\n");

    printf("server is listening...\n");

    if(CreateIoCompletionPort((HANDLE)listenfd, CompletionPort, (DWORD)listenfd, 0) == NULL)
        err_sys("associate lisenfd error!\n");

    GetSystemInfo(&systeminfo);
    
    //创建工作线程
    for (i = 0; i < systeminfo.dwNumberOfProcessors; i++)
    {
        hWorkThread = (HANDLE)_beginthreadex(NULL, 0, &WorkThread, CompletionPort, 0, &threadID);
        CloseHandle(hWorkThread);
    }

    for (;;)
    {    
        if((connfd = accept(listenfd, (struct sockaddr*)&cliaddr, &clilen)) == SOCKET_ERROR )
        {
            int code = WSAGetLastError();
            printf("accept error!\n");
            continue;
        }
        
        //将连接句柄关联到I/O完成端口
        if(CreateIoCompletionPort((HANDLE)connfd, CompletionPort, (DWORD)connfd, 0) == NULL)
        {
            printf("associate connfd error!\n");
            closesocket(connfd);
            continue;
        }
        
        lpPerIOData = (LPPER_IO_OPERATION_DATA)malloc(sizeof(PER_IO_OPERATION_DATA));
        ZeroMemory(lpPerIOData, sizeof(PER_IO_OPERATION_DATA));
        lpPerIOData->DataBuf.len= DATA_BUFSIZE;
        lpPerIOData->DataBuf.buf = lpPerIOData->buffer;
        ret = WSARecv(connfd, &lpPerIOData->DataBuf, 1, &lpPerIOData->NumberOfBytesRecvd, \
                    &lpPerIOData->Flags, &lpPerIOData->overlap, NULL);
        if (ret == SOCKET_ERROR)
        {
            ret = WSAGetLastError();
            if(ret != WSA_IO_PENDING)
            {
                printf("WSARecv error!\n");
                closesocket(connfd);
            }
        }
        
    }
    
    free(lpPerIOData);
    CloseHandle(CompletionPort);
    closesocket(listenfd);
    WSACleanup();
    return 0;
}

//工作线程
unsigned __stdcall WorkThread( void* param)
{
    int            ret;
    DWORD        BytesTransferred;
    SOCKET        connfd;

    LPPER_IO_OPERATION_DATA lpPerIOData = NULL;

    HANDLE        cport = (HANDLE) param;

    printf( "In work thread...\n" );
    for(;;)
    {
        //等待I/O完成通知
        ret = GetQueuedCompletionStatus(cport, &BytesTransferred, (LPDWORD)&connfd,\
                                        (LPOVERLAPPED *)&lpPerIOData, INFINITE);
        if (ret == 0 || BytesTransferred == 0)
        {
            //close connection
            closesocket(connfd);
        }
        else
        {
            printf("recv %d bytes!\n", BytesTransferred);
            ret = send(connfd, lpPerIOData->buffer, BytesTransferred, 0);
            if (ret == SOCKET_ERROR)
            {
                printf("send error!\n");
                closesocket(connfd);
                continue;
            }
            else
                printf("send %d bytes!\n", ret);
            
            ZeroMemory(&lpPerIOData->overlap, sizeof(OVERLAPPED));
            ret = WSARecv(connfd, &lpPerIOData->DataBuf, 1, &lpPerIOData->NumberOfBytesRecvd,\
                            &lpPerIOData->Flags, &lpPerIOData->overlap, NULL);
            if (ret == SOCKET_ERROR)
            {
                ret = WSAGetLastError();
                if(ret != WSA_IO_PENDING)
                {
                    printf("WSARecv error!\n");
                    closesocket(connfd);
                }
            }
        }
    }
    _endthreadex( 0 );
    return 0;
}