Merge branch 'master' of https://github.com/divyagayathri-hcl/sonic-s…

…wss-common into zmq

aminclude_static.am

@@ -0,0 +1,126 @@
+
+# aminclude_static.am generated automatically by Autoconf
+# from AX_AM_MACROS_STATIC on Thu Dec  5 08:01:58 UTC 2024
+
+
+# Code coverage
+#
+# Optional:
+#  - CODE_COVERAGE_DIRECTORY: Top-level directory for code coverage reporting.
+#    Multiple directories may be specified, separated by whitespace.
+#    (Default: $(top_builddir))
+#  - CODE_COVERAGE_OUTPUT_FILE: Filename and path for the .info file generated
+#    by lcov for code coverage. (Default:
+#    $(PACKAGE_NAME)-$(PACKAGE_VERSION)-coverage.info)
+#  - CODE_COVERAGE_OUTPUT_DIRECTORY: Directory for generated code coverage
+#    reports to be created. (Default:
+#    $(PACKAGE_NAME)-$(PACKAGE_VERSION)-coverage)
+#  - CODE_COVERAGE_BRANCH_COVERAGE: Set to 1 to enforce branch coverage,
+#    set to 0 to disable it and leave empty to stay with the default.
+#    (Default: empty)
+#  - CODE_COVERAGE_LCOV_SHOPTS_DEFAULT: Extra options shared between both lcov
+#    instances. (Default: based on )
+#  - CODE_COVERAGE_LCOV_SHOPTS: Extra options to shared between both lcov
+#    instances. (Default: )
+#  - CODE_COVERAGE_LCOV_OPTIONS_GCOVPATH: --gcov-tool pathtogcov
+#  - CODE_COVERAGE_LCOV_OPTIONS_DEFAULT: Extra options to pass to the
+#    collecting lcov instance. (Default: )
+#  - CODE_COVERAGE_LCOV_OPTIONS: Extra options to pass to the collecting lcov
+#    instance. (Default: )
+#  - CODE_COVERAGE_LCOV_RMOPTS_DEFAULT: Extra options to pass to the filtering
+#    lcov instance. (Default: empty)
+#  - CODE_COVERAGE_LCOV_RMOPTS: Extra options to pass to the filtering lcov
+#    instance. (Default: )
+#  - CODE_COVERAGE_GENHTML_OPTIONS_DEFAULT: Extra options to pass to the
+#    genhtml instance. (Default: based on )
+#  - CODE_COVERAGE_GENHTML_OPTIONS: Extra options to pass to the genhtml
+#    instance. (Default: )
+#  - CODE_COVERAGE_IGNORE_PATTERN: Extra glob pattern of files to ignore
+#
+# The generated report will be titled using the $(PACKAGE_NAME) and
+# $(PACKAGE_VERSION). In order to add the current git hash to the title,
+# use the git-version-gen script, available online.
+# Optional variables
+# run only on top dir
+if CODE_COVERAGE_ENABLED
+ ifeq ($(abs_builddir), $(abs_top_builddir))
+CODE_COVERAGE_DIRECTORY ?= $(top_builddir)
+CODE_COVERAGE_OUTPUT_FILE ?= $(PACKAGE_NAME)-$(PACKAGE_VERSION)-coverage.info
+CODE_COVERAGE_OUTPUT_DIRECTORY ?= $(PACKAGE_NAME)-$(PACKAGE_VERSION)-coverage
+
+CODE_COVERAGE_BRANCH_COVERAGE ?=
+CODE_COVERAGE_LCOV_SHOPTS_DEFAULT ?= $(if $(CODE_COVERAGE_BRANCH_COVERAGE),--rc lcov_branch_coverage=$(CODE_COVERAGE_BRANCH_COVERAGE))
+CODE_COVERAGE_LCOV_SHOPTS ?= $(CODE_COVERAGE_LCOV_SHOPTS_DEFAULT)
+CODE_COVERAGE_LCOV_OPTIONS_GCOVPATH ?= --gcov-tool "$(GCOV)"
+CODE_COVERAGE_LCOV_OPTIONS_DEFAULT ?= $(CODE_COVERAGE_LCOV_OPTIONS_GCOVPATH)
+CODE_COVERAGE_LCOV_OPTIONS ?= $(CODE_COVERAGE_LCOV_OPTIONS_DEFAULT)
+CODE_COVERAGE_LCOV_RMOPTS_DEFAULT ?=
+CODE_COVERAGE_LCOV_RMOPTS ?= $(CODE_COVERAGE_LCOV_RMOPTS_DEFAULT)
+CODE_COVERAGE_GENHTML_OPTIONS_DEFAULT ?=$(if $(CODE_COVERAGE_BRANCH_COVERAGE),--rc genhtml_branch_coverage=$(CODE_COVERAGE_BRANCH_COVERAGE))
+CODE_COVERAGE_GENHTML_OPTIONS ?= $(CODE_COVERAGE_GENHTML_OPTIONS_DEFAULT)
+CODE_COVERAGE_IGNORE_PATTERN ?=
+
+GITIGNOREFILES = $(GITIGNOREFILES) $(CODE_COVERAGE_OUTPUT_FILE) $(CODE_COVERAGE_OUTPUT_DIRECTORY)
+code_coverage_v_lcov_cap = $(code_coverage_v_lcov_cap_$(V))
+code_coverage_v_lcov_cap_ = $(code_coverage_v_lcov_cap_$(AM_DEFAULT_VERBOSITY))
+code_coverage_v_lcov_cap_0 = @echo "  LCOV   --capture" $(CODE_COVERAGE_OUTPUT_FILE);
+code_coverage_v_lcov_ign = $(code_coverage_v_lcov_ign_$(V))
+code_coverage_v_lcov_ign_ = $(code_coverage_v_lcov_ign_$(AM_DEFAULT_VERBOSITY))
+code_coverage_v_lcov_ign_0 = @echo "  LCOV   --remove /tmp/*" $(CODE_COVERAGE_IGNORE_PATTERN);
+code_coverage_v_genhtml = $(code_coverage_v_genhtml_$(V))
+code_coverage_v_genhtml_ = $(code_coverage_v_genhtml_$(AM_DEFAULT_VERBOSITY))
+code_coverage_v_genhtml_0 = @echo "  GEN   " "$(CODE_COVERAGE_OUTPUT_DIRECTORY)";
+code_coverage_quiet = $(code_coverage_quiet_$(V))
+code_coverage_quiet_ = $(code_coverage_quiet_$(AM_DEFAULT_VERBOSITY))
+code_coverage_quiet_0 = --quiet
+
+# sanitizes the test-name: replaces with underscores: dashes and dots
+code_coverage_sanitize = $(subst -,_,$(subst .,_,$(1)))
+
+# Use recursive makes in order to ignore errors during check
+check-code-coverage:
+	-$(AM_V_at)$(MAKE) $(AM_MAKEFLAGS) -k check
+	$(AM_V_at)$(MAKE) $(AM_MAKEFLAGS) code-coverage-capture
+
+# Capture code coverage data
+code-coverage-capture: code-coverage-capture-hook
+	$(code_coverage_v_lcov_cap)$(LCOV) $(code_coverage_quiet) $(addprefix --directory ,$(CODE_COVERAGE_DIRECTORY)) --capture --output-file "$(CODE_COVERAGE_OUTPUT_FILE).tmp" --test-name "$(call code_coverage_sanitize,$(PACKAGE_NAME)-$(PACKAGE_VERSION))" --no-checksum --compat-libtool $(CODE_COVERAGE_LCOV_SHOPTS) $(CODE_COVERAGE_LCOV_OPTIONS)
+	$(code_coverage_v_lcov_ign)$(LCOV) $(code_coverage_quiet) $(addprefix --directory ,$(CODE_COVERAGE_DIRECTORY)) --remove "$(CODE_COVERAGE_OUTPUT_FILE).tmp" "/tmp/*" $(CODE_COVERAGE_IGNORE_PATTERN) --output-file "$(CODE_COVERAGE_OUTPUT_FILE)" $(CODE_COVERAGE_LCOV_SHOPTS) $(CODE_COVERAGE_LCOV_RMOPTS)
+	-@rm -f "$(CODE_COVERAGE_OUTPUT_FILE).tmp"
+	$(code_coverage_v_genhtml)LANG=C $(GENHTML) $(code_coverage_quiet) $(addprefix --prefix ,$(CODE_COVERAGE_DIRECTORY)) --output-directory "$(CODE_COVERAGE_OUTPUT_DIRECTORY)" --title "$(PACKAGE_NAME)-$(PACKAGE_VERSION) Code Coverage" --legend --show-details "$(CODE_COVERAGE_OUTPUT_FILE)" $(CODE_COVERAGE_GENHTML_OPTIONS)
+	@echo "file://$(abs_builddir)/$(CODE_COVERAGE_OUTPUT_DIRECTORY)/index.html"
+
+code-coverage-clean:
+	-$(LCOV) --directory $(top_builddir) -z
+	-rm -rf "$(CODE_COVERAGE_OUTPUT_FILE)" "$(CODE_COVERAGE_OUTPUT_FILE).tmp" "$(CODE_COVERAGE_OUTPUT_DIRECTORY)"
+	-find . \( -name "*.gcda" -o -name "*.gcno" -o -name "*.gcov" \) -delete
+
+code-coverage-dist-clean:
+
+AM_DISTCHECK_CONFIGURE_FLAGS = $(AM_DISTCHECK_CONFIGURE_FLAGS) --disable-code-coverage
+ else # ifneq ($(abs_builddir), $(abs_top_builddir))
+check-code-coverage:
+
+code-coverage-capture: code-coverage-capture-hook
+
+code-coverage-clean:
+
+code-coverage-dist-clean:
+ endif # ifeq ($(abs_builddir), $(abs_top_builddir))
+else #! CODE_COVERAGE_ENABLED
+# Use recursive makes in order to ignore errors during check
+check-code-coverage:
+	@echo "Need to reconfigure with --enable-code-coverage"
+# Capture code coverage data
+code-coverage-capture: code-coverage-capture-hook
+	@echo "Need to reconfigure with --enable-code-coverage"
+
+code-coverage-clean:
+
+code-coverage-dist-clean:
+
+endif #CODE_COVERAGE_ENABLED
+# Hook rule executed before code-coverage-capture, overridable by the user
+code-coverage-capture-hook:
+
+.PHONY: check-code-coverage code-coverage-capture code-coverage-dist-clean code-coverage-clean code-coverage-capture-hook

