* HIGHLY EXPERIMENTAL - USE AT OWN RISK *
Database Layer: - Implement connection pooling: Instead of 1 delay thread per database, you can configure between 1 and 32 worker threads that have a seperate thread in the core and have a seperate connection to the MySQL server (based on raczman/Albator´s database layer for Trinitycore3) - Implement a configurable thread bundle for synchroneous requests from seperate core threads (see worldserver.conf.dist for more info) - Every mapupdate thread now has its seperate MySQL connection to the world and characters database - Drop inconsistent PExecuteLog function - query logging will be implemented CONSISTENTLY later - Drop current prepared statement interface - this will be done *properly* later - You´ll need to update your worldserver.conf and authserver.conf - You´re recommended to make a backup of your databases before using this. * HIGHLY EXPERIMENTAL - USE AT OWN RISK * * HIGHLY EXPERIMENTAL - USE AT OWN RISK * etc. --HG-- branch : trunk
This commit is contained in:
Machiavelli
@@ -0,0 +1,216 @@
|
||||
/*
|
||||
* Copyright (C) 2008-2010 Trinity <http://www.trinitycore.org/>
|
||||
*
|
||||
* This program is free software; you can redistribute it and/or modify
|
||||
* it under the terms of the GNU General Public License as published by
|
||||
* the Free Software Foundation; either version 2 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU General Public License
|
||||
* along with this program; if not, write to the Free Software
|
||||
* Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
|
||||
*/
|
||||
|
||||
#include "DatabaseWorkerPool.h"
|
||||
#include "MySQLConnection.h"
|
||||
#include "DatabaseEnv.h"
|
||||
#include "SQLOperation.h"
|
||||
|
||||
DatabaseWorkerPool::DatabaseWorkerPool() :
|
||||
m_queue(new ACE_Activation_Queue(new ACE_Message_Queue<ACE_MT_SYNCH>)),
|
||||
m_connections(0)
|
||||
{
|
||||
m_infoString = "";
|
||||
|
||||
mysql_library_init(-1, NULL, NULL);
|
||||
WPFatal (mysql_thread_safe(), "Used MySQL library isn't thread-safe.");
|
||||
}
|
||||
|
||||
DatabaseWorkerPool::~DatabaseWorkerPool()
|
||||
{
|
||||
}
|
||||
|
||||
bool DatabaseWorkerPool::Open(const std::string& infoString, uint8 num_threads)
|
||||
{
|
||||
sLog.outDebug("Creating bundled/master MySQL connection.");
|
||||
m_bundle_conn = new MySQLConnection();
|
||||
m_bundle_conn->Open(infoString);
|
||||
++m_connections;
|
||||
|
||||
m_async_connections.resize(num_threads);
|
||||
|
||||
/// Open the Async pool
|
||||
for (uint8 i = 0; i < num_threads; i++)
|
||||
{
|
||||
m_async_connections[i] = new MySQLConnection(m_queue);
|
||||
m_async_connections[i]->Open(infoString);
|
||||
++m_connections;
|
||||
sLog.outDebug("Async database thread pool opened. Worker thread count: %u", num_threads);
|
||||
}
|
||||
|
||||
m_infoString = infoString;
|
||||
return true;
|
||||
}
|
||||
|
||||
void DatabaseWorkerPool::Close()
|
||||
{
|
||||
/// Shuts down worker threads for this connection pool.
|
||||
for (uint8 i = 0; i < m_async_connections.size(); i++)
|
||||
Enqueue(NULL);
|
||||
|
||||
//- MySQL::Thread_End() should be called manually from the aborting calling threads
|
||||
|
||||
delete m_bundle_conn;
|
||||
m_bundle_conn = NULL;
|
||||
}
|
||||
|
||||
/*! This function creates a new MySQL connection for every MapUpdate thread
|
||||
and every unbundled task.
|
||||
*/
|
||||
void DatabaseWorkerPool::Init_MySQL_Connection()
|
||||
{
|
||||
MySQLConnection* conn = new MySQLConnection();
|
||||
conn->Open(m_infoString);
|
||||
|
||||
{
|
||||
ACE_Guard<ACE_Thread_Mutex> guard(m_connectionMap_mtx);
|
||||
m_sync_connections[ACE_Based::Thread::current()] = conn;
|
||||
}
|
||||
|
||||
sLog.outDebug("Core thread with ID ["UI64FMTD"] initializing MySQL connection.",
|
||||
(uint64)ACE_Based::Thread::currentId());
|
||||
}
|
||||
|
||||
void DatabaseWorkerPool::End_MySQL_Connection()
|
||||
{
|
||||
MySQLConnection* conn;
|
||||
{
|
||||
ACE_Guard<ACE_Thread_Mutex> guard(m_connectionMap_mtx);
|
||||
conn = m_sync_connections[ACE_Based::Thread::current()];
|
||||
}
|
||||
delete conn;
|
||||
conn = NULL;
|
||||
}
|
||||
|
||||
void DatabaseWorkerPool::Execute(const char* sql)
|
||||
{
|
||||
if (!sql)
|
||||
return;
|
||||
|
||||
BasicStatementTask* task = new BasicStatementTask(sql);
|
||||
Enqueue(task);
|
||||
}
|
||||
|
||||
void DatabaseWorkerPool::PExecute(const char* sql, ...)
|
||||
{
|
||||
if (!sql)
|
||||
return;
|
||||
|
||||
va_list ap;
|
||||
char szQuery[MAX_QUERY_LEN];
|
||||
va_start(ap, sql);
|
||||
int res = vsnprintf(szQuery, MAX_QUERY_LEN, sql, ap);
|
||||
va_end(ap);
|
||||
|
||||
Execute(szQuery);
|
||||
}
|
||||
|
||||
void DatabaseWorkerPool::DirectExecute(const char* sql)
|
||||
{
|
||||
if (sql)
|
||||
GetConnection()->Execute(sql);
|
||||
}
|
||||
|
||||
void DatabaseWorkerPool::DirectPExecute(const char* sql, ...)
|
||||
{
|
||||
if (!sql)
|
||||
return;
|
||||
|
||||
va_list ap;
|
||||
char szQuery[MAX_QUERY_LEN];
|
||||
va_start(ap, sql);
|
||||
int res = vsnprintf(szQuery, MAX_QUERY_LEN, sql, ap);
|
||||
va_end(ap);
|
||||
|
||||
return DirectExecute(szQuery);
|
||||
}
|
||||
|
||||
QueryResult_AutoPtr DatabaseWorkerPool::Query(const char* sql)
|
||||
{
|
||||
return GetConnection()->Query(sql);
|
||||
}
|
||||
|
||||
QueryResult_AutoPtr DatabaseWorkerPool::PQuery(const char* sql, ...)
|
||||
{
|
||||
if (!sql)
|
||||
return QueryResult_AutoPtr(NULL);
|
||||
|
||||
va_list ap;
|
||||
char szQuery[MAX_QUERY_LEN];
|
||||
va_start(ap, sql);
|
||||
int res = vsnprintf(szQuery, MAX_QUERY_LEN, sql, ap);
|
||||
va_end(ap);
|
||||
|
||||
return Query(szQuery);
|
||||
}
|
||||
|
||||
void DatabaseWorkerPool::BeginTransaction()
|
||||
{
|
||||
ACE_Guard<ACE_Thread_Mutex> guard(m_transQueues_mtx);
|
||||
ACE_Based::Thread* tranThread = ACE_Based::Thread::current(); // owner of this transaction
|
||||
TransactionQueues::iterator itr = m_tranQueues.find(tranThread);
|
||||
if (itr != m_tranQueues.end() && itr->second != NULL)
|
||||
{
|
||||
itr->second->ForcefulDelete();
|
||||
delete itr->second;
|
||||
}
|
||||
m_tranQueues[tranThread] = new TransactionTask();
|
||||
return;
|
||||
}
|
||||
|
||||
void DatabaseWorkerPool::RollbackTransaction()
|
||||
{
|
||||
ACE_Guard<ACE_Thread_Mutex> guard(m_transQueues_mtx);
|
||||
ACE_Based::Thread* tranThread = ACE_Based::Thread::current(); // owner of this transaction
|
||||
TransactionQueues::iterator itr = m_tranQueues.find(tranThread);
|
||||
if (itr != m_tranQueues.end() && itr->second != NULL)
|
||||
{
|
||||
itr->second->ForcefulDelete();
|
||||
delete itr->second;
|
||||
}
|
||||
}
|
||||
|
||||
void DatabaseWorkerPool::CommitTransaction()
|
||||
{
|
||||
ACE_Guard<ACE_Thread_Mutex> guard(m_transQueues_mtx);
|
||||
ACE_Based::Thread* tranThread = ACE_Based::Thread::current(); // owner of this transaction
|
||||
TransactionQueues::iterator itr = m_tranQueues.find(tranThread);
|
||||
if (itr != m_tranQueues.end() && itr->second != NULL)
|
||||
{
|
||||
Enqueue(itr->second);
|
||||
m_tranQueues.erase(itr);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
MySQLConnection* DatabaseWorkerPool::GetConnection()
|
||||
{
|
||||
MySQLConnection* conn;
|
||||
ConnectionMap::const_iterator itr;
|
||||
{
|
||||
/*! MapUpdate + unbundled threads */
|
||||
ACE_Guard<ACE_Thread_Mutex> guard(m_connectionMap_mtx);
|
||||
itr = m_sync_connections.find(ACE_Based::Thread::current());
|
||||
if (itr != m_sync_connections.end())
|
||||
conn = itr->second;
|
||||
}
|
||||
/*! Bundled threads */
|
||||
conn = m_bundle_conn;
|
||||
ASSERT (conn);
|
||||
return conn;
|
||||
}
|
||||
Reference in New Issue
Block a user