socket.c

Go to the documentation of this file.

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <ctype.h>
#include "ctl/ctldef.h"
#include "ctl/pool.h"
#include "ctl/ctlnew.h"
#include "ctl/socket.h"

/* Some things to glue Winsock/BSD Sockets together more nicely */
#ifdef WIN32
#include <ws2tcpip.h>
#define EWOULDBLOCK         WSAEWOULDBLOCK
#define SHUT_RDRW           SD_BOTH
#define SocketError()       WSAGetLastError()
#define SocketSetError(e)   WSASetLastError(e)
#else
#define INVALID_SOCKET      -1
#define SOCKET_ERROR        -1
#define closesocket         close
#define SocketError()       errno
#define SocketSetError(e)   (errno = (e))
#include <fcntl.h>
#include <stropts.h>
#endif

/* Something to write at ends of buffers to detect overwrites */
static const uint32 GUARD_BYTES = 0xDEADBAB4; 

/*
 * Declare socket memory pool
 */
ctl_fp_decl( Socket, MAX_SOCKETS, SocketPool );
static SocketPool* sockPool;

/*
 * Declare list of active Sockets
 */
dll_autolist(Socket,active, socksActive);
static void Socket_Free( Socket* sock );
/*
 * Declare testpoint data 
 */
static size_t cSockActive = 0;
static size_t cSockFree = MAX_SOCKETS;
/*
 * Declare socket buffer pool
 */
ppWindows(static WSADATA wsaData;)
static const size_t cSockTotal = MAX_SOCKETS;
static size_t sockInSize;
static size_t sockOutSize;
static size_t sockHeapSize;
static size_t sockBufferSize;
static size_t sockOffsetOut;
static size_t sockOffsetEnd;
static size_t bufferSize;
static uint8* bufferSpace = NULL;   /* Space reserved for socket I/O buffers */
static size_t sockUserSize;
static size_t userSize;
static uint8* userSpace = NULL;     /* Space reserved for user heaps */
#define SOCK_INPUT_DEFAULTPTR(index)    (bufferSpace + (index * sockBufferSize))
#define SOCK_OUTPUT_DEFAULTPTR(index)   ((bufferSpace + (index * sockBufferSize)) + sockOffsetOut)
#define SOCK_HEAP_DEFAULTPTR(index)     (userSpace + (index * sockUserSize))

uint32 SocketID( const Socket* sock )
{
    assertobjconst(sock);
    if( !ctl_fp_valid(SocketPool, sockPool, sock ) )
        return 0;
    if( Socket_ShouldIgnore(sock) )
        return 0;
    return sock->id;
}

Socket* SocketLut( uint32 id )
{
    Socket* sock = ctl_fp_lut(SocketPool, sockPool, id&0xffff );
    if( NULL == sock )
        return NULL;
    if( sock->id != id )
        return NULL;
    if( Socket_ShouldIgnore(sock) )
        return NULL;
    return sock;
}

Socket* SocketFind( const char* label )
{
    assertconst(label,1);
    {
        socket_foreach( sock )
        {
            assertobjptr(sock);
            if( !Socket_ShouldIgnore(sock) && !strcmp(sock->label,label) )
            {
                return sock;
            }
        }
    }
    return NULL;
}

/* Default do-nothing socket callbacks */
static void Socket_DefaultReceived( Socket* sock, ctl_serial* packet )
{
    assertobjptr(sock);
    assertobjptr(packet);
}
/* Default do-nothing socket callbacks */
static void Socket_DefaultCallback( Socket* sock )
{
    assertobjptr(sock);
}

static NewSocketCallback login_callback = Socket_DefaultCallback;
static NewSocketCallback user_callback = Socket_DefaultCallback;

NewSocketCallback SocketsSetLogin( NewSocketCallback callback )
{
    NewSocketCallback cb_old = login_callback;
    login_callback = NULL == callback ? Socket_DefaultCallback : callback;
    assertcodeptr(login_callback);
    return cb_old;
}

NewSocketCallback SocketsSetUser( NewSocketCallback callback )
{
    NewSocketCallback cb_old = user_callback;
    user_callback = NULL == callback ? Socket_DefaultCallback : callback;
    assertcodeptr(user_callback);
    return cb_old;
}


bool SocketsInit( size_t input, size_t output, size_t heap )
{
    trace(("SocketsInit()\n"));
    sockInSize = input;
    sockOutSize = output;
    sockHeapSize = heap;
    sockBufferSize = (sockInSize + sockOutSize + (2*sizeof(uint32)));
    sockOffsetOut = sockInSize + sizeof(uint32);
    sockOffsetEnd = sockOffsetOut + sizeof(uint32);
    bufferSize = MAX_SOCKETS * sockBufferSize;
    sockUserSize = sockHeapSize + sizeof(uint32);
    userSize = MAX_SOCKETS * sockUserSize;

#ifdef WIN32
    {
        /* Setup Winsock */
        int iResult = WSAStartup(MAKEWORD(2,2), &wsaData);
        if (iResult != 0) 
        {
            SocketsLogLastError( LOG_ERROR, "WSAStartup failed: %d\n", iResult );
            return false;
        }
        LOG( LOG_STARTUP,"%s - %s\n", wsaData.szDescription, wsaData.szSystemStatus  );
    }
#endif

    /* Initialize socket pool */
    ctl_alloctype(sockPool,SocketPool);
    ctl_fp_init( SocketPool, sockPool );
    /* 
     * Allocate all the socket storage at once
     * I'd rather die right away for insufficient space than at 2:00am when the 834th user logs in
     */
    ctl_malloc( bufferSpace, bufferSize );
    ctl_malloc( userSpace, userSize );
    return true;
}

