Problem: cannot monitor state of queues at runtime

Solution: add API and ZMQ_EVENT_PIPES_STATS event which generates 2
values, one for the egress and one for the ingress pipes respectively.
Refactor the events code to be able to send multiple values.

Author: bluca

doc/zmq_socket_monitor_versioned.txt

@@ -11,6 +11,7 @@ zmq_socket_monitor_versioned - monitor socket events
 SYNOPSIS
 --------
 *int zmq_socket_monitor_versioned (void '*socket', char '*endpoint', uint64_t 'events', int 'event_version');*
+*void zmq_socket_monitor_pipes_stats (void '*socket');*
 
 
 DESCRIPTION
@@ -43,16 +44,21 @@ event number.
 
 Unless it is specified differently, the second frame contains an event
 value (64 bits) that provides additional data according to the event number.
-Some events might define additional value frames following the second one.
-The third and fourth frames contain strings that specifies the affected
-connection or endpoint. The third frame contains a string denoting the local
-endpoint, while the fourth frame contains a string denoting the remote endpoint.
+The third frame contains the number of value frames that will follow it as a 64
+bits integer. Each event type might have a different number of values.
+The second-to-last and last frames contain strings that specifies the affected
+connection or endpoint. The former frame contains a string denoting the local
+endpoint, while the latter frame contains a string denoting the remote endpoint.
 Either of these may be empty, depending on the event type and whether the
 connection uses a bound or connected local endpoint.
 
 Note that the format of the second and further frames, and also the number of
 frames, may be different for events added in the future.
 
+The _zmq_socket_monitor_pipes_stats()_ method triggers an event of type
+ZMQ_EVENT_PIPES_STATS for each connected peer of the monitored socket.
+NOTE: _zmq_socket_monitor_pipes_stats()_ is in DRAFT state.
+
 ----
 Monitoring events are only generated by some transports: At the moment these
 are SOCKS, TCP, IPC, and TIPC. Note that it is not an error to call
@@ -168,6 +174,24 @@ The ZMTP security mechanism handshake failed due to an authentication failure.
 The event value is the status code returned by the ZAP handler (i.e. 300, 
 400 or 500).
 
+----
+
+Supported events (v2)
+----------------
+
+ZMQ_EVENT_PIPE_STATS
+~~~~~~~~~~~~~~~~~~~~
+This event provides two values, the number of messages in each of the two
+queues associated with the returned endpoint (respectively egress and ingress).
+This event only triggers after calling the function
+_zmq_socket_monitor_pipes_stats()_.
+NOTE: this measurement is asynchronous, so by the time the message is received
+the internal state might have already changed.
+NOTE: when the monitored socket and the monitor are not used in a poll, the
+event might not be delivered until an API has been called on the monitored
+socket, like zmq_getsockopt for example (the option is irrelevant).
+NOTE: in DRAFT state, not yet available in stable releases.
+
 
 
 RETURN VALUE

include/zmq.h

@@ -726,16 +726,20 @@ ZMQ_EXPORT int zmq_socket_get_peer_state (void *socket,
                                           const void *routing_id,
                                           size_t routing_id_size);
 
+/*  DRAFT Socket monitoring events                                            */
+#define ZMQ_EVENT_PIPES_STATS 0x10000
+
 #define ZMQ_CURRENT_EVENT_VERSION 1
 #define ZMQ_CURRENT_EVENT_VERSION_DRAFT 2
 
 #define ZMQ_EVENT_ALL_V1 ZMQ_EVENT_ALL
-#define ZMQ_EVENT_ALL_V2 ZMQ_EVENT_ALL_V1
+#define ZMQ_EVENT_ALL_V2 ZMQ_EVENT_ALL_V1 | ZMQ_EVENT_PIPES_STATS
 
 ZMQ_EXPORT int zmq_socket_monitor_versioned (void *s_,
                                              const char *addr_,
                                              uint64_t events_,
                                              int event_version_);
+ZMQ_EXPORT void zmq_socket_monitor_pipes_stats (void *s);
 
 #endif // ZMQ_BUILD_DRAFT_API
 

src/command.hpp

