netlib.cpp

Go to the documentation of this file.

/*
 * netlib.cpp
 *
 * Copyright (C) 2007-2009  Thomas A. Vaughan
 * All rights reserved.
 *
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *     * Redistributions of source code must retain the above copyright
 *       notice, this list of conditions and the following disclaimer.
 *     * Redistributions in binary form must reproduce the above copyright
 *       notice, this list of conditions and the following disclaimer in the
 *       documentation and/or other materials provided with the distribution.
 *     * Neither the name of the <organization> nor the
 *       names of its contributors may be used to endorse or promote products
 *       derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THOMAS A. VAUGHAN ''AS IS'' AND ANY
 * EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED. IN NO EVENT SHALL THOMAS A. VAUGHAN BE LIABLE FOR ANY
 * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 *
 * Implementation of the networking library.  See netlib.h
 */

// includes --------------------------------------------------------------------
#include "netlib.h"             // always include our own header first!
#include "wavesock.h"

#include <deque>

#ifndef _XOPEN_SOURCE
#define _XOPEN_SOURCE 600
#endif  // _XOPEN_SOURCE

#include <string.h>

#include "common/wave_ex.h"
#include "perf/perf.h"
#include "util/parsing.h"

#ifdef WIN32
typedef long ssize_t;
#endif  // WIN32


namespace netlib {


// use a small size for testing, large size for production
static const int s_chunkSize            = 8192;

static const int s_maxHeaderLine        = 64;

struct request_t {
        // constructor, manipulators
        request_t(void) throw() { this->clear(); }
        void clear(void) throw() {
                        connId = 0;
                        msgbuf = NULL;
                }
        bool is_empty(void) const throw() {
                        return (!connId && !msgbuf);
                }

        // data fields
        conn_id_t                       connId;
        smart_ptr<MessageBuffer>        msgbuf;
};



// TODO: don't require memory alloc/free!  Keep a free list
typedef std::deque<request_t> message_queue_t;



// conn_rec_t : connection record
struct conn_rec_t {
        // constructor, manipulators
        conn_rec_t(void) throw() : socket(-1) { this->clear(); }
        ~conn_rec_t(void) { this->clear(); }
        void clear(void) throw() { 
                        if (wsIsValidSocket(socket)) {
                                DPRINTF("Closing down socket!");
                                DPRINTF("Connection id = 0x%lx", (long) conn_id);
                                wsCloseSocket(socket);
                        }

                        conn_id = 0;
                        local = 0;
                        socket = -1;
                        need_bytes = -1;
                        type = eType_Invalid;
                        msgbuf = NULL;
                        message_queue.clear();
                        send_byte = -1;
                        buffer[0] = 0;
                        buff_idx = -1;  // empty
                        buff_len = 0;
                        header[0] = 0;
                        head_idx = 0;
                        udpFrom.clear();
                        address.clear();
                }
        void dump(IN const char * text) const throw() {
                        DPRINTF("%s", text);
                        DPRINTF("  connId: 0x%04lx", (long) conn_id);
                        DPRINTF("  socket: %d", socket);
                        DPRINTF("  type: %d", type);
                        address.dump(text);
                }

