Windows Socket I/O模型之 阻塞模式

#include <winsock2.h>
#include <ws2tcpip.h>

#include "resolve.h"
#include "public.h"

typedef SINGLE_LIST_HEADER BuffHeader;

typedef struct _CONNECTION_OBJ
{
    SOCKET s; // Client socket

    HANDLE hRecvSema; // Semaphore incremented for each receive


    BuffHeader buff;
} CONNECTION_OBJ;

volatile LONG gConnectedClients = 0;



CONNECTION_OBJ* GetConnectObj(SOCKET s) {
    CONNECTION_OBJ *newobj = NULL;

    newobj = (CONNECTION_OBJ *)HeapAlloc(GetProcessHeap(), HEAP_ZERO_MEMORY, sizeof(CONNECTION_OBJ));
    if (newobj == NULL) {
        fprintf(stderr, "heapalloc failed.\n");
        ExitProcess(-1);
    }

    newobj->s = s;
    newobj->hRecvSema = CreateSemaphore(NULL, 0, 0x0FFFFFFF, NULL);
    if (newobj->hRecvSema == NULL) {
        fprintf(stderr, "createsimaphore failed.\n");
        ExitProcess(-1);
    }

    newobj->buff.head = newobj->buff.tail = NULL;

    InitializeCriticalSection(&newobj->buff.SendRecvQueueCritSec);

    return newobj;
}

void FreeConnectionObj(CONNECTION_OBJ *obj) {
    DeleteCriticalSection(&obj->buff.SendRecvQueueCritSec);
    HeapFree(GetProcessHeap(), 0, obj);
}

DWORD WINAPI ReceiveThread(LPVOID lpParam) {
    CONNECTION_OBJ *ConnObj=NULL;
    BUFFER_OBJ *BuffObj=NULL;
    int rc;

    ConnObj = (CONNECTION_OBJ *)lpParam;

    if (gProtocol == IPPROTO_TCP) {
        while (true) {
            BuffObj = GetBufferObj(gBufferSize);

            rc = recv(ConnObj->s, BuffObj->buf, BuffObj->buflen, 0);
            BuffObj->buflen = rc;

            EnqueueBufferObj(&ConnObj->buff, &(BuffObj->next));
            
            ReleaseSemaphore(ConnObj->hRecvSema, 1, NULL);

            if (rc == 0 || rc == SOCKET_ERROR) {
                break;
            } else if (rc != SOCKET_ERROR){
                printf("Read Bytes: %d\n", rc);
            }
        }
    }

    ExitProcess(0);
    return 0;
}

DWORD WINAPI SendThread(LPVOID lpParam) {
    CONNECTION_OBJ *ConnObj=NULL;
    BUFFER_OBJ *BuffObj=NULL;
    int rc,
                    nleft,
                    idx;

    SINGLE_LIST* block;

    ConnObj = (CONNECTION_OBJ*)lpParam;

    while (true) {
        rc = WaitForSingleObject(ConnObj->hRecvSema, INFINITE);
        if ((rc == WAIT_FAILED) || (rc == WAIT_TIMEOUT)) {
            fprintf(stderr, "Send thread failed.\n");
            ExitProcess(-1);
        }

        block = DequeueBufferObj(&ConnObj->buff);
        if (block == NULL)
            ExitProcess(-1);
        BuffObj = (BUFFER_OBJ*) container_of(BUFFER_OBJ, next, block);

        if (((gProtocol == IPPROTO_TCP) && (BuffObj->buflen == 0)) || (BuffObj->buflen == SOCKET_ERROR)) {
            FreeBufferObj(BuffObj);
            BuffObj = NULL;
            break;
        }

        if (gProtocol == IPPROTO_TCP) {
            nleft = BuffObj->buflen;
            idx = 0;

            while (nleft > 0) {
                rc = send(ConnObj->s, BuffObj->buf, BuffObj->buflen, 0);
                if (rc == SOCKET_ERROR)
                {
                    break;
                }
                else
                {
                    nleft -= rc;
                    idx += rc;
                }
            }
        }

        FreeBufferObj(BuffObj);
        BuffObj = NULL;
    }

    closesocket(ConnObj->s);
    FreeConnectionObj(ConnObj);
    ExitProcess(-1);

    return 0;
}

