iocpserver.cpp

< WINDOWS网络编程>>英文版,一本详细讲解WINDOWS平台下网络编程的国外经典书籍,适合英文水平高的牛人

int PostRecv(SOCKET_OBJ *sock, BUFFER_OBJ *recvobj)
{
    WSABUF  wbuf;
    DWORD   bytes,
            flags;
    int     rc;


    recvobj->operation = OP_READ;

    wbuf.buf = recvobj->buf;
    wbuf.len = recvobj->buflen;

    flags = 0;

    EnterCriticalSection(&sock->SockCritSec);

    // Assign the IO order to this receive. This must be performned within
    //    the critical section. The operation of assigning the IO count and posting
    //    the receive cannot be interupted.
    recvobj->IoOrder = sock->IoCountIssued;
    sock->IoCountIssued++;

    if (gProtocol == IPPROTO_TCP)
    {
        rc = WSARecv(
                sock->s,
               &wbuf,
                1,
               &bytes,
               &flags,
               &recvobj->ol,
                NULL
                );
    }
    else
    {
        rc = WSARecvFrom(
                sock->s,
               &wbuf,
                1,
               &bytes,
               &flags,
                (SOCKADDR *)&recvobj->addr,
               &recvobj->addrlen,
               &recvobj->ol,
                NULL
                );
    }

    LeaveCriticalSection(&sock->SockCritSec);

    if (rc == SOCKET_ERROR)
    {
        if (WSAGetLastError() != WSA_IO_PENDING)
        {
            dbgprint("PostRecv: WSARecv* failed: %d\n", WSAGetLastError());
            return SOCKET_ERROR;
        }
    }

    // Increment outstanding overlapped operations
    InterlockedIncrement(&sock->OutstandingOps);

    return NO_ERROR;
}

//
// Function: PostSend
// 
// Description:
//    Post an overlapped send operation on the socket.
//
int PostSend(SOCKET_OBJ *sock, BUFFER_OBJ *sendobj)
{
    WSABUF  wbuf;
    DWORD   bytes;
    int     rc;

    sendobj->operation = OP_WRITE;

    wbuf.buf = sendobj->buf;
    wbuf.len = sendobj->buflen;

    EnterCriticalSection(&sock->SockCritSec);

    // Incrmenting the last send issued and issuing the send should not be
    //    interuptable.
    sock->LastSendIssued++;

    if (gProtocol == IPPROTO_TCP)
    {
        rc = WSASend(
                sock->s,
               &wbuf,
                1,
               &bytes,
                0,
               &sendobj->ol,
                NULL
                );
    }
    else
    {
        rc = WSASendTo(
                sock->s,
               &wbuf,
                1,
               &bytes,
                0,
                (SOCKADDR *)&sendobj->addr,
                sendobj->addrlen,
               &sendobj->ol,
                NULL
                );
    }

    LeaveCriticalSection(&sock->SockCritSec);

    if (rc == SOCKET_ERROR)
    {
        if (WSAGetLastError() != WSA_IO_PENDING)
        {
            dbgprint("PostSend: WSASend* failed: %d\n", WSAGetLastError());
            return SOCKET_ERROR;
        }
    }

    // Increment the outstanding operation count
    InterlockedIncrement(&sock->OutstandingOps);

    return NO_ERROR;
}

//
// Function: PostAccept
// 
// Description:
//    Post an overlapped accept on a listening socket.
//
int PostAccept(SOCKET_OBJ *sock, BUFFER_OBJ *acceptobj)
{
    DWORD   bytes;
    int     rc;

    acceptobj->operation = OP_ACCEPT;

    // Create the client socket for an incoming connection
    acceptobj->sclient = socket(sock->af, SOCK_STREAM, IPPROTO_TCP);
    if (acceptobj->sclient == INVALID_SOCKET)
    {
        fprintf(stderr, "PostAccept: socket failed: %d\n", WSAGetLastError());
        return -1;
    }

    rc = sock->lpfnAcceptEx(
            sock->s,
            acceptobj->sclient,
            acceptobj->buf,
            acceptobj->buflen - ((sizeof(SOCKADDR_STORAGE) + 16) * 2),
            sizeof(SOCKADDR_STORAGE) + 16,
            sizeof(SOCKADDR_STORAGE) + 16,
           &bytes,
           &acceptobj->ol
            );
    if (rc == FALSE)
    {
        if (WSAGetLastError() != WSA_IO_PENDING)
        {
            dbgprint("PostAccept: AcceptEx failed: %d\n",
                    WSAGetLastError());
            return SOCKET_ERROR;
        }
    }

    // Increment the outstanding overlapped count for this socket
    InterlockedIncrement(&sock->OutstandingOps);

    return NO_ERROR;
}