@@ -32,6 +32,7 @@
 
 #include <string>
 #include "stdint.hpp"
+#include "endpoint.hpp"
 
 namespace zmq
 {
@@ -73,6 +74,8 @@ __declspec(align (64))
         reap,
         reaped,
         inproc_connected,
+        pipe_peer_stats,
+        pipe_stats_publish,
         done
     } type;
 
@@ -186,6 +189,23 @@ __declspec(align (64))
         {
         } reaped;
 
+        //  Send application-side pipe count and ask to send monitor event
+        struct
+        {
+            uint64_t queue_count;
+            zmq::own_t *socket_base;
+            endpoint_uri_pair_t *endpoint_pair;
+        } pipe_peer_stats;
+
+        //  Collate application thread and I/O thread pipe counts and endpoints
+        //  and send as event
+        struct
+        {
+            uint64_t outbound_queue_count;
+            uint64_t inbound_queue_count;
+            endpoint_uri_pair_t *endpoint_pair;
+        } pipe_stats_publish;
+
         //  Sent by reaper thread to the term thread when all the sockets
         //  are successfully deallocated.
         struct

src/object.cpp

@@ -107,6 +107,19 @@ void zmq::object_t::process_command (command_t &cmd_)
             process_hiccup (cmd_.args.hiccup.pipe);
             break;
 
+        case command_t::pipe_peer_stats:
+            process_pipe_peer_stats (cmd_.args.pipe_peer_stats.queue_count,
+                                     cmd_.args.pipe_peer_stats.socket_base,
+                                     cmd_.args.pipe_peer_stats.endpoint_pair);
+            break;
+
+        case command_t::pipe_stats_publish:
+            process_pipe_stats_publish (
+              cmd_.args.pipe_stats_publish.outbound_queue_count,
+              cmd_.args.pipe_stats_publish.inbound_queue_count,
+              cmd_.args.pipe_stats_publish.endpoint_pair);
+            break;
+
         case command_t::pipe_term:
             process_pipe_term ();
             break;
@@ -285,6 +298,35 @@ void zmq::object_t::send_hiccup (pipe_t *destination_, void *pipe_)
     send_command (cmd);
 }
 
+void zmq::object_t::send_pipe_peer_stats (pipe_t *destination_,
+                                          uint64_t queue_count_,
+                                          own_t *socket_base_,
+                                          endpoint_uri_pair_t *endpoint_pair_)
+{
+    command_t cmd;
+    cmd.destination = destination_;
+    cmd.type = command_t::pipe_peer_stats;
+    cmd.args.pipe_peer_stats.queue_count = queue_count_;
+    cmd.args.pipe_peer_stats.socket_base = socket_base_;
+    cmd.args.pipe_peer_stats.endpoint_pair = endpoint_pair_;
+    send_command (cmd);
+}
+
+void zmq::object_t::send_pipe_stats_publish (
+  own_t *destination_,
+  uint64_t outbound_queue_count_,
+  uint64_t inbound_queue_count_,
+  endpoint_uri_pair_t *endpoint_pair_)
+{
+    command_t cmd;
+    cmd.destination = destination_;
+    cmd.type = command_t::pipe_stats_publish;
+    cmd.args.pipe_stats_publish.outbound_queue_count = outbound_queue_count_;
+    cmd.args.pipe_stats_publish.inbound_queue_count = inbound_queue_count_;
+    cmd.args.pipe_stats_publish.endpoint_pair = endpoint_pair_;
+    send_command (cmd);
+}
+
 void zmq::object_t::send_pipe_term (pipe_t *destination_)
 {
     command_t cmd;
@@ -422,6 +464,20 @@ void zmq::object_t::process_hiccup (void *)
     zmq_assert (false);
 }
 
