Problem: Thread-safe solution for modify hwm of pipe

Solution: where change pipe hwm, send a command (new type pipe_hwm) to peer, so peer pipe can modify hwm thread-safely
This commit is contained in:
laplaceyang 2017-01-11 10:49:54 +08:00
parent 4fc313d152
commit 107f2441d5
6 changed files with 43 additions and 0 deletions

View File

@ -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 {

View File

@ -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);

View File

@ -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 ();

View File

@ -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_);
}

View File

@ -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 ();

View File

@ -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);
}
}