Replaced ACE_Task_Base based LogWorker with ProducerConsumerQueue

This commit is contained in:
leak
2014-06-30 14:44:52 +02:00
parent a5c742dafe
commit eb36acd152
7 changed files with 141 additions and 34 deletions

View File

@@ -67,7 +67,6 @@ using boost::asio::ip::tcp;
void SignalHandler(const boost::system::error_code& error, int signalNumber)
{
TC_LOG_ERROR("server.authserver", "SIGNAL HANDLER WORKING");
if (!error)
{
switch (signalNumber)

View File

@@ -71,7 +71,7 @@ bool ConfigMgr::GetBoolDefault(const char* name, bool def)
{
try
{
std::string val = _config.get<std::string>(name);
std::string val = _config.get<std::string>(ptree::path_type(name, '/'));
val.erase(std::remove(val.begin(), val.end(), '"'), val.end());
return (val == "true" || val == "TRUE" || val == "yes" || val == "YES" || val == "1");
}
@@ -83,12 +83,12 @@ bool ConfigMgr::GetBoolDefault(const char* name, bool def)
int ConfigMgr::GetIntDefault(const char* name, int def)
{
return _config.get<int>(name, def);
return _config.get<int>(ptree::path_type(name, '/'), def);
}
float ConfigMgr::GetFloatDefault(const char* name, float def)
{
return _config.get<float>(name, def);
return _config.get<float>(ptree::path_type(name, '/'), def);
}
std::string const& ConfigMgr::GetFilename()

View File

@@ -273,7 +273,7 @@ void Log::write(LogMessage* msg) const
msg->text.append("\n");
if (worker)
worker->enqueue(new LogOperation(logger, msg));
worker->Enqueue(new LogOperation(logger, msg));
else
{
logger->write(*msg);

View File

@@ -23,6 +23,7 @@
#include "Appender.h"
#include "Logger.h"
#include "LogWorker.h"
#include <stdarg.h>
#include <unordered_map>
#include <string>

View File

@@ -16,35 +16,41 @@
*/
#include "LogWorker.h"
#include <thread>
LogWorker::LogWorker()
: m_queue(HIGH_WATERMARK, LOW_WATERMARK)
{
ACE_Task_Base::activate(THR_NEW_LWP | THR_JOINABLE | THR_INHERIT_SCHED, 1);
_cancelationToken = false;
_workerThread = std::thread(&LogWorker::WorkerThread, this);
}
LogWorker::~LogWorker()
{
m_queue.deactivate();
wait();
_cancelationToken = true;
_queue.Cancel();
_workerThread.join();
}
int LogWorker::enqueue(LogOperation* op)
void LogWorker::Enqueue(LogOperation* op)
{
return m_queue.enqueue(op);
return _queue.Push(op);
}
int LogWorker::svc()
void LogWorker::WorkerThread()
{
while (1)
{
LogOperation* request;
if (m_queue.dequeue(request) == -1)
break;
LogOperation* operation = nullptr;
_queue.WaitAndPop(operation);
request->call();
delete request;
if (_cancelationToken)
return;
operation->call();
delete operation;
}
return 0;
}

View File

@@ -18,30 +18,29 @@
#ifndef LOGWORKER_H
#define LOGWORKER_H
#include <atomic>
#include <thread>
#include "LogOperation.h"
#include "ProducerConsumerQueue.h"
#include <ace/Task.h>
#include <ace/Activation_Queue.h>
class LogWorker: protected ACE_Task_Base
class LogWorker
{
public:
LogWorker();
~LogWorker();
typedef ACE_Message_Queue_Ex<LogOperation, ACE_MT_SYNCH> LogMessageQueueType;
enum
{
HIGH_WATERMARK = 8 * 1024 * 1024,
LOW_WATERMARK = 8 * 1024 * 1024
};
int enqueue(LogOperation *op);
void Enqueue(LogOperation *op);
private:
virtual int svc();
LogMessageQueueType m_queue;
ProducerConsumerQueue<LogOperation*> _queue;
void WorkerThread();
std::thread _workerThread;
std::atomic_bool _cancelationToken;
};
#endif

View File

@@ -0,0 +1,102 @@
/*
* Copyright (C) 2008-2014 TrinityCore <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, see <http://www.gnu.org/licenses/>.
*/
#ifndef _PCQ_H
#define _PCQ_H
#include <condition_variable>
#include <mutex>
#include <queue>
template <typename T>
class ProducerConsumerQueue
{
private:
std::mutex _queueLock;
std::queue<T> _queue;
std::condition_variable _condition;
public:
void Push(const T& value)
{
_queueLock.lock();
_queue.push(std::move(value));
_queueLock.unlock();
_condition.notify_one();
}
bool Empty() const
{
std::lock_guard<std::mutex> lock(_queueLock);
return _queue.empty();
}
bool Pop(T& value)
{
std::lock_guard<std::mutex> lock(_queueLock);
if (_queue.empty())
return false;
value = _queue.front();
_queue.pop();
return true;
}
void WaitAndPop(T& value)
{
std::unique_lock<std::mutex> lock(_queueLock);
_condition.wait(lock, [this](){ return !_queue.empty(); });
if (_queue.empty())
return;
value = _queue.front();
_queue.pop();
}
void Cancel()
{
_queueLock.lock();
while (!_queue.empty())
{
T& value = _queue.front();
delete &value;
_queue.pop();
}
_queueLock.unlock();
_condition.notify_all();
}
};
#endif