+void zmq::object_t::process_pipe_peer_stats (uint64_t,
+                                             own_t *,
+                                             endpoint_uri_pair_t *)
+{
+    zmq_assert (false);
+}
+
+void zmq::object_t::process_pipe_stats_publish (uint64_t,
+                                                uint64_t,
+                                                endpoint_uri_pair_t *)
+{
+    zmq_assert (false);
+}
+
 void zmq::object_t::process_pipe_term ()
 {
     zmq_assert (false);

src/object.hpp

@@ -32,6 +32,7 @@
 
 #include <string>
 #include "stdint.hpp"
+#include "endpoint.hpp"
 
 namespace zmq
 {
@@ -96,6 +97,14 @@ class object_t
     void send_activate_read (zmq::pipe_t *destination_);
     void send_activate_write (zmq::pipe_t *destination_, uint64_t msgs_read_);
     void send_hiccup (zmq::pipe_t *destination_, void *pipe_);
+    void send_pipe_peer_stats (zmq::pipe_t *destination_,
+                               uint64_t queue_count_,
+                               zmq::own_t *socket_base,
+                               endpoint_uri_pair_t *endpoint_pair_);
+    void send_pipe_stats_publish (zmq::own_t *destination_,
+                                  uint64_t outbound_queue_count_,
+                                  uint64_t inbound_queue_count_,
+                                  endpoint_uri_pair_t *endpoint_pair_);
     void send_pipe_term (zmq::pipe_t *destination_);
     void send_pipe_term_ack (zmq::pipe_t *destination_);
     void send_pipe_hwm (zmq::pipe_t *destination_, int inhwm_, int outhwm_);
@@ -117,6 +126,13 @@ class object_t
     virtual void process_activate_read ();
     virtual void process_activate_write (uint64_t msgs_read_);
     virtual void process_hiccup (void *pipe_);
+    virtual void process_pipe_peer_stats (uint64_t queue_count_,
+                                          zmq::own_t *socket_base_,
+                                          endpoint_uri_pair_t *endpoint_pair_);
+    virtual void
+    process_pipe_stats_publish (uint64_t outbound_queue_count_,
+                                uint64_t inbound_queue_count_,
+                                endpoint_uri_pair_t *endpoint_pair_);
     virtual void process_pipe_term ();
     virtual void process_pipe_term_ack ();
     virtual void process_pipe_hwm (int inhwm_, int outhwm_);

src/pipe.cpp

@@ -563,3 +563,19 @@ const zmq::endpoint_uri_pair_t &zmq::pipe_t::get_endpoint_pair () const
 {
     return _endpoint_pair;
 }
+
+void zmq::pipe_t::send_stats_to_peer (own_t *socket_base_)
+{
+    endpoint_uri_pair_t *ep =
+      new (std::nothrow) endpoint_uri_pair_t (_endpoint_pair);
+    send_pipe_peer_stats (_peer, _msgs_written - _peers_msgs_read, socket_base_,
+                          ep);
+}
+
+void zmq::pipe_t::process_pipe_peer_stats (uint64_t queue_count_,
+                                           own_t *socket_base_,
+                                           endpoint_uri_pair_t *endpoint_pair_)
+{
+    send_pipe_stats_publish (socket_base_, queue_count_,
+                             _msgs_written - _peers_msgs_read, endpoint_pair_);
+}

src/pipe.hpp

@@ -145,6 +145,8 @@ class pipe_t : public object_t,
     void set_endpoint_pair (endpoint_uri_pair_t endpoint_pair_);
     const endpoint_uri_pair_t &get_endpoint_pair () const;
 
+    void send_stats_to_peer (own_t *socket_base_);
+
   private:
     //  Type of the underlying lock-free pipe.
     typedef ypipe_base_t<msg_t> upipe_t;
@@ -153,6 +155,9 @@ class pipe_t : public object_t,
     void process_activate_read ();
     void process_activate_write (uint64_t msgs_read_);
     void process_hiccup (void *pipe_);
+    void process_pipe_peer_stats (uint64_t queue_count_,
+                                  own_t *socket_base_,
+                                  endpoint_uri_pair_t *endpoint_pair_);
     void process_pipe_term ();
     void process_pipe_term_ack ();
     void process_pipe_hwm (int inhwm_, int outhwm_);

src/session_base.cpp

@@ -409,6 +409,11 @@ void zmq::session_base_t::process_attach (i_engine *engine_)
         zmq_assert (!_pipe);
         _pipe = pipes[0];
 
+        //  The endpoints strings are not set on bind, set them here so that
+        //  events can use them.
+        pipes[0]->set_endpoint_pair (engine_->get_endpoint ());
+        pipes[1]->set_endpoint_pair (engine_->get_endpoint ());
+
         //  Ask socket to plug into the remote end of the pipe.
         send_bind (_socket, pipes[1]);
     }

src/socket_base.cpp

@@ -1421,6 +1421,32 @@ void zmq::socket_base_t::process_term_endpoint (std::string *endpoint_)
     delete endpoint_;
 }
 