void SocketsShutdown(void)
{
    trace(("SocketsShutdown()\n"));
    {
        /* Shutdown all sockets */
        socket_foreach( sock )
        {
            assertobjptr(sock);
            Socket_Close(sock);
        }
    }
    {
        /* Free all sockets */
        socket_foreach_deletable( sock )
        {
            assertobjptr(sock);
            Socket_Free(sock);
        }
    }
    /* Free socket pool */
    ctl_free(sockPool);
    /* Free reserved socket buffers */
    ctl_free(bufferSpace);
    /* Free reserved user data */
    ctl_free(userSpace);
    /* Make sure active list is clear */
    dll_initlist(Socket,active, socksActive);
    /* Reset statistics */
    assert( 0 == cSockActive );
    assert( MAX_SOCKETS == cSockFree );
    cSockActive = 0;
    cSockFree = MAX_SOCKETS;
#ifdef WIN32
    WSACleanup();
#endif
}

bool SocketsCycle( unsigned ms )
{
    mstime tick = ctl_mstime();
#ifdef WIN32
    /*
     * Under Windows, select() SUCKS.  Under WinSock, fd_set is an array of  
     * handles that is searched linearly.  A call to FD_SET in that case could  
     * scan through hundreds, even thousands of handles every time, times  
     * hundreds, even thousands of handles.  It makes it much more efficient to  
     * poll every socket and Sleep(1), even with the OS context switches.   
     * The replacement in the Winsock API is to set up threads for every  
     * socket, or have a WindowProc receive a message, or there's one that will 
     * check up to 64 handles at a time....  Just ugly.
     * 
     * For WIN32, we yield up time to the rest of the OS, since this will 
     * otherwise be a hard loop.  Makes CPU usage (on a given core) go from 
     * 100% to around 3%, so though not quite as good as select, it will do 
     * to keep the system running for testing in a Windows environment.
     */
    Sleep( 1 );

    /* Free dead sockets */
    {
        socket_foreach_deletable( sock )
        {
            assertobjptr(sock);
            if( tick - sock->lastmsg > sock->timeout )
            {   /* Close timed-out sockets */
                SocketLogIP(sock,LOG_INFO,"Timed out:");
                Socket_Close(sock);
            }
            if( stUninitialized == sock->type )
            {
                Socket_Free(sock);
            }
        }
    }
#else
    /*
     * Under Linux/Unix/etc. select() is THE way to go.  The server can go to 
     * sleep nicely when there's nothing to do, and you don't have to do  
     * thousands of context switches between application and OS to see if  
     * anything needs to happen.  The fd_set structure and macros under a BSD  
     * implementation are a set of bit masks.  Accessing the list is an indexed
     * lookup with some masking, and entries in the list are one bit, so the 
     * data is 1/32 the size of the list of handles you get in Windows, too.
     */
    static fd_set fdreads;
    static fd_set fdwrites;
    /* Maybe see about cacheing fd_set data...? */
    FD_ZERO(&fdreads);
    FD_ZERO(&fdwrites);
    {
        int nfds = -1;
        socket_foreach_deletable( sock )
        {
            assertobjptr(sock);
            if( stUninitialized != sock->type && INVALID_SOCKET != sock->socket )
            {
                if( tick - sock->lastmsg > sock->timeout )
                {   /* Close timed-out sockets */
                    SocketLogIP(sock,LOG_INFO,"Timed out:");
                    Socket_Close(sock);
                }
                else
                {
                    SOCKET sockCurr = sock->socket;
                    if( stSendingFinal != sock->type )
                    {
                        /* Don't listen for more incoming messages from closing sockets */
                        FD_SET( sockCurr, &fdreads );
                    }
                    if( !RingBuff_IsEmpty(&sock->output) )
                    {
                        FD_SET( sockCurr, &fdwrites );
                    }
                    if( sock->socket > nfds )
                    {
                        nfds = sockCurr;
                    }
                }
            }
            if( stUninitialized == sock->type )
            {
                Socket_Free(sock);
            }
        }
        /* Do select on all sockets, time-out appropriately as passed in */
        {
            struct timeval timeout = { ms/1000, (ms%1000)*1000 };
            nfds = select( 1+nfds, &fdreads, &fdwrites, NULL, 0 == ms ? NULL : &timeout );
        }
        if( 0 == nfds )
        {
            return true;
        }
        else if( SOCKET_ERROR == nfds )
        {
            int save_errno = SocketError();
            switch( save_errno )
            {
            case EBADF:
                /*
                 * One handle might be invalid - scan them all
                 * Shouldn't ever happen, but if one fd somehow becomes invalid,
                 * should we crash & nuke everyone, or just the 'invalid' one?
                 * I'll just assume you want to close the invalid one.
                 */
                {
                    SocketsLogLastError( LOG_ERROR, "select() EBADF.\n" );
                    bool bFound = false;
                    socket_foreach_deletable( suspect )
                    {
                        int tmp;
                        int ret = ioctl( suspect->socket, I_GETSIG, &tmp );
                        if( (-1 == ret) && (EBADF == SocketError()) )
                        {
                            Socket_LogError( suspect, LOG_ERROR, "select() BAD HANDLE: %d.\n", suspect->socket );
                            Socket_Close( suspect );
                            bFound = true;
                        }
                    }
                    /* If we couldn't find the 'bad' socket, something else is 
                     * wrong, and that's probably fatal. */
                    return bFound;
                }
                break;
            case EAGAIN:
            case EINTR:
                /* Time-out. */
                break;
            case EINVAL:
                
                /* nfds arg might be <0 or >FD_SETSIZE */
                if( (1+nfds < 0) || (1+nfds > FD_SETSIZE) )
                {
                    SocketsLogLastError( LOG_ERROR, "select() nfds out of range.\n" );
                    return false;
                }
                
                /* Invalid timeout interval (not likely, and a bug, anyways) */

                /* STREAM or multiplexer downstream from multiplexer */
                
                /* Fall-through to default error handler */
            default:
                SocketsLogLastError( LOG_ERROR, "select() failed.\n" );
                // Make sure previous function(s) don't overwrite error status
                SocketSetError(save_errno);
                return false;
            }
        }
    }

#endif

    /* Attempt to write sockets first, to make room for side-effects of reading sockets */
    {
        socket_foreach( sock )
        {
            assertobjptr(sock);
            if( stUninitialized != sock->type ppNotWindows(&& FD_ISSET(sock->socket,&fdwrites)) )
            {
                assertcodeptr(sock->vtab->WriteCycle);
                sock->vtab->WriteCycle(sock);
            }
        }
    }

    /* Attempt to read sockets */
    {
        socket_foreach( sock )
        {
            assertobjptr(sock);
            if( !Socket_ShouldIgnore(sock) ppNotWindows(&& FD_ISSET(sock->socket,&fdreads)) )
            {
                assertcodeptr(sock->vtab->ReadCycle);
                sock->vtab->ReadCycle(sock);
            }
        }
    }

    /* Normal-looking status */
    return true;
}

