[orchagent]: In-process SAI notification queue for ZMQ mode (Option 3)

[vpandian-nokia]

What I did

• Implemented Option 3 (HLD: sonic-buildimage PR #27791): in ZMQ mode, SAI notifications are delivered on the libsairedis callback thread, enqueued into a shared in-process SaiNotificationQueue, and drained by the existing orchagent main Select loop through a queue executor — instead of re-posting to ASIC_DB:NOTIFICATIONS and consuming via Redis NotificationConsumer.
• Added SaiNotificationQueue, SaiNotificationQueueSelectable, SaiNotificationQueueExecutor, and SaiNotificationDispatcher.
• Added SaiNotificationOrch as a thin infrastructure orch (not a feature orch): it owns the shared queue executor and exposes registerHandler(op, handler, readinessPredicate) so feature orchs attach per-op dispatch callbacks. Feature orchs do not register separate executors for queue-path notifications.
• Updated SAI notification callbacks in notifications.cpp (and ICMP callback) to call enqueueSaiNotification() in ZMQ mode; non-ZMQ mode continues to use the existing Redis notification path unchanged.
• Migrated notification handlers to shared handleNotification(entry) in:
  • PortsOrch (SAI_SWITCH_NOTIFICATION_NAME_PORT_STATE_CHANGE, SAI_SWITCH_NOTIFICATION_NAME_PORT_HOST_TX_READY)
  • FdbOrch (SAI_SWITCH_NOTIFICATION_NAME_FDB_EVENT)
  • BfdOrch (SAI_SWITCH_NOTIFICATION_NAME_BFD_SESSION_STATE_CHANGE)
  • IcmpOrch (SAI_SWITCH_NOTIFICATION_NAME_ICMP_ECHO_SESSION_STATE_CHANGE)
  • TwampOrch (SAI_SWITCH_NOTIFICATION_NAME_TWAMP_SESSION_EVENT)
  • DashHaOrch (SAI_SWITCH_NOTIFICATION_NAME_HA_SET_EVENT, SAI_SWITCH_NOTIFICATION_NAME_HA_SCOPE_EVENT)
  • DashHaFlowOrch (SAI_SWITCH_NOTIFICATION_NAME_FLOW_BULK_GET_SESSION_EVENT)
  • MACsecOrch (SAI_SWITCH_NOTIFICATION_NAME_SWITCH_MACSEC_POST_STATUS, SAI_SWITCH_NOTIFICATION_NAME_MACSEC_POST_STATUS)
  • HFTelOrch (SAI_SWITCH_NOTIFICATION_NAME_TAM_TEL_TYPE_CONFIG_CHANGE)
• Added/extended unit tests in tests/mock_tests/notifications_ut.cpp.
• Fixed duplicate Orch::addExecutor(icmpStateNotifier) introduced during rebase in icmporch.cpp.

Why I did it
For ZMQ southbound mode, SAI notifications are delivered in-process. Re-posting them through Redis adds latency and extra serialization. Option 3 routes notifications directly from the SAI callback into an in-process queue and dispatches them to the owning orch, avoiding the Redis round-trip while keeping the existing orch handler logic.
Related:

• HLD: sonic-buildimage PR #27791
• Issue: sonic-buildimage #27541

How I verified it

• Built swss deb; mock_tests passed, including notifications_ut (12/12 ZMQ enqueue tests).
• DUT validation on Nokia IXR-7220 H5-64D:
  • Port state: verified via port bounce (admin down/up); handlePortStateChangeNotification observed in orchagent syslog with correct oper status and speed.
  • FDB: verified via actual MAC learning on a VLAN member port; FDB notification reached FdbOrch and entry appeared in FDB_TABLE.
  • FDB aging: verified via actual FDB aging/removal on a learned entry.

Details if related
ZMQ-mode dispatch path (Option 3):

1. libsairedis callback (ZMQ thread) → enqueueSaiNotification(op, data, values) + wake queue selectable
2. main loop → SaiNotificationQueueExecutor::execute()
3. readiness predicate check (per-op, e.g. allPortsReady() for port notifications)
4. SaiNotificationDispatcher → registered feature-orch handleNotification(entry)
   Non-ZMQ mode: unchanged — syncd publishes to ASIC_DB:NOTIFICATIONS; orchs consume via existing NotificationConsumer / Notifier path.
   There is no hybrid Option 2 + Option 3 path for the same notification type in ZMQ mode.

[mssonicbld]

/azp run

[azure-pipelines]

Azure Pipelines:
There may be pipelines that require an authorized user to comment /azp run to 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).

[azure-pipelines]

Azure Pipelines:
There may be pipelines that require an authorized user to comment /azp run to run.