//
// Function: InsertPendingSend
// 
// Description:
//    This routine inserts a send buffer object into the socket's list
//    of out of order sends. The routine DoSends will go through this
//    list to issue those sends that are in the correct order.
//
void InsertPendingSend(SOCKET_OBJ *sock, BUFFER_OBJ *send)
{
    BUFFER_OBJ *ptr=NULL,
               *prev=NULL;


    EnterCriticalSection(&sock->SockCritSec);

    send->next = NULL;

    // This loop finds the place to put the send within the list.
    //    The send list is in the same order as the receives were
    //    posted.
    ptr = sock->OutOfOrderSends;
    while (ptr)
    {
        if (send->IoOrder < ptr->IoOrder)
        {
            break;
        }

        prev = ptr;
        ptr = ptr->next;
    }
    if (prev == NULL)
    {
        // Inserting at head
        sock->OutOfOrderSends = send;
        send->next = ptr;
    }
    else
    {
        // Insertion somewhere in the middle
        prev->next = send;
        send->next = ptr;
    }

    LeaveCriticalSection(&sock->SockCritSec);
}

//
// Function: DoSends
//
// Description:
//    This routine goes through a socket object's list of out of order send
//    buffers and sends as many of them up to the current send count. For each
//    send posted, the LastSendIssued is incremented. This means that the next
//    buffer sent must have an IO sequence nubmer equal to the LastSendIssued.
//    This is to preserve the order of data echoed back.
//
int DoSends(SOCKET_OBJ *sock)
{
    BUFFER_OBJ *sendobj=NULL;
    int         ret;

    ret = NO_ERROR;

    EnterCriticalSection(&sock->SockCritSec);

    sendobj = sock->OutOfOrderSends;
    while ((sendobj) && (sendobj->IoOrder == sock->LastSendIssued))
    {
        if (PostSend(sock, sendobj) != NO_ERROR)
        {
            FreeBufferObj(sendobj);
            
            ret = SOCKET_ERROR;
            break;
        }
        sock->OutOfOrderSends = sendobj = sendobj->next;
    }

    LeaveCriticalSection(&sock->SockCritSec);

    return ret;
}

