From 49e1b8b75f0e737a7235fce986d84d5f48aeaf9b Mon Sep 17 00:00:00 2001 From: sigiesec Date: Tue, 29 Aug 2017 10:03:17 +0200 Subject: [PATCH 01/10] Problem: test case in main function Solution: extracted test_basic function --- tests/test_router_mandatory.cpp | 23 +++++++++++++++-------- 1 file changed, 15 insertions(+), 8 deletions(-) diff --git a/tests/test_router_mandatory.cpp b/tests/test_router_mandatory.cpp index 40325625..1388a4f0 100644 --- a/tests/test_router_mandatory.cpp +++ b/tests/test_router_mandatory.cpp @@ -29,9 +29,8 @@ #include "testutil.hpp" -int main (void) +void test_basic () { - setup_test_environment(); size_t len = MAX_SOCKET_STRING; char my_endpoint[MAX_SOCKET_STRING]; void *ctx = zmq_ctx_new (); @@ -55,7 +54,8 @@ int main (void) // Send a message to an unknown peer with mandatory routing // This will fail int mandatory = 1; - rc = zmq_setsockopt (router, ZMQ_ROUTER_MANDATORY, &mandatory, sizeof (mandatory)); + rc = zmq_setsockopt (router, ZMQ_ROUTER_MANDATORY, &mandatory, + sizeof (mandatory)); assert (rc == 0); rc = zmq_send (router, "UNKNOWN", 7, ZMQ_SNDMORE); assert (rc == -1 && errno == EHOSTUNREACH); @@ -69,12 +69,12 @@ int main (void) assert (rc == 0); // Get message from dealer to know when connection is ready - char buffer [255]; + char buffer[255]; rc = zmq_send (dealer, "Hello", 5, 0); assert (rc == 5); rc = zmq_recv (router, buffer, 255, 0); assert (rc == 1); - assert (buffer [0] == 'X'); + assert (buffer[0] == 'X'); // Send a message to connected dealer now // It should work @@ -82,7 +82,7 @@ int main (void) assert (rc == 1); rc = zmq_send (router, "Hello", 5, 0); assert (rc == 5); - + rc = zmq_close (router); assert (rc == 0); @@ -91,6 +91,13 @@ int main (void) rc = zmq_ctx_term (ctx); assert (rc == 0); - - return 0 ; +} + +int main (void) +{ + setup_test_environment (); + + test_basic (); + + return 0; } From efa86fe62972565eb52bb6c8886e4eef95919b50 Mon Sep 17 00:00:00 2001 From: sigiesec Date: Tue, 29 Aug 2017 10:04:06 +0200 Subject: [PATCH 02/10] Problem: no test case using the proposed zmq_socket_get_peer_state function Solution: added test case (with dummy implementation of zmq_socket_get_peer_state) --- include/zmq.h | 4 ++ src/zmq.cpp | 10 +++ tests/test_router_mandatory.cpp | 104 +++++++++++++++++++++++++++++++- 3 files changed, 117 insertions(+), 1 deletion(-) diff --git a/include/zmq.h b/include/zmq.h index 62318fbd..dceccb31 100644 --- a/include/zmq.h +++ b/include/zmq.h @@ -661,6 +661,10 @@ ZMQ_EXPORT int zmq_poller_modify_fd (void *poller, int fd, short events); ZMQ_EXPORT int zmq_poller_remove_fd (void *poller, int fd); #endif +ZMQ_EXPORT int zmq_socket_get_peer_state (void *socket, + const void *identity, + size_t identity_size); + /******************************************************************************/ /* Scheduling timers */ /******************************************************************************/ diff --git a/src/zmq.cpp b/src/zmq.cpp index dca9088c..d4f3efba 100644 --- a/src/zmq.cpp +++ b/src/zmq.cpp @@ -1389,6 +1389,16 @@ int zmq_poller_wait_all (void *poller_, zmq_poller_event_t *events_, int n_event return rc; } +// Peer-specific state + +int zmq_socket_get_peer_state (void *socket, + const void *identity, + size_t identity_size) +{ + errno = ENOTSUP; + return -1; +} + // Timers void *zmq_timers_new (void) diff --git a/tests/test_router_mandatory.cpp b/tests/test_router_mandatory.cpp index 1388a4f0..36cdb42d 100644 --- a/tests/test_router_mandatory.cpp +++ b/tests/test_router_mandatory.cpp @@ -29,6 +29,107 @@ #include "testutil.hpp" +void test_get_peer_state () +{ +#ifdef ZMQ_BUILD_DRAFT_API + size_t len = MAX_SOCKET_STRING; + char my_endpoint[MAX_SOCKET_STRING]; + void *ctx = zmq_ctx_new (); + assert (ctx); + void *router = zmq_socket (ctx, ZMQ_ROUTER); + assert (router); + + int rc = zmq_bind (router, "tcp://127.0.0.1:*"); + assert (rc == 0); + + rc = zmq_getsockopt (router, ZMQ_LAST_ENDPOINT, my_endpoint, &len); + assert (rc == 0); + + int mandatory = 1; + rc = zmq_setsockopt (router, ZMQ_ROUTER_MANDATORY, &mandatory, + sizeof (mandatory)); + + // Create dealer called "X" and connect it to our router + void *dealer1 = zmq_socket (ctx, ZMQ_DEALER); + assert (dealer1); + rc = zmq_setsockopt (dealer1, ZMQ_IDENTITY, "X", 1); + assert (rc == 0); + rc = zmq_connect (dealer1, my_endpoint); + assert (rc == 0); + + // Create dealer called "Y" and connect it to our router + void *dealer2 = zmq_socket (ctx, ZMQ_DEALER); + assert (dealer2); + rc = zmq_setsockopt (dealer2, ZMQ_IDENTITY, "Y", 1); + assert (rc == 0); + rc = zmq_connect (dealer2, my_endpoint); + assert (rc == 0); + + // Get message from dealer to know when connection is ready + char buffer[255]; + rc = zmq_send (dealer1, "Hello", 5, 0); + assert (rc == 5); + rc = zmq_recv (router, buffer, 255, 0); + assert (rc == 1); + assert (buffer[0] == 'X'); + + void *poller = zmq_poller_new (); + assert (poller); + + // Poll on router and dealer1, but not on dealer2 + rc = zmq_poller_add (poller, router, NULL, ZMQ_POLLOUT); + assert (rc == 0); + rc = zmq_poller_add (poller, dealer1, NULL, ZMQ_POLLIN); + assert (rc == 0); + + const size_t count = 10000; + const size_t event_size = 2; + zmq_poller_event_t events[event_size]; + for (size_t iterations = 0; + iterations < count + && zmq_poller_wait_all (poller, events, event_size, -1) != -1; + ++iterations) { + for (size_t i = 0; i < event_size; ++i) { + if (events[i].socket == router) { + rc = zmq_socket_get_peer_state (router, "X", 1); + assert (rc != -1); + if (rc & ZMQ_POLLOUT) { + rc = zmq_send (router, "X", 1, ZMQ_SNDMORE); + assert (rc == 1); + rc = zmq_send (router, "Hello", 5, 0); + assert (rc == 5); + } + rc = zmq_socket_get_peer_state (router, "Y", 1); + assert (rc != -1); + if (rc & ZMQ_POLLOUT) { + rc = zmq_send (router, "Y", 1, ZMQ_SNDMORE); + assert (rc == 1); + rc = zmq_send (router, "Hello", 5, 0); + assert (rc == 5); + } + } + if (events[i].socket == dealer1) { + rc = zmq_recv (dealer1, buffer, 255, ZMQ_DONTWAIT); + assert (rc == 5); + } + // never read from dealer2, so its pipe becomes full eventually + } + } + + rc = zmq_close (router); + assert (rc == 0); + + rc = zmq_close (dealer1); + assert (rc == 0); + + rc = zmq_close (dealer2); + assert (rc == 0); + + rc = zmq_ctx_term (ctx); + assert (rc == 0); +#endif +} + void test_basic () { size_t len = MAX_SOCKET_STRING; @@ -98,6 +199,7 @@ int main (void) setup_test_environment (); test_basic (); + test_get_peer_state (); - return 0; + return 0 ; } From f4d139bd165b6d7219ea36d4ac9de0a60aebf141 Mon Sep 17 00:00:00 2001 From: sigiesec Date: Tue, 29 Aug 2017 10:30:03 +0200 Subject: [PATCH 03/10] Problem: duplicated code in socket-related functions Solution: extract as_socket_base_t function --- src/zmq.cpp | 141 ++++++++++++++++++++-------------------------------- 1 file changed, 54 insertions(+), 87 deletions(-) diff --git a/src/zmq.cpp b/src/zmq.cpp index d4f3efba..23a877c0 100644 --- a/src/zmq.cpp +++ b/src/zmq.cpp @@ -247,6 +247,16 @@ int zmq_ctx_destroy (void *ctx_) // Sockets +static zmq::socket_base_t *as_socket_base_t (void *s_) +{ + zmq::socket_base_t *s = static_cast (s_); + if (!s_ || !s->check_tag ()) { + errno = ENOTSOCK; + return NULL; + } + return s; +} + void *zmq_socket (void *ctx_, int type_) { if (!ctx_ || !((zmq::ctx_t *) ctx_)->check_tag ()) { @@ -260,109 +270,83 @@ void *zmq_socket (void *ctx_, int type_) int zmq_close (void *s_) { - if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) { - errno = ENOTSOCK; + zmq::socket_base_t *s = as_socket_base_t (s_); + if (!s) return -1; - } - ((zmq::socket_base_t*) s_)->close (); + s->close (); return 0; } int zmq_setsockopt (void *s_, int option_, const void *optval_, size_t optvallen_) { - if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) { - errno = ENOTSOCK; + zmq::socket_base_t *s = as_socket_base_t (s_); + if (!s) return -1; - } - zmq::socket_base_t *s = (zmq::socket_base_t *) s_; - int result = s->setsockopt (option_, optval_, optvallen_); - return result; + return s->setsockopt (option_, optval_, optvallen_); } int zmq_getsockopt (void *s_, int option_, void *optval_, size_t *optvallen_) { - if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) { - errno = ENOTSOCK; + zmq::socket_base_t *s = as_socket_base_t (s_); + if (!s) return -1; - } - zmq::socket_base_t *s = (zmq::socket_base_t *) s_; - int result = s->getsockopt (option_, optval_, optvallen_); - return result; + return s->getsockopt (option_, optval_, optvallen_); } int zmq_socket_monitor (void *s_, const char *addr_, int events_) { - if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) { - errno = ENOTSOCK; + zmq::socket_base_t *s = as_socket_base_t (s_); + if (!s) return -1; - } - zmq::socket_base_t *s = (zmq::socket_base_t *) s_; - int result = s->monitor (addr_, events_); - return result; + return s->monitor (addr_, events_); } int zmq_join (void *s_, const char* group_) { - if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) { - errno = ENOTSOCK; + zmq::socket_base_t *s = as_socket_base_t (s_); + if (!s) return -1; - } - zmq::socket_base_t *s = (zmq::socket_base_t *) s_; - int result = s->join (group_); - return result; + return s->join (group_); } int zmq_leave (void *s_, const char* group_) { - if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) { - errno = ENOTSOCK; + zmq::socket_base_t *s = as_socket_base_t (s_); + if (!s) return -1; - } - zmq::socket_base_t *s = (zmq::socket_base_t *) s_; - int result = s->leave (group_); - return result; + return s->leave (group_); } int zmq_bind (void *s_, const char *addr_) { - if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) { - errno = ENOTSOCK; + zmq::socket_base_t *s = as_socket_base_t (s_); + if (!s) return -1; - } - zmq::socket_base_t *s = (zmq::socket_base_t *) s_; - int result = s->bind (addr_); - return result; + return s->bind (addr_); } int zmq_connect (void *s_, const char *addr_) { - if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) { - errno = ENOTSOCK; + zmq::socket_base_t *s = as_socket_base_t (s_); + if (!s) return -1; - } - zmq::socket_base_t *s = (zmq::socket_base_t *) s_; - int result = s->connect (addr_); - return result; + return s->connect (addr_); } int zmq_unbind (void *s_, const char *addr_) { - if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) { - errno = ENOTSOCK; + zmq::socket_base_t *s = as_socket_base_t (s_); + if (!s) return -1; - } - zmq::socket_base_t *s = (zmq::socket_base_t *) s_; return s->term_endpoint (addr_); } int zmq_disconnect (void *s_, const char *addr_) { - if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) { - errno = ENOTSOCK; + zmq::socket_base_t *s = as_socket_base_t (s_); + if (!s) return -1; - } - zmq::socket_base_t *s = (zmq::socket_base_t *) s_; return s->term_endpoint (addr_); } @@ -392,10 +376,9 @@ int zmq_sendmsg (void *s_, zmq_msg_t *msg_, int flags_) int zmq_send (void *s_, const void *buf_, size_t len_, int flags_) { - if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) { - errno = ENOTSOCK; + zmq::socket_base_t *s = as_socket_base_t (s_); + if (!s) return -1; - } zmq_msg_t msg; if (zmq_msg_init_size (&msg, len_)) return -1; @@ -405,7 +388,6 @@ int zmq_send (void *s_, const void *buf_, size_t len_, int flags_) assert (buf_); memcpy (zmq_msg_data (&msg), buf_, len_); } - zmq::socket_base_t *s = (zmq::socket_base_t *) s_; int rc = s_sendmsg (s, &msg, flags_); if (unlikely (rc < 0)) { int err = errno; @@ -421,16 +403,14 @@ int zmq_send (void *s_, const void *buf_, size_t len_, int flags_) int zmq_send_const (void *s_, const void *buf_, size_t len_, int flags_) { - if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) { - errno = ENOTSOCK; + zmq::socket_base_t *s = as_socket_base_t (s_); + if (!s) return -1; - } zmq_msg_t msg; int rc = zmq_msg_init_data (&msg, (void *)buf_, len_, NULL, NULL); if (rc != 0) return -1; - zmq::socket_base_t *s = (zmq::socket_base_t *) s_; rc = s_sendmsg (s, &msg, flags_); if (unlikely (rc < 0)) { int err = errno; @@ -454,10 +434,9 @@ int zmq_send_const (void *s_, const void *buf_, size_t len_, int flags_) // int zmq_sendiov (void *s_, iovec *a_, size_t count_, int flags_) { - if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) { - errno = ENOTSOCK; + zmq::socket_base_t *s = as_socket_base_t (s_); + if (!s) return -1; - } if (unlikely (count_ <= 0 || !a_)) { errno = EINVAL; return -1; @@ -465,7 +444,6 @@ int zmq_sendiov (void *s_, iovec *a_, size_t count_, int flags_) int rc = 0; zmq_msg_t msg; - zmq::socket_base_t *s = (zmq::socket_base_t *) s_; for (size_t i = 0; i < count_; ++i) { rc = zmq_msg_init_size (&msg, a_[i].iov_len); @@ -512,15 +490,13 @@ int zmq_recvmsg (void *s_, zmq_msg_t *msg_, int flags_) int zmq_recv (void *s_, void *buf_, size_t len_, int flags_) { - if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) { - errno = ENOTSOCK; + zmq::socket_base_t *s = as_socket_base_t (s_); + if (!s) return -1; - } zmq_msg_t msg; int rc = zmq_msg_init (&msg); errno_assert (rc == 0); - zmq::socket_base_t *s = (zmq::socket_base_t *) s_; int nbytes = s_recvmsg (s, &msg, flags_); if (unlikely (nbytes < 0)) { int err = errno; @@ -562,17 +538,14 @@ int zmq_recv (void *s_, void *buf_, size_t len_, int flags_) // int zmq_recviov (void *s_, iovec *a_, size_t *count_, int flags_) { - if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) { - errno = ENOTSOCK; + zmq::socket_base_t *s = as_socket_base_t (s_); + if (!s) return -1; - } if (unlikely (!count_ || *count_ <= 0 || !a_)) { errno = EINVAL; return -1; } - zmq::socket_base_t *s = (zmq::socket_base_t *) s_; - size_t count = *count_; int nread = 0; bool recvmore = true; @@ -634,24 +607,18 @@ int zmq_msg_init_data (zmq_msg_t *msg_, void *data_, size_t size_, int zmq_msg_send (zmq_msg_t *msg_, void *s_, int flags_) { - if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) { - errno = ENOTSOCK; + zmq::socket_base_t *s = as_socket_base_t (s_); + if (!s) return -1; - } - zmq::socket_base_t *s = (zmq::socket_base_t *) s_; - int result = s_sendmsg (s, msg_, flags_); - return result; + return s_sendmsg (s, msg_, flags_); } int zmq_msg_recv (zmq_msg_t *msg_, void *s_, int flags_) { - if (!s_ || !((zmq::socket_base_t*) s_)->check_tag ()) { - errno = ENOTSOCK; + zmq::socket_base_t *s = as_socket_base_t (s_); + if (!s) return -1; - } - zmq::socket_base_t *s = (zmq::socket_base_t *) s_; - int result = s_recvmsg (s, msg_, flags_); - return result; + return s_recvmsg (s, msg_, flags_); } int zmq_msg_close (zmq_msg_t *msg_) From cda20260b3242ba71ec1c08fff84e17a6accfaa3 Mon Sep 17 00:00:00 2001 From: sigiesec Date: Tue, 29 Aug 2017 11:50:30 +0200 Subject: [PATCH 04/10] Problem: missing call to zmq_poller_destroy Solution: added call --- tests/test_router_mandatory.cpp | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/tests/test_router_mandatory.cpp b/tests/test_router_mandatory.cpp index 36cdb42d..bc6677b3 100644 --- a/tests/test_router_mandatory.cpp +++ b/tests/test_router_mandatory.cpp @@ -65,13 +65,24 @@ void test_get_peer_state () rc = zmq_connect (dealer2, my_endpoint); assert (rc == 0); - // Get message from dealer to know when connection is ready + // Get message from both dealers to know when connection is ready char buffer[255]; rc = zmq_send (dealer1, "Hello", 5, 0); assert (rc == 5); rc = zmq_recv (router, buffer, 255, 0); assert (rc == 1); assert (buffer[0] == 'X'); + rc = zmq_recv (router, buffer, 255, 0); + assert (rc == 5); + + rc = zmq_send (dealer2, "Hello", 5, 0); + assert (rc == 5); + rc = zmq_recv (router, buffer, 255, 0); + assert (rc == 1); + assert (buffer[0] == 'Y'); + rc = zmq_recv (router, buffer, 255, 0); + assert (rc == 5); + void *poller = zmq_poller_new (); assert (poller); @@ -92,6 +103,8 @@ void test_get_peer_state () for (size_t i = 0; i < event_size; ++i) { if (events[i].socket == router) { rc = zmq_socket_get_peer_state (router, "X", 1); + if (rc == -1) + printf ("zmq_socket_get_peer_state failed: %i\n", errno); assert (rc != -1); if (rc & ZMQ_POLLOUT) { rc = zmq_send (router, "X", 1, ZMQ_SNDMORE); @@ -100,6 +113,8 @@ void test_get_peer_state () assert (rc == 5); } rc = zmq_socket_get_peer_state (router, "Y", 1); + if (rc == -1) + printf ("zmq_socket_get_peer_state failed: %i\n", errno); assert (rc != -1); if (rc & ZMQ_POLLOUT) { rc = zmq_send (router, "Y", 1, ZMQ_SNDMORE); @@ -115,6 +130,7 @@ void test_get_peer_state () // never read from dealer2, so its pipe becomes full eventually } } + zmq_poller_destroy (&poller); rc = zmq_close (router); assert (rc == 0); From 48a1e637b6225fec266cb94ccdcf9ada59fa83fe Mon Sep 17 00:00:00 2001 From: sigiesec Date: Tue, 29 Aug 2017 11:51:01 +0200 Subject: [PATCH 05/10] Problem: zmq_socket_get_peer_state is not implemented Solution: add initial implementation --- src/router.cpp | 21 +++++++++++++++++++++ src/router.hpp | 1 + src/socket_base.cpp | 8 ++++++++ src/socket_base.hpp | 6 +++++- src/zmq.cpp | 9 ++++++--- 5 files changed, 41 insertions(+), 4 deletions(-) diff --git a/src/router.cpp b/src/router.cpp index 0e28c2ce..e9e71636 100644 --- a/src/router.cpp +++ b/src/router.cpp @@ -443,6 +443,27 @@ zmq::blob_t zmq::router_t::get_credential () const return fq.get_credential (); } +int zmq::router_t::get_peer_state (const void *identity, + size_t identity_size) const +{ + int res = 0; + + blob_t identity_blob ((unsigned char *) identity, identity_size); + outpipes_t::const_iterator it = outpipes.find (identity_blob); + if (it == outpipes.end ()) { + errno = EHOSTUNREACH; + return -1; + } + + const outpipe_t &outpipe = it->second; + if (outpipe.pipe->check_hwm ()) + res |= ZMQ_POLLOUT; + + /** \todo does it make any sense to check the inpipe as well? */ + + return res; +} + bool zmq::router_t::identify_peer (pipe_t *pipe_) { msg_t msg; diff --git a/src/router.hpp b/src/router.hpp index f693000e..f991aa67 100644 --- a/src/router.hpp +++ b/src/router.hpp @@ -64,6 +64,7 @@ namespace zmq void xread_activated (zmq::pipe_t *pipe_); void xwrite_activated (zmq::pipe_t *pipe_); void xpipe_terminated (zmq::pipe_t *pipe_); + int get_peer_state (const void *identity, size_t identity_size) const; protected: diff --git a/src/socket_base.cpp b/src/socket_base.cpp index 44f904fd..60a22e52 100644 --- a/src/socket_base.cpp +++ b/src/socket_base.cpp @@ -221,6 +221,14 @@ zmq::socket_base_t::socket_base_t (ctx_t *parent_, uint32_t tid_, int sid_, bool } } +int zmq::socket_base_t::get_peer_state (const void *identity, + size_t identity_size) const +{ + // Only ROUTER sockets support this + errno = ENOTSUP; + return -1; +} + zmq::socket_base_t::~socket_base_t () { if (mailbox) diff --git a/src/socket_base.hpp b/src/socket_base.hpp index a8b100e1..b54af708 100644 --- a/src/socket_base.hpp +++ b/src/socket_base.hpp @@ -138,6 +138,11 @@ namespace zmq void event_handshake_failed_auth(const std::string &addr_, int err_); void event_handshake_succeeded(const std::string &addr_, int err_); + // Query the state of a specific peer. The default implementation + // always returns an ENOTSUP error. + virtual int get_peer_state (const void *identity, + size_t identity_size) const; + protected: socket_base_t (zmq::ctx_t *parent_, uint32_t tid_, int sid_, bool thread_safe_ = false); @@ -180,7 +185,6 @@ namespace zmq // Delay actual destruction of the socket. void process_destroy (); - // Next assigned name on a zmq_connect() call used by ROUTER and STREAM socket types std::string connect_rid; diff --git a/src/zmq.cpp b/src/zmq.cpp index 23a877c0..e39bfa65 100644 --- a/src/zmq.cpp +++ b/src/zmq.cpp @@ -1358,12 +1358,15 @@ int zmq_poller_wait_all (void *poller_, zmq_poller_event_t *events_, int n_event // Peer-specific state -int zmq_socket_get_peer_state (void *socket, +int zmq_socket_get_peer_state (void *s_, const void *identity, size_t identity_size) { - errno = ENOTSUP; - return -1; + zmq::socket_base_t *s = as_socket_base_t (s_); + if (!s) + return -1; + + return s->get_peer_state (identity, identity_size); } // Timers From f70097c1cf277068e5aea459ceeee1e043adb413 Mon Sep 17 00:00:00 2001 From: sigiesec Date: Tue, 29 Aug 2017 13:00:41 +0200 Subject: [PATCH 06/10] Problem: test does not trigger HWM Solution: modify order of operations, add diagnostic output --- tests/test_router_mandatory.cpp | 69 +++++++++++++++++++++------------ 1 file changed, 45 insertions(+), 24 deletions(-) diff --git a/tests/test_router_mandatory.cpp b/tests/test_router_mandatory.cpp index bc6677b3..93c7fe48 100644 --- a/tests/test_router_mandatory.cpp +++ b/tests/test_router_mandatory.cpp @@ -32,34 +32,44 @@ void test_get_peer_state () { #ifdef ZMQ_BUILD_DRAFT_API - size_t len = MAX_SOCKET_STRING; - char my_endpoint[MAX_SOCKET_STRING]; void *ctx = zmq_ctx_new (); assert (ctx); void *router = zmq_socket (ctx, ZMQ_ROUTER); assert (router); - int rc = zmq_bind (router, "tcp://127.0.0.1:*"); - assert (rc == 0); - - rc = zmq_getsockopt (router, ZMQ_LAST_ENDPOINT, my_endpoint, &len); - assert (rc == 0); - + int rc; int mandatory = 1; rc = zmq_setsockopt (router, ZMQ_ROUTER_MANDATORY, &mandatory, sizeof (mandatory)); - // Create dealer called "X" and connect it to our router + rc = zmq_bind (router, "tcp://127.0.0.1:*"); + assert (rc == 0); + + size_t my_endpoint_len = MAX_SOCKET_STRING; + char my_endpoint[MAX_SOCKET_STRING]; + rc = + zmq_getsockopt (router, ZMQ_LAST_ENDPOINT, my_endpoint, &my_endpoint_len); + assert (rc == 0); + void *dealer1 = zmq_socket (ctx, ZMQ_DEALER); assert (dealer1); + + void *dealer2 = zmq_socket (ctx, ZMQ_DEALER); + assert (dealer2); + + // Lower HWMs to allow doing the test with fewer messages + int hwm = 100; + zmq_setsockopt (router, ZMQ_SNDHWM, &hwm, sizeof (int)); + zmq_setsockopt (dealer1, ZMQ_RCVHWM, &hwm, sizeof (int)); + zmq_setsockopt (dealer2, ZMQ_RCVHWM, &hwm, sizeof (int)); + + // Name dealer1 "X" and connect it to our router rc = zmq_setsockopt (dealer1, ZMQ_IDENTITY, "X", 1); assert (rc == 0); rc = zmq_connect (dealer1, my_endpoint); assert (rc == 0); - // Create dealer called "Y" and connect it to our router - void *dealer2 = zmq_socket (ctx, ZMQ_DEALER); - assert (dealer2); + // Name dealer2 "Y" and connect it to our router rc = zmq_setsockopt (dealer2, ZMQ_IDENTITY, "Y", 1); assert (rc == 0); rc = zmq_connect (dealer2, my_endpoint); @@ -83,7 +93,6 @@ void test_get_peer_state () rc = zmq_recv (router, buffer, 255, 0); assert (rc == 5); - void *poller = zmq_poller_new (); assert (poller); @@ -95,41 +104,53 @@ void test_get_peer_state () const size_t count = 10000; const size_t event_size = 2; + bool dealer2_blocked = false; + size_t dealer1_sent = 0, dealer2_sent = 0, dealer1_received = 0; zmq_poller_event_t events[event_size]; - for (size_t iterations = 0; - iterations < count - && zmq_poller_wait_all (poller, events, event_size, -1) != -1; - ++iterations) { + for (size_t iterations = 0; iterations < count; ++iterations) { + rc = zmq_poller_wait_all (poller, events, event_size, -1); + assert (rc != -1); for (size_t i = 0; i < event_size; ++i) { - if (events[i].socket == router) { + if (events[i].socket == router && events[i].events & ZMQ_POLLOUT) { rc = zmq_socket_get_peer_state (router, "X", 1); if (rc == -1) printf ("zmq_socket_get_peer_state failed: %i\n", errno); assert (rc != -1); if (rc & ZMQ_POLLOUT) { - rc = zmq_send (router, "X", 1, ZMQ_SNDMORE); + rc = zmq_send (router, "X", 1, ZMQ_SNDMORE | ZMQ_DONTWAIT); assert (rc == 1); - rc = zmq_send (router, "Hello", 5, 0); + rc = zmq_send (router, "Hello", 5, ZMQ_DONTWAIT); assert (rc == 5); + + ++dealer1_sent; } + rc = zmq_socket_get_peer_state (router, "Y", 1); if (rc == -1) printf ("zmq_socket_get_peer_state failed: %i\n", errno); assert (rc != -1); if (rc & ZMQ_POLLOUT) { - rc = zmq_send (router, "Y", 1, ZMQ_SNDMORE); + rc = zmq_send (router, "Y", 1, ZMQ_SNDMORE | ZMQ_DONTWAIT); assert (rc == 1); - rc = zmq_send (router, "Hello", 5, 0); + rc = zmq_send (router, "Hello", 5, ZMQ_DONTWAIT); assert (rc == 5); + ++dealer2_sent; + } else { + dealer2_blocked = true; } } - if (events[i].socket == dealer1) { + if (events[i].socket == dealer1 && events[i].events & ZMQ_POLLIN) { rc = zmq_recv (dealer1, buffer, 255, ZMQ_DONTWAIT); assert (rc == 5); + + ++dealer1_received; } // never read from dealer2, so its pipe becomes full eventually } } + printf ("dealer1_sent = %zu, dealer2_sent = %zu, dealer1_received = %zu\n", + dealer1_sent, dealer2_sent, dealer1_received); + assert (dealer2_blocked); zmq_poller_destroy (&poller); rc = zmq_close (router); @@ -217,5 +238,5 @@ int main (void) test_basic (); test_get_peer_state (); - return 0 ; + return 0; } From fc334bc759addf7fcd6d6f710af812813cbcc01c Mon Sep 17 00:00:00 2001 From: sigiesec Date: Wed, 30 Aug 2017 09:38:44 +0200 Subject: [PATCH 07/10] Problem: unclean and duplicated test code Solution: refactoring --- tests/test_router_mandatory.cpp | 81 +++++++++++++++++++-------------- 1 file changed, 48 insertions(+), 33 deletions(-) diff --git a/tests/test_router_mandatory.cpp b/tests/test_router_mandatory.cpp index 93c7fe48..e467eab1 100644 --- a/tests/test_router_mandatory.cpp +++ b/tests/test_router_mandatory.cpp @@ -29,6 +29,26 @@ #include "testutil.hpp" +#ifdef ZMQ_BUILD_DRAFT_API +bool send_msg_to_peer_if_ready (void *router, const char *peer_identity) +{ + int rc = zmq_socket_get_peer_state (router, peer_identity, 1); + if (rc == -1) + printf ("zmq_socket_get_peer_state failed for %s: %i\n", peer_identity, + errno); + assert (rc != -1); + if (rc & ZMQ_POLLOUT) { + rc = zmq_send (router, peer_identity, 1, ZMQ_SNDMORE | ZMQ_DONTWAIT); + assert (rc == 1); + rc = zmq_send (router, "Hello", 5, ZMQ_DONTWAIT); + assert (rc == 5); + + return true; + } + return false; +} +#endif + void test_get_peer_state () { #ifdef ZMQ_BUILD_DRAFT_API @@ -59,18 +79,24 @@ void test_get_peer_state () // Lower HWMs to allow doing the test with fewer messages int hwm = 100; - zmq_setsockopt (router, ZMQ_SNDHWM, &hwm, sizeof (int)); - zmq_setsockopt (dealer1, ZMQ_RCVHWM, &hwm, sizeof (int)); - zmq_setsockopt (dealer2, ZMQ_RCVHWM, &hwm, sizeof (int)); + rc = zmq_setsockopt (router, ZMQ_SNDHWM, &hwm, sizeof (int)); + assert (rc == 0); + rc = zmq_setsockopt (dealer1, ZMQ_RCVHWM, &hwm, sizeof (int)); + assert (rc == 0); + rc = zmq_setsockopt (dealer2, ZMQ_RCVHWM, &hwm, sizeof (int)); + assert (rc == 0); + + const char *dealer1_identity = "X"; + const char *dealer2_identity = "Y"; // Name dealer1 "X" and connect it to our router - rc = zmq_setsockopt (dealer1, ZMQ_IDENTITY, "X", 1); + rc = zmq_setsockopt (dealer1, ZMQ_IDENTITY, dealer1_identity, 1); assert (rc == 0); rc = zmq_connect (dealer1, my_endpoint); assert (rc == 0); // Name dealer2 "Y" and connect it to our router - rc = zmq_setsockopt (dealer2, ZMQ_IDENTITY, "Y", 1); + rc = zmq_setsockopt (dealer2, ZMQ_IDENTITY, dealer2_identity, 1); assert (rc == 0); rc = zmq_connect (dealer2, my_endpoint); assert (rc == 0); @@ -81,7 +107,7 @@ void test_get_peer_state () assert (rc == 5); rc = zmq_recv (router, buffer, 255, 0); assert (rc == 1); - assert (buffer[0] == 'X'); + assert (0 == memcmp (buffer, dealer1_identity, rc)); rc = zmq_recv (router, buffer, 255, 0); assert (rc == 5); @@ -89,7 +115,7 @@ void test_get_peer_state () assert (rc == 5); rc = zmq_recv (router, buffer, 255, 0); assert (rc == 1); - assert (buffer[0] == 'Y'); + assert (0 == memcmp (buffer, dealer2_identity, rc)); rc = zmq_recv (router, buffer, 255, 0); assert (rc == 5); @@ -107,41 +133,30 @@ void test_get_peer_state () bool dealer2_blocked = false; size_t dealer1_sent = 0, dealer2_sent = 0, dealer1_received = 0; zmq_poller_event_t events[event_size]; - for (size_t iterations = 0; iterations < count; ++iterations) { + for (size_t iteration = 0; iteration < count; ++iteration) { rc = zmq_poller_wait_all (poller, events, event_size, -1); assert (rc != -1); - for (size_t i = 0; i < event_size; ++i) { - if (events[i].socket == router && events[i].events & ZMQ_POLLOUT) { - rc = zmq_socket_get_peer_state (router, "X", 1); - if (rc == -1) - printf ("zmq_socket_get_peer_state failed: %i\n", errno); - assert (rc != -1); - if (rc & ZMQ_POLLOUT) { - rc = zmq_send (router, "X", 1, ZMQ_SNDMORE | ZMQ_DONTWAIT); - assert (rc == 1); - rc = zmq_send (router, "Hello", 5, ZMQ_DONTWAIT); - assert (rc == 5); - + for (size_t event_no = 0; event_no < event_size; ++event_no) { + const zmq_poller_event_t ¤t_event = events[event_no]; + if (current_event.socket == router + && current_event.events & ZMQ_POLLOUT) { + if (send_msg_to_peer_if_ready (router, dealer1_identity)) ++dealer1_sent; - } - rc = zmq_socket_get_peer_state (router, "Y", 1); - if (rc == -1) - printf ("zmq_socket_get_peer_state failed: %i\n", errno); - assert (rc != -1); - if (rc & ZMQ_POLLOUT) { - rc = zmq_send (router, "Y", 1, ZMQ_SNDMORE | ZMQ_DONTWAIT); - assert (rc == 1); - rc = zmq_send (router, "Hello", 5, ZMQ_DONTWAIT); - assert (rc == 5); + if (send_msg_to_peer_if_ready (router, dealer2_identity)) ++dealer2_sent; - } else { + else dealer2_blocked = true; - } } - if (events[i].socket == dealer1 && events[i].events & ZMQ_POLLIN) { + if (current_event.socket == dealer1 + && current_event.events & ZMQ_POLLIN) { rc = zmq_recv (dealer1, buffer, 255, ZMQ_DONTWAIT); assert (rc == 5); + int more; + size_t more_size = sizeof (more); + rc = zmq_getsockopt (dealer1, ZMQ_RCVMORE, &more, &more_size); + assert (rc == 0); + assert (!more); ++dealer1_received; } From eeccbbd6f89df3248ad5f88116d4098d7c43104c Mon Sep 17 00:00:00 2001 From: sigiesec Date: Fri, 1 Sep 2017 15:51:57 +0200 Subject: [PATCH 08/10] Problem: test case fails with tcp transport Solution: use inproc transport instead --- tests/test_router_mandatory.cpp | 9 ++------- 1 file changed, 2 insertions(+), 7 deletions(-) diff --git a/tests/test_router_mandatory.cpp b/tests/test_router_mandatory.cpp index e467eab1..2062e513 100644 --- a/tests/test_router_mandatory.cpp +++ b/tests/test_router_mandatory.cpp @@ -62,13 +62,8 @@ void test_get_peer_state () rc = zmq_setsockopt (router, ZMQ_ROUTER_MANDATORY, &mandatory, sizeof (mandatory)); - rc = zmq_bind (router, "tcp://127.0.0.1:*"); - assert (rc == 0); - - size_t my_endpoint_len = MAX_SOCKET_STRING; - char my_endpoint[MAX_SOCKET_STRING]; - rc = - zmq_getsockopt (router, ZMQ_LAST_ENDPOINT, my_endpoint, &my_endpoint_len); + const char *my_endpoint = "inproc://test_get_peer_state"; + rc = zmq_bind (router, my_endpoint); assert (rc == 0); void *dealer1 = zmq_socket (ctx, ZMQ_DEALER); From f3b268d84f82fb13fe9c92882484ee5480664a6d Mon Sep 17 00:00:00 2001 From: sigiesec Date: Fri, 1 Sep 2017 16:27:53 +0200 Subject: [PATCH 09/10] Problem: no tests for error cases of zmq_socket_get_peer_state Solution: added tests --- tests/test_router_mandatory.cpp | 40 +++++++++++++++++++++++++++++++++ 1 file changed, 40 insertions(+) diff --git a/tests/test_router_mandatory.cpp b/tests/test_router_mandatory.cpp index 2062e513..11f3fe5b 100644 --- a/tests/test_router_mandatory.cpp +++ b/tests/test_router_mandatory.cpp @@ -177,6 +177,45 @@ void test_get_peer_state () #endif } +void test_get_peer_state_corner_cases () +{ +#ifdef ZMQ_BUILD_DRAFT_API + const char peer_identity[] = "foo"; + + // call get_peer_state with NULL socket + int rc = + zmq_socket_get_peer_state (NULL, peer_identity, strlen (peer_identity)); + assert (rc == -1 && errno == ENOTSOCK); + + void *ctx = zmq_ctx_new (); + assert (ctx); + void *dealer = zmq_socket (ctx, ZMQ_DEALER); + assert (dealer); + void *router = zmq_socket (ctx, ZMQ_ROUTER); + assert (router); + + // call get_peer_state with a non-ROUTER socket + rc = + zmq_socket_get_peer_state (dealer, peer_identity, strlen (peer_identity)); + assert (rc == -1 && errno == ENOTSUP); + + // call get_peer_state for an unknown identity + rc = + zmq_socket_get_peer_state (router, peer_identity, strlen (peer_identity)); + assert (rc == -1 && errno == EHOSTUNREACH); + + rc = zmq_close (router); + assert (rc == 0); + + rc = zmq_close (dealer); + assert (rc == 0); + + rc = zmq_ctx_term (ctx); + assert (rc == 0); + +#endif +} + void test_basic () { size_t len = MAX_SOCKET_STRING; @@ -247,6 +286,7 @@ int main (void) test_basic (); test_get_peer_state (); + test_get_peer_state_corner_cases (); return 0; } From 79e28af4ceec3e4710460dbb7f598777031cc0fc Mon Sep 17 00:00:00 2001 From: sigiesec Date: Fri, 1 Sep 2017 17:15:23 +0200 Subject: [PATCH 10/10] Problem: new function zmq_socket_get_peer_state not in zmq_draft.h Solution: added function to zmq_draft.h --- src/zmq_draft.h | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/zmq_draft.h b/src/zmq_draft.h index c0a88980..30371662 100644 --- a/src/zmq_draft.h +++ b/src/zmq_draft.h @@ -141,6 +141,10 @@ int zmq_poller_modify_fd (void *poller, int fd, short events); int zmq_poller_remove_fd (void *poller, int fd); #endif +int zmq_socket_get_peer_state (void *socket, + const void *identity, + size_t identity_size); + /******************************************************************************/ /* Scheduling timers */ /******************************************************************************/