Add sync Buffer according to mantis #550

The game server buffers all changes made to a character in a sync buffer.
The buffer is sent to the account server if the buffer contains more then
20 message, reaches size of 1kb or at least every 10 seconds.
ATM Character attributes, corr points and attribute points and skills are
synchronized. TODO: items, location, money...

11 files changed, 316 insertions, 54 deletions

diff --git a/src/account-server/dalstorage.cpp b/src/account-server/dalstorage.cpp
index 06a1c83a..20815a7d 100644
--- a/src/account-server/dalstorage.cpp
+++ b/src/account-server/dalstorage.cpp
@@ -67,11 +67,6 @@ DALStorage::~DALStorage()
 /**
  * Connect to the database and initialize it if necessary.
  *
- * TODO: <b>Exceptionfault:</b> after connecting to the database, we have to
- *       verify if the version matches a supported version. Maybe implement a
- *       "version table" to check after connect. Raise an error with verbose
- *       informations about the discrepancy between the versions.
- *
  */
 void DALStorage::open()
 {
@@ -538,7 +533,8 @@ bool DALStorage::updateCharacter(Character *character,
     {
         for (unsigned int skill_id = 0; skill_id < CHAR_SKILL_NB; skill_id++)
         {
-            flushSkill(character, skill_id);
+            updateExperience(character->getDatabaseID(), skill_id,
+                character->getExperience(skill_id));
         }
     }
     catch (const dal::DbSqlQueryExecFailure& e)
@@ -635,57 +631,15 @@ bool DALStorage::updateCharacter(Character *character,
 
 /**
  * Save changes of a skill to the database permanently.
+ * @deprecated Use DALStorage::updateExperience instead!!!
 */
 void DALStorage::flushSkill(const Character* const character,
                             const int skill_id )
 {
-    try
-    {
-        const unsigned int exp = character->getExperience(skill_id);
-
-        // if experience has decreased to 0 we don't store is anymore,
-        // its the default
-        if (exp == 0)
-        {
-            std::ostringstream sql;
-            sql << "DELETE FROM " << CHAR_SKILLS_TBL_NAME << " "
-                << "WHERE char_id = '" << character->getDatabaseID() << "' "
-                << "AND skill_id = '" << skill_id << "'";
-            mDb->execSql(sql.str());
-            return;
-        }
-
-        // try to update the skill
-        std::ostringstream sql;
-        sql << "UPDATE " << CHAR_SKILLS_TBL_NAME << " "
-            << "SET skill_exp = '" << exp << "' "
-            << "WHERE char_id = '" << character->getDatabaseID() << "' "
-            << "AND skill_id = '" << skill_id << "'";
-        mDb->execSql(sql.str());
-
-        // check if the update has modified a row
-        if (mDb->getModifiedRows() > 0)
-        {
-            return;
-        }
-
-        sql.clear();
-        sql.str("");
-        sql << "INSERT INTO " << CHAR_SKILLS_TBL_NAME << " "
-            << "(char_id, skill_id, skill_exp) VALUES ( "
-            << "'" << character->getDatabaseID() << "', "
-            << "'" << skill_id << "', "
-            << "'" << exp << "' )";
-        mDb->execSql(sql.str());
-    }
-    catch (const dal::DbSqlQueryExecFailure &e)
-    {
-        LOG_ERROR("DALStorage::flushSkill: " << e.what());
-        throw;
-    }
+    updateExperience(character->getDatabaseID(), skill_id,
+        character->getExperience(skill_id));
 }
 
-
 /**
  * Add an account to the database.
  */
@@ -804,7 +758,8 @@ void DALStorage::flush(Account *account)
                 // update the characters skills
                 for (unsigned int skill_id = 0; skill_id < CHAR_SKILL_NB; skill_id++)
                 {
-                    flushSkill((*it), skill_id);
+                    updateExperience((*it)->getDatabaseID(), skill_id,
+                        (*it)->getExperience(skill_id));
                 }
             }
         } //
@@ -886,6 +841,78 @@ void DALStorage::updateLastLogin(const Account *account)
     mDb->execSql(sql.str());
 }
 
