Commit 6117a2b0 authored by Ian Barber's avatar Ian Barber

Revert "Replace incomplete count with a std::set"

This reverts commit 4aa5ba3d.
Unintentional merge
parent 4aa5ba3d
...@@ -111,6 +111,7 @@ zmq::session_base_t::session_base_t (class io_thread_t *io_thread_, ...@@ -111,6 +111,7 @@ zmq::session_base_t::session_base_t (class io_thread_t *io_thread_,
io_object_t (io_thread_), io_object_t (io_thread_),
connect (connect_), connect (connect_),
pipe (NULL), pipe (NULL),
incomplete_detach (0),
incomplete_in (false), incomplete_in (false),
pending (false), pending (false),
engine (NULL), engine (NULL),
...@@ -230,26 +231,31 @@ void zmq::session_base_t::clean_pipes () ...@@ -230,26 +231,31 @@ void zmq::session_base_t::clean_pipes ()
void zmq::session_base_t::terminated (pipe_t *pipe_) void zmq::session_base_t::terminated (pipe_t *pipe_)
{ {
// Drop the reference to the deallocated pipe if required. // Drop the reference to the deallocated pipe if required.
zmq_assert (pipe == pipe_ || incomplete_pipes.size () > 0); zmq_assert (pipe == pipe_ || incomplete_detach > 0);
if (pipe == pipe_) // If we still have pipes outstanding, decrement.
// If this is our current pipe, remove it // This will only have been set in a disconnect situation
pipe = NULL; // where delay_attach_on_connect is 1.
else if (incomplete_detach > 0)
// Remove the pipe from the detached pipes set incomplete_detach --;
incomplete_pipes.erase (pipe_);
// If there are still extra detached pipes, don't continue
if (incomplete_detach > 0)
return;
pipe = NULL;
// If we are waiting for pending messages to be sent, at this point // If we are waiting for pending messages to be sent, at this point
// we are sure that there will be no more messages and we can proceed // we are sure that there will be no more messages and we can proceed
// with termination safely. // with termination safely.
if (pending && !pipe && incomplete_pipes.size () == 0) if (pending)
proceed_with_term (); proceed_with_term ();
} }
void zmq::session_base_t::read_activated (pipe_t *pipe_) void zmq::session_base_t::read_activated (pipe_t *pipe_)
{ {
// Skip activating if we're detaching this pipe // Skip activating if we're detaching this pipe
if (incomplete_pipes.size () > 0 && pipe_ != pipe) if (incomplete_detach > 0 && pipe_ != pipe)
return; return;
zmq_assert (pipe == pipe_); zmq_assert (pipe == pipe_);
...@@ -263,7 +269,7 @@ void zmq::session_base_t::read_activated (pipe_t *pipe_) ...@@ -263,7 +269,7 @@ void zmq::session_base_t::read_activated (pipe_t *pipe_)
void zmq::session_base_t::write_activated (pipe_t *pipe_) void zmq::session_base_t::write_activated (pipe_t *pipe_)
{ {
// Skip activating if we're detaching this pipe // Skip activating if we're detaching this pipe
if (incomplete_pipes.size () > 0 && pipe_ != pipe) if (incomplete_detach > 0 && pipe_ != pipe)
return; return;
zmq_assert (pipe == pipe_); zmq_assert (pipe == pipe_);
...@@ -305,7 +311,7 @@ void zmq::session_base_t::process_attach (i_engine *engine_) ...@@ -305,7 +311,7 @@ void zmq::session_base_t::process_attach (i_engine *engine_)
zmq_assert (engine_ != NULL); zmq_assert (engine_ != NULL);
// Create the pipe if it does not exist yet. // Create the pipe if it does not exist yet.
if (!pipe && !is_terminating ()) { if ((!pipe || incomplete_detach > 0) && !is_terminating ()) {
object_t *parents [2] = {this, socket}; object_t *parents [2] = {this, socket};
pipe_t *pipes [2] = {NULL, NULL}; pipe_t *pipes [2] = {NULL, NULL};
int hwms [2] = {options.rcvhwm, options.sndhwm}; int hwms [2] = {options.rcvhwm, options.sndhwm};
...@@ -415,8 +421,7 @@ void zmq::session_base_t::detached () ...@@ -415,8 +421,7 @@ void zmq::session_base_t::detached ()
&& addr->protocol != "pgm" && addr->protocol != "epgm") { && addr->protocol != "pgm" && addr->protocol != "epgm") {
pipe->hiccup (); pipe->hiccup ();
pipe->terminate (false); pipe->terminate (false);
incomplete_pipes.insert (pipe); incomplete_detach ++;
pipe = NULL;
} }
reset (); reset ();
......
...@@ -104,8 +104,8 @@ namespace zmq ...@@ -104,8 +104,8 @@ namespace zmq
// Pipe connecting the session to its socket. // Pipe connecting the session to its socket.
zmq::pipe_t *pipe; zmq::pipe_t *pipe;
// This set is added to with pipes we are disconnecting, but haven't yet completed // This flag is set if we are disconnecting, but haven't yet completed
std::set<pipe_t *> incomplete_pipes; int incomplete_detach;
// This flag is true if the remainder of the message being processed // This flag is true if the remainder of the message being processed
// is still in the in pipe. // is still in the in pipe.
......
...@@ -33,6 +33,8 @@ static void *server (void *c) ...@@ -33,6 +33,8 @@ static void *server (void *c)
char buffer[16]; char buffer[16];
int rc, val; int rc, val;
shoulddie = *(long *)sd;
context = zmq_init (1); context = zmq_init (1);
assert (context); assert (context);
......
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