Windows 完成端口例程

#define _WINSOCK_DEPRECATED_NO_WARNINGS

#include <WinSock2.h>
#include <Mswsock.h>
#include <stdio.h>
#include <string.h>
#include <time.h>
#include <stdlib.h>

#pragma comment(lib, "ws2_32.lib")

#define MAX_BUFFER        8192

#define TIMEOUT            1800

typedef BOOL(WINAPI *LPFAcceptEx)(SOCKET sListenSocket, SOCKET sAcceptSocket, PVOID lpOutputBuffer, DWORD dwReceiveDataLength,
    DWORD dwLocalAddressLength, DWORD dwRemoteAddressLength, LPDWORD lpdwBytesReceived, LPOVERLAPPED lpOverlapped);

struct client_t {
    OVERLAPPED ov;
    SOCKET sock;
    WSABUF _buf;
    unsigned char data[MAX_BUFFER];//数据缓冲区
    time_t timeout;
    struct client_t* next;
    struct client_t* prev;
};


struct server_t {
    SOCKET sock;
    HANDLE iocp;
    LPFAcceptEx accept;
    struct client_t* head;
    struct client_t* tail;
};

static void client_exit(struct client_t* pc)
{
    closesocket(pc->sock);
    //TODO:断开连接
    printf("断开连接：%d\t%X\n", pc->sock, pc);
    free(pc);
}

static void insert_into_list(struct server_t* ps, struct client_t* pc)
{
    if (NULL == ps->head || NULL == ps->tail) {
        pc->next = pc->prev = NULL;
        ps->head = ps->tail = pc;
    } else {
        pc->prev = ps->tail;
        pc->next = NULL;
        ps->tail->next = pc;
        ps->tail = pc;
    }
}

static void delete_from_list(struct server_t* ps, struct client_t* pc)
{
    struct client_t* next = pc->next;
    struct client_t* prev = pc->prev;

    if (NULL == next) {
        ps->tail = prev;
    } else {
        next->prev = prev;
    }
    if (NULL == prev) {
        ps->head = next;
    } else {
        prev->next = next;
    }
}

static void server_exit(struct server_t* ps)
{
    struct client_t* pc = ps->head, *temp = pc;
    while (pc) {
        temp = pc->next;
        CancelIo((HANDLE)pc->sock);
        client_exit(pc);
        pc = temp;
    }
    CancelIo((HANDLE)ps->sock);
    CloseHandle(ps->iocp);
    closesocket(ps->sock);
    free(ps);
}

static SOCKET init_socket(const char* ip, unsigned short port)
{
    SOCKET sock = WSASocket(AF_INET, SOCK_STREAM, 0, NULL, 0, WSA_FLAG_OVERLAPPED);
    if (INVALID_SOCKET == sock) {
    _e1:
        return INVALID_SOCKET;
    }
    struct sockaddr_in addr_in;
    memset(&addr_in, 0, sizeof(struct sockaddr_in));
    addr_in.sin_addr.S_un.S_addr = inet_addr(ip);
    addr_in.sin_family = AF_INET;
    addr_in.sin_port = htons(port);
    if (SOCKET_ERROR == bind(sock, (struct sockaddr*)&addr_in, sizeof(struct sockaddr_in))) {
    _e2:
        closesocket(sock);
        goto _e1;
    }
    if (SOCKET_ERROR == listen(sock, SOMAXCONN)) {
        goto _e2;
    }
    return sock;
}

static int post_accept(struct server_t* ps)
{
    SOCKET sock = WSASocket(AF_INET, SOCK_STREAM, 0, NULL, 0, WSA_FLAG_OVERLAPPED);
    if (INVALID_SOCKET == sock) {
    _e1:
        return -1;
    }
    struct client_t* pc = (struct client_t*)malloc(sizeof(struct client_t));
    if (NULL == pc) {
    _e2:
        closesocket(sock);
        goto _e1;
    }
    memset(pc, 0, sizeof(struct client_t));
    pc->sock = sock;
    pc->_buf.buf = pc->data;
    pc->_buf.len = MAX_BUFFER;

    DWORD rec = 0;
    if (FALSE == ps->accept(ps->sock, pc->sock, pc->data, 0, sizeof(struct sockaddr) + 16, 
        sizeof(struct sockaddr) + 16, &rec, &pc->ov) && ERROR_IO_PENDING != WSAGetLastError()) {
        free(pc);
        goto _e2;
    }
    return 0;
}

