Add thread safe guard to throw exception when share DBConnection cross threads

common/Makefile.am

@@ -69,7 +69,8 @@ libswsscommon_la_SOURCES = \
     warm_restart.cpp          \
     luatable.cpp              \
     countertable.cpp          \
-    redisutility.cpp
+    redisutility.cpp          \
+    threadsafeguard.cpp
 
 libswsscommon_la_CXXFLAGS = $(DBGFLAGS) $(AM_CFLAGS) $(CFLAGS_COMMON) $(LIBNL_CFLAGS) $(CODE_COVERAGE_CXXFLAGS)
 libswsscommon_la_CPPFLAGS = $(DBGFLAGS) $(AM_CFLAGS) $(CFLAGS_COMMON) $(LIBNL_CPPFLAGS) $(CODE_COVERAGE_CPPFLAGS)

common/dbconnector.cpp

@@ -410,10 +410,12 @@ RedisContext::~RedisContext()
 
 RedisContext::RedisContext()
     : m_conn(NULL)
+    , m_running(false)
 {
 }
 
 RedisContext::RedisContext(const RedisContext &other)
+    : m_running(false)
 {
     auto octx = other.getContext();
     const char *unixPath = octx->unix_sock.path;
@@ -478,6 +480,11 @@ void RedisContext::setClientName(const string& clientName)
     r.checkStatusOK();
 }
 
+std::atomic<bool> &RedisContext::getRunningFlag()
+{
+    return m_running;
+}
+
 string RedisContext::getClientName()
 {
     string command("CLIENT GETNAME");

common/dbconnector.h

@@ -7,10 +7,12 @@
 #include <utility>
 #include <memory>
 #include <mutex>
+#include <atomic>
 
 #include <hiredis/hiredis.h>
 #include "rediscommand.h"
 #include "redisreply.h"
+#include "threadsafeguard.h"
 #define EMPTY_NAMESPACE std::string()
 
 namespace swss {
@@ -127,6 +129,8 @@ class RedisContext
 
     std::string getClientName();
 
+    std::atomic<bool> &getRunningFlag();
+
 protected:
     RedisContext();
     void initContext(const char *host, int port, const timeval *tv);
@@ -135,6 +139,11 @@ class RedisContext
 
 private:
     redisContext *m_conn;
+
+    /*
+     * Use this flag and ThreadSafeGuard to check and throw exception when multiple thread run command on same redis context.
+     */
+    std::atomic<bool> m_running;
 };
 
 class DBConnector : public RedisContext

common/redisreply.cpp

@@ -33,6 +33,7 @@ inline void guard(FUNC func, const char* command)
 
 RedisReply::RedisReply(RedisContext *ctx, const RedisCommand& command)
 {
+    ThreadSafeGuard safeguard(ctx->getRunningFlag(), string(command.c_str()));
     int rc = redisAppendFormattedCommand(ctx->getContext(), command.c_str(), command.length());
     if (rc != REDIS_OK)
     {
@@ -51,6 +52,7 @@ RedisReply::RedisReply(RedisContext *ctx, const RedisCommand& command)
 
 RedisReply::RedisReply(RedisContext *ctx, const string& command)
 {
+    ThreadSafeGuard safeguard(ctx->getRunningFlag(), command);
     int rc = redisAppendCommand(ctx->getContext(), command.c_str());
     if (rc != REDIS_OK)
     {

common/threadsafeguard.cpp

@@ -0,0 +1,26 @@
+#include <iostream>
+#include <sstream>
+#include <system_error>
+
+#include "common/logger.h"
+#include "common/threadsafeguard.h"
+
+using namespace std;
+using namespace swss;
+
+ThreadSafeGuard::ThreadSafeGuard(atomic<bool> &running, const string &info)
+:m_running(running)
+{
+    auto currentlyRunning = m_running.exchange(true);
+    if (currentlyRunning)
+    {
+        string errmsg = "Current thread '" + info + "' conflict with other thread.";
+        SWSS_LOG_ERROR("%s", errmsg.c_str());
+        throw system_error(make_error_code(errc::operation_in_progress), errmsg.c_str());
+    }
+}
+
+ThreadSafeGuard::~ThreadSafeGuard()
+{
+    m_running = false;
+}

common/threadsafeguard.h

@@ -0,0 +1,15 @@
+#pragma once
+
+namespace swss {
+
+class ThreadSafeGuard
+{
+public:
+    ThreadSafeGuard(std::atomic<bool> &running, const std::string &info);
+    ~ThreadSafeGuard();
+
+private:
+    std::atomic<bool> &m_running;
+};
+
+}

tests/redis_ut.cpp

@@ -1057,3 +1057,21 @@ TEST(Connector, hmset)
     // test empty multi hash
     db.hmset({});
 }
+
+TEST(Connector, ThreadSafeGuard)
+{
+    DBConnector db("TEST_DB", 0, true);
+
+    string exceptionMessage;
+    try
+    {
+        ThreadSafeGuard safeguard1(db.getRunningFlag(), "command 1");
+        ThreadSafeGuard safeguard2(db.getRunningFlag(), "command 2");
+    }
+    catch (const system_error& ex)
+    {
+        exceptionMessage = ex.what();
+    }
+
+    ASSERT_EQ(exceptionMessage, "Current thread 'command 2' conflict with other thread.: Operation now in progress");
+}