diff --git a/src/libs/mcu/host/test_zmq_transport.cpp b/src/libs/mcu/host/test_zmq_transport.cpp index cd818d9..f7c695b 100644 --- a/src/libs/mcu/host/test_zmq_transport.cpp +++ b/src/libs/mcu/host/test_zmq_transport.cpp @@ -1,17 +1,24 @@ #include #include +#include #include #include #include +#include +#include #include #include +#include #include +#include #include #include +#include #include #include "dispatcher.hpp" +#include "libs/common/logger.hpp" #include "zmq_transport.hpp" namespace mcu { @@ -126,5 +133,170 @@ TEST_F(ZmqTransportTest, SendReceive) { ASSERT_EQ(response.value(), "World"); } +// Counts log lines, so a test can assert how many times Send() went round its +// retry loop rather than merely that it eventually failed. The mutex guards +// against the server thread logging concurrently with the test thread. +class CountingLogger : public common::Logger { + public: + auto Debug(std::string_view msg) -> void override { Record(msg); } + auto Info(std::string_view msg) -> void override { Record(msg); } + auto Warning(std::string_view msg) -> void override { Record(msg); } + auto Error(std::string_view msg) -> void override { Record(msg); } + + auto Count(std::string_view needle) const -> std::size_t { + const std::lock_guard lock(mutex_); + return static_cast( + std::ranges::count_if(messages_, [needle](const std::string& msg) { + return msg.find(needle) != std::string::npos; + })); + } + + private: + auto Record(std::string_view msg) -> void { + const std::lock_guard lock(mutex_); + messages_.emplace_back(msg); + } + + mutable std::mutex mutex_; + std::vector messages_; +}; + +// Drives Send() into genuine backpressure, and reports the send that met it. +// +// The obvious setup -- point the transport at an endpoint nobody binds -- does +// NOT work, despite the mute state PAIR sockets are documented to have. +// connect() creates the outbound pipe immediately whether or not a peer is +// reachable, and libzmq queues into it, so a peerless send returns success. +// Backpressure begins only once ZMQ_SNDHWM messages (1000 by default) are +// outstanding, and that is where the retry loop finally has something to +// absorb. Measuring the first send to block gives the retry path a clean +// reading: every send before it succeeded on its first attempt and logged +// nothing. +struct BlockedSend { + std::expected result; + std::chrono::steady_clock::duration elapsed; + int sends; +}; + +auto SendUntilQueueBlocks(ZmqTransport& transport) -> BlockedSend { + // Comfortably past the default high-water mark, and a backstop against a + // regression that made sends succeed forever rather than hanging the suite. + constexpr int kMaxSends{5000}; + + BlockedSend blocked{}; + for (blocked.sends = 1; blocked.sends <= kMaxSends; ++blocked.sends) { + const auto start{std::chrono::steady_clock::now()}; + blocked.result = transport.Send("Hello"); + blocked.elapsed = std::chrono::steady_clock::now() - start; + if (!blocked.result) { + break; + } + } + return blocked; +} + +class ZmqTransportRetryTest : public ::testing::Test { + protected: + void TearDown() override { + std::error_code error{}; + const std::string path{ + own_endpoint_.substr(std::string_view{"ipc://"}.size())}; + std::filesystem::remove(path, error); + std::filesystem::remove(path + ".lock", error); + } + + static auto MakeConfig( + common::Logger& logger, std::chrono::milliseconds send_timeout, + std::chrono::milliseconds total_timeout) -> TransportConfig { + // TransportConfig has user-provided constructors, so it is not an + // aggregate: designated initialisers will not compile. Assign after + // construction. + TransportConfig config{logger}; + config.send_timeout = send_timeout; + config.retry.max_attempts = kAttempts; + config.retry.retry_delay = kRetryDelay; + config.retry.total_timeout = total_timeout; + return config; + } + + static constexpr uint32_t kAttempts{3}; + static constexpr std::chrono::milliseconds kRetryDelay{10}; + + const std::string absent_peer_endpoint_{Endpoint("absent_peer")}; + const std::string own_endpoint_{Endpoint("retry_own")}; + CountingLogger logger_; + const ReceiverMap receiver_map_; +}; + +// The regression test the fix owes: a blocked send must make every attempt it +// was configured for. +// +// Three assertions that fail independently, because each admits a different +// wrong answer on its own. kTimeout says the outcome was classified as +// retryable at all -- before the fix an expired ZMQ_SNDTIMEO came back as a +// falsy result rather than an exception and was reported as kOperationFailed, +// so the loop never even reached its deadline check. The log count says the +// loop iterated. The elapsed time says each iteration blocked on the socket for +// its own timeout instead of failing instantly. +TEST_F(ZmqTransportRetryTest, SendAttemptsEveryRetryWhenSocketBlocks) { + constexpr std::chrono::milliseconds kSendTimeout{100}; + constexpr std::chrono::milliseconds kTotalTimeout{2000}; + + Dispatcher dispatcher{receiver_map_}; + auto config = MakeConfig(logger_, kSendTimeout, kTotalTimeout); + auto transport = ZmqTransport::Create(absent_peer_endpoint_, own_endpoint_, + dispatcher, config); + ASSERT_TRUE(transport.has_value()); + ASSERT_TRUE((*transport)->IsReady()); + + const auto blocked = SendUntilQueueBlocks(**transport); + + ASSERT_FALSE(blocked.result.has_value()) + << "no send ever met backpressure in " << blocked.sends << " attempts"; + EXPECT_EQ(blocked.result.error(), common::Error::kTimeout); + EXPECT_EQ(logger_.Count("Send retrying"), kAttempts - 1) + << "the retry loop ran " << logger_.Count("Send retrying") + 1 + << " of its " << kAttempts << " configured attempts"; + // Lower bound only: an upper bound here would measure the machine's load + // rather than this code. Two attempts' worth of blocking cannot fit in one. + EXPECT_GE(blocked.elapsed, 2 * kSendTimeout); + EXPECT_LT(blocked.elapsed, kTotalTimeout) << "the whole-send budget overran"; +} + +// The shape of the configuration that shipped: one attempt allowed to consume +// the entire retry budget, so the first EAGAIN arrives at the deadline and +// max_attempts and retry_delay describe nothing that can happen. +// +// The transport now shrinks the per-attempt slice until the configured attempts +// fit, so the caller's stated intent -- three tries within this budget -- is +// what they get. +TEST_F(ZmqTransportRetryTest, SendRetriesWhenSendTimeoutClaimsTheWholeBudget) { + constexpr std::chrono::milliseconds kBudget{600}; + // (600ms - 2 * 10ms of retry delay) / 3 attempts. + constexpr std::chrono::milliseconds kExpectedSlice{193}; + + Dispatcher dispatcher{receiver_map_}; + auto config = MakeConfig(logger_, kBudget, kBudget); + auto transport = ZmqTransport::Create(absent_peer_endpoint_, own_endpoint_, + dispatcher, config); + ASSERT_TRUE(transport.has_value()); + ASSERT_TRUE((*transport)->IsReady()); + + const auto blocked = SendUntilQueueBlocks(**transport); + + ASSERT_FALSE(blocked.result.has_value()) + << "no send ever met backpressure in " << blocked.sends << " attempts"; + EXPECT_EQ(blocked.result.error(), common::Error::kTimeout); + EXPECT_EQ(logger_.Count("send_timeout exceeds"), 1U) + << "an incoherent budget should be clamped, and said so"; + EXPECT_EQ(logger_.Count("Send retrying"), kAttempts - 1); + // Bounded both ways, each side ruling out a different wrong answer. Below: + // the attempts genuinely blocked rather than failing instantly. Above: they + // were shortened to share the budget, not repeated at full length -- which + // would have taken three times as long. + EXPECT_GE(blocked.elapsed, 2 * kExpectedSlice); + EXPECT_LT(blocked.elapsed, 2 * kBudget); +} + } // namespace } // namespace mcu diff --git a/src/libs/mcu/host/zmq_transport.cpp b/src/libs/mcu/host/zmq_transport.cpp index 286db9e..708ecbb 100644 --- a/src/libs/mcu/host/zmq_transport.cpp +++ b/src/libs/mcu/host/zmq_transport.cpp @@ -7,7 +7,9 @@ #include #include +#include #include +#include #include #include #include @@ -151,6 +153,7 @@ ZmqTransport::ZmqTransport(const std::string& to_emulator, // NOLINT // an exception leaving this constructor destroys a *joinable* std::thread, // which calls std::terminate -- Create()'s handler never gets a look in. try { + ClampSendTimeoutToRetryBudget(); SetSocketOptions(); // Start server thread FIRST (it will BIND) @@ -212,6 +215,38 @@ auto ZmqTransport::SignalBind(BindOutcome outcome) -> void { bind_cv_.notify_all(); } +// Makes RetryConfig's invariant true rather than merely documented. +// +// ZMQ_SNDTIMEO bounds a single attempt; retry.total_timeout bounds the whole +// Send(). If one attempt may consume the entire budget then the first EAGAIN +// arrives at or after the deadline and Send() returns having tried once -- +// max_attempts and retry_delay become dead configuration. That was the shipped +// default: send_timeout and total_timeout were both 1000ms. +// +// Shrinking the per-attempt slice is preferred to rejecting the config. A +// caller who asks for three attempts within a second has said something +// coherent about what they want; the arithmetic that makes it fit is ours to +// do, and failing startup over a tuning number would be a worse answer. +auto ZmqTransport::ClampSendTimeoutToRetryBudget() -> void { + // Zero attempts would skip the loop entirely and report a timeout without + // ever touching the socket. + config_.retry.max_attempts = std::max(config_.retry.max_attempts, 1U); + const auto attempts{config_.retry.max_attempts}; + + const auto delays{(attempts - 1) * config_.retry.retry_delay}; + const auto sending{delays < config_.retry.total_timeout + ? config_.retry.total_timeout - delays + : std::chrono::milliseconds::zero()}; + // A floor of 1ms, because 0 means "never block" and a negative value means + // "block forever" -- both worse than a very short attempt. + const auto slice{std::max(std::chrono::milliseconds{1}, sending / attempts)}; + + if (config_.send_timeout > slice) { + LogWarning("send_timeout exceeds the per-attempt retry budget; clamping"); + config_.send_timeout = slice; + } +} + auto ZmqTransport::SetSocketOptions() -> void { // Set linger to 0 to discard messages immediately on close to_emulator_socket_.set(zmq::sockopt::linger, config_.linger_ms); @@ -292,6 +327,30 @@ ZmqTransport::~ZmqTransport() { } } +// One attempt at the socket, classified. +// +// The classification is the substance here. cppzmq's send() reports an expired +// ZMQ_SNDTIMEO by returning an EMPTY result and throws error_t only for +// everything else -- so the timeout this retry loop exists to absorb is a falsy +// return, not an exception. The previous code looked for it exclusively in a +// catch block, treated the falsy return as a hard failure, and so left the +// retry path unreachable even once the timeouts allowed for it. +// +// ETIMEDOUT is still caught for the same outcome: no libzmq version in use +// raises it here, but it means precisely what EAGAIN means and costs one line. +auto ZmqTransport::TrySendOnce(std::string_view data) -> SendAttempt { + try { + const auto result{ + to_emulator_socket_.send(zmq::buffer(data), zmq::send_flags::none)}; + return result ? SendAttempt::kSent : SendAttempt::kWouldBlock; + } catch (const zmq::error_t& e) { + if (e.num() == EAGAIN || e.num() == ETIMEDOUT) { + return SendAttempt::kWouldBlock; + } + return SendAttempt::kFailed; + } +} + auto ZmqTransport::Send(std::string_view data) -> std::expected { if (state_.load() != TransportState::kReady) { @@ -299,53 +358,40 @@ auto ZmqTransport::Send(std::string_view data) return std::unexpected(common::Error::kInvalidState); } - // Calculate deadline for retry timeout + // The whole-send budget. Each attempt is separately bounded by the socket's + // ZMQ_SNDTIMEO, which the constructor sized to fit max_attempts of them in + // here; this deadline is the backstop for a peer that keeps us just under it. const auto deadline{std::chrono::steady_clock::now() + config_.retry.total_timeout}; - // Retry loop for (uint32_t attempt = 0; attempt < config_.retry.max_attempts; ++attempt) { - try { - auto result{ - to_emulator_socket_.send(zmq::buffer(data), zmq::send_flags::none)}; - if (result) { + switch (TrySendOnce(data)) { + case SendAttempt::kSent: if (attempt > 0) { LogDebug("Send succeeded after retry"); } - return {}; // Success! - } - } catch (const zmq::error_t& e) { - // Check if error is retryable - if (e.num() == EAGAIN || e.num() == ETIMEDOUT) { - // Check if we've exceeded total timeout - if (std::chrono::steady_clock::now() >= deadline) { - LogError("Send timeout after retries"); - return std::unexpected(common::Error::kTimeout); - } - - if (attempt + 1 < config_.retry.max_attempts) { - LogDebug("Send retrying after transient error"); - } - - // Wait before retry (unless this was the last attempt) - if (attempt + 1 < config_.retry.max_attempts) { - std::this_thread::sleep_for(config_.retry.retry_delay); - } - continue; // Retry - } - - // Non-retryable error - LogError("Send failed with non-retryable error"); - return std::unexpected(common::Error::kOperationFailed); + return {}; + case SendAttempt::kFailed: + LogError("Send failed with non-retryable error"); + return std::unexpected(common::Error::kOperationFailed); + case SendAttempt::kWouldBlock: + break; } - // result was false but no exception - operation failed - LogError("Send operation returned false"); - return std::unexpected(common::Error::kOperationFailed); + if (attempt + 1 >= config_.retry.max_attempts) { + break; // Out of attempts. + } + if (std::chrono::steady_clock::now() >= deadline) { + break; // Out of budget. + } + LogDebug("Send retrying after transient error"); + std::this_thread::sleep_for(config_.retry.retry_delay); } - // Max attempts exceeded - LogError("Send failed: max attempts exceeded"); + // Every exhausted path is a timeout: a mute PAIR socket blocks rather than + // dropping, so running out of attempts and running out of budget both mean + // the peer never took the message. + LogError("Send timeout after retries"); return std::unexpected(common::Error::kTimeout); } diff --git a/src/libs/mcu/host/zmq_transport.hpp b/src/libs/mcu/host/zmq_transport.hpp index 0d989ac..852a813 100644 --- a/src/libs/mcu/host/zmq_transport.hpp +++ b/src/libs/mcu/host/zmq_transport.hpp @@ -25,8 +25,8 @@ namespace mcu { // ZMQ_PAIR, and ZMQ offers no connection callback without a socket monitor, // which this class deliberately does not use. Whether a peer is attached is // only ever knowable from the result of the operation you just attempted -- -// which is why a missing emulator surfaces as Send() returning kTimeout rather -// than as a state value. +// which is why a missing emulator surfaces as a failed Send() or Receive() +// rather than as a state value. enum class TransportState : uint8_t { // Constructed, startup not yet begun. Not observable from outside: the // constructor moves to kStarting before any other thread holds a reference. @@ -37,14 +37,30 @@ enum class TransportState : uint8_t { // socket. The only state in which Send()/Receive() are permitted. // // This does NOT mean a peer is attached. connect() is asynchronous and - // returns before any peer exists, so a Send() in this state can still sit in - // ZMQ's mute state for send_timeout and come back as kTimeout. + // returns before any peer exists. + // + // Nor does a successful Send() in this state mean one is: connect() creates + // the outbound pipe whether or not the far end is reachable, and libzmq + // queues into it, so sends to nobody succeed until ZMQ_SNDHWM messages are + // outstanding. Only past that point does a send block for send_timeout and + // come back as kTimeout. kReady, // Startup failed; terminal. Send()/Receive() return kInvalidState from here // on, and StartupStatus() carries the reason. kFailed, }; +// Bounds Send()'s retry loop. Read together with TransportConfig::send_timeout, +// which is the budget for ONE attempt while total_timeout is the budget for all +// of them: the two are only meaningful if +// +// max_attempts * send_timeout + (max_attempts - 1) * retry_delay +// <= total_timeout +// +// Give a single attempt the whole budget and the first EAGAIN already arrives +// at the deadline, so Send() gives up having tried exactly once and the two +// fields below describe behaviour that cannot happen. ZmqTransport enforces the +// invariant at construction by shrinking send_timeout to fit. struct RetryConfig { uint32_t max_attempts{3}; std::chrono::milliseconds retry_delay{10}; @@ -58,7 +74,10 @@ struct TransportConfig { // connect() on a PAIR socket is asynchronous and completes without a peer, // so there would be nothing to wait for. std::chrono::milliseconds startup_timeout{5000}; - std::chrono::milliseconds send_timeout{1000}; + // One attempt's worth of ZMQ_SNDTIMEO, not the whole send. The default is + // sized so all of retry.max_attempts fit inside retry.total_timeout with the + // inter-attempt delays: 3 * 300ms + 2 * 10ms = 920ms <= 1000ms. + std::chrono::milliseconds send_timeout{300}; std::chrono::milliseconds recv_timeout{5000}; int linger_ms{0}; // Discard pending messages on close RetryConfig retry{}; @@ -152,6 +171,13 @@ class ZmqTransport : public Transport { private: enum class BindOutcome : uint8_t { kPending, kBound, kFailed }; + // How one send attempt ended. kWouldBlock is the only retryable outcome, and + // it deliberately covers both ways ZMQ reports a full/mute socket -- see + // TrySendOnce. + enum class SendAttempt : uint8_t { kSent, kWouldBlock, kFailed }; + + auto TrySendOnce(std::string_view data) -> SendAttempt; + auto ClampSendTimeoutToRetryBudget() -> void; auto ServerThread(const std::string& endpoint) -> void; auto ServeLoop(zmq::socket_t& socket) -> void; auto SignalBind(BindOutcome outcome) -> void;