diff --git a/common/zmqclient.cpp b/common/zmqclient.cpp index 5ef7817b5..8aa648aad 100644 --- a/common/zmqclient.cpp +++ b/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(retry_delay) * 2; + retry_delay = (doubled > MQ_SEND_RETRY_DELAY_CEILING_MS) + ? MQ_SEND_RETRY_DELAY_CEILING_MS + : static_cast(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(retry_delay) > observed && + !m_sendBackoffMaxMs.compare_exchange_weak(observed, + static_cast(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); diff --git a/common/zmqclient.h b/common/zmqclient.h index 487f68fab..07c22b6ea 100644 --- a/common/zmqclient.h +++ b/common/zmqclient.h @@ -5,6 +5,8 @@ #include #include #include +#include +#include #include "zmqserver.h" namespace swss { @@ -34,6 +36,19 @@ class ZmqClient std::string& tableName, std::vector>& 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 m_sendbuffer; + + // Inner send-retry caps (see setSendRetryConfig); defaults: MQ_MAX_RETRY, uncapped back-off. + std::atomic m_sendMaxRetries{MQ_MAX_RETRY}; + std::atomic 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 m_sendEagainTotal{0}; + std::atomic m_sendBlipAbsorbedTotal{0}; + std::atomic m_sendBackoffMaxMs{0}; }; } diff --git a/tests/zmq_state_ut.cpp b/tests/zmq_state_ut.cpp index d81ef0f49..ce848ac01 100644 --- a/tests/zmq_state_ut.cpp +++ b/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 kcos; + kcos.push_back(KeyOpFieldsValuesTuple("k0", SET_COMMAND, + std::vector{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(1 + maxRetries + 1)); + EXPECT_EQ(client.getSendBlipAbsorbedTotal(), 0u); + EXPECT_EQ(client.getSendBackoffMaxMs(), static_cast(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";