+void DALStorage::updateCharacterPoints(const int CharId, const int CharPoints,
+    const int CorrPoints, const int AttribId, const int AttribValue )
+{
+    std::ostringstream sql;
+    sql << "UPDATE " << CHARACTERS_TBL_NAME
+        << " SET char_pts = " << CharPoints << ", "
+        << " correct_pts = " << CorrPoints << ", ";
+
+    switch (AttribId)
+    {
+        case CHAR_ATTR_STRENGTH:     sql << "str = "; break;
+        case CHAR_ATTR_AGILITY:      sql << "agi = "; break;
+        case CHAR_ATTR_DEXTERITY:    sql << "dex = "; break;
+        case CHAR_ATTR_VITALITY:     sql << "vit = "; break;
+        case CHAR_ATTR_INTELLIGENCE: sql << "int = "; break;
+        case CHAR_ATTR_WILLPOWER:    sql << "will = "; break;
+    }
+    sql << AttribValue
+        << " WHERE id = " << CharId;
+
+    mDb->execSql(sql.str());
+}
+
+void DALStorage::updateExperience(const int CharId, const int SkillId,
+    const int SkillValue)
+{
+    try
+    {
+        // if experience has decreased to 0 we don't store it anymore,
+        // its the default
+        if (SkillValue == 0)
+        {
+            std::ostringstream sql;
+            sql << "DELETE FROM " << CHAR_SKILLS_TBL_NAME
+                << " WHERE char_id = " << CharId
+                << " AND skill_id = " << SkillId;
+            mDb->execSql(sql.str());
+            return;
+        }
+
+        // try to update the skill
+        std::ostringstream sql;
+        sql << "UPDATE " << CHAR_SKILLS_TBL_NAME
+            << " SET skill_exp = " << SkillValue
+            << " WHERE char_id = " << CharId
+            << " AND skill_id = " << SkillId;
+        mDb->execSql(sql.str());
+
+        // check if the update has modified a row
+        if (mDb->getModifiedRows() > 0)
+        {
+            return;
+        }
+
+        sql.clear();
+        sql.str("");
+        sql << "INSERT INTO " << CHAR_SKILLS_TBL_NAME << " "
+            << "(char_id, skill_id, skill_exp) VALUES ( "
+            << CharId << ", "
+            << SkillId << ", "
+            << SkillValue << ")";
+        mDb->execSql(sql.str());
+    }
+    catch (const dal::DbSqlQueryExecFailure &e)
+    {
+        LOG_ERROR("DALStorage::updateExperience: " << e.what());
+        throw;
+    }
+}
+
+
+
 /**
  * Add a guild
  */
