* Only call Database::ThreadStart()/ThreadEnd() once per thread.

* Some cleanups in MapUpdater/DelayExecutor.

--HG--
branch : trunk
This commit is contained in:
XTZGZoReX
2010-04-05 13:08:47 +02:00
parent a795a1110c
commit 7c277c87d8
7 changed files with 94 additions and 102 deletions

View File

@@ -6,103 +6,102 @@
#include <ace/Guard_T.h>
#include <ace/Method_Request.h>
//the reason this things are here is that i want to make
//the netcode patch and the multithreaded maps independant
//once they are merged 1 class should be used
class WDBThreadStartReq1 : public ACE_Method_Request
{
public:
WDBThreadStartReq1(){}
virtual int
call (void)
{
WorldDatabase.ThreadStart();
CharacterDatabase.ThreadStart();
loginDatabase.ThreadStart();
return 0;
}
WDBThreadStartReq1()
{
}
virtual int call()
{
WorldDatabase.ThreadStart();
return 0;
}
};
class WDBThreadEndReq1 : public ACE_Method_Request
{
public:
WDBThreadEndReq1(){}
virtual int
call (void)
{
WorldDatabase.ThreadEnd();
CharacterDatabase.ThreadEnd();
loginDatabase.ThreadEnd();
return 0;
}
WDBThreadEndReq1()
{
}
virtual int call()
{
WorldDatabase.ThreadEnd();
return 0;
}
};
class MapUpdateRequest : public ACE_Method_Request
{
public:
private:
Map& m_map;
MapUpdater& m_updater;
ACE_UINT32 m_diff;
MapUpdateRequest(Map& m, MapUpdater& u, ACE_UINT32 d) : m_map(m), m_updater(u), m_diff(d){}
virtual int
call (void)
{
m_map.Update (m_diff);
m_updater.update_finished ();
return 0;
}
public:
MapUpdateRequest(Map& m, MapUpdater& u, ACE_UINT32 d)
: m_map(m), m_updater(u), m_diff(d)
{
}
virtual int call()
{
m_map.Update (m_diff);
m_updater.update_finished ();
return 0;
}
};
MapUpdater::MapUpdater() :
m_mutex(),
m_condition(m_mutex),
m_executor(),
pedning_requests(0)
MapUpdater::MapUpdater()
: m_mutex(), m_condition(m_mutex), m_executor(), pending_requests(0)
{
return;
}
MapUpdater::~MapUpdater()
{
this->deactivate();
deactivate();
}
int MapUpdater::activate(size_t num_threads)
{
return this->m_executor.activate(static_cast<int> (num_threads), new WDBThreadStartReq1, new WDBThreadEndReq1);
return m_executor.activate((int)num_threads, new WDBThreadStartReq1, new WDBThreadEndReq1);
}
int MapUpdater::deactivate(void)
int MapUpdater::deactivate()
{
this->wait();
wait();
return this->m_executor.deactivate();
return m_executor.deactivate();
}
int MapUpdater::wait()
{
ACE_GUARD_RETURN(ACE_Thread_Mutex, guard, this->m_mutex, -1);
ACE_GUARD_RETURN(ACE_Thread_Mutex, guard, m_mutex, -1);
while(this->pedning_requests > 0)
this->m_condition.wait();
while (pending_requests > 0)
m_condition.wait();
return 0;
}
int MapUpdater::schedule_update(Map& map, ACE_UINT32 diff)
{
ACE_GUARD_RETURN(ACE_Thread_Mutex, guard, this->m_mutex, -1);
ACE_GUARD_RETURN(ACE_Thread_Mutex, guard, m_mutex, -1);
++this->pedning_requests;
++pending_requests;
if (this->m_executor.execute(new MapUpdateRequest(map, *this, diff)) == -1)
if (m_executor.execute(new MapUpdateRequest(map, *this, diff)) == -1)
{
ACE_DEBUG((LM_ERROR, ACE_TEXT("(%t) \n"), ACE_TEXT("Failed to schedule Map Update")));
--this->pedning_requests;
--pending_requests;
return -1;
}
@@ -116,15 +115,15 @@ bool MapUpdater::activated()
void MapUpdater::update_finished()
{
ACE_GUARD(ACE_Thread_Mutex, guard, this->m_mutex);
ACE_GUARD(ACE_Thread_Mutex, guard, m_mutex);
if (this->pedning_requests == 0)
if (pending_requests == 0)
{
ACE_ERROR((LM_ERROR,ACE_TEXT("(%t)\n"), ACE_TEXT("MapUpdater::update_finished BUG, report to devs")));
ACE_ERROR((LM_ERROR, ACE_TEXT("(%t)\n"), ACE_TEXT("MapUpdater::update_finished BUG, report to devs")));
return;
}
--this->pedning_requests;
--pending_requests;
this->m_condition.broadcast();
m_condition.broadcast();
}

View File

@@ -11,6 +11,7 @@ class Map;
class MapUpdater
{
public:
MapUpdater();
virtual ~MapUpdater();
@@ -22,15 +23,18 @@ class MapUpdater
int activate(size_t num_threads);
int deactivate(void);
int deactivate();
bool activated();
private:
void update_finished();
DelayExecutor m_executor;
ACE_Condition_Thread_Mutex m_condition;
ACE_Thread_Mutex m_mutex;
size_t pedning_requests;
size_t pending_requests;
void update_finished();
};
#endif //_MAP_UPDATER_H_INCLUDED

View File

@@ -155,8 +155,6 @@ class ReactorRunnable : protected ACE_Task_Base
DEBUG_LOG ("Network Thread Starting");
WorldDatabase.ThreadStart();
CharacterDatabase.ThreadStart();
loginDatabase.ThreadStart();
ACE_ASSERT (m_Reactor);
@@ -190,8 +188,6 @@ class ReactorRunnable : protected ACE_Task_Base
}
WorldDatabase.ThreadEnd();
CharacterDatabase.ThreadEnd();
loginDatabase.ThreadEnd();
DEBUG_LOG ("Network Thread Exitting");

