3party/libzmq
0
0
Fork 0
mirror of https://github.com/zeromq/libzmq.git synced 2024-12-27 23:51:04 +08:00
Code Issues Projects Releases Wiki Activity

Merge pull request #3225 from sigiesec/fix-bug-router-duplicate-id-no-handover

Browse Source 
Problem: inconsistent _out_pipes when using the same routing id witho…
This commit is contained in:
Luca Boccassi 2018-08-17 15:27:20 +01:00 committed by GitHub
commit 1c5e81788e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 75 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
src/router.cpp
View File

@ -198,7 +198,7 @@ int zmq::router_t::xsend (msg_t *msg_)
// Find the pipe associated with the routing id stored in the prefix.
// If there's no such pipe just silently ignore the message, unless
// router_mandatory is set. ;
// router_mandatory is set.
out_pipe_t *out_pipe = lookup_out_pipe (
blob_t (static_cast<unsigned char *> (msg_->data ()),
msg_->size (), zmq::reference_tag_t ()));
@ -487,9 +487,9 @@ bool zmq::router_t::identify_peer (pipe_t *pipe_, bool locally_initiated_)
// Try to remove an existing routing id entry to allow the new
// connection to take the routing id.
out_pipe_t existing_outpipe = try_erase_out_pipe (routing_id);
out_pipe_t *existing_outpipe = lookup_out_pipe (routing_id);
if (existing_outpipe.pipe) {
if (existing_outpipe) {
if (!_handover)
// Ignore peers with duplicate ID
return false;
@ -502,15 +502,16 @@ bool zmq::router_t::identify_peer (pipe_t *pipe_, bool locally_initiated_)
put_uint32 (buf + 1, _next_integral_routing_id++);
blob_t new_routing_id (buf, sizeof buf);
existing_outpipe.pipe->set_router_socket_routing_id (
new_routing_id);
pipe_t *const old_pipe = existing_outpipe->pipe;
add_out_pipe (ZMQ_MOVE (new_routing_id), existing_outpipe.pipe);
erase_out_pipe (old_pipe);
old_pipe->set_router_socket_routing_id (new_routing_id);
add_out_pipe (ZMQ_MOVE (new_routing_id), old_pipe);
if (existing_outpipe.pipe == _current_in)
if (old_pipe == _current_in)
_terminate_current_in = true;
else
existing_outpipe.pipe->terminate (true);
old_pipe->terminate (true);
}
}
}

73
tests/test_router_handover.cpp
View File

@ -43,7 +43,7 @@ void tearDown ()
teardown_test_context ();
}
void test_x ()
void test_with_handover ()
{
size_t len = MAX_SOCKET_STRING;
char my_endpoint[MAX_SOCKET_STRING];
@ -85,15 +85,16 @@ void test_x ()
recv_string_expect_success (router, "Hello", 0);
// Send a message to 'X' routing id. This should be delivered
// to the second dealer, instead of the first beccause of the handover.
// to the second dealer, instead of the first because of the handover.
send_string_expect_success (router, "X", ZMQ_SNDMORE);
send_string_expect_success (router, "Hello", 0);
// Ensure that the first dealer doesn't receive the message
// but the second one does
int rc = zmq_recv (dealer_one, buffer, 255, ZMQ_NOBLOCK);
TEST_ASSERT_EQUAL_INT (-1, rc);
TEST_ASSERT_EQUAL_INT (EAGAIN, errno);
const int timeout = SETTLE_TIME;
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (dealer_one, ZMQ_RCVTIMEO, &timeout, sizeof timeout));
TEST_ASSERT_FAILURE_ERRNO (EAGAIN, zmq_recv (dealer_one, buffer, 255, 0));
recv_string_expect_success (dealer_two, "Hello", 0);
@ -102,11 +103,69 @@ void test_x ()
test_context_socket_close (dealer_two);
}
int main (void)
void test_without_handover ()
{
size_t len = MAX_SOCKET_STRING;
char my_endpoint[MAX_SOCKET_STRING];
void *router = test_context_socket (ZMQ_ROUTER);
TEST_ASSERT_SUCCESS_ERRNO (zmq_bind (router, "tcp://127.0.0.1:*"));
TEST_ASSERT_SUCCESS_ERRNO (
zmq_getsockopt (router, ZMQ_LAST_ENDPOINT, my_endpoint, &len));
// Create dealer called "X" and connect it to our router
void *dealer_one = test_context_socket (ZMQ_DEALER);
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (dealer_one, ZMQ_ROUTING_ID, "X", 1));
TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (dealer_one, my_endpoint));
// Get message from dealer to know when connection is ready
char buffer[255];
send_string_expect_success (dealer_one, "Hello", 0);
recv_string_expect_success (router, "X", 0);
recv_string_expect_success (router, "Hello", 0);
// Now create a second dealer that uses the same routing id
void *dealer_two = test_context_socket (ZMQ_DEALER);
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (dealer_two, ZMQ_ROUTING_ID, "X", 1));
TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (dealer_two, my_endpoint));
// Send message from second dealer
send_string_expect_success (dealer_two, "Hello", 0);
// This should be ignored by the router
const int timeout = SETTLE_TIME;
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (router, ZMQ_RCVTIMEO, &timeout, sizeof timeout));
TEST_ASSERT_FAILURE_ERRNO (EAGAIN, zmq_recv (router, buffer, 255, 0));
// Send a message to 'X' routing id. This should be delivered
// to the second dealer, instead of the first because of the handover.
send_string_expect_success (router, "X", ZMQ_SNDMORE);
send_string_expect_success (router, "Hello", 0);
// Ensure that the second dealer doesn't receive the message
// but the first one does
TEST_ASSERT_SUCCESS_ERRNO (
zmq_setsockopt (dealer_two, ZMQ_RCVTIMEO, &timeout, sizeof timeout));
TEST_ASSERT_FAILURE_ERRNO (EAGAIN, zmq_recv (dealer_two, buffer, 255, 0));
recv_string_expect_success (dealer_one, "Hello", 0);
test_context_socket_close (router);
test_context_socket_close (dealer_one);
test_context_socket_close (dealer_two);
}
int main ()
{
setup_test_environment ();
UNITY_BEGIN ();
RUN_TEST (test_x);
RUN_TEST (test_with_handover);
RUN_TEST (test_without_handover);
return UNITY_END ();
}