diff --git a/src/account-server/dalstorage.hpp b/src/account-server/dalstorage.hpp
index e0cfead5..896151d1 100644
--- a/src/account-server/dalstorage.hpp
+++ b/src/account-server/dalstorage.hpp
@@ -123,6 +123,27 @@ class DALStorage
         void updateLastLogin(const Account *account);
 
         /**
+         * Write a modification message about Character points to the database.
+         *
+         * @param CharId      ID of the character
+         * @param CharPoints  Number of character points left for the character
+         * @param CorrPoints  Number of correction points left for the character
+         * @param AttribId    ID of the modified attribute
+         * @param AttribValue New value of the modified attribute
+         */
+        void updateCharacterPoints(const int CharId, const int CharPoints,
+            const int CorrPoints, const int AttribId, const int AttribValue );
+
+        /**
+         * Write a modification message about character skills to the database.
+         * @param CharId      ID of the character
+         * @param SkillId     ID of the skill
+         * @param SkillValue  new skill points
+         */
+        void updateExperience(const int CharId, const int SkillId,
+            const int SkillValue);
+
+        /**
          * Sets a ban on an account (hence on all its characters).
          *
          * @param id character identifier.
diff --git a/src/account-server/serverhandler.cpp b/src/account-server/serverhandler.cpp
index bdbdb1f6..1c4a5219 100644
--- a/src/account-server/serverhandler.cpp
+++ b/src/account-server/serverhandler.cpp
@@ -31,7 +31,6 @@
 #include "account-server/dalstorage.hpp"
 #include "chat-server/post.hpp"
 #include "net/connectionhandler.hpp"
-#include "net/messagein.hpp"
 #include "net/messageout.hpp"
 #include "net/netcomputer.hpp"
 #include "serialize/characterdata.hpp"
@@ -241,6 +240,12 @@ void ServerHandler::processMessage(NetComputer *comp, MessageIn &msg)
             }
         } break;
 
+        case GAMSG_PLAYER_SYNC:
+        {
+            LOG_DEBUG("GAMSG_PLAYER_SYNC");
+            GameServerHandler::syncDatabase(msg);
+        } break;
+
         case GAMSG_REDIRECT:
         {
             LOG_DEBUG("GAMSG_REDIRECT");
@@ -511,3 +516,37 @@ void GameServerHandler::sendPartyChange(Character *ptr, int partyId)
         s->send(msg);
     }
 }
+
+void GameServerHandler::syncDatabase(MessageIn &msg)
+{
+    int msgType = msg.readByte();
+    while( msgType != SYNC_END_OF_BUFFER )
+    {
+        switch (msgType)
+        {
+            case SYNC_CHARACTER_POINTS:
+            {
+                LOG_DEBUG("received SYNC_CHARACTER_POINTS");
+                int CharId = msg.readLong();
+                int CharPoints = msg.readLong();
+                int CorrPoints = msg.readLong();
+                int AttribId = msg.readByte();
+                int AttribValue = msg.readLong();
+                storage->updateCharacterPoints(CharId, CharPoints, CorrPoints,
+                                               AttribId, AttribValue);
+            } break;
+
+            case SYNC_CHARACTER_SKILL:
+            {
+                LOG_DEBUG("received SYNC_CHARACTER_SKILL");
+                int CharId = msg.readLong();
+                int SkillId = msg.readByte();
+                int SkillValue = msg.readLong();
+                storage->updateExperience(CharId, SkillId, SkillValue);
+            } break;
+        }
+
+        // read next message type from buffer
+        msgType = msg.readByte();
+    }
+}
diff --git a/src/account-server/serverhandler.hpp b/src/account-server/serverhandler.hpp
index 0d75219c..646ebf4c 100644
--- a/src/account-server/serverhandler.hpp
+++ b/src/account-server/serverhandler.hpp
@@ -25,6 +25,8 @@
 #include <iosfwd>
 #include <string>
 
+#include "net/messagein.hpp"
+
 class Character;
 
 namespace GameServerHandler
@@ -64,6 +66,12 @@ namespace GameServerHandler
      * Sends chat party information
      */
     void sendPartyChange(Character *ptr, int partyId);
+
+    /**
+     * Takes a GAMSG_PLAYER_SYNC from the gameserver and stores all changes in
+     * the database.
+     */
+    void syncDatabase(MessageIn &msg);
 }
 
 #endif
diff --git a/src/defines.h b/src/defines.h
index 7ee92bdd..a0499377 100644
--- a/src/defines.h
+++ b/src/defines.h
@@ -96,6 +96,7 @@ enum
  * - CPMSG_*: from chat server to client
  * - PGMSG_*: from client to game server
  * - GPMSG_*: from game server to client
+ * - GAMSG_*: from game server to account server
  *
  * Components: B byte, W word, L long, S variable-size string
  *             C tile-based coordinates (B*3)
