Merge pull request #2302 from laplaceyang/pr_thread_safe_modify_pipe

Problem: Thread-safe solution for modify hwm of pipe

src/command.hpp

@@ -64,6 +64,7 @@ namespace zmq
             hiccup,
             pipe_term,
             pipe_term_ack,
+            pipe_hwm,
             term_req,
             term,
             term_ack,
@@ -129,6 +130,12 @@ namespace zmq
             struct {
             } pipe_term_ack;
 
+            //  Sent by one of pipe to another part for modify hwm
+            struct {
+                int inhwm;
+                int outhwm;
+            } pipe_hwm;
+
             //  Sent by I/O object ot the socket to request the shutdown of
             //  the I/O object.
             struct {

src/object.cpp

@@ -118,6 +118,10 @@ void zmq::object_t::process_command (command_t &cmd_)
         process_pipe_term_ack ();
         break;
 
+    case command_t::pipe_hwm:
+        process_pipe_hwm (cmd_.args.pipe_hwm.inhwm, cmd_.args.pipe_hwm.outhwm);
+        break;
+
     case command_t::term_req:
         process_term_req (cmd_.args.term_req.object);
         break;
@@ -291,6 +295,16 @@ void zmq::object_t::send_pipe_term_ack (pipe_t *destination_)
     send_command (cmd);
 }
 
+void zmq::object_t::send_pipe_hwm (pipe_t *destination_, int inhwm_, int outhwm_)
+{
+    command_t cmd;
+    cmd.destination = destination_;
+    cmd.type = command_t::pipe_hwm;
+    cmd.args.pipe_hwm.inhwm = inhwm_;
+    cmd.args.pipe_hwm.outhwm = outhwm_;
+    send_command (cmd);
+}
+
 void zmq::object_t::send_term_req (own_t *destination_,
     own_t *object_)
 {
@@ -401,6 +415,11 @@ void zmq::object_t::process_pipe_term_ack ()
     zmq_assert (false);
 }
 
+void zmq::object_t::process_pipe_hwm (int, int)
+{
+    zmq_assert (false);
+}
+
 void zmq::object_t::process_term_req (own_t *)
 {
     zmq_assert (false);

src/object.hpp

@@ -102,6 +102,7 @@ namespace zmq
         void send_hiccup (zmq::pipe_t *destination_, void *pipe_);
         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_);
         void send_term_req (zmq::own_t *destination_,
             zmq::own_t *object_);
         void send_term (zmq::own_t *destination_, int linger_);
@@ -122,6 +123,7 @@ namespace zmq
         virtual void process_hiccup (void *pipe_);
         virtual void process_pipe_term ();
         virtual void process_pipe_term_ack ();
+        virtual void process_pipe_hwm (int inhwm_, int outhwm_);
         virtual void process_term_req (zmq::own_t *object_);
         virtual void process_term (int linger_);
         virtual void process_term_ack ();

src/pipe.cpp

@@ -377,6 +377,11 @@ void zmq::pipe_t::process_pipe_term_ack ()
     delete this;
 }
 
+void zmq::pipe_t::process_pipe_hwm (int inhwm_, int outhwm_)
+{
+    set_hwms(inhwm_, outhwm_);
+}
+
 void zmq::pipe_t::set_nodelay ()
 {
     this->delay = false;
@@ -533,3 +538,8 @@ bool zmq::pipe_t::check_hwm () const
     bool full = hwm > 0 && msgs_written - peers_msgs_read >= uint64_t (hwm);
     return( !full );
 }
+
+void zmq::pipe_t::send_hwms_to_peer(int inhwm_, int outhwm_)
+{
+    send_pipe_hwm(peer, inhwm_, outhwm_);
+}

src/pipe.hpp

@@ -136,6 +136,9 @@ namespace zmq
         //  Set the boost to high water marks, used by inproc sockets so total hwm are sum of connect and bind sockets watermarks
         void set_hwms_boost(int inhwmboost_, int outhwmboost_);
 
+        // send command to peer for notify the change of hwm
+        void send_hwms_to_peer(int inhwm_, int outhwm_);
+
         //  Returns true if HWM is not reached
         bool check_hwm () const;
     private:
@@ -149,6 +152,7 @@ namespace zmq
         void process_hiccup (void *pipe_);
         void process_pipe_term ();
         void process_pipe_term_ack ();
+        void process_pipe_hwm (int inhwm_, int outhwm_);
 
         //  Handler for delimiter read from the pipe.
         void process_delimiter ();

src/socket_base.cpp

@@ -1414,6 +1414,7 @@ void zmq::socket_base_t::update_pipe_options(int option_)
         for (pipes_t::size_type i = 0; i != pipes.size(); ++i)
         {
             pipes[i]->set_hwms(options.rcvhwm, options.sndhwm);
+            pipes[i]->send_hwms_to_peer(options.sndhwm, options.rcvhwm);
         }
     }