const char* SocketsLastErrorString(void)
{
#ifdef WIN32
    static char szError[512]; // <- Multithread risk
    DWORD dw = SocketError(); 
    FormatMessage(
        FORMAT_MESSAGE_FROM_SYSTEM | 
        FORMAT_MESSAGE_IGNORE_INSERTS |
        FORMAT_MESSAGE_MAX_WIDTH_MASK,
        NULL, 
        dw, 
        MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
        szError, sizeof(szError)/sizeof(*szError), 
        NULL);
    return szError;
#else
    return strerror(SocketError());
#endif
}

void SocketsLogLastError( unsigned mask, const char*fmt, ... )
{
    va_list args;
    va_start( args, fmt );
    assertconst(fmt,1);
    debug_vlog(mask, fmt, args );
    LOG( mask, "%s\n", SocketsLastErrorString() );
}

struct addrinfo *SocketsParseAddress( const char* szAddress )
{
    char ipbuff[256];
    const char* port;
    const char* address;
    const char* scan;
    assertconst(szAddress,1);
    scan = address = szAddress;
    /* Look for colon */
    while( ':' != *scan )
    {
        if( 0 == *scan )
            return NULL;
        scan++;
    }
    /* If we found '://', skip over protocol and search for ':' */
    if( *scan == '/' )
    {
        ++scan;
        if( *scan == '/' )
        {
            address = ++scan;
            while( ':' != *scan )
            {
                if( 0 == *scan )
                {
                    LOG( LOG_ERROR, "SocketsParseAddress(%s) no ':port'.\n", szAddress );
                    return NULL;
                }
                scan++;
            }
        }
        else
        {
            LOG( LOG_ERROR, "SocketsParseAddress(%s) found ':/' instead of '://'.\n", szAddress );
            return NULL;
        }
    }
    if( *scan == ':' )
    {
        size_t address_len;
        port = scan+1;
        address_len = (scan-address);
        if( address_len > sizeof(ipbuff) )
        {
            LOG( LOG_ERROR, "SocketsParseAddress(%s) address too long.\n", szAddress );
            return NULL;
        }
        memcpy(ipbuff,address,address_len);
        ipbuff[address_len] = 0;
        SocketsLogHostent( LOG_INFO, ipbuff );
        {
            struct addrinfo aiHints;
            struct addrinfo *aiList = NULL;
            memset(&aiHints, 0, sizeof(aiHints));
            aiHints.ai_family = AF_INET;
            aiHints.ai_socktype = SOCK_STREAM;
            aiHints.ai_flags = AI_CANONNAME;
            aiHints.ai_protocol = IPPROTO_TCP;
            if( 0 != getaddrinfo(ipbuff, port, &aiHints, &aiList) )
            {
                SocketsLogLastError( LOG_ERROR, "SocketsParseAddress(%s) failed.\n", szAddress );
                return NULL;
            }
#if 1
            {
                struct addrinfo *ai = aiList;
                while( NULL != ai )
                {
                    trace(( "%s: ", ai->ai_canonname ));
                    traceip("Address:",ai->ai_addr);
                    ai = ai->ai_next;
                }
            }
#endif
            return aiList;
        }
    }
    return NULL;
}

void SocketsLogHostent( uint32 logmask, const char* name )
{
    /* Mine connection details out of sockets */
    struct hostent* hostEntry;
    assertconst(name,1);
    hostEntry = gethostbyname(name);
    if( NULL == hostEntry )
    {
        struct in_addr ina;
#ifdef WIN32
        uint32 inu = inet_addr(name);
        ina = *((struct in_addr*)&inu);
#else
        if( 0 == inet_aton( name, &ina) )
        {
            SocketsLogLastError( LOG_ERROR, "inet_aton  failed.\n" );
            return;
        }
#endif
        hostEntry = gethostbyaddr( ppWindows((const char*))&ina, sizeof(ina), AF_INET );
        if( NULL == hostEntry )
        {
            SocketsLogLastError( LOG_ERROR, "gethostbyaddr  failed.\n" );
            return;
        }
    }
    LOG( logmask, "Official name: %s\n", hostEntry->h_name);
    LOG( logmask, "Alternate names:\n");
    {
        const char** p = (const char** )hostEntry->h_aliases;
        while( *p )
        {
            LOG( logmask, "\t%s\n", *p++ );
        }
    }
    LOG( logmask, "Address type: ");
    switch (hostEntry->h_addrtype) 
    {
    case AF_INET:
        LOG( logmask, "AF_INET\n");
        break;
    case AF_INET6:
        LOG( logmask, "AF_INET6\n");
        break;
    default:
        LOG( logmask, " %d\n", hostEntry->h_addrtype);
        break;
    }
    /*LOG( logmask, "Address length: %d\n", hostEntry->h_length); */
    if( hostEntry->h_length >= 4 )
    {
        struct in_addr **p = (struct in_addr **)hostEntry->h_addr_list;
        LOG( logmask, "Addresses Available:\n" );
        while(*p)
        {
            LOG( logmask, "\t%s\n", inet_ntoa(**p) );
            ++p;
        }
    }
}


static void Socket_Options( Socket* sock )
{
    assertobjptr(sock);
#ifdef WIN32
    { 
        /* Set non-blocking */
        uint32 iMode = 1; 
        if( SOCKET_ERROR == ioctlsocket(sock->socket, FIONBIO, &iMode) )
        {
            SocketsLogLastError( LOG_ERROR, "%s ioctlsocket(%d)n", sock->label, sock->socket );
        }
    }
#if 0
    {
        /* Needs <mstcpip.h> */
        /* Ping connection to make sure it's alive */
        BOOL bOptVal = TRUE;
        DWORD dwError = 0L ;
        DWORD dwBytes = 0L ;
        struct tcp_keepalive sKA_Settings = {0}, sReturned = {0} ;
        setsockopt( sock->socket, SOL_SOCKET, SO_KEEPALIVE, (char*)&bOptVal, sizeof(BOOL) );
        sKA_Settings.onoff = 1 ;
        sKA_Settings.keepalivetime = 30000 ; /* Keep Alive in 30 sec. */
        sKA_Settings.keepaliveinterval = 5000 ; /* Resend if No-Reply */
        if (WSAIoctl( sock->socket, SIO_KEEPALIVE_VALS, &sKA_Settings,
            sizeof(sKA_Settings), &sReturned, sizeof(sReturned), &dwBytes,
            NULL, NULL) != 0)
        {
            SocketsLogLastError( LOG_ERROR, "Keepalive setup(%s)\n", sock->label );
        }
    }
#endif
#else
    {
        /* Set non-blocking */
        int flags = fcntl( sock->socket, F_GETFL, 0 );
        if (-1 == flags )
        {
            SocketsLogLastError( LOG_ERROR, "%s fcntl(%d)n", sock->label, sock->socket );
        }
        else
        {
            flags |= O_NONBLOCK;
            fcntl( sock->socket, F_SETFL, flags  );
        }
    }
#endif
}


static void Generic_WriteCycle( Socket* sock )
{
    int attempts;
    assertobjptr(sock);
    for( attempts = 0; attempts < 2; ++attempts )
    {
        size_t remain;
        if( (remain = RingBuff_Remain( &sock->output )) )
        {
            const uint8* buff;
            RingBuff_Peek( &sock->output, &buff, remain );
            int bytes = send( sock->socket, buff, remain, 0 );
            if( SOCKET_ERROR == bytes )
            {
                switch( SocketError() )
                {
                case 0:
                    /* No error */
                case EWOULDBLOCK:
                    /* No space to send */
                    return;
                default:
                    /* Some error or other */
                    SocketsLogLastError( LOG_ERROR, "Generic_WriteCycle(%s) receive failed\n", sock->label );
                    Socket_Close( sock );
                    return;
                }
            }
            else if( bytes > 0 )
            {
                /* Consume sent bytes off queue */
                RingBuff_Pop( &sock->output, &buff, bytes );
                if( bytes < remain )
                {
                    /* Don't loop on partial send */
                    return;
                }
            }
            if( stSendingFinal == sock->type && RingBuff_IsEmpty(&sock->output) )
            {
                /* Complete graceful shutdown */
                Socket_Close( sock );
                return;
            }
        }
        else
        {
            /* Nothing to do here. */
            return;
        }
    }
}

static void Generic_ReadCycle( Socket* sock )
{
    /*
     * This is based on reading one, and only one packet from the input stream at a time.
     *
     * I don't want to fill local memory with unhandled packets, giving the 
     * network layer permission to ask the client for more.  
     *
     * Let them back up on the client.  This is a single-threaded environment 
     * and we're popping one command every other cycle not only because it's
     * easier, but because I don't want assholes to write their own clients
     * (very easy to do) and flood everybody else out while 'experimenting'.
     */
    /* Try to read data into buffer */
    size_t available;
    assertobjptr(sock);
    available = RingBuff_Free(&sock->input);
    if( 0 < available )
    {   /* Try to get non-zero bytes when there's space */
        uint8* space;
        RingBuff_Space( &sock->input, &space, available );
        int bytes = recv( sock->socket, space, available, 0 );
        Socket_Validate(sock);
        if( 0 == bytes )
        {   /* Disconnected */
            Socket_Close( sock );
            return;
        }
        if( SOCKET_ERROR == bytes )
        {
            if( EWOULDBLOCK != SocketError() )
            {
                SocketsLogLastError( LOG_ERROR, "Generic_ReadCycle(%s) receive failed\n", sock->label );
                Socket_Close( sock );
            }
            return;
        }
        //debug_loghex( LOG_INFO, sock->input.tail, bytes );
        RingBuff_Add( &sock->input, &space, (size_t)bytes );
    }
    ctl_serial serial;
    if( RingBuff_PopSerial( &sock->input, &serial ) )
    {
        sock->lastmsg = ctl_mstime();
        sock->appReceived( sock, &serial );
    }
}

/*
 * Client connection has been shut down
 */
static void Client_Close( Socket* sock )
{
}
SockVTAB ClientVTAB =
{
    Generic_ReadCycle,
    Generic_WriteCycle,
    Client_Close
};


Socket* Socket_Create_Client( const char* address )
{
    Socket* sock;
    assertconst(address,1);
    sock = Socket_Create();
    if( NULL == sock )
        return NULL;
    sock->type = stClient;
    sock->timeout = int64_max;
    LOG( LOG_INFO, "Socket_Create_Client(%s)\n", address );
    sock->vtab = &ClientVTAB;
    strncpy(sock->label, "Client", countof(sock->label) );
    return sock;
}
Socket* Socket_Client_Connect( Socket* sock, const char* address )
{
    /* Resolve address and connect */
    struct addrinfo * service = SocketsParseAddress( address );
    if( NULL == service )
    {
        Socket_Close( sock );
        return NULL;
    }
    sock->socket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
    if( sock->socket == INVALID_SOCKET ) 
    {
        SocketsLogLastError( LOG_ERROR, "Socket_Create_Client(%s) socket failed\n", address );
        Socket_Close( sock );
        return NULL;
    }
    /* Setup socket */
    Socket_Options( sock );
    memcpy( &sock->socka_buff, service->ai_addr, min(service->ai_addrlen,sizeof(sock->socka_buff)) );
    LOG( LOG_INFO, "Client: %s\n", service->ai_canonname );
    freeaddrinfo(service);
    SocketLogIP(sock,LOG_INFO,"Address: ");
    if( SOCKET_ERROR == connect( sock->socket, (struct sockaddr*)&sock->address, sizeof(sock->address)) ) 
    {
        SocketsLogLastError( LOG_ERROR, "Socket_Create_Client(%s) connect failed.\n", address );
        Socket_Close( sock );
        return NULL;
    }
    return sock;
}
/*
 * Accept incoming connections and make Logins out of them
 */