@@ -269,6 +270,7 @@ enum {
     GAMSG_REDIRECT              = 0x0530, // L id
     AGMSG_REDIRECT_RESPONSE     = 0x0531, // L id, B*32 token, S game address, W game port
     GAMSG_PLAYER_RECONNECT      = 0x0532, // L id, B*32 token
+    GAMSG_PLAYER_SYNC           = 0x0533, // serialised sync data
     GAMSG_SET_QUEST             = 0x0540, // L id, S name, S value
     GAMSG_GET_QUEST             = 0x0541, // L id, S name
     AGMSG_GET_QUEST_RESPONSE    = 0x0542, // L id, S name, S value
@@ -307,6 +309,13 @@ enum {
     DATA_VERSION_OUTDATED = 0x01
 };
 
+// used to identify part of sync message
+enum {
+    SYNC_CHARACTER_POINTS = 0x01,       // L charId, L charPoints, L corrPoints, B attribute id, L attribute value
+    SYNC_CHARACTER_SKILL  = 0x02,       // L charId, B skillId, L skill value
+    SYNC_END_OF_BUFFER    = 0xFF        // shows, that the buffer ends here.
+};
+
 // Login specific return values
 enum {
     LOGIN_INVALID_VERSION = 0x40,       // the user is using an incompatible protocol
diff --git a/src/game-server/accountconnection.cpp b/src/game-server/accountconnection.cpp
index 43c4f20d..bbdcdd13 100644
--- a/src/game-server/accountconnection.cpp
+++ b/src/game-server/accountconnection.cpp
@@ -33,12 +33,19 @@
 #include "game-server/quest.hpp"
 #include "game-server/state.hpp"
 #include "net/messagein.hpp"
-#include "net/messageout.hpp"
 #include "serialize/characterdata.hpp"
 #include "utils/logger.h"
 #include "utils/tokendispenser.hpp"
 #include "utils/tokencollector.hpp"
 
+AccountConnection::~AccountConnection()
+{
+    if (mSyncBuffer)
+    {
+        delete (mSyncBuffer);
+    }
+}
+
 bool AccountConnection::start()
 {
     const std::string accountServerAddress =
@@ -72,6 +79,10 @@ bool AccountConnection::start()
     }
     send(msg);
 
+    // initialize sync buffer
+    mSyncBuffer = new MessageOut(GAMSG_PLAYER_SYNC);
+    mSyncMessages = 0;
+
     return true;
 }
 
@@ -297,3 +308,57 @@ void AccountConnection::changeAccountLevel(Character *c, int level)
     msg.writeShort(level);
     send(msg);
 }
+
+void AccountConnection::syncChanges(bool force)
+{
+    if (mSyncMessages == 0)
+        return;
+
+    // send buffer if:
+    //    a.) forced by any process
+    //    b.) every 10 seconds
+    //    c.) buffer reaches size of 1kb
+    //    d.) buffer holds more then 20 messages
+    if (force ||
+        mSyncMessages > SYNC_BUFFER_LIMIT ||
+        mSyncBuffer->getLength() > SYNC_BUFFER_SIZE )
+    {
+        LOG_DEBUG("Sending GAMSG_PLAYER_SYNC with " << mSyncMessages << " messages." );
+
+        // attach end-of-buffer flag
+        mSyncBuffer->writeByte(SYNC_END_OF_BUFFER);
+        send(*mSyncBuffer);
+        delete (mSyncBuffer);
+
+        mSyncBuffer = new MessageOut(GAMSG_PLAYER_SYNC);
+        mSyncMessages = 0;
+    }
+    else
+    {
+        LOG_DEBUG("No changes to sync with account server.");
+    }
+}
+
+void AccountConnection::updateCharacterPoints(const int CharId, const int CharPoints,
+    const int CorrPoints, const int AttribId, const int AttribValue )
+{
+    mSyncMessages++;
+    mSyncBuffer->writeByte(SYNC_CHARACTER_POINTS);
+    mSyncBuffer->writeLong(CharId);
+    mSyncBuffer->writeLong(CharPoints);
+    mSyncBuffer->writeLong(CorrPoints);
+    mSyncBuffer->writeByte(AttribId);
+    mSyncBuffer->writeLong(AttribValue);
+    syncChanges();
+}
+
+void AccountConnection::updateExperience(const int CharId, const int SkillId,
+    const int SkillValue)
+{
+    mSyncMessages++;
+    mSyncBuffer->writeByte(SYNC_CHARACTER_SKILL);
+    mSyncBuffer->writeLong(CharId);
+    mSyncBuffer->writeByte(SkillId);
+    mSyncBuffer->writeLong(SkillValue);
+    syncChanges();
+}
diff --git a/src/game-server/accountconnection.hpp b/src/game-server/accountconnection.hpp
index 2c63e17c..22c422a7 100644
--- a/src/game-server/accountconnection.hpp
+++ b/src/game-server/accountconnection.hpp
@@ -22,16 +22,39 @@
 #ifndef _TMW_ACCOUNTCONNECTION_H_
 #define _TMW_ACCOUNTCONNECTION_H_
 