View File

@@ -4,16 +4,15 @@
#include "DelayExecutor.h"
DelayExecutor*
DelayExecutor::instance()
DelayExecutor* DelayExecutor::instance()
{
return ACE_Singleton<DelayExecutor, ACE_Thread_Mutex>::instance();
}
DelayExecutor::DelayExecutor():
activated_ (false),
pre_svc_hook_ (0),
post_svc_hook_ (0) {}
DelayExecutor::DelayExecutor()
: activated_(false), pre_svc_hook_(0), post_svc_hook_(0)
{
}
DelayExecutor::~DelayExecutor()
{
@@ -23,49 +22,46 @@ DelayExecutor::~DelayExecutor()
if (post_svc_hook_)
delete post_svc_hook_;
this->deactivate ();
deactivate();
}
int DelayExecutor::deactivate()
{
if (!this->activated())
if (!activated())
return -1;
this->activated(false);
this->queue_.queue()->deactivate();
this->wait();
activated(false);
queue_.queue()->deactivate();
wait();
return 0;
}
int DelayExecutor::svc (void)
int DelayExecutor::svc()
{
if (pre_svc_hook_)
pre_svc_hook_->call();
for (;;)
{
ACE_Method_Request* rq = this->queue_.dequeue();
ACE_Method_Request* rq = queue_.dequeue();
if (!rq)
break;
if (!rq)
break;
rq->call();
delete rq;
rq->call();
delete rq;
}
if (post_svc_hook_)
post_svc_hook_->call();
return 0;
return 0;
}
int DelayExecutor::activate(int num_threads, ACE_Method_Request* pre_svc_hook, ACE_Method_Request* post_svc_hook)
{
if (this->activated())
if (activated())
return -1;
if (num_threads < 1)
@@ -77,15 +73,15 @@ int DelayExecutor::activate(int num_threads, ACE_Method_Request* pre_svc_hook, A
if (post_svc_hook_)
delete post_svc_hook_;
this->pre_svc_hook_ = pre_svc_hook;
this->post_svc_hook_ = post_svc_hook;
pre_svc_hook_ = pre_svc_hook;
post_svc_hook_ = post_svc_hook;
this->queue_.queue ()->activate ();
queue_.queue()->activate();
if (ACE_Task_Base::activate(THR_NEW_LWP | THR_JOINABLE | THR_INHERIT_SCHED, num_threads) == -1)
return -1;
return -1;
this->activated(true);
activated(true);
return true;
}
@@ -95,20 +91,21 @@ int DelayExecutor::execute(ACE_Method_Request* new_req)
if (new_req == NULL)
return -1;
if (this->queue_.enqueue(new_req,(ACE_Time_Value*)&ACE_Time_Value::zero) == -1)
if (queue_.enqueue(new_req, (ACE_Time_Value*)&ACE_Time_Value::zero) == -1)
{
delete new_req;
ACE_ERROR_RETURN((LM_ERROR, ACE_TEXT("(%t) %p\n"), ACE_TEXT("DelayExecutor::execute enqueue")), -1);
}
return 0;
}
bool DelayExecutor::activated()
{
return this->activated_;
return activated_;
}
void DelayExecutor::activated(bool s)
{
this->activated_ = s;
activated_ = s;
}