+void zmq::socket_base_t::process_pipe_stats_publish (
+  uint64_t outbound_queue_count_,
+  uint64_t inbound_queue_count_,
+  endpoint_uri_pair_t *endpoint_pair_)
+{
+    uint64_t values[2] = {outbound_queue_count_, inbound_queue_count_};
+    event (*endpoint_pair_, values, 2, ZMQ_EVENT_PIPES_STATS);
+    delete endpoint_pair_;
+}
+
+/*
+ * There are 2 pipes per connection, and the inbound one _must_ be queried from
+ * the I/O thread. So ask the outbound pipe, in the application thread, to send
+ * a message (pipe_peer_stats) to its peer. The message will carry the outbound
+ * pipe stats and endpoint, and the reference to the socket object.
+ * The inbound pipe on the I/O thread will then add its own stats and endpoint,
+ * and write back a message to the socket object (pipe_stats_publish) which
+ * will raise an event with the data.
+ */
+void zmq::socket_base_t::query_pipes_stats ()
+{
+    for (pipes_t::size_type i = 0; i != _pipes.size (); ++i) {
+        _pipes[i]->send_stats_to_peer (this);
+    }
+}
+
 void zmq::socket_base_t::update_pipe_options (int option_)
 {
     if (option_ == ZMQ_SNDHWM || option_ == ZMQ_RCVHWM) {

src/socket_base.hpp

@@ -157,6 +157,8 @@ class socket_base_t : public own_t,
     virtual int get_peer_state (const void *routing_id_,
                                 size_t routing_id_size_) const;
 
+    void query_pipes_stats ();
+
   protected:
     socket_base_t (zmq::ctx_t *parent_,
                    uint32_t tid_,
@@ -278,6 +280,9 @@ class socket_base_t : public own_t,
     //  Handlers for incoming commands.
     void process_stop ();
     void process_bind (zmq::pipe_t *pipe_);
+    void process_pipe_stats_publish (uint64_t outbound_queue_count_,
+                                     uint64_t inbound_queue_count_,
+                                     endpoint_uri_pair_t *endpoint_pair_);
     void process_term (int linger_);
     void process_term_endpoint (std::string *endpoint_);
 

src/zmq.cpp

@@ -1452,3 +1452,10 @@ int zmq_has (const char *capability_)
     //  Whatever the application asked for, we don't have
     return false;
 }
+
+void zmq_socket_monitor_pipes_stats (void *s_)
+{
+    zmq::socket_base_t *s = as_socket_base_t (s_);
+    if (s)
+        s->query_pipes_stats ();
+}

src/zmq_draft.h

@@ -123,10 +123,20 @@ int zmq_socket_get_peer_state (void *socket_,
                                const void *routing_id_,
                                size_t routing_id_size_);
 
+/*  DRAFT Socket monitoring events                                            */
+#define ZMQ_EVENT_PIPES_STATS 0x10000
+
+#define ZMQ_CURRENT_EVENT_VERSION 1
+#define ZMQ_CURRENT_EVENT_VERSION_DRAFT 2
+
+#define ZMQ_EVENT_ALL_V1 ZMQ_EVENT_ALL
+#define ZMQ_EVENT_ALL_V2 ZMQ_EVENT_ALL_V1 | ZMQ_EVENT_PIPES_STATS
+
 int zmq_socket_monitor_versioned (void *s_,
                                   const char *addr_,
                                   uint64_t events_,
                                   int event_version_);
+void zmq_socket_monitor_pipes_stats (void *s_);
 
 #endif // ZMQ_BUILD_DRAFT_API