static int post_recv(struct client_t* pc)
{
    memset(pc->data, 0, sizeof(pc->data));
    DWORD rec = 0, flag = 0;
    int ret = WSARecv(pc->sock, &pc->_buf, 1, &rec, &flag, &pc->ov, NULL);
    if (SOCKET_ERROR == ret && WSA_IO_PENDING != WSAGetLastError()) {
        return -1;
    }
    return 0;
}

static void accept_client(struct server_t* ps, struct client_t* pc)
{
    do {
        if (NULL == CreateIoCompletionPort((HANDLE)pc->sock, ps->iocp, pc->sock, 0)) {
        _e1:
            client_exit(pc);
            break;
        }
        if (-1 == post_recv(pc)) {
            goto _e1;
        }
        pc->timeout = time(NULL) + TIMEOUT;
        insert_into_list(ps, pc);
        //TODO:建立连接
        printf("建立连接：%d\t%X\n", pc->sock, pc);
    } while (0);
    post_accept(ps);
}

static void data_transfer(struct server_t* ps, struct client_t* pc, int size)
{
    delete_from_list(ps, pc);
    do {
        if (0 == size) {    
        _e1:
            client_exit(pc);
            break;
        }
        //TODO: 数据处理
        printf("数据处理：%d\t%X\n", pc->sock, pc);
        memset(pc->data, 0, sizeof(pc->data));
        if (-1 == post_recv(pc)) {
            goto _e1;
        }
        pc->timeout = time(NULL) + TIMEOUT;
        insert_into_list(ps, pc);
    } while (0);
}

static int timer(struct server_t* ps)
{
    struct client_t* pc = ps->head, *temp = pc;
    time_t now = time(NULL);
    while (pc && pc->timeout <= now) {
        temp = pc->next;
        CancelIo((HANDLE)pc->sock);
        pc = temp;
    }
    if (NULL == pc) {
        ps->head = ps->tail = NULL;
        return INFINITE;
    } else {
        ps->head = pc;
        pc->prev = NULL;
        int timeout = pc->timeout - now;
        return 0 >= timeout ? 0 : 1000 * timeout;
    }
}

void start_server(struct server_t* ps)
{
    DWORD size = 0, timeout = INFINITE;
    ULONG_PTR socket_ctx = 0;
    LPOVERLAPPED pov = NULL;
    
    while (1) {
        BOOL ret = GetQueuedCompletionStatus(ps->iocp, &size, &socket_ctx, &pov, timeout);
        struct client_t* pc = CONTAINING_RECORD(pov, struct client_t, ov);
        if (NULL != pc) {
            if (FALSE == ret) {
                delete_from_list(ps, pc);
                client_exit(pc);
            } else {
                if (socket_ctx == ps->sock) {
                    accept_client(ps, pc);
                } else if (0 == socket_ctx) {
                    break;
                } else {
                    data_transfer(ps, pc, size);
                }
            }
        }
        timeout = timer(ps);
    }
    server_exit(ps);
}


int init_server(const char* ip, unsigned short port, struct server_t* ps)
{
    memset(ps, 0, sizeof(struct server_t));
    ps->iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);
    if (NULL == ps->iocp) {
    _e1:
        return -1;
    }
    ps->sock = init_socket(ip, port);
    if (INVALID_SOCKET == ps->sock) {
    _e2:
        CloseHandle(ps->iocp);
        goto _e1;
    }
    if (NULL == CreateIoCompletionPort((HANDLE)ps->sock, ps->iocp, ps->sock, 0)) {
    _e3:
        closesocket(ps->sock);
        goto _e2;
    }
    GUID guid = WSAID_ACCEPTEX;
    DWORD ret = 0;
    if (SOCKET_ERROR == WSAIoctl(ps->sock, SIO_GET_EXTENSION_FUNCTION_POINTER, &guid,
        sizeof(guid), &ps->accept, sizeof(ps->accept), &ret, NULL, NULL) || NULL == ps->accept) {
        goto _e3;
    }
    if (-1 == post_accept(ps)) {
        goto _e3;
    }
    return 0;
}

void close_client(struct server_t* ps, struct client_t* pc)
{
    delete_from_list(ps, pc);
    CancelIo((HANDLE)pc->sock);
}

void stop_server(struct server_t* ps)
{
    PostQueuedCompletionStatus(ps->iocp, 0, 0, NULL);
}

int main()
{
    WSADATA ws;
    WSAStartup(MAKEWORD(2, 2), &ws);
    struct server_t gs;
    init_server("127.0.0.1", 9080, &gs);

    start_server(&gs);

    WSACleanup();
}