diff options
author | Martin Sustrik <sustrik@250bpm.com> | 2010-08-06 17:49:37 +0200 |
---|---|---|
committer | Martin Sustrik <sustrik@250bpm.com> | 2010-08-25 15:39:20 +0200 |
commit | 05d908492dc382941fc633ad7082b5bd86e84e67 (patch) | |
tree | ae10e49766152e42521a6c100e622dc616998143 /src/fq.cpp | |
parent | b7e0fa972f45d21e45cacb93a1a92d38fdc11f40 (diff) |
WIP: Socket migration between threads, new zmq_close() semantics
Sockets may now be migrated between OS threads; sockets may not be used by
more than one thread at any time. To migrate a socket to another thread the
caller must ensure that a full memory barrier is called before using the
socket from the target thread.
The new zmq_close() semantics implement the behaviour discussed at:
http://lists.zeromq.org/pipermail/zeromq-dev/2010-July/004244.html
Specifically, zmq_close() is now deterministic and while it still returns
immediately, it does not discard any data that may still be queued for
sending. Further, zmq_term() will now block until all outstanding data has
been sent.
TODO: Many bugs have been introduced, needs testing. Further, SO_LINGER or
an equivalent mechanism (possibly a configurable timeout to zmq_term())
needs to be implemented.
Diffstat (limited to 'src/fq.cpp')
-rw-r--r-- | src/fq.cpp | 29 |
1 files changed, 19 insertions, 10 deletions
@@ -32,18 +32,19 @@ zmq::fq_t::fq_t () : zmq::fq_t::~fq_t () { - for (pipes_t::size_type i = 0; i != pipes.size (); i++) - pipes [i]->term (); + zmq_assert (pipes.empty ()); } void zmq::fq_t::attach (reader_t *pipe_) { + pipe_->set_event_sink (this); + pipes.push_back (pipe_); pipes.swap (active, pipes.size () - 1); active++; } -void zmq::fq_t::detach (reader_t *pipe_) +void zmq::fq_t::terminated (reader_t *pipe_) { zmq_assert (!more || pipes [current] != pipe_); @@ -57,16 +58,18 @@ void zmq::fq_t::detach (reader_t *pipe_) pipes.erase (pipe_); } -void zmq::fq_t::kill (reader_t *pipe_) +bool zmq::fq_t::has_pipes () { - // Move the pipe to the list of inactive pipes. - active--; - if (current == active) - current = 0; - pipes.swap (pipes.index (pipe_), active); + return !pipes.empty (); +} + +void zmq::fq_t::term_pipes () +{ + for (pipes_t::size_type i = 0; i != pipes.size (); i++) + pipes [i]->terminate (); } -void zmq::fq_t::revive (reader_t *pipe_) +void zmq::fq_t::activated (reader_t *pipe_) { // Move the pipe to the list of active pipes. pipes.swap (pipes.index (pipe_), active); @@ -98,6 +101,12 @@ int zmq::fq_t::recv (zmq_msg_t *msg_, int flags_) } return 0; } + else { + active--; + pipes.swap (current, active); + if (current == active) + current = 0; + } } // No message is available. Initialise the output parameter |