1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
|
/* SPDX-License-Identifier: MPL-2.0 */
#include "testutil.hpp"
#include "testutil_unity.hpp"
SETUP_TEARDOWN_TESTCONTEXT
const int MAX_SENDS = 10000;
void test_change_before_connected ()
{
int rc;
void *bind_socket = test_context_socket (ZMQ_PUSH);
void *connect_socket = test_context_socket (ZMQ_PULL);
int val = 2;
rc = zmq_setsockopt (connect_socket, ZMQ_RCVHWM, &val, sizeof (val));
TEST_ASSERT_EQUAL_INT (0, rc);
rc = zmq_setsockopt (bind_socket, ZMQ_SNDHWM, &val, sizeof (val));
TEST_ASSERT_EQUAL_INT (0, rc);
zmq_connect (connect_socket, "inproc://a");
zmq_bind (bind_socket, "inproc://a");
size_t placeholder = sizeof (val);
val = 0;
rc = zmq_getsockopt (bind_socket, ZMQ_SNDHWM, &val, &placeholder);
TEST_ASSERT_EQUAL_INT (0, rc);
TEST_ASSERT_EQUAL_INT (2, val);
int send_count = 0;
while (send_count < MAX_SENDS
&& zmq_send (bind_socket, NULL, 0, ZMQ_DONTWAIT) == 0)
++send_count;
TEST_ASSERT_EQUAL_INT (4, send_count);
test_context_socket_close (bind_socket);
test_context_socket_close (connect_socket);
}
void test_change_after_connected ()
{
int rc;
void *bind_socket = test_context_socket (ZMQ_PUSH);
void *connect_socket = test_context_socket (ZMQ_PULL);
int val = 1;
rc = zmq_setsockopt (connect_socket, ZMQ_RCVHWM, &val, sizeof (val));
TEST_ASSERT_EQUAL_INT (0, rc);
rc = zmq_setsockopt (bind_socket, ZMQ_SNDHWM, &val, sizeof (val));
TEST_ASSERT_EQUAL_INT (0, rc);
zmq_connect (connect_socket, "inproc://a");
zmq_bind (bind_socket, "inproc://a");
val = 5;
rc = zmq_setsockopt (bind_socket, ZMQ_SNDHWM, &val, sizeof (val));
TEST_ASSERT_EQUAL_INT (0, rc);
size_t placeholder = sizeof (val);
val = 0;
rc = zmq_getsockopt (bind_socket, ZMQ_SNDHWM, &val, &placeholder);
TEST_ASSERT_EQUAL_INT (0, rc);
TEST_ASSERT_EQUAL_INT (5, val);
int send_count = 0;
while (send_count < MAX_SENDS
&& zmq_send (bind_socket, NULL, 0, ZMQ_DONTWAIT) == 0)
++send_count;
TEST_ASSERT_EQUAL_INT (6, send_count);
test_context_socket_close (bind_socket);
test_context_socket_close (connect_socket);
}
int send_until_wouldblock (void *socket_)
{
int send_count = 0;
while (send_count < MAX_SENDS
&& zmq_send (socket_, &send_count, sizeof (send_count), ZMQ_DONTWAIT)
== sizeof (send_count)) {
++send_count;
}
return send_count;
}
int test_fill_up_to_hwm (void *socket_, int sndhwm_)
{
int send_count = send_until_wouldblock (socket_);
fprintf (stderr, "sndhwm==%i, send_count==%i\n", sndhwm_, send_count);
TEST_ASSERT_LESS_OR_EQUAL_INT (sndhwm_ + 1, send_count);
TEST_ASSERT_GREATER_THAN_INT (sndhwm_ / 10, send_count);
return send_count;
}
void test_decrease_when_full ()
{
int rc;
void *bind_socket = test_context_socket (ZMQ_PUSH);
void *connect_socket = test_context_socket (ZMQ_PULL);
int val = 1;
rc = zmq_setsockopt (connect_socket, ZMQ_RCVHWM, &val, sizeof (val));
TEST_ASSERT_EQUAL_INT (0, rc);
int sndhwm = 100;
rc = zmq_setsockopt (bind_socket, ZMQ_SNDHWM, &sndhwm, sizeof (sndhwm));
TEST_ASSERT_EQUAL_INT (0, rc);
zmq_bind (bind_socket, "inproc://a");
zmq_connect (connect_socket, "inproc://a");
// we must wait for the connect to succeed here, unfortunately we don't
// have monitoring events for inproc, so we just hope SETTLE_TIME suffices
msleep (SETTLE_TIME);
// Fill up to hwm
int send_count = test_fill_up_to_hwm (bind_socket, sndhwm);
// Decrease snd hwm
sndhwm = 70;
rc = zmq_setsockopt (bind_socket, ZMQ_SNDHWM, &sndhwm, sizeof (sndhwm));
TEST_ASSERT_EQUAL_INT (0, rc);
int sndhwm_read = 0;
size_t sndhwm_read_size = sizeof (sndhwm_read);
rc =
zmq_getsockopt (bind_socket, ZMQ_SNDHWM, &sndhwm_read, &sndhwm_read_size);
TEST_ASSERT_EQUAL_INT (0, rc);
TEST_ASSERT_EQUAL_INT (sndhwm, sndhwm_read);
msleep (SETTLE_TIME);
// Read out all data (should get up to previous hwm worth so none were dropped)
int read_count = 0;
int read_data = 0;
while (
read_count < MAX_SENDS
&& zmq_recv (connect_socket, &read_data, sizeof (read_data), ZMQ_DONTWAIT)
== sizeof (read_data)) {
TEST_ASSERT_EQUAL_INT (read_data, read_count);
++read_count;
}
TEST_ASSERT_EQUAL_INT (send_count, read_count);
// Give io thread some time to catch up
msleep (SETTLE_TIME);
// Fill up to new hwm
test_fill_up_to_hwm (bind_socket, sndhwm);
test_context_socket_close (bind_socket);
test_context_socket_close (connect_socket);
}
int main ()
{
setup_test_environment ();
UNITY_BEGIN ();
RUN_TEST (test_change_before_connected);
RUN_TEST (test_change_after_connected);
RUN_TEST (test_decrease_when_full);
return UNITY_END ();
}
|
