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

Code cleanup

Browse Source
This commit is contained in:
Martin Hurton 2014-07-09 13:49:40 +02:00
parent bf74c0cfb1
commit 816299f11c
6 changed files with 69 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
src/ctx.cpp
View File

@ -44,7 +44,7 @@ int clipped_maxsocket(int max_requested)
if (max_requested >= zmq::poller_t::max_fds () && zmq::poller_t::max_fds () != -1) if (max_requested >= zmq::poller_t::max_fds () && zmq::poller_t::max_fds () != -1)
// -1 because we need room for the reaper mailbox. // -1 because we need room for the reaper mailbox.
max_requested = zmq::poller_t::max_fds () - 1; max_requested = zmq::poller_t::max_fds () - 1;
return max_requested; return max_requested;
} }
@ -115,9 +115,9 @@ int zmq::ctx_t::terminate ()
// we are a forked child process. Close all file descriptors // we are a forked child process. Close all file descriptors
// inherited from the parent. // inherited from the parent.
for (sockets_t::size_type i = 0; i != sockets.size (); i++) for (sockets_t::size_type i = 0; i != sockets.size (); i++)
sockets[i]->get_mailbox()->forked(); sockets [i]->get_mailbox ()->forked ();
term_mailbox.forked(); term_mailbox.forked ();
} }
#endif #endif
@ -374,8 +374,8 @@ int zmq::ctx_t::register_endpoint (const char *addr_,
{ {
endpoints_sync.lock (); endpoints_sync.lock ();
bool inserted = endpoints.insert (endpoints_t::value_type ( const bool inserted = endpoints.insert (
std::string (addr_), endpoint_)).second; endpoints_t::value_type (std::string (addr_), endpoint_)).second;
endpoints_sync.unlock (); endpoints_sync.unlock ();
@ -420,6 +420,7 @@ void zmq::ctx_t::unregister_endpoints (socket_base_t *socket_)
} }
++it; ++it;
} }
endpoints_sync.unlock (); endpoints_sync.unlock ();
} }
@ -484,7 +485,7 @@ void zmq::ctx_t::connect_inproc_sockets (zmq::socket_base_t *bind_socket_,
options_t& bind_options, const pending_connection_t &pending_connection_, side side_) options_t& bind_options, const pending_connection_t &pending_connection_, side side_)
{ {
bind_socket_->inc_seqnum(); bind_socket_->inc_seqnum();
pending_connection_.bind_pipe->set_tid(bind_socket_->get_tid()); pending_connection_.bind_pipe->set_tid (bind_socket_->get_tid ());
if (!bind_options.recv_identity) { if (!bind_options.recv_identity) {
msg_t msg; msg_t msg;

6
src/mailbox.cpp
View File

@ -25,7 +25,7 @@ zmq::mailbox_t::mailbox_t ()
// Get the pipe into passive state. That way, if the users starts by // Get the pipe into passive state. That way, if the users starts by
// polling on the associated file descriptor it will get woken up when // polling on the associated file descriptor it will get woken up when
// new command is posted. // new command is posted.
bool ok = cpipe.read (NULL); const bool ok = cpipe.read (NULL);
zmq_assert (!ok); zmq_assert (!ok);
active = false; active = false;
} }
@ -40,7 +40,7 @@ zmq::mailbox_t::~mailbox_t ()
sync.unlock (); sync.unlock ();
} }
zmq::fd_t zmq::mailbox_t::get_fd () zmq::fd_t zmq::mailbox_t::get_fd () const
{ {
return signaler.get_fd (); return signaler.get_fd ();
} }
@ -49,7 +49,7 @@ void zmq::mailbox_t::send (const command_t &cmd_)
{ {
sync.lock (); sync.lock ();
cpipe.write (cmd_, false); cpipe.write (cmd_, false);
bool ok = cpipe.flush (); const bool ok = cpipe.flush ();
sync.unlock (); sync.unlock ();
if (!ok) if (!ok)
signaler.send (); signaler.send ();

6
src/mailbox.hpp
View File

@ -40,15 +40,15 @@ namespace zmq
mailbox_t (); mailbox_t ();
~mailbox_t (); ~mailbox_t ();
fd_t get_fd (); fd_t get_fd () const;
void send (const command_t &cmd_); void send (const command_t &cmd_);
int recv (command_t *cmd_, int timeout_); int recv (command_t *cmd_, int timeout_);
#ifdef HAVE_FORK #ifdef HAVE_FORK
// close the file descriptors in the signaller. This is used in a forked // close the file descriptors in the signaller. This is used in a forked
// child process to close the file descriptors so that they do not interfere // child process to close the file descriptors so that they do not interfere
// with the context in the parent process. // with the context in the parent process.
void forked() { signaler.forked(); } void forked () { signaler.forked (); }
#endif #endif
private: private:

72
src/signaler.cpp
View File

@ -70,7 +70,7 @@ zmq::signaler_t::signaler_t ()
unblock_socket (r); unblock_socket (r);
} }
#ifdef HAVE_FORK #ifdef HAVE_FORK
pid = getpid(); pid = getpid ();
#endif #endif
} }
@ -80,9 +80,9 @@ zmq::signaler_t::~signaler_t ()
int rc = close (r); int rc = close (r);
errno_assert (rc == 0); errno_assert (rc == 0);
#elif defined ZMQ_HAVE_WINDOWS #elif defined ZMQ_HAVE_WINDOWS
struct linger so_linger = { 1, 0 }; const struct linger so_linger = { 1, 0 };
int rc = setsockopt (w, SOL_SOCKET, SO_LINGER, int rc = setsockopt (w, SOL_SOCKET, SO_LINGER,
(char *)&so_linger, sizeof (so_linger)); (const char *) &so_linger, sizeof so_linger);
wsa_assert (rc != SOCKET_ERROR); wsa_assert (rc != SOCKET_ERROR);
rc = closesocket (w); rc = closesocket (w);
wsa_assert (rc != SOCKET_ERROR); wsa_assert (rc != SOCKET_ERROR);
@ -96,15 +96,15 @@ zmq::signaler_t::~signaler_t ()
#endif #endif
} }
zmq::fd_t zmq::signaler_t::get_fd () zmq::fd_t zmq::signaler_t::get_fd () const
{ {
return r; return r;
} }
void zmq::signaler_t::send () void zmq::signaler_t::send ()
{ {
#if defined(HAVE_FORK) #if defined HAVE_FORK
if (unlikely(pid != getpid())) { if (unlikely (pid != getpid ())) {
//printf("Child process %d signaler_t::send returning without sending #1\n", getpid()); //printf("Child process %d signaler_t::send returning without sending #1\n", getpid());
return; // do not send anything in forked child context return; // do not send anything in forked child context
} }
@ -125,13 +125,13 @@ void zmq::signaler_t::send ()
if (unlikely (nbytes == -1 && errno == EINTR)) if (unlikely (nbytes == -1 && errno == EINTR))
continue; continue;
#if defined(HAVE_FORK) #if defined(HAVE_FORK)
if (unlikely(pid != getpid())) { if (unlikely (pid != getpid ())) {
//printf("Child process %d signaler_t::send returning without sending #2\n", getpid()); //printf("Child process %d signaler_t::send returning without sending #2\n", getpid());
errno = EINTR; errno = EINTR;
break; break;
} }
#endif #endif
zmq_assert (nbytes == sizeof (dummy)); zmq_assert (nbytes == sizeof dummy);
break; break;
} }
#endif #endif
@ -140,8 +140,7 @@ void zmq::signaler_t::send ()
int zmq::signaler_t::wait (int timeout_) int zmq::signaler_t::wait (int timeout_)
{ {
#ifdef HAVE_FORK #ifdef HAVE_FORK
if (unlikely(pid != getpid())) if (unlikely (pid != getpid ())) {
{
// we have forked and the file descriptor is closed. Emulate an interupt // we have forked and the file descriptor is closed. Emulate an interupt
// response. // response.
//printf("Child process %d signaler_t::wait returning simulating interrupt #1\n", getpid()); //printf("Child process %d signaler_t::wait returning simulating interrupt #1\n", getpid());
@ -151,7 +150,6 @@ int zmq::signaler_t::wait (int timeout_)
#endif #endif
#ifdef ZMQ_POLL_BASED_ON_POLL #ifdef ZMQ_POLL_BASED_ON_POLL
struct pollfd pfd; struct pollfd pfd;
pfd.fd = r; pfd.fd = r;
pfd.events = POLLIN; pfd.events = POLLIN;
@ -166,7 +164,8 @@ int zmq::signaler_t::wait (int timeout_)
return -1; return -1;
} }
#ifdef HAVE_FORK #ifdef HAVE_FORK
if (unlikely(pid != getpid())) { else
if (unlikely (pid != getpid ())) {
// we have forked and the file descriptor is closed. Emulate an interupt // we have forked and the file descriptor is closed. Emulate an interupt
// response. // response.
//printf("Child process %d signaler_t::wait returning simulating interrupt #2\n", getpid()); //printf("Child process %d signaler_t::wait returning simulating interrupt #2\n", getpid());
@ -245,7 +244,7 @@ void zmq::signaler_t::recv ()
} }
#ifdef HAVE_FORK #ifdef HAVE_FORK
void zmq::signaler_t::forked() void zmq::signaler_t::forked ()
{ {
// Close file descriptors created in the parent and create new pair // Close file descriptors created in the parent and create new pair
close (r); close (r);
@ -274,13 +273,13 @@ int zmq::signaler_t::make_fdpair (fd_t *r_, fd_t *w_)
// Windows CE does not manage security attributes // Windows CE does not manage security attributes
SECURITY_DESCRIPTOR sd; SECURITY_DESCRIPTOR sd;
SECURITY_ATTRIBUTES sa; SECURITY_ATTRIBUTES sa;
memset (&sd, 0, sizeof (sd)); memset (&sd, 0, sizeof sd);
memset (&sa, 0, sizeof (sa)); memset (&sa, 0, sizeof sa);
InitializeSecurityDescriptor(&sd, SECURITY_DESCRIPTOR_REVISION); InitializeSecurityDescriptor (&sd, SECURITY_DESCRIPTOR_REVISION);
SetSecurityDescriptorDacl(&sd, TRUE, 0, FALSE); SetSecurityDescriptorDacl (&sd, TRUE, 0, FALSE);
sa.nLength = sizeof(SECURITY_ATTRIBUTES); sa.nLength = sizeof (SECURITY_ATTRIBUTES);
sa.lpSecurityDescriptor = &sd; sa.lpSecurityDescriptor = &sd;
# endif # endif
@ -310,9 +309,10 @@ int zmq::signaler_t::make_fdpair (fd_t *r_, fd_t *w_)
win_assert (sync != NULL); win_assert (sync != NULL);
} }
else if (signaler_port != 0) { else
wchar_t mutex_name[MAX_PATH]; if (signaler_port != 0) {
swprintf(mutex_name, MAX_PATH, L"Global\\zmq-signaler-port-%d", signaler_port); wchar_t mutex_name [MAX_PATH];
swprintf (mutex_name, MAX_PATH, L"Global\\zmq-signaler-port-%d", signaler_port);
# if !defined _WIN32_WCE # if !defined _WIN32_WCE
sync = CreateMutexW (&sa, FALSE, mutex_name); sync = CreateMutexW (&sa, FALSE, mutex_name);
@ -338,16 +338,16 @@ int zmq::signaler_t::make_fdpair (fd_t *r_, fd_t *w_)
// Set SO_REUSEADDR and TCP_NODELAY on listening socket. // Set SO_REUSEADDR and TCP_NODELAY on listening socket.
BOOL so_reuseaddr = 1; BOOL so_reuseaddr = 1;
int rc = setsockopt (listener, SOL_SOCKET, SO_REUSEADDR, int rc = setsockopt (listener, SOL_SOCKET, SO_REUSEADDR,
(char *)&so_reuseaddr, sizeof (so_reuseaddr)); (char *)&so_reuseaddr, sizeof so_reuseaddr);
wsa_assert (rc != SOCKET_ERROR); wsa_assert (rc != SOCKET_ERROR);
BOOL tcp_nodelay = 1; BOOL tcp_nodelay = 1;
rc = setsockopt (listener, IPPROTO_TCP, TCP_NODELAY, rc = setsockopt (listener, IPPROTO_TCP, TCP_NODELAY,
(char *)&tcp_nodelay, sizeof (tcp_nodelay)); (char *)&tcp_nodelay, sizeof tcp_nodelay);
wsa_assert (rc != SOCKET_ERROR); wsa_assert (rc != SOCKET_ERROR);
// Init sockaddr to signaler port. // Init sockaddr to signaler port.
struct sockaddr_in addr; struct sockaddr_in addr;
memset (&addr, 0, sizeof (addr)); memset (&addr, 0, sizeof addr);
addr.sin_family = AF_INET; addr.sin_family = AF_INET;
addr.sin_addr.s_addr = htonl (INADDR_LOOPBACK); addr.sin_addr.s_addr = htonl (INADDR_LOOPBACK);
addr.sin_port = htons (signaler_port); addr.sin_port = htons (signaler_port);
@ -358,7 +358,7 @@ int zmq::signaler_t::make_fdpair (fd_t *r_, fd_t *w_)
// Set TCP_NODELAY on writer socket. // Set TCP_NODELAY on writer socket.
rc = setsockopt (*w_, IPPROTO_TCP, TCP_NODELAY, rc = setsockopt (*w_, IPPROTO_TCP, TCP_NODELAY,
(char *)&tcp_nodelay, sizeof (tcp_nodelay)); (char *) &tcp_nodelay, sizeof tcp_nodelay);
wsa_assert (rc != SOCKET_ERROR); wsa_assert (rc != SOCKET_ERROR);
if (sync != NULL) { if (sync != NULL) {
@ -368,11 +368,11 @@ int zmq::signaler_t::make_fdpair (fd_t *r_, fd_t *w_)
} }
// Bind listening socket to signaler port. // Bind listening socket to signaler port.
rc = bind (listener, (const struct sockaddr*) &addr, sizeof (addr)); rc = bind (listener, (const struct sockaddr*) &addr, sizeof addr);
if (rc != SOCKET_ERROR && signaler_port == 0) { if (rc != SOCKET_ERROR && signaler_port == 0) {
// Retrieve ephemeral port number // Retrieve ephemeral port number
int addrlen = sizeof (addr); int addrlen = sizeof addr;
rc = getsockname (listener, (struct sockaddr*) &addr, &addrlen); rc = getsockname (listener, (struct sockaddr*) &addr, &addrlen);
} }
@ -382,7 +382,7 @@ int zmq::signaler_t::make_fdpair (fd_t *r_, fd_t *w_)
// Connect writer to the listener. // Connect writer to the listener.
if (rc != SOCKET_ERROR) if (rc != SOCKET_ERROR)
rc = connect (*w_, (struct sockaddr*) &addr, sizeof (addr)); rc = connect (*w_, (struct sockaddr*) &addr, sizeof addr);
// Accept connection from writer. // Accept connection from writer.
if (rc != SOCKET_ERROR) if (rc != SOCKET_ERROR)
@ -439,7 +439,7 @@ int zmq::signaler_t::make_fdpair (fd_t *r_, fd_t *w_)
// The bug will be fixed in V5.6 ECO4 and beyond. In the meantime, we'll // The bug will be fixed in V5.6 ECO4 and beyond. In the meantime, we'll
// create the socket pair manually. // create the socket pair manually.
struct sockaddr_in lcladdr; struct sockaddr_in lcladdr;
memset (&lcladdr, 0, sizeof (lcladdr)); memset (&lcladdr, 0, sizeof lcladdr);
lcladdr.sin_family = AF_INET; lcladdr.sin_family = AF_INET;
lcladdr.sin_addr.s_addr = htonl (INADDR_LOOPBACK); lcladdr.sin_addr.s_addr = htonl (INADDR_LOOPBACK);
lcladdr.sin_port = 0; lcladdr.sin_port = 0;
@ -448,16 +448,16 @@ int zmq::signaler_t::make_fdpair (fd_t *r_, fd_t *w_)
errno_assert (listener != -1); errno_assert (listener != -1);
int on = 1; int on = 1;
int rc = setsockopt (listener, IPPROTO_TCP, TCP_NODELAY, &on, sizeof (on)); int rc = setsockopt (listener, IPPROTO_TCP, TCP_NODELAY, &on, sizeof on);
errno_assert (rc != -1); errno_assert (rc != -1);
rc = setsockopt (listener, IPPROTO_TCP, TCP_NODELACK, &on, sizeof (on)); rc = setsockopt (listener, IPPROTO_TCP, TCP_NODELACK, &on, sizeof on);
errno_assert (rc != -1); errno_assert (rc != -1);
rc = bind (listener, (struct sockaddr*) &lcladdr, sizeof (lcladdr)); rc = bind (listener, (struct sockaddr*) &lcladdr, sizeof lcladdr);
errno_assert (rc != -1); errno_assert (rc != -1);
socklen_t lcladdr_len = sizeof (lcladdr); socklen_t lcladdr_len = sizeof lcladdr;
rc = getsockname (listener, (struct sockaddr*) &lcladdr, &lcladdr_len); rc = getsockname (listener, (struct sockaddr*) &lcladdr, &lcladdr_len);
errno_assert (rc != -1); errno_assert (rc != -1);
@ -468,13 +468,13 @@ int zmq::signaler_t::make_fdpair (fd_t *r_, fd_t *w_)
*w_ = open_socket (AF_INET, SOCK_STREAM, 0); *w_ = open_socket (AF_INET, SOCK_STREAM, 0);
errno_assert (*w_ != -1); errno_assert (*w_ != -1);
rc = setsockopt (*w_, IPPROTO_TCP, TCP_NODELAY, &on, sizeof (on)); rc = setsockopt (*w_, IPPROTO_TCP, TCP_NODELAY, &on, sizeof on);
errno_assert (rc != -1); errno_assert (rc != -1);
rc = setsockopt (*w_, IPPROTO_TCP, TCP_NODELACK, &on, sizeof (on)); rc = setsockopt (*w_, IPPROTO_TCP, TCP_NODELACK, &on, sizeof on);
errno_assert (rc != -1); errno_assert (rc != -1);
rc = connect (*w_, (struct sockaddr*) &lcladdr, sizeof (lcladdr)); rc = connect (*w_, (struct sockaddr*) &lcladdr, sizeof lcladdr);
errno_assert (rc != -1); errno_assert (rc != -1);
*r_ = accept (listener, NULL, NULL); *r_ = accept (listener, NULL, NULL);

6
src/signaler.hpp
View File

@ -41,7 +41,7 @@ namespace zmq
signaler_t (); signaler_t ();
~signaler_t (); ~signaler_t ();
fd_t get_fd (); fd_t get_fd () const;
void send (); void send ();
int wait (int timeout_); int wait (int timeout_);
void recv (); void recv ();
@ -49,7 +49,7 @@ namespace zmq
#ifdef HAVE_FORK #ifdef HAVE_FORK
// close the file descriptors in a forked child process so that they // close the file descriptors in a forked child process so that they
// do not interfere with the context in the parent process. // do not interfere with the context in the parent process.
void forked(); void forked ();
#endif #endif
private: private:
@ -72,7 +72,7 @@ namespace zmq
pid_t pid; pid_t pid;
// idempotent close of file descriptors that is safe to use by destructor // idempotent close of file descriptors that is safe to use by destructor
// and forked(). // and forked().
void close_internal(); void close_internal ();
#endif #endif
}; };
} }