DWORD WINAPI ServerListenThread(LPVOID param) {
    CONNECTION_OBJ *ConnObj=NULL;
    HANDLE hThread = NULL;
    SOCKET s;
    int rc;

    s = (SOCKET)param;

    if (gProtocol == IPPROTO_TCP) {
        SOCKADDR_STORAGE saAccept;
        SOCKET ns;
        int acceptlen = sizeof(SOCKADDR_STORAGE);

        rc = listen(s, 200);
        if (rc == SOCKET_ERROR) {
            fprintf(stderr, "listen failed");
            return -1;
        }

        while (true) {
            ns = accept(s, (SOCKADDR*)&saAccept, &acceptlen);
            if (ns == INVALID_SOCKET) {
                fprintf(stderr, "accept failed.\n");
                return -1;
            }

            InterlockedIncrement(&gConnectedClients);

            ConnObj = GetConnectObj(ns);

            hThread = CreateThread(NULL, 0, SendThread, (LPVOID)ConnObj, 0, NULL);
            if (hThread == NULL) {
                fprintf(stderr, "Create thread 0.\n");
                return -1;
            }
            CloseHandle(hThread);

            hThread = CreateThread(NULL, 0, ReceiveThread, (LPVOID)ConnObj, 0, NULL);
            if (hThread == NULL) {
                fprintf(stderr, "Create thread 1.\n");
                return -1;
            }
            CloseHandle(hThread);
        }
    }

    closesocket(s);
    ExitThread(0);
    return 0;
}

int _tmain(int argc, _TCHAR* argv[])
{
    WSADATA wsd;

    SOCKET s[MAX_LISTEN_SOCKETS];
    HANDLE threads[MAX_LISTEN_SOCKETS];

    struct addrinfo *res = NULL;
    struct addrinfo *ptr = NULL;

    int listencount = 0;
    int rc = 0;
    int i;

    if (WSAStartup(MAKEWORD(2, 2), &wsd) != 0) {
        fprintf(stderr, "unable load to winsock!\n");
        return -1;
    }

    res = ResolveAddress(gSrvAddr, gPort, gAddressFamily, gSocketType, gProtocol);
    if (res == NULL) {
        fprintf(stderr, "resolve addr failed.\n");
        return -1;
    }

    ptr = res;
    while (ptr) {
        if (listencount > MAX_LISTEN_SOCKETS) {
            fprintf(stderr, "big than MAX_LISTEN_SOCKETS.\n");
            return -1;
        }

        s[listencount] = socket(ptr->ai_family, ptr->ai_socktype, ptr->ai_protocol);
        if (s[listencount] == INVALID_SOCKET) {
            fprintf(stderr, "socket failed.\n");
            return -1;
        }

        rc = bind(s[listencount], ptr->ai_addr, ptr->ai_addrlen);
        if (rc == SOCKET_ERROR) {
            fprintf(stderr, "bind failed.%d\n", WSAGetLastError());
            return -1;
        }

        threads[listencount] = CreateThread(NULL, 0, ServerListenThread, (LPVOID)s[listencount], 0, NULL);
        if (threads[listencount] == NULL) {
            fprintf(stderr, "Create listen task faileds.\n");
            return -1;
        }

        ++listencount;
        ptr = ptr->ai_next;
    }

    freeaddrinfo(res);

    while (true) {
        rc = WaitForMultipleObjects(listencount, threads, TRUE, INFINITE);
        if (rc == WAIT_TIMEOUT) {
            fprintf(stderr, "wait timeout.\n");
            return -1;
        } else if (rc == WAIT_FAILED) {
            fprintf(stderr, "wait error.\n");
            return -1;
        } else
            break;
    }

    for (i=0; i < listencount ;i++)
    {
        CloseHandle(threads[i]);
    }

    WSACleanup();

    return 0;
}