common/redispipeline.h

@@ -164,7 +164,7 @@ class RedisPipeline {
             return;
 
         m_channels.insert(channel);
-        m_luaPub += "redis.call('PUBLISH', '" + channel + "', 'G');";
+        m_luaPub += "redis.call('PUBLISH', '" + channel + "', 'G')\n";
         m_shaPub = loadRedisScript(m_luaPub);
     }
 

common/zmqclient.cpp

@@ -25,6 +25,13 @@ ZmqClient::ZmqClient(const std::string& endpoint, const std::string& vrf)
     initialize(endpoint, vrf);
 }
 
+ZmqClient::ZmqClient(const std::string& endpoint, uint32_t waitTimeMs) :
+    m_waitTimeMs(waitTimeMs)
+{
+//    m_waitTimeMs = waitTimeMs;
+    initialize(endpoint);
+}
+
 ZmqClient::~ZmqClient()
 {
     std::lock_guard<std::mutex> lock(m_socketMutex);
@@ -55,6 +62,17 @@ void ZmqClient::initialize(const std::string& endpoint, const std::string& vrf)
 
     connect();
 }
+
+void ZmqClient::initialize(const std::string& endpoint)
+{
+    m_connected = false;
+    m_endpoint = endpoint;
+    m_context = nullptr;
+    m_socket = nullptr;
+    m_sendbuffer.resize(MQ_RESPONSE_MAX_COUNT);
+
+    connect();
+}
 
 bool ZmqClient::isConnected()
 {
@@ -137,7 +155,7 @@ void ZmqClient::sendMsg(
     int zmq_err = 0;
     int retry_delay = 10;
     int rc = 0;
-    for (int i = 0; i <=  MQ_MAX_RETRY; ++i)
+    for (int i = 0; i <= MQ_MAX_RETRY; ++i)
     {
         {
             // ZMQ socket is not thread safe: http://api.zeromq.org/2-1:zmq
@@ -202,8 +220,32 @@ bool ZmqClient::wait(std::string& dbName,
                      std::vector<std::shared_ptr<KeyOpFieldsValuesTuple>>& kcos)
 {
     SWSS_LOG_ENTER();
+
+    zmq_pollitem_t items [1] = { };
+    items[0].socket = m_socket;
+    items[0].events = ZMQ_POLLIN;
+
     int rc;
-    for (int i = 0; true ; ++i)
+    for (int i = 0; true; ++i)
+    {
+        rc = zmq_poll(items, 1, (int)m_waitTimeMs);
+        if (rc == 0)
+        {
+            SWSS_LOG_ERROR("zmq_poll timed out");
+            return false;
+        }
+        if (rc > 0)
+        {
+            break;
+        }
+        if (zmq_errno() == EINTR && i <= MQ_MAX_RETRY)
+        {
+            continue;
+        }
+        SWSS_LOG_ERROR("zmq_poll failed, zmqerrno: %d", zmq_errno());
+    }
+
+    for (int i = 0; true; ++i)
     {
         rc = zmq_recv(m_socket, m_sendbuffer.data(), m_sendbuffer.size(), 0);
         if (rc < 0)
@@ -212,13 +254,14 @@ bool ZmqClient::wait(std::string& dbName,
             {
                 continue;
             }
-            SWSS_LOG_THROW("zmq_recv failed, zmqerrno: %d", zmq_errno());
+            SWSS_LOG_ERROR("zmq_recv failed, zmqerrno: %d", zmq_errno());
         }
         if (rc >= (int)m_sendbuffer.size())
         {
-            SWSS_LOG_THROW(
+            SWSS_LOG_ERROR(
                 "zmq_recv message was truncated (over %d bytes, received %d), increase buffer size, message DROPPED",
                 (int)m_sendbuffer.size(), rc);
+            return false;
         }
         break;
     }

common/zmqclient.h

@@ -15,6 +15,7 @@ class ZmqClient
 
     ZmqClient(const std::string& endpoint);
     ZmqClient(const std::string& endpoint, const std::string& vrf);
+    ZmqClient(const std::string& endpoint, uint32_t waitTimeMs);
     ~ZmqClient();
 
     bool isConnected();
@@ -31,6 +32,7 @@ class ZmqClient
 
 private:
     void initialize(const std::string& endpoint, const std::string& vrf);
+    void initialize(const std::string& endpoint);
 
     std::string m_endpoint;
 
@@ -42,6 +44,8 @@ class ZmqClient
 
     bool m_connected;
 
+    uint32_t m_waitTimeMs;
+
     std::mutex m_socketMutex;
 
     std::vector<char> m_sendbuffer;