aesop-srv.cpp

Go to the documentation of this file.

/*
 * aesop-srv.cpp
 *
 * Copyright (C) 2007-2010  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 basic aesop server.  See aesop-srv.h
 *
 * WARNING: try not to put anything interesting in here.  The Server
 * object is mostly just a simple coordinator.  Any rich state etc. should
 * be handled by other objects.  Some objects may also need to run on the
 * client, for example.
 */

// includes -------------------------------------------------------------------
#include "aesop-srv.h"          // always include our own header first!

#include <math.h>

//#include "srv-connection-mgr.h"       // connection management
#include "srv-converse.h"       // conversations we manage
#include "srv-hosts.h"          // host management
#include "srv-msg-router.h"     // message router
#include "srv-physics.h"        // physics messages
#include "srv-players.h"        // player management
#include "srv-users.h"          // user management

#include "common/wave_ex.h"
#include "conversation/conversation.h"  // conversation manager
#include "aesop-proto/message.h"
#include "aesop-proto/protocol.h"
#include "crypto/crypto.h"
#include "datahash/datahash_util.h"
#include "map-dynamics/map-dynamics.h"
#include "map-manager/map-manager.h"
#include "netrq/netrq.h"                // network request queue
#include "perf/perf.h"
#include "story/story.h"
#include "threadpool/threadpool.h"
#include "threadsafe/threadsafe_counter.h"
#include "threadsafe/threadsafe_map.h"
#include "threadsafe/threadsafe_queue.h"
#include "xdrbuf/xdrbuf.h"


namespace aesop {


// public interface destructor implementation
Server::~Server(void) throw() { }


typedef threadsafe_map<int, std::string> name_map_t;


// interface between handlers and thread pool
class handler_intercept_t {
public:
        // constructor, manipulators
        handler_intercept_t(IN const threadsafe_refcount& in_trc) throw() :
                trc(in_trc) {
                        fn = NULL;
                        state.clear();
                        map = NULL;
                }

        // public data fields
        msg_handler_fn          fn;
        handler_state_t         state;
        name_map_t *            map;

private:
        // private data fields
        threadsafe_refcount     trc;    // server keeps track of worker counts
};



typedef std::vector<conn_id_t> vec_conn_t;


// should be long enough to hold most objects and requests, but not insane...
static const int s_maxNetQueueLength            = 256;

// max size of a broadcast message (keep small!)
static const int s_bcastBuffer                  = 128;


//
//      static helper methods
//

static void
interceptHandler
(
IN void * context
)
{
        perf::Timer timer("aesop_srv::interceptHandler");
        ASSERT(context, "null");
        handler_intercept_t * ctx = (handler_intercept_t *) context;
        ASSERT(ctx, "null");

        // wrap context in a smart pointer so it cleans up on exit
        smart_ptr<handler_intercept_t> smart_hi = ctx;

        ASSERT(ctx->fn, "null handler fn?");
        ASSERT(ctx->state.story, "null story?");
        ASSERT(ctx->state.updates, "null updates interface");

        ctx->fn(ctx->state);
}



//
//      Objects to construct network messages (used for network request queue).
//

class QueryContext : public netrq::Request {
public:
        QueryContext(IN dword_t objectId) throw() {
                        sprintf(m_id, "qr%ld", (long) objectId);
                        m_objectId = objectId;
                }
        ~QueryContext(void) throw() { }

        void write(IO xdrbuf::Output * output) {
                        // get this physics object
                        smart_ptr<PhysicsObject> obj =
                            getPhysicsObjectById(m_objectId);
                        if (!obj) {
                                DPRINTF("Client asked for object not there");

                                // object is destroyed!
                        //      output->addInt32Packlet('d', &m_objectId, 1);
                        } else {
                                DPRINTF("Telling client about object %ld",
                                    (long) m_objectId);
                                smart_ptr<Instance> instance =
                                    getInstanceFromPhysicsObject(obj);
                                if (!instance)
                                        return;

                                const char * typeId = instance->getTypeId();
                                int tiLength = strlen(typeId);

                                point3d_t pos = obj->getPosition();
                                //pos.dump("  position");

                                // look up map ID
                                MapDynamics * dyn =
                                    getMapDynamicsFromPhysicsObject(obj);
                                if (!dyn) {
                                        return;
                                }

                                Map * map = dyn->getMap();
                                ASSERT(map, "null");

                                const char * mapId = map->getId();
                                int miLength = strlen(mapId);

                                // for now, we just provide type ID and mapID
                                output->openPacklet('q');       // query response
                                output->addInt32Packlet('i',
                                    (int32_t *) &m_objectId, 1);
                                output->addStringPacklet(
                                    't', typeId, tiLength);
                                output->addStringPacklet(
                                    'm', mapId, miLength);
                                output->closePacklet('q');
                        }
                }

        int getMaxBytes(void) const throw() {
                        return 255 + 255 + 4 + 4 + 4 * 2;
                }

        const char * getId(void) const throw() { return m_id; }

private:
        char            m_id[32];
        dword_t         m_objectId;
};



class AuthContext : public netrq::Request {
public:
        AuthContext(IN dword_t token,
                    IN const char * key) {
                        ASSERT(token, "null");
                        ASSERT(key, "null");
                        sprintf(m_id, "auth%ld", (long) token);
                        m_token = token;
                        m_encryptedDesKey = key;
                }
        ~AuthContext(void) throw() { }

        void write(IN xdrbuf::Output * output) {
                        ASSERT(output, "null");

                        output->openPacklet('c');
                        output->addInt32Packlet('t',
                            (int32_t *) &m_token, 1);
                        output->openPacklet('k');

                        const char * key = m_encryptedDesKey.c_str();
                        int len = m_encryptedDesKey.length();
                        while (len) {
                                int send = len;
                                if (send > 255) {
                                        send = 255;
                                }
                                output->addStringPacklet('a', key, send);
                                key += send;
                                len -= send;
                        }
                        output->closePacklet('k');
                        output->closePacklet('c');
                }

        int getMaxBytes(void) const throw() {
                        return 4 + 512 + 5 * 2;
                }

        const char * getId(void) const throw() { return m_id; }

private:
        char            m_id[32];
        dword_t         m_token;
        std::string     m_encryptedDesKey;
};



class ObjectExistsResponse : public netrq::Request {
public:
        ~ObjectExistsResponse(void) throw() { }

        // netrq::Request class interface methods ------------------------------
        const char * getId(void) const throw() { return m_requestId; }
        int getMaxBytes(void) const throw() { return 6; }
        void write(IO xdrbuf::Output * output) {
                        ASSERT(output, "null");

                        // well, does our object exist?
                        smart_ptr<PhysicsObject> obj =
                            getPhysicsObjectById(m_objectId);
                        if (obj && obj->getId()) {
                                // yes, object exists and is healthy!
                                return;         // nothing to do
                        }

                        // object is destroyed!
                        output->addInt32Packlet('d',
                            (int32_t *) &m_objectId, 1);
                }

