From 4b8eafd6cbef89d3e3712855ad2f9fbc37745a9d Mon Sep 17 00:00:00 2001 From: fhgffy <102001626+fhgffy@users.noreply.github.com> Date: Fri, 2 Oct 2026 10:02:27 -0700 Subject: [PATCH 1/2] Problem: multipart send drops unsent frames Solution: remove each part only after a successful send and test retry behavior after EAGAIN, backpressure and EFSM. Fixes #659. --- tests/multipart.cpp | 113 ++++++++++++++++++++++++++++++++++++++++++++ zmq_addon.hpp | 5 +- 2 files changed, 116 insertions(+), 2 deletions(-) diff --git a/tests/multipart.cpp b/tests/multipart.cpp index 7a67372..504f75e 100644 --- a/tests/multipart.cpp +++ b/tests/multipart.cpp @@ -1,10 +1,123 @@ #include #include +#include "testutil.hpp" #include #ifdef ZMQ_HAS_RVALUE_REFS +TEST_CASE("multipart send retains parts on EAGAIN", "[multipart]") +{ + const int part_count = GENERATE(1, 3); + const std::string first = GENERATE(std::string(), std::string("first")); + zmq::context_t context; + zmq::socket_t output(context, zmq::socket_type::push); + output.set(zmq::sockopt::linger, 0); + output.bind("inproc://multipart.send.retry"); + zmq::multipart_t message; + message.addstr(first); + if (part_count > 1) { + message.addstr(std::string(256, '\0')); + message.addstr("last"); + } + const zmq::multipart_t expected = message.clone(); + + SECTION("nonblocking integer flags") + { + for (int attempt = 0; attempt < 2; ++attempt) { + REQUIRE_FALSE(message.send(output, ZMQ_DONTWAIT)); + REQUIRE(message == expected); + } + } + SECTION("timed send_flags overload") + { + output.set(zmq::sockopt::sndtimeo, 1); + for (int attempt = 0; attempt < 2; ++attempt) { + REQUIRE_FALSE(message.send(output, zmq::send_flags::none)); + REQUIRE(message == expected); + } + } + + zmq::socket_t input(context, zmq::socket_type::pull); + input.set(zmq::sockopt::rcvtimeo, 1000); + input.connect("inproc://multipart.send.retry"); + output.set(zmq::sockopt::sndtimeo, 1000); + REQUIRE(message.send(output)); + CHECK(message.empty()); + zmq::multipart_t received; + REQUIRE(received.recv(input)); + CHECK(received == expected); +} + +TEST_CASE("multipart send retains parts at high water mark", "[multipart]") +{ + zmq::context_t context; + zmq::socket_t output(context, zmq::socket_type::push); + zmq::socket_t input(context, zmq::socket_type::pull); + output.set(zmq::sockopt::linger, 0); + output.set(zmq::sockopt::sndhwm, 1); + output.set(zmq::sockopt::sndtimeo, 1000); + input.set(zmq::sockopt::rcvhwm, 1); + input.set(zmq::sockopt::rcvtimeo, 1000); + output.bind("inproc://multipart.send.hwm"); + input.connect("inproc://multipart.send.hwm"); + + int queued = 0; + while (queued < 16 + && output.send(zmq::str_buffer("queued"), zmq::send_flags::dontwait)) + ++queued; + REQUIRE(queued > 0); + REQUIRE(queued < 16); + + zmq::multipart_t message; + message.addstr("first"); + message.addstr("last"); + const zmq::multipart_t expected = message.clone(); + REQUIRE_FALSE(message.send(output, zmq::send_flags::dontwait)); + REQUIRE(message == expected); + + for (int i = 0; i < queued; ++i) { + zmq::message_t part; + REQUIRE(input.recv(part)); + CHECK(part.to_string() == "queued"); + } + REQUIRE(message.send(output)); + CHECK(message.empty()); + zmq::multipart_t received; + REQUIRE(received.recv(input)); + CHECK(received == expected); +} + +TEST_CASE("multipart send retains parts on exception", "[multipart]") +{ + zmq::context_t context; + zmq::socket_t output(context, zmq::socket_type::rep); + zmq::socket_t input(context, zmq::socket_type::req); + output.set(zmq::sockopt::linger, 0); + output.set(zmq::sockopt::sndtimeo, 1000); + output.set(zmq::sockopt::rcvtimeo, 1000); + input.set(zmq::sockopt::linger, 0); + input.set(zmq::sockopt::sndtimeo, 1000); + input.set(zmq::sockopt::rcvtimeo, 1000); + output.bind("inproc://multipart.send.exception"); + input.connect("inproc://multipart.send.exception"); + zmq::multipart_t message; + message.addstr("first"); + message.addstr("last"); + const zmq::multipart_t expected = message.clone(); + CHECK_THROWS_ZMQ_ERROR(EFSM, message.send(output)); + REQUIRE(message == expected); + + REQUIRE(input.send(zmq::str_buffer("request"))); + zmq::message_t request; + REQUIRE(output.recv(request)); + REQUIRE(message.send(output)); + CHECK(message.empty()); + zmq::multipart_t received; + REQUIRE(received.recv(input)); + CHECK(received == expected); +} + #ifdef ZMQ_CPP17 using multipart_send_int_t = bool (zmq::multipart_t::*)(zmq::socket_ref, int); using multipart_send_flags_t = bool (zmq::multipart_t::*)(zmq::socket_ref, diff --git a/zmq_addon.hpp b/zmq_addon.hpp index f5a3b19..01401af 100644 --- a/zmq_addon.hpp +++ b/zmq_addon.hpp @@ -476,8 +476,8 @@ class multipart_t flags &= ~(ZMQ_SNDMORE); bool more = size() > 0; while (more) { - message_t message = pop(); - more = size() > 0; + message_t &message = m_parts.front(); + more = size() > 1; #ifdef ZMQ_CPP11 if (!socket.send(message, static_cast( (more ? ZMQ_SNDMORE : 0) | flags))) @@ -486,6 +486,7 @@ class multipart_t if (!socket.send(message, (more ? ZMQ_SNDMORE : 0) | flags)) return false; #endif + m_parts.pop_front(); } clear(); return true; From 9347af5e4fd89179e322a59d653e60e00141ff47 Mon Sep 17 00:00:00 2001 From: fhgffy <102001626+fhgffy@users.noreply.github.com> Date: Mon, 5 Oct 2026 05:47:49 -0400 Subject: [PATCH 2/2] Problem: multipart tests fail formatting Solution: format the existing static_assert and cover empty sends and caller SNDMORE through both flag overloads. Keep the corrected send loop after peer-close testing exposed an abort in direct helper delegation. --- tests/multipart.cpp | 54 ++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 49 insertions(+), 5 deletions(-) diff --git a/tests/multipart.cpp b/tests/multipart.cpp index 504f75e..4a49646 100644 --- a/tests/multipart.cpp +++ b/tests/multipart.cpp @@ -6,6 +6,51 @@ #ifdef ZMQ_HAS_RVALUE_REFS +TEST_CASE("multipart send handles empty messages", "[multipart]") +{ + zmq::context_t context; + zmq::socket_t output(context, zmq::socket_type::push); + zmq::multipart_t message; + REQUIRE(message.send(output, ZMQ_DONTWAIT | ZMQ_SNDMORE)); + REQUIRE( + message.send(output, zmq::send_flags::dontwait | zmq::send_flags::sndmore)); + CHECK(message.empty()); +} + +TEST_CASE("multipart send ignores caller SNDMORE", "[multipart]") +{ + const int part_count = GENERATE(1, 3); + zmq::context_t context; + zmq::socket_t output(context, zmq::socket_type::push); + zmq::socket_t input(context, zmq::socket_type::pull); + output.set(zmq::sockopt::linger, 0); + output.set(zmq::sockopt::sndtimeo, 1000); + input.set(zmq::sockopt::rcvtimeo, 1000); + output.bind("inproc://multipart.send.sndmore"); + input.connect("inproc://multipart.send.sndmore"); + zmq::multipart_t message; + message.addstr(""); + if (part_count > 1) { + message.addstr("middle"); + message.addstr("last"); + } + const zmq::multipart_t expected = message.clone(); + + SECTION("integer flags") + { + REQUIRE(message.send(output, ZMQ_SNDMORE)); + } + SECTION("send_flags overload") + { + REQUIRE(message.send(output, zmq::send_flags::sndmore)); + } + + CHECK(message.empty()); + zmq::multipart_t received; + REQUIRE(received.recv(input)); + CHECK(received == expected); +} + TEST_CASE("multipart send retains parts on EAGAIN", "[multipart]") { const int part_count = GENERATE(1, 3); @@ -122,11 +167,10 @@ TEST_CASE("multipart send retains parts on exception", "[multipart]") using multipart_send_int_t = bool (zmq::multipart_t::*)(zmq::socket_ref, int); using multipart_send_flags_t = bool (zmq::multipart_t::*)(zmq::socket_ref, zmq::send_flags); -static_assert(std::is_invocable::value, - "Can't multipart_t::send with socket_ref"); +static_assert( + std::is_invocable:: + value, + "Can't multipart_t::send with socket_ref"); static_assert(std::is_invocable