Commit 42000d2c authored by Martin Sustrik's avatar Martin Sustrik

terminology unified: revive->activate

parent 92923cd4
...@@ -39,8 +39,8 @@ namespace zmq ...@@ -39,8 +39,8 @@ namespace zmq
own, own,
attach, attach,
bind, bind,
revive, activate_reader,
reader_info, activate_writer,
pipe_term, pipe_term,
pipe_term_ack, pipe_term_ack,
term_req, term_req,
...@@ -83,14 +83,13 @@ namespace zmq ...@@ -83,14 +83,13 @@ namespace zmq
// Sent by pipe writer to inform dormant pipe reader that there // Sent by pipe writer to inform dormant pipe reader that there
// are messages in the pipe. // are messages in the pipe.
struct { struct {
} revive; } activate_reader;
// Sent by pipe reader to inform pipe writer // Sent by pipe reader to inform pipe writer about how many
// about how many messages it has read so far. // messages it has read so far.
// Used to implement the flow control.
struct { struct {
uint64_t msgs_read; uint64_t msgs_read;
} reader_info; } activate_writer;
// Sent by pipe reader to pipe writer to ask it to terminate // Sent by pipe reader to pipe writer to ask it to terminate
// its end of the pipe. // its end of the pipe.
......
...@@ -103,8 +103,8 @@ int zmq::fq_t::recv (zmq_msg_t *msg_, int flags_) ...@@ -103,8 +103,8 @@ int zmq::fq_t::recv (zmq_msg_t *msg_, int flags_)
// without blocking. // without blocking.
zmq_assert (!(more && !fetched)); zmq_assert (!(more && !fetched));
// Note that when message is not fetched, current pipe is killed and // Note that when message is not fetched, current pipe is deactivated
// replaced by another active pipe. Thus we don't have to increase // and replaced by another active pipe. Thus we don't have to increase
// the 'current' pointer. // the 'current' pointer.
if (fetched) { if (fetched) {
more = msg_->flags & ZMQ_MSG_MORE; more = msg_->flags & ZMQ_MSG_MORE;
......
...@@ -57,8 +57,12 @@ void zmq::object_t::process_command (command_t &cmd_) ...@@ -57,8 +57,12 @@ void zmq::object_t::process_command (command_t &cmd_)
{ {
switch (cmd_.type) { switch (cmd_.type) {
case command_t::revive: case command_t::activate_reader:
process_revive (); process_activate_reader ();
break;
case command_t::activate_writer:
process_activate_writer (cmd_.args.activate_writer.msgs_read);
break; break;
case command_t::stop: case command_t::stop:
...@@ -89,10 +93,6 @@ void zmq::object_t::process_command (command_t &cmd_) ...@@ -89,10 +93,6 @@ void zmq::object_t::process_command (command_t &cmd_)
process_seqnum (); process_seqnum ();
break; break;
case command_t::reader_info:
process_reader_info (cmd_.args.reader_info.msgs_read);
break;
case command_t::pipe_term: case command_t::pipe_term:
process_pipe_term (); process_pipe_term ();
return; return;
...@@ -248,18 +248,18 @@ void zmq::object_t::send_bind (own_t *destination_, reader_t *in_pipe_, ...@@ -248,18 +248,18 @@ void zmq::object_t::send_bind (own_t *destination_, reader_t *in_pipe_,
send_command (cmd); send_command (cmd);
} }
void zmq::object_t::send_revive (object_t *destination_) void zmq::object_t::send_activate_reader (reader_t *destination_)
{ {
command_t cmd; command_t cmd;
#if defined ZMQ_MAKE_VALGRIND_HAPPY #if defined ZMQ_MAKE_VALGRIND_HAPPY
memset (&cmd, 0, sizeof (cmd)); memset (&cmd, 0, sizeof (cmd));
#endif #endif
cmd.destination = destination_; cmd.destination = destination_;
cmd.type = command_t::revive; cmd.type = command_t::activate_reader;
send_command (cmd); send_command (cmd);
} }
void zmq::object_t::send_reader_info (writer_t *destination_, void zmq::object_t::send_activate_writer (writer_t *destination_,
uint64_t msgs_read_) uint64_t msgs_read_)
{ {
command_t cmd; command_t cmd;
...@@ -267,8 +267,8 @@ void zmq::object_t::send_reader_info (writer_t *destination_, ...@@ -267,8 +267,8 @@ void zmq::object_t::send_reader_info (writer_t *destination_,
memset (&cmd, 0, sizeof (cmd)); memset (&cmd, 0, sizeof (cmd));
#endif #endif
cmd.destination = destination_; cmd.destination = destination_;
cmd.type = command_t::reader_info; cmd.type = command_t::activate_writer;
cmd.args.reader_info.msgs_read = msgs_read_; cmd.args.activate_writer.msgs_read = msgs_read_;
send_command (cmd); send_command (cmd);
} }
...@@ -356,12 +356,12 @@ void zmq::object_t::process_bind (reader_t *in_pipe_, writer_t *out_pipe_, ...@@ -356,12 +356,12 @@ void zmq::object_t::process_bind (reader_t *in_pipe_, writer_t *out_pipe_,
zmq_assert (false); zmq_assert (false);
} }
void zmq::object_t::process_revive () void zmq::object_t::process_activate_reader ()
{ {
zmq_assert (false); zmq_assert (false);
} }
void zmq::object_t::process_reader_info (uint64_t msgs_read_) void zmq::object_t::process_activate_writer (uint64_t msgs_read_)
{ {
zmq_assert (false); zmq_assert (false);
} }
......
...@@ -68,8 +68,8 @@ namespace zmq ...@@ -68,8 +68,8 @@ namespace zmq
void send_bind (class own_t *destination_, void send_bind (class own_t *destination_,
class reader_t *in_pipe_, class writer_t *out_pipe_, class reader_t *in_pipe_, class writer_t *out_pipe_,
const blob_t &peer_identity_, bool inc_seqnum_ = true); const blob_t &peer_identity_, bool inc_seqnum_ = true);
void send_revive (class object_t *destination_); void send_activate_reader (class reader_t *destination_);
void send_reader_info (class writer_t *destination_, void send_activate_writer (class writer_t *destination_,
uint64_t msgs_read_); uint64_t msgs_read_);
void send_pipe_term (class writer_t *destination_); void send_pipe_term (class writer_t *destination_);
void send_pipe_term_ack (class reader_t *destination_); void send_pipe_term_ack (class reader_t *destination_);
...@@ -87,8 +87,8 @@ namespace zmq ...@@ -87,8 +87,8 @@ namespace zmq
const blob_t &peer_identity_); const blob_t &peer_identity_);
virtual void process_bind (class reader_t *in_pipe_, virtual void process_bind (class reader_t *in_pipe_,
class writer_t *out_pipe_, const blob_t &peer_identity_); class writer_t *out_pipe_, const blob_t &peer_identity_);
virtual void process_revive (); virtual void process_activate_reader ();
virtual void process_reader_info (uint64_t msgs_read_); virtual void process_activate_writer (uint64_t msgs_read_);
virtual void process_pipe_term (); virtual void process_pipe_term ();
virtual void process_pipe_term_ack (); virtual void process_pipe_term_ack ();
virtual void process_term_req (class own_t *object_); virtual void process_term_req (class own_t *object_);
......
...@@ -112,7 +112,7 @@ bool zmq::reader_t::read (zmq_msg_t *msg_) ...@@ -112,7 +112,7 @@ bool zmq::reader_t::read (zmq_msg_t *msg_)
msgs_read++; msgs_read++;
if (lwm > 0 && msgs_read % lwm == 0) if (lwm > 0 && msgs_read % lwm == 0)
send_reader_info (writer, msgs_read); send_activate_writer (writer, msgs_read);
return true; return true;
} }
...@@ -127,7 +127,7 @@ void zmq::reader_t::terminate () ...@@ -127,7 +127,7 @@ void zmq::reader_t::terminate ()
send_pipe_term (writer); send_pipe_term (writer);
} }
void zmq::reader_t::process_revive () void zmq::reader_t::process_activate_reader ()
{ {
// Forward the event to the sink (either socket or session). // Forward the event to the sink (either socket or session).
sink->activated (this); sink->activated (this);
...@@ -258,7 +258,7 @@ void zmq::writer_t::rollback () ...@@ -258,7 +258,7 @@ void zmq::writer_t::rollback ()
void zmq::writer_t::flush () void zmq::writer_t::flush ()
{ {
if (!pipe->flush ()) if (!pipe->flush ())
send_revive (reader); send_activate_reader (reader);
} }
void zmq::writer_t::terminate () void zmq::writer_t::terminate ()
...@@ -288,7 +288,7 @@ void zmq::writer_t::write_delimiter () ...@@ -288,7 +288,7 @@ void zmq::writer_t::write_delimiter ()
flush (); flush ();
} }
void zmq::writer_t::process_reader_info (uint64_t msgs_read_) void zmq::writer_t::process_activate_writer (uint64_t msgs_read_)
{ {
zmq_msg_t msg; zmq_msg_t msg;
......
...@@ -81,7 +81,7 @@ namespace zmq ...@@ -81,7 +81,7 @@ namespace zmq
void set_writer (class writer_t *writer_); void set_writer (class writer_t *writer_);
// Command handlers. // Command handlers.
void process_revive (); void process_activate_reader ();
void process_pipe_term_ack (); void process_pipe_term_ack ();
// Returns true if the message is delimiter; false otherwise. // Returns true if the message is delimiter; false otherwise.
...@@ -150,7 +150,7 @@ namespace zmq ...@@ -150,7 +150,7 @@ namespace zmq
uint64_t hwm_, int64_t swap_size_); uint64_t hwm_, int64_t swap_size_);
~writer_t (); ~writer_t ();
void process_reader_info (uint64_t msgs_read_); void process_activate_writer (uint64_t msgs_read_);
// Command handlers. // Command handlers.
void process_pipe_term (); void process_pipe_term ();
......
...@@ -495,8 +495,9 @@ int zmq::socket_base_t::recv (::zmq_msg_t *msg_, int flags_) ...@@ -495,8 +495,9 @@ int zmq::socket_base_t::recv (::zmq_msg_t *msg_, int flags_)
errno = err; errno = err;
// If the message cannot be fetched immediately, there are two scenarios. // If the message cannot be fetched immediately, there are two scenarios.
// For non-blocking recv, commands are processed in case there's a revive // For non-blocking recv, commands are processed in case there's an
// command already waiting int a command pipe. If it's not, return EAGAIN. // activate_reader command already waiting int a command pipe.
// If it's not, return EAGAIN.
if (flags_ & ZMQ_NOBLOCK) { if (flags_ & ZMQ_NOBLOCK) {
if (errno != EAGAIN) if (errno != EAGAIN)
return -1; return -1;
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment