generic-library/libzmq
1
0
Fork 0
mirror of https://github.com/zeromq/libzmq.git synced 2024-12-12 10:33:52 +01:00
Code Issues Projects Releases Wiki Activity

problem:closed poller still associated with socket

Browse Source
This commit is contained in:
Doron Somech 2016-05-05 12:53:55 +03:00
parent e15da4b38c
commit a747f72450
5 changed files with 25 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
src/mailbox_safe.cpp
View File

@ -52,12 +52,12 @@ zmq::mailbox_safe_t::~mailbox_safe_t ()
sync->unlock ();
}
void zmq::mailbox_safe_t::add_signaler(signaler_t* signaler)
void zmq::mailbox_safe_t::add_signaler (signaler_t* signaler)
{
signalers.push_back(signaler);
}
void zmq::mailbox_safe_t::remove_signaler(signaler_t* signaler)
void zmq::mailbox_safe_t::remove_signaler (signaler_t* signaler)
{
std::vector<signaler_t*>::iterator it = signalers.begin();
@ -71,6 +71,11 @@ void zmq::mailbox_safe_t::remove_signaler(signaler_t* signaler)
signalers.erase(it);
}
void zmq::mailbox_safe_t::clear_signalers ()
{
signalers.clear ();
}
void zmq::mailbox_safe_t::send (const command_t &cmd_)
{
sync->lock ();

5
src/mailbox_safe.hpp
View File

@ -57,8 +57,9 @@ namespace zmq
int recv (command_t *cmd_, int timeout_);
// Add signaler to mailbox which will be called when a message is ready
void add_signaler(signaler_t* signaler);
void remove_signaler(signaler_t* signaler);
void add_signaler (signaler_t* signaler);
void remove_signaler (signaler_t* signaler);
void clear_signalers ();
#ifdef HAVE_FORK
// close the file descriptors in the signaller. This is used in a forked

7
src/socket_base.cpp
View File

@ -1281,6 +1281,12 @@ int zmq::socket_base_t::recv (msg_t *msg_, int flags_)
int zmq::socket_base_t::close ()
{
ENTER_MUTEX ();
// Remove all existing signalers for thread safe sockets
if (thread_safe)
((mailbox_safe_t*)mailbox)->clear_signalers();
// Mark the socket as dead
tag = 0xdeadbeef;
@ -1289,6 +1295,7 @@ int zmq::socket_base_t::close ()
// process.
send_reap (this);
EXIT_MUTEX ();
return 0;
}

15
src/socket_poller.cpp
View File

@ -186,19 +186,14 @@ int zmq::socket_poller_t::remove (socket_base_t *socket_)
return -1;
}
items.erase(it);
need_rebuild = true;
int thread_safe;
size_t thread_safe_size = sizeof(int);
if (socket_->getsockopt (ZMQ_THREAD_SAFE, &thread_safe, &thread_safe_size) == -1)
return -1;
if (thread_safe) {
if (socket_->remove_signaler (&signaler) == -1)
return -1;
}
items.erase (it);
need_rebuild = true;
if (socket_->getsockopt (ZMQ_THREAD_SAFE, &thread_safe, &thread_safe_size) == 0 && thread_safe)
socket_->remove_signaler (&signaler);
return 0;
}

6
tests/test_poller.cpp
View File

@ -143,9 +143,7 @@ int main (void)
assert (event.events == ZMQ_POLLOUT);
#endif
// Destory poller, sockets and ctx
rc = zmq_poller_destroy (&poller);
assert (rc == 0);
// Destory sockets, poller and ctx
rc = zmq_close (sink);
assert (rc == 0);
rc = zmq_close (vent);
@ -158,6 +156,8 @@ int main (void)
rc = zmq_close (client);
assert (rc == 0);
#endif
rc = zmq_poller_destroy(&poller);
assert(rc == 0);
rc = zmq_ctx_term (ctx);
assert (rc == 0);