Add ZmqRouteServer and ZmqRouteConsumerStateTable for burst-coalesced route ingress

[venkit-nexthop]

Why I did it

Under sustained route ingress, ZmqConsumerStateTable fires its SelectableEvent once per ZMQ message. Each notification wakes the consumer's main loop, which runs execute() -> drain() -> bulker.flush() even though the bulker exists specifically to amortize that flush across many tuples. We want the consumer main loop to wake once per burst, not once per message.

How I did it

Two new classes plus small refactors of the existing ones:

• ZmqRouteServer (new, common/zmqrouteserver.{h,cpp}) — derives from ZmqServer, overrides mqPollThread(). Replaces the base's "1 zmq_poll -> 1 zmq_recv -> handle" loop with:

  1. zmq_poll with adaptive timeout (1 s when idle, 5 ms BURST_QUIESCE_MS while a burst is in progress).
  2. Inner drain loop calling zmq_recv(ZMQ_DONTWAIT) until EAGAIN, accumulating touched handlers in a set.
  3. After the burst quiesces (a BURST_QUIESCE_MS poll timeout with non-empty dirty-handlers set), call notifyPending() on each touched handler exactly once. eventfd writes coalesce, so a mid-burst notify from the consumer side is harmless.
  4. The oneToOneSync (request/response) mode supported by ZmqServer is intentionally not available here — burst coalescing assumes streaming ingress.

• ZmqRouteConsumerStateTable (new, common/zmqrouteconsumerstatetable.{h,cpp}) — derives from ZmqConsumerStateTable. Adds a public setIngressCallback(IngressCallback) so a consumer can merge tuples directly from mqPollThread into its own sync map (skipping the internal std::queue + per-message eventfd notify). notifyPending() overrides the base no-op and signals the SelectableEvent. hasData() always returns true and hasCachedData() returns false: the Selectable's readiness is event-driven via notifyPending.

• ZmqMessageHandler gains a virtual void notifyPending() default no-op so ZmqConsumerStateTable (which doesn't need it) inherits silently.

• ZmqServer: destructor is now virtual (the class has a virtual mqPollThread, so a non-virtual dtor would UB-delete a derived ZmqRouteServer through a base pointer). A few private members (m_socket, m_endpoint, m_runThread, m_buffer, m_oneToOneSync, m_allowZmqPoll) and handleReceivedData(buf, size) / mqPollThread() are exposed as protected so the route subclass can access them. Other internals (m_mqPollThread, m_HandlerMap, m_vrf, m_context, findMessageHandler, startMqPollThread) stay private. handleReceivedData(buf, size) now returns the matched ZmqMessageHandler* (or nullptr) so the route server can record dirty handlers without re-doing the lookup.

• ZmqConsumerStateTable::handleReceivedData(kcos) is now virtual (overridden by ZmqRouteConsumerStateTable), and m_selectableEvent / m_asyncDBUpdater are protected so the override can fire the eventfd and forward to the async updater.

How to verify it

6 new unit tests in tests/zmq_route_ut.cpp:

• ZmqMessageHandler.NotifyPendingDefaultIsNoOp — base default doesn't crash.
• ZmqRouteConsumerStateTable.ConstSelectableProperties — hasData()=true, hasCachedData()=false, initializedWithData()=false.
• ZmqRouteConsumerStateTable.NotifyPendingFiresSelectableEvent — direct notifyPending() wakes Select.
• ZmqRouteConsumerStateTable.IngressCallbackReceivesData — end-to-end: ZmqProducerStateTable -> ZmqRouteServer -> ingress callback gets the deserialized tuple; Select wakes on the post-burst notify.
• ZmqRouteConsumerStateTable.BurstCoalescingFiresFewerWakeups — the core new behavior: send 200 messages back-to-back, all reach the callback but Select wakes far fewer times (asserts wakeups < N/4; observed 1–3 in practice).
• ZmqRouteConsumerStateTable.NoIngressCallbackIsSafe — callback unset case doesn't crash; notifyPending still wakes Select.

```
$ ./tests/tests --gtest_filter="ZmqMessageHandler.:ZmqRouteConsumerStateTable."
[==========] 6 tests from 2 test suites ran. (3134 ms total)
[ PASSED ] 6 tests.
```

Coverage on the new files: zmqrouteserver.cpp 90.0% lines, zmqrouteconsumerstatetable.cpp 71.4% (gap: dbPersistence=true async-updater path), zmqrouteconsumerstatetable.h 100%. Remaining uncovered lines are defensive throws (SWSS_LOG_THROW on non-EAGAIN/EINTR zmq_recv errors and on message truncation past 16 MiB) that aren't reachable from a unit test.

Which release branch to backport (provide reason below if selected)

[mssonicbld]

/azp run

[azure-pipelines]

Azure Pipelines successfully started running 1 pipeline(s).

[securely1g]

Thanks for the PR — the core optimization (drain-then-notify via adaptive zmq_poll timeout) is solid and well-motivated. A few concerns:

1. zmqserver.h access control layout

There are now 4 alternating protected:/private: sections. Should consolidate into one protected: block and one private: block for readability.

2. hasData() always returns true

If orch code ever checks hasData() to decide whether to call pops(), it'll always enter that path even when there's nothing. Could this cause spinning or unexpected behavior in any existing consumer patterns?

3. No pops() override — consumption pattern unclear

ZmqRouteConsumerStateTable inherits ZmqConsumerStateTable::pops() which drains m_receivedOperationQueue. But with the ingress callback path, tuples never enter that queue — they go straight to the callback. So pops() would always return empty. If the orch main loop calls pops() after Select wakes it (the normal pattern), it gets nothing.

How does the consumer actually retrieve data after Select wakes? The ingress callback presumably merges into m_toSync directly, but that contract isn't enforced or documented in the class interface. This is the main thing that needs clarification.

4. Thread safety of m_ingressCallback

The callback is set from the main thread via setIngressCallback() but invoked from mqPollThread. There's no synchronization on the std::function itself. If setIngressCallback is called after server.bind() (after the poll thread starts), there's a data race. Should either:

• Document that setIngressCallback must be called before bind()
• Use std::atomic or a mutex around the callback check

5. oneToOneSync not explicitly blocked

The comment says "intentionally not available here" but nothing prevents it structurally. Fine now since the 3-arg constructor doesn't expose it, but fragile if someone adds constructors later. A runtime check or static_assert would harden this.

6. Test coverage

• Hardcoded ports 1238-1242 could collide in parallel test runs
• The dbPersistence=true path in handleReceivedData forwards to m_asyncDBUpdater but has no test. Since this is a new class, at least one test with dbPersistence=true would be good.

Minor

zmqrouteserver.h uses 2-space indentation for public/private but the rest of swss-common uses 4. Should match codebase style.

---

Overall the burst coalescing approach is sound — the adaptive timeout with dirty handler tracking is clean. Main blocker is #3 (clarifying the data consumption contract).

[mssonicbld]

/azp run

[azure-pipelines]

Azure Pipelines successfully started running 1 pipeline(s).

[mssonicbld]

/azp run

[azure-pipelines]

Azure Pipelines successfully started running 1 pipeline(s).

[mssonicbld]

/azp run

[azure-pipelines]

Azure Pipelines successfully started running 1 pipeline(s).

[mssonicbld]

Hi, there are workflow run(s) waiting for approval, you may be first-time contributor. I will notify maintainers to help approve once PR is approved. Thanks!

_{---Powered by SONiC BuildBot}

[deepak-singhal0408]

the burst drain+once-per-burst-wakeup model is sound and cleanly scoped and implementation looks good to me.