Commit f9d7eea6 authored by sigiesec's avatar sigiesec

Problem: code duplication

Solution: unified Windows & non-Windows code further
parent e7817ad3
...@@ -54,7 +54,6 @@ zmq::select_t::select_t (const zmq::ctx_t &ctx_) : ...@@ -54,7 +54,6 @@ zmq::select_t::select_t (const zmq::ctx_t &ctx_) :
current_family_entry_it (family_entries.end ()), current_family_entry_it (family_entries.end ()),
#else #else
maxfd (retired_fd), maxfd (retired_fd),
retired (false),
#endif #endif
stopping (false) stopping (false)
{ {
...@@ -79,12 +78,11 @@ zmq::select_t::handle_t zmq::select_t::add_fd (fd_t fd_, i_poll_events *events_) ...@@ -79,12 +78,11 @@ zmq::select_t::handle_t zmq::select_t::add_fd (fd_t fd_, i_poll_events *events_)
u_short family = get_fd_family (fd_); u_short family = get_fd_family (fd_);
wsa_assert (family != AF_UNSPEC); wsa_assert (family != AF_UNSPEC);
family_entry_t &family_entry = family_entries [family]; family_entry_t &family_entry = family_entries [family];
#endif
family_entry.fd_entries.push_back (fd_entry); family_entry.fd_entries.push_back (fd_entry);
FD_SET (fd_, &family_entry.fds_set.error); FD_SET (fd_, &family_entry.fds_set.error);
#else
fd_entries.push_back (fd_entry);
FD_SET (fd_, &fds_set.error);
#if !defined ZMQ_HAVE_WINDOWS
if (fd_ > maxfd) if (fd_ > maxfd)
maxfd = fd_; maxfd = fd_;
#endif #endif
...@@ -147,6 +145,7 @@ void zmq::select_t::trigger_events (const fd_entries_t &fd_entries_, ...@@ -147,6 +145,7 @@ void zmq::select_t::trigger_events (const fd_entries_t &fd_entries_,
} }
} }
#if defined ZMQ_HAVE_WINDOWS
bool zmq::select_t::try_remove_fd_entry ( bool zmq::select_t::try_remove_fd_entry (
family_entries_t::iterator family_entry_it, zmq::fd_t &handle_) family_entries_t::iterator family_entry_it, zmq::fd_t &handle_)
{ {
...@@ -170,6 +169,7 @@ bool zmq::select_t::try_remove_fd_entry ( ...@@ -170,6 +169,7 @@ bool zmq::select_t::try_remove_fd_entry (
family_entry.fds_set.remove_fd (handle_); family_entry.fds_set.remove_fd (handle_);
return true; return true;
} }
#endif
void zmq::select_t::rm_fd (handle_t handle_) void zmq::select_t::rm_fd (handle_t handle_)
{ {
...@@ -195,21 +195,21 @@ void zmq::select_t::rm_fd (handle_t handle_) ...@@ -195,21 +195,21 @@ void zmq::select_t::rm_fd (handle_t handle_)
} }
#else #else
fd_entries_t::iterator fd_entry_it = fd_entries_t::iterator fd_entry_it =
find_fd_entry_by_handle (fd_entries, handle_); find_fd_entry_by_handle (family_entry.fd_entries, handle_);
assert (fd_entry_it != fd_entries.end ()); assert (fd_entry_it != fd_entries.end ());
fd_entry_it->fd = retired_fd; fd_entry_it->fd = retired_fd;
fds_set.remove_fd (handle_); family_entry.fds_set.remove_fd (handle_);
if (handle_ == maxfd) { if (handle_ == maxfd) {
maxfd = retired_fd; maxfd = retired_fd;
for (fd_entry_it = fd_entries.begin (); for (fd_entry_it = family_entry.fd_entries.begin ();
fd_entry_it != fd_entries.end (); ++fd_entry_it) fd_entry_it != family_entry.fd_entries.end (); ++fd_entry_it)
if (fd_entry_it->fd > maxfd) if (fd_entry_it->fd > maxfd)
maxfd = fd_entry_it->fd; maxfd = fd_entry_it->fd;
} }
retired = true; family_entry.retired = true;
#endif #endif
adjust_load (-1); adjust_load (-1);
} }
...@@ -219,10 +219,9 @@ void zmq::select_t::set_pollin (handle_t handle_) ...@@ -219,10 +219,9 @@ void zmq::select_t::set_pollin (handle_t handle_)
#if defined ZMQ_HAVE_WINDOWS #if defined ZMQ_HAVE_WINDOWS
u_short family = get_fd_family (handle_); u_short family = get_fd_family (handle_);
wsa_assert (family != AF_UNSPEC); wsa_assert (family != AF_UNSPEC);
FD_SET (handle_, &family_entries [family].fds_set.read); family_entry_t &family_entry = family_entries [family];
#else
FD_SET (handle_, &fds_set.read);
#endif #endif
FD_SET (handle_, &family_entry.fds_set.read);
} }
void zmq::select_t::reset_pollin (handle_t handle_) void zmq::select_t::reset_pollin (handle_t handle_)
...@@ -230,10 +229,9 @@ void zmq::select_t::reset_pollin (handle_t handle_) ...@@ -230,10 +229,9 @@ void zmq::select_t::reset_pollin (handle_t handle_)
#if defined ZMQ_HAVE_WINDOWS #if defined ZMQ_HAVE_WINDOWS
u_short family = get_fd_family (handle_); u_short family = get_fd_family (handle_);
wsa_assert (family != AF_UNSPEC); wsa_assert (family != AF_UNSPEC);
FD_CLR (handle_, &family_entries [family].fds_set.read); family_entry_t &family_entry = family_entries [family];
#else
FD_CLR (handle_, &fds_set.read);
#endif #endif
FD_CLR (handle_, &family_entry.fds_set.read);
} }
void zmq::select_t::set_pollout (handle_t handle_) void zmq::select_t::set_pollout (handle_t handle_)
...@@ -241,10 +239,9 @@ void zmq::select_t::set_pollout (handle_t handle_) ...@@ -241,10 +239,9 @@ void zmq::select_t::set_pollout (handle_t handle_)
#if defined ZMQ_HAVE_WINDOWS #if defined ZMQ_HAVE_WINDOWS
u_short family = get_fd_family (handle_); u_short family = get_fd_family (handle_);
wsa_assert (family != AF_UNSPEC); wsa_assert (family != AF_UNSPEC);
FD_SET (handle_, &family_entries [family].fds_set.write); family_entry_t &family_entry = family_entries [family];
#else
FD_SET (handle_, &fds_set.write);
#endif #endif
FD_SET (handle_, &family_entry.fds_set.write);
} }
void zmq::select_t::reset_pollout (handle_t handle_) void zmq::select_t::reset_pollout (handle_t handle_)
...@@ -252,10 +249,9 @@ void zmq::select_t::reset_pollout (handle_t handle_) ...@@ -252,10 +249,9 @@ void zmq::select_t::reset_pollout (handle_t handle_)
#if defined ZMQ_HAVE_WINDOWS #if defined ZMQ_HAVE_WINDOWS
u_short family = get_fd_family (handle_); u_short family = get_fd_family (handle_);
wsa_assert (family != AF_UNSPEC); wsa_assert (family != AF_UNSPEC);
FD_CLR (handle_, &family_entries [family].fds_set.write); family_entry_t &family_entry = family_entries [family];
#else
FD_CLR (handle_, &fds_set.write);
#endif #endif
FD_CLR (handle_, &family_entry.fds_set.write);
} }
void zmq::select_t::start () void zmq::select_t::start ()
...@@ -364,54 +360,53 @@ void zmq::select_t::loop () ...@@ -364,54 +360,53 @@ void zmq::select_t::loop ()
++current_family_entry_it) { ++current_family_entry_it) {
family_entry_t &family_entry = current_family_entry_it->second; family_entry_t &family_entry = current_family_entry_it->second;
// select will fail when run with empty sets.
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 (use_wsa_events) { if (use_wsa_events) {
// There is no reason to wait again after WSAWaitForMultipleEvents. // There is no reason to wait again after WSAWaitForMultipleEvents.
// Simply collect what is ready. // Simply collect what is ready.
struct timeval tv_nodelay = {0, 0}; struct timeval tv_nodelay = {0, 0};
rc = select (0, &local_fds_set.read, &local_fds_set.write, select_family_entry (family_entry, 0, true, tv_nodelay);
&local_fds_set.error, &tv_nodelay); } else {
} else select_family_entry (family_entry, 0, timeout > 0, tv);
rc = select (0, &local_fds_set.read, &local_fds_set.write,
&local_fds_set.error, timeout > 0 ? &tv : NULL);
wsa_assert (rc != SOCKET_ERROR);
trigger_events (fd_entries, local_fds_set, rc);
if (family_entry.retired) {
family_entry.retired = false;
family_entry.fd_entries.erase (
std::remove_if (fd_entries.begin (), fd_entries.end (),
is_retired_fd),
family_entry.fd_entries.end ());
} }
} }
#else #else
fds_set_t local_fds_set = fds_set; select_family_entry (family_entry, maxfd, timeout > 0, tv);
rc = select (maxfd + 1, &local_fds_set.read, &local_fds_set.write, #endif
&local_fds_set.error, timeout ? &tv : NULL); }
}
if (rc == -1) { void zmq::select_t::select_family_entry (family_entry_t &family_entry_,
errno_assert (errno == EINTR); const int max_fd_,
continue; const bool use_timeout_,
} struct timeval &tv_)
{
// select will fail when run with empty sets.
fd_entries_t &fd_entries = family_entry_.fd_entries;
if (fd_entries.empty ())
return;
trigger_events (fd_entries, local_fds_set, rc); fds_set_t local_fds_set = family_entry_.fds_set;
int rc = select (max_fd_, &local_fds_set.read, &local_fds_set.write,
&local_fds_set.error, use_timeout_ ? &tv_ : NULL);
if (retired) { #if defined ZMQ_HAVE_WINDOWS
retired = false; wsa_assert (rc != SOCKET_ERROR);
fd_entries.erase (std::remove_if (fd_entries.begin (), #else
fd_entries.end (), is_retired_fd), if (rc == -1) {
fd_entries.end ()); errno_assert (errno == EINTR);
} return;
}
#endif #endif
trigger_events (fd_entries, local_fds_set, rc);
if (family_entry_.retired) {
family_entry_.retired = false;
family_entry_.fd_entries.erase (std::remove_if (fd_entries.begin (),
fd_entries.end (),
is_retired_fd),
family_entry_.fd_entries.end ());
} }
} }
......
...@@ -53,129 +53,130 @@ ...@@ -53,129 +53,130 @@
namespace zmq namespace zmq
{ {
struct i_poll_events;
struct i_poll_events; // Implements socket polling mechanism using POSIX.1-2001 select()
// function.
// Implements socket polling mechanism using POSIX.1-2001 select() class select_t : public poller_base_t
// function. {
public:
typedef fd_t handle_t;
class select_t : public poller_base_t select_t (const ctx_t &ctx_);
{ ~select_t ();
public:
typedef fd_t handle_t; // "poller" concept.
handle_t add_fd (fd_t fd_, zmq::i_poll_events *events_);
void rm_fd (handle_t handle_);
void set_pollin (handle_t handle_);
void reset_pollin (handle_t handle_);
void set_pollout (handle_t handle_);
void reset_pollout (handle_t handle_);
void start ();
void stop ();
select_t (const ctx_t &ctx_); static int max_fds ();
~select_t ();
// "poller" concept. private:
handle_t add_fd (fd_t fd_, zmq::i_poll_events *events_); // Main worker thread routine.
void rm_fd (handle_t handle_); static void worker_routine (void *arg_);
void set_pollin (handle_t handle_);
void reset_pollin (handle_t handle_);
void set_pollout (handle_t handle_);
void reset_pollout (handle_t handle_);
void start ();
void stop ();
static int max_fds (); // Main event loop.
void loop ();
private: // Reference to ZMQ context.
const ctx_t &ctx;
// Main worker thread routine. // Internal state.
static void worker_routine (void *arg_); struct fds_set_t
{
fds_set_t ();
fds_set_t (const fds_set_t &other_);
fds_set_t &operator= (const fds_set_t &other_);
// Convenience method to descriptor from all sets.
void remove_fd (const fd_t &fd_);
fd_set read;
fd_set write;
fd_set error;
};
// Main event loop. struct fd_entry_t
void loop (); {
fd_t fd;
zmq::i_poll_events *events;
};
typedef std::vector<fd_entry_t> fd_entries_t;
// Reference to ZMQ context. void trigger_events (const fd_entries_t &fd_entries_,
const ctx_t &ctx; const fds_set_t &local_fds_set_,
int event_count_);
// Internal state. struct family_entry_t
struct fds_set_t {
{ family_entry_t ();
fds_set_t ();
fds_set_t (const fds_set_t& other_);
fds_set_t& operator=(const fds_set_t& other_);
// Convenience method to descriptor from all sets.
void remove_fd (const fd_t& fd_);
fd_set read; fd_entries_t fd_entries;
fd_set write; fds_set_t fds_set;
fd_set error; bool retired;
}; };
struct fd_entry_t void select_family_entry (family_entry_t &family_entry_,
{ int max_fd_,
fd_t fd; bool use_timeout_,
zmq::i_poll_events* events; struct timeval &tv_);
};
typedef std::vector<fd_entry_t> fd_entries_t;
#if defined ZMQ_HAVE_WINDOWS #if defined ZMQ_HAVE_WINDOWS
struct family_entry_t typedef std::map<u_short, family_entry_t> family_entries_t;
{
family_entry_t ();
fd_entries_t fd_entries;
fds_set_t fds_set;
bool retired;
};
typedef std::map<u_short, family_entry_t> family_entries_t;
struct wsa_events_t
{
wsa_events_t ();
~wsa_events_t ();
// read, write, error and readwrite
WSAEVENT events [4];
};
#endif
void trigger_events(const fd_entries_t &fd_entries_, const fds_set_t &local_fds_set_, int event_count_); struct wsa_events_t
{
wsa_events_t ();
~wsa_events_t ();
#if defined ZMQ_HAVE_WINDOWS // read, write, error and readwrite
family_entries_t family_entries; WSAEVENT events [4];
// See loop for details. };
family_entries_t::iterator current_family_entry_it;
bool try_remove_fd_entry (family_entries_t::iterator family_entry_it, family_entries_t family_entries;
zmq::fd_t &handle_); // See loop for details.
#else family_entries_t::iterator current_family_entry_it;
fd_entries_t fd_entries;
fds_set_t fds_set;
fd_t maxfd;
bool retired;
#endif
#if defined ZMQ_HAVE_WINDOWS bool try_remove_fd_entry (family_entries_t::iterator family_entry_it,
static const size_t fd_family_cache_size = 8; zmq::fd_t &handle_);
std::pair<fd_t, u_short> fd_family_cache [fd_family_cache_size];
u_short get_fd_family (fd_t fd_); static const size_t fd_family_cache_size = 8;
std::pair<fd_t, u_short> fd_family_cache [fd_family_cache_size];
// Socket's family or AF_UNSPEC on error. u_short get_fd_family (fd_t fd_);
static u_short determine_fd_family (fd_t fd_);
// Socket's family or AF_UNSPEC on error.
static u_short determine_fd_family (fd_t fd_);
#else
// on non-Windows, we can treat all fds as one family
family_entry_t family_entry;
fd_t maxfd;
bool retired;
#endif #endif
// Checks if an fd_entry_t is retired.
static bool is_retired_fd (const fd_entry_t &entry);
static fd_entries_t::iterator // Checks if an fd_entry_t is retired.
find_fd_entry_by_handle (fd_entries_t &fd_entries, handle_t handle_); static bool is_retired_fd (const fd_entry_t &entry);
// If true, thread is shutting down. static fd_entries_t::iterator
bool stopping; find_fd_entry_by_handle (fd_entries_t &fd_entries, handle_t handle_);
// Handle of the physical thread doing the I/O work. // If true, thread is shutting down.
thread_t worker; bool stopping;
select_t (const select_t&); // Handle of the physical thread doing the I/O work.
const select_t &operator = (const select_t&); thread_t worker;
};
typedef select_t poller_t; select_t (const select_t &);
const select_t &operator= (const select_t &);
};
typedef select_t poller_t;
} }
#endif #endif
......
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