+#include "net/messageout.hpp"
 #include "net/connection.hpp"
 
 class Character;
 
+/** \fn void AccountConnection::syncChanges(bool force = false)
+ *
+ *  The gameserver holds a buffer with all changes made by a character. The
+ *  changes are added at the time they occur. When the buffer reaches one of
+ *  the following limits, the buffer is sent to the account server and applied
+ *  to the database.
+ *
+ *  The sync buffer is sent when:
+ *  - forced by any process (param force = true)
+ *  - every 10 seconds
+ *  - buffer reaches size of 1kb (defined in #SYNC_BUFFER_SIZE)
+ *  - buffer holds more then 20 messages (defined in #SYNC_BUFFER_LIMIT)
+ */
+#define SYNC_BUFFER_SIZE    1024   /**< maximum size of sync buffer in bytes. */
+#define SYNC_BUFFER_LIMIT     20   /**< maximum number of messages in sync buffer. */
+
 /**
  * A connection to the account server.
  */
 class AccountConnection : public Connection
 {
     public:
+
+        /**
+        * Destructor
+        */
+        ~AccountConnection();
+
         /**
          * Initializes a connection to the account server described in the
          * configuration file. Registers the maps known by MapManager.
@@ -84,12 +107,49 @@ class AccountConnection : public Connection
          */
         void changeAccountLevel(Character *, int);
 
+        /**
+         * Sends all changed player data to the account server to minimize
+         * dataloss due to failure of one server component.
+         *
+         * @param force Send changes even if buffer hasn't reached its size
+         *              or message limit. (used to send in timed schedules)
+         */
+        void syncChanges(bool force = false);
+
+        /**
+         * Write a modification message about character points to the sync buffer.
+         *
+         * @param CharId      ID of the character
+         * @param CharPoints  Number of character points left for the character
+         * @param CorrPoints  Number of correction points left for the character
+         * @param AttribId    ID of the modified attribute
+         * @param AttribValue New value of the modified attribute
+         */
+        void updateCharacterPoints(const int CharId, const int CharPoints,
+                                   const int CorrPoints, const int AttribId,
+                                   const int AttribValue);
+
+
+        /**
+         * Write a modification message about character skills to the sync buffer.
+         * @param CharId      ID of the character
+         * @param SkillId     ID of the skill
+         * @param SkillValue  new skill points
+         */
+        void updateExperience(const int CharId, const int SkillId,
+                              const int SkillValue);
+
     protected:
         /**
          * Processes server messages.
          */
         virtual void processMessage(MessageIn &);
 
+    private:
+
+        MessageOut* mSyncBuffer;     /**< Message buffer to store sync data. */
+        int mSyncMessages;           /**< Number of messages in the sync buffer. */
+
 };
 
 extern AccountConnection *accountHandler;