static void Listener_ReadCycle(struct Socket* sock)
{
    /* Accept any incoming connection. */
    SOCKET client;
    assertobjptr(sock);
    client = accept( sock->socket, NULL, NULL );
    if( INVALID_SOCKET == client ) 
    {
        if( EWOULDBLOCK != SocketError() )
        {
            SocketsLogLastError( LOG_ERROR, "Listener_ReadCycle(%s) accept failed\n", sock->label );
        }
    }
    else
    {
        sock->lastmsg = ctl_mstime();
        Socket_Create_Login( client );
    }
}
/*
 * Listener has been shut down
 */
static void Listener_Close( Socket* sock )
{
    LOG( LOG_INFO, "%s Closed\n", sock->label );
}
/*
 * Listener does not write
 */
static void Listener_WriteCycle( Socket* sock )
{
    RingBuff_Reset(&sock->output);
}
SockVTAB ListenerVTAB =
{
    Listener_ReadCycle,
    Listener_WriteCycle,
    Listener_Close
};

Socket* Socket_Create_Listener( const char* label, const char* address )
{
    Socket* sock;
    struct addrinfo * service;
    assertconst(label,1);
    assertconst(address,1);
    LOG(LOG_STARTUP,"Socket_Create_Listener(%s,%s)\n", label, address );
    service = SocketsParseAddress( address );
    if( NULL == service )
        return NULL;
    sock = Socket_Create();
    if( NULL == sock )
        return NULL;
    sock->vtab = &ListenerVTAB;
    sock->type = stListener;
    sock->timeout = int64_max;
    strncpy( sock->label, label, countof(sock->label) );
    sock->socket = socket( AF_INET, SOCK_STREAM, IPPROTO_TCP);
    if( sock->socket == INVALID_SOCKET ) 
    {
        SocketsLogLastError( LOG_ERROR, "Socket_Create_Listener(%s,%s) socket failed\n", sock->label,address );
        Socket_Close( sock );
        return NULL;
    }
    /* Setup socket */
    Socket_Options( sock );
    {
        /* Resolve port and listen */
        memcpy( &sock->socka_buff, service->ai_addr, min(service->ai_addrlen,sizeof(sock->socka_buff)) );
        LOG( LOG_STARTUP, "Listen: %s\n", service->ai_canonname );
        freeaddrinfo(service);
        SocketLogIP(sock,LOG_INFO,"Address: ");
        if( SOCKET_ERROR == bind( sock->socket, (struct sockaddr*)&sock->ip, sizeof(sock->ip)) ) 
        {
            SocketsLogLastError( LOG_ERROR, "Socket_Create_Listener(%s,%s) bind failed.\n", sock->label,address );
            Socket_Close( sock );
            return NULL;
        }
    }
    /* Listen for connections, and allow a back-log of half the users trying to get on at once */
    if( SOCKET_ERROR == listen( sock->socket, MAX_SOCKETS/2 ) )
    {
        SocketsLogLastError( LOG_ERROR, "Socket_Create_Listener(%s,%s) listen failed.\n", sock->label,address );
        Socket_Close( sock );
        return NULL;
    }
    return sock;
}

/*
 * Default bogus HTTP response string
 *
 * This response just puts up a 'NOT A WEB SERVER' message.
 * 
 * YOUR response should immediately redirect to an appropriate web site 
 */
const char* defcbHTTP(void)
{
    return
        "<html xmlns=\"http://www.w3.org/1999/xhtml\" lang=\"en\">"
        "<body bgcolor=\"#000000\" text=\"#ff0000\" link=\"#ff0000\" vlink=\"#800000\">"
        "<font size=\"7\"><center><br/><br/><br/><br/><b>"
            "<a href=\"http://en.wikipedia.org/wiki/Liar_paradox\">NOT A WEB SERVER</a>"
        "</b></center></font></body></html>"
        "\n";
}

/* 
 * Default FLASH policy file string
 *
 * This response tells Flash to connect wherever it likes.
 * 
 * YOUR response should be a policy file containing a list of explicit URLs to
 * web sites and ports that Flash can connect to and/or receive content from.  
 */
const char* defcbFlashPolicy(void)
{
    return
        "<?xml version=\"1.0\"?>\n"
        "<!DOCTYPE cross-domain-policy SYSTEM \"http://www.macromedia.com/xml/dtds/cross-domain-policy.dtd\">\n"
        "<cross-domain-policy>\n"
        "   <allow-access-from domain=\"*\" to-ports=\"*\" />\n"    /* This is the important bit */
        "</cross-domain-policy>\n";
}

static MLCallback cbHTTP = defcbHTTP;
static MLCallback cbFlashPolicy = defcbFlashPolicy;
MLCallback SocketsSetHTTP( MLCallback callback )
{
    MLCallback prev = cbHTTP;
    cbHTTP = NULL == callback ? defcbHTTP: callback;
    assertcodeptr(cbHTTP);
    return prev;
}
MLCallback SocketsSetFlashPolicy( MLCallback callback )
{
    MLCallback prev = cbFlashPolicy;
    cbFlashPolicy = NULL == callback ? defcbFlashPolicy: callback;
    assertcodeptr(cbFlashPolicy);
    return prev;
}

/*
 * Do read cycle for Login, watching for special cases and doing lots of 
 * checks for bogus inputs from the great unwashed.
 */