        // static helper methods -----------------------------------------------
        static smart_ptr<netrq::Request> create(IN const char * requestId,
                                IN long objectId) {
                        ASSERT(requestId, "null");
                        ASSERT(objectId, "null");

                        smart_ptr<ObjectExistsResponse> local =
                            new ObjectExistsResponse;
                        ASSERT(local, "out of memory");

                        strcpy(local->m_requestId, requestId);
                        local->m_objectId = objectId;

                        return local;
                }
private:
        // private helper methods ----------------------------------------------
        ObjectExistsResponse(void) throw() { }

        // private member data -------------------------------------------------
        char            m_requestId[64];
        dword_t         m_objectId;
};



//
//      ServerImpl
//

class ServerImpl : public StateUpdates,
                        public Server,
                        public converse::ConversationRouter {
public:
        ServerImpl(void) throw();
        ~ServerImpl(void) throw() { }

        // public member functions ---------------------------------------------
        void initialize(IN smart_ptr<Datahash>& params,
                                IN smart_ptr<ServerGameLogic>& gameLogic);

        // aesop::Server class interface methods ------------------------------
        int exec(void);
        void requestStopTS(IN int retval);

        // aesop::StateUpdates class interface methods ------------------------
        //      REMEMBER: these must be threadsafe!
        void enqueueMessageTS(IN conn_id_t connId,
                                IN smart_ptr<MessageBuffer>& message);
        void newTcpConnectionTS(IN conn_id_t connId,
                                IN long token);
        void notifyDialogReplyTS(IN conn_id_t conn_id,
                                IN const char * conversationGuid,
                                IN int dialogId,
                                IN int playerId,
                                IN const Datahash * reply);
        bool setPlayerUserAccountTS(IN conn_id_t conn_id,
                                IN int playerId,
                                IN const char * username,
                                IN const char * usertag,
                                OUT std::string& diagnostic);
        void newGameTS(IN conn_id_t connId,
                                IN int playerId);
        void requestShutdownTS(void);

        // converse::ConversationRouter class interface methods ----------------
        void routeConversationTS(IN const char * guid,
                                IN int dialogId,
                                IN int playerId,
                                IN conn_id_t connId,
                                IN const char * dialogData);

private:
        // private typedefs ----------------------------------------------------
        struct request_t {
                // constructor, manipulators
                request_t(void) throw() { this->clear(); }
                void clear(void) throw() {
                                conn_id = 0;
                                message = NULL;
                        }
                bool is_empty(void) const throw() {
                                return (!conn_id && !message);
                        }

                // data fields
                conn_id_t                       conn_id;
                smart_ptr<MessageBuffer>        message;
        };
        typedef threadsafe_queue<request_t> msg_queue_t;

        // private helper methods ----------------------------------------------
        void checkMessages(IN long microseconds);
        void handleMessage(IN const envelope_t& env,
                                IN const MessageBuffer * msg);
        void handleUdp(IN const envelope_t& env,
                                IN const MessageBuffer * msg);
        void parseHostConnectUdp(IN const netlib::envelope_t& env,
                                IO xdrbuf::Input * input);
        void parseHostAuthUdp(IN host_rec_t& hr,
                                IN const xdrbuf::PackletHeader& ph,
                                IO xdrbuf::Input * input);
        void parsePlayerUdp(IN host_rec_t& hr,
                                IO xdrbuf::Input * input);
        void parseQueryUdp(IN host_rec_t& hr,
                                IN const xdrbuf::PackletHeader& ph,
                                IO xdrbuf::Input * input);
        void parseGameUdp(IN host_rec_t& hr,
                                IN const xdrbuf::PackletHeader& ph,
                                IO xdrbuf::Input * input);
        void parseExistsUdp(IN host_rec_t& hr,
                                IO xdrbuf::Input * input);
        void sendUdp(void);
        const char * getStoryGuidTS(void);
        void newPlayer(IN host_rec_t& hr, IN int playerId);
        bool isAdmin(IN conn_id_t udpConnID, IN int playerId);
        void checkPlayers(void);
        void sendBroadcast(void);

        // private member data -------------------------------------------------
        smart_ptr<story::Story>         m_story;
        smart_ptr<MessageRouter>        m_router;
        smart_ptr<threadpool::Pool>     m_threadpool;
        smart_ptr<Datahash>             m_params;
        smart_ptr<converse::ConversationManager> m_conversationMgr;
        smart_ptr<UserManager>          m_userMgr;
        smart_ptr<PlayerManager>        m_playerMgr;
        smart_ptr<HostManager>          m_hostMgr;
        smart_ptr<MapManager>           m_mapMgr;
        smart_ptr<ServerGameLogic>      m_gameLogic;
        smart_ptr<xdrbuf::Input>        m_xdrInput;     // input parsing
        smart_ptr<xdrbuf::Output>       m_xdrBcast;     // broadcast output
        msg_queue_t                     m_outboundMessages;
        float                           m_requestedHz;  // main loop cycles
        float                           m_udpHz;        // UDP frequency
        bool                            m_shutdown;
        int                             m_retval;
        conn_id_t                       m_tcpConn;      // TCP listening socket
        int                             m_udpPort;      // clients send UDP here
        int                             m_tcpPort;      // clients send TCP here
        conn_id_t                       m_udpConn;      // listening connection
        conn_id_t                       m_udpBcast;     // broadcast UDP point
        threadsafe_refcount             m_workerTrc;    // refcount of workers
        name_map_t                      m_threadNames;
        int                             m_threadCounter;
        eServerMode                     m_smode;
        dword_t                         m_udpClock;     // clock for UDP messages
        int                             m_bcastDelay;   // seconds between bcasts
        std::string                     m_publicName;   // server display name
};



ServerImpl::ServerImpl(void)
throw()
{
        m_shutdown = false;
        m_retval = 0;
        m_threadCounter = 0;
        m_requestedHz = 0;
        m_udpHz = 0;
        m_udpPort = 0;
        m_tcpPort = 0;
        m_udpConn = 0;
        m_tcpConn = 0;
        m_udpBcast = 0;
        m_smode = eServerMode_Invalid;
        m_udpClock = 0;
        m_bcastDelay = 4;       // make this config driven?
}



void
ServerImpl::initialize
(
IN smart_ptr<Datahash>& params,
IN smart_ptr<ServerGameLogic>& gameLogic
)
{
        ASSERT(params, "null");
        ASSERT(gameLogic, "null");

        DPRINTF("In Server::initialize()...");

        // save parameters
        m_params = params;

        // game logic object
        m_gameLogic = gameLogic;

        // create xdr input parser for UDP packets
        ASSERT(!m_xdrInput, "Alread have an input parser?");
        m_xdrInput = xdrbuf::Input::create();
        ASSERT(m_xdrInput, "Failed to create xdr input parser");

        // create xdr output buffer for UDP broadcasts
        ASSERT(!m_xdrBcast, "Already have an output buffer");
        m_xdrBcast = xdrbuf::Output::create(s_bcastBuffer);
        ASSERT(m_xdrBcast, "null");

        // requested server loop speed?
        m_requestedHz = getFloat(params, "requestedHz");
        if (m_requestedHz < 1.0 || m_requestedHz > 200.0) {
                WAVE_EX(wex);
                wex << "Invalid requested Hz: " << m_requestedHz;
                wex << ". Should be be between 1 and 200 Hz";
        }

        // cycles per UDP send?
        m_udpHz = getFloat(params, "udpHz");
        if (m_udpHz < 1 || m_udpHz > 20) {
                WAVE_EX(wex);
                wex << "Invalid udp Hz: " << m_udpHz;
        }

        int udpSize = getInt(params, "udpSize");
        if (udpSize < 64 || udpSize > 1024) {
                WAVE_EX(wex);
                wex << "Invalid udp packet size: " << udpSize;
        }

        // default locale (see TODO file regarding locales!)
        const char * locale = i18n::getHostLocale();
        ASSERT_THROW(locale, "failed to determine host locale?");
        DPRINTF("Using locale: '%s'", locale);

        // load story
        const char * story_dir = getString(params, "storyDir");
        m_story = story::Story::create(locale, story_dir);
        ASSERT(m_story, "failed to load story");

        // initialize thread pool
        smart_ptr<Datahash> pool_params = getSubhash(params, "threadpool");
        ASSERT(pool_params, "should have subhash for threadpool init");
        m_threadpool = threadpool::Pool::create(pool_params);
        ASSERT(m_threadpool, "failed to create thread pool for server");

        // figure out where to broadcast
        // Our "best guess" is the last non-loopback interface found
        netlib::map_ip_addr_t ips;
        netlib::getLocalInterfaces(ips);
        netlib::ip_addr_t bestGuess;
        DPRINTF("Checking server interfaces:");
        for (netlib::map_ip_addr_t::const_iterator i = ips.begin();
             i != ips.end(); ++i) {
                const char * name = i->first.c_str();
                const netlib::ip_addr_t& ip = i->second;

                ip.dump(name);
                if (!ip.isLoopback()) {
                        bestGuess = ip;
                }
        }
        if (!bestGuess.isValid()) {
                DPRINTF("No non-loopback interfaces available!");
                DPRINTF("Server is using loopback interface only");
                bestGuess.setLoopback();
        }
        bestGuess.setBroadcast();

        // broadcast data
        netlib::address_t bcastAddress;
        int clientPort = eDefaultPort_clientUdp;
        const char * clientPortStr =
            getString(params, "clientPort", eDatahash_Optional);
        if (!clientPortStr) {
                DPRINTF("No client UDP port specified, using default: %d",
                    clientPort);
        } else {
                DPRINTF("Overriding default client port of %d", clientPort);
                clientPort = atoi(clientPortStr);
                DPRINTF("  with new port value: %d", clientPort);
                ASSERT2(clientPort > 0, "Bad client udp port: " << clientPort);
        }
        const char * bcastIP =
            getString(params, "broadcastAddress", eDatahash_Optional);
        if (!bcastIP) {
                DPRINTF("No broadcastAddress specified, using auto-detected address");
                bcastAddress.ip = bestGuess;
                bcastAddress.port = clientPort;
        } else {
                DPRINTF("Overriding autodetected broadcast address");
                bestGuess.dump("autodetected broadcast address");
                bcastAddress.set(bcastIP, clientPort);
        }
        bcastAddress.dump("UDP broadcast address");
        m_publicName = getString(params, "publicName");

        // udp ports
        m_udpPort = getInt(params, "udpPort");
        ASSERT(m_udpPort > 0, "Bad udp port: %d", m_udpPort);
        ASSERT(!m_udpConn, "already have a UDP listener?");
        netlib::address_t udpAddress;
        udpAddress.setlocal(m_udpPort);
        udpAddress.dump("UDP address");

        m_udpConn = netlib::createUdpLocal(udpAddress);
        ASSERT(m_udpConn, "null");
        ASSERT(!m_udpBcast, "already have a UDP broadcaster?");
        m_udpBcast = netlib::createUdpBroadcast(bcastAddress);
        ASSERT2(m_udpBcast, "null");

        // message router
        m_router = MessageRouter::create();
        ASSERT(m_router, "Failed to create message router");

        // conversation manager
        m_conversationMgr = converse::ConversationManager::create(this);
        ASSERT(m_conversationMgr, "failed to create conversation manager");

        // user manager
        m_userMgr = UserManager::create(params);
        ASSERT(m_userMgr, "failed to create user manager");

        // player manager
        m_playerMgr = PlayerManager::create(m_userMgr);
        ASSERT(m_playerMgr, "failed to create player manager");

        // host manager
        m_hostMgr = HostManager::create(udpSize);
        ASSERT(m_hostMgr, "failed to create host manager");

        // map manager
        m_mapMgr = MapManager::create(params, m_story);
        ASSERT(m_mapMgr, "failed to create map manager");

        // hook up game logic object
        m_gameLogic->setPlayerManager(m_playerMgr);
        m_gameLogic->setMapManager(m_mapMgr);

        // set up tcp listener
        m_tcpPort = getInt(params, "tcpPort");
        ASSERT(m_tcpPort > 0, "Bad listening port: %d", m_tcpPort);
        int backlog = getInt(params, "tcpBacklog");
        ASSERT(backlog > 0, "Bad max backlog: %d", backlog);
        ASSERT(!m_tcpConn, "already have a tcp listener?");
        netlib::address_t tcpAddress;
        tcpAddress.setlocal(m_tcpPort);
        tcpAddress.dump("TCP Address");
        m_tcpConn = netlib::createTcpListener(tcpAddress, backlog);
        ASSERT(m_tcpConn, "failed to create tcp listener socket");

        // initialize type instance library
        initializeTypeInstanceLibrary(m_story);

        // successful initialization
        m_smode = eServerMode_Idle;
}



// TODO: there is one weirdness with the networking library.  Based
// on the core messaging loop, TCP reads and sends still happen on this
// thread.  At the moment, this is considered desirable because some
// important connection state is shared for all threads, so it may
// make sense to keep the low-level state and routing on the main
// thread rather than having a lot of locking.  If this turns into a
// performance issue then even the low level TCP read/writes should be
// moved off this thread.  This isn't expected to be an issue because
// all TCP reads/writes are non-blocking.


int
ServerImpl::exec
(
void
)
{
        perf::Timer timer("aesop-srv::main-loop");

        // given requested Hz (loops/second), determine microseconds per loop
        ASSERT(m_requestedHz > 0 && m_requestedHz < 1000,
            "Bad server frequency?  %f hz", m_requestedHz);
        int loop_us = (int) (1.0e6 / m_requestedHz);
        int max_usec = loop_us + 1000;  // maximum allowed value
        loop_us -= 1000;        // take off another ms for good measure
        DPRINTF("Server loop requested to run at %.1f hz", m_requestedHz);
        DPRINTF("  That means %d microseconds per loop", loop_us);
        ASSERT(loop_us > 0 && loop_us < 1000 * 1000,
            "Bad loop microseconds? %d", loop_us);

        // how many loop cycles per UDP send?
        DPRINTF("UDP packets requested at %.1f hz", m_udpHz);
        int udpSend = (int) ((m_requestedHz / m_udpHz) + 0.5);
        DPRINTF("Sending udp packets every %d cycles", udpSend);

        // time delta per loop
        perf::time_t dt(0, loop_us);

        // time delta between broadcasts
        ASSERT2(m_bcastDelay > 0, "Bad delay: " << m_bcastDelay);
        perf::time_t bcastDt(m_bcastDelay, 0);

        // target time for next loop (now + dt)
        perf::time_t target = perf::getNow();
        target.increment(dt);

        // target time for next broadcast
        perf::time_t targetBcast = perf::getNow();
        targetBcast.increment(bcastDt);

        perf::time_t benchmark = perf::getNow();
        static const int s_benchmarkCount = (int) (3 * m_requestedHz);

        // main loop
        int udpCount = 0;
        perf::time_t lastLoopTime = perf::getNow();
        for (long loop_counter = 0; !m_shutdown; ++loop_counter, udpCount++) {

                // what time is now?
                perf::time_t now = perf::getNow();
                perf::time_t delta = now;
                delta.decrement(lastLoopTime);
                lastLoopTime = now;

                // need to broadcast?
                if (now.isGreaterThan(targetBcast)) {
                        this->sendBroadcast();
                        targetBcast = now;
                        targetBcast.increment(bcastDt);
                }

                // recalibrate time delta.  This loop will try to stay close
                //   to the desired Hz rate.
                if (s_benchmarkCount == loop_counter) {
                        loop_counter = 0;
                        perf::time_t delta = now;
                        delta.decrement(benchmark);
                        double delta_t = delta.getSeconds();
                        if (delta_t > 0.01) {
                                double hz = s_benchmarkCount / delta_t;
                                double dr = (hz / m_requestedHz) - 1.0;

                        //      DPRINTF("%6.2lf Hz server loop", hz);

                                double s = dt.getSeconds();
                                s *= 1.0 + 0.15 * dr;           // s is now requested dt
                                int usec = (int) (1000 * 1000 * s);
                                if (usec >= 0 && usec < max_usec) {
                                        perf::time_t new_dt(0, usec);
                                        dt = new_dt;
                                }
                        }
                        benchmark = now;
                }

                // check on player state
                this->checkPlayers();

                // update all game state
                float delta_seconds = (float) delta.getSeconds();
                if (delta_seconds < 1.0e-6) {
                        // force to be a small positive number
                        delta_seconds = 1.0e-6;
                }
                m_mapMgr->tickMaps(delta_seconds);

                // let the game logic run
                {
                        perf::Timer timer("game-logic-tick");
                        m_gameLogic->tick(delta_seconds);
                }

                // send udp messages?
                if (udpCount >= udpSend) {
                        udpCount = 0;
                        try {
                                this->sendUdp();
                        } catch (std::exception& e) {
                                DPRINTF("Failed to send UDP packets: %s",
                                        e.what());
                        }
                }

                // message pump
                while (true) {
                        perf::time_t now = perf::getNow();
                        if (now.isGreaterThan(target)) {
                                target = now;
                                target.increment(dt);
                                break;
                        }

                        // send an enqueued message
                        request_t req;
                        if (m_outboundMessages.pop_front(req)) {
                                ASSERT(!req.is_empty(),
                                    "empty message in queue");
                                netlib::enqueueMessage(req.conn_id,
                                    req.message);
                        }

                        // read messages
                        perf::time_t delta;
                        delta = target;
                        delta.decrement(now);
                        this->checkMessages(delta.getMicroseconds());
                }
        }
        DPRINTF("Shutting down server!");

        // we need to wait until all worker threads have exited
        DPRINTF("Waiting for worker threads to exit...");
        for (;;) {
                long lCount = m_workerTrc.getRefCount();
                if (lCount < 2) {
                        break;          // all done!
                }
                DPRINTF("  Waiting for %ld workers to exit...", lCount - 1);

                // debugging: who is working?
                name_map_t::iterator_t i;
                m_threadNames.getIterator(i);
                int nThread;
                std::string name;
                while (m_threadNames.getNextElement(i, nThread, name)) {
                        DPRINTF("    Thread %3d = %s", nThread, name.c_str());
                }

                // TODO: could force exit after 30 seconds etc...
                sleep(1);
        }

        // any closing stuff?
        netlib::dumpStats();

        // all done
        return m_retval;
}



void
ServerImpl::requestStopTS
(
IN int retval
)
{
        // caller is requesting that we stop
        // this method needs to be threadsafe without any locking!  This is
        // typically used by signal handlers etc.

        // these fields are safe to be set by an arbitrary thread, since we
        // don't care about race conditions on reading
        m_retval = retval;
        m_shutdown = true;
}



//
//      ServerImpl -- aesop::StateUpdates class interface methods
//
//      All methods should be threadsafe, hence the TS suffix.
//

void
ServerImpl::enqueueMessageTS
(
IN conn_id_t connId,
IN smart_ptr<MessageBuffer>& message
)
{
        ASSERT(connId, "null");
        ASSERT(message, "null");

        // threadsafe!  add to our threadsafe list
        request_t req;
        req.conn_id = connId;
        req.message = message;
        m_outboundMessages.push_back(req);
}



void
ServerImpl::newTcpConnectionTS
(
IN conn_id_t connId,
IN long token
)
{
        ASSERT(connId, "null");
        ASSERT(token, "null");

        // must be threadsafe!

        // see if we can find the host based on the token
        host_rec_t hr;
        if (!m_hostMgr->getHostByUdpConnectionID(token, hr)) {
                DPRINTF("Unrecognized host in new tcp connection");
                return;
        }

        // have a host--does it match the IP?
//      netlib::connection_info_t ci1, ci2;

//      netlib::getConnectionInfo(token, ci1);
//      netlib::getConnectionInfo(connId, ci2);

//      ci1.dump("Local host information");
//      ci2.dump("Remote host information");

        // for now, just compare IP addresses
//      if (ci1.address.ip != ci2.address.ip) {
//              DPRINTF("TCP connect info does not match UDP?");
//              return;         // disallow TCP connection
//      }

        // okay, valid!
        hr.tcpConn = connId;
        m_hostMgr->updateHost(hr);
}



void
ServerImpl::newPlayer
(
IN host_rec_t& hr,
IN int playerId
)
{
        ASSERT(hr.isValid(), "invalid");
        ASSERT(playerId > 0, "Bad player ID: %d", playerId);

        // NOTE: this entire logic chain needs to be idempotent!

        // this routine must be threadsafe!
        // this happens right now because hostMgr and playerMgr are threadsafe

        // only allowed if host is in good state
        if (eHostMode_Authorized != hr.mode) {
                DPRINTF("Host is not authorized to add players");
                return;
        }

        // also only allowed if host has TCP connection
        if (!hr.tcpConn) {
                DPRINTF("Cannot add players: no tcp connection");
                return;
        }

        // add player!
        player_rec_t pr;
        pr.udpConnId = hr.udpConn;
        pr.playerId = playerId;
        m_playerMgr->updatePlayer(pr);

        // fire up a new player conversation
        std::ostringstream oss;
        oss << "new-player:" << pr.udpConnId << ":" << playerId;

        // create conversation host
        new_player_conv_data_t data;
        data.updates = this;
        data.userMgr = m_userMgr;
        data.connId = hr.tcpConn;
        data.playerId = playerId;
        ASSERT(data.isValid(), "should be valid now");

        smart_ptr<converse::ConversationHost> host =
            createNewPlayerConversationHost(hr.desKey, data);
        ASSERT(host, "null");

        // start up conversation
        ASSERT(m_conversationMgr, "null");
        m_conversationMgr->updateConversationTS(oss.str().c_str(), hr.tcpConn,
            playerId, host);
}



void
ServerImpl::notifyDialogReplyTS
(
IN conn_id_t conn_id,
IN const char * conversationGuid,
IN int dialogId,
IN int playerId,
IN const Datahash * reply
)
{
        ASSERT(conn_id, "Null");
        ASSERT(conversationGuid, "null");
        ASSERT(dialogId > 0, "Bad dialog id: %d", dialogId);
        ASSERT(playerId > 0, "Bad player id: %d", playerId);
        ASSERT(reply, "null");

        // threadsafe!
        // use threadsafe conversation manager
        ASSERT(m_conversationMgr, "null");
        m_conversationMgr->handleDialogReplyTS(conversationGuid,
                                        dialogId,
                                        conn_id,
                                        playerId,
                                        reply);
}



bool
ServerImpl::setPlayerUserAccountTS
(
IN conn_id_t conn_id,
IN int playerId,
IN const char * username,
IN const char * usertag,
OUT std::string& diagnostic
)
{
        ASSERT(conn_id, "null");
        ASSERT(playerId > 0, "bad player id: %d", playerId);
        ASSERT(username, "null");
        ASSERT(usertag, "null");
        ASSERT(m_userMgr, "null");
        diagnostic.clear();

        // check about admin first (MUST happen before lock, disk hit)
        bool isAdmin = m_userMgr->isAdminTS(username);
        if (isAdmin) {
                DPRINTF("User '%s' is an admin!", username);
        }

        // get host
        host_rec_t hr;
        if (!m_hostMgr->getHostByTcpConnectionID(conn_id, hr)) {
                DPRINTF("Cannot set player user account: bad host");
                return false;
        }

        // get player record
        player_rec_t pr;
        if (!m_playerMgr->getPlayerByHostAndPlayerId(hr.udpConn, playerId, pr)) {
                diagnostic = "Internal Error: No record for this player!";
                return false;
        }

        // update local copy
        pr.username.set(username);
        pr.playerTag.set(usertag);
        if (isAdmin) {
                pr.flags |= ePrivilege_Admin;
        }

        // update player record
        m_playerMgr->updatePlayer(pr);

        // allowed!
        return true;
}



void
ServerImpl::newGameTS
(
IN conn_id_t connId,
IN int playerId
)
{
        ASSERT(connId, "null");
        ASSERT(playerId > 0, "Bad player id: %d", playerId);

        // threadsafe!

        // get host
        host_rec_t hr;
        if (!m_hostMgr->getHostByTcpConnectionID(connId, hr)) {
                DPRINTF("new game request ignored: bad host?");
                return;
        }
        conn_id_t hostId = hr.udpConn;

        // first verify this is a valid admin
        // no lock needed
        if (!this->isAdmin(hr.udpConn, playerId)) {
                DPRINTF("Player %d is not admin, cannot start game",
                    playerId);
                return;
        }

        // okay!
        DPRINTF("playerId=%d from connection 0x%04lx has requested a new game",
            playerId, (long) hostId);

        // nuke all existing maps!
        m_mapMgr->clear();

        // request that starting map be loaded
        std::string id = m_story->getStartingMapId();
        DPRINTF("  Starting map id='%s'", id.c_str());
        m_mapMgr->requestLoad(id.c_str());
}



void
ServerImpl::requestShutdownTS
(
void
)
{
        // This needs to be threadsafe!
        // However, we can cheat here.  the shutdown flag is special:
        //   it is always false, until it goes true, and stays true
        // So we can just set it to true here, and not worry if another
        //   thread reads it as false at the same time.  Other threads
        //   will eventually figure out we are shutting down...

        DPRINTF("Requested to shut down!  Setting shutdown flag...");
        m_shutdown = true;
}



//
//      ServerImpl -- converse::ConversationRouter class interface methods
//

void
ServerImpl::routeConversationTS
(
IN const char * guid,
IN int dialogId,
IN int playerId,
IN conn_id_t connId,
IN const char * dialogData
)
{
        ASSERT(guid, "null");
        ASSERT(dialogId >= 0, "Bad dialog id: %d", dialogId);
        ASSERT(playerId > 0, "Bad player id: %d", playerId);
        ASSERT(connId, "null");
        ASSERT(dialogData, "null");

        smart_ptr<MessageBuffer> msg;
        // threadsafe!
        // end of conversation?
        if (!dialogId) {
                DPRINTF("Ending conversation: %s", guid);
                msg = createTerminateConversationMessage(playerId, guid);
        } else {
                // construct dialog message
                msg = createConversationDialogMessage(playerId, guid,
                    dialogId, dialogData);
        }
        ASSERT(msg, "null");

        // threadsafe -- enqueue
        this->enqueueMessageTS(connId, msg);
}



//
//      ServerImpl -- private helper methods
//

void
ServerImpl::checkMessages
(
IN long microseconds
)
{
        perf::Timer timer("aesop-srv::checkMessages");
        ASSERT(microseconds >= 0, "Bad microseconds: %ld", microseconds);

        envelope_t env;
        smart_ptr<MessageBuffer> msg;
        if (!netlib::getNextMessage(microseconds, env, msg))
                return;         // nothing to do!

        try {
                // what kind of message?
                if (netlib::eType_UDPRemote == env.type) {
                        // udp packet from client
                        this->handleUdp(env, msg);
                } else if (netlib::eType_TCP == env.type) {
                        // tcp packet from client
                        this->handleMessage(env, msg);
                } else {
                        DPRINTF("Unknown message type!  %d", env.type);
                }

        } catch (std::exception& e) {
                dumpMessage(std::cerr, "Exception handling this message",
                    env, msg);
                DPRINTF("Exception: %s", e.what());
        }
}



void
ServerImpl::handleMessage
(
IN const envelope_t& env,
IN const MessageBuffer * msg
)
{
        perf::Timer timer("aesop-srv::handleMessage");
        ASSERT(!env.is_empty(), "empty envelope");
        ASSERT(msg, "null");

        // TCP?  get payload
        tcp_payload_t payload;
        getTcpPayload(msg->getData(), payload);
        ASSERT(!payload.is_empty(), "empty");

        DPRINTF("Got message: %s", payload.command.c_str());
        
        // get handler for this
        msg_handler_fn handler = m_router->getHandler(payload);
        if (!handler) {
                return;         // TODO: send back an error code?
        }

        // construct state object
        smart_ptr<handler_intercept_t> hi =
            new handler_intercept_t(m_workerTrc);
        ASSERT(hi, "out of memory");
        hi->fn = handler;
        hi->state.src_env = env;
        hi->state.payload = payload;
        hi->state.story = m_story;
        hi->state.updates = this;
        hi->map = &m_threadNames;

        // construct name
        ++m_threadCounter;
        m_threadNames.insert(m_threadCounter, payload.command.c_str());

//      DPRINTF("Worker count: %ld", m_workerTrc.getRefCount());

        // pass to thread pool
        // TODO: return busy if threadpool is at capacity
        void * ctx = (void *) ((handler_intercept_t *) hi);
        m_threadpool->addRequest(interceptHandler, ctx);
        hi.disown();    // thread will free this when it is done

        // nothing else to do!
}



void
ServerImpl::handleUdp
(
IN const envelope_t& env,
IN const MessageBuffer * msg
)
{
        perf::Timer timer("Server::handleUdp");
        ASSERT(msg, "null");

        // for now, we handle UDP messages in the main thread
        // Any processing that requires different threads should
        // be moved to TCP messages!

        // NOTE: this must be threadsafe!  Other threads could be accessing
        //      data structures as we are parsing the UDP datagram.
        // All data access is behind threadsafe structures.

        // reset parser with input data
        ASSERT(m_xdrInput, "should have parser by now");
        m_xdrInput->reset((const byte_t *) msg->getData(), msg->getBytes());

        // first thing should be the client (host) global state
        xdrbuf::PackletHeader ph = m_xdrInput->getNextPackletHeader();
        if ('h' != ph.getName()) {
                // 'h' -- header
                DPRINTF("Malformed UDP packet--no header?");
                return;
        }
        int32_t host_state[2];  // header is 2 longs: token and clock
        m_xdrInput->readInt32s(host_state, 2);

        // verify host token
        dword_t token = host_state[0];
        conn_id_t conn_id = 0;
        host_rec_t hr;

        if (token) {
                // do netlib lookups outside of mutex
                //  (safe since we're on the main thread)
                netlib::connection_info_t ci;
                if (!netlib::getConnectionInfo(token, ci)) {
                        DPRINTF("Input UDP token was not valid");
                        return;
                }

                // do address and port match?
                if (ci.address != env.address) {
                        DPRINTF("UDP sender does not match token");
                        return;
                }

                // appears to be a valid token!
                conn_id = (conn_id_t) token;
        }

        if (conn_id) {
                // appears to be a valid UDP packet from client
                if (!m_hostMgr->getHostByUdpConnectionID(conn_id, hr)) {
                        DPRINTF("No host record for conn id: 0x%04lx",
                            (long) conn_id);
                        return;
                }
        
                // remember client's udp clock
                dword_t clock = (dword_t) host_state[1];
                hr.lastUdpReceived = clock;
                m_hostMgr->updateHost(hr);
        }

        // parse message
        while (!m_xdrInput->endOfStream()) {
                xdrbuf::PackletHeader ph = m_xdrInput->getNextPackletHeader();
                char name = ph.getName();
                //DPRINTF("Parsing packlet with name: '%c'", name);
                switch (name) {

                case 'a':       // auth attempt
                        this->parseHostAuthUdp(hr, ph, m_xdrInput);
                        break;

                case 'c':       // connection attempt
                        this->parseHostConnectUdp(env, m_xdrInput);
                        break;

                case 'e':       // client is asking if object exists
                        this->parseExistsUdp(hr, m_xdrInput);
                        break;

                case 'g':       // client is sending game data/events
                        this->parseGameUdp(hr, ph, m_xdrInput);
                        break;

                case 'p':       // player data
                        this->parsePlayerUdp(hr, m_xdrInput);
                        break;

                case 'q':       // client is sending a query request
                        this->parseQueryUdp(hr, ph, m_xdrInput);
                        break;

                default:
                        DPRINTF("Unknown packlet name! '%c'", name);
                }
        }
//      DPRINTF("End of parsing!");
}



void
ServerImpl::parseHostConnectUdp
(
IN const netlib::envelope_t& env,
IO xdrbuf::Input * input
)
{
        ASSERT(input, "null");

        // host is attempting to connect
        // must be threadsafe!

        DPRINTF("Connection attempt...");
        // ASSERT(hr.isValid()) -- host can be invalid for connection attempts!

        // next should be version and token
        xdrbuf::PackletHeader ph = input->getNextPackletHeader();
        if ('v' != ph.getName()) {
                WAVE_EX(wex);
                wex << "Invalid packlet header during connection";
        }
        int32_t l[3];
        input->readInt32s(l, 3);
        if (eProtocol_Long != l[0]) {
                WAVE_EX(wex);
                wex << "Protocol versions do not match";
        }
        dword_t token = l[1];
        if (!token) {
                WAVE_EX(wex);
                wex << "Invalid client connection token";
        }
        int port = (int) l[2];
        if (!port) {
                WAVE_EX(wex);
                wex << "Invalid client UDP port";
        }

        DPRINTF("  token = %ld", (long) token);
        DPRINTF("  port = %d", port);

        // next should be public key
        ph = input->getNextPackletHeader();
        if ('p' != ph.getName() ||
            xdrbuf::ePacklet_ParentBegin != ph.getType()) {
                WAVE_EX(wex);
                wex << "Invalid packlet header during connection";
        }

        // keep reading pieces of the public key...
        const long maxSize = 1024;
        char buffer[maxSize];
        char * offset = buffer;
        int read = 0;
        while (true) {
                xdrbuf::PackletHeader subph = input->getNextPackletHeader();
                if ('p' == subph.getName()) {
                        break;          // end of key pieces
                }
                if (xdrbuf::ePacklet_String != subph.getType()) {
                        WAVE_EX(wex);
                        wex << "Unexpected packlet while reading public key";
                }

                int len = subph.getDataCount();
                if (read + len > maxSize - 1) {
                        WAVE_EX(wex);
                        wex << "Public key is too long!";
                }

                input->readString(offset, len);
                offset += len;
                read += len;
        }
        buffer[read] = 0;               // buffer now contains public key

        // should have an end-of-packlet tag now
        ph = input->getNextPackletHeader();
        if ('c' != ph.getName() ||
            xdrbuf::ePacklet_ParentEnd != ph.getType()) {
                WAVE_EX(wex);
                wex << "Invalid end of connection packlet";
        }

        // do we already have a connection to this client?
        // TODO / WARNING : This is inefficient.  We have to loop through
        // all the known hosts to see if one of them matches.  Could make
        // this a quicker lookup.  This isn't an issue for the expected
        // use case of these libraries (16 connected hosts at max) but
        // the host manager (HostManager) should be updated to find
        // hosts based on server and port (or other ID) faster--or keep
        // this lookup on the client (that is, in this server object).
        host_rec_t hr;
        aesop::HostManager::iterator_t i;
        m_hostMgr->getIterator(i);
        while (m_hostMgr->getNextHost(i, hr)) {
                netlib::connection_info_t ci;
                if (netlib::getConnectionInfo(hr.udpConn, ci)) {
                        if (env.address == ci.address) {
                                DPRINTF("Already connected!");
                                DPRINTF("  Removing old host record");
                                m_hostMgr->removeHost(hr.udpConn);
                        //      return;
                        }
                }
        }

        // okay, looks to be a legit request!  create a udp connection
        std::string server = netlib::getServerFromIP(env.address.ip);
        DPRINTF("  server='%s'", server.c_str());
        netlib::address_t address(server.c_str(), port);
        ASSERT2(address.isValid(), "Invalid client udp address");
        conn_id_t hostId = netlib::createUdpRemote(m_udpConn, address);
        if (!hostId) {
                WAVE_EX(wex);
                wex << "Failed to create remote UDP connection";
        }

        // success!  create host record and queue a connection response
        m_hostMgr->addHost(hostId, token, buffer);
        if (!m_hostMgr->getHostByUdpConnectionID(hostId, hr)) {
                WAVE_EX(wex);
                wex << "Cannot find a just-added host?";
        }
        ASSERT(hr.netQueue, "null network request queue for host");
        ASSERT(hr.desKey, "null");
        ASSERT(hr.publicKey, "null");

        // create context to send back the encrypted DES key
        std::string desKey = hr.desKey->serialize();
        std::string encryptedDesKey = hr.publicKey->encrypt(desKey.c_str());
        smart_ptr<netrq::Request> rc =
            new AuthContext(hostId, encryptedDesKey.c_str());
        hr.netQueue->addRequest(m_udpClock + 2, rc);

        DPRINTF("Host should have connected!  connId=%08lx", (long) hostId);
}



void
ServerImpl::parseHostAuthUdp
(
IN host_rec_t& hr,
IN const xdrbuf::PackletHeader& ph,
IN xdrbuf::Input * input
)
{
        ASSERT(input, "null");

        if (!hr.isValid()) {
                WAVE_EX(wex);
                wex << "Ignoring auth attempt from invalid host";
        }
        if (eHostMode_Authorized == hr.mode) {
                DPRINTF("Host is already authorized!");
                return;
        }
        DPRINTF("Host is attempting to authenticate");

        // must be threadsafe!

        // this should be a single string!
        if (xdrbuf::ePacklet_String != ph.getType()) {
                WAVE_EX(wex);
                wex << "Invalid packlet type";
        }

        const int maxSize = 1024;
        int len = ph.getDataCount();
        if (len < 1 || len > maxSize - 1) {
                WAVE_EX(wex);
                wex << "Bad password length: " << len;
        }

        char buffer[maxSize];
        input->readString(buffer, len);
        buffer[len] = 0;

        DPRINTF("Received string: %s", buffer);
        ASSERT(hr.desKey, "null");
        std::string decrypted = hr.desKey->decrypt(buffer);
        DPRINTF("Got host password: '%s'", decrypted.c_str());

        hr.mode = eHostMode_Authorized;
        m_hostMgr->updateHost(hr);
}



void
ServerImpl::parsePlayerUdp
(
IN host_rec_t& hr,
IO xdrbuf::Input * input
)
{
        ASSERT(input, "null");

        // must be threadsafe!

        // check host
        if (!hr.isValid()) {
                WAVE_EX(wex);
                wex << "Cannot process player data from invalid host";
        }

        // should have ID packlet now
        xdrbuf::PackletHeader ph = input->getNextPackletHeader();
        char name = ph.getName();
        if ('i' != name) {
                WAVE_EX(wex);
                wex << "Invalid packlet in player UDP data.  Expected ID 'i' ";
                wex << "but received '" << name << "' instead.";
        }
        int32_t playerId = -1;
        input->readInt32s(&playerId, 1);
        if (playerId < 1) {
                WAVE_EX(wex);
                wex << "Invalid player id: " << playerId;
        }

        // get this player's record
        player_rec_t rec;
        player_rec_t * pr = NULL;
        if (m_playerMgr->getPlayerByHostAndPlayerId(hr.udpConn, playerId, rec))
        {
                // found the record!
                pr = &rec;
        }
        if (!pr) {
                // create the player record, but leave pr null
                // this means we will ignore further player data in this udp
                this->newPlayer(hr, playerId);
        }

        // parse all data for player
        for (;;) {
                ph = input->getNextPackletHeader();
                name = ph.getName();

                switch (name) {
                case 'p':
                        return;         // end of player packlet

                case 'm':
                        {
                                // player wants to move
                                float f[7];
                                input->readFloats(f, 7);
                                float dt = f[3];
                                if (dt > 0.0 && pr && pr->obj) {
                                        // TODO: game logic gets this!
                                        // for now we do some simple logic here
                                        point3d_t delta(f[0], f[1], f[2]);
                                        //delta.dump("  requested move");

                                        // NOTE: euler angles are f[4] - f[6]

                                        // what if delta is too big?  Can lead
                                        // to funky oscillations.  Keep to max
                                        float r2 = dotProduct(delta, delta);
                                        float r = sqrt(r2);
                                        if (r > 0.4) {
                                                delta = (0.4 / r) * delta;
                                        }
                                        pr->obj->requestMove(delta, dt);

                                        // set orientation
                                        // (we only use y-rotation for now...)
                                        point3d_t euler(0, f[5], 0);
                                        quaternion_t q;
                                        q.setEulerZYX(euler);
                                        pr->obj->setOrientation(q);
                                }
                        }
                        break;

                default:
                        {
                                WAVE_EX(wex);
                                wex << "Unrecognized player packet: '";
                                wex << name << "'";
                        }
                }       // end of switch statement
        }       // end of packlet header loop
}



void
ServerImpl::parseQueryUdp
(
IN host_rec_t& hr,
IN const xdrbuf::PackletHeader& ph,
IO xdrbuf::Input * input
)
{
        ASSERT(input, "null");

        // must be threadsafe!

        // check host
        if (!hr.isValid()) {
                WAVE_EX(wex);
                wex << "Cannot process query from invalid host";
        }

        // validate packlet header
        if (xdrbuf::ePacklet_Int32s != ph.getType()) {
                WAVE_EX(wex);
                wex << "query packlets should be of type int32";
        }
        dword_t objId;
        input->readInt32s((int32_t *) &objId, 1);

        //DPRINTF("Client wants information about object: %ld", objId);
        if (!objId) {
                WAVE_EX(wex);
                wex << "Invalid object ID requested";
        }

        // add response to this client's request queue
        ASSERT(hr.netQueue, "null");
        if (hr.netQueue->size() > s_maxNetQueueLength) {
                DPRINTF("Dropping client request");
                return;         // don't add anything, we have enough to do
        }

        smart_ptr<netrq::Request> ctx = new QueryContext(objId);
        ASSERT(ctx, "out of memory");
        hr.netQueue->addRequest(m_udpClock + 2, ctx);
}



void
ServerImpl::parseGameUdp
(
IN host_rec_t& hr,
IN const xdrbuf::PackletHeader& ph,
IO xdrbuf::Input * input
)
{
        perf::Timer timer("parseGameUdp");
        ASSERT(input, "null");
        ASSERT(m_gameLogic, "null");

        m_gameLogic->parseGameData(hr.udpConn, input);
}



void
ServerImpl::parseExistsUdp
(
IN host_rec_t& hr,
IO xdrbuf::Input * input
)
{
        ASSERT(input, "null");

        // host is asking if this object exists.
        // get the object ID the client is interested in
        dword_t objectId;
        input->readInt32s((int32_t *) &objectId, 1);
        if (objectId < 1) {
                WAVE_EX(wex);
                wex << "Bad object ID: " << objectId;
        }

        // create a message ID for this response
        char buffer[64];
        sprintf(buffer, "er-%ld", (long) objectId);
        if (hr.netQueue->containsRequest(buffer))
                return;         // already pending!

        // need to send a message back
        smart_ptr<netrq::Request> request =
            ObjectExistsResponse::create(buffer, objectId);
        ASSERT(request, "null");
        hr.netQueue->addRequest(m_udpClock, request);
}



void
ServerImpl::sendUdp
(
void
)
{
        // loop through all hosts and send!
        // this must be threadsafe.  We don't need to acquire the server
        //      mutex since all collections are threadsafe.
        // increment UDP clock
        ++m_udpClock;

        // set up outbound physics updates
        addPhysicsMessagesToQueue(m_udpClock, m_hostMgr, m_playerMgr, m_mapMgr);

        // send to each host
        vec_conn_t badConns;            // bad connections
        HostManager::iterator_t i;
        host_rec_t hr;
        m_hostMgr->getIterator(i);
        while (m_hostMgr->getNextHost(i, hr)) {
                if (!hr.outbuf || !hr.udpConn) {
                        DPRINTF("No way to send");
                        continue;       // skip this one
                }

                // is this still a good host connection?
                if (!netlib::isValidConnection(hr.udpConn)) {
                        DPRINTF("Removing host record!  0x%04lx",
                            (long) hr.udpConn);
                        badConns.push_back(hr.udpConn);
                        continue;               // skip host
                }

                // construct data!
                hr.outbuf->clear();

                // UDP header is 6 longs:
                //      0: token
                //      1: current server clock
                //      2: last clock received from client
                //      3: server status
                //      4: host status
                //      5: server's listening TCP port
                int32_t l[6];
                l[0] = hr.udpToken;
                l[1] = m_udpClock;
                l[2] = hr.lastUdpReceived;
                l[3] = m_smode;
                l[4] = hr.mode;
                l[5] = m_tcpPort;

                // send general state ('h' -- header)
                hr.outbuf->addInt32Packlet('h', l, 6);

                // players on this host
                PlayerManager::iterator_t i;
                m_playerMgr->getIterator(hr.udpConn, i);
                player_rec_t pr;
                while (m_playerMgr->getNextPlayer(i, pr)) {
                        // send player status
                        dword_t objectId = pr.obj ? pr.obj->getId() : 0;

                        // Player status is 3 longs right now:
                        //   0: playerId
                        //   1: player state (enum)
                        //   2: object ID (can be null)
                        int32_t l[3];
                        l[0] = pr.playerId;     // playerId
                        l[1] = pr.getMode();
                        l[2] = objectId;
                        hr.outbuf->addInt32Packlet('p', l, 3);
                }

                // add any pending messages, as space permits
                netrq::sendMessagesFromQueue(hr.outbuf, m_udpClock, hr.netQueue);

                // other data?

                // construct netlib message
                // TODO: avoid the copy here!
                smart_ptr<MessageBuffer> msg = MessageBuffer::create();
                ASSERT(msg, "null");
                const byte_t * data = hr.outbuf->getData();
                int bytes = hr.outbuf->getDataBytes();
                ASSERT(bytes > 0, "Bad bytes: %d", bytes);
                msg->reserve(bytes);
                msg->appendData((const char *) data, bytes);
                msg->close();

                // send
                netlib::enqueueMessage(hr.udpConn, msg);
        }
//      DPRINTF("Done sending data to hosts");

        // remove bad hosts
        for (vec_conn_t::iterator ib = badConns.begin(); ib != badConns.end();
             ++ib) {
                conn_id_t conn_id = *ib;
                DPRINTF("  Removing host with connection ID 0x%04lx",
                    (long) conn_id);
                m_hostMgr->removeHost(conn_id);
        }
}



const char *
ServerImpl::getStoryGuidTS
(
void
)
{
        ASSERT(m_story, "null");

        return m_story->getUuid();
}



bool
ServerImpl::isAdmin
(
IN conn_id_t udpConnID,
IN int playerId
)
{
        ASSERT(udpConnID, "null");
        ASSERT(playerId > 0, "bad player id: %d", playerId);
        ASSERT(m_playerMgr, "null");

        // NOTE: this method is threadsafe, since playerMgr is threadsafe
        player_rec_t pr;
        if (!m_playerMgr->getPlayerByHostAndPlayerId(udpConnID, playerId, pr)) {
                return false;
        }
        return ePrivilege_Admin & pr.flags;
}



void
ServerImpl::checkPlayers
(
void
)
{
        // threadsafe!

        // TODO: this should go away!  Game Logic object should handle this
        // in its tick() call.

        // loop through all players
        PlayerManager::iterator_t i;
        m_playerMgr->getIterator(0, i);
        player_rec_t pr;
        while (m_playerMgr->getNextPlayer(i, pr)) {
                if (pr.isAuthenticated() &&
                    !pr.isInMap()) {
                        try {
                                // game logic object can handle this
                                if (m_gameLogic->requestPlayerStartTS(pr)) {
                                        m_playerMgr->updatePlayer(pr);
                                }
                        } catch (std::exception& e) {
                                DPRINTF("Exception: %s", e.what());
                        }
                }
        }
}



void
ServerImpl::sendBroadcast
(
void
)
{
        ASSERT2(m_udpBcast, "No UDP broadcast point?");
        ASSERT2(m_xdrBcast, "null");
        ASSERT2(m_story, "null");

//      DPRINTF("Broadcasting!");

        m_xdrBcast->clear();

        // write public name ('server' for player)
        int len = m_publicName.length();
        m_xdrBcast->addStringPacklet('s', m_publicName.c_str(), len);

        // write story UUID
        const char * uuid = m_story->getUuid();
        ASSERT2(uuid, "null");
        len = strlen(uuid);
        m_xdrBcast->addStringPacklet('u', uuid, len);

        // write port
        int32_t port = m_udpPort;
        m_xdrBcast->addInt32Packlet('p', &port, 1);

        // construct netlib message
        // TODO: avoid the copy here!
        smart_ptr<MessageBuffer> msg = MessageBuffer::create();
        ASSERT(msg, "null");
        const byte_t * data = m_xdrBcast->getData();
        int bytes = m_xdrBcast->getDataBytes();
        ASSERT(bytes > 0, "Bad bytes: %d", bytes);
        msg->reserve(bytes);
        msg->appendData((const char *) data, bytes);
        msg->close();

        // send
        netlib::enqueueMessage(m_udpBcast, msg);
        
        // that's it!
}



//
//      public API
//

smart_ptr<Server>
Server::create
(
IN smart_ptr<Datahash>& params,
IN smart_ptr<ServerGameLogic>& gameLogic
)
{
        ASSERT(params, "null");
        ASSERT(gameLogic, "null");

        smart_ptr<ServerImpl> local = new ServerImpl;
        ASSERT(local, "out of memory");

        // initialize the server
        local->initialize(params, gameLogic);

        return local;
}


};      // aesop namespace