mirror of
https://github.com/mangosfour/server.git
synced 2025-12-15 19:37:02 +00:00
[11045] Rewrite internals of DB layer. Simplify code and use less locking. Spawn and use separate connections for sync and async DB requests. Implement database connection pool for SELECT queries. Up to maximum 16 connections supported. Disable 'autocommit' mode for MySQL.
UPDATE YOUR CONFIGS! Defaults: LoginDatabaseConnections = 1 WorldDatabaseConnections = 1 CharacterDatabaseConnections = 1 If you are not using <mtmaps> patch do not change the default settings - this is useless. You can try following option in your MySQL config to squeeze even more performance from your DB: [mysqld] transaction-isolation = READ-COMMITTED Great thanks to Undergarun, kero99 and selector for making tests and providing very useful feedback and DB statistics! Have fun :) Signed-off-by: Ambal <pogrebniak@gala.net>
This commit is contained in:
Ambal
parent
9bc37afa28
commit
631ce36680
19 changed files with 655 additions and 547 deletions
|
|
@ -23,18 +23,16 @@
|
|||
#include "Platform/Define.h"
|
||||
#include "Threading.h"
|
||||
#include "DatabaseEnv.h"
|
||||
#include "Database/PGSQLDelayThread.h"
|
||||
#include "Database/SqlOperations.h"
|
||||
#include "Timer.h"
|
||||
|
||||
size_t DatabasePostgre::db_count = 0;
|
||||
|
||||
DatabasePostgre::DatabasePostgre() : Database(), mPGconn(NULL)
|
||||
DatabasePostgre::DatabasePostgre()
|
||||
{
|
||||
// before first connection
|
||||
if( db_count++ == 0 )
|
||||
{
|
||||
|
||||
if (!PQisthreadsafe())
|
||||
{
|
||||
sLog.outError("FATAL ERROR: PostgreSQL libpq isn't thread-safe.");
|
||||
|
|
@ -45,26 +43,32 @@ DatabasePostgre::DatabasePostgre() : Database(), mPGconn(NULL)
|
|||
|
||||
DatabasePostgre::~DatabasePostgre()
|
||||
{
|
||||
|
||||
if (m_delayThread)
|
||||
HaltDelayThread();
|
||||
|
||||
if( mPGconn )
|
||||
//destroy SqlConnection objects
|
||||
if(m_pQueryConnections.size())
|
||||
{
|
||||
PQfinish(mPGconn);
|
||||
mPGconn = NULL;
|
||||
for (int i = 0; i < m_pQueryConnections.size(); ++i)
|
||||
delete m_pQueryConnections[i];
|
||||
|
||||
m_pQueryConnections.clear();
|
||||
}
|
||||
|
||||
if(m_pAsyncConn)
|
||||
{
|
||||
delete m_pAsyncConn;
|
||||
m_pAsyncConn = NULL;
|
||||
}
|
||||
}
|
||||
|
||||
bool DatabasePostgre::Initialize(const char *infoString)
|
||||
SqlConnection * DatabasePostgre::CreateConnection()
|
||||
{
|
||||
if(!Database::Initialize(infoString))
|
||||
return false;
|
||||
|
||||
tranThread = NULL;
|
||||
|
||||
InitDelayThread();
|
||||
return new PostgreSQLConnection();
|
||||
}
|
||||
|
||||
bool PostgreSQLConnection::Initialize(const char *infoString)
|
||||
{
|
||||
Tokens tokens = StrSplit(infoString, ";");
|
||||
|
||||
Tokens::iterator iter;
|
||||
|
|
@ -98,26 +102,18 @@ bool DatabasePostgre::Initialize(const char *infoString)
|
|||
mPGconn = NULL;
|
||||
return false;
|
||||
}
|
||||
else
|
||||
{
|
||||
sLog.outDetail( "Connected to Postgre database at %s",
|
||||
host.c_str());
|
||||
sLog.outString( "PostgreSQL server ver: %d",PQserverVersion(mPGconn));
|
||||
return true;
|
||||
}
|
||||
|
||||
sLog.outDetail( "Connected to Postgre database at %s", host.c_str());
|
||||
sLog.outString( "PostgreSQL server ver: %d", PQserverVersion(mPGconn));
|
||||
return true;
|
||||
}
|
||||
|
||||
bool DatabasePostgre::_Query(const char *sql, PGresult** pResult, uint64* pRowCount, uint32* pFieldCount)
|
||||
bool PostgreSQLConnection::_Query(const char *sql, PGresult** pResult, uint64* pRowCount, uint32* pFieldCount)
|
||||
{
|
||||
if (!mPGconn)
|
||||
return 0;
|
||||
return false;
|
||||
|
||||
// guarded block for thread-safe request
|
||||
ACE_Guard<ACE_Thread_Mutex> query_connection_guard(mMutex);
|
||||
#ifdef MANGOS_DEBUG
|
||||
uint32 _s = getMSTime();
|
||||
#endif
|
||||
uint32 _s = WorldTimer::getMSTime();
|
||||
// Send the query
|
||||
*pResult = PQexec(mPGconn, sql);
|
||||
if(!*pResult )
|
||||
|
|
@ -132,9 +128,7 @@ bool DatabasePostgre::_Query(const char *sql, PGresult** pResult, uint64* pRowCo
|
|||
}
|
||||
else
|
||||
{
|
||||
#ifdef MANGOS_DEBUG
|
||||
sLog.outDebug("[%u ms] SQL: %s", getMSTime() - _s, sql );
|
||||
#endif
|
||||
DEBUG_FILTER_LOG(LOG_FILTER_SQL_TEXT, "[%u ms] SQL: %s", WorldTimer::getMSTimeDiff(_s,WorldTimer::getMSTime()), sql );
|
||||
}
|
||||
|
||||
*pRowCount = PQntuples(*pResult);
|
||||
|
|
@ -146,13 +140,14 @@ bool DatabasePostgre::_Query(const char *sql, PGresult** pResult, uint64* pRowCo
|
|||
PQclear(*pResult);
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
QueryResult* DatabasePostgre::Query(const char *sql)
|
||||
QueryResult* PostgreSQLConnection::Query(const char *sql)
|
||||
{
|
||||
if (!mPGconn)
|
||||
return 0;
|
||||
return NULL;
|
||||
|
||||
PGresult* result = NULL;
|
||||
uint64 rowCount = 0;
|
||||
|
|
@ -162,15 +157,15 @@ QueryResult* DatabasePostgre::Query(const char *sql)
|
|||
return NULL;
|
||||
|
||||
QueryResultPostgre * queryResult = new QueryResultPostgre(result, rowCount, fieldCount);
|
||||
queryResult->NextRow();
|
||||
|
||||
queryResult->NextRow();
|
||||
return queryResult;
|
||||
}
|
||||
|
||||
QueryNamedResult* DatabasePostgre::QueryNamed(const char *sql)
|
||||
QueryNamedResult* PostgreSQLConnection::QueryNamed(const char *sql)
|
||||
{
|
||||
if (!mPGconn)
|
||||
return 0;
|
||||
return NULL;
|
||||
|
||||
PGresult* result = NULL;
|
||||
uint64 rowCount = 0;
|
||||
|
|
@ -184,67 +179,35 @@ QueryNamedResult* DatabasePostgre::QueryNamed(const char *sql)
|
|||
names[i] = PQfname(result, i);
|
||||
|
||||
QueryResultPostgre * queryResult = new QueryResultPostgre(result, rowCount, fieldCount);
|
||||
queryResult->NextRow();
|
||||
|
||||
queryResult->NextRow();
|
||||
return new QueryNamedResult(queryResult,names);
|
||||
}
|
||||
|
||||
bool DatabasePostgre::Execute(const char *sql)
|
||||
bool PostgreSQLConnection::Execute(const char *sql)
|
||||
{
|
||||
|
||||
if (!mPGconn)
|
||||
return false;
|
||||
|
||||
// don't use queued execution if it has not been initialized
|
||||
if (!m_threadBody)
|
||||
return DirectExecute(sql);
|
||||
uint32 _s = WorldTimer::getMSTime();
|
||||
|
||||
tranThread = ACE_Based::Thread::current(); // owner of this transaction
|
||||
TransactionQueues::iterator i = m_tranQueues.find(tranThread);
|
||||
if (i != m_tranQueues.end() && i->second != NULL)
|
||||
{ // Statement for transaction
|
||||
i->second->DelayExecute(sql);
|
||||
PGresult *res = PQexec(mPGconn, sql);
|
||||
if (PQresultStatus(res) != PGRES_COMMAND_OK)
|
||||
{
|
||||
sLog.outErrorDb( "SQL: %s", sql );
|
||||
sLog.outErrorDb( "SQL %s", PQerrorMessage(mPGconn) );
|
||||
return false;
|
||||
}
|
||||
else
|
||||
{
|
||||
// Simple sql statement
|
||||
m_threadBody->Delay(new SqlStatement(sql));
|
||||
DEBUG_FILTER_LOG(LOG_FILTER_SQL_TEXT, "[%u ms] SQL: %s", WorldTimer::getMSTimeDiff(_s,WorldTimer::getMSTime()), sql );
|
||||
}
|
||||
|
||||
PQclear(res);
|
||||
return true;
|
||||
}
|
||||
|
||||
bool DatabasePostgre::DirectExecute(const char* sql)
|
||||
{
|
||||
if (!mPGconn)
|
||||
return false;
|
||||
{
|
||||
// guarded block for thread-safe request
|
||||
ACE_Guard<ACE_Thread_Mutex> query_connection_guard(mMutex);
|
||||
#ifdef MANGOS_DEBUG
|
||||
uint32 _s = getMSTime();
|
||||
#endif
|
||||
PGresult *res = PQexec(mPGconn, sql);
|
||||
if (PQresultStatus(res) != PGRES_COMMAND_OK)
|
||||
{
|
||||
sLog.outErrorDb( "SQL: %s", sql );
|
||||
sLog.outErrorDb( "SQL %s", PQerrorMessage(mPGconn) );
|
||||
return false;
|
||||
}
|
||||
else
|
||||
{
|
||||
#ifdef MANGOS_DEBUG
|
||||
sLog.outDebug("[%u ms] SQL: %s", getMSTime() - _s, sql );
|
||||
#endif
|
||||
}
|
||||
PQclear(res);
|
||||
|
||||
// end guarded block
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
bool DatabasePostgre::_TransactionCmd(const char *sql)
|
||||
bool PostgreSQLConnection::_TransactionCmd(const char *sql)
|
||||
{
|
||||
if (!mPGconn)
|
||||
return false;
|
||||
|
|
@ -263,88 +226,22 @@ bool DatabasePostgre::_TransactionCmd(const char *sql)
|
|||
return true;
|
||||
}
|
||||
|
||||
bool DatabasePostgre::BeginTransaction()
|
||||
bool PostgreSQLConnection::BeginTransaction()
|
||||
{
|
||||
if (!mPGconn)
|
||||
return false;
|
||||
// don't use queued execution if it has not been initialized
|
||||
if (!m_threadBody)
|
||||
{
|
||||
if (tranThread == ACE_Based::Thread::current())
|
||||
return false; // huh? this thread already started transaction
|
||||
mMutex.acquire();
|
||||
if (!_TransactionCmd("START TRANSACTION"))
|
||||
{
|
||||
mMutex.release(); // can't start transaction
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
// transaction started
|
||||
tranThread = ACE_Based::Thread::current(); // owner of this transaction
|
||||
TransactionQueues::iterator i = m_tranQueues.find(tranThread);
|
||||
if (i != m_tranQueues.end() && i->second != NULL)
|
||||
// If for thread exists queue and also contains transaction
|
||||
// delete that transaction (not allow trans in trans)
|
||||
delete i->second;
|
||||
|
||||
m_tranQueues[tranThread] = new SqlTransaction();
|
||||
|
||||
return true;
|
||||
return _TransactionCmd("START TRANSACTION");
|
||||
}
|
||||
|
||||
bool DatabasePostgre::CommitTransaction()
|
||||
bool PostgreSQLConnection::CommitTransaction()
|
||||
{
|
||||
if (!mPGconn)
|
||||
return false;
|
||||
|
||||
// don't use queued execution if it has not been initialized
|
||||
if (!m_threadBody)
|
||||
{
|
||||
if (tranThread != ACE_Based::Thread::current())
|
||||
return false;
|
||||
bool _res = _TransactionCmd("COMMIT");
|
||||
tranThread = NULL;
|
||||
mMutex.release();
|
||||
return _res;
|
||||
}
|
||||
tranThread = ACE_Based::Thread::current();
|
||||
TransactionQueues::iterator i = m_tranQueues.find(tranThread);
|
||||
if (i != m_tranQueues.end() && i->second != NULL)
|
||||
{
|
||||
m_threadBody->Delay(i->second);
|
||||
i->second = NULL;
|
||||
return true;
|
||||
}
|
||||
else
|
||||
return false;
|
||||
return _TransactionCmd("COMMIT");
|
||||
}
|
||||
|
||||
bool DatabasePostgre::RollbackTransaction()
|
||||
{
|
||||
if (!mPGconn)
|
||||
return false;
|
||||
// don't use queued execution if it has not been initialized
|
||||
if (!m_threadBody)
|
||||
{
|
||||
if (tranThread != ACE_Based::Thread::current())
|
||||
return false;
|
||||
bool _res = _TransactionCmd("ROLLBACK");
|
||||
tranThread = NULL;
|
||||
mMutex.release();
|
||||
return _res;
|
||||
}
|
||||
tranThread = ACE_Based::Thread::current();
|
||||
TransactionQueues::iterator i = m_tranQueues.find(tranThread);
|
||||
if (i != m_tranQueues.end() && i->second != NULL)
|
||||
{
|
||||
delete i->second;
|
||||
i->second = NULL;
|
||||
}
|
||||
return true;
|
||||
bool PostgreSQLConnection::RollbackTransaction()
|
||||
{
|
||||
return _TransactionCmd("ROLLBACK");
|
||||
}
|
||||
|
||||
unsigned long DatabasePostgre::escape_string(char *to, const char *from, unsigned long length)
|
||||
unsigned long PostgreSQLConnection::escape_string(char *to, const char *from, unsigned long length)
|
||||
{
|
||||
if (!mPGconn || !to || !from || !length)
|
||||
return 0;
|
||||
|
|
@ -352,23 +249,4 @@ unsigned long DatabasePostgre::escape_string(char *to, const char *from, unsigne
|
|||
return PQescapeString(to, from, length);
|
||||
}
|
||||
|
||||
void DatabasePostgre::InitDelayThread()
|
||||
{
|
||||
assert(!m_delayThread);
|
||||
|
||||
//New delay thread for delay execute
|
||||
m_threadBody = new PGSQLDelayThread(this); // Will be deleted on m_delayThread delete
|
||||
m_delayThread = new ACE_Based::Thread(m_threadBody);
|
||||
}
|
||||
|
||||
void DatabasePostgre::HaltDelayThread()
|
||||
{
|
||||
if (!m_threadBody || !m_delayThread) return;
|
||||
|
||||
m_threadBody->Stop(); //Stop event
|
||||
m_delayThread->wait(); //Wait for flush to DB
|
||||
delete m_delayThread; //This also deletes m_threadBody
|
||||
m_delayThread = NULL;
|
||||
m_threadBody = NULL;
|
||||
}
|
||||
#endif
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue