[zmqclient]: Add send-path blip absorber, back-pressure counters, and inner retry cap

common/zmqclient.cpp

@@ -15,6 +15,10 @@ using namespace std;
 
 namespace swss {
 
+// Ceiling for the exponential send-retry backoff; bounds the doubling so it
+// cannot overflow int and feed usleep() a garbage duration.
+static const int MQ_SEND_RETRY_DELAY_CEILING_MS = 60000;
+
 ZmqClient::ZmqClient(const std::string& endpoint)
     : ZmqClient(endpoint, "")
 {
@@ -150,6 +154,15 @@ void ZmqClient::connect()
     m_connected = true;
 }
 
+void ZmqClient::setSendRetryConfig(int maxRetries, int maxBackoffMs)
+{
+    // Clamp to MQ_MAX_RETRY so this can only shorten the default ladder.
+    m_sendMaxRetries.store((maxRetries < 0) ? MQ_MAX_RETRY
+                                            : std::min(maxRetries, MQ_MAX_RETRY),
+                           std::memory_order_relaxed);
+    m_sendMaxBackoffMs.store(maxBackoffMs, std::memory_order_relaxed);
+}
+
 void ZmqClient::sendMsg(
         const std::string& dbName,
         const std::string& tableName,
@@ -173,7 +186,12 @@ void ZmqClient::sendMsg(
     int zmq_err = 0;
     int retry_delay = 10;
     int rc = 0;
-    for (int i = 0; i <= MQ_MAX_RETRY; ++i)
+    bool saw_eagain = false;
+    // Snapshot the caps so this send reads stable values (they are atomic and
+    // may be reconfigured between sends).
+    const int max_retries = m_sendMaxRetries.load(std::memory_order_relaxed);
+    const int max_backoff_ms = m_sendMaxBackoffMs.load(std::memory_order_relaxed);
+    for (int i = 0; i <= max_retries; ++i)
     {
         {
             // ZMQ socket is not thread safe: http://api.zeromq.org/2-1:zmq
@@ -191,13 +209,27 @@ void ZmqClient::sendMsg(
         }
         if (rc >= 0)
         {
+            // Absorbed a transient full-socket blip; caller never had to re-queue.
+            if (saw_eagain)
+            {
+                m_sendBlipAbsorbedTotal.fetch_add(1, std::memory_order_relaxed);
+            }
             SWSS_LOG_DEBUG("zmq sended %d bytes", serializedlen);
             return;
         }
 
         zmq_err = zmq_errno();
-        // sleep (2 ^ retry time) * 10 ms
-        retry_delay *= 2;
+        // sleep (2 ^ retry time) * 10 ms. Double in a wider type and clamp to the
+        // ceiling before narrowing, so the multiply cannot overflow int; then
+        // apply the caller's cap. All three (log/record/sleep) see the real delay.
+        int64_t doubled = static_cast<int64_t>(retry_delay) * 2;
+        retry_delay = (doubled > MQ_SEND_RETRY_DELAY_CEILING_MS)
+                          ? MQ_SEND_RETRY_DELAY_CEILING_MS
+                          : static_cast<int>(doubled);
+        if (max_backoff_ms >= 0 && retry_delay > max_backoff_ms)
+        {
+            retry_delay = max_backoff_ms;
+        }
         if (zmq_err == EINTR
             || zmq_err== EFSM)
         {
@@ -213,7 +245,21 @@ void ZmqClient::sendMsg(
         else if (zmq_err == EAGAIN)
         {
             // EAGAIN: ZMQ is full to need try again
-            SWSS_LOG_WARN("zmq is full, will retry in %d ms, endpoint: %s, error: %d", retry_delay, m_endpoint.c_str(), zmq_err);
+            saw_eagain = true;
+            m_sendEagainTotal.fetch_add(1, std::memory_order_relaxed);
+            // Record/log backoff only when a retry follows; the final attempt throws without waiting.
+            if (i < max_retries)
+            {
+                // Deepest backoff waited (congestion-depth gauge).
+                uint64_t observed = m_sendBackoffMaxMs.load(std::memory_order_relaxed);
+                while (static_cast<uint64_t>(retry_delay) > observed &&
+                       !m_sendBackoffMaxMs.compare_exchange_weak(observed,
+                                                                 static_cast<uint64_t>(retry_delay),
+                                                                 std::memory_order_relaxed))
+                {
+                }
+                SWSS_LOG_WARN("zmq is full, will retry in %d ms, endpoint: %s, error: %d", retry_delay, m_endpoint.c_str(), zmq_err);
+            }
         }
         else if (zmq_err == ETERM)
         {
@@ -230,10 +276,14 @@ void ZmqClient::sendMsg(
             throw system_error(make_error_code(errc::io_error), message);
         }
 
-        usleep(retry_delay * 1000);
+        // No sleep after the final attempt: the loop is about to throw.
+        if (i < max_retries)
+        {
+            usleep(retry_delay * 1000);
+        }
     }
 
-    // failed after retry
+    // Inner retries exhausted; caller owns the outer retry. Surface by throwing.
     auto message =  "zmq send failed, endpoint: " + m_endpoint + ", zmqerrno: " + to_string(zmq_err) + ":" + zmq_strerror(zmq_err) + ", msg length:" + to_string(serializedlen);
     SWSS_LOG_ERROR("%s", message.c_str());
     throw system_error(make_error_code(errc::io_error), message);

common/zmqclient.h

@@ -5,6 +5,8 @@
 #include <queue>
 #include <thread> 
 #include <mutex> 
+#include <atomic>
+#include <cstdint>
 #include "zmqserver.h"
 
 namespace swss {
@@ -34,6 +36,19 @@ class ZmqClient
               std::string& tableName,
               std::vector<std::shared_ptr<KeyOpFieldsValuesTuple>>& kcos);
 
+    // Optionally shorten the inner send-retry loop. Typically configured once
+    // at setup, but the values are atomic so they may also be changed between
+    // sends (each send snapshots them for a consistent read).
+    // maxRetries   < 0 keeps the default ladder; otherwise clamped to MQ_MAX_RETRY.
+    // maxBackoffMs < 0 keeps the exponential backoff; >= 0 caps each retry sleep (ms).
+    void setSendRetryConfig(int maxRetries, int maxBackoffMs = -1);
+
+    // Process-local send-path back-pressure counters (the socket-full signal
+    // sendMsg() otherwise only logs). ZmqClient only counts; the owner publishes.
+    uint64_t getSendEagainTotal() const { return m_sendEagainTotal.load(std::memory_order_relaxed); }        // total EAGAIN occurrences
+    uint64_t getSendBlipAbsorbedTotal() const { return m_sendBlipAbsorbedTotal.load(std::memory_order_relaxed); } // EAGAIN sends that still succeeded
+    uint64_t getSendBackoffMaxMs() const { return m_sendBackoffMaxMs.load(std::memory_order_relaxed); }       // deepest backoff waited (ms)
+
 private:
     void initialize(const std::string& endpoint, const std::string& vrf = "");
 
@@ -55,6 +70,17 @@ class ZmqClient
     std::mutex m_socketMutex;
 
     std::vector<char> m_sendbuffer;
+
+    // Inner send-retry caps (see setSendRetryConfig); defaults: MQ_MAX_RETRY, uncapped back-off.
+    std::atomic<int> m_sendMaxRetries{MQ_MAX_RETRY};
+    std::atomic<int> m_sendMaxBackoffMs{-1};
+
+    // Send-path back-pressure counters (see getters above). Appended members;
+    // all swss-common consumers rebuild from source, and no existing call site
+    // changes.
+    std::atomic<uint64_t> m_sendEagainTotal{0};
+    std::atomic<uint64_t> m_sendBlipAbsorbedTotal{0};
+    std::atomic<uint64_t> m_sendBackoffMaxMs{0};
 };
 
 }

tests/zmq_state_ut.cpp

@@ -455,6 +455,67 @@ TEST(ZmqConsumerStateTableBatchBufferOverflow, test)
     EXPECT_ANY_THROW(p.send(kcos));
 }
 
+TEST(ZmqClientSendPathCounters, blipAbsorberExhaustionCountsEagainAndClampsBackoff)
+{
+    // A peerless PUSH socket buffers locally up to SNDHWM, then EAGAINs
+    // deterministically once full (no peer ever drains it) — lets us assert
+    // exact counters without timing dependence. Per-process ipc endpoint avoids
+    // path collisions across parallel runs.
+    std::string deadEndpoint = "ipc:///tmp/zmqclient_ut_mute_" + std::to_string(getpid());
+    ZmqClient client(deadEndpoint, 0);
+
+    std::vector<KeyOpFieldsValuesTuple> kcos;
+    kcos.push_back(KeyOpFieldsValuesTuple("k0", SET_COMMAND,
+                   std::vector<FieldValueTuple>{FieldValueTuple("f0", "v0")}));
+
+    EXPECT_EQ(client.getSendEagainTotal(), 0u);
+    EXPECT_EQ(client.getSendBlipAbsorbedTotal(), 0u);
+    EXPECT_EQ(client.getSendBackoffMaxMs(), 0u);
+
+    // Phase 1 — fill the buffer (single attempt, zero backoff): the overflowing
+    // send throws after exactly one EAGAIN.
+    client.setSendRetryConfig(0, 0);
+    bool bufferFull = false;
+    for (int i = 0; i < MQ_WATERMARK * 2; ++i)
+    {
+        try
+        {
+            client.sendMsg(TEST_DB, "SEND_PATH_COUNTER_UT", kcos);
+        }
+        catch (const std::system_error&)
+        {
+            bufferFull = true;
+            break;
+        }
+    }
+    ASSERT_TRUE(bufferFull);
+    EXPECT_EQ(client.getSendEagainTotal(), 1u);
+    EXPECT_EQ(client.getSendBlipAbsorbedTotal(), 0u);
+    EXPECT_EQ(client.getSendBackoffMaxMs(), 0u);
+
+    // Phase 2 — buffer stays saturated: one capped sendMsg() makes
+    // (maxRetries + 1) attempts, each counted, then throws. Exercises the eagain
+    // counter, the caller backoff cap, and the overflow guard.
+    const int maxRetries = 3;
+    const int maxBackoffMs = 5;
+    client.setSendRetryConfig(maxRetries, maxBackoffMs);
+    EXPECT_THROW(client.sendMsg(TEST_DB, "SEND_PATH_COUNTER_UT", kcos), std::system_error);
+
+    EXPECT_EQ(client.getSendEagainTotal(), static_cast<uint64_t>(1 + maxRetries + 1));
+    EXPECT_EQ(client.getSendBlipAbsorbedTotal(), 0u);
+    EXPECT_EQ(client.getSendBackoffMaxMs(), static_cast<uint64_t>(maxBackoffMs));
+}
+
+TEST(ZmqClientSendPathCounters, defaultCountersStartZeroAndAccessorsLink)
+{
+    // Counters read zero before any send and the appended accessors link.
+    std::string endpoint = "ipc:///tmp/zmqclient_ut_default_" + std::to_string(getpid());
+    ZmqClient client(endpoint, 0);
+    EXPECT_EQ(client.getSendEagainTotal(), 0u);
+    EXPECT_EQ(client.getSendBlipAbsorbedTotal(), 0u);
+    EXPECT_EQ(client.getSendBackoffMaxMs(), 0u);
+}
+
 TEST(ZmqProducerStateTableDeleteAfterSend, test)
 {
     std::string testTableName = "ZMQ_PROD_DELETE_UT";