From e7817ad38d3972f9f522985d13c76ad1dc94e3e9 Mon Sep 17 00:00:00 2001 From: sigiesec Date: Thu, 26 Oct 2017 10:31:19 +0200 Subject: [PATCH] Problem: code duplication Solution: reduced code duplication by introducing local variables and new function trigger_events --- src/select.cpp | 250 ++++++++++++++++++++++++++----------------------- src/select.hpp | 4 +- 2 files changed, 137 insertions(+), 117 deletions(-) diff --git a/src/select.cpp b/src/select.cpp index 33eb9af5..ec1d2f88 100644 --- a/src/select.cpp +++ b/src/select.cpp @@ -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 ¤t_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, timeout > 0 ? &tv : NULL); + &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); 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 &entry = fd_family_cache[i]; - if (entry.first == fd_) - { + for (i = 0; i < fd_family_cache_size; ++i) { + const std::pair &entry = fd_family_cache [i]; + if (entry.first == fd_) { return entry.second; } if (entry.first == retired_fd) break; } - std::pair res = std::make_pair(fd_, determine_fd_family (fd_)); + std::pair 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) { } diff --git a/src/select.hpp b/src/select.hpp index bad2cd51..1bad80d4 100644 --- a/src/select.hpp +++ b/src/select.hpp @@ -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.