static void Login_ReadCycle(struct Socket* sock)
{
    /* Try to read data into buffer */
    size_t available;
    assertobjptr(sock);
    available = RingBuff_Free(&sock->input);
    if( 0 < available )
    {
        uint8* space;
        RingBuff_Space(&sock->input,&space,available);
        int bytes = recv( sock->socket, space, available, 0 );
        Socket_Validate(sock);
        if( 0 == bytes )
        {   /* Disconnected */
            Socket_Close( sock );
            return;
        }
        if( SOCKET_ERROR == bytes )
        {
            if( EWOULDBLOCK != SocketError() )
            {
                SocketsLogLastError( LOG_ERROR, "Login_ReadCycle(%s) receive failed\n", sock->label );
                Socket_Close( sock );
            }
            return;
        }
        //debug_loghex( LOG_INFO, sock->input.tail, bytes );
        RingBuff_Add( &sock->input, &space, (size_t)bytes );
    }
    available = RingBuff_Remain(&sock->input);
    {   
        /* Check for Flash 'security' input */
        const char szGET[] = "GET / HTTP";  /* Substring */
        const char szQuery[] = "<policy-file-request/>"; /* Includes NUL */
        if( sizeof(szQuery) == available )
        {
            const uint8* peek;
            if( RingBuff_Peek( &sock->input, &peek, sizeof(szQuery) ) 
             && 0 == memcmp( peek, szQuery, sizeof(szQuery) ) )
            {
                /* Flash client asked for assurance that I allow connections. */
                /* Send a policy file to let it know it CAN connect. */
                const char* xmlSpew = cbFlashPolicy();
                RingBuff_Reset( &sock->input );
                Socket_Write( sock, xmlSpew, 1+strlen(xmlSpew) );
                /* Client re-opens a NEW connection once this is received, so just blow it away once the data has cleared */
                sock->type = stSendingFinal;
                sock->timeout = 10000;  /* Allow ten seconds, then just drop it */
                Socket_Validate(sock);
                strncpy(sock->label,szconst(char,szQuery),countof(sock->label));
                return;
            }
        }
        if( sizeof(szGET) < available )
        {
            const uint8* peek;
            if( RingBuff_Peek( &sock->input, &peek, sizeof(szQuery) ) 
             && 0 == memcmp( peek, szGET, sizeof(szGET)-1 ) )
            {
                /* Client asked for web page. */
                /* Send an html snippet to tell it no. */
                /* When 'live', should probably link to the service home page */
                /* It may be a good idea to use the www port, as this is almost never blocked anywhere */
                const char htmlSpewHead[] = 
                    "HTTP/1.1 200 OK\n"
                    "Content-Type: text/html\n"
                    "\n";
                const char* htmlSpewBody = cbHTTP();
                RingBuff_Reset( &sock->input );
                Socket_Write( sock, htmlSpewHead, sizeof(htmlSpewHead)-1 );
                Socket_Write( sock, htmlSpewBody, strlen(htmlSpewBody) );
                /* Client re-opens a NEW connection once this is received, so just blow it away once the data has cleared */
                sock->type = stSendingFinal;
                sock->timeout = 10000; /* Allow ten seconds, then drop it */
                Socket_Validate(sock);
                strncpy(sock->label,szconst(char,szGET),countof(sock->label));
                SocketLogIP( sock, LOG_SECURITY, "Bogus HTTP connection: " );
                return;
            }
        }
        if( available >= 2 )
        {   /* Try to pop a 16 bit length-encoded serial packet */
            ctl_serial serial;
            if( RingBuff_PopSerial( &sock->input, &serial ) )
            {
                sock->lastmsg = ctl_mstime();
                sock->appReceived( sock, &serial );
            }
        }
    }
}

/*
 * Login connection has been shut down
 */
static void Login_Close( Socket* sock )
{
}
SockVTAB LoginVTAB =
{
    Login_ReadCycle,
    Generic_WriteCycle,
    Login_Close
};

Socket* Socket_Create_Login( SOCKET isock )
{
    Socket* sock = Socket_Create();
    if( NULL == sock )
        return NULL;
    sock->vtab = &LoginVTAB;
    sock->type = stLogin;
    sock->socket = isock;
    sock->timeout = 5000; /* We expect a messages immediately upon connection */
    {
        /* Grab address from connecting client */
        size_t addrlen = sizeof(sock->socka_buff);
        getpeername( sock->socket, &sock->address, &addrlen );
    }
    cstrncpy(char)( sock->label, "Login", countof(sock->label) );
    /*traceip("Socket_Create_Login:",&sock->address); */
    /* Setup socket */
    Socket_Options( sock );
    /* Call application callback for dynamically created Login socket */
    assertcodeptr(login_callback);
    login_callback(sock);
    return sock;
}

/*
 * User connection has been shut down
 */
static void User_Close( Socket* sock )
{
}
SockVTAB UserVTAB =
{
    Generic_ReadCycle,
    Generic_WriteCycle,
    User_Close
};
Socket* Socket_Create_User( Socket* sock, const char* label )
{
    assertobjptr(sock);
    assertconst(label,1);
    sock->vtab = &UserVTAB;
    sock->type = stUser;
    sock->timeout = (15*60*1000); /* Let user go idle for 15 minutes */
    cstrncpy(char)(sock->label, label, countof(sock->label));
    /*trace(("Socket_Create_User(%s)\n", sock->label )); */
    /* Call application callback for dynamically created User socket */
    assertcodeptr(user_callback);
    user_callback(sock);
    return sock;
}

/*
 * Raw connection read cycle
 */
