Sequencer class with lambda response message

meta/RedisSelectableChannel.cpp

@@ -1,18 +1,37 @@
 #include "RedisSelectableChannel.h"
 
 #include "swss/logger.h"
+#include "syncd/Logger.h"
 
 using namespace sairedis;
 
+#define MY_LOCK() \
+if(m_protected) \
+{ \
+    LogToModuleFile("1", "before MY_LOCK()"); \
+    m_mutex->lock(); \
+    LogToModuleFile("1", "after MY_LOCK()"); \
+} 
+
+#define MY_UNLOCK() \
+if(m_protected) \
+{ \
+    LogToModuleFile("1", "before MY_UNLOCK()"); \
+    m_mutex->unlock(); \
+    LogToModuleFile("1", "after MY_UNLOCK()"); \
+}
+
 RedisSelectableChannel::RedisSelectableChannel(
         _In_ std::shared_ptr<swss::DBConnector> dbAsic,
         _In_ const std::string& asicStateTable,
         _In_ const std::string& getResponseTable,
         _In_ const std::string& tempPrefix,
-        _In_ bool modifyRedis):
+        _In_ bool modifyRedis,
+        _In_ std::shared_ptr<std::mutex> t_mutex):
     m_dbAsic(dbAsic),
     m_tempPrefix(tempPrefix),
-    m_modifyRedis(modifyRedis)
+    m_modifyRedis(modifyRedis),
+    m_mutex(t_mutex)
 {
     SWSS_LOG_ENTER();
 
@@ -29,13 +48,23 @@ RedisSelectableChannel::RedisSelectableChannel(
     m_getResponse = std::make_shared<swss::ProducerTable>(m_dbAsic.get(), getResponseTable);
 
     SWSS_LOG_NOTICE("opened redis channel");
+    if(t_mutex != NULL)
+    {
+        m_protected = true;
+    }
+    else
+    {
+        m_protected = false;
+    }
 }
 
 bool RedisSelectableChannel::empty()
 {
     SWSS_LOG_ENTER();
-
-    return m_asicState->empty();
+    MY_LOCK();
+    bool ans = m_asicState->empty();
+    MY_UNLOCK();
+    return ans;
 }
 
 void RedisSelectableChannel::set(
@@ -44,16 +73,17 @@ void RedisSelectableChannel::set(
         _In_ const std::string& op)
 {
     SWSS_LOG_ENTER();
-
+    MY_LOCK();
     m_getResponse->set(key, values, op);
+    MY_UNLOCK();
 }
 
 void RedisSelectableChannel::pop(
         _Out_ swss::KeyOpFieldsValuesTuple& kco,
         _In_ bool initViewMode)
 {
     SWSS_LOG_ENTER();
-
+    MY_LOCK();
     if (initViewMode)
     {
         m_asicState->pop(kco, m_tempPrefix);
@@ -62,55 +92,69 @@ void RedisSelectableChannel::pop(
     {
         m_asicState->pop(kco);
     }
+    MY_UNLOCK();
 }
 
 // Selectable overrides
 
 int RedisSelectableChannel::getFd()
 {
     SWSS_LOG_ENTER();
-
-    return m_asicState->getFd();
+    MY_LOCK();
+    int ans = m_asicState->getFd();
+    MY_UNLOCK();
+    return ans;
 }
 
 uint64_t RedisSelectableChannel::readData()
 {
     SWSS_LOG_ENTER();
-
-    return m_asicState->readData();
+    MY_LOCK();
+    uint64_t ans = m_asicState->readData();
+    MY_UNLOCK();
+    return ans;
 }
 
 bool RedisSelectableChannel::hasData()
 {
     SWSS_LOG_ENTER();
-
-    return m_asicState->hasData();
+    MY_LOCK();
+    bool ans = m_asicState->hasData();
+    MY_UNLOCK();
+    return ans;
 }
 
 bool RedisSelectableChannel::hasCachedData()
 {
     SWSS_LOG_ENTER();
-
-    return m_asicState->hasCachedData();
+    MY_LOCK();
+    bool ans = m_asicState->hasCachedData();
+    MY_UNLOCK();
+    return ans;
 }
 
 bool RedisSelectableChannel::initializedWithData()
 {
     SWSS_LOG_ENTER();
-
-    return m_asicState->initializedWithData();
+    MY_LOCK();
+    bool ans = m_asicState->initializedWithData();
+    MY_UNLOCK();
+    return ans;
 }
 
 void RedisSelectableChannel::updateAfterRead()
 {
     SWSS_LOG_ENTER();
-
-    return m_asicState->updateAfterRead();
+    MY_LOCK();
+    m_asicState->updateAfterRead();
+    MY_UNLOCK();
 }
 
 int RedisSelectableChannel::getPri() const
 {
     SWSS_LOG_ENTER();
-
-    return m_asicState->getPri();
+    MY_LOCK();
+    auto ans = m_asicState->getPri();
+    MY_UNLOCK();
+    return ans;
 }

meta/RedisSelectableChannel.h

@@ -4,6 +4,7 @@
 
 #include "swss/consumertable.h"
 #include "swss/producertable.h"
+#include <mutex>
 
 namespace sairedis
 {
@@ -17,7 +18,8 @@ namespace sairedis
                     _In_ const std::string& asicStateTable,
                     _In_ const std::string& getResponseTable,
                     _In_ const std::string& tempPrefix,
-                    _In_ bool modifyRedis);
+                    _In_ bool modifyRedis,
+                    _In_ std::shared_ptr<std::mutex> mutex = nullptr);
 
             virtual ~RedisSelectableChannel() = default;
 
@@ -61,5 +63,9 @@ namespace sairedis
             std::string m_tempPrefix;
 
             bool m_modifyRedis;
+
+            bool m_protected;
+
+            std::shared_ptr<std::mutex> m_mutex;
     };
 }

syncd/Logger.h

@@ -0,0 +1,108 @@
+#ifndef LOGGER_H
+#define LOGGER_H
+
+#define MODULE_NAME     "syncd"
+
+#include <string>
+#include <iostream>
+#include <fstream>
+#include <sys/stat.h>
+#include <chrono>
+#include <cstdio>
+#include <mutex>
+#include <sstream>
+#include <iomanip>
+#include "fmt/format.h"
+#include <ctime>
+#include <thread>
+
+#define MAX_LOG_SIZE    (10 *50 * 1024) /* 50 KB */
+#define ENABLE_LOGGING  1
+
+// Define a mutex for thread safety
+static std::mutex logMutex;
+
+// Improved logging function with thread safety
+static void writeToLogFile(const std::string& funcName, const std::string& fileNum, const std::string& message) {
+    // Lock the mutex to ensure thread safety
+    std::lock_guard<std::mutex> lock(logMutex);
+
+    std::string filePath = "/" + std::string(MODULE_NAME) + "_debugLog_" + fileNum + ".txt";
+    std::string backupFilePath = filePath + ".history";
+    struct stat fileStat;
+
+    // Check if the log file exists and if its size exceeds the maximum limit
+    if (stat(filePath.c_str(), &fileStat) == 0) {
+        if (fileStat.st_size > MAX_LOG_SIZE) {
+            // Remove the old backup file
+            std::remove(backupFilePath.c_str());
+            // Rename the current log file to the backup file
+            if (std::rename(filePath.c_str(), backupFilePath.c_str()) != 0) {
+                std::cerr << "Error: Could not rename file " << filePath << " to " << backupFilePath << std::endl;
+                return;
+            }
+        }
+    }
+
+    // Open the log file in append mode
+    std::ofstream logFile(filePath, std::ios_base::app);
+    if (!logFile.is_open()) {
+        std::cerr << "Error: Could not open file " << filePath << std::endl;
+        return;
+    }
+
+    auto now = std::chrono::system_clock::now();
+    auto in_time_t = std::chrono::system_clock::to_time_t(now);
+    auto milliseconds = std::chrono::duration_cast<std::chrono::milliseconds>(now.time_since_epoch()) % 1000;
+
+    std::ostringstream oss;
+    oss << std::put_time(std::localtime(&in_time_t), "%Y-%m-%d %H:%M:%S")
+        << '.' << std::setw(3) << std::setfill('0') << milliseconds.count();
+    std::string formatted_time = oss.str();
+
+    std::thread::id this_id = std::this_thread::get_id();
+
+    // Write the timestamp, function name, and message to the log file
+    logFile << formatted_time << " V4 " << this_id << " " << funcName << ": " << message << std::endl;
+
+
+    logFile.close();
+}
+
+template<typename... Args>
+static void logFormattedMessage(const std::string& funcName, const std::string& fileNum, const std::string& format, Args... messageArgs) {
+    std::ostringstream oss;
+
+    std::string remainingFormat = format;
+
+    // Helper function to process a single argument
+    auto processArg = [&oss, &remainingFormat](const auto& arg) {
+        size_t pos = remainingFormat.find("{}");
+        if (pos != std::string::npos) {
+            oss << remainingFormat.substr(0, pos) << arg;
+            remainingFormat = remainingFormat.substr(pos + 2);
+        }
+    };
+
+    // Helper function to recursively process all arguments
+    auto processAllArgs = [&](auto&&... innerArgs) -> void {
+        (void)std::initializer_list<int>{(processArg(innerArgs), 0)...};
+    };
+
+    // Process all arguments
+    processAllArgs(messageArgs...);
+
+    // Add any remaining format string
+    oss << remainingFormat;
+
+    // Write the full message to the log file
+    writeToLogFile(funcName, fileNum, oss.str());
+}
+#if ENABLE_LOGGING
+// Macro for easy logging with formatting
+#define LogToModuleFile(fileNum, format, ...) logFormattedMessage(__func__, fileNum, format, ##__VA_ARGS__)
+#else
+#define LogToModuleFile(fileNum, format, ...)
+#endif
+
+#endif // LOGGER_H

syncd/Makefile.am

@@ -56,7 +56,8 @@ libSyncd_a_SOURCES = \
 				WatchdogScope.cpp \
 				Workaround.cpp \
 				ZeroMQNotificationProducer.cpp \
-				syncd_main.cpp
+				syncd_main.cpp \
+				Sequencer.cpp
 
 libSyncd_a_CPPFLAGS = $(CODE_COVERAGE_CPPFLAGS)
 libSyncd_a_CXXFLAGS = $(DBGFLAGS) $(AM_CXXFLAGS) $(CXXFLAGS_COMMON) $(CODE_COVERAGE_CXXFLAGS)