View File

@@ -1,5 +1,5 @@
#ifndef _M_DELAY_EXECUTOR_H
#define _M_DELAY_EXECUTOR_H
#define _M_DELAY_EXECUTOR_H
#include <ace/Task.h>
#include <ace/Activation_Queue.h>
@@ -8,6 +8,7 @@
class DelayExecutor : protected ACE_Task_Base
{
public:
DelayExecutor();
virtual ~DelayExecutor();
@@ -15,19 +16,22 @@ class DelayExecutor : protected ACE_Task_Base
int execute(ACE_Method_Request* new_req);
int activate(int num_threads = 1, ACE_Method_Request* pre_svc_hook = 0, ACE_Method_Request* post_svc_hook = 0);
int activate(int num_threads = 1, ACE_Method_Request* pre_svc_hook = NULL, ACE_Method_Request* post_svc_hook = NULL);
int deactivate();
bool activated();
virtual int svc(void);
virtual int svc();
private:
ACE_Activation_Queue queue_;
ACE_Method_Request* pre_svc_hook_;
ACE_Method_Request* post_svc_hook_;
bool activated_;
void activated(bool s);
bool activated_;
};
#endif // _M_DELAY_EXECUTOR_H

View File

@@ -373,8 +373,6 @@ void CliRunnable::run()
{
///- Init new SQL thread for the world database (one connection call enough)
WorldDatabase.ThreadStart(); // let thread do safe mySQL requests
CharacterDatabase.ThreadStart();
loginDatabase.ThreadStart();
char commandbuf[256];
bool canflush = true;
@@ -444,6 +442,4 @@ void CliRunnable::run()
///- End the database thread
WorldDatabase.ThreadEnd(); // free mySQL thread resources
CharacterDatabase.ThreadEnd();
loginDatabase.ThreadEnd();
}

View File

@@ -45,8 +45,6 @@ void WorldRunnable::run()
{
///- Init new SQL thread for the world database
WorldDatabase.ThreadStart(); // let thread do safe mySQL requests (one connection call enough)
CharacterDatabase.ThreadStart();
loginDatabase.ThreadStart();
sWorld.InitResultQueue();
@@ -96,6 +94,4 @@ void WorldRunnable::run()
///- End the database thread
WorldDatabase.ThreadEnd(); // free mySQL thread resources
CharacterDatabase.ThreadEnd();
loginDatabase.ThreadEnd();
}