[orchagent]: Add ZmqRouteServer for concurrent route updates

lib/orch_zmq_config.cpp

@@ -78,6 +78,22 @@ std::shared_ptr<swss::ZmqServer> swss::create_zmq_server(std::string zmq_address
     return std::make_shared<ZmqServer>(zmq_address, vrf, true);
 }
 
+std::shared_ptr<swss::ZmqRouteServer> swss::create_zmq_route_server(std::string zmq_address, std::string vrf)
+{
+    if (!std::regex_search(zmq_address, ZMQ_NONE_IPV6_ADDRESS_WITH_PORT)
+            && !std::regex_search(zmq_address, ZMQ_IPV6_ADDRESS_WITH_PORT))
+    {
+        auto zmq_port = get_zmq_port();
+        zmq_address = zmq_address + ":" + std::to_string(zmq_port);
+    }
+
+    SWSS_LOG_NOTICE("Create ZMQ route server with address: %s, vrf: %s", zmq_address.c_str(), vrf.c_str());
+
+    // To prevent message loss between ZmqServer's bind operation and the creation of ZmqProducerStateTable,
+    // use lazy binding and call bind() only after the handler has been registered.
+    return std::make_shared<ZmqRouteServer>(zmq_address, vrf, true);
+}
+
 bool swss::get_feature_status(std::string feature, bool default_value)
 {
     std::shared_ptr<std::string> enabled = nullptr;

lib/orch_zmq_config.h

@@ -9,6 +9,7 @@
 #include "zmqclient.h"
 #include "zmqserver.h"
 #include "zmqproducerstatetable.h"
+#include "zmqrouteserver.h"
 
 /*
  * swssconfig will only connect to local orchagent ZMQ endpoint.
@@ -35,6 +36,7 @@ int get_zmq_port();
 std::shared_ptr<ZmqClient> create_zmq_client(std::string zmq_address, std::string vrf="");
 
 std::shared_ptr<ZmqServer> create_zmq_server(std::string zmq_address, std::string vrf="");
+std::shared_ptr<ZmqRouteServer> create_zmq_route_server(std::string zmq_address, std::string vrf="");
 
 bool get_feature_status(std::string feature, bool default_value);
 

orchagent/Makefile.am

@@ -122,6 +122,7 @@ orchagent_SOURCES = \
             response_publisher.cpp \
             nvgreorch.cpp \
             zmqorch.cpp \
+            zmqrouteorch.cpp \
             dash/dashenifwdorch.cpp \
             dash/dashenifwdinfo.cpp \
             dash/dashcounter.cpp \

orchagent/main.cpp

@@ -651,6 +651,9 @@ int main(int argc, char **argv)
     DBConnector config_db("CONFIG_DB", 0);
     DBConnector state_db("STATE_DB", 0);
 
+    // Get switch_type
+    getCfgSwitchType(&config_db, gMySwitchType, gMySwitchSubType);
+
     // Instantiate ZMQ server
     shared_ptr<ZmqServer> zmq_server = nullptr;
     if (zmq_server_address.empty())
@@ -660,12 +663,12 @@ int main(int argc, char **argv)
     else
     {
         SWSS_LOG_NOTICE("The ZMQ channel on the northbound side of orchagent has been initialized: %s, %s", zmq_server_address.c_str(), vrf.c_str());
-        zmq_server = create_zmq_server(zmq_server_address);
+        if (gMySwitchType == "fabric" || gMySwitchType == "dpu")
+            zmq_server = create_zmq_server(zmq_server_address);
+        else
+            zmq_server = create_zmq_route_server(zmq_server_address);
     }
 
-    // Get switch_type
-    getCfgSwitchType(&config_db, gMySwitchType, gMySwitchSubType);
-
     sai_attribute_t attr;
     vector<sai_attribute_t> attrs;
 

orchagent/orchdaemon.cpp

@@ -366,7 +366,7 @@ bool OrchDaemon::init()
 
     // Enable the fpmsyncd service to send Route events to orchagent via the ZMQ channel.
     auto enable_route_zmq = get_route_perf_zmq_enabled();
-    auto route_zmq_server = enable_route_zmq ? m_zmqServer : nullptr;
+    auto route_zmq_server = enable_route_zmq ? dynamic_cast<ZmqRouteServer *>(m_zmqServer) : nullptr;
 
     gRouteOrch = new RouteOrch(m_applDb, route_tables, gSwitchOrch, gNeighOrch, gIntfsOrch, vrf_orch, gFgNhgOrch, gSrv6Orch, route_zmq_server);
     gNhgOrch = new NhgOrch(m_applDb, APP_NEXTHOP_GROUP_TABLE_NAME);

orchagent/p4orch/tests/Makefile.am

@@ -29,6 +29,7 @@ p4orch_tests_SOURCES = $(ORCHAGENT_DIR)/orch.cpp \
 		       $(ORCHAGENT_DIR)/request_parser.cpp \
 		       $(top_srcdir)/lib/recorder.cpp \
 		       $(ORCHAGENT_DIR)/zmqorch.cpp \
+		       $(ORCHAGENT_DIR)/zmqrouteorch.cpp \
 		       $(ORCHAGENT_DIR)/namelabelmapper.cpp \
 		       $(ORCHAGENT_DIR)/flex_counter/flex_counter_manager.cpp \
 		       $(ORCHAGENT_DIR)/flex_counter/flow_counter_handler.cpp \

orchagent/p4orch/tests/fake_routeorch.cpp

@@ -19,7 +19,7 @@ extern sai_mpls_api_t*              sai_mpls_api;
 extern sai_switch_api_t*            sai_switch_api;
 extern size_t gMaxBulkSize;
 
-RouteOrch::RouteOrch(DBConnector *db, vector<table_name_with_pri_t> &tableNames, SwitchOrch *switchOrch, NeighOrch *neighOrch, IntfsOrch *intfsOrch, VRFOrch *vrfOrch, FgNhgOrch *fgNhgOrch, Srv6Orch *srv6Orch, swss::ZmqServer *zmqServer) :
+RouteOrch::RouteOrch(DBConnector *db, vector<table_name_with_pri_t> &tableNames, SwitchOrch *switchOrch, NeighOrch *neighOrch, IntfsOrch *intfsOrch, VRFOrch *vrfOrch, FgNhgOrch *fgNhgOrch, Srv6Orch *srv6Orch, swss::ZmqRouteServer *zmqServer) :
         gRouteBulker(sai_route_api, gMaxBulkSize),
         gLabelRouteBulker(sai_mpls_api, gMaxBulkSize),
         gNextHopGroupMemberBulker(sai_next_hop_group_api, gSwitchId, gMaxBulkSize),

orchagent/p4orch/tests/mock_routeorch.h

@@ -15,6 +15,7 @@
 #include "bulker.h"
 #include "fgnhgorch.h"
 #include <map>
+#include "zmqrouteserver.h"
 
 /* Maximum next hop group number */
 #define NHGRP_MAX_SIZE 128
@@ -182,7 +183,7 @@ struct LabelRouteBulkContext
 class RouteOrch : public Orch, public Subject
 {
 public:
-    RouteOrch(DBConnector *db, vector<table_name_with_pri_t> &tableNames, SwitchOrch *switchOrch, NeighOrch *neighOrch, IntfsOrch *intfsOrch, VRFOrch *vrfOrch, FgNhgOrch *fgNhgOrch, Srv6Orch *srv6Orch, swss::ZmqServer *zmqServer = nullptr);
+    RouteOrch(DBConnector *db, vector<table_name_with_pri_t> &tableNames, SwitchOrch *switchOrch, NeighOrch *neighOrch, IntfsOrch *intfsOrch, VRFOrch *vrfOrch, FgNhgOrch *fgNhgOrch, Srv6Orch *srv6Orch, swss::ZmqRouteServer *zmqServer = nullptr);
 
     bool hasNextHopGroup(const NextHopGroupKey&) const;
     sai_object_id_t getNextHopGroupId(const NextHopGroupKey&);

orchagent/routeorch.cpp

@@ -40,11 +40,11 @@ extern bool gEnableFibSuppress;
 #define DEFAULT_NUMBER_OF_ECMP_GROUPS   128
 #define DEFAULT_MAX_ECMP_GROUP_SIZE     32
 
-RouteOrch::RouteOrch(DBConnector *db, vector<table_name_with_pri_t> &tableNames, SwitchOrch *switchOrch, NeighOrch *neighOrch, IntfsOrch *intfsOrch, VRFOrch *vrfOrch, FgNhgOrch *fgNhgOrch, Srv6Orch *srv6Orch, swss::ZmqServer *zmqServer) :
+RouteOrch::RouteOrch(DBConnector *db, vector<table_name_with_pri_t> &tableNames, SwitchOrch *switchOrch, NeighOrch *neighOrch, IntfsOrch *intfsOrch, VRFOrch *vrfOrch, FgNhgOrch *fgNhgOrch, Srv6Orch *srv6Orch, swss::ZmqRouteServer *zmqRouteServer) :
         gRouteBulker(sai_route_api, gMaxBulkSize),
         gLabelRouteBulker(sai_mpls_api, gMaxBulkSize),
         gNextHopGroupMemberBulker(sai_next_hop_group_api, gSwitchId, gMaxBulkSize),
-        ZmqRouteOrch(db, tableNames, zmqServer, /*dbPersistence=*/false),
+        ZmqRouteOrch(db, tableNames, zmqRouteServer),
         m_switchOrch(switchOrch),
         m_neighOrch(neighOrch),
         m_intfsOrch(intfsOrch),

orchagent/routeorch.h

@@ -16,8 +16,8 @@
 #include "bulker.h"
 #include "fgnhgorch.h"
 #include <map>
-#include "zmqorch.h"
-#include "zmqserver.h"
+#include "zmqrouteorch.h"
+#include "zmqrouteserver.h"
 #include <unordered_map>
 
 extern bool gRouteStateAsyncPublish;
@@ -224,7 +224,7 @@ struct LabelRouteBulkContext
 class RouteOrch : public ZmqRouteOrch, public Subject
 {
 public:
-    RouteOrch(DBConnector *db, vector<table_name_with_pri_t> &tableNames, SwitchOrch *switchOrch, NeighOrch *neighOrch, IntfsOrch *intfsOrch, VRFOrch *vrfOrch, FgNhgOrch *fgNhgOrch, Srv6Orch *srv6Orch, swss::ZmqServer *zmqServer = nullptr);
+    RouteOrch(DBConnector *db, vector<table_name_with_pri_t> &tableNames, SwitchOrch *switchOrch, NeighOrch *neighOrch, IntfsOrch *intfsOrch, VRFOrch *vrfOrch, FgNhgOrch *fgNhgOrch, Srv6Orch *srv6Orch, swss::ZmqRouteServer *zmqServer = nullptr);
 
     bool hasNextHopGroup(const NextHopGroupKey&) const;
     sai_object_id_t getNextHopGroupId(const NextHopGroupKey&);

orchagent/zmqorch.cpp

@@ -31,22 +31,6 @@ void ZmqConsumer::execute()
     drain();
 }
 
-void ZmqRouteConsumer::execute()
-{
-    SWSS_LOG_ENTER();
-
-    size_t update_size = 0;
-    auto table = static_cast<swss::ZmqConsumerStateTable*>(getSelectable());
-    do
-    {
-        std::deque<KeyOpFieldsValuesTuple> entries;
-        table->pops(entries);
-        update_size = addToSync(entries);
-    } while (update_size != 0);
-
-    drain();
-}
-
 void ZmqConsumer::drain()
 {
     if (!m_toSync.empty() || !m_toSyncQueue.empty())
@@ -97,42 +81,3 @@ void ZmqOrch::doTask(Consumer &consumer)
     // When ZMQ disabled, forward data from Consumer
     doTask((ConsumerBase &)consumer);
 }
-
-ZmqRouteOrch::ZmqRouteOrch(DBConnector *db, const vector<string> &tableNames, ZmqServer *zmqServer, bool dbPersistence)
-: ZmqOrch()
-{
-    for (auto it : tableNames)
-    {
-        addConsumer(db, it, default_orch_pri, zmqServer, dbPersistence);
-    }
-}
-
-ZmqRouteOrch::ZmqRouteOrch(DBConnector *db, const vector<table_name_with_pri_t> &tableNames_with_pri, ZmqServer *zmqServer, bool dbPersistence)
-: ZmqOrch()
-{
-    for (const auto& it : tableNames_with_pri)
-    {
-        addConsumer(db, it.first, it.second, zmqServer, dbPersistence);
-    }
-}
-
-void ZmqRouteOrch::addConsumer(DBConnector *db, string tableName, int pri, ZmqServer *zmqServer, bool dbPersistence)
-{
-    if (db->getDbId() == APPL_DB || db->getDbId() == DPU_APPL_DB)
-    {
-        if (zmqServer != nullptr)
-        {
-            SWSS_LOG_DEBUG("ZmqRouteConsumer initialize for: %s", tableName.c_str());
-            addExecutor(new ZmqRouteConsumer(new ZmqConsumerStateTable(db, tableName, *zmqServer, gBatchSize, pri, dbPersistence), this, tableName));
-        }
-        else
-        {
-            SWSS_LOG_DEBUG("Consumer initialize for: %s", tableName.c_str());
-            addExecutor(new Consumer(new ConsumerStateTable(db, tableName, gBatchSize, pri), this, tableName));
-        }
-    }
-    else
-    {
-        SWSS_LOG_WARN("ZmqRouteOrch does not support create consumer for db: %d, table: %s", db->getDbId(), tableName.c_str());
-    }
-}

orchagent/zmqorch.h

@@ -23,17 +23,6 @@ class ZmqConsumer : public ConsumerBase {
     void drain() override;
 };
 
-class ZmqRouteConsumer : public ZmqConsumer
-{
-public:
-    ZmqRouteConsumer(swss::ZmqConsumerStateTable *select, Orch *orch, const std::string &name)
-        : ZmqConsumer(select, orch, name)
-    {
-    }
-
-    void execute() override;
-};
-
 class ZmqOrch : public Orch
 {
 public:
@@ -52,13 +41,3 @@ class ZmqOrch : public Orch
 private:
     void addConsumer(swss::DBConnector *db, std::string tableName, int pri, swss::ZmqServer *zmqServer, bool dbPersistence = true);
 };
-
-class ZmqRouteOrch : public ZmqOrch
-{
-public:
-    ZmqRouteOrch(swss::DBConnector *db, const std::vector<std::string> &tableNames, swss::ZmqServer *zmqServer, bool dbPersistence = true);
-    ZmqRouteOrch(swss::DBConnector *db, const std::vector<table_name_with_pri_t> &tableNames_with_pri, swss::ZmqServer *zmqServer, bool dbPersistence = true);
-
-private:
-    void addConsumer(swss::DBConnector *db, std::string tableName, int pri, swss::ZmqServer *zmqServer, bool dbPersistence = true);
-};

orchagent/zmqrouteorch.cpp

@@ -0,0 +1,121 @@
+#include "zmqrouteorch.h"
+
+using namespace swss;
+using namespace std;
+
+extern int gBatchSize;
+extern size_t gMaxBulkSize;
+
+ZmqRouteConsumer::ZmqRouteConsumer(swss::ZmqRouteConsumerStateTable *select, Orch *orch, const std::string &name)
+    : ConsumerBase(select, orch, name)
+{
+    // mqPollThread delivers bursts of tuples through this callback. Stage them
+    // in the plain m_ingress map under m_ingressMutex rather than merging into
+    // m_toSync here; the merge into m_toSync happens on the orch main thread in
+    // execute(). The eventfd is fired once the staged batch reaches
+    // gMaxBulkSize (so the main loop has a real batch to drain); otherwise
+    // mqPollThread fires it once per burst after the burst quiesces.
+    select->setIngressCallback(
+        [this, select](const std::vector<std::shared_ptr<KeyOpFieldsValuesTuple>> &kcos) {
+            std::lock_guard<std::mutex> lk(m_ingressMutex);
+            for (const auto &kco : kcos)
+            {
+                // Plain last-writer-wins staging by key. The SyncMap merge into
+                // m_toSync is applied later by execute()'s addToSync().
+                m_ingress[kfvKey(*kco)] = *kco;
+            }
+            if (m_ingress.size() >= gMaxBulkSize)
+            {
+                select->notifyPending();
+            }
+        });
+}
+
+void ZmqRouteConsumer::execute()
+{
+    SWSS_LOG_ENTER();
+
+    std::deque<KeyOpFieldsValuesTuple> entries;
+    {
+        // Move the staged tuples out of m_ingress under the lock, mirroring
+        // ZmqConsumer::execute()'s pops() + addToSync(entries). The lock is
+        // dropped before addToSync() below: m_toSync is owned by this (main)
+        // thread alone, so the merge needs no lock, and releasing early keeps
+        // mqPollThread free to stage the next burst while the merge runs.
+        std::lock_guard<std::mutex> lk(m_ingressMutex);
+        for (auto &kv : m_ingress)
+        {
+            entries.push_back(std::move(kv.second));
+        }
+        m_ingress.clear();
+    }
+    addToSync(entries);
+
+    // m_toSync is mutated only by this (main) thread, so drain() — which reads
+    // m_toSync and hands it to doTask — does not need to hold m_ingressMutex.
+    // Releasing the lock before drain() also keeps mqPollThread free to stage
+    // further tuples into m_ingress while doTask is running.
+    drain();
+}
+
+void ZmqRouteConsumer::drain()
+{
+    // TODO: doTask() currently walks the whole of m_toSync in one go. Per the
+    // route programming HLD -- doc/orchagent/orchagent_route_redesign.md
+    // sections 7.4 and 7.5 (sonic-net/SONiC#2328) -- this becomes a yieldable
+    // walk bounded by a time quantum, so the main loop returns to execute()
+    // and keeps draining m_ingress (and hence the ZMQ socket) instead of
+    // stalling ingress behind one large batch.
+    if (!m_toSync.empty())
+        (static_cast<ZmqRouteOrch*>(m_orch))->doTask(*this);
+}
+
+
+ZmqRouteOrch::ZmqRouteOrch(DBConnector *db, const vector<string> &tableNames, swss::ZmqRouteServer *zmqServer)
+: Orch()
+{
+    for (const auto& it : tableNames)
+    {
+        addConsumer(db, it, default_orch_pri, zmqServer);
+    }
+}
+
+
+ZmqRouteOrch::ZmqRouteOrch(DBConnector *db, const vector<table_name_with_pri_t> &tableNames_with_pri, swss::ZmqRouteServer *zmqServer)
+{
+    for (const auto& it : tableNames_with_pri)
+    {
+        addConsumer(db, it.first, it.second, zmqServer);
+    }
+}
+
+void ZmqRouteOrch::addConsumer(DBConnector *db, string tableName, int pri, swss::ZmqRouteServer *zmqServer)
+{
+    if (db->getDbId() == APPL_DB || db->getDbId() == DPU_APPL_DB)
+    {
+        if (zmqServer != nullptr)
+        {
+            SWSS_LOG_DEBUG("ZmqRouteConsumer initialize for: %s", tableName.c_str());
+            addExecutor(
+                new ZmqRouteConsumer(
+                  new swss::ZmqRouteConsumerStateTable(
+                    db, tableName, *zmqServer, pri, /* dbPersistence= */false),
+                this, tableName));
+        }
+        else
+        {
+            SWSS_LOG_DEBUG("Consumer initialize for: %s", tableName.c_str());
+            addExecutor(new Consumer(new ConsumerStateTable(db, tableName, gBatchSize, pri), this, tableName));
+        }
+    }
+    else
+    {
+        SWSS_LOG_WARN("ZmqRouteOrch does not support create consumer for db: %d, table: %s", db->getDbId(), tableName.c_str());
+    }
+}
+
+void ZmqRouteOrch::doTask(Consumer &consumer)
+{
+    // When ZMQ disabled, forward data from Consumer
+    doTask(static_cast<ConsumerBase &>(consumer));
+}