[venkit-nexthop]

Reviewed against the HLD in sonic-net/sonic-buildimage#27791. The design is sound and matches Option 3 — the queue/dispatcher/executor split is clean, handler lookup correctly copies under the lock and invokes outside it, and the non-ZMQ Redis path is left untouched. Four issues below, one of which I would treat as blocking.

1. [Major] Busy-spin when a readiness predicate is false

SaiNotificationQueueExecutor::execute() returns without consuming anything when the front op is not ready:

std::string frontOp;
if (!m_queue->peekFrontOp(frontOp) || !m_dispatcher->isReady(frontOp))
{
    return;
}

But hasCachedData() returns hasData(), i.e. "queue non-empty". In Select::select() (sonic-swss-common common/select.cpp:153):

if (sel->hasCachedData())
{
    m_ready.insert(sel);   // reinserted; the loop at :137 calls it again immediately
}

So with a non-empty queue and a false predicate, execute() returns, hasCachedData() is true, the selectable is reinserted and called again — a tight loop inside a single select() call, burning a core until the predicate flips.

This looks reachable in practice. FdbOrch registers [this]() { return m_portsOrch->allPortsReady(); }, and FDB events arriving during startup are exactly the case where ports are not yet ready. Because the predicate can only become true through progress on the same main thread, this risks livelock rather than a brief spin.

Suggestion: when the front op is not ready, either return false from hasCachedData() until readiness changes, or defer with a short backoff/timer instead of relying on immediate reinsertion.

2. [Major] Readiness is checked only for the front entry, then the whole batch is dispatched

if (!m_queue->peekFrontOp(frontOp) || !m_dispatcher->isReady(frontOp)) return;

std::deque<swss::KeyOpFieldsValuesTuple> entries;
m_queue->pops(entries);          // up to popBatchSize (128)

for (auto &entry : entries)
{
    m_dispatcher->dispatch(entry);   // no per-entry readiness check
}

pops() takes up to 128 mixed-op entries, but readiness was validated only for entry 0. If a port_state_change is at the head and an fdb_event is at position 5, the FDB handler runs even though allPortsReady() is false — the exact condition the predicate exists to prevent.

Suggestion: check isReady() per entry inside the dispatch loop (deferring or re-queuing those that are not ready), or stop popping at the first not-ready op.

3. [Minor] The queue's priority is silently discarded

SaiNotificationQueue is constructed with pri = 100, but Select never sees it. Executor is handed a wrapper:

Executor(new SaiNotificationQueueSelectable(queue), orch, name)

and the wrapper does not forward the priority:

explicit SaiNotificationQueueSelectable(SaiNotificationQueue *queue) : m_queue(queue) {}

swss::Selectable(int pri = 0), so the wrapper registers at priority 0 while the intended 100 sits unused on the inner object. Either forward the priority through the wrapper or drop the parameter so it does not mislead.

Related: I assume the wrapper exists because Executor takes ownership of its Selectable and would otherwise delete the process-global queue. That is a reasonable lifetime shim, but it is not stated anywhere, and the indirection otherwise looks redundant given SaiNotificationQueue is already a Selectable. Worth a comment.

4. [Minor] The queue is unbounded

enqueue() has no cap. m_highWatermark is tracked and exposed but never enforced or logged. The previous Redis path had implicit backpressure; an in-process std::queue has none, so a notification burst against a stalled main loop grows without limit. Combined with issue 1, which can stall the drain indefinitely, these compound.

Suggestion: a maximum depth with a drop policy, plus a rate-limited warning when the watermark crosses a threshold.

Smaller notes

• registerHandler() silently overwrites an existing handler for the same op. A warning would catch accidental double-registration during the phased migration.
• enqueueSaiNotification() drops notifications when gOrchShutdownRequested != 0 with no counter or log. Intentional, but invisible if it ever happens mid-operation.
• getSaiNotificationQueue() / getSaiNotificationDispatcher() leak by design as process-lifetime singletons. That is fine, but std::call_once or a function-local static would express it more directly than a mutex plus null check.

What looks right

• The dispatcher copies the handler under the lock and calls it outside, so no handler runs while holding the dispatcher mutex.
• pops() re-notifies the eventfd when entries remain, so a partial drain does not stall.
• Non-ZMQ mode is genuinely untouched; the gRedisCommunicationMode == ZMQ_SYNC guard was already present and only the body changed.
• The old handleNotification(NotificationConsumer&, ...) overloads are removed rather than left as dead alternatives, avoiding two live paths.

This is a diff-only review; I have not built or run the change. Issue 1 in particular is worth confirming against Select's actual behaviour before acting on it.