static void Raw_ReadCycle( Socket* sock )
{
    /* Try to read data into buffer */
    size_t available;
    assertobjptr(sock);
    available = RingBuff_Free(&sock->input);
    if( 0 < available )
    {   /* Try to get non-zero bytes when there's space */
        uint8* space;
        RingBuff_Space(&sock->input, &space, available );
        int bytes = recv( sock->socket, space, available, 0 );
        if( 0 == bytes )
        {   /* Disconnected */
            Socket_Close( sock );
            return;
        }
        if( SOCKET_ERROR == bytes )
        {
            if( EWOULDBLOCK != SocketError() )
            {
                SocketsLogLastError( LOG_ERROR, "Raw_ReadCycle(%s) receive failed\n", sock->label );
                Socket_Close( sock );
            }
            return;
        }
        //debug_loghex( LOG_INFO, sock->input.tail, bytes );
        RingBuff_Add( &sock->input, &space, (size_t)bytes );
    }
    {
        size_t remain = RingBuff_Remain(&sock->input);
        const uint8* peek;
        if( RingBuff_Peek( &sock->input, &peek, remain ) )
        {
            ctl_serial packet;
            sock->lastmsg = ctl_mstime();
            ctl_serial_init( &packet, (uint8*)peek, peek+remain );
            /* 
             * App must call Socket_Raw_Consume() to eat bytes in the queue
             * This is because at this level, 'Raw' doesn't know what is a complete packet 
             * App is expectected to keep looking at the data until it knows what to do
             * with it, then {do it} and remove the packet data from the input queue.
             */
            sock->appReceived( sock, &packet );
        }
    }
}
const uint8* Socket_Raw_Consume( Socket* sock, size_t size )
{
    const uint8* data;
    assertobjptr(sock);
    RingBuff_Pop( &sock->input, &data, size );
    return data;
}

/*
 * Raw connection has been shut down
 */
static void Raw_Close( Socket* sock )
{
}
SockVTAB RawVTAB =
{
    Raw_ReadCycle,
    Generic_WriteCycle,
    Raw_Close
};
Socket* Socket_Create_Raw( Socket* sock )
{
    assertobjptr(sock);
    sock->vtab = &RawVTAB;
    sock->type = stRaw;
    sock->timeout = 15000;
    cstrncpy(char)( sock->label, "Raw", countof(sock->label) );
    /*trace(("Socket_Create_Raw(%s)\n", sock->label )); */
    /* Call application callback for dynamically created Raw socket */
    assertcodeptr(user_callback);
    user_callback(sock);
    return sock;
}

Socket* Socket_Create(void)
{
    Socket* sock;
    size_t index;
    /* Allocate & clear it; all values should be zero */
    ctl_fp_calloc(SocketPool, sockPool, sock );
    if( NULL == sock )
        return NULL;
    sock->type = stUninitialized;
    sock->timeout = 5000;
    ++cSockActive;
    --cSockFree;
    sock->created = sock->lastmsg = ctl_mstime();

    {   
        /*
         * Confusion reduction (or confusion enhancement)
         * Socket->id is 16 bits AND a 16 bit sequence counter.  
         *
         * This will reduce the likelihood that a socket ID freed and reused 
         * will end up being the same ID, and help spot slightly stale identifier 
         * references.
         *
         * Very, very stale identifiers may not be identified because there's still 
         * a very small chance that the counter will wrap around and make an old 
         * identifier look valid again.
         */
        static uint32 seq = 0;
        index = ctl_fp_index(SocketPool, sockPool, sock );
        sock->id = index | (seq++&0xffff)<<16;
    }


    /* Add to active list */
    dll_push_back(Socket,active, socksActive, sock );
        
    /* Set socket to invalid value */
    sock->socket = INVALID_SOCKET;

    /* Give it a default label */
    strncpy(sock->label, "Socket", countof(sock->label) );

    /* Alocate socket storage separately */
    /* Keep the socket structure data small because it's iterated */
    /* Preallocate everything ONCE to keep server from dying at some odd time due to insufficient resources.  Die immediately, please. */
    /* Allocated from pool we initialized at startup */
    Socket_Customize( sock, 0,0,0 );

    /* Setup default application callbacks */
    sock->appReceived = Socket_DefaultReceived;
    sock->appDisconnect = Socket_DefaultCallback;
    return sock;
}

/* Free up dead socket */
static void Socket_Free( Socket* sock )
{
    size_t index;
    Socket_Validate(sock);
    index = ctl_fp_index(SocketPool, sockPool, sock );
    assert( index >= 0 && index < MAX_SOCKETS );
    sock->type = stUninitialized;
    sock->id = 0;

    /* Free any customized buffers */
    if( NULL != sock->pInput && SOCK_INPUT_DEFAULTPTR(index) != sock->pInput )
        ctl_free( sock->pInput );
    if( NULL != sock->pOutput && SOCK_OUTPUT_DEFAULTPTR(index) != sock->pOutput )
        ctl_free( sock->pOutput );
    if( NULL != sock->pAppHeap && SOCK_HEAP_DEFAULTPTR(index) != sock->pAppHeap )
        ctl_free( sock->pAppHeap );

    dll_erase(Socket,active, socksActive, sock );
    ctl_fp_free(SocketPool, sockPool, sock );
    --cSockActive;
    ++cSockFree;
}