diff --git a/src/game-server/character.cpp b/src/game-server/character.cpp
index 8ae2e85d..780a142b 100644
--- a/src/game-server/character.cpp
+++ b/src/game-server/character.cpp
@@ -26,6 +26,7 @@
 #include "game-server/character.hpp"
 
 #include "defines.h"
+#include "game-server/accountconnection.hpp"
 #include "game-server/attackzone.hpp"
 #include "game-server/buysell.hpp"
 #include "game-server/eventlistener.hpp"
@@ -341,6 +342,10 @@ void Character::receiveExperience(size_t skill, int experience)
         mExperience.at(skill - CHAR_SKILL_BEGIN) = newExp;
         mModifiedExperience.insert(skill - CHAR_SKILL_BEGIN);
 
+        // inform account server
+        accountHandler->updateExperience(getDatabaseID(),
+            skill - CHAR_SKILL_BEGIN, newExp);
+
         // check for skill levelup
         while (newExp >= Character::expForLevel(getAttribute(skill) + 1))
         {
diff --git a/src/game-server/gamehandler.cpp b/src/game-server/gamehandler.cpp
index 7886ac18..675a0fc0 100644
--- a/src/game-server/gamehandler.cpp
+++ b/src/game-server/gamehandler.cpp
@@ -443,6 +443,16 @@ void GameHandler::processMessage(NetComputer *comp, MessageIn &message)
             result.writeShort(GPMSG_RAISE_ATTRIBUTE_RESPONSE);
             result.writeByte(retCode);
             result.writeByte(attribute);
+
+            if (retCode == ATTRIBMOD_OK )
+            {
+                accountHandler->updateCharacterPoints(
+                    computer.character->getDatabaseID(),
+                    computer.character->getCharacterPoints(),
+                    computer.character->getCorrectionPoints(),
+                    attribute,
+                    computer.character->getAttribute(attribute));
+            }
         } break;
 
         case PGMSG_LOWER_ATTRIBUTE:
@@ -453,6 +463,16 @@ void GameHandler::processMessage(NetComputer *comp, MessageIn &message)
             result.writeShort(GPMSG_LOWER_ATTRIBUTE_RESPONSE);
             result.writeByte(retCode);
             result.writeByte(attribute);
+
+            if (retCode == ATTRIBMOD_OK )
+            {
+                accountHandler->updateCharacterPoints(
+                    computer.character->getDatabaseID(),
+                    computer.character->getCharacterPoints(),
+                    computer.character->getCorrectionPoints(),
+                    attribute,
+                    computer.character->getAttribute(attribute));
+            }
         } break;
 
         case PGMSG_RESPAWN:
diff --git a/src/game-server/main-game.cpp b/src/game-server/main-game.cpp
index a0723e31..93c2f1fa 100644
--- a/src/game-server/main-game.cpp
+++ b/src/game-server/main-game.cpp
@@ -310,6 +310,8 @@ int main(int argc, char *argv[])
             // Print world time at 10 second intervals to show we're alive
             if (worldTime % 100 == 0) {
                 LOG_INFO("World time: " << worldTime);
+                // force sending changes to the account serber every 10 secs.
+                accountHandler->syncChanges(true);
             }
 
             if (accountHandler->isConnected())
diff --git a/src/game-server/state.cpp b/src/game-server/state.cpp
index 61829a57..3b6dd624 100644
--- a/src/game-server/state.cpp
+++ b/src/game-server/state.cpp
@@ -450,10 +450,16 @@ void GameState::update(int worldTime)
         for (CharacterIterator p(map->getWholeMapIterator()); p; ++p)
         {
             informPlayer(map, *p);
+            /*
+             sending the whole character is overhead for the database, it should
+             be replaced by a syncbuffer. see: game-server/accountconnection:
+             AccountConnection::syncChanges()
+
             if (worldTime % 2000 == 0)
             {
                 accountHandler->sendCharacterData(*p);
             }
+            */
         }
 
         for (ObjectIterator i(map->getWholeMapIterator()); i; ++i)