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 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432
|
/* SPDX-License-Identifier: MPL-2.0 */
#include "testutil.hpp"
#if defined(ZMQ_HAVE_WINDOWS)
#include <winsock2.h>
#include <ws2tcpip.h>
#include <stdexcept>
#define close closesocket
typedef SOCKET raw_socket;
#else
#include <arpa/inet.h>
#include <unistd.h>
typedef int raw_socket;
#endif
#include <limits.h>
#include <stdlib.h>
#include <string.h>
// TODO remove this here, either ensure that UINT16_MAX is always properly
// defined or handle this at a more central location
#ifndef UINT16_MAX
#define UINT16_MAX 65535
#endif
#include "testutil_unity.hpp"
SETUP_TEARDOWN_TESTCONTEXT
// Read one event off the monitor socket; return value and address
// by reference, if not null, and event number by value. Returns -1
// in case of error.
static int get_monitor_event (void *monitor_)
{
for (int i = 0; i < 10; i++) {
// First frame in message contains event number and value
zmq_msg_t msg;
TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_init (&msg));
if (zmq_msg_recv (&msg, monitor_, ZMQ_DONTWAIT) == -1) {
msleep (SETTLE_TIME);
continue; // Interrupted, presumably
}
TEST_ASSERT_TRUE (zmq_msg_more (&msg));
uint8_t *data = static_cast<uint8_t *> (zmq_msg_data (&msg));
uint16_t event = *reinterpret_cast<uint16_t *> (data);
// Second frame in message contains event address
TEST_ASSERT_SUCCESS_ERRNO (zmq_msg_init (&msg));
if (zmq_msg_recv (&msg, monitor_, 0) == -1) {
return -1; // Interrupted, presumably
}
TEST_ASSERT_FALSE (zmq_msg_more (&msg));
return event;
}
return -1;
}
static void recv_with_retry (raw_socket fd_, char *buffer_, int bytes_)
{
int received = 0;
while (true) {
int rc = TEST_ASSERT_SUCCESS_RAW_ERRNO (
recv (fd_, buffer_ + received, bytes_ - received, 0));
TEST_ASSERT_GREATER_THAN_INT (0, rc);
received += rc;
TEST_ASSERT_LESS_OR_EQUAL_INT (bytes_, received);
if (received == bytes_)
break;
}
}
static void mock_handshake (raw_socket fd_, int mock_ping_)
{
char buffer[128];
memset (buffer, 0, sizeof (buffer));
memcpy (buffer, zmtp_greeting_null, sizeof (zmtp_greeting_null));
int rc = TEST_ASSERT_SUCCESS_RAW_ERRNO (
send (fd_, buffer, sizeof (zmtp_greeting_null), 0));
TEST_ASSERT_EQUAL_INT (sizeof (zmtp_greeting_null), rc);
recv_with_retry (fd_, buffer, sizeof (zmtp_greeting_null));
memset (buffer, 0, sizeof (buffer));
memcpy (buffer, zmtp_ready_dealer, sizeof (zmtp_ready_dealer));
rc = TEST_ASSERT_SUCCESS_RAW_ERRNO (
send (fd_, buffer, sizeof (zmtp_ready_dealer), 0));
TEST_ASSERT_EQUAL_INT (sizeof (zmtp_ready_dealer), rc);
// greeting
recv_with_retry (fd_, buffer, sizeof (zmtp_ready_dealer));
if (mock_ping_) {
// test PING context - should be replicated in the PONG
// to avoid timeouts, do a bulk send
const uint8_t zmtp_ping[12] = {4, 10, 4, 'P', 'I', 'N',
'G', 0, 0, 'L', 'O', 'L'};
uint8_t zmtp_pong[10] = {4, 8, 4, 'P', 'O', 'N', 'G', 'L', 'O', 'L'};
memset (buffer, 0, sizeof (buffer));
memcpy (buffer, zmtp_ping, 12);
rc = TEST_ASSERT_SUCCESS_RAW_ERRNO (send (fd_, buffer, 12, 0));
TEST_ASSERT_EQUAL_INT (12, rc);
// test a larger body that won't fit in a small message and should get
// truncated
memset (buffer, 'z', sizeof (buffer));
memcpy (buffer, zmtp_ping, 12);
buffer[1] = 65;
rc = TEST_ASSERT_SUCCESS_RAW_ERRNO (send (fd_, buffer, 67, 0));
TEST_ASSERT_EQUAL_INT (67, rc);
// small pong
recv_with_retry (fd_, buffer, 10);
TEST_ASSERT_EQUAL_INT (0, memcmp (zmtp_pong, buffer, 10));
// large pong
recv_with_retry (fd_, buffer, 23);
uint8_t zmtp_pooong[65] = {4, 21, 4, 'P', 'O', 'N', 'G', 'L', 'O', 'L'};
memset (zmtp_pooong + 10, 'z', 55);
TEST_ASSERT_EQUAL_INT (0, memcmp (zmtp_pooong, buffer, 23));
}
}
static void setup_curve (void *socket_, int is_server_)
{
const char *secret_key;
const char *public_key;
const char *server_key;
if (is_server_) {
secret_key = "JTKVSB%%)wK0E.X)V>+}o?pNmC{O&4W4b!Ni{Lh6";
public_key = "rq:rM>}U?@Lns47E1%kR.o@n%FcmmsL/@{H8]yf7";
server_key = NULL;
} else {
secret_key = "D:)Q[IlAW!ahhC2ac:9*A}h:p?([4%wOTJ%JR%cs";
public_key = "Yne@$w-vo<fVvi]a<NY6T1ed:M$fCG*[IaLV{hID";
server_key = "rq:rM>}U?@Lns47E1%kR.o@n%FcmmsL/@{H8]yf7";
}
zmq_setsockopt (socket_, ZMQ_CURVE_SECRETKEY, secret_key,
strlen (secret_key));
zmq_setsockopt (socket_, ZMQ_CURVE_PUBLICKEY, public_key,
strlen (public_key));
if (is_server_)
zmq_setsockopt (socket_, ZMQ_CURVE_SERVER, &is_server_,
sizeof (is_server_));
else
zmq_setsockopt (socket_, ZMQ_CURVE_SERVERKEY, server_key,
strlen (server_key));
}
static void prep_server_socket (int set_heartbeats_,
int is_curve_,
void **server_out_,
void **mon_out_,
char *endpoint_,
size_t ep_length_,
int socket_type_)
{
// We'll be using this socket in raw mode
void *server = test_context_socket (socket_type_);
int value = 0;
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (server, ZMQ_LINGER, &value, sizeof (value)));
if (set_heartbeats_) {
value = 50;
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (server, ZMQ_HEARTBEAT_IVL, &value, sizeof (value)));
}
if (is_curve_)
setup_curve (server, 1);
bind_loopback_ipv4 (server, endpoint_, ep_length_);
// Create and connect a socket for collecting monitor events on dealer
void *server_mon = test_context_socket (ZMQ_PAIR);
TEST_ASSERT_SUCCESS_ERRNO (zmq_socket_monitor (
server, "inproc://monitor-dealer",
ZMQ_EVENT_CONNECTED | ZMQ_EVENT_DISCONNECTED | ZMQ_EVENT_ACCEPTED));
// Connect to the inproc endpoint so we'll get events
TEST_ASSERT_SUCCESS_ERRNO (
zmq_connect (server_mon, "inproc://monitor-dealer"));
*server_out_ = server;
*mon_out_ = server_mon;
}
// This checks for a broken TCP connection (or, in this case a stuck one
// where the peer never responds to PINGS). There should be an accepted event
// then a disconnect event.
static void test_heartbeat_timeout (int server_type_, int mock_ping_)
{
int rc;
char my_endpoint[MAX_SOCKET_STRING];
void *server, *server_mon;
prep_server_socket (!mock_ping_, 0, &server, &server_mon, my_endpoint,
MAX_SOCKET_STRING, server_type_);
fd_t s = connect_socket (my_endpoint);
// Mock a ZMTP 3 client so we can forcibly time out a connection
mock_handshake (s, mock_ping_);
// By now everything should report as connected
rc = get_monitor_event (server_mon);
TEST_ASSERT_EQUAL_INT (ZMQ_EVENT_ACCEPTED, rc);
if (!mock_ping_) {
// We should have been disconnected
rc = get_monitor_event (server_mon);
TEST_ASSERT_EQUAL_INT (ZMQ_EVENT_DISCONNECTED, rc);
}
close (s);
test_context_socket_close (server);
test_context_socket_close (server_mon);
}
// This checks that peers respect the TTL value in ping messages
// We set up a mock ZMTP 3 client and send a ping message with a TLL
// to a server that is not doing any heartbeating. Then we sleep,
// if the server disconnects the client, then we know the TTL did
// its thing correctly.
static void test_heartbeat_ttl (int client_type_, int server_type_)
{
int rc, value;
char my_endpoint[MAX_SOCKET_STRING];
void *server, *server_mon, *client;
prep_server_socket (0, 0, &server, &server_mon, my_endpoint,
MAX_SOCKET_STRING, server_type_);
client = test_context_socket (client_type_);
// Set the heartbeat TTL to 0.1 seconds
value = 100;
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (client, ZMQ_HEARTBEAT_TTL, &value, sizeof (value)));
// Set the heartbeat interval to much longer than the TTL so that
// the socket times out oon the remote side.
value = 250;
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (client, ZMQ_HEARTBEAT_IVL, &value, sizeof (value)));
TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (client, my_endpoint));
// By now everything should report as connected
rc = get_monitor_event (server_mon);
TEST_ASSERT_EQUAL_INT (ZMQ_EVENT_ACCEPTED, rc);
msleep (SETTLE_TIME);
// We should have been disconnected
rc = get_monitor_event (server_mon);
TEST_ASSERT_EQUAL_INT (ZMQ_EVENT_DISCONNECTED, rc);
test_context_socket_close (server);
test_context_socket_close (server_mon);
test_context_socket_close (client);
}
// This checks for normal operation - that is pings and pongs being
// exchanged normally. There should be an accepted event on the server,
// and then no event afterwards.
static void
test_heartbeat_notimeout (int is_curve_, int client_type_, int server_type_)
{
int rc;
char my_endpoint[MAX_SOCKET_STRING];
void *server, *server_mon;
prep_server_socket (1, is_curve_, &server, &server_mon, my_endpoint,
MAX_SOCKET_STRING, server_type_);
void *client = test_context_socket (client_type_);
if (is_curve_)
setup_curve (client, 0);
rc = zmq_connect (client, my_endpoint);
// Give it a sec to connect and handshake
msleep (SETTLE_TIME);
// By now everything should report as connected
rc = get_monitor_event (server_mon);
TEST_ASSERT_EQUAL_INT (ZMQ_EVENT_ACCEPTED, rc);
// We should still be connected because pings and pongs are happenin'
rc = get_monitor_event (server_mon);
// TODO: this fails ~1% of the runs on OBS but it does not seem to be reproducible anywhere else
if (rc == 512)
TEST_IGNORE_MESSAGE (
"Unreliable test occasionally fails on slow CIs, ignoring");
TEST_ASSERT_EQUAL_INT (-1, rc);
test_context_socket_close (client);
test_context_socket_close (server);
test_context_socket_close (server_mon);
}
void test_heartbeat_timeout_router ()
{
test_heartbeat_timeout (ZMQ_ROUTER, 0);
}
void test_heartbeat_timeout_router_mock_ping ()
{
test_heartbeat_timeout (ZMQ_ROUTER, 1);
}
#define DEFINE_TESTS(first, second, first_define, second_define) \
void test_heartbeat_ttl_##first##_##second () \
{ \
test_heartbeat_ttl (first_define, second_define); \
} \
void test_heartbeat_notimeout_##first##_##second () \
{ \
test_heartbeat_notimeout (0, first_define, second_define); \
} \
void test_heartbeat_notimeout_##first##_##second##_with_curve () \
{ \
test_heartbeat_notimeout (1, first_define, second_define); \
}
DEFINE_TESTS (dealer, router, ZMQ_DEALER, ZMQ_ROUTER)
DEFINE_TESTS (req, rep, ZMQ_REQ, ZMQ_REP)
DEFINE_TESTS (pull, push, ZMQ_PULL, ZMQ_PUSH)
DEFINE_TESTS (sub, pub, ZMQ_SUB, ZMQ_PUB)
DEFINE_TESTS (pair, pair, ZMQ_PAIR, ZMQ_PAIR)
#ifdef ZMQ_BUILD_DRAFT_API
DEFINE_TESTS (gather, scatter, ZMQ_GATHER, ZMQ_SCATTER)
DEFINE_TESTS (client, server, ZMQ_CLIENT, ZMQ_SERVER)
#endif
const int deciseconds_per_millisecond = 100;
const int heartbeat_ttl_max =
(UINT16_MAX + 1) * deciseconds_per_millisecond - 1;
void test_setsockopt_heartbeat_success (const int value_)
{
void *const socket = test_context_socket (ZMQ_PAIR);
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (socket, ZMQ_HEARTBEAT_TTL, &value_, sizeof (value_)));
int value_read;
size_t value_read_size = sizeof (value_read);
TEST_ASSERT_SUCCESS_ERRNO (zmq_getsockopt (socket, ZMQ_HEARTBEAT_TTL,
&value_read, &value_read_size));
TEST_ASSERT_EQUAL_INT (value_ - value_ % deciseconds_per_millisecond,
value_read);
test_context_socket_close (socket);
}
void test_setsockopt_heartbeat_ttl_max ()
{
test_setsockopt_heartbeat_success (heartbeat_ttl_max);
}
void test_setsockopt_heartbeat_ttl_more_than_max_fails ()
{
void *const socket = test_context_socket (ZMQ_PAIR);
const int value = heartbeat_ttl_max + 1;
TEST_ASSERT_FAILURE_ERRNO (
EINVAL,
zmq_setsockopt (socket, ZMQ_HEARTBEAT_TTL, &value, sizeof (value)));
test_context_socket_close (socket);
}
void test_setsockopt_heartbeat_ttl_near_zero ()
{
test_setsockopt_heartbeat_success (deciseconds_per_millisecond - 1);
}
int main (void)
{
// The test cases are very long-running. The default timeout of 60 seconds
// is not always enough.
setup_test_environment (90);
UNITY_BEGIN ();
RUN_TEST (test_heartbeat_timeout_router);
RUN_TEST (test_heartbeat_timeout_router_mock_ping);
RUN_TEST (test_heartbeat_ttl_dealer_router);
RUN_TEST (test_heartbeat_ttl_req_rep);
RUN_TEST (test_heartbeat_ttl_pull_push);
RUN_TEST (test_heartbeat_ttl_sub_pub);
RUN_TEST (test_heartbeat_ttl_pair_pair);
RUN_TEST (test_setsockopt_heartbeat_ttl_max);
RUN_TEST (test_setsockopt_heartbeat_ttl_more_than_max_fails);
RUN_TEST (test_setsockopt_heartbeat_ttl_near_zero);
RUN_TEST (test_heartbeat_notimeout_dealer_router);
RUN_TEST (test_heartbeat_notimeout_req_rep);
RUN_TEST (test_heartbeat_notimeout_pull_push);
RUN_TEST (test_heartbeat_notimeout_sub_pub);
RUN_TEST (test_heartbeat_notimeout_pair_pair);
RUN_TEST (test_heartbeat_notimeout_dealer_router_with_curve);
RUN_TEST (test_heartbeat_notimeout_req_rep_with_curve);
RUN_TEST (test_heartbeat_notimeout_pull_push_with_curve);
RUN_TEST (test_heartbeat_notimeout_sub_pub_with_curve);
RUN_TEST (test_heartbeat_notimeout_pair_pair_with_curve);
#ifdef ZMQ_BUILD_DRAFT_API
RUN_TEST (test_heartbeat_ttl_client_server);
RUN_TEST (test_heartbeat_ttl_gather_scatter);
RUN_TEST (test_heartbeat_notimeout_client_server);
RUN_TEST (test_heartbeat_notimeout_gather_scatter);
RUN_TEST (test_heartbeat_notimeout_client_server_with_curve);
RUN_TEST (test_heartbeat_notimeout_gather_scatter_with_curve);
#endif
return UNITY_END ();
}
|