diff --git a/lib/orch_zmq_config.cpp b/lib/orch_zmq_config.cpp index 825c725c6b0..5783330c136 100644 --- a/lib/orch_zmq_config.cpp +++ b/lib/orch_zmq_config.cpp @@ -78,6 +78,22 @@ std::shared_ptr swss::create_zmq_server(std::string zmq_address return std::make_shared(zmq_address, vrf, true); } +std::shared_ptr 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 server with address: %s", zmq_address.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(zmq_address, vrf, true); +} + bool swss::get_feature_status(std::string feature, bool default_value) { std::shared_ptr enabled = nullptr; diff --git a/lib/orch_zmq_config.h b/lib/orch_zmq_config.h index b6167059423..d329ac60d7a 100644 --- a/lib/orch_zmq_config.h +++ b/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 create_zmq_client(std::string zmq_address, std::string vrf=""); std::shared_ptr create_zmq_server(std::string zmq_address, std::string vrf=""); +std::shared_ptr create_zmq_route_server(std::string zmq_address, std::string vrf=""); bool get_feature_status(std::string feature, bool default_value); diff --git a/orchagent/Makefile.am b/orchagent/Makefile.am index b0cbceb2414..f09b2dc0fff 100644 --- a/orchagent/Makefile.am +++ b/orchagent/Makefile.am @@ -121,6 +121,7 @@ orchagent_SOURCES = \ response_publisher.cpp \ nvgreorch.cpp \ zmqorch.cpp \ + zmqrouteorch.cpp \ dash/dashenifwdorch.cpp \ dash/dashenifwdinfo.cpp \ dash/dashcounter.cpp \ diff --git a/orchagent/main.cpp b/orchagent/main.cpp index 4c2f80d4c70..7622b5f50fa 100644 --- a/orchagent/main.cpp +++ b/orchagent/main.cpp @@ -646,6 +646,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 zmq_server = nullptr; if (zmq_server_address.empty()) @@ -655,12 +658,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 attrs; diff --git a/orchagent/orchdaemon.cpp b/orchagent/orchdaemon.cpp index 9064c6f75df..71d5236f91d 100644 --- a/orchagent/orchdaemon.cpp +++ b/orchagent/orchdaemon.cpp @@ -363,7 +363,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(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); diff --git a/orchagent/p4orch/tests/Makefile.am b/orchagent/p4orch/tests/Makefile.am index 416536dc8ff..d6316aa2893 100644 --- a/orchagent/p4orch/tests/Makefile.am +++ b/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)/flex_counter/flex_counter_manager.cpp \ $(ORCHAGENT_DIR)/flex_counter/flow_counter_handler.cpp \ $(ORCHAGENT_DIR)/port/port_capabilities.cpp \ diff --git a/orchagent/routeorch.cpp b/orchagent/routeorch.cpp index 5ef3ba6f7ea..feb8b828fa5 100644 --- a/orchagent/routeorch.cpp +++ b/orchagent/routeorch.cpp @@ -39,11 +39,11 @@ extern bool gRouteStateAsyncPublish; #define DEFAULT_NUMBER_OF_ECMP_GROUPS 128 #define DEFAULT_MAX_ECMP_GROUP_SIZE 32 -RouteOrch::RouteOrch(DBConnector *db, vector &tableNames, SwitchOrch *switchOrch, NeighOrch *neighOrch, IntfsOrch *intfsOrch, VRFOrch *vrfOrch, FgNhgOrch *fgNhgOrch, Srv6Orch *srv6Orch, swss::ZmqServer *zmqServer) : +RouteOrch::RouteOrch(DBConnector *db, vector &tableNames, SwitchOrch *switchOrch, NeighOrch *neighOrch, IntfsOrch *intfsOrch, VRFOrch *vrfOrch, FgNhgOrch *fgNhgOrch, Srv6Orch *srv6Orch, 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), diff --git a/orchagent/routeorch.h b/orchagent/routeorch.h index 6123c75d589..5ae1a8bde6a 100644 --- a/orchagent/routeorch.h +++ b/orchagent/routeorch.h @@ -16,8 +16,8 @@ #include "bulker.h" #include "fgnhgorch.h" #include -#include "zmqorch.h" -#include "zmqserver.h" +#include "zmqrouteorch.h" +#include "zmqrouteserver.h" #include extern bool gRouteStateAsyncPublish; @@ -224,7 +224,7 @@ struct LabelRouteBulkContext class RouteOrch : public ZmqRouteOrch, public Subject { public: - RouteOrch(DBConnector *db, vector &tableNames, SwitchOrch *switchOrch, NeighOrch *neighOrch, IntfsOrch *intfsOrch, VRFOrch *vrfOrch, FgNhgOrch *fgNhgOrch, Srv6Orch *srv6Orch, swss::ZmqServer *zmqServer = nullptr); + RouteOrch(DBConnector *db, vector &tableNames, SwitchOrch *switchOrch, NeighOrch *neighOrch, IntfsOrch *intfsOrch, VRFOrch *vrfOrch, FgNhgOrch *fgNhgOrch, Srv6Orch *srv6Orch, ZmqRouteServer *zmqServer = nullptr); bool hasNextHopGroup(const NextHopGroupKey&) const; sai_object_id_t getNextHopGroupId(const NextHopGroupKey&); diff --git a/orchagent/zmqorch.cpp b/orchagent/zmqorch.cpp index 7586d5afd8f..80c77f93bc6 100644 --- a/orchagent/zmqorch.cpp +++ b/orchagent/zmqorch.cpp @@ -32,22 +32,6 @@ void ZmqConsumer::execute() drain(); } -void ZmqRouteConsumer::execute() -{ - SWSS_LOG_ENTER(); - - size_t update_size = 0; - auto table = static_cast(getSelectable()); - do - { - std::deque entries; - table->pops(entries); - update_size = addToSync(entries); - } while (update_size != 0); - - drain(); -} - void ZmqConsumer::drain() { if (!m_toSync.empty() || !m_queue.empty()) @@ -98,42 +82,3 @@ void ZmqOrch::doTask(Consumer &consumer) // When ZMQ disabled, forward data from Consumer doTask((ConsumerBase &)consumer); } - -ZmqRouteOrch::ZmqRouteOrch(DBConnector *db, const vector &tableNames, ZmqServer *zmqServer, bool dbPersistence) -: ZmqOrch() -{ - for (auto it : tableNames) - { - addConsumer(db, it, default_orch_pri, zmqServer, dbPersistence); - } -} - -ZmqRouteOrch::ZmqRouteOrch(DBConnector *db, const vector &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()); - } -} diff --git a/orchagent/zmqorch.h b/orchagent/zmqorch.h index a0e7750995e..3662df5f1f6 100644 --- a/orchagent/zmqorch.h +++ b/orchagent/zmqorch.h @@ -28,17 +28,6 @@ class ZmqConsumer : public ConsumerBase { std::deque m_queue; }; -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: @@ -57,13 +46,3 @@ class ZmqOrch : public Orch private: void addConsumer(swss::DBConnector *db, std::string tableName, int pri, swss::ZmqServer *zmqServer, bool orderedQueue = false, bool dbPersistence = true); }; - -class ZmqRouteOrch : public ZmqOrch -{ -public: - ZmqRouteOrch(swss::DBConnector *db, const std::vector &tableNames, swss::ZmqServer *zmqServer, bool dbPersistence = true); - ZmqRouteOrch(swss::DBConnector *db, const std::vector &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); -}; diff --git a/orchagent/zmqrouteorch.cpp b/orchagent/zmqrouteorch.cpp new file mode 100644 index 00000000000..7f89565b32f --- /dev/null +++ b/orchagent/zmqrouteorch.cpp @@ -0,0 +1,111 @@ +#include "zmqrouteorch.h" + +using namespace swss; +using namespace std; + +extern int gBatchSize; +extern size_t gMaxBulkSize; + +ZmqRouteConsumer::ZmqRouteConsumer(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_toSyncMutex 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> &kcos) { + std::lock_guard lk(m_toSyncMutex); + 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(); + + { + // Drain the staged tuples into m_toSync under the lock, mirroring + // ZmqConsumer::execute()'s pops() + addToSync(entries). The lock is + // held only while moving tuples out of m_ingress. + std::lock_guard lk(m_toSyncMutex); + std::deque entries; + 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_toSyncMutex. + drain(); +} + +void ZmqRouteConsumer::drain() +{ + if (!m_toSync.empty()) + (static_cast(m_orch))->doTask(*this); +} + + +ZmqRouteOrch::ZmqRouteOrch(DBConnector *db, const vector &tableNames, ZmqRouteServer *zmqServer) +: Orch() +{ + for (auto it : tableNames) + { + addConsumer(db, it, default_orch_pri, zmqServer); + } +} + + +ZmqRouteOrch::ZmqRouteOrch(DBConnector *db, const vector &tableNames_with_pri, 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, 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 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((ConsumerBase &)consumer); +} diff --git a/orchagent/zmqrouteorch.h b/orchagent/zmqrouteorch.h new file mode 100644 index 00000000000..80315f3449f --- /dev/null +++ b/orchagent/zmqrouteorch.h @@ -0,0 +1,49 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include "zmqrouteserver.h" +#include "zmqrouteconsumerstatetable.h" + +extern int gZmqExecuteTimeQuantaMsecs; + +class ZmqRouteConsumer : public ConsumerBase { +public: + ZmqRouteConsumer(ZmqRouteConsumerStateTable *select, Orch *orch, const std::string &name); + + swss::TableBase *getConsumerTable() const override + { + // ZmqRouteConsumerStateTable is a subclass of TableBase + return static_cast(getSelectable()); + } + + void execute() override; + void drain() override; + +private: + // Staging buffer for tuples delivered by the ZmqRouteServer mqPollThread + // ingress callback. The callback writes here under m_toSyncMutex (rather + // than merging into m_toSync directly); execute() drains it into m_toSync + // under the same lock. This keeps m_toSync single-threaded (touched only + // by the orch main thread), so the base ConsumerBase paths need no locking. + std::mutex m_toSyncMutex; + std::unordered_map m_ingress; +}; + +class ZmqRouteOrch : public Orch +{ +public: + ZmqRouteOrch(swss::DBConnector *db, const std::vector &tableNames, ZmqRouteServer *zmqServer); + ZmqRouteOrch(swss::DBConnector *db, const std::vector &tableNames_with_pri, ZmqRouteServer *zmqServer); + + virtual void doTask(ConsumerBase &consumer) { }; + void doTask(Consumer &consumer) override; + +private: + void addConsumer(swss::DBConnector *db, std::string tableName, int pri, ZmqRouteServer *zmqServer); +}; diff --git a/tests/mock_tests/Makefile.am b/tests/mock_tests/Makefile.am index 47b06c3b7b8..6ec28795a9c 100644 --- a/tests/mock_tests/Makefile.am +++ b/tests/mock_tests/Makefile.am @@ -92,6 +92,7 @@ tests_SOURCES = aclorch_ut.cpp \ mock_orch_test.cpp \ mock_dash_orch_test.cpp \ zmq_orch_ut.cpp \ + zmq_route_orch_ut.cpp \ retrycache_ut.cpp \ saihelper_ut.cpp \ mock_saihelper.cpp \ @@ -180,6 +181,7 @@ tests_SOURCES = aclorch_ut.cpp \ $(top_srcdir)/cfgmgr/portmgr.cpp \ $(top_srcdir)/cfgmgr/sflowmgr.cpp \ $(top_srcdir)/orchagent/zmqorch.cpp \ + $(top_srcdir)/orchagent/zmqrouteorch.cpp \ $(top_srcdir)/orchagent/dash/dashenifwdorch.cpp \ $(top_srcdir)/orchagent/dash/dashenifwdinfo.cpp \ $(top_srcdir)/orchagent/dash/dashaclorch.cpp \ diff --git a/tests/mock_tests/zmq_route_orch_ut.cpp b/tests/mock_tests/zmq_route_orch_ut.cpp new file mode 100644 index 00000000000..3ae59bea123 --- /dev/null +++ b/tests/mock_tests/zmq_route_orch_ut.cpp @@ -0,0 +1,337 @@ +#include +#include +#include +#include +#include +#include +#include + +#include "gtest/gtest.h" +#include "schema.h" +#include "ut_helper.h" +#include "orch_zmq_config.h" +#include "dbconnector.h" +#include "mock_table.h" +#include "select.h" +#include "zmqclient.h" +#include "zmqproducerstatetable.h" +#include "zmqrouteserver.h" +#include "zmqrouteconsumerstatetable.h" + +#define protected public +#include "orch.h" +#include "zmqrouteorch.h" +#undef protected + +using namespace std; +using namespace swss; + +extern size_t gMaxBulkSize; + +namespace { + +// Wait until pred() becomes true or deadlineMs elapses; returns the final value. +template +bool waitFor(int deadlineMs, Pred pred) +{ + auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(deadlineMs); + while (std::chrono::steady_clock::now() < deadline) + { + if (pred()) + return true; + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + return pred(); +} + +// Minimal subclass of ZmqRouteOrch that records doTask invocations, so tests +// can assert that drain() forwards correctly without needing a full RouteOrch. +class RecordingZmqRouteOrch : public ZmqRouteOrch +{ +public: + RecordingZmqRouteOrch(swss::DBConnector *db, + const std::vector &tables, + ZmqRouteServer *zmqServer) + : ZmqRouteOrch(db, tables, zmqServer) + { + } + + void doTask(ConsumerBase &consumer) override + { + ++doTaskCount; + // Record the keys handed to doTask before clearing. m_toSync is cleared + // below, so tests that want to verify specific entries must capture + // them here. doTask runs only on the orch main thread, so no lock. + for (const auto &kv : consumer.m_toSync) + { + seenKeys.insert(kv.first); + } + // Drain the consumer's m_toSync so subsequent drain() calls observe it + // as empty (matches the contract a real orch would honor). + consumer.m_toSync.clear(); + } + + std::atomic doTaskCount{0}; + std::set seenKeys; +}; + +} // namespace + +// ZmqRouteOrch with a nullptr server falls back to plain Consumer (legacy +// non-ZMQ path) for APPL_DB tables. +TEST(ZmqRouteOrchTest, NullServerFallsBackToConsumer) +{ + vector tables = { + { "ZMQ_ROUTE_UT_T1", 1 }, + { "ZMQ_ROUTE_UT_T2", 2 }, + }; + auto app_db = make_shared("APPL_DB", 0); + auto orch = make_shared(app_db.get(), tables, nullptr); + + EXPECT_EQ(orch->getSelectables().size(), tables.size()); + // Ensure the executor is a plain Consumer (not a ZmqRouteConsumer): the + // legacy fallback shouldn't pull in the ZmqRouteConsumer machinery. + auto exec = orch->m_consumerMap.begin()->second.get(); + EXPECT_EQ(dynamic_cast(exec), nullptr); +} + +// vector ctor (no per-table priority) — exercises the +// default_orch_pri code path in ZmqRouteOrch::ZmqRouteOrch(vector,...). +TEST(ZmqRouteOrchTest, VectorOfStringsCtor) +{ + vector tables = { "ZMQ_ROUTE_UT_TS1", "ZMQ_ROUTE_UT_TS2" }; + auto app_db = make_shared("APPL_DB", 0); + auto orch = make_shared(app_db.get(), tables, nullptr); + EXPECT_EQ(orch->getSelectables().size(), tables.size()); +} + +// Non-APPL_DB databases are unsupported; addConsumer should warn and create +// no executor. +TEST(ZmqRouteOrchTest, UnsupportedDbProducesNoExecutor) +{ + vector tables = { { "ZMQ_ROUTE_UT_T1", 1 } }; + auto state_db = make_shared("STATE_DB", 0); + auto orch = make_shared(state_db.get(), tables, nullptr); + EXPECT_EQ(orch->getSelectables().size(), 0u); +} + +// With a real ZmqRouteServer, ZmqRouteOrch creates a ZmqRouteConsumer (not a +// plain Consumer). The server must outlive the orch. +TEST(ZmqRouteOrchTest, RealServerCreatesZmqRouteConsumer) +{ + vector tables = { { "ZMQ_ROUTE_UT_T1", 1 } }; + auto app_db = make_shared("APPL_DB", 0); + ZmqRouteServer server("tcp://*:1260", "", /*lazyBind=*/true); + + auto orch = make_shared(app_db.get(), tables, &server); + ASSERT_EQ(orch->getSelectables().size(), tables.size()); + + auto exec = orch->m_consumerMap.begin()->second.get(); + EXPECT_NE(dynamic_cast(exec), nullptr); +} + +// doTask(Consumer&) on the base ZmqRouteOrch is a stub that forwards to the +// virtual doTask(ConsumerBase&) — this is the only piece that ZmqRouteOrch +// itself implements (besides ctors / addConsumer). Cover it. +TEST(ZmqRouteOrchTest, DoTaskConsumerForwardsToConsumerBase) +{ + vector tables = { { "ZMQ_ROUTE_UT_T1", 1 } }; + auto app_db = make_shared("APPL_DB", 0); + auto orch = make_shared(app_db.get(), tables, nullptr); + + auto *exec = orch->m_consumerMap.begin()->second.get(); + auto *consumer = dynamic_cast(exec); + ASSERT_NE(consumer, nullptr); + + // Forge a single entry into m_toSync so that the recording doTask can see + // something and so the subsequent clear() actually does work. SyncMap is a + // multimap, so use insert rather than operator[]. + consumer->m_toSync.insert({ + "k1", + std::make_tuple(std::string("k1"), std::string(SET_COMMAND), + std::vector{{"f", "v"}}) + }); + + // ZmqRouteOrch::doTask(Consumer&) forwards to doTask(ConsumerBase&). + static_cast(orch.get())->doTask(*consumer); + EXPECT_EQ(orch->doTaskCount.load(), 1); + EXPECT_TRUE(consumer->m_toSync.empty()); +} + +// Drain on a ZmqRouteConsumer with empty m_toSync must NOT call doTask. +// Drain on a non-empty m_toSync must call doTask exactly once and the lock +// must allow re-entry afterwards. +TEST(ZmqRouteConsumerTest, DrainGatedByToSyncEmptiness) +{ + vector tables = { { "ZMQ_ROUTE_UT_T1", 1 } }; + auto app_db = make_shared("APPL_DB", 0); + ZmqRouteServer server("tcp://*:1261", "", /*lazyBind=*/true); + auto orch = make_shared(app_db.get(), tables, &server); + + auto *exec = orch->m_consumerMap.begin()->second.get(); + auto *zrc = dynamic_cast(exec); + ASSERT_NE(zrc, nullptr); + + // Empty m_toSync: drain is a no-op. + zrc->drain(); + EXPECT_EQ(orch->doTaskCount.load(), 0); + + // Stage one entry via addToSync; drain forwards to doTask exactly once. + // RecordingZmqRouteOrch::doTask clears m_toSync. + KeyOpFieldsValuesTuple kfv("route_a", SET_COMMAND, + vector{{"f", "v"}}); + zrc->addToSync(kfv); + EXPECT_EQ(zrc->m_toSync.size(), 1u); + + zrc->drain(); + EXPECT_EQ(orch->doTaskCount.load(), 1); + EXPECT_TRUE(zrc->m_toSync.empty()); + + // A subsequent empty drain still doesn't call doTask, and the lock + // re-acquires cleanly. + zrc->drain(); + EXPECT_EQ(orch->doTaskCount.load(), 1); +} + +// execute() drains m_ingress into m_toSync then calls drain(); with m_ingress +// empty here it reduces to drain(). Cover that override. +TEST(ZmqRouteConsumerTest, ExecuteDelegatesToDrain) +{ + vector tables = { { "ZMQ_ROUTE_UT_T1", 1 } }; + auto app_db = make_shared("APPL_DB", 0); + ZmqRouteServer server("tcp://*:1262", "", /*lazyBind=*/true); + auto orch = make_shared(app_db.get(), tables, &server); + + auto *zrc = dynamic_cast( + orch->m_consumerMap.begin()->second.get()); + ASSERT_NE(zrc, nullptr); + + KeyOpFieldsValuesTuple kfv("route_b", SET_COMMAND, + vector{{"f", "v"}}); + zrc->addToSync(kfv); + + zrc->execute(); + EXPECT_EQ(orch->doTaskCount.load(), 1); +} + +// Deque-form addToSync forwards to ConsumerBase::addToSync(deque) and returns +// the count. +TEST(ZmqRouteConsumerTest, AddToSyncDequeReturnsCount) +{ + vector tables = { { "ZMQ_ROUTE_UT_T1", 1 } }; + auto app_db = make_shared("APPL_DB", 0); + ZmqRouteServer server("tcp://*:1263", "", /*lazyBind=*/true); + auto orch = make_shared(app_db.get(), tables, &server); + + auto *zrc = dynamic_cast( + orch->m_consumerMap.begin()->second.get()); + ASSERT_NE(zrc, nullptr); + + std::deque entries; + for (int i = 0; i < 5; ++i) + { + entries.emplace_back("k" + std::to_string(i), SET_COMMAND, + vector{{"f", "v"}}); + } + + EXPECT_EQ(zrc->addToSync(entries), 5u); + EXPECT_EQ(zrc->m_toSync.size(), 5u); +} + +// dumpPendingTasks returns the staged entries as strings. +TEST(ZmqRouteConsumerTest, DumpPendingTasksReturnsEntries) +{ + vector tables = { { "ZMQ_ROUTE_UT_T1", 1 } }; + auto app_db = make_shared("APPL_DB", 0); + ZmqRouteServer server("tcp://*:1264", "", /*lazyBind=*/true); + auto orch = make_shared(app_db.get(), tables, &server); + + auto *zrc = dynamic_cast( + orch->m_consumerMap.begin()->second.get()); + ASSERT_NE(zrc, nullptr); + + zrc->addToSync(KeyOpFieldsValuesTuple("kA", SET_COMMAND, + vector{{"f", "v"}})); + zrc->addToSync(KeyOpFieldsValuesTuple("kB", DEL_COMMAND, + vector{})); + + std::vector ts; + zrc->dumpPendingTasks(ts); + EXPECT_EQ(ts.size(), 2u); +} + +// End-to-end: ZmqProducerStateTable → ZmqRouteServer → ZmqRouteConsumer. +// The ingress callback (on mqPollThread) stages the tuple into m_ingress; +// execute() on this thread then drains m_ingress into m_toSync and forwards it +// to doTask. Verifies the callback wiring set up by ZmqRouteConsumer's +// constructor delivers the entry all the way through to doTask. +TEST(ZmqRouteConsumerTest, IngressCallbackDeliversToDoTask) +{ + const string tableName = "ZMQ_ROUTE_UT_INGRESS"; + const string pushEndpoint = "tcp://localhost:1266"; + const string pullEndpoint = "tcp://*:1266"; + + vector tables = { { tableName, 1 } }; + auto app_db = make_shared("APPL_DB", 0); + ZmqRouteServer server(pullEndpoint, "", /*lazyBind=*/true); + auto orch = make_shared(app_db.get(), tables, &server); + auto *zrc = dynamic_cast( + orch->m_consumerMap.begin()->second.get()); + ASSERT_NE(zrc, nullptr); + + server.bind(); + + ZmqClient client(pushEndpoint, 0); + ZmqProducerStateTable p(app_db.get(), tableName, client, /*dbPersistence=*/false); + p.set("route_x", vector{{"nh", "1.1.1.1"}}); + + // The tuple arrives asynchronously on mqPollThread and lands in m_ingress. + // Repeatedly run execute() (drains m_ingress → m_toSync → doTask) until + // doTask observes the entry, then confirm the specific key was delivered. + ASSERT_TRUE(waitFor(2000, [&] { + zrc->execute(); + return orch->doTaskCount.load() >= 1; + })); + EXPECT_EQ(orch->seenKeys.count("route_x"), 1u); +} + +// When the ingress callback stages past gMaxBulkSize entries, it must fire +// notifyPending mid-burst (rather than waiting for the burst quiesce timer) +// so the orch main loop wakes up and drains immediately. We lower +// gMaxBulkSize to 1 to make this trivially observable. +TEST(ZmqRouteConsumerTest, IngressCallbackFiresNotifyAtMaxBulkSize) +{ + const string tableName = "ZMQ_ROUTE_UT_BULK"; + const string pushEndpoint = "tcp://localhost:1267"; + const string pullEndpoint = "tcp://*:1267"; + + vector tables = { { tableName, 1 } }; + auto app_db = make_shared("APPL_DB", 0); + ZmqRouteServer server(pullEndpoint, "", /*lazyBind=*/true); + auto orch = make_shared(app_db.get(), tables, &server); + auto *zrc = dynamic_cast( + orch->m_consumerMap.begin()->second.get()); + ASSERT_NE(zrc, nullptr); + + server.bind(); + + // Force the mid-burst notify branch to trip on the very first callback. + const size_t savedMaxBulk = gMaxBulkSize; + gMaxBulkSize = 1; + + Select sel; + sel.addSelectable(zrc); + + ZmqClient client(pushEndpoint, 0); + ZmqProducerStateTable p(app_db.get(), tableName, client, /*dbPersistence=*/false); + p.set("route_bulk", vector{{"nh", "2.2.2.2"}}); + + // The ingress callback fires notifyPending the moment m_ingress reaches + // gMaxBulkSize=1, so the Select loop wakes on the consumer's event without + // waiting for the BURST_QUIESCE_MS post-burst notify. + Selectable *out = nullptr; + EXPECT_EQ(sel.select(&out, 2000), Select::OBJECT); + EXPECT_EQ(out, zrc); + + gMaxBulkSize = savedMaxBulk; +}