36
src/socket_base.cpp
View File

@ -178,7 +178,7 @@ int zmq::socket_base_t::parse_uri (const char *uri_,
} }
protocol_ = uri.substr (0, pos); protocol_ = uri.substr (0, pos);
address_ = uri.substr (pos + 3); address_ = uri.substr (pos + 3);
if (protocol_.empty () || address_.empty ()) { if (protocol_.empty () || address_.empty ()) {
errno = EINVAL; errno = EINVAL;
return -1; return -1;
@ -207,7 +207,7 @@ int zmq::socket_base_t::check_protocol (const std::string &protocol_)
return -1; return -1;
} }
#endif #endif
#if !defined ZMQ_HAVE_NORM #if !defined ZMQ_HAVE_NORM
if (protocol_ == "norm") { if (protocol_ == "norm") {
errno = EPROTONOSUPPORT; errno = EPROTONOSUPPORT;
@ -365,9 +365,9 @@ int zmq::socket_base_t::bind (const char *addr_)
if (protocol == "inproc") { if (protocol == "inproc") {
const endpoint_t endpoint = { this, options }; const endpoint_t endpoint = { this, options };
int rc = register_endpoint (addr_, endpoint); const int rc = register_endpoint (addr_, endpoint);
if (rc == 0) { if (rc == 0) {
connect_pending(addr_, this); connect_pending (addr_, this);
last_endpoint.assign (addr_); last_endpoint.assign (addr_);
} }
return rc; return rc;
@ -485,7 +485,8 @@ int zmq::socket_base_t::connect (const char *addr_)
int rcvhwm = 0; int rcvhwm = 0;
if (peer.socket == NULL) if (peer.socket == NULL)
rcvhwm = options.rcvhwm; rcvhwm = options.rcvhwm;
else if (options.rcvhwm != 0 && peer.options.sndhwm != 0) else
if (options.rcvhwm != 0 && peer.options.sndhwm != 0)
rcvhwm = options.rcvhwm + peer.options.sndhwm; rcvhwm = options.rcvhwm + peer.options.sndhwm;
// Create a bi-directional pipe to connect the peers. // Create a bi-directional pipe to connect the peers.
@ -524,11 +525,9 @@ int zmq::socket_base_t::connect (const char *addr_)
const endpoint_t endpoint = {this, options}; const endpoint_t endpoint = {this, options};
pend_connection (std::string (addr_), endpoint, new_pipes); pend_connection (std::string (addr_), endpoint, new_pipes);
} }
else else {
{
// If required, send the identity of the local socket to the peer. // If required, send the identity of the local socket to the peer.
if (peer.options.recv_identity) { if (peer.options.recv_identity) {
msg_t id; msg_t id;
rc = id.init_size (options.identity_size); rc = id.init_size (options.identity_size);
errno_assert (rc == 0); errno_assert (rc == 0);
@ -561,7 +560,7 @@ int zmq::socket_base_t::connect (const char *addr_)
last_endpoint.assign (addr_); last_endpoint.assign (addr_);
// remember inproc connections for disconnect // remember inproc connections for disconnect
inprocs.insert (inprocs_t::value_type (std::string (addr_), new_pipes[0])); inprocs.insert (inprocs_t::value_type (std::string (addr_), new_pipes [0]));
return 0; return 0;
} }
@ -569,7 +568,7 @@ int zmq::socket_base_t::connect (const char *addr_)
options.type == ZMQ_SUB || options.type == ZMQ_SUB ||
options.type == ZMQ_REQ); options.type == ZMQ_REQ);
if (unlikely (is_single_connect)) { if (unlikely (is_single_connect)) {
endpoints_t::iterator it = endpoints.find (addr_); const endpoints_t::iterator it = endpoints.find (addr_);
if (it != endpoints.end ()) { if (it != endpoints.end ()) {
// There is no valid use for multiple connects for SUB-PUB nor // There is no valid use for multiple connects for SUB-PUB nor
// DEALER-ROUTER nor REQ-REP. Multiple connects produces // DEALER-ROUTER nor REQ-REP. Multiple connects produces
@ -712,7 +711,7 @@ void zmq::socket_base_t::add_endpoint (const char *addr_, own_t *endpoint_, pipe
{ {
// Activate the session. Make it a child of this socket. // Activate the session. Make it a child of this socket.
launch_child (endpoint_); launch_child (endpoint_);
endpoints.insert (endpoints_t::value_type (std::string (addr_), endpoint_pipe_t(endpoint_, pipe))); endpoints.insert (endpoints_t::value_type (std::string (addr_), endpoint_pipe_t (endpoint_, pipe)));
} }
int zmq::socket_base_t::term_endpoint (const char *addr_) int zmq::socket_base_t::term_endpoint (const char *addr_)
@ -750,9 +749,9 @@ int zmq::socket_base_t::term_endpoint (const char *addr_)
errno = ENOENT; errno = ENOENT;
return -1; return -1;
} }
for (inprocs_t::iterator it = range.first; it != range.second; ++it) for (inprocs_t::iterator it = range.first; it != range.second; ++it)
it->second->terminate(true); it->second->terminate (true);
inprocs.erase (range.first, range.second); inprocs.erase (range.first, range.second);
return 0; return 0;
} }
@ -767,7 +766,7 @@ int zmq::socket_base_t::term_endpoint (const char *addr_)
for (endpoints_t::iterator it = range.first; it != range.second; ++it) { for (endpoints_t::iterator it = range.first; it != range.second; ++it) {
// If we have an associated pipe, terminate it. // If we have an associated pipe, terminate it.
if (it->second.second != NULL) if (it->second.second != NULL)
it->second.second->terminate(false); it->second.second->terminate (false);
term_child (it->second.first); term_child (it->second.first);
} }
endpoints.erase (range.first, range.second); endpoints.erase (range.first, range.second);
@ -983,7 +982,7 @@ int zmq::socket_base_t::process_commands (int timeout_, bool throttle_)
// commands recently, so that we can throttle the new commands. // commands recently, so that we can throttle the new commands.
// Get the CPU's tick counter. If 0, the counter is not available. // Get the CPU's tick counter. If 0, the counter is not available.
uint64_t tsc = zmq::clock_t::rdtsc (); const uint64_t tsc = zmq::clock_t::rdtsc ();
// Optimised version of command processing - it doesn't have to check // Optimised version of command processing - it doesn't have to check
// for incoming commands each time. It does so only if certain time // for incoming commands each time. It does so only if certain time
@ -1176,12 +1175,11 @@ void zmq::socket_base_t::pipe_terminated (pipe_t *pipe_)
xpipe_terminated (pipe_); xpipe_terminated (pipe_);
// Remove pipe from inproc pipes // Remove pipe from inproc pipes
for (inprocs_t::iterator it = inprocs.begin(); it != inprocs.end(); ++it) { for (inprocs_t::iterator it = inprocs.begin (); it != inprocs.end (); ++it)
if (it->second == pipe_) { if (it->second == pipe_) {
inprocs.erase(it); inprocs.erase (it);
break; break;
} }
}
// Remove the pipe from the list of attached pipes and confirm its // Remove the pipe from the list of attached pipes and confirm its
// termination if we are already shutting down. // termination if we are already shutting down.
@ -1195,7 +1193,7 @@ void zmq::socket_base_t::extract_flags (msg_t *msg_)
// Test whether IDENTITY flag is valid for this socket type. // Test whether IDENTITY flag is valid for this socket type.
if (unlikely (msg_->flags () & msg_t::identity)) if (unlikely (msg_->flags () & msg_t::identity))
zmq_assert (options.recv_identity); zmq_assert (options.recv_identity);
// Remove MORE flag. // Remove MORE flag.
rcvmore = msg_->flags () & msg_t::more ? true : false; rcvmore = msg_->flags () & msg_t::more ? true : false;
} }