        // data fields
        conn_id_t       conn_id;
        conn_id_t       local;          // local peer connection (UDP)
        int             socket;
        long            need_bytes;     // bytes needed to complete message
        smart_ptr<MessageBuffer> msgbuf;// long-lived message buffer
        eConnectionType type;           // what sort of connection?
        message_queue_t message_queue;  // pending messages to write
        address_t       address;        // where messages go
        address_t       udpFrom;        // for received UDP messages
        long            send_byte;      // current send byte
        char            buffer[s_chunkSize];    // buffer for reading
        int             buff_idx;       // where in buffer are we?
        ssize_t         buff_len;       // how much did we read?
        char            header[s_maxHeaderLine];
        int             head_idx;       // index into header line
};

// connection ID --> connection record
typedef std::map<conn_id_t, smart_ptr<conn_rec_t> > conn_map_t;

static conn_map_t s_connection_map;



// stats!
static dword_t s_messagesSent                   = 0;
static dword_t s_messagesReceived               = 0;
static qword_t s_bytesWritten                   = 0;
static qword_t s_bytesRead                      = 0;


// connection type names


////////////////////////////////////////////////////////////////////////////////
//
//      static helper methods
//
////////////////////////////////////////////////////////////////////////////////

static const char *
getTypeName
(
IN eConnectionType type
)
{
        switch (type) {

        case eType_TCP:
                return "TCP client";

        case eType_UDPLocal:
                return "Local UDP port";

        case eType_UDPRemote:
                return "Remote UDP port";

        case eType_TCPListener:
                return "Local TCP Listener";

        default:
                break;
        }
        return "Unknown connection type!";
}



static conn_id_t
getNewConnectionId
(
void
)
{
        // originally I was just using a dumb counter, to avoid having
        // to look for collisions etc.  But counters can wrap!  So I'm
        // using random numbers
        static const dword_t s_dwMax = 0x10000 - 1;
        for (;;) {
                conn_id_t conn_id = 1 + (rand() % s_dwMax);
                if (s_connection_map.end() == s_connection_map.find(conn_id))
                        return conn_id;
        }
}



static void
dumpErrorInfo
(
IN const char * msg
)
{
        const int s_bufsize = 256;
        char buffer[s_bufsize];

        wsGetErrorMessage(buffer, s_bufsize);

        DPRINTF("%s", msg);
        DPRINTF("%s", buffer);
}



static void
verify
(
IN bool isOK,
IN const char * msg
)
{
        if (isOK)
                return;         // no problem!

        // if we're here, there is a big problem!
        dumpErrorInfo(msg);
        ASSERT(false, "halting");
}



static void
verifyThrow
(
IN bool isOK,
IN const char * msg
)
{
        if (isOK)
                return;         // no problem!

        const int s_bufsize = 256;
        char buffer[s_bufsize];
        wsGetErrorMessage(buffer, s_bufsize);

        ASSERT(msg, "null");
        DPRINTF("Error!  '%s' on '%s'", buffer, msg);

        WAVE_EX(wex);
        wex << msg << "\n";
        wex << "Error: " << buffer;
}



static conn_rec_t *
getConnectionRecord
(
IN conn_id_t conn_id
)
{
        ASSERT(conn_id, "null");

        conn_map_t::iterator i = s_connection_map.find(conn_id);
        if (s_connection_map.end() == i) {
                DPRINTF("Connection ID 0x%lx not found!", (long) conn_id);
                return NULL;
        }

        return i->second;
}



static bool
readBuffer
(
IN conn_rec_t * rec
)
{
        // timer itself impacts timing!
        // perf::Timer timer("netlib::readBuffer");
        ASSERT(rec, "null");

        // reset
        rec->buffer[0] = 0;
        rec->buff_len = 0;
        rec->buff_idx = -1;

        // make nonblocking call to see if we have any data to read
        // NOTE: the type of read call depends on the type of socket!
        ssize_t bytes = -2;
        if (eType_TCP == rec->type) {
                bytes = wsReceive(rec->socket, rec->buffer, s_chunkSize - 1);
        } else if (eType_UDPLocal == rec->type) {
                // DPRINTF("Reading UDP packet...");
                bytes = wsReceiveFrom(rec->socket, rec->buffer, s_chunkSize - 1,
                    rec->udpFrom);
        } else {
                DPRINTF("Bad local connection type?  Disconnecting...");
                bytes = 0;
        }

        // error?
        if (bytes < 0) {
                if (eWS_Again == wsGetError()) {
                        // no problem, try again later
                        return false;
                }
                DPRINTF("Error receiving!  Disconnecting client...");
                bytes = 0;
        }

        // client gave up?
        if (0 == bytes) {
                DPRINTF("Client has disconnected");
                closeConnection(rec->conn_id);
                return false;
        }

        s_bytesRead += bytes;

        // received data!
        if (bytes >= s_chunkSize) {
                DPRINTF("ERROR: client sent too many bytes!");
                DPRINTF("Our buffer size is %d bytes",
                    s_chunkSize - 1);
                DPRINTF("Client sent %ld bytes", (long) bytes);
                DPRINTF("Truncating data!");
                ASSERT(false, "HALT");  // this is bad
                bytes = s_chunkSize - 1;
        }
        rec->buffer[bytes] = 0; // force null-termination
        rec->buff_len = bytes;
        rec->buff_idx = 0;

        return true;
}



static void
parseHeaderLine
(
IN conn_rec_t * rec
)
{
        // at the moment, this routine is fast enough that the timer itself
        //   adds significant time!
        // perf::Timer timer("netlib::parseHeaderLine");
        ASSERT(rec, "null");
        ASSERT(rec->head_idx >= 0 && rec->head_idx < s_maxHeaderLine,
            "Bad header byte index: %d", rec->head_idx);

        // TODO: avoid allocations here!  (use of std::string)
//      rec->dump("Parsing header");

        // end of line!  Interesting?
        rec->header[rec->head_idx] = 0; // null-terminate
        std::string key;
        const char * p = getNextTokenFromString(rec->header, key, eParse_None);
        //DPRINTF("key = '%s'", key.c_str());

        std::string val;
        getNextTokenFromString(p, val, eParse_None);
        //DPRINTF("val = '%s'", val.c_str());

        // now what?
        // At the moment, the header consists of a single line: the byte
        //  count (size), which is identified by a leading "s" character.
        if ("s" == key) {
                rec->need_bytes = atol(val.c_str());
                //DPRINTF("Message bytes: %ld", rec->need_bytes);
                if (rec->need_bytes <= 0) {
                        DPRINTF("Bad byte count? %ld", rec->need_bytes);
                        rec->need_bytes = 0;
                }
        } else {
                //DPRINTF("Unknown message header key!");
        }
}



static bool
handleData
(
IN conn_rec_t * rec,
IO envelope_t& envelope,
IO smart_ptr<MessageBuffer>& msgbuf
)
{
        // timer itself impacts timing!
        // perf::Timer timer("netlib::handleData");
        ASSERT(rec, "null");
        ASSERT(envelope.is_empty(), "not empty");
        ASSERT(!msgbuf, "should be null");

//      rec->dump("Reading data");

        // DPRINTF("Got data to read!");

        // keep reading!
        const char * p = NULL;
        while (true) {

                if (rec->buff_idx >= rec->buff_len) {
                        rec->buff_idx = -1;
                }

                // DPRINTF("Starting loop, idx = %d", rec->buff_idx);
                if (rec->buff_idx < 0) {
                        // need to read buffer!
                        if (!readBuffer(rec))
                                return false;   // couldn't read
                        if (rec->buff_idx < 0) {
                                return false;   // failed to read anyway
                        }
                        //DPRINTF("  Read %d bytes", rec->buff_len);
                }
                p = rec->buffer + rec->buff_idx;
                //DPRINTF("  After message read, idx=%d", rec->buff_idx);
                //DPRINTF("  %d bytes remaining in buffer",
                //    rec->buff_len - rec->buff_idx);

                // parse message headers if necessary
                const char * maxP = rec->buffer + rec->buff_len;
                for (; p < maxP && rec->need_bytes < 0; ++p) {
                        
                        // are we getting a lot of garbage from client?
                        if (rec->head_idx >= s_maxHeaderLine - 1) {
                                // too big!  Reset
                                DPRINTF("Garbage from remote host!  Resetting");
                                rec->head_idx = 0;
                        }

                        // push new byte to end of our header buffer
                        rec->header[rec->head_idx] = *p;
                        rec->head_idx++;

                        if ('\n' == *p) {
                                parseHeaderLine(rec);
                                rec->head_idx = 0;
                        } else if (!*p) {
                                // null in header line?  Weird!
                                rec->head_idx = 0;
                        }
                }

                // should be message data
                int remain = rec->buff_len - (p - rec->buffer);
                // DPRINTF("%d bytes remain", remain);
                rec->buff_idx = p - rec->buffer;

                if (!remain) {
//                      DPRINTF("end of buffer");       // very common!
                        continue;       // end of buffer
                }

                // no point in proceeding if we aren't ready
                if (rec->need_bytes < 0) {
                        DPRINTF("aren't ready");
                        ASSERT(!*p, "should be out of data");
//                      rec->need_bytes = 0;
                        return false;   // ran out of data from read
                }

                // what we expected?
                if (rec->need_bytes < remain) {
//                      DPRINTF("ERROR: received more bytes than expected!");
//                      DPRINTF("  expected: %ld", rec->need_bytes);
//                      DPRINTF("  received: %d", remain);
//                      DPRINTF("  truncating!!!");  - NO!  Not truncating...
                        // not a problem: we just take what we need
                        remain = rec->need_bytes;
                }

                // need to create buffer?
                if (!rec->msgbuf) {
                        rec->msgbuf = MessageBuffer::create();
                        ASSERT(rec->msgbuf, "failed to create message buffer?");
                        rec->msgbuf->reserve(rec->need_bytes + 1);
                }

                // append
                rec->msgbuf->appendData(p, remain);

                // decrement
                //DPRINTF("  Copied %d bytes...", remain);
                rec->need_bytes -= remain;
                rec->buff_idx = p + remain - rec->buffer;
                if (rec->need_bytes < 1) {
                        //DPRINTF("Message is now complete!");
                        rec->msgbuf->close();
                        //DPRINTF("Message size: %ld bytes",
                        //    rec->msgbuf->getBytes());

                        // hand buffer over to message
                        msgbuf = rec->msgbuf;

                        // construct envelope information
                        envelope.fromConnId = rec->conn_id;
                        envelope.type = rec->type;

                        // Need to swap out for UDP!
                        // We read from local UDP port (of course), but client
                        // needs to know which remote UDP client sent this.
                        if (eType_UDPLocal == envelope.type) {
                                envelope.type = eType_UDPRemote;
                                envelope.fromConnId = 0;
                                envelope.address = rec->udpFrom;
                        }

                        // give up ownership and clean up
                        rec->msgbuf = 0;
                        rec->need_bytes = -1;
                        s_messagesReceived++;

                        return true;
                }
        }

        // nope
        return false;
}



static conn_id_t
addConnectionRecord
(
IN eConnectionType type,
IN int socket,
IN const address_t& address
)
{
        ASSERT(eType_Invalid != type, "Bad type");
        ASSERT(socket > -2, "bad socket");

//      DPRINTF("Creating connection record for '%s':%d ...", host, port);

        if (-1 == socket) {
                ASSERT(eType_UDPRemote == type,
                    "Bad connection type (%d) for socket %d", type, socket);
        }

        // construct connection record and put in threadsafe map...
        smart_ptr<conn_rec_t> rec = new conn_rec_t;
        ASSERT(rec, "out of memory?");
        rec->socket = socket;
        rec->address = address;
        rec->type = type;
        rec->conn_id = getNewConnectionId();
//      DPRINTF("  Assigning connection id = 0x%04lx", rec->conn_id);
//      rec->dump("Just created");

        // add to map
        s_connection_map[rec->conn_id] = rec;
        ASSERT(2 == rec.get_ref_count(), "should have 2 refs!");

        //DPRINTF("Currently have %d connections", s_connection_map.size());

        return rec->conn_id;
}



static conn_id_t
handleConnection
(
IN conn_rec_t * rec
)
{
        perf::Timer timer("netlib::handleConnection");
        ASSERT(rec, "null");
        ASSERT(eType_TCPListener == rec->type,
            "Requesting to listen on a non-listening socket?");
        ASSERT(wsIsValidSocket(rec->socket), "bad socket? %d", rec->socket);

        address_t address;
        int c = wsAccept(rec->socket, address);
        if (!wsIsValidSocket(c)) {
                // nobody wanted to connect!
                DPRINTF("Not a valid connection?");
                return 0;
        }
//      address.dump("New connection");

        return addConnectionRecord(eType_TCP, c, address);
}



static void
writeMessage
(
IN conn_rec_t * rec
)
{
        perf::Timer timer("netlib::writeMessage");
        ASSERT(rec, "null");

        ASSERT(rec->message_queue.size(), "empty queue?");

        //DPRINTF("Have message to write!");

        // get the first message in the queue
        const request_t& req = rec->message_queue.front();
        ASSERT(!req.is_empty(), "empty message in queue?");
        ASSERT(req.connId, "null");

        // for UDP, need additional stuff...
        conn_rec_t * recTo = NULL;
        if (eType_UDPLocal == rec->type) {
                // we must be sending to a remote UDP port
                recTo = getConnectionRecord(req.connId);
                ASSERT(recTo, "null entry for UDP receiver");
                ASSERT(eType_UDPRemote == recTo->type,
                    "receiver is wrong type of connection");
        } else if (eType_UDPBroadcast == rec->type) {
                // broadcast: we know where to send
                recTo = rec;
        }

        // try to write header?
        if (rec->send_byte < 0) {
                // yes, header has not yet been sent

                // construct header and send
                const int s_headerBytes = 32;
                char header[s_headerBytes];
                sprintf(header, "\ns %ld\n", req.msgbuf->getBytes());
                //DPRINTF("Header:\n-----%s-----", header);
                int hbytes = strlen(header);
                //DPRINTF("About to send header...");
                //DPRINTF(" hbytes = %d", hbytes);
                //DPRINTF(" s = %d", rec->socket);
                long bytes;
                if (!recTo) {
                        bytes = wsSend(rec->socket, header, hbytes);
                } else {
                        // recTo->address.dump("Sending here");
                        bytes = wsSendTo(rec->socket, header, hbytes,
                            recTo->address);
                }
                if (bytes < 0) {
                        // error!  bad or not?
                        if (eWS_Again == wsGetError()) {
                                //DPRINTF("header send failed, will try again");
                                return;         // quietly fail, try again later
                        }
                        DPRINTF("Error writing to client!  will close connection");
                        closeConnection(rec->conn_id);
                        return;
                }
                verifyThrow(bytes == hbytes, "failed to send complete header!");
                s_bytesWritten += bytes;

                // header successfully sent!
                //DPRINTF("Successfully sent message header");
                rec->send_byte = 0;     // no data sent yet
        }

        // keep sending...
        while (1) {
                long to_send = req.msgbuf->getBytes() - rec->send_byte;
                if (to_send < 1) {
                        //DPRINTF("Completed message send!");

                        // all done
                        s_messagesSent++;
                        break;
                }

                // don't send more than the reader can handle
                if (to_send >= s_chunkSize)
                        to_send = s_chunkSize - 1;

                const char * data = req.msgbuf->getData() + rec->send_byte;
                long sent;
                if (!recTo) {
                        sent = wsSend(rec->socket, data, to_send);
                } else {
                        sent = wsSendTo(rec->socket, data, to_send,
                            recTo->address);
                }
                if (sent < 0) {
                        // error!  bad or not?
                        if (eWS_Again == wsGetError()) {
                                DPRINTF("socket send() failed, will try again");
                                return;
                        }
                        DPRINTF("Error writing to client!  will close connection");
                        closeConnection(rec->conn_id);
                        return;
                }
                s_bytesWritten += sent;
                // DPRINTF("  sent %ld bytes", sent);

                // update stats on bytes sent
                rec->send_byte += sent;
        }

        // update socket data (can do this in place)
        rec->message_queue.pop_front();         // done with message!
        rec->send_byte = -1;                    // reset sent count
}



static void
handleWrites
(
IN ws_set_t writeable
)
{
        // timer itself impacts timing!
        // perf::Timer timer("netlib::handleWrites");
        ASSERT(writeable, "null");

        for (conn_map_t::iterator i = s_connection_map.begin();
             i != s_connection_map.end(); ++i) {
                conn_rec_t * rec = i->second;
                ASSERT(rec, "null connection record");
                int s = rec->socket;
                if (!wsIsValidSocket(s))
                        continue;       // not a real socket!

                if (!wsIsSocketInSet(writeable, s))
                        continue;       // socket isn't writeable

                // DPRINTF("Have a message to write!");
                if (!rec->message_queue.size())
                        continue;       // nothing to write anyway!

                writeMessage(rec);
        }
}



static bool
handleRead
(
IN ws_set_t readers,
IO envelope_t& envelope,
IO smart_ptr<MessageBuffer>& msgbuf
)
{
        // timer itself affects timing!
        // perf::Timer timer("netlib::handleRead");
        ASSERT(readers, "null");
        ASSERT(envelope.is_empty(), "not empty");
        ASSERT(!msgbuf, "should be null");

        for (conn_map_t::iterator i = s_connection_map.begin();
             i != s_connection_map.end(); ++i) {
                conn_rec_t * rec = i->second;
                ASSERT(rec, "null connection record");
                int s = rec->socket;
                if (!wsIsValidSocket(s))
                        continue;               // not a real socket

                if (rec->buff_idx >= 0) {
                        // rec->dump("More data in buffer");
                        if (handleData(rec, envelope, msgbuf))
                                return true;
                }

                if (!wsIsSocketInSet(readers, s))
                        continue;       // socket not impacted

                if (eType_TCPListener == rec->type) {
                        // DPRINTF("Got a request to connect!");
                        handleConnection(rec);
                        return false;
                } else {
                        // DPRINTF("Received data!");
                        // rec->dump("New incoming data");
                        if (handleData(rec, envelope, msgbuf))
                                return true;
                }
        }

        // can get here if no sockets were available for reads!
        return false;
}



////////////////////////////////////////////////////////////////////////////////
//
//      public API
//
////////////////////////////////////////////////////////////////////////////////

std::string
getServerFromIP
(
IN const ip_addr_t& ip
)
{
        // TODO: DNS reverse lookup!
        // for now, just string-ify the IP address
        ASSERT(1 == ip.flags, "Only works with IPv4 addresses for now...");

        char buffer[64];
        sprintf(buffer, "%d.%d.%d.%d",
            ip.addr[0], ip.addr[1], ip.addr[2], ip.addr[3]);

        return buffer;
}



conn_id_t
createTcpListener
(
IN const address_t& address,
IN int maxBacklog
)
{
        perf::Timer timer("netlib::createTcpListener");
        ASSERT(address.isValid(), "Invalid address");
        ASSERT(maxBacklog > 0, "bad max backlog: %d", maxBacklog);

        // set up listening socket
        int s = wsCreateTcpSocket();
        // DPRINTF("TCP listening socket: %d", s);
        verify(wsIsValidSocket(s), "Failed to create tcp listening socket");

        // bind (name) the socket
        verify(!wsBindToPort(s, address.port),
            "Failed to bind tcp listening socket");

        // set up for listening
        verify(!wsListen(s, maxBacklog),
            "Failed to set up tcp socket for listening");

        // add to our map
        return addConnectionRecord(eType_TCPListener, s, address);
}



conn_id_t
createTcpConnection
(
IN const address_t& address
)
{
        perf::Timer timer("netlib::createTcpConnection");
        ASSERT2(address.isValid(),
            "invalid address--cannot create TCP connection");

        // create socket
        int c = wsCreateTcpSocket();

        // DPRINTF("TCP connection socket: %d", c);
        verifyThrow(wsIsValidSocket(c),
            "Failed to create socket for tcp connection");

        // set up address data
        verifyThrow(-1 != wsConnect(c, address),
            "Failed to connect to server");

//      DPRINTF("Connected!");
        return addConnectionRecord(eType_TCP, c, address);
}



conn_id_t
createUdpLocal
(
IN const address_t& address
)
{
        perf::Timer timer("netlib::createUdpLocal");
        ASSERT(address.isValid(), "invalid address");

        // create socket
        int c = wsCreateUdpSocket(false);       // not broadcast
        DPRINTF("Local UDP socket: %d", c);
        verifyThrow(wsIsValidSocket(c), "Failed to create socket for local udp");

        // bind to local port
        verifyThrow(-1 != wsBindToPort(c, address.port),
            "Failed to bind to local UDP socket");

        // all done!
        return addConnectionRecord(eType_UDPLocal, c, address);
}



conn_id_t
createUdpRemote
(
IN conn_id_t localUdp,
IN const address_t& address
)
{
        perf::Timer timer("netlib::createUdpRemote");
        ASSERT(localUdp, "null");
        ASSERT2(address.isValid(),
            "Address is invalid--cannot create remote UDP connection");

        // create connection record
        conn_id_t conn_id =
            addConnectionRecord(eType_UDPRemote, -1, address);
        ASSERT(conn_id, "null");

        // retrieve it because we're going to tweak it...
        conn_rec_t * rec = getConnectionRecord(conn_id);
        ASSERT(rec, "null");
        rec->local = localUdp;
        DPRINTF("Added remote udp connection");

        // all done!
        return conn_id;
}



conn_id_t
createUdpBroadcast
(
IN const address_t& broadcastAddress
)
{
        ASSERT2(broadcastAddress.isValid(), "Invalid broadcast address");

        // create socket
        int s = wsCreateUdpSocket(true);        // yes, broadcast
        verifyThrow(wsIsValidSocket(s),
            "Failed to create socket for udp broadcast");

        // create connection record
        return addConnectionRecord(eType_UDPBroadcast, s, broadcastAddress);
}



bool
enqueueMessage
(
IN conn_id_t conn_id,
IN smart_ptr<MessageBuffer>& message
)
{
        // at the moment, this routine is fast enough that the timer itself
        //      adds significant time!
        // perf::Timer timer("netlib::enqueueMessage");
        ASSERT(conn_id, "null");
        ASSERT(message, "null");

        // look for this connection
        conn_rec_t * rec = getConnectionRecord(conn_id);
        if (!rec) {
                DPRINTF("Connection id not recognized? 0x%04lx", (long) conn_id);
                return false;
        }
        ASSERT(rec, "null record in map for connection id 0x%04lx", (long) conn_id);
        // DPRINTF("Enqueuing message for connection 0x%04lx", conn_id);
        // DPRINTF("  Local protocol is %d", rec->type);

        // improper connection type for sending?
        if (eType_UDPLocal == rec->type) {
                DPRINTF("Cannot send messages to local UDP!  Send to remote");
                return false;
        }

        // udp?  In that case, need to swap local + remote
        if (eType_UDPRemote == rec->type) {
                ASSERT(rec->local, "null local UDP?");
                conn_rec_t * localRec = getConnectionRecord(rec->local);
                ASSERT(localRec, "local udp sender disappeared");
                rec = localRec;         // swap out and queue here
        }
        ASSERT(wsIsValidSocket(rec->socket),
            "null socket in connection record for 0x%04lx", (long) conn_id);

        // add to message queue
        request_t req;
        req.connId = conn_id;
        req.msgbuf = message;
        rec->message_queue.push_back(req);

        // all done
        return true;
}




bool
getNextMessage
(
IN long wait_microseconds,
OUT envelope_t& envelope,
OUT smart_ptr<MessageBuffer>& msgbuf
)
{
        perf::Timer timer("netlib::getNextMessage");
        ASSERT(wait_microseconds >= 0, "Bad wait: %ld", wait_microseconds);
        envelope.clear();
        ASSERT(!msgbuf, "unecessary free?");

        // arbitrary assertion here
        ASSERT(wait_microseconds < 1000 * 1000, "Wait is too long! %ld usec",
            wait_microseconds);

        // mask of types that can read
        dword_t readMask = eType_TCP | eType_UDPLocal | eType_TCPListener;

        // see if any sockets have data for us
        static ws_set_t readers = 0;
        if (!readers) {
                readers = wsCreateSet();
        }
        wsClearSet(readers);

        // also see if any sockets are ready to be sent out on
        static ws_set_t writeable = 0;
        if (!writeable) {
                writeable = wsCreateSet();
        }
        wsClearSet(writeable);

        // add all sockets we care about
        int max = 0;
        for (conn_map_t::iterator i = s_connection_map.begin();
             i != s_connection_map.end(); ++i) {
                const conn_rec_t * rec = i->second;
                ASSERT(eType_Invalid != rec->type, "invalid connection type?");

                // only read from certain sockets
                if (readMask & rec->type) {
                        wsAddSocketToSet(readers, rec->socket);
                }

                // interested in writing if message is pending
                if (rec->message_queue.size() > 0) {
                        wsAddSocketToSet(writeable, rec->socket);
                }

                // update max socket ID?
                if (rec->socket > max)
                        max = rec->socket;
        }

        // go get 'em
        int count = wsSelect(max + 1, readers, writeable, wait_microseconds);
        if (count < 0) {
                dumpErrorInfo("select() call failed");
                return false;
        }

        // first see if anything can be written
        handleWrites(writeable);

        // okay, see if there is anything to read!
        return handleRead(readers, envelope, msgbuf);
}



bool
isValidConnection
(
IN conn_id_t conn_id
)
{
        return (NULL != getConnectionRecord(conn_id));
}



bool
getConnectionInfo
(
IN conn_id_t conn_id,
OUT connection_info_t& info
)
{
        ASSERT(conn_id, "null");
        info.clear();

        conn_rec_t * rec = getConnectionRecord(conn_id);
        if (!rec) {
                return false;
        }

        info.type = rec->type;
        info.address = rec->address;

        // all done
        return true;
}



void
closeConnection
(
IN conn_id_t conn_id
)
{
        ASSERT(conn_id, "null");

        DPRINTF("Closing connection 0x%lx...", (long) conn_id);

        conn_map_t::iterator i = s_connection_map.find(conn_id);
        if (s_connection_map.end() == i) {
                DPRINTF("Error in closeConnection() -- connection id 0x%lx not found",
                    (long) conn_id);
                return;
        }

        // remove the connection
        s_connection_map.erase(i);
}



void
dumpMessage
(
IO std::ostream& stream,
IN const char * title,
IN const envelope_t& envelope,
IN const MessageBuffer * buffer
)
{
        ASSERT(stream.good(), "bad?");
        ASSERT(title, "null");

        DPRINTF("Message dump: %s", title);
        if (envelope.is_empty()) {
                DPRINTF("  Envelope is empty!");
        } else {
                DPRINTF("  From: 0x%04lx", (long) envelope.fromConnId);
                DPRINTF("  Connection Type: %d (%s)", envelope.type,
                    getTypeName(envelope.type));
                envelope.address.dump(title);
        }

        if (!buffer) {
                DPRINTF("  Null message buffer!");
        } else {
                DPRINTF("  Message: '%s'", buffer->getData());
        }
}



void
dumpStats
(
void
)
{
        DPRINTF("Networking stats:");
        DPRINTF("  Total messages received: %6u", s_messagesReceived);
        DPRINTF("  Total messages sent:     %6u", s_messagesSent);
//#ifdef WIN32
//      // sigh... Windows has a non-standard format for long long
//      DPRINTF("  Total bytes read:    %10I64d  (%4I64d MB)", s_bytesRead,
//          (s_bytesRead + 512 * 1024) / (1024 * 1024));
//      DPRINTF("  Total bytes written: %10I64d  (%4I64d MB)", s_bytesWritten,
//          (s_bytesWritten + 512 * 1024) / (1024 * 1024));
//#else // WIN32
        std::cerr << "  Total bytes received: " << s_bytesRead;
        std::cerr << "  (" << (s_bytesRead + 512 * 1024) / (1024 * 1024) << " MB)\n";
        std::cerr << "  Total bytes sent: " << s_bytesWritten;
        std::cerr << "  (" << (s_bytesWritten + 512 * 1024) / (1024 * 1024) << " MB)\n";
//#endif        // WIN32

        if (s_messagesReceived > 0) {
                long avg = (long) (s_bytesRead / s_messagesReceived);
                DPRINTF("  Average size of message read: %5ld bytes", avg);
        }
        if (s_messagesSent > 0) {
                long avg = (long) (s_bytesWritten / s_messagesSent);
                DPRINTF("  Average size of message sent: %5ld bytes", avg);
        }
}


void
connection_info_t::dump
(
IN const char * title
)
const
throw()
{
        DPRINTF("Connection info: %s", title);
        DPRINTF("  type: %d", type);
        address.dump(title);
}


};      // netlib namespace