Socket* Socket_Customize( Socket* sock, size_t input, size_t output, size_t heap )
{
    size_t index;
    assertobjptr(sock);
    index = ctl_fp_index(SocketPool, sockPool, sock );
    assert( index >= 0 && index < MAX_SOCKETS );
    /* Free previously customized buffers (I don't wanna leak even if you're doing it wrong) */
    if( NULL != sock->pInput && SOCK_INPUT_DEFAULTPTR(index) != sock->pInput )
        ctl_free( sock->pInput );
    if( NULL != sock->pOutput && SOCK_OUTPUT_DEFAULTPTR(index) != sock->pOutput )
        ctl_free( sock->pOutput );
    if( NULL != sock->pAppHeap && SOCK_HEAP_DEFAULTPTR(index) != sock->pAppHeap )
        ctl_free( sock->pAppHeap );
    if( input > sockInSize )
    {
        ctl_malloc( sock->pInput, input + sizeof(uint32));
    }
    else
    {
        input = sockInSize;
        sock->pInput = SOCK_INPUT_DEFAULTPTR(index);
    }
    if( output > sockOutSize )
    {
        ctl_malloc( sock->pOutput, output + sizeof(uint32) );
    }
    else
    {
        output = sockOutSize;
        sock->pOutput= SOCK_OUTPUT_DEFAULTPTR(index);
    }
    if( heap > sockHeapSize )
    {
        ctl_malloc( sock->pAppHeap, heap + sizeof(uint32) );
    }
    else
    {
        heap = sockHeapSize;
        sock->pAppHeap = SOCK_HEAP_DEFAULTPTR(index);
    }
    /* Setup input/output ring buffers */
    memset( sock->pInput, 0, input );
    RingBuff_Init( &sock->input, sock->pInput, sockInSize );
    memset( sock->pInput, 0, output );
    RingBuff_Init( &sock->output, sock->pOutput, sockOutSize );
    /* Setup application heap */
    memset( sock->pAppHeap, 0, heap );
    ctl_heap_init( &sock->appheap, sock->pAppHeap, sockHeapSize );
    /* Initialize guard bytes whether we're checking them or not (memory dump friendly) */
    *((uint32*)(sock->pInput + input)) = GUARD_BYTES;
    *((uint32*)(sock->pOutput + output)) = GUARD_BYTES;
    *((uint32*)(sock->pAppHeap + heap)) = GUARD_BYTES;
    return sock;
}


void Socket_LogError( Socket* sock, unsigned mask, const char*fmt, ... )
{
    va_list args;
    va_start( args, fmt );
    int save_errno = SocketError();
    assertconst(fmt,1);
    debug_vlog(mask, fmt, args );
    SocketLogIP(sock,mask,"");
    SocketSetError(save_errno);
    LOG( mask, "%s\n", SocketsLastErrorString() );
}

#ifdef DEBUG

void Socket_Validate( Socket* sock )
{
    size_t index;
    assertobjptr(sock);
    index = ctl_fp_index(SocketPool, sockPool, sock );
    assert( NULL != sock->pInput );
    assert( NULL != sock->pOutput );
    assert( NULL != sock->pAppHeap );
    if( SOCK_INPUT_DEFAULTPTR(index) == sock->pInput )
    {
        assert( *((uint32*)(sock->pInput + sockInSize)) == GUARD_BYTES );
    }
    else
    {
        assert( *((uint32*)(sock->pInput + ctl_size(sock->pInput) - sizeof(uint32))) == GUARD_BYTES );
    }
    if( SOCK_OUTPUT_DEFAULTPTR(index) == sock->pOutput )
    {
        assert( *((uint32*)(sock->pOutput + sockOutSize)) == GUARD_BYTES );
    }
    else
    {
        assert( *((uint32*)(sock->pOutput + ctl_size(sock->pOutput) - sizeof(uint32))) == GUARD_BYTES );
    }
    if( SOCK_HEAP_DEFAULTPTR(index) == sock->pAppHeap )
    {
        assert( *((uint32*)(sock->pAppHeap + sockHeapSize)) == GUARD_BYTES );
    }
    else
    {
        assert( *((uint32*)(sock->pAppHeap + ctl_size(sock->pAppHeap) - sizeof(uint32))) == GUARD_BYTES );
    }
}
#endif

void Socket_Close( Socket* sock )
{
    assertobjptr(sock);
    trace(( "Socket_Close(%s)\n", sock->label ));
    /* Debug: See if socket got shredded */
    Socket_Validate(sock);
    sock->type = stUninitialized;
    sock->id = 0;

    /* Close socket handle */
    if( INVALID_SOCKET != sock->socket )
    {
        if( SOCKET_ERROR == closesocket(sock->socket) )
            SocketsLogLastError( LOG_ERROR, "Socket_Close(%d,%s) closesocket failed.\n", sock->socket, sock->label );
        sock->socket = INVALID_SOCKET;
    }
    
    /* Tell app to clean up after its self */
    assertcodeptr(sock->appDisconnect);
    sock->appDisconnect(sock);

    /* Tell Socket to clean up after its self */
    assertcodeptr(sock->vtab->Close);
    sock->vtab->Close(sock);

    /* Restore default application callbacks. */
    /* App will receive nothing more about this instance. */
    sock->appReceived = Socket_DefaultReceived;
    sock->appDisconnect = Socket_DefaultCallback;
}

bool Socket_Write( Socket* sock, const void* buff, size_t count )
{
    assertobjptr(sock);
    assertconst(buff,count);
    if( Socket_ShouldIgnore(sock) )
    {
        return false;
    }
    else
    {
        bool bRet = RingBuff_CopyTo( &sock->output, buff, count );
        if( !bRet )
        {
            SocketsLogLastError( LOG_ERROR, "Socket_Write(%s,%u) failed\n", sock->label, count );
            Socket_Close( sock );
            return false;
        }
        return true;
    }
}

bool Socket_Alloc_Serial( Socket* sock, ctl_serial* serial, size_t size )
{
    assertobjptr(sock);
    assertobjptr(serial);
    if( Socket_ShouldIgnore(sock) )
    {
        return false;
    }
    else
    {
        bool bRet = RingBuff_AddSerial( &sock->output, serial, size + 2 );
        if( bRet )
        {
            return true;
        }
        else
        {   /* Not enough room left! */
            SocketsLogLastError( LOG_ERROR, "Socket_Alloc_Serial(%s,%u) failed\n", sock->label, size );
            Socket_Close( sock );
            return false;
        }
    }
}

bool Socket_Write_Serial( Socket* sock, const ctl_serial* serial )
{
    assertobjptr(sock);
    assertobjconst(serial);
    if( Socket_ShouldIgnore(sock) )
    {
        return false;
    }
    else
    {
        size_t length = ctl_serial_total( serial );
        bool bRet = RingBuff_CopyTo( &sock->output, serial->begin, length );
        if( !bRet )
        {   /* Not enough room left! */
            SocketsLogLastError( LOG_ERROR, "Socket_Write_Serial(%s,%u) failed\n", sock->label, length );
            Socket_Close( sock );
            return false;
        }
        return true;
    }
}