Merge branch 'bugfix/StopResponding_#58' into develop closes #58

BaseCommon/Log.h

@@ -16,7 +16,6 @@
  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  */
 
-
 #pragma once
 
 #include <ctime>
@@ -39,10 +38,10 @@ private:
     static uint m_prevMsgCount;
     static uint m_repeatThreshold;
 
-    static void getTimeStamp(std::string& s);
+    static void getTimeStamp(std::string &s);
 
-    template<typename... Args>
-    static void formatLogMessage(std::string& output, LOG_SEVERITY severity, const std::string & f, Args... args)
+    template <typename... Args>
+    static void formatLogMessage(std::string &output, LOG_SEVERITY severity, const std::string &f, Args... args)
     {
         assert(severity != LOG_NONE);
 
@@ -58,7 +57,7 @@ private:
         case LOG_FATAL:
             severityStr.assign("FATAL  ");
             break;
-        case LOG_INFO :
+        case LOG_INFO:
             severityStr.assign("INFO   ");
             break;
         case LOG_WARNING:
@@ -74,84 +73,113 @@ private:
         std::string f2("[%s] ");
         f2.append(f);
         CStringUtils::string_format_in_place(output, f2, severityStr.c_str(), args...);
-        boost::trim_if(output, [](char c){ return c == '\n' || c == '\r' || c == ' ' || c == '\t'; });
+        boost::trim_if(output, [](char c)
+                       { return c == '\n' || c == '\r' || c == ' ' || c == '\t'; });
         output.push_back('\n');
     }
 
 public:
-    static void addTarget(CLogTarget * target);
+    static void addTarget(CLogTarget *target);
     static void finalise();
-    static uint& getRepeatThreshold();
+    static uint &getRepeatThreshold();
 
-    template<typename... Args> static void logTrace(const std::string & f, Args... args)
+    template <typename... Args>
+    static void logTrace(const std::string &f, Args... args)
     {
         log(LOG_TRACE, f, args...);
     }
 
-    template<typename... Args> static void logDebug(const std::string & f, Args... args)
+    template <typename... Args>
+    static void logDebug(const std::string &f, Args... args)
     {
         log(LOG_DEBUG, f, args...);
     }
 
-    template<typename... Args> static void logInfo(const std::string & f, Args... args)
+    template <typename... Args>
+    static void logInfo(const std::string &f, Args... args)
     {
         log(LOG_INFO, f, args...);
     }
 
-    template<typename... Args> static void logWarning(const std::string & f, Args... args)
+    template <typename... Args>
+    static void logWarning(const std::string &f, Args... args)
     {
         log(LOG_WARNING, f, args...);
     }
 
-    template<typename... Args> static void logError(const std::string & f, Args... args)
+    template <typename... Args>
+    static void logError(const std::string &f, Args... args)
     {
         log(LOG_ERROR, f, args...);
     }
 
-    template<typename... Args> static void logFatal(const std::string & f, Args... args)
+    template <typename... Args>
+    static void logFatal(const std::string &f, Args... args)
     {
         log(LOG_FATAL, f, args...);
     }
 
