Fix buffer leak on contended CONFLATE socket

Makefile.am

@@ -1374,6 +1374,7 @@ if ENABLE_STATIC
 test_apps += \
 	unittests/unittest_poller \
 	unittests/unittest_ypipe \
+	unittests/unittest_dbuffer \
 	unittests/unittest_mtrie \
 	unittests/unittest_ip_resolver \
 	unittests/unittest_udp_address \
@@ -1398,6 +1399,15 @@ unittests_unittest_ypipe_LDADD = \
         ${src_libzmq_la_LIBADD} \
         $(CODE_COVERAGE_LDFLAGS)
 
+unittests_unittest_dbuffer_SOURCES = unittests/unittest_dbuffer.cpp
+unittests_unittest_dbuffer_CPPFLAGS = -I$(top_srcdir)/src ${TESTUTIL_CPPFLAGS} $(CODE_COVERAGE_CPPFLAGS)
+unittests_unittest_dbuffer_CXXFLAGS = $(CODE_COVERAGE_CXXFLAGS)
+unittests_unittest_dbuffer_LDADD = \
+        ${TESTUTIL_LIBS} \
+        $(top_builddir)/src/.libs/libzmq.a \
+        ${src_libzmq_la_LIBADD} \
+        $(CODE_COVERAGE_LDFLAGS)
+
 unittests_unittest_mtrie_SOURCES = unittests/unittest_mtrie.cpp
 unittests_unittest_mtrie_CPPFLAGS = -I$(top_srcdir)/src ${TESTUTIL_CPPFLAGS} $(CODE_COVERAGE_CPPFLAGS)
 unittests_unittest_mtrie_CXXFLAGS = $(CODE_COVERAGE_CXXFLAGS)

src/dbuffer.hpp

@@ -46,6 +46,13 @@ template <> class dbuffer_t<msg_t>
     void write (const msg_t &value_)
     {
         zmq_assert (value_.check ());
+
+        //  If a previous write faced a locked buffer (the reader was busy),
+        //  the unswapped message is still owned by _back.
+        //  Release it before overwriting, otherwise its content is leaked.
+        const int rc = _back->close ();
+        errno_assert (rc == 0);
+
         *_back = value_;
 
         zmq_assert (_back->check ());

unittests/CMakeLists.txt

@@ -3,6 +3,7 @@ cmake_minimum_required(VERSION 2.8.1...3.31)
 
 set(unittests
     unittest_ypipe
+    unittest_dbuffer
     unittest_poller
     unittest_mtrie
     unittest_ip_resolver

unittests/unittest_dbuffer.cpp

@@ -0,0 +1,162 @@
+/* SPDX-License-Identifier: MPL-2.0 */
+
+#include "../tests/testutil.hpp"
+
+//  On Windows testutil.hpp maps close () to closesocket ().
+//  Undo this so it doesn't rewrite zmq::msg_t::close in other headers.
+#ifdef close
+#undef close
+#endif
+
+#include <atomic_counter.hpp>
+#include <dbuffer.hpp>
+#include <msg.hpp>
+#include <thread.hpp>
+
+#include <unity.h>
+
+void setUp ()
+{
+}
+void tearDown ()
+{
+}
+
+//  All test messages share one payload buffer.
+//  Ownership is tracked per message using the hint field, which points at
+//  a counter for how many times the content was released.
+static char shared_payload[129];
+
+static void count_free (void *data_, void *hint_)
+{
+    LIBZMQ_UNUSED (data_);
+    ++*static_cast<int *> (hint_);
+}
+
+//  Writes a message owning 'shared_payload' into the dbuffer and gives up
+//  ownership, the way pipe writers do. 'freed_' is incremented when the
+//  dbuffer releases the message content.
+static void write_counted_msg (zmq::dbuffer_t<zmq::msg_t> &buf_, int *freed_)
+{
+    zmq::msg_t msg;
+    TEST_ASSERT_EQUAL_INT (0,
+                           msg.init_data (shared_payload, sizeof shared_payload,
+                                          count_free, freed_));
+    buf_.write (msg);
+    TEST_ASSERT_EQUAL_INT (0, msg.init ());
+}
+
+void test_create ()
+{
+    zmq::dbuffer_t<zmq::msg_t> buf;
+    TEST_ASSERT_FALSE (buf.check_read ());
+}
+
+void test_read_empty ()
+{
+    zmq::dbuffer_t<zmq::msg_t> buf;
+    zmq::msg_t msg;
+    TEST_ASSERT_EQUAL_INT (0, msg.init ());
+    TEST_ASSERT_FALSE (buf.read (&msg));
+    TEST_ASSERT_EQUAL_INT (0, msg.close ());
+}
+
+//  Uncontended conflation.
+//  Second write should replace the first message and release its content.
+//  The reader must only observe the second message.
+void test_conflation_releases_overwritten_message ()
+{
+    int freed_first = 0, freed_second = 0;
+
+    zmq::dbuffer_t<zmq::msg_t> buf;
+    write_counted_msg (buf, &freed_first);
+    write_counted_msg (buf, &freed_second);
+    TEST_ASSERT_EQUAL_INT (1, freed_first);
+    TEST_ASSERT_EQUAL_INT (0, freed_second);
+
+    zmq::msg_t msg;
+    TEST_ASSERT_EQUAL_INT (0, msg.init ());
+    TEST_ASSERT_TRUE (buf.read (&msg));
+    TEST_ASSERT_EQUAL_INT (sizeof shared_payload, msg.size ());
+    TEST_ASSERT_EQUAL_INT (0, msg.close ());
+    TEST_ASSERT_EQUAL_INT (1, freed_second);
+
+    TEST_ASSERT_FALSE (buf.read (&msg));
+}
+
+//  Helpers for parking a reader inside probe (), holding the dbuffer lock.
+//  This allows deterministically causing writes to fail try_lock.
+static zmq::atomic_counter_t probe_entered;
+static zmq::atomic_counter_t probe_release;
+
+static bool blocking_probe_fn (const zmq::msg_t &)
+{
+    probe_entered.add (1);
+    while (probe_release.get () == 0)
+        msleep (1);
+    return true;
+}
+
+static void probe_thread_fn (void *arg_)
+{
+    static_cast<zmq::dbuffer_t<zmq::msg_t> *> (arg_)->probe (blocking_probe_fn);
+}
+
+//  A write to a dbuffer locked by a reader stores the msg in the _back slot.
+//  Subsequent writes must release that message instead of overwriting.
+//  Previously, this would leak one message per reader/writer collision
+//  on sockets with ZMQ_CONFLATE set.
+void test_write_while_reader_busy_does_not_leak ()
+{
+    int freed_first = 0, freed_second = 0, freed_third = 0;
+
+    {
+        zmq::dbuffer_t<zmq::msg_t> buf;
+
+        probe_entered.set (0);
+        probe_release.set (0);
+        zmq::thread_t reader;
+        reader.start (probe_thread_fn, &buf, "dbuffer_probe");
+        while (probe_entered.get () == 0)
+            msleep (1);
+
+        //  The reader now sits inside probe () holding the lock.
+        //  Both writes fail try_lock, and the second overwrites _back.
+        write_counted_msg (buf, &freed_first);
+        write_counted_msg (buf, &freed_second);
+        TEST_ASSERT_EQUAL_INT (1, freed_first);
+        TEST_ASSERT_EQUAL_INT (0, freed_second);
+
+        probe_release.add (1);
+        reader.stop ();
+
+        //  With the reader gone this write swaps successfully.
+        //  It must still release the message stored by the failed swap.
+        write_counted_msg (buf, &freed_third);
+        TEST_ASSERT_EQUAL_INT (1, freed_second);
+
+        zmq::msg_t msg;
+        TEST_ASSERT_EQUAL_INT (0, msg.init ());
+        TEST_ASSERT_TRUE (buf.read (&msg));
+        TEST_ASSERT_EQUAL_INT (0, msg.close ());
+        TEST_ASSERT_EQUAL_INT (1, freed_third);
+    }
+
+    //  Destruction must not double-release anything.
+    TEST_ASSERT_EQUAL_INT (1, freed_first);
+    TEST_ASSERT_EQUAL_INT (1, freed_second);
+    TEST_ASSERT_EQUAL_INT (1, freed_third);
+}
+
+int main (void)
+{
+    setup_test_environment ();
+
+    UNITY_BEGIN ();
+    RUN_TEST (test_create);
+    RUN_TEST (test_read_empty);
+    RUN_TEST (test_conflation_releases_overwritten_message);
+    RUN_TEST (test_write_while_reader_busy_does_not_leak);
+
+    return UNITY_END ();
+}