Add support for default zmq synchronous mode flag (sonic-net#711)

Add support for default zmq synchronous mode flag

lib/inc/RedisRemoteSaiInterface.h

@@ -445,6 +445,8 @@ namespace sairedis
 
             bool m_syncMode;
 
+            sai_redis_communication_mode_t m_redisCommunicationMode;
+
             bool m_initialized;
 
             std::shared_ptr<Recorder> m_recorder;

lib/inc/sairedis.h

@@ -23,6 +23,36 @@ typedef enum _sai_redis_notify_syncd_t
 
 } sai_redis_notify_syncd_t;
 
+typedef enum _sai_redis_communication_mode_t
+{
+    /**
+     * @brief Asynchronous mode using Redis DB.
+     */
+    SAI_REDIS_COMMUNICATION_MODE_REDIS_ASYNC,
+
+    /**
+     * @brief Synchronous mode using Redis DB.
+     */
+    SAI_REDIS_COMMUNICATION_MODE_REDIS_SYNC,
+
+    /**
+     * @brief Synchronous mode using ZMQ library.
+     *
+     * When enabled syncd also needs to be running in zmq synchronous mode.
+     * Command pipeline will be disabled when this flag will be set to true.
+     *
+     * This attribute is only introduced to help kick start using synchronous
+     * mode with zmq library. This mode requires some additional configuration
+     * like main channel string and notification channel string. When using
+     * this attribute those channels are set to default values:
+     * "ipc:///tmp/zmq_ep" and "ipc:///tmp/zmq_ntf_ep". To take control of
+     * those values a context config json file must be provided via
+     * SAI_REDIS_KEY_CONTEXT_CONFIG profile argument.
+     */
+    SAI_REDIS_COMMUNICATION_MODE_ZMQ_SYNC,
+
+} sai_redis_communication_mode_t;
+
 typedef enum _sai_redis_switch_attr_t
 {
     /**
@@ -113,12 +143,28 @@ typedef enum _sai_redis_switch_attr_t
      * running in synchronous mode. Command pipeline will be disabled when this
      * flag will be set to true.
      *
+     * NOTE: This attribute is depreacated by
+     * SAI_REDIS_SWITCH_ATTR_REDIS_COMMUNICATION_MODE.  When set to true it
+     * will set SAI_REDIS_SWITCH_ATTR_REDIS_COMMUNICATION_MODE to
+     * SAI_REDIS_COMMUNICATION_MODE_REDIS_SYNC.
+     *
+     * TODO: remove this attribute.
+     *
      * @type bool
      * @flags CREATE_AND_SET
      * @default false
      */
     SAI_REDIS_SWITCH_ATTR_SYNC_MODE,
 
+    /**
+     * @brief Redis communication mode.
+     *
+     * @type sai_redis_communication_mode_t
+     * @flags CREATE_AND_SET
+     * @default SAI_REDIS_COMMUNICATION_MODE_REDIS_ASYNC
+     */
+    SAI_REDIS_SWITCH_ATTR_REDIS_COMMUNICATION_MODE,
+
     /**
      * @brief Record statistics counters API calls.
      *

lib/src/ContextConfig.cpp

@@ -17,7 +17,9 @@ ContextConfig::ContextConfig(
     m_dbCounters(dbCounters),
     m_dbFlex(dbFlex),
     m_dbState(dbState),
-    m_zmqEnable(false)
+    m_zmqEnable(false),
+    m_zmqEndpoint("ipc:///tmp/zmq_ep"),
+    m_zmqNtfEndpoint("ipc:///tmp/zmq_ntf_ep")
 {
     SWSS_LOG_ENTER();
 
@@ -38,4 +40,3 @@ void ContextConfig::insert(
 
     m_scc->insert(config);
 }
-

lib/src/RedisRemoteSaiInterface.cpp

@@ -33,6 +33,7 @@ RedisRemoteSaiInterface::RedisRemoteSaiInterface(
         _In_ std::function<sai_switch_notifications_t(std::shared_ptr<Notification>)> notificationCallback,
         _In_ std::shared_ptr<Recorder> recorder):
     m_contextConfig(contextConfig),
+    m_redisCommunicationMode(SAI_REDIS_COMMUNICATION_MODE_REDIS_ASYNC),
     m_recorder(recorder),
     m_notificationCallback(notificationCallback)
 {
@@ -71,6 +72,7 @@ sai_status_t RedisRemoteSaiInterface::initialize(
     m_asicInitViewMode = false; // default mode is apply mode
     m_useTempView = false;
     m_syncMode = false;
+    m_redisCommunicationMode = SAI_REDIS_COMMUNICATION_MODE_REDIS_ASYNC;
 
     if (m_contextConfig->m_zmqEnable)
     {
@@ -355,6 +357,8 @@ sai_status_t RedisRemoteSaiInterface::setRedisExtensionAttribute(
 
         case SAI_REDIS_SWITCH_ATTR_SYNC_MODE:
 
+            SWSS_LOG_WARN("sync mode is depreacated, use communication mode");
+
             m_syncMode = attr->value.booldata;
 
             if (m_contextConfig->m_zmqEnable)
@@ -373,6 +377,78 @@ sai_status_t RedisRemoteSaiInterface::setRedisExtensionAttribute(
 
             return SAI_STATUS_SUCCESS;
 
+        case SAI_REDIS_SWITCH_ATTR_REDIS_COMMUNICATION_MODE:
+
+            m_redisCommunicationMode = (sai_redis_communication_mode_t)attr->value.s32;
+
+            if (m_contextConfig->m_zmqEnable)
+            {
+                SWSS_LOG_NOTICE("zmq enabled via context config");
+
+                m_redisCommunicationMode = SAI_REDIS_COMMUNICATION_MODE_ZMQ_SYNC;
+            }
+
+            m_communicationChannel = nullptr;
+
+            switch (m_redisCommunicationMode)
+            {
+                case SAI_REDIS_COMMUNICATION_MODE_REDIS_ASYNC:
+
+                    SWSS_LOG_NOTICE("enabling redis async mode");
+
+                    m_syncMode = false;
+
+                    m_communicationChannel = std::make_shared<RedisChannel>(
+                            m_contextConfig->m_dbAsic,
+                            std::bind(&RedisRemoteSaiInterface::handleNotification, this, _1, _2, _3));
+
+                    m_communicationChannel->setBuffered(true);
+
+                    return SAI_STATUS_SUCCESS;
+
+                case SAI_REDIS_COMMUNICATION_MODE_REDIS_SYNC:
+
+                    SWSS_LOG_NOTICE("enabling redis sync mode");
+
+                    m_syncMode = true;
+
+                    m_communicationChannel = std::make_shared<RedisChannel>(
+                            m_contextConfig->m_dbAsic,
+                            std::bind(&RedisRemoteSaiInterface::handleNotification, this, _1, _2, _3));
+
+                    m_communicationChannel->setBuffered(false);
+
+                    return SAI_STATUS_SUCCESS;
+
+                case SAI_REDIS_COMMUNICATION_MODE_ZMQ_SYNC:
+
+                    m_contextConfig->m_zmqEnable = true;
+
+                    // main communication channel was created at initialize method
+                    // so this command will replace it with zmq channel
+
+                    m_communicationChannel = std::make_shared<ZeroMQChannel>(
+                            m_contextConfig->m_zmqEndpoint,
+                            m_contextConfig->m_zmqNtfEndpoint,
+                            std::bind(&RedisRemoteSaiInterface::handleNotification, this, _1, _2, _3));
+
+                    SWSS_LOG_NOTICE("zmq enabled, forcing sync mode");
+
+                    m_syncMode = true;
+
+                    SWSS_LOG_NOTICE("disabling buffered pipeline in sync mode");
+
+                    m_communicationChannel->setBuffered(false);
+
+                    return SAI_STATUS_SUCCESS;
+
+                default:
+
+                    SWSS_LOG_ERROR("invalid communication mode value: %d", m_redisCommunicationMode);
+
+                    return SAI_STATUS_NOT_SUPPORTED;
+            }
+
         case SAI_REDIS_SWITCH_ATTR_USE_PIPELINE:
 
             if (m_syncMode)
@@ -657,12 +733,22 @@ sai_status_t RedisRemoteSaiInterface::waitForGetResponse(
 
     if (status == SAI_STATUS_SUCCESS)
     {
+        if (values.size() == 0)
+        {
+            SWSS_LOG_THROW("logic error, get response returned 0 values!, send api response or sync/async issue?");
+        }
+
         SaiAttributeList list(objectType, values, false);
 
         transfer_attributes(objectType, attr_count, list.get_attr_list(), attr_list, false);
     }
     else if (status == SAI_STATUS_BUFFER_OVERFLOW)
     {
+        if (values.size() == 0)
+        {
+            SWSS_LOG_THROW("logic error, get response returned 0 values!, send api response or sync/async issue?");
+        }
+
         SaiAttributeList list(objectType, values, true);
 
         // no need for id fix since this is overflow

meta/sai_serialize.h

@@ -206,6 +206,9 @@ std::string sai_serialize_queue_deadlock_ntf(
 std::string sai_serialize(
         _In_ const sai_redis_notify_syncd_t& value);
 
+std::string sai_serialize_redis_communication_mode(
+        _In_ sai_redis_communication_mode_t value);
+
 // deserialize
 
 void sai_deserialize_enum(
@@ -363,4 +366,8 @@ void sai_deserialize(
 sai_redis_notify_syncd_t sai_deserialize_redis_notify_syncd(
         _In_ const std::string& s);
 
+void sai_deserialize_redis_communication_mode(
+        _In_ const std::string& s,
+        _Out_ sai_redis_communication_mode_t& value);
+
 #endif // __SAI_SERIALIZE__

meta/saiserialize.cpp

@@ -1981,6 +1981,34 @@ std::string sai_serialize(
     }
 }
 
+#define REDIS_COMMUNICATION_MODE_REDIS_ASYNC_STRING "redis_async"
+#define REDIS_COMMUNICATION_MODE_REDIS_SYNC_STRING  "redis_sync"
+#define REDIS_COMMUNICATION_MODE_ZMQ_SYNC_STRING    "zmq_sync"
+
+std::string sai_serialize_redis_communication_mode(
+        _In_ sai_redis_communication_mode_t value)
+{
+    SWSS_LOG_ENTER();
+
+    switch (value)
+    {
+        case SAI_REDIS_COMMUNICATION_MODE_REDIS_ASYNC:
+            return REDIS_COMMUNICATION_MODE_REDIS_ASYNC_STRING;
+
+        case SAI_REDIS_COMMUNICATION_MODE_REDIS_SYNC:
+            return REDIS_COMMUNICATION_MODE_REDIS_SYNC_STRING;
+
+        case SAI_REDIS_COMMUNICATION_MODE_ZMQ_SYNC:
+            return REDIS_COMMUNICATION_MODE_ZMQ_SYNC_STRING;
+
+        default:
+
+            SWSS_LOG_WARN("unknown value on sai_redis_communication_mode_t: %d", value);
+
+            return std::to_string(value);
+    }
+}
+
 // deserialize
 
 void sai_deserialize_bool(
@@ -3686,3 +3714,27 @@ sai_redis_notify_syncd_t sai_deserialize_redis_notify_syncd(
 
     return value;
 }
+
+void sai_deserialize_redis_communication_mode(
+        _In_ const std::string& s,
+        _Out_ sai_redis_communication_mode_t& value)
+{
+    SWSS_LOG_ENTER();
+
+    if (s == REDIS_COMMUNICATION_MODE_REDIS_ASYNC_STRING)
+    {
+        value = SAI_REDIS_COMMUNICATION_MODE_REDIS_ASYNC;
+    }
+    else if (s == REDIS_COMMUNICATION_MODE_REDIS_SYNC_STRING)
+    {
+        value = SAI_REDIS_COMMUNICATION_MODE_REDIS_SYNC;
+    }
+    else if (s == REDIS_COMMUNICATION_MODE_ZMQ_SYNC_STRING)
+    {
+        value = SAI_REDIS_COMMUNICATION_MODE_ZMQ_SYNC;
+    }
+    else
+    {
+        SWSS_LOG_THROW("enum '%s' not found in sai_redis_communication_mode_t", s.c_str());
+    }
+}

saiplayer/CommandLineOptions.cpp

@@ -1,5 +1,7 @@
 #include "CommandLineOptions.h"
 
+#include "meta/sai_serialize.h"
+
 #include "swss/logger.h"
 
 #include <sstream>
@@ -18,6 +20,9 @@ CommandLineOptions::CommandLineOptions()
     m_enableDebug = false;
     m_sleep = false;
     m_syncMode = false;
+    m_enableRecording = false;
+
+    m_redisCommunicationMode = SAI_REDIS_COMMUNICATION_MODE_REDIS_ASYNC;
 
     m_profileMapFile = "";
     m_contextConfig = "";
@@ -35,6 +40,8 @@ std::string CommandLineOptions::getCommandLineString() const
     ss << " EnableDebug=" << (m_enableDebug ? "YES" : "NO");
     ss << " Sleep=" << (m_sleep ? "YES" : "NO");
     ss << " SyncMode=" << (m_syncMode ? "YES" : "NO");
+    ss << " RedisCommunicationMode=" << sai_serialize_redis_communication_mode(m_redisCommunicationMode);
+    ss << " EnableRecording=" << (m_enableRecording ? "YES" : "NO");
     ss << " ProfileMapFile=" << m_profileMapFile;
     ss << " ContextConfig=" << m_contextConfig;
 

saiplayer/CommandLineOptions.h

@@ -1,5 +1,7 @@
 #pragma once
 
+#include "sairedis.h"
+
 #include "swss/sal.h"
 
 #include <string>
@@ -33,6 +35,10 @@ namespace saiplayer
 
             bool m_syncMode;
 
+            sai_redis_communication_mode_t m_redisCommunicationMode;
+
+            bool m_enableRecording;
+
             std::string m_profileMapFile;
 
             std::string m_contextConfig;