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/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 dca9088c..e39bfa65 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_) @@ -1389,6 +1356,19 @@ 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 *s_, + const void *identity, + size_t identity_size) +{ + zmq::socket_base_t *s = as_socket_base_t (s_); + if (!s) + return -1; + + return s->get_peer_state (identity, identity_size); +} + // Timers void *zmq_timers_new (void) 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 */ /******************************************************************************/ diff --git a/tests/test_router_mandatory.cpp b/tests/test_router_mandatory.cpp index 40325625..11f3fe5b 100644 --- a/tests/test_router_mandatory.cpp +++ b/tests/test_router_mandatory.cpp @@ -29,9 +29,195 @@ #include "testutil.hpp" -int main (void) +#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 + void *ctx = zmq_ctx_new (); + assert (ctx); + void *router = zmq_socket (ctx, ZMQ_ROUTER); + assert (router); + + int rc; + int mandatory = 1; + rc = zmq_setsockopt (router, ZMQ_ROUTER_MANDATORY, &mandatory, + sizeof (mandatory)); + + 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); + assert (dealer1); + + void *dealer2 = zmq_socket (ctx, ZMQ_DEALER); + assert (dealer2); + + // Lower HWMs to allow doing the test with fewer messages + int hwm = 100; + 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, 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, dealer2_identity, 1); + assert (rc == 0); + rc = zmq_connect (dealer2, my_endpoint); + assert (rc == 0); + + // 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 (0 == memcmp (buffer, dealer1_identity, rc)); + 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 (0 == memcmp (buffer, dealer2_identity, rc)); + rc = zmq_recv (router, buffer, 255, 0); + assert (rc == 5); + + 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; + 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 iteration = 0; iteration < count; ++iteration) { + rc = zmq_poller_wait_all (poller, events, event_size, -1); + assert (rc != -1); + 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; + + if (send_msg_to_peer_if_ready (router, dealer2_identity)) + ++dealer2_sent; + else + dealer2_blocked = true; + } + 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; + } + // 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); + 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_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 () { - setup_test_environment(); size_t len = MAX_SOCKET_STRING; char my_endpoint[MAX_SOCKET_STRING]; void *ctx = zmq_ctx_new (); @@ -55,7 +241,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 +256,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 +269,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 +278,15 @@ int main (void) rc = zmq_ctx_term (ctx); assert (rc == 0); - - return 0 ; +} + +int main (void) +{ + setup_test_environment (); + + test_basic (); + test_get_peer_state (); + test_get_peer_state_corner_cases (); + + return 0; }