-    template<typename... Args> static void log(LOG_SEVERITY severity, const std::string & f, Args... args)
+    template <typename... Args>
+    static void log(LOG_SEVERITY severity, const std::string &f, Args... args)
     {
+        // Protect against concurrent access to log targets
         std::lock_guard lockTarget(m_targetsMutex);
 
-        if(m_targets.empty())
+        if (m_targets.empty())
             return;
 
+        // Format the message with the given arguments
         std::string msg;
         formatLogMessage(msg, severity, f, args...);
 
-        bool repeatedMsg = (msg.compare(m_prevMsg) == 0);
+        bool repeatedMsg = (msg == m_prevMsg);
 
-        if(repeatedMsg && m_repeatThreshold > 0U) {
+        // Handle repeated messages
+        if (repeatedMsg && m_repeatThreshold > 0U) {
             m_prevMsgCount++;
-            if(m_prevMsgCount >= m_repeatThreshold)
+            if (m_prevMsgCount >= m_repeatThreshold)
+            {
+                // If threshold reached, skip logging this duplicate
                 return;
+            }
         }
 
-        m_prevMsg.assign(msg);
+        // If we are leaving a repetition sequence, log a summary first
+        if (!repeatedMsg && m_repeatThreshold > 0U && m_prevMsgCount >= m_repeatThreshold) {
+            std::string summary;
+            formatLogMessage(summary, severity,
+                             "Previous message repeated %d times",
+                             m_prevMsgCount - m_repeatThreshold + 1);
 
-        if(m_prevMsgCount >= m_repeatThreshold && !repeatedMsg && m_repeatThreshold > 0U) {
-            formatLogMessage(msg, severity, "Previous message repeated %d times", m_prevMsgCount - m_repeatThreshold + 1);
-            m_prevMsg.clear();
-        }
+            std::string ts;
+            getTimeStamp(ts);
 
-        std::string timestamp;
-        getTimeStamp(timestamp);
-        std::string msgts;
-        CStringUtils::string_format_in_place(msgts, "[%s] %s", timestamp.c_str(), msg.c_str());
+            std::string summaryLine;
+            CStringUtils::string_format_in_place(summaryLine, "[%s] %s", ts.c_str(), summary.c_str());
 
-        for(auto target : m_targets) {
-            if(severity >= target->getLevel()) {
-                target->printLog(msgts);
+            for (auto target : m_targets)
+            {
+                if (severity >= target->getLevel())
+                    target->printLog(summaryLine);
             }
-        }
 
-        if(m_prevMsgCount != 0 && !repeatedMsg) {
+            // Reset repetition counter after summary
             m_prevMsgCount = 0;
-            log(severity, f, args ...);
         }
+
+        // Always log the current message
+        std::string ts;
+        getTimeStamp(ts);
+
+        std::string msgLine;
+        CStringUtils::string_format_in_place(msgLine, "[%s] %s", ts.c_str(), msg.c_str());
+
+        for (auto target : m_targets) {
+            if (severity >= target->getLevel())
+                target->printLog(msgLine);
+        }
+
+        // Save current message for repetition detection
+        m_prevMsg = msg;
     }
 };

BaseCommon/Utils.h

@@ -19,6 +19,14 @@
 #include <vector>
 #include <climits>
 #include <memory>
+#include <thread>
+#include <sstream>
+
+#define THREAD_ID_STR(tid) ([](std::thread::id id){ \
+    std::stringstream ss; \
+    ss << id; \
+    return ss.str(); \
+}(tid).c_str())
 
 enum TRISTATE {
 	STATE_FALSE,

DStarGateway/DStarGatewayApp.cpp

@@ -118,6 +118,7 @@ int main(int argc, char *argv[])
 	TLog logConf;
 	config->getLog(logConf);
 	CLog::finalise();
+	CLog::getRepeatThreshold() = logConf.repeatThreshold;
 	if(logConf.displayLevel	!= LOG_NONE && !daemon.daemon) CLog::addTarget(new CLogConsoleTarget(logConf.displayLevel));
 	if(logConf.fileLevel		!= LOG_NONE) CLog::addTarget(new CLogFileTarget(logConf.fileLevel, logConf.logDir, logConf.fileRoot, logConf.fileRotate));
 
@@ -170,6 +171,7 @@ void CDStarGatewayApp::run()
 
 bool CDStarGatewayApp::createThread()
 {
+	CLog::logTrace("Entering CDStarGatewayApp::createThread - Thread ID %s", THREAD_ID_STR(std::this_thread::get_id()));
 	// Log
 	TLog log;
 	m_config->getLog(log);
@@ -257,6 +259,7 @@ bool CDStarGatewayApp::createThread()
 	bool atLeastOneRepeater = false;
 	CRepeaterProtocolHandlerFactory repeaterProtocolFactory;
 	for(unsigned int i = 0U; i < m_config->getRepeaterCount(); i++) {
+		CLog::logTrace("Adding repeaters - CDStarGatewayApp::createThread - Rpt Idx %i - Thread ID %s", i, THREAD_ID_STR(std::this_thread::get_id()));
 		TRepeater rptrConfig;
 		m_config->getRepeater(i, rptrConfig);
 		auto  repeaterProtocolHandler = repeaterProtocolFactory.getRepeaterProtocolHandler(rptrConfig.hwType, gatewayConfig, rptrConfig.address, rptrConfig.port);
@@ -294,6 +297,8 @@ bool CDStarGatewayApp::createThread()
 		if(!ddEnabled) ddEnabled = rptrConfig.band.length() > 1U;
 	}
 
+	CLog::logTrace("Repeaters Added - CDStarGatewayApp::createThread - Thread ID %s", THREAD_ID_STR(std::this_thread::get_id()));
+
 	if(!atLeastOneRepeater) {
 		CLog::logError("Error: no repeaters are enabled or opening network communication to repeater failed");
 		return false;
@@ -306,40 +311,46 @@ bool CDStarGatewayApp::createThread()
 	auto ircddbVersionInfo = "linux_" + PRODUCT_NAME + "-" + VERSION;
 	std::vector<CIRCDDB *> clients;
 	for(unsigned int i=0; i < m_config->getIrcDDBCount(); i++) {
+		CLog::logTrace("Adding Ircddb - CDStarGatewayApp::createThread - Ircddb  Idx %i - Thread ID %s", i, THREAD_ID_STR(std::this_thread::get_id()));
 		TircDDB ircDDBConfig;
 		m_config->getIrcDDB(i, ircDDBConfig);
 		CLog::logInfo("ircDDB Network %d set to %s user: %s, Quadnet %d", i + 1,ircDDBConfig.hostname.c_str(), ircDDBConfig.username.c_str(), ircDDBConfig.isQuadNet);
 		CIRCDDB * ircDDB = new CIRCDDBClient(ircDDBConfig.hostname, 9007U, ircDDBConfig.username, ircDDBConfig.password, ircddbVersionInfo, gatewayConfig.address, ircDDBConfig.isQuadNet);
 		clients.push_back(ircDDB);
 	}
+	CLog::logTrace("Added Ircddb - CDStarGatewayApp::createThread - Ircddb  Count %i - Thread ID %s", clients.size(), THREAD_ID_STR(std::this_thread::get_id()));
 	if(clients.size() > 0U) {
 		CIRCDDBMultiClient* multiClient = new CIRCDDBMultiClient(clients);
 		bool res = multiClient->open();
 		if (!res) {
-			CLog::logInfo("Cannot initialise the ircDDB protocol handler\n");
+			CLog::logError("Cannot initialise the ircDDB protocol handler\n");
 			return false;
 		}
 		m_thread->setIRC(multiClient);
 	}
 
+	CLog::logTrace("Setting Up Dextra CDStarGatewayApp::createThread - Thread ID %s", THREAD_ID_STR(std::this_thread::get_id()));
 	// Setup Dextra
 	TDextra dextraConfig;
 	m_config->getDExtra(dextraConfig);
 	CLog::logInfo("DExtra enabled: %d, max. dongles: %u, url: %s", int(dextraConfig.enabled), dextraConfig.maxDongles, dextraConfig.hostfileUrl.c_str());
 	m_thread->setDExtra(dextraConfig.enabled, dextraConfig.maxDongles);
 
+	CLog::logTrace("Setting Up DCS CDStarGatewayApp::createThread - Thread ID %s", THREAD_ID_STR(std::this_thread::get_id()));
 	// Setup DCS
 	TDCS dcsConfig;
 	m_config->getDCS(dcsConfig);
 	CLog::logInfo("DCS enabled: %d, url: %s", int(dcsConfig.enabled), dcsConfig.hostfileUrl.c_str());
 	m_thread->setDCS(dcsConfig.enabled);
 
+	CLog::logTrace("Setting Up DPlus CDStarGatewayApp::createThread - Thread ID %s", THREAD_ID_STR(std::this_thread::get_id()));
 	// Setup DPlus
 	TDplus dplusConfig;
 	m_config->getDPlus(dplusConfig);
 	CLog::logInfo("D-Plus enabled: %d, max. dongles: %u, login: %s, url: %s", int(dplusConfig.enabled), dplusConfig.maxDongles, dplusConfig.login.c_str(), dplusConfig.hostfileUrl.c_str());
 	m_thread->setDPlus(dplusConfig.enabled, dplusConfig.maxDongles, dplusConfig.login);
 
+	CLog::logTrace("Setting Up XLX CDStarGatewayApp::createThread - Thread ID %s", THREAD_ID_STR(std::this_thread::get_id()));
 	// Setup XLX
 	TXLX xlxConfig;
 	m_config->getXLX(xlxConfig);

IRCDDB/IRCClient.cpp

@@ -58,19 +58,24 @@ IRCClient::IRCClient(IRCApplication *app, const std::string& update_channel, con
 
 IRCClient::~IRCClient()
 {
-	delete m_proto;
+    stopWork();       
+    delete m_proto;
 }
 
 void IRCClient::startWork()
 {
-	m_terminateThread = false;
-	m_future = std::async(std::launch::async, &IRCClient::Entry, this);
+    if (m_thread.joinable())
+        return;
+
+    m_terminateThread.store(false, std::memory_order_relaxed);
+    m_thread = std::thread(&IRCClient::Entry, this);
 }
 
 void IRCClient::stopWork()
 {
-	m_terminateThread = true;
-	m_future.get();
+    m_terminateThread.store(true, std::memory_order_relaxed);
+    if (m_thread.joinable())
+        m_thread.join();
 }
 
 void IRCClient::Entry()
@@ -98,7 +103,7 @@ void IRCClient::Entry()
 
 		switch (state) {
 			case 0:
-				if (m_terminateThread) {
+				if (m_terminateThread.load(std::memory_order_relaxed)) {
 					CLog::logInfo("IRCClient::Entry: thread terminated at state=%d\n", state);
 					return;
 				}
@@ -118,7 +123,7 @@ void IRCClient::Entry()
 				break;
 
 			case 1:
-				if (m_terminateThread) {
+				if (m_terminateThread.load(std::memory_order_relaxed)) {
 					CLog::logInfo("IRCClient::Entry: thread terminated at state=%d\n", state);
 					return;
 				}
@@ -261,7 +266,7 @@ void IRCClient::Entry()
 
 
 			case 5:
-				if (m_terminateThread)
+				if (m_terminateThread.load(std::memory_order_relaxed))
 					state = 6;
 				else {
 					if (m_recvQ->isEOF()) {

IRCDDB/IRCClient.h

@@ -22,7 +22,8 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #pragma once
 
 #include <string>
-#include <future>
+#include <thread>
+#include <atomic>   
 
 #include "IRCReceiver.h"
 #include "IRCMessageQueue.h"
@@ -47,13 +48,13 @@ private:
 	unsigned int m_port;
 	std::string m_callsign;
 	std::string m_password;
-	bool m_terminateThread;
+	std::atomic<bool> m_terminateThread{false};
 
 	IRCReceiver *m_recv;
 	IRCMessageQueue *m_recvQ;
 	IRCMessageQueue *m_sendQ;
 	IRCProtocol *m_proto;
 	IRCApplication *m_app;
-	std::future<void> m_future;
-};
+	std::thread m_thread;
 
+};

IRCDDB/IRCDDBApp.cpp

@@ -112,7 +112,7 @@ public:
 	std::regex m_fromPattern;
 
 	bool m_initReady;
-	bool m_terminateThread;
+	std::atomic<bool> m_terminateThread{false};
 
 	std::map<std::string, IRCDDBAppUserObject> m_userMap;
 	std::mutex m_userMapMutex;
@@ -151,8 +151,10 @@ IRCDDBApp::IRCDDBApp(const std::string& u_chan)
 
 IRCDDBApp::~IRCDDBApp()
 {
-	delete m_d->m_sendQ;
-	delete m_d;
+    stopWork();
+
+    delete m_d->m_sendQ;
+    delete m_d;
 }
 
 void IRCDDBApp::rptrQTH(const std::string& callsign, double latitude, double longitude, const std::string& desc1, const std::string& desc2, const std::string& infoURL)
@@ -279,14 +281,18 @@ IRCMessage *IRCDDBApp::getReplyMessage()
 
 void IRCDDBApp::startWork()
 {
-	m_d->m_terminateThread = false;
-	m_future = std::async(std::launch::async, &IRCDDBApp::Entry, this);
+    if (m_thread.joinable())
+        return;
+
+    m_d->m_terminateThread.store(false, std::memory_order_relaxed);
+    m_thread = std::thread(&IRCDDBApp::Entry, this);
 }
 
 void IRCDDBApp::stopWork()
 {
-    m_d->m_terminateThread = true;
-	m_future.get();
+    m_d->m_terminateThread.store(true, std::memory_order_relaxed);
+    if (m_thread.joinable())
+        m_thread.join();
 }
 
 unsigned int IRCDDBApp::calculateUsn(const std::string& nick)
@@ -996,7 +1002,7 @@ static bool needsDatabaseUpdate(int tableID)
 void IRCDDBApp::Entry()
 {
 	int sendlistTableID = 0;
-	while (!m_d->m_terminateThread) {
+	while (!m_d->m_terminateThread.load(std::memory_order_relaxed)) {
 		if (m_d->m_timer > 0)
 			m_d->m_timer--;
 		switch(m_d->m_state) {

IRCDDB/IRCDDBApp.h

@@ -25,7 +25,8 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #include "IRCApplication.h"
 
 #include <string>
-#include <future>
+#include <thread>
+#include <atomic>
 #include <ctime>
 #include <vector>
 
@@ -99,6 +100,7 @@ private:
 
 	IRCDDBAppPrivate *m_d;
 	time_t m_maxTime;
-	std::future<void> m_future;
+	std::thread m_thread;
+
 };
 

IRCDDB/IRCDDBClient.cpp

@@ -51,7 +51,7 @@ CIRCDDBClient::~CIRCDDBClient()
 	// A false return implies a network error, or unable to log in
 bool CIRCDDBClient::open()
 {
-	CLog::logInfo("start client and app\n");
+	CLog::logInfo("IRCDDB start client and app\n");
 	m_d->client->startWork();
 	m_d->m_app->startWork();
 	return true;

README.md

@@ -158,6 +158,7 @@ The testing framwework used is Google Test.
 
 # 6. Version History
 ## 6.1. Version 1.0
+- [**Improvement**] Improve threading handling ([#58](https://github.com/F4FXL/DStarGateway/issues/58))
 - [**Improvement**] Add Add an option to disable logging of ircddb traffic ([#59](https://github.com/F4FXL/DStarGateway/issues/59))
 - [**Bugfix**] Fix repeater not reverting to startup reflector after issueing a command through remote control. ([#57](https://github.com/F4FXL/DStarGateway/issues/57))
 - [**Bugfix**] Fix corrupted slow data leading to DV Text Message not being sent to ircddb. Thanks to Manfred DL1JM for all the testing. ([#55](https://github.com/F4FXL/DStarGateway/issues/55))

Tests/Log/logRepeat.cpp

@@ -52,6 +52,18 @@ namespace LogRepeatTests
         EXPECT_THAT(m_logTarget->m_messages[1].c_str(), EndsWith("[ERROR  ] Two Message\n"));
     }
 
+    TEST_F(LogRepeat, ThreeIdenticalMessageThreshold0) {
+        CLog::getRepeatThreshold() = 0U;
+        CLog::logError("One Message");
+        CLog::logError("One Message");
+        CLog::logError("One Message");
+
+        EXPECT_EQ(3, m_logTarget->m_messages.size()) << "There should be 3 messages in the log.";
+        EXPECT_THAT(m_logTarget->m_messages[0].c_str(), EndsWith("[ERROR  ] One Message\n"));
+        EXPECT_THAT(m_logTarget->m_messages[1].c_str(), EndsWith("[ERROR  ] One Message\n"));
+        EXPECT_THAT(m_logTarget->m_messages[2].c_str(), EndsWith("[ERROR  ] One Message\n"));
+    }
+
     TEST_F(LogRepeat, ThreeIdenticalMessageThreshold1) {
         CLog::getRepeatThreshold() = 1U;
         CLog::logError("One Message");
@@ -62,6 +74,34 @@ namespace LogRepeatTests
         EXPECT_THAT(m_logTarget->m_messages[0].c_str(), EndsWith("[ERROR  ] One Message\n"));
     }
 
+    TEST_F(LogRepeat, NineIdenticalMessageTwoDifferentThreshold0) {
+        CLog::getRepeatThreshold() = 0U;
+        CLog::logError("One Message");
+        CLog::logError("One Message");
+        CLog::logError("One Message");        
+        CLog::logError("One Message");
+        CLog::logError("One Message");       
+        CLog::logError("One Message");
+        CLog::logError("One Message");        
+        CLog::logError("One Message");
+        CLog::logError("One Message");
+        CLog::logError("Another Message");
+        CLog::logError("And here is another Message");
+
+        EXPECT_EQ(11, m_logTarget->m_messages.size()) << "There should be two message in the log.";
+        EXPECT_THAT(m_logTarget->m_messages[0].c_str(), EndsWith("[ERROR  ] One Message\n"));
+        EXPECT_THAT(m_logTarget->m_messages[1].c_str(), EndsWith("[ERROR  ] One Message\n"));
+        EXPECT_THAT(m_logTarget->m_messages[2].c_str(), EndsWith("[ERROR  ] One Message\n"));
+        EXPECT_THAT(m_logTarget->m_messages[3].c_str(), EndsWith("[ERROR  ] One Message\n"));
+        EXPECT_THAT(m_logTarget->m_messages[4].c_str(), EndsWith("[ERROR  ] One Message\n"));
+        EXPECT_THAT(m_logTarget->m_messages[5].c_str(), EndsWith("[ERROR  ] One Message\n"));
+        EXPECT_THAT(m_logTarget->m_messages[6].c_str(), EndsWith("[ERROR  ] One Message\n"));
+        EXPECT_THAT(m_logTarget->m_messages[7].c_str(), EndsWith("[ERROR  ] One Message\n"));
+        EXPECT_THAT(m_logTarget->m_messages[8].c_str(), EndsWith("[ERROR  ] One Message\n"));
+        EXPECT_THAT(m_logTarget->m_messages[9].c_str(), EndsWith("[ERROR  ] Another Message\n"));
+        EXPECT_THAT(m_logTarget->m_messages[10].c_str(), EndsWith("[ERROR  ] And here is another Message\n"));
+    }
+
     TEST_F(LogRepeat, NineIdenticalMessageTwoDifferentThreshold1) {
         CLog::getRepeatThreshold() = 1U;
         CLog::logError("One Message");