Introduce keepalives for ZmqClient and ZmqServer

Fixes issue: sonic-net/sonic-buildimage#23110

When a DPU is powered off and back on, the ZMQ client on the switch still
holds a stale TCP connection. The first message sent after DPU restart is
delivered over the dead connection, gets a TCP RST, and is silently lost.
ZMQ then auto-reconnects, so subsequent messages succeed.

This patch enables:
1. TCP keepalive on ZmqClient PUSH sockets to detect dead connections
   proactively (within ~8 seconds of peer going down).
2. ZMQ_IMMEDIATE on ZmqClient PUSH sockets to prevent queueing messages
   to peers whose underlying TCP connection is not yet completed.
3. TCP keepalive on ZmqServer PULL sockets as defense-in-depth.

With these changes, after DPU power-off:
- TCP keepalive probes will fail, causing ZMQ to tear down the stale
  connection and reconnect
- ZMQ_IMMEDIATE prevents the first message from being queued to a peer
  with an incomplete connection, so it stays in the send queue until
  the reconnection completes

Signed-off-by: Prabhat Aravind <[email protected]>

Author: prabhataravind

common/zmqclient.cpp

@@ -109,6 +109,30 @@ void ZmqClient::connect()
         // Increase send buffer for use all bandwidth: http://api.zeromq.org/4-2:zmq-setsockopt
         int high_watermark = MQ_WATERMARK;
         zmq_setsockopt(m_socket, ZMQ_SNDHWM, &high_watermark, sizeof(high_watermark));
+
+        // Enable TCP keepalive to detect dead connections (e.g. DPU power-off).
+        // Without keepalive, a stale TCP connection persists indefinitely, and the
+        // first message sent after peer restart is silently lost on the dead connection.
+        // Keepalive probes will detect the dead peer within ~8s (5s idle + 3x1s probes),
+        // triggering ZMQ to reconnect before any new data is sent.
+        int keepalive = 1;
+        zmq_setsockopt(m_socket, ZMQ_TCP_KEEPALIVE, &keepalive, sizeof(keepalive));
+
+        int keepalive_idle = 5;  // seconds before first probe
+        zmq_setsockopt(m_socket, ZMQ_TCP_KEEPALIVE_IDLE, &keepalive_idle, sizeof(keepalive_idle));
+
+        int keepalive_intvl = 1; // seconds between probes
+        zmq_setsockopt(m_socket, ZMQ_TCP_KEEPALIVE_INTVL, &keepalive_intvl, sizeof(keepalive_intvl));
+
+        int keepalive_cnt = 3;   // number of failed probes before connection is considered dead
+        zmq_setsockopt(m_socket, ZMQ_TCP_KEEPALIVE_CNT, &keepalive_cnt, sizeof(keepalive_cnt));
+
+        // ZMQ_IMMEDIATE: only queue messages to completed connections.
+        // Prevents the first message from being sent over a not-yet-reconnected
+        // peer, which would result in silent message loss.
+        int immediate = 1;
+        zmq_setsockopt(m_socket, ZMQ_IMMEDIATE, &immediate, sizeof(immediate));
+
     }
 
     if (!m_vrf.empty())

common/zmqserver.cpp

@@ -90,6 +90,20 @@ void ZmqServer::bind()
         zmq_setsockopt(m_socket, ZMQ_RCVHWM, &high_watermark, sizeof(high_watermark));
     }
 
+    // Enable TCP keepalive on the server socket as defense-in-depth.
+    // This allows the server to detect and clean up stale client connections.
+    int keepalive = 1;
+    zmq_setsockopt(m_socket, ZMQ_TCP_KEEPALIVE, &keepalive, sizeof(keepalive));
+
+    int keepalive_idle = 5;
+    zmq_setsockopt(m_socket, ZMQ_TCP_KEEPALIVE_IDLE, &keepalive_idle, sizeof(keepalive_idle));
+
+    int keepalive_intvl = 1;
+    zmq_setsockopt(m_socket, ZMQ_TCP_KEEPALIVE_INTVL, &keepalive_intvl, sizeof(keepalive_intvl));
+
+    int keepalive_cnt = 3;
+    zmq_setsockopt(m_socket, ZMQ_TCP_KEEPALIVE_CNT, &keepalive_cnt, sizeof(keepalive_cnt));
+
     if (!m_vrf.empty())
     {   
         zmq_setsockopt(m_socket, ZMQ_BINDTODEVICE, m_vrf.c_str(), m_vrf.length());