//
// Function: HandleIo
//
// Description:
//    This function handles the IO on a socket. In the event of a receive, the 
//    completed receive is posted again. For completed accepts, another AcceptEx
//    is posted. For completed sends, the buffer is freed.
//
void HandleIo(SOCKET_OBJ *sock, BUFFER_OBJ *buf, HANDLE CompPort, DWORD BytesTransfered, DWORD error)
{
    SOCKET_OBJ *clientobj=NULL;     // New client object for accepted connections
    BUFFER_OBJ *recvobj=NULL,       // Used to post new receives on accepted connections
               *sendobj=NULL;       // Used to post new sends for data received
    BOOL        bCleanupSocket;
    char       *tmp;
    int         i;

    if (error != 0)
        dbgprint("OP = %d; Error = %d\n", buf->operation, error);

    bCleanupSocket = FALSE;

    if ((error != NO_ERROR) && (gProtocol == IPPROTO_TCP))
    {
        // An error occured on a TCP socket, free the associated per I/O buffer
        // and see if there are any more outstanding operations. If so we must
        // wait until they are complete as well.
        //
        FreeBufferObj(buf);

        if (InterlockedDecrement(&sock->OutstandingOps) == 0)
        {
            dbgprint("Freeing socket obj in GetOverlappedResult\n");
            FreeSocketObj(sock);
        }
        return;
    }

    EnterCriticalSection(&sock->SockCritSec);
    if (buf->operation == OP_ACCEPT)
    {
        HANDLE            hrc;
        SOCKADDR_STORAGE *LocalSockaddr=NULL,
                         *RemoteSockaddr=NULL;
        int               LocalSockaddrLen,
                          RemoteSockaddrLen;

        // Update counters
        InterlockedExchangeAdd(&gBytesRead, BytesTransfered);
        InterlockedExchangeAdd(&gBytesReadLast, BytesTransfered);

        // Print the client's addresss
        sock->lpfnGetAcceptExSockaddrs(
                buf->buf,
                buf->buflen - ((sizeof(SOCKADDR_STORAGE) + 16) * 2),
                sizeof(SOCKADDR_STORAGE) + 16,
                sizeof(SOCKADDR_STORAGE) + 16,
                (SOCKADDR **)&LocalSockaddr,
               &LocalSockaddrLen,
                (SOCKADDR **)&RemoteSockaddr,
               &RemoteSockaddrLen
                );

        /*
        printf("Received connection from: ");
        PrintAddress((SOCKADDR *)RemoteSockaddr, RemoteSockaddrLen);
        printf("\n");
        */

        // Get a new SOCKET_OBJ for the client connection
        clientobj = GetSocketObj(buf->sclient, sock->af);

        // Associate the new connection to our completion port
        hrc = CreateIoCompletionPort(
                (HANDLE)buf->sclient,
                CompPort,
                (ULONG_PTR)clientobj,
                0
                );
        if (hrc == NULL)
        {
            fprintf(stderr, "CompletionThread: CreateIoCompletionPort failed: %d\n",
                    GetLastError());
            return;
        }

        // Get a BUFFER_OBJ to echo the data received with the accept back to the client
        sendobj = GetBufferObj(clientobj, BytesTransfered);

        // Copy the buffer to the sending object
        memcpy(sendobj->buf, buf->buf, BytesTransfered);

        // Post the send
        if (PostSend(clientobj, sendobj) == NO_ERROR)
        {
            // Now post some receives on this new connection
            for(i=0; i < gOverlappedCount ;i++)
            {
                recvobj = GetBufferObj(clientobj, gBufferSize);

                if (PostRecv(clientobj, recvobj) != NO_ERROR)
                {
                    // If for some reason the send call fails, clean up the connection
                    FreeBufferObj(recvobj);
                    error = SOCKET_ERROR;
                    break;
                }
            }
        }
        else
        {
            // If for some reason the send call fails, clean up the connection
            FreeBufferObj(sendobj);
            error = SOCKET_ERROR;
        }
        
        // Re-post the AcceptEx
        PostAccept(sock, buf);

		if (error != NO_ERROR)
		{
            if (clientobj->OutstandingOps == 0)
            {
                closesocket(clientobj->s);
                clientobj->s = INVALID_SOCKET;
                FreeSocketObj(clientobj);
            }
            else
            {
                clientobj->bClosing = TRUE;
            }
            error = NO_ERROR;
		}
    }
    else if ((buf->operation == OP_READ) && (error == NO_ERROR))
    {
        //
        // Receive completed successfully
        //
        if ((BytesTransfered > 0) || (gProtocol == IPPROTO_UDP))
        {
            InterlockedExchangeAdd(&gBytesRead, BytesTransfered);
            InterlockedExchangeAdd(&gBytesReadLast, BytesTransfered);

            // Create a buffer to send
            sendobj = GetBufferObj(sock, gBufferSize);

            if (gProtocol == IPPROTO_UDP)
            {
                memcpy(&sendobj->addr, &buf->addr, buf->addrlen);
            }

            // Swap the buffers (i.e. buffer we just received becomes the send buffer)
            tmp              = sendobj->buf;
            sendobj->buflen  = BytesTransfered;
            sendobj->buf     = buf->buf;
            sendobj->IoOrder = buf->IoOrder;

            buf->buf    = tmp;
            buf->buflen = gBufferSize;

            InsertPendingSend(sock, sendobj);

            if (DoSends(sock) != NO_ERROR)
            {
                error = SOCKET_ERROR;
            }
            else
            {
                // Post another receive
                if (PostRecv(sock, buf) != NO_ERROR)
                {
                    // In the event the recv fails, clean up the connection
                    FreeBufferObj(buf);
                    error = SOCKET_ERROR;
                }
            }