Skip to content

L
libzmq
Commit e7817ad3 authored by sigiesec's avatar sigiesec
Browse files
Options

Problem: code duplication 

Solution: reduced code duplication by introducing local variables and
new function trigger_events
parent 37914d1b
Hide whitespace changes
Inline Side-by-side
Showing with 137 additions and 117 deletions
src/select.cpp
View file @ e7817ad3
......@@ -60,7 +60,7 @@ zmq::select_t::select_t (const zmq::ctx_t &ctx_) :
{
#if defined ZMQ_HAVE_WINDOWS
for (size_t i = 0; i < fd_family_cache_size; ++i)
fd_family_cache[i] = std::make_pair(retired_fd, 0);
fd_family_cache [i] = std::make_pair (retired_fd, 0);
#endif
}
......@@ -78,7 +78,7 @@ zmq::select_t::handle_t zmq::select_t::add_fd (fd_t fd_, i_poll_events *events_)
#if defined ZMQ_HAVE_WINDOWS
u_short family = get_fd_family (fd_);
wsa_assert (family != AF_UNSPEC);
family_entry_t& family_entry = family_entries [family];
family_entry_t &family_entry = family_entries [family];
family_entry.fd_entries.push_back (fd_entry);
FD_SET (fd_, &family_entry.fds_set.error);
#else
......@@ -107,6 +107,46 @@ zmq::select_t::find_fd_entry_by_handle (fd_entries_t &fd_entries,
return fd_entry_it;
}
void zmq::select_t::trigger_events (const fd_entries_t &fd_entries_,
const fds_set_t &local_fds_set_,
int event_count_)
{
// Size is cached to avoid iteration through recently added descriptors.
for (fd_entries_t::size_type i = 0, size = fd_entries_.size ();
i < size && event_count_ > 0; ++i) {
const fd_entry_t &current_fd_entry = fd_entries_ [i];
if (is_retired_fd (current_fd_entry))
continue;
if (FD_ISSET (current_fd_entry.fd, &local_fds_set_.read)) {
current_fd_entry.events->in_event ();
--event_count_;
}
// TODO: can the is_retired_fd be true at this point? if it
// was retired before, we would already have continued, and I
// don't see where it might have been modified
// And if rc == 0, we can break instead of continuing
if (is_retired_fd (current_fd_entry) || event_count_ == 0)
continue;
if (FD_ISSET (current_fd_entry.fd, &local_fds_set_.write)) {
current_fd_entry.events->out_event ();
--event_count_;
}
// TODO: same as above
if (is_retired_fd (current_fd_entry) || event_count_ == 0)
continue;
if (FD_ISSET (current_fd_entry.fd, &local_fds_set_.error)) {
current_fd_entry.events->in_event ();
--event_count_;
}
}
}
bool zmq::select_t::try_remove_fd_entry (
family_entries_t::iterator family_entry_it, zmq::fd_t &handle_)
{
......@@ -163,8 +203,8 @@ void zmq::select_t::rm_fd (handle_t handle_)
if (handle_ == maxfd) {
maxfd = retired_fd;
for (fd_entry_it = fd_entries.begin (); fd_entry_it != fd_entries.end ();
++fd_entry_it)
for (fd_entry_it = fd_entries.begin ();
fd_entry_it != fd_entries.end (); ++fd_entry_it)
if (fd_entry_it->fd > maxfd)
maxfd = fd_entry_it->fd;
}
......@@ -240,9 +280,10 @@ void zmq::select_t::loop ()
int timeout = (int) execute_timers ();
#if defined ZMQ_HAVE_OSX
struct timeval tv = { (long) (timeout / 1000), timeout % 1000 * 1000 };
struct timeval tv = {(long) (timeout / 1000), timeout % 1000 * 1000};
#else
struct timeval tv = { (long) (timeout / 1000), (long) (timeout % 1000 * 1000) };
struct timeval tv = {(long) (timeout / 1000),
(long) (timeout % 1000 * 1000)};
#endif
int rc = 0;
......@@ -257,38 +298,51 @@ void zmq::select_t::loop ()
cannot be used alone, because it does not support more than 64 events
which is not enough.
To reduce unncessary overhead, WSA is only used when there are more
To reduce unnecessary overhead, WSA is only used when there are more
than one family. Moreover, AF_INET and AF_INET6 are considered the same
family because Windows seems to handle them properly.
See get_fd_family for details.
*/
// If there is just one family, there is no reason to use WSA events.
if (family_entries.size () > 1) {
const bool use_wsa_events = family_entries.size () > 1;
if (use_wsa_events) {
// TODO: I don't really understand why we are doing this. If any of
// the events was signaled, we will call select for each fd_family
// afterwards. The only benefit is if none of the events was
// signaled, then we continue early.
// IMHO, either WSAEventSelect/WSAWaitForMultipleEvents or select
// should be used, but not both
wsa_events_t wsa_events;
for (family_entries_t::iterator family_entry_it = family_entries.begin ();
family_entry_it != family_entries.end (); ++family_entry_it) {
family_entry_t& family_entry = family_entry_it->second;
for (family_entries_t::iterator family_entry_it =
family_entries.begin ();
family_entry_it != family_entries.end (); ++family_entry_it) {
family_entry_t &family_entry = family_entry_it->second;
for (fd_entries_t::iterator fd_entry_it = family_entry.fd_entries.begin ();
fd_entry_it != family_entry.fd_entries.end (); ++fd_entry_it) {
for (fd_entries_t::iterator fd_entry_it =
family_entry.fd_entries.begin ();
fd_entry_it != family_entry.fd_entries.end ();
++fd_entry_it) {
fd_t fd = fd_entry_it->fd;
// http://stackoverflow.com/q/35043420/188530
if (FD_ISSET (fd, &family_entry.fds_set.read) &&
FD_ISSET (fd, &family_entry.fds_set.write))
rc = WSAEventSelect (fd, wsa_events.events [3],
FD_READ | FD_ACCEPT | FD_CLOSE | FD_WRITE | FD_CONNECT | FD_OOB);
if (FD_ISSET (fd, &family_entry.fds_set.read)
&& FD_ISSET (fd, &family_entry.fds_set.write))
rc =
WSAEventSelect (fd, wsa_events.events [3],
FD_READ | FD_ACCEPT | FD_CLOSE
| FD_WRITE | FD_CONNECT | FD_OOB);
else if (FD_ISSET (fd, &family_entry.fds_set.read))
rc = WSAEventSelect (fd, wsa_events.events [0],
FD_READ | FD_ACCEPT | FD_CLOSE | FD_OOB);
FD_READ | FD_ACCEPT | FD_CLOSE
| FD_OOB);
else if (FD_ISSET (fd, &family_entry.fds_set.write))
rc = WSAEventSelect (fd, wsa_events.events [1],
FD_WRITE | FD_CONNECT | FD_OOB);
FD_WRITE | FD_CONNECT | FD_OOB);
else if (FD_ISSET (fd, &family_entry.fds_set.error))
rc = WSAEventSelect (fd, wsa_events.events [2],
FD_OOB);
rc = WSAEventSelect (fd, wsa_events.events [2], FD_OOB);
else
rc = 0;
......@@ -297,8 +351,8 @@ void zmq::select_t::loop ()
}
rc = WSAWaitForMultipleEvents (4, wsa_events.events, FALSE,
timeout ? timeout : INFINITE, FALSE);
wsa_assert (rc != (int)WSA_WAIT_FAILED);
timeout ? timeout : INFINITE, FALSE);
wsa_assert (rc != (int) WSA_WAIT_FAILED);
zmq_assert (rc != WSA_WAIT_IO_COMPLETION);
if (rc == WSA_WAIT_TIMEOUT)
......@@ -306,103 +360,56 @@ void zmq::select_t::loop ()
}
for (current_family_entry_it = family_entries.begin ();
current_family_entry_it != family_entries.end (); ++current_family_entry_it) {
family_entry_t& family_entry = current_family_entry_it->second;
current_family_entry_it != family_entries.end ();
++current_family_entry_it) {
family_entry_t &family_entry = current_family_entry_it->second;
// select will fail when run with empty sets.
if (family_entry.fd_entries.empty ())
fd_entries_t &fd_entries = family_entry.fd_entries;
if (fd_entries.empty())
continue;
fds_set_t local_fds_set = family_entry.fds_set;
if (family_entries.size () > 1) {
if (use_wsa_events) {
// There is no reason to wait again after WSAWaitForMultipleEvents.
// Simply collect what is ready.
struct timeval tv_nodelay = { 0, 0 };
rc = select (0, &local_fds_set.read, &local_fds_set.write, &local_fds_set.error,
&tv_nodelay);
}
else
struct timeval tv_nodelay = {0, 0};
rc = select (0, &local_fds_set.read, &local_fds_set.write,
&local_fds_set.error, &tv_nodelay);
} else
rc = select (0, &local_fds_set.read, &local_fds_set.write,
&local_fds_set.error, timeout > 0 ? &tv : NULL);
&local_fds_set.error, timeout > 0 ? &tv : NULL);
wsa_assert (rc != SOCKET_ERROR);
// Size is cached to avoid iteration through recently added descriptors.
for (fd_entries_t::size_type i = 0, size = family_entry.fd_entries.size (); i < size && rc > 0; ++i) {
if (family_entry.fd_entries[i].fd == retired_fd)
continue;
if (FD_ISSET(family_entry.fd_entries[i].fd, &local_fds_set.read)) {
family_entry.fd_entries[i].events->in_event();
--rc;
}
if (family_entry.fd_entries[i].fd == retired_fd || rc == 0)
continue;
if (FD_ISSET(family_entry.fd_entries[i].fd, &local_fds_set.write)) {
family_entry.fd_entries[i].events->out_event();
--rc;
}
if (family_entry.fd_entries[i].fd == retired_fd || rc == 0)
continue;
if (FD_ISSET(family_entry.fd_entries[i].fd, &local_fds_set.error)) {
family_entry.fd_entries[i].events->in_event();
--rc;
}
}
trigger_events (fd_entries, local_fds_set, rc);
if (family_entry.retired) {
family_entry.retired = false;
family_entry.fd_entries.erase (std::remove_if (family_entry.fd_entries.begin (),
family_entry.fd_entries.end (), is_retired_fd), family_entry.fd_entries.end ());
family_entry.fd_entries.erase (
std::remove_if (fd_entries.begin (), fd_entries.end (),
is_retired_fd),
family_entry.fd_entries.end ());
}
}
#else
fds_set_t local_fds_set = fds_set;
rc = select (maxfd + 1, &local_fds_set.read, &local_fds_set.write,
&local_fds_set.error, timeout ? &tv : NULL);
&local_fds_set.error, timeout ? &tv : NULL);
if (rc == -1) {
errno_assert (errno == EINTR);
continue;
}
// Size is cached to avoid iteration through just added descriptors.
for (fd_entries_t::size_type i = 0, size = fd_entries.size (); i < size && rc > 0; ++i) {
if (fd_entries [i].fd == retired_fd)
continue;
if (FD_ISSET (fd_entries [i].fd, &local_fds_set.read)) {
fd_entries [i].events->in_event ();
--rc;
}
if (fd_entries [i].fd == retired_fd || rc == 0)
continue;
if (FD_ISSET (fd_entries [i].fd, &local_fds_set.write)) {
fd_entries [i].events->out_event ();
--rc;
}
if (fd_entries [i].fd == retired_fd || rc == 0)
continue;
if (FD_ISSET (fd_entries [i].fd, &local_fds_set.error)) {
fd_entries [i].events->in_event ();
--rc;
}
}
trigger_events (fd_entries, local_fds_set, rc);
if (retired) {
retired = false;
fd_entries.erase (std::remove_if (fd_entries.begin (), fd_entries.end (),
is_retired_fd), fd_entries.end ());
fd_entries.erase (std::remove_if (fd_entries.begin (),
fd_entries.end (), is_retired_fd),
fd_entries.end ());
}
#endif
}
......@@ -410,7 +417,7 @@ void zmq::select_t::loop ()
void zmq::select_t::worker_routine (void *arg_)
{
((select_t*) arg_)->loop ();
((select_t *) arg_)->loop ();
}
zmq::select_t::fds_set_t::fds_set_t ()
......@@ -420,16 +427,22 @@ zmq::select_t::fds_set_t::fds_set_t ()
FD_ZERO (&error);
}
zmq::select_t::fds_set_t::fds_set_t (const fds_set_t& other_)
zmq::select_t::fds_set_t::fds_set_t (const fds_set_t &other_)
{
#if defined ZMQ_HAVE_WINDOWS
// On Windows we don't need to copy the whole fd_set.
// SOCKETS are continuous from the beginning of fd_array in fd_set.
// We just need to copy fd_count elements of fd_array.
// We gain huge memcpy() improvement if number of used SOCKETs is much lower than FD_SETSIZE.
memcpy (&read, &other_.read, (char *) (other_.read.fd_array + other_.read.fd_count ) - (char *) &other_.read );
memcpy (&write, &other_.write, (char *) (other_.write.fd_array + other_.write.fd_count) - (char *) &other_.write);
memcpy (&error, &other_.error, (char *) (other_.error.fd_array + other_.error.fd_count) - (char *) &other_.error);
memcpy (&read, &other_.read,
(char *) (other_.read.fd_array + other_.read.fd_count)
- (char *) &other_.read);
memcpy (&write, &other_.write,
(char *) (other_.write.fd_array + other_.write.fd_count)
- (char *) &other_.write);
memcpy (&error, &other_.error,
(char *) (other_.error.fd_array + other_.error.fd_count)
- (char *) &other_.error);
#else
memcpy (&read, &other_.read, sizeof other_.read);
memcpy (&write, &other_.write, sizeof other_.write);
......@@ -437,16 +450,23 @@ zmq::select_t::fds_set_t::fds_set_t (const fds_set_t& other_)
#endif
}
zmq::select_t::fds_set_t& zmq::select_t::fds_set_t::operator= (const fds_set_t& other_)
zmq::select_t::fds_set_t &zmq::select_t::fds_set_t::
operator= (const fds_set_t &other_)
{
#if defined ZMQ_HAVE_WINDOWS
// On Windows we don't need to copy the whole fd_set.
// SOCKETS are continuous from the beginning of fd_array in fd_set.
// We just need to copy fd_count elements of fd_array.
// We gain huge memcpy() improvement if number of used SOCKETs is much lower than FD_SETSIZE.
memcpy (&read, &other_.read, (char *) (other_.read.fd_array + other_.read.fd_count ) - (char *) &other_.read );
memcpy (&write, &other_.write, (char *) (other_.write.fd_array + other_.write.fd_count) - (char *) &other_.write);
memcpy (&error, &other_.error, (char *) (other_.error.fd_array + other_.error.fd_count) - (char *) &other_.error);
memcpy (&read, &other_.read,
(char *) (other_.read.fd_array + other_.read.fd_count)
- (char *) &other_.read);
memcpy (&write, &other_.write,
(char *) (other_.write.fd_array + other_.write.fd_count)
- (char *) &other_.write);
memcpy (&error, &other_.error,
(char *) (other_.error.fd_array + other_.error.fd_count)
- (char *) &other_.error);
#else
memcpy (&read, &other_.read, sizeof other_.read);
memcpy (&write, &other_.write, sizeof other_.write);
......@@ -455,7 +475,7 @@ zmq::select_t::fds_set_t& zmq::select_t::fds_set_t::operator= (const fds_set_t&
return *this;
}
void zmq::select_t::fds_set_t::remove_fd (const fd_t& fd_)
void zmq::select_t::fds_set_t::remove_fd (const fd_t &fd_)
{
FD_CLR (fd_, &read);
FD_CLR (fd_, &write);
......@@ -473,25 +493,23 @@ u_short zmq::select_t::get_fd_family (fd_t fd_)
// cache the results of determine_fd_family, as this is frequently called
// for the same sockets, and determine_fd_family is expensive
size_t i;
for (i = 0; i < fd_family_cache_size; ++i)
{
const std::pair<fd_t, u_short> &entry = fd_family_cache[i];
if (entry.first == fd_)
{
for (i = 0; i < fd_family_cache_size; ++i) {
const std::pair<fd_t, u_short> &entry = fd_family_cache [i];
if (entry.first == fd_) {
return entry.second;
}
if (entry.first == retired_fd)
break;
}
std::pair<fd_t, u_short> res = std::make_pair(fd_, determine_fd_family (fd_));
std::pair<fd_t, u_short> res =
std::make_pair (fd_, determine_fd_family (fd_));
if (i < fd_family_cache_size) {
fd_family_cache [i] = res;
}
else {
} else {
// just overwrite a random entry
// could be optimized by some LRU strategy
fd_family_cache [rand() % fd_family_cache_size] = res;
fd_family_cache [rand () % fd_family_cache_size] = res;
}
return res.second;
......@@ -500,19 +518,20 @@ u_short zmq::select_t::get_fd_family (fd_t fd_)
u_short zmq::select_t::determine_fd_family (fd_t fd_)
{
// Use sockaddr_storage instead of sockaddr to accommodate different structure sizes
sockaddr_storage addr = { 0 };
sockaddr_storage addr = {0};
int addr_size = sizeof addr;
int type;
int type_length = sizeof(int);
int type_length = sizeof (int);
int rc = getsockopt(fd_, SOL_SOCKET, SO_TYPE, (char*) &type, &type_length);
int rc =
getsockopt (fd_, SOL_SOCKET, SO_TYPE, (char *) &type, &type_length);
if (rc == 0) {
if (type == SOCK_DGRAM)
return AF_INET;
else {
rc = getsockname(fd_, (sockaddr *)&addr, &addr_size);
rc = getsockname (fd_, (sockaddr *) &addr, &addr_size);
// AF_INET and AF_INET6 can be mixed in select
// TODO: If proven otherwise, should simply return addr.sa_family
......@@ -524,8 +543,7 @@ u_short zmq::select_t::determine_fd_family (fd_t fd_)
return AF_UNSPEC;
}
zmq::select_t::family_entry_t::family_entry_t () :
retired (false)
zmq::select_t::family_entry_t::family_entry_t () : retired (false)
{
}
......
src/select.hpp
View file @ e7817ad3
......@@ -97,7 +97,7 @@ namespace zmq
fds_set_t ();
fds_set_t (const fds_set_t& other_);
fds_set_t& operator=(const fds_set_t& other_);
// Convinient method to descriptor from all sets.
// Convenience method to descriptor from all sets.
void remove_fd (const fd_t& fd_);
fd_set read;
......@@ -133,6 +133,8 @@ namespace zmq
};
#endif
void trigger_events(const fd_entries_t &fd_entries_, const fds_set_t &local_fds_set_, int event_count_);
#if defined ZMQ_HAVE_WINDOWS
family_entries_t family_entries;
// See loop for details.
......
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