From fb35fbd350ed5d1282f7d4aad733208cfd46e9ed Mon Sep 17 00:00:00 2001 From: Sven Luethi Date: Thu, 1 Apr 2021 00:17:38 +0200 Subject: [PATCH] Implements rudimentary thread-safety for EventBus and Logger classes. --- src/core/EventBus.cpp | 20 ++++++++++++++------ src/core/Logger.cpp | 33 +++++++++++++++++++++------------ src/core/include/EventBus.hpp | 9 +++++++-- src/core/include/Logger.hpp | 4 ++++ src/core/main.cpp | 23 +++++++++++++---------- src/resolver/Server.cpp | 8 +++++++- 6 files changed, 66 insertions(+), 31 deletions(-) diff --git a/src/core/EventBus.cpp b/src/core/EventBus.cpp index 4b458f9..d34c621 100644 --- a/src/core/EventBus.cpp +++ b/src/core/EventBus.cpp @@ -5,21 +5,28 @@ namespace Core { // ============================== EventBus Class ============================== -EventBus::EventBus(Logger& logger) : _logger(logger) { +EventBus::EventBus(Logger& logger) : + _logger(logger) +{ _logger.debug.ln ("Starting Event Bus..."); _thread = thread(&EventBus::run, this); + _logger.debug.ln ("Event Bus Started"); } EventBus::~EventBus() { + _logger.debug.ln ("Stopping Event Bus..."); _thread.join(); // destroy all events in _events vector - _logger.debug.ln ("EventBus has been shut down."); + for (auto event : _events) { delete event; } + _logger.debug.ln ("Event Bus stopped"); } Event& EventBus::registerEvent(const string& eventID) { + _mutex.lock(); for ( const auto event : _events ) { if (event->getID() == eventID) { _logger.debug << "Event '" << eventID << "' found" << Logger::commit; + _mutex.unlock(); return *event; } } @@ -27,27 +34,28 @@ Event& EventBus::registerEvent(const string& eventID) { _logger.debug << "Registering new Event '" << eventID << "'" << Logger::commit; Event* event = new Event(eventID); _events.push_back(event); + _mutex.unlock(); return *event; } void EventBus::run() { - _logger.warn.ln("Hi from EventBus run method! From within Event Thread! Dont forget to make Logger class thread-safe!"); + _logger.warn.ln("Hi from EventBus run method, from within Event Thread!"); } // =============================== Event Class ================================ Event::Event(const string id) : _id(id) {} -Event::~Event() { - -} +Event::~Event() {} const string& Event::getID() const { return _id; }; void Event::subscribe(EventSubscriber* es) { + _mutex.lock(); _subscribers.push_back(es); + _mutex.unlock(); } void Event::fire() { diff --git a/src/core/Logger.cpp b/src/core/Logger.cpp index d26221d..ee2b1ad 100644 --- a/src/core/Logger.cpp +++ b/src/core/Logger.cpp @@ -63,17 +63,19 @@ Logger::Sink::Sink (const Level level, Logger& parent) : _lineIsPrefixed(false) {} void Logger::Sink::ln ( const string& msg ) { - ln(msg.c_str()); + ln (msg.c_str()); } void Logger::Sink::ln ( const char* msg ) { + _parent._mutex.lock(); if ( _level >= _parent._logLevel ) { // If line has already been started, abort it and start new one: if(_lineIsPrefixed) *_parent._output << endl; prefixLine(true); - *_parent._output << msg << endl; + *(_parent._output) << msg << endl; cout << msg << CC_RESET "\n"; } + _parent._mutex.unlock(); } void Logger::Sink::prefixLine(bool resetToFalse) { @@ -82,19 +84,19 @@ void Logger::Sink::prefixLine(bool resetToFalse) { switch ( _level) { case DEBUG: - *_parent._output << timestamp << " " << "DEBUG "; + *(_parent._output) << timestamp << " " << "DEBUG "; cout << timestamp << " " << CC_BLUE "DEBUG "; break; case INFO: - *_parent._output << timestamp << " " << "INFO "; + *(_parent._output) << timestamp << " " << "INFO "; cout << timestamp << " " << CC_GREEN "INFO "; break; case WARN: - *_parent._output << timestamp << " " << "WARN "; + *(_parent._output) << timestamp << " " << "WARN "; cout << timestamp << " " << CC_YELLOW "WARN "; break; case FATAL: - *_parent._output << timestamp << " " << "FATAL "; + *(_parent._output) << timestamp << " " << "FATAL "; cout << timestamp << " " << CC_RED "FATAL "; break; } @@ -103,61 +105,68 @@ void Logger::Sink::prefixLine(bool resetToFalse) { } Logger::Sink& Logger::Sink::operator<<(const string& msg) { + _parent._mutex.try_lock(); if ( _level >= _parent._logLevel ) { prefixLine(); - *_parent._output << msg; + *(_parent._output) << msg; cout << msg; } return *this; } Logger::Sink& Logger::Sink::operator<<(const char* msg) { + _parent._mutex.try_lock(); if ( _level >= _parent._logLevel ) { prefixLine(); - *_parent._output << msg; + *(_parent._output) << msg; cout << msg; } return *this; } Logger::Sink& Logger::Sink::operator<<(const int msg) { + _parent._mutex.try_lock(); if ( _level >= _parent._logLevel ) { prefixLine(); - *_parent._output << msg; + *(_parent._output) << msg; cout << msg; } return *this; } Logger::Sink& Logger::Sink::operator<<(const long msg) { + _parent._mutex.try_lock(); if ( _level >= _parent._logLevel ) { prefixLine(); - *_parent._output << msg; + *(_parent._output) << msg; cout << msg; } return *this; } Logger::Sink& Logger::Sink::operator<<(const unsigned long msg) { + _parent._mutex.try_lock(); if ( _level >= _parent._logLevel ) { prefixLine(); - *_parent._output << msg; + *(_parent._output) << msg; cout << msg; } return *this; } Logger::Sink& Logger::Sink::operator<<(const char character) { + _parent._mutex.try_lock(); if ( _level >= _parent._logLevel ) { switch(character) { // COMMIT - End message line case 0: - *_parent._output << endl; + *(_parent._output) << endl; cout << CC_RESET "\n"; _lineIsPrefixed = false; break; } } + _parent._mutex.unlock(); return *this; } diff --git a/src/core/include/EventBus.hpp b/src/core/include/EventBus.hpp index 48c0fcf..65bcef5 100644 --- a/src/core/include/EventBus.hpp +++ b/src/core/include/EventBus.hpp @@ -5,6 +5,7 @@ #include #include #include +#include namespace Plains { namespace Core { @@ -12,6 +13,7 @@ namespace Core { using std::thread; using std::string; using std::vector; +using std::mutex; using std::unique_ptr; class EventSubscriber { @@ -30,6 +32,7 @@ class Event { private: const string _id; vector _subscribers; + mutex _mutex; }; class EventBus { @@ -38,12 +41,14 @@ class EventBus { ~EventBus(); Event& registerEvent(const string& eventID); + private: + void run(); + private: Logger& _logger; thread _thread; vector _events; - - void run(); + mutex _mutex; }; }} diff --git a/src/core/include/Logger.hpp b/src/core/include/Logger.hpp index 8ac0a6a..152da68 100644 --- a/src/core/include/Logger.hpp +++ b/src/core/include/Logger.hpp @@ -2,6 +2,7 @@ #define LOGGER_HPP #include #include +#include #define CC(COLOR) CC_##COLOR #define CC_RESET "\033[0m" @@ -20,6 +21,7 @@ namespace Core { using std::string; using std::ofstream; +using std::mutex; //using std::ostream; //using std::streambuf; @@ -39,6 +41,7 @@ class Logger { Logger& _parent; const Logger::Level _level; void prefixLine(bool resetToFalse = false); + public: Sink(const Level level, Logger& parent); void ln (const string& msg); @@ -56,6 +59,7 @@ class Logger { private: enum Level _logLevel; ofstream* _output; + mutex _mutex; static string genTimeStamp(); }; diff --git a/src/core/main.cpp b/src/core/main.cpp index 4cb318e..e7dcf39 100644 --- a/src/core/main.cpp +++ b/src/core/main.cpp @@ -3,6 +3,7 @@ #include "DNS.hpp" #include #include +#include using namespace Plains; using namespace Core; @@ -15,13 +16,13 @@ int main(int args, char **argv) { logger.info.ln ("Starting PlaiNS..."); + vector arguments; + for ( unsigned index = 0; index < args; ++index ) { + arguments.push_back (argv[index]); + } + try { - vector arguments; - for ( unsigned index = 0; index < args; ++index ) { - arguments.push_back (argv[index]); - } - ConfigManager cfg(logger, arguments); EventBus eventBus(logger); Resolver::Server resolver(logger, eventBus); @@ -54,15 +55,17 @@ int main(int args, char **argv) { //resolver.run(); logger.info.ln ("PlaiNS successfully started."); // Run something on main thread... - logger.info.ln ("Shutting down PlaiNS. Good Bye!"); + logger.info.ln ("Shutting down PlaiNS. Good Bye!"); } - catch (Exception& e) - { + catch (Exception& e) { logger.fatal << "Uncaught Exception: " << e.what() << Logger::commit; return 1; } - catch (int e) - { + catch (std::system_error& e) { + logger.fatal << "Uncaught system_error exception: " << e.what() << Logger::commit; + return 1; + } + catch (int e) { logger.fatal << "Uncaught Exception with error code " << e << Logger::commit; return 1; } diff --git a/src/resolver/Server.cpp b/src/resolver/Server.cpp index 2e763e4..eea877b 100644 --- a/src/resolver/Server.cpp +++ b/src/resolver/Server.cpp @@ -12,6 +12,7 @@ namespace Plains { namespace Resolver { using asio::ip::udp; +using asio::ip::tcp; static char* byteToHex(char c) { static char* out = new char[3]; @@ -93,9 +94,12 @@ Server::Server(Logger& logger, EventBus& eventBus) _udpSocket = new udp::socket(_ioContext, udp::endpoint(udp::v4(), 8853)); _ioContext.run(); doReceive(); + _logger.debug.ln ("Resolver started"); } -Server::~Server() {} +Server::~Server() { + _udpSocket->close(); +} //void Server::run() { // _ioContext.run(); @@ -129,4 +133,6 @@ void Server::doSend(std::size_t length) + + }}