Event Driven Database

If you need a high performance database in your game engine, but you don't want database access slowing down game-play then this resource is for you.

Disclaimer

Please note that this is an Advanced Resource and if you're not a strong C++ programmer that has experience with multi-threading then proceed with caution (and consider getting outside help).

Notes:

I used TGE 1.5 as a starting point and I'm not sure when the multi-threading stuff was added to Torque. If it wasn't before TGE 1.5 then you're going to need to upgrade (which I highly recommend anyway :P)

Although I use MySQL++ here, you could use any database.

In this example I have references to a login server, which is not part of this resource, so this resource will not compile correctly without some adjustment, but you'll have to make modifications anyway to add new queries.

Edit: If you follow along this entire thread then you'll see how to get rid of the login server stuff since you probably don't need it and to add a generic SQL handler to get the record sets back to your Torque Script code.

I have #ifdef DEDICATED / #endif around the code because I only include MySQL code in my dedicated server. If you're using SQLite or some other embedded database on the client, you'll want to not do this. If you're using both MySQL and SQLite then you'll have a bit of refactoring to do. In my game I'll probably use both and quite likely I'll use two implementations like MySQLDatabaseInterface* gServerDatabase and SQLiteDatabaseInterface* gClientCache.

This resource takes a bit too much CPU time due to the fact that Torque doesn't have a built-in condition variable (yet). I'll add that as a separate resource eventually. If you use that resource, simply follow the source-code comments and uncomment every line that references m_pNotEmptyEvent. You could possibly use a semaphore too, but that's not as clean.

I did my best to match the rest of Torque's coding style, other than using m_ instead of m... Sorry, some old habits die hard. I didn't make a private copy constructor like I should have, but just remember that this is coded as a singleton and you shouldn't be creating new one's and/or copying the thing around without knowing what you're doing, ok? :P

Feel free to use this resource for personal and/or commercial use, but please give me credit (Tony "SgtFlame" Richards) where possible.

I placed my code in "core", but in retrospect, it might have been better to place it in "platform". I'm not making that change right now for other reasons, but my suggestion to you is you add the new files to the "engine/core" directory.

databaseEvent.h

Add this new file.

#ifndef _DATABASE_EVENT_H_
#define _DATABASE_EVENT_H_

// This is only used on dedicated servers
#ifdef DEDICATED

#include "platform/event.h"

enum DatabaseEventType
{
   LoginQuery
};

struct DatabaseRequest
: public Event
{
   DatabaseEventType type2;

   DatabaseRequest() { type = DatabaseRequestEventType; }
};

struct DatabaseReply
: public Event
{
   DatabaseEventType type2;

   DatabaseReply() { type = DatabaseReplyEventType; }
};

#define  MAX_USER_ID_SIZE     128
#define  MAX_PASSWORD_SIZE    128

struct LoginRequest
: public DatabaseRequest
{
   LoginRequest() { size = sizeof(LoginRequest); type2 = LoginQuery; }

   char        szUserId[MAX_USER_ID_SIZE];
   char        szPassword[MAX_PASSWORD_SIZE];
   NetSocket   connection;
};

struct LoginReply
: public DatabaseReply
{
   LoginReply() { size = sizeof(LoginReply); type2 = LoginQuery; }

   bool        passwordMatched;
   NetSocket   connection;
};

#endif // DEDICATED

#endif // _DATABASE_EVENT_H_

databaseInterface.h

//-----------------------------------------------------------------------------
// Fractured Universe
// Copyright (C) Tony Richards
//-----------------------------------------------------------------------------

#ifndef _DATABASE_INTERFACE_H_
#define _DATABASE_INTERFACE_H_

// This is only used on dedicated servers
#ifdef DEDICATED

#include "core/tVector.h"

#include "platform/event.h"
#include "core/databaseEvent.h"

#include "platform/platformThread.h"

#include <mysql++.h>

class DatabaseInterface
{
   //-~-~-~-~-~-~
   // Structors
public:
            DatabaseInterface();
   virtual ~DatabaseInterface();

   //-~-~-~-~-~-~-~-~-~
   // Implementation
public:

   /// Place an event in Game's event queue.
   virtual void postEvent(DatabaseRequest& event);

   /// Static thread startup method.
   /// @note Don't call this; the thread lifetime
   ///          is automatically handled.
   static void processThread(void *udata);

   //-~-~-~-~-~-~-~-~-~-~-~-~-~-~
   // Additional implementation
protected:
   /// Main process loop
   void processLoop();

   /// Process a single event
   void processEvent(DatabaseRequest* event);

   //-~-~-~-~-~-~-~-~-~
   // Member Variables
private:

   /// Events are stored here by any thread, for processing by the main thread.
   Vector<DatabaseRequest*> m_EventQueue1, m_EventQueue2, *m_pEventQueue;

   volatile bool   m_ShuttingDown;
   void *          m_pEventQueueMutex;
   //void *          m_pNotEmptyEvent;
   Thread*         m_ProcessThread;

	mysqlpp::Connection m_con;

};

/// Global database instance.
extern DatabaseInterface *gDatabase;

#endif   // DEDICATED

#endif // _DATABASE_INTERFACE_H_

databaseInterface.cc

//-----------------------------------------------------------------------------
// Fractured Universe
// Copyright (C) Tony Richards
//-----------------------------------------------------------------------------

// This module is only used on Dedicated Servers
#ifdef DEDICATED

#include "core/databaseInterface.h"

#include "platform/platform.h"
#include "platform/platformMutex.h"
#include "platform/gameInterface.h"

#include "console/console.h"

#include "game/mmo/login/loginServer.h"

DatabaseInterface *gDatabase = NULL;

DatabaseInterface::DatabaseInterface()
:  m_pEventQueue(&m_EventQueue1)
,  m_pEventQueueMutex(Mutex::createMutex())
//,   m_pNotEmptyEvent(Condition::createCondition())
,  m_ShuttingDown(false)
,  m_con(false)
{
   m_ProcessThread = new Thread((ThreadRunFunction)processThread, (S32) this, 1);
}

DatabaseInterface::~DatabaseInterface()
{
   // TODO: Shutdown gracefully.  For now everything just quieces correctly
   // anyway, but it'd be nice to do this correctly.
}

void DatabaseInterface::postEvent(DatabaseRequest& event)
{
   // Only one thread can post at a time.
   Mutex::lockMutex(m_pEventQueueMutex);

   // Create a deep copy of event, and save a pointer to the copy in a vector.
   DatabaseRequest* copy = (DatabaseRequest*)dMalloc(event.size);
   dMemcpy(copy, &event, event.size);
   m_pEventQueue->push_back(copy);

   // TODO: Assert the condition variable
   // Condition::assert(m_pNotEmptyEvent);

   Mutex::unlockMutex(m_pEventQueueMutex);

}

void DatabaseInterface::processEvent(DatabaseRequest* event)
{
   switch(event->type2)
   {
      case LoginQuery:
      {
         LoginRequest* pRequest = (LoginRequest*)event;

         // Create default reply
         LoginReply reply;
         reply.passwordMatched = false;
         reply.connection = pRequest->connection;

      	mysqlpp::Query query = m_con.query();

	      query << "select * from Player where UserName = " 
            "\'" << pRequest->szUserId << "\'";

	      mysqlpp::ResUse res = query.use();

         if (res)
         {
            try
            {
		         mysqlpp::Row row;
	            while (row = res.fetch_row()) 
               {
                  if (dStricmp(row["EncryptedPassword"].c_str(), pRequest->szPassword) == 0)
                  {
                     // Password matched
                     reply.passwordMatched = true;

                     Con::printf("%s matched", row["UserName"].c_str());
                     break;
                  }
                  else
                  {
                     Con::printf("%s didn't match", row["UserName"].c_str());
                  }
	            }
            }
            catch(...)
            {
            }
         }

         Game->postEvent(reply);
      }
      break;
   }
}

void DatabaseInterface::processLoop()
{
   if (!m_con.connect(
      Con::getVariable("Database::database"), 
      Con::getVariable("Database::host"), 
      Con::getVariable("Database::user"), 
      Con::getVariable("Database::password")))
   {
      Con::printf("Error connecting to database!");

      // TODO: Loop a few times before giving up?
      return;
   }

   while(!m_ShuttingDown)
   {
      Mutex::lockMutex(m_pEventQueueMutex);

      // swap event queue pointers
      Vector<DatabaseRequest*> &fullEventQueue = *m_pEventQueue;
      if(m_pEventQueue == &m_EventQueue1)
         m_pEventQueue = &m_EventQueue2;
      else
         m_pEventQueue = &m_EventQueue1;

      // TODO: signal that the queue is now empty
      // Condition::retract(m_pNotEmptyEvent);

      Mutex::unlockMutex(m_pEventQueueMutex);

      // Keep track of the original size
      const int size = fullEventQueue.size();

      // Walk the event queue in fifo order, processing the events, then clear the queue.
      for(int i=0; i < size; i++)
      {
         processEvent(fullEventQueue[i]);
         dFree(fullEventQueue[i]);
      }
      fullEventQueue.clear();

      if (size == 0)
      {
         // HACK: This is a bad hack.  Instead of doing this polling, use a true thread-safe queue.
         // TODO: Check to see if the next queue is empty, and if it is, wait for a condition variable
         // and don't do the Platform::sleep(0);
         // Condition::require(m_pNotEmptyEvent);
         Platform::sleep(0);
      }
   }
}

void DatabaseInterface::processThread( void *udata )
{
   DatabaseInterface* pThis = (DatabaseInterface *)udata;
   pThis->processLoop();
}

void GameInterface::processDatabaseReplyEventType(DatabaseReply* event)
{
   switch(event->type2)
   {
   case LoginQuery:
      // Note that this won't compile.
      g_pLoginServer->processLoginReply((LoginReply*)event);
      break;
   default:
      Con::printf("Got an illegal database reply type: %i", event->type2);
      break;
   }
}

#endif

platform/event.h

Modify the enum EventTypes:

/// Available event types.
enum EventTypes
{
   InputEventType,
   MouseMoveEventType,
   PacketReceiveEventType,
   TimeEventType,
   QuitEventType,
   ConsoleEventType,
   ConnectedReceiveEventType,
   ConnectedAcceptEventType,
   ConnectedNotifyEventType,
   DatabaseRequestEventType,
   DatabaseReplyEventType
};

platform/gameInterface.h

Add this somewhere at the top where the other #includes are:

#ifdef DEDICATED
#include "core/databaseEvent.h"
#endif

In the Event Handlers section, add this:

virtual void processConnectedNotifyEvent(ConnectedNotifyEvent *event);

#ifdef DEDICATED
   virtual void processDatabaseReplyEventType(DatabaseReply* event);
#endif
   /// @}

platform/gameInterface.cc

In GameInterface::processEvent add this:

void GameInterface::processEvent(Event *event)
{

...

      case ConnectedNotifyEventType:
         processConnectedNotifyEvent( (ConnectedNotifyEvent *) event );
         break;
#ifdef DEDICATED
      case DatabaseReplyEventType:
         processDatabaseReplyEventType( (DatabaseReply*) event );
         break;
#endif
   }

...

}

game/main.cc

Somewhere in initGame() you have to create gDatabase, and in shutdownGame() you have to delete it. Since the script needs to be executed first (because the database credentials are initialized in the script) you really need to do something like this in initGame().

At the top you'll need to include the database interface header. Remember, it's either core or platform, depending on the choice you made earlier.

#include "platform/databaseInterface.h"

Change this:

// run the entry script and return.
   return runEntryScript(argc, argv);
}

To this:

// run the entry script and return.
   bool rc = runEntryScript(argc, argv);

#ifdef DEDICATED
   // Create the database object
   gDatabase = new DatabaseInterface;
#endif

   return rc;
}

And in shutdownGame() add this somewhere at the top, but after calling the script onExit:

/// Shutdown the game and delete core objects
void shutdownGame()
{
   //exec the script onExit() function
   Con::executef(1, "onExit");

#ifdef DEDICATED
   delete gDatabase;
#endif

   BadWordFilter::destroy();
   ParticleEngine::destroy();

Adding more queries:

To add another query, add new events in databaseEvent.h like LoginRequest and LoginReply and make sure in the constructor of these events you correctly initialize type2 to a new DatabaseEventType (and don't forget to add the new enum to DatabaseEventType.)

Remember that these new events are copied using memcpy(), so be sure if you pass pointers around to evaluate the lifetime of the object pointers you're passing so as to not leave a reference to a deleted object hanging around. If you must, make sure processEvent correctly deletes any pointers to things in DatabaseRequest, and make sure your final DatabaseReply handler deletes anything in DatabaseReply.

In databaseInterface.cc in processEvent, add a new case statement to handle the new type2. As you add more queries you're going to want to extract these cases to call other methods instead of making them all in-line, otherwise the processEvent function will become huge and unwieldy.

Also add a new case statement in processDatabaseReplyEventType to handle the new type2.

So there you have it. This mechanism will allow you to do your slower database calls in a separate thread while allowing your main game thread to continue processing. Everything in GameInterface::processDatabaseReplyEventType is executed in the main game thread, and everything in DatabaseInterface::processEvent is executed in the database thread.

Let me know if I left something out. Happy Database Torquing!