Skip to content

Commit 86e0770

Browse files
rschlaikjerrschlaikjer
committed
Add unit test on contended dbuffer not leaking
1 parent 2bb3b96 commit 86e0770

3 files changed

Lines changed: 173 additions & 0 deletions

File tree

‎Makefile.am‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1374,6 +1374,7 @@ if ENABLE_STATIC
13741374
test_apps += \
13751375
unittests/unittest_poller \
13761376
unittests/unittest_ypipe \
1377+
unittests/unittest_dbuffer \
13771378
unittests/unittest_mtrie \
13781379
unittests/unittest_ip_resolver \
13791380
unittests/unittest_udp_address \
@@ -1398,6 +1399,15 @@ unittests_unittest_ypipe_LDADD = \
13981399
${src_libzmq_la_LIBADD} \
13991400
$(CODE_COVERAGE_LDFLAGS)
14001401

1402+
unittests_unittest_dbuffer_SOURCES = unittests/unittest_dbuffer.cpp
1403+
unittests_unittest_dbuffer_CPPFLAGS = -I$(top_srcdir)/src ${TESTUTIL_CPPFLAGS} $(CODE_COVERAGE_CPPFLAGS)
1404+
unittests_unittest_dbuffer_CXXFLAGS = $(CODE_COVERAGE_CXXFLAGS)
1405+
unittests_unittest_dbuffer_LDADD = \
1406+
${TESTUTIL_LIBS} \
1407+
$(top_builddir)/src/.libs/libzmq.a \
1408+
${src_libzmq_la_LIBADD} \
1409+
$(CODE_COVERAGE_LDFLAGS)
1410+
14011411
unittests_unittest_mtrie_SOURCES = unittests/unittest_mtrie.cpp
14021412
unittests_unittest_mtrie_CPPFLAGS = -I$(top_srcdir)/src ${TESTUTIL_CPPFLAGS} $(CODE_COVERAGE_CPPFLAGS)
14031413
unittests_unittest_mtrie_CXXFLAGS = $(CODE_COVERAGE_CXXFLAGS)

‎unittests/CMakeLists.txt‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ cmake_minimum_required(VERSION 2.8.1...3.31)
33

44
set(unittests
55
unittest_ypipe
6+
unittest_dbuffer
67
unittest_poller
78
unittest_mtrie
89
unittest_ip_resolver

‎unittests/unittest_dbuffer.cpp‎

Lines changed: 162 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,162 @@
1+
/* SPDX-License-Identifier: MPL-2.0 */
2+
3+
#include "../tests/testutil.hpp"
4+
5+
// On Windows testutil.hpp maps close () to closesocket ().
6+
// Undo this so it doesn't rewrite zmq::msg_t::close in other headers.
7+
#ifdef close
8+
#undef close
9+
#endif
10+
11+
#include <atomic_counter.hpp>
12+
#include <dbuffer.hpp>
13+
#include <msg.hpp>
14+
#include <thread.hpp>
15+
16+
#include <unity.h>
17+
18+
void setUp ()
19+
{
20+
}
21+
void tearDown ()
22+
{
23+
}
24+
25+
// All test messages share one payload buffer.
26+
// Ownership is tracked per message using the hint field, which points at
27+
// a counter for how many times the content was released.
28+
static char shared_payload[129];
29+
30+
static void count_free (void *data_, void *hint_)
31+
{
32+
LIBZMQ_UNUSED (data_);
33+
++*static_cast<int *> (hint_);
34+
}
35+
36+
// Writes a message owning 'shared_payload' into the dbuffer and gives up
37+
// ownership, the way pipe writers do. 'freed_' is incremented when the
38+
// dbuffer releases the message content.
39+
static void write_counted_msg (zmq::dbuffer_t<zmq::msg_t> &buf_, int *freed_)
40+
{
41+
zmq::msg_t msg;
42+
TEST_ASSERT_EQUAL_INT (0,
43+
msg.init_data (shared_payload, sizeof shared_payload,
44+
count_free, freed_));
45+
buf_.write (msg);
46+
TEST_ASSERT_EQUAL_INT (0, msg.init ());
47+
}
48+
49+
void test_create ()
50+
{
51+
zmq::dbuffer_t<zmq::msg_t> buf;
52+
TEST_ASSERT_FALSE (buf.check_read ());
53+
}
54+
55+
void test_read_empty ()
56+
{
57+
zmq::dbuffer_t<zmq::msg_t> buf;
58+
zmq::msg_t msg;
59+
TEST_ASSERT_EQUAL_INT (0, msg.init ());
60+
TEST_ASSERT_FALSE (buf.read (&msg));
61+
TEST_ASSERT_EQUAL_INT (0, msg.close ());
62+
}
63+
64+
// Uncontended conflation.
65+
// Second write should replace the first message and release its content.
66+
// The reader must only observe the second message.
67+
void test_conflation_releases_overwritten_message ()
68+
{
69+
int freed_first = 0, freed_second = 0;
70+
71+
zmq::dbuffer_t<zmq::msg_t> buf;
72+
write_counted_msg (buf, &freed_first);
73+
write_counted_msg (buf, &freed_second);
74+
TEST_ASSERT_EQUAL_INT (1, freed_first);
75+
TEST_ASSERT_EQUAL_INT (0, freed_second);
76+
77+
zmq::msg_t msg;
78+
TEST_ASSERT_EQUAL_INT (0, msg.init ());
79+
TEST_ASSERT_TRUE (buf.read (&msg));
80+
TEST_ASSERT_EQUAL_INT (sizeof shared_payload, msg.size ());
81+
TEST_ASSERT_EQUAL_INT (0, msg.close ());
82+
TEST_ASSERT_EQUAL_INT (1, freed_second);
83+
84+
TEST_ASSERT_FALSE (buf.read (&msg));
85+
}
86+
87+
// Helpers for parking a reader inside probe (), holding the dbuffer lock.
88+
// This allows deterministically causing writes to fail try_lock.
89+
static zmq::atomic_counter_t probe_entered;
90+
static zmq::atomic_counter_t probe_release;
91+
92+
static bool blocking_probe_fn (const zmq::msg_t &)
93+
{
94+
probe_entered.add (1);
95+
while (probe_release.get () == 0)
96+
msleep (1);
97+
return true;
98+
}
99+
100+
static void probe_thread_fn (void *arg_)
101+
{
102+
static_cast<zmq::dbuffer_t<zmq::msg_t> *> (arg_)->probe (blocking_probe_fn);
103+
}
104+
105+
// A write to a dbuffer locked by a reader stores the msg in the _back slot.
106+
// Subsequent writes must release that message instead of overwriting.
107+
// Previously, this would leak one message per reader/writer collision
108+
// on sockets with ZMQ_CONFLATE set.
109+
void test_write_while_reader_busy_does_not_leak ()
110+
{
111+
int freed_first = 0, freed_second = 0, freed_third = 0;
112+
113+
{
114+
zmq::dbuffer_t<zmq::msg_t> buf;
115+
116+
probe_entered.set (0);
117+
probe_release.set (0);
118+
zmq::thread_t reader;
119+
reader.start (probe_thread_fn, &buf, "dbuffer_probe");
120+
while (probe_entered.get () == 0)
121+
msleep (1);
122+
123+
// The reader now sits inside probe () holding the lock.
124+
// Both writes fail try_lock, and the second overwrites _back.
125+
write_counted_msg (buf, &freed_first);
126+
write_counted_msg (buf, &freed_second);
127+
TEST_ASSERT_EQUAL_INT (1, freed_first);
128+
TEST_ASSERT_EQUAL_INT (0, freed_second);
129+
130+
probe_release.add (1);
131+
reader.stop ();
132+
133+
// With the reader gone this write swaps successfully.
134+
// It must still release the message stored by the failed swap.
135+
write_counted_msg (buf, &freed_third);
136+
TEST_ASSERT_EQUAL_INT (1, freed_second);
137+
138+
zmq::msg_t msg;
139+
TEST_ASSERT_EQUAL_INT (0, msg.init ());
140+
TEST_ASSERT_TRUE (buf.read (&msg));
141+
TEST_ASSERT_EQUAL_INT (0, msg.close ());
142+
TEST_ASSERT_EQUAL_INT (1, freed_third);
143+
}
144+
145+
// Destruction must not double-release anything.
146+
TEST_ASSERT_EQUAL_INT (1, freed_first);
147+
TEST_ASSERT_EQUAL_INT (1, freed_second);
148+
TEST_ASSERT_EQUAL_INT (1, freed_third);
149+
}
150+
151+
int main (void)
152+
{
153+
setup_test_environment ();
154+
155+
UNITY_BEGIN ();
156+
RUN_TEST (test_create);
157+
RUN_TEST (test_read_empty);
158+
RUN_TEST (test_conflation_releases_overwritten_message);
159+
RUN_TEST (test_write_while_reader_busy_does_not_leak);
160+
161+
return UNITY_END ();
162+
}

0 commit comments

Comments
 (0)