common base for all pollers created; the only thing it handles at the moment is 'load'

13 files changed, 125 insertions, 86 deletions

diff --git a/src/Makefile.am b/src/Makefile.am
index d2cc174..89fc44a 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -89,6 +89,7 @@ libzmq_la_SOURCES = \
     platform.hpp \
     poll.hpp \
     poller.hpp \
+    poller_base.hpp \
     pair.hpp \
     pub.hpp \
     pull.hpp \
@@ -148,6 +149,7 @@ libzmq_la_SOURCES = \
     pgm_socket.cpp \
     pipe.cpp \
     poll.cpp \
+    poller_base.cpp \
     pull.cpp \
     push.cpp \
     pub.cpp \
diff --git a/src/devpoll.cpp b/src/devpoll.cpp
index 7054c2b..32aca50 100644
--- a/src/devpoll.cpp
+++ b/src/devpoll.cpp
@@ -56,10 +56,6 @@ zmq::devpoll_t::devpoll_t () :
 zmq::devpoll_t::~devpoll_t ()
 {
     worker.stop ();
-
-    //  Make sure there are no fds registered on shutdown.
-    zmq_assert (load.get () == 0);
-
     close (devpoll_fd);
 }
 
@@ -84,7 +80,7 @@ zmq::devpoll_t::handle_t zmq::devpoll_t::add_fd (fd_t fd_,
     pending_list.push_back (fd_);
 
     //  Increase the load metric of the thread.
-    load.add (1);
+    adjust_load (1);
 
     return fd_;
 }
@@ -97,7 +93,7 @@ void zmq::devpoll_t::rm_fd (handle_t handle_)
     fd_table [handle_].valid = false;
 
     //  Decrease the load metric of the thread.
-    load.sub (1);
+    adjust_load (-1);
 }
 
 void zmq::devpoll_t::set_pollin (handle_t handle_)
@@ -140,11 +136,6 @@ void zmq::devpoll_t::cancel_timer (i_poll_events *events_, int id_)
         timers.erase (it);
 }
 
-int zmq::devpoll_t::get_load ()
-{
-    return load.get ();
-}
-
 void zmq::devpoll_t::start ()
 {
     worker.start (worker_routine, this);
diff --git a/src/devpoll.hpp b/src/devpoll.hpp
index 00be385..ae16583 100644
--- a/src/devpoll.hpp
+++ b/src/devpoll.hpp
@@ -28,14 +28,14 @@
 
 #include "fd.hpp"
 #include "thread.hpp"
-#include "atomic_counter.hpp"
+#include "poller_base.hpp"
 
 namespace zmq
 {
 
     //  Implements socket polling mechanism using the "/dev/poll" interface.
 
-    class devpoll_t
+    class devpoll_t : public poller_base_t
     {
     public:
 
@@ -53,7 +53,6 @@ namespace zmq
         void reset_pollout (handle_t handle_);
         void add_timer (int timeout_, struct i_poll_events *events_, int id_);
         void cancel_timer (struct i_poll_events *events_, int id_);
-        int get_load ();
         void start ();
         void stop ();
 
@@ -94,10 +93,6 @@ namespace zmq
         //  Handle of the physical thread doing the I/O work.
         thread_t worker;
 
-        //  Load of the poller. Currently number of file descriptors
-        //  registered with the poller.
-        atomic_counter_t load;
-
         devpoll_t (const devpoll_t&);
         void operator = (const devpoll_t&);
     };
diff --git a/src/epoll.cpp b/src/epoll.cpp
index bbad639..584a13f 100644
--- a/src/epoll.cpp
+++ b/src/epoll.cpp
@@ -45,9 +45,6 @@ zmq::epoll_t::~epoll_t ()
     //  Wait till the worker thread exits.
     worker.stop ();
 
-    //  Make sure there are no fds registered on shutdown.
-    zmq_assert (load.get () == 0);
-
     close (epoll_fd);
     for (retired_t::iterator it = retired.begin (); it != retired.end (); it ++)
         delete *it;
@@ -71,7 +68,7 @@ zmq::epoll_t::handle_t zmq::epoll_t::add_fd (fd_t fd_, i_poll_events *events_)
     errno_assert (rc != -1);
 
     //  Increase the load metric of the thread.
-    load.add (1);
+    adjust_load (1);
 
     return pe;
 }
@@ -85,7 +82,7 @@ void zmq::epoll_t::rm_fd (handle_t handle_)
     retired.push_back (pe);
 
     //  Decrease the load metric of the thread.
-    load.sub (1);
+    adjust_load (-1);
 }
 
 void zmq::epoll_t::set_pollin (handle_t handle_)
@@ -133,11 +130,6 @@ void zmq::epoll_t::cancel_timer (i_poll_events *events_, int id_)
     timers.erase (it);
 }
 
-int zmq::epoll_t::get_load ()
-{
-    return load.get ();
-}
-
 void zmq::epoll_t::start ()
 {
     worker.start (worker_routine, this);
diff --git a/src/epoll.hpp b/src/epoll.hpp
index a68f055..a19fc0d 100644
--- a/src/epoll.hpp
+++ b/src/epoll.hpp
@@ -29,7 +29,7 @@
 
 #include "fd.hpp"
 #include "thread.hpp"
-#include "atomic_counter.hpp"
+#include "poller_base.hpp"
 
 namespace zmq
 {
@@ -37,7 +37,7 @@ namespace zmq
     //  This class implements socket polling mechanism using the Linux-specific
     //  epoll mechanism.
 
-    class epoll_t
+    class epoll_t : public poller_base_t
     {
     public:
 
@@ -55,7 +55,6 @@ namespace zmq
         void reset_pollout (handle_t handle_);
         void add_timer (int timeout_, struct i_poll_events *events_, int id_);
         void cancel_timer (struct i_poll_events *events_, int id_);
-        int get_load ();
         void start ();
         void stop ();
 
@@ -91,10 +90,6 @@ namespace zmq
         //  Handle of the physical thread doing the I/O work.
         thread_t worker;
 
-        //  Load of the poller. Currently number of file descriptors
-        //  registered with the poller.
-        atomic_counter_t load;
-
         epoll_t (const epoll_t&);
         void operator = (const epoll_t&);
     };
diff --git a/src/kqueue.cpp b/src/kqueue.cpp
index f76a08f..47178d3 100644
--- a/src/kqueue.cpp
+++ b/src/kqueue.cpp
@@ -54,10 +54,6 @@ zmq::kqueue_t::kqueue_t () :
 zmq::kqueue_t::~kqueue_t ()
 {
     worker.stop ();
-
-    //  Make sure there are no fds registered on shutdown.
-    zmq_assert (load.get () == 0);
-
     close (kqueue_fd);
 }
 
@@ -74,7 +70,7 @@ void zmq::kqueue_t::kevent_delete (fd_t fd_, short filter_)
 {
     struct kevent ev;
 
-    EV_SET (&ev, fd_, filter_, EV_DELETE, 0, 0, (kevent_udata_t)NULL);
+    EV_SET (&ev, fd_, filter_, EV_DELETE, 0, 0, (kevent_udata_t) NULL);
     int rc = kevent (kqueue_fd, &ev, 1, NULL, 0, NULL);
     errno_assert (rc != -1);
 }
@@ -90,6 +86,8 @@ zmq::kqueue_t::handle_t zmq::kqueue_t::add_fd (fd_t fd_,
     pe->flag_pollout = 0;
     pe->reactor = reactor_;
 
+    adjust_load (1);
+
     return pe;
 }
 
@@ -102,6 +100,8 @@ void zmq::kqueue_t::rm_fd (handle_t handle_)
         kevent_delete (pe->fd, EVFILT_WRITE);
     pe->fd = retired_fd;
     retired.push_back (pe);
+
+    adjust_load (-1);
 }
 
 void zmq::kqueue_t::set_pollin (handle_t handle_)
@@ -144,11 +144,6 @@ void zmq::kqueue_t::cancel_timer (i_poll_events *events_, int id_)
         timers.erase (it);
 }
 
-int zmq::kqueue_t::get_load ()
-{
-    return load.get ();
-}
-
 void zmq::kqueue_t::start ()
 {
     worker.start (worker_routine, this);
diff --git a/src/kqueue.hpp b/src/kqueue.hpp
index 43c2a39..6a27260 100644
--- a/src/kqueue.hpp
+++ b/src/kqueue.hpp
@@ -29,7 +29,7 @@
 
 #include "fd.hpp"
 #include "thread.hpp"
-#include "atomic_counter.hpp"
+#include "poller_base.hpp"
 
 namespace zmq
 {
@@ -37,7 +37,7 @@ namespace zmq
     //  Implements socket polling mechanism using the BSD-specific
     //  kqueue interface.
 
-    class kqueue_t
+    class kqueue_t : public poller_base_t
     {
     public:
 
@@ -55,7 +55,6 @@ namespace zmq
         void reset_pollout (handle_t handle_);
         void add_timer (int timeout_, struct i_poll_events *events_, int id_);
         void cancel_timer (struct i_poll_events *events_, int id_);
-        int get_load ();
         void start ();
         void stop ();
 
@@ -98,10 +97,6 @@ namespace zmq
         //  Handle of the physical thread doing the I/O work.
         thread_t worker;
 
-        //  Load of the poller. Currently number of file descriptors
-        //  registered with the poller.
-        atomic_counter_t load;
-
         kqueue_t (const kqueue_t&);
         void operator = (const kqueue_t&);
     };
diff --git a/src/poll.cpp b/src/poll.cpp
index 513c405..9afa6b5 100644
--- a/src/poll.cpp
+++ b/src/poll.cpp
@@ -54,9 +54,6 @@ zmq::poll_t::poll_t () :
 zmq::poll_t::~poll_t ()
 {
     worker.stop ();
-
-    //  Make sure there are no fds registered on shutdown.
-    zmq_assert (load.get () == 0);
 }
 
 zmq::poll_t::handle_t zmq::poll_t::add_fd (fd_t fd_, i_poll_events *events_)
@@ -69,7 +66,7 @@ zmq::poll_t::handle_t zmq::poll_t::add_fd (fd_t fd_, i_poll_events *events_)
     fd_table [fd_].events = events_;
 
     //  Increase the load metric of the thread.
-    load.add (1);
+    adjust_load (1);
 
     return fd_;
 }
@@ -85,7 +82,7 @@ void zmq::poll_t::rm_fd (handle_t handle_)
     retired = true;
 
     //  Decrease the load metric of the thread.
-    load.sub (1);
+    adjust_load (-1);
 }
 
 void zmq::poll_t::set_pollin (handle_t handle_)
@@ -124,11 +121,6 @@ void zmq::poll_t::cancel_timer (i_poll_events *events_, int id_)
         timers.erase (it);
 }
 
-int zmq::poll_t::get_load ()
-{
-    return load.get ();
-}
-
 void zmq::poll_t::start ()
 {
     worker.start (worker_routine, this);
diff --git a/src/poll.hpp b/src/poll.hpp
index 96d18dd..e88c39d 100644
--- a/src/poll.hpp
+++ b/src/poll.hpp
@@ -34,7 +34,7 @@
 
 #include "fd.hpp"
 #include "thread.hpp"
-#include "atomic_counter.hpp"
+#include "poller_base.hpp"
 
 namespace zmq
 {
@@ -42,7 +42,7 @@ namespace zmq
     //  Implements socket polling mechanism using the POSIX.1-2001
     //  poll() system call.
 
-    class poll_t
+    class poll_t : public poller_base_t
     {
     public:
 
@@ -60,7 +60,6 @@ namespace zmq
         void reset_pollout (handle_t handle_);
         void add_timer (int timeout_, struct i_poll_events *events_, int id_);
         void cancel_timer (struct i_poll_events *events_, int id_);
-        int get_load ();
         void start ();
         void stop ();
 
@@ -98,10 +97,6 @@ namespace zmq
         //  Handle of the physical thread doing the I/O work.
         thread_t worker;
 
-        //  Load of the poller. Currently number of file descriptors
-        //  registered with the poller.
-        atomic_counter_t load;
-
         poll_t (const poll_t&);
         void operator = (const poll_t&);
     };
diff --git a/src/poller_base.cpp b/src/poller_base.cpp
new file mode 100644
index 0000000..f1de7e9
--- /dev/null
+++ b/src/poller_base.cpp
@@ -0,0 +1,44 @@
+/*
+    Copyright (c) 2007-2010 iMatix Corporation
+
+    This file is part of 0MQ.
+
+    0MQ is free software; you can redistribute it and/or modify it under
+    the terms of the Lesser GNU General Public License as published by
+    the Free Software Foundation; either version 3 of the License, or
+    (at your option) any later version.
+
+    0MQ is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    Lesser GNU General Public License for more details.
+
+    You should have received a copy of the Lesser GNU General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include "poller_base.hpp"
+#include "err.hpp"
+
+zmq::poller_base_t::poller_base_t ()
+{
+}
+
+zmq::poller_base_t::~poller_base_t ()
+{
+    //  Make sure there are no fds registered on shutdown.
+    zmq_assert (get_load () == 0);
+}
+
+int zmq::poller_base_t::get_load ()
+{
+    return load.get ();
+}
+
+void zmq::poller_base_t::adjust_load (int amount_)
+{
+    if (amount_ > 0)
+        load.add (amount_);
+    else if (amount_ < 0)
+        load.sub (-amount_);
+}
diff --git a/src/poller_base.hpp b/src/poller_base.hpp
new file mode 100644
index 0000000..0b0d53d
--- /dev/null
+++ b/src/poller_base.hpp
@@ -0,0 +1,56 @@
+/*
+    Copyright (c) 2007-2010 iMatix Corporation
+
+    This file is part of 0MQ.
+
+    0MQ is free software; you can redistribute it and/or modify it under
+    the terms of the Lesser GNU General Public License as published by
+    the Free Software Foundation; either version 3 of the License, or
+    (at your option) any later version.
+
+    0MQ is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    Lesser GNU General Public License for more details.
+
+    You should have received a copy of the Lesser GNU General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#ifndef __ZMQ_POLLER_BASE_HPP_INCLUDED__
+#define __ZMQ_POLLER_BASE_HPP_INCLUDED__
+
+#include "atomic_counter.hpp"
+
+namespace zmq
+{
+
+    class poller_base_t
+    {
+    public:
+
+        poller_base_t ();
+        ~poller_base_t ();
+
+        //  Returns load of the poller. Note that this function can be
+        //  invoked from a different thread!
+        int get_load ();
+
+    protected:
+
+        //  Called by individual poller implementations to manage the load.
+        void adjust_load (int amount_);
+
+    private:
+
+        //  Load of the poller. Currently the number of file descriptors
+        //  registered.
+        atomic_counter_t load;
+
+        poller_base_t (const poller_base_t&);
+        void operator = (const poller_base_t&);
+    };
+
+}
+
+#endif
diff --git a/src/select.cpp b/src/select.cpp
index f169239..a04ee66 100644
--- a/src/select.cpp
+++ b/src/select.cpp
@@ -54,9 +54,6 @@ zmq::select_t::select_t () :
 zmq::select_t::~select_t ()
 {
     worker.stop ();
-
-    //  Make sure there are no fds registered on shutdown.
-    zmq_assert (load.get () == 0);
 }
 
 zmq::select_t::handle_t zmq::select_t::add_fd (fd_t fd_, i_poll_events *events_)
@@ -77,7 +74,7 @@ zmq::select_t::handle_t zmq::select_t::add_fd (fd_t fd_, i_poll_events *events_)
         maxfd = fd_;
 
     //  Increase the load metric of the thread.
-    load.add (1);
+    adjust_load (1);
 
     return fd_;
 }
@@ -113,7 +110,7 @@ void zmq::select_t::rm_fd (handle_t handle_)
     }
 
     //  Decrease the load metric of the thread.
-    load.sub (1);
+    adjust_load (-1);
 }
 
 void zmq::select_t::set_pollin (handle_t handle_)
@@ -148,11 +145,6 @@ void zmq::select_t::cancel_timer (i_poll_events *events_, int id_)
         timers.erase (it);
 }
 
-int zmq::select_t::get_load ()
-{
-    return load.get ();
-}
-
 void zmq::select_t::start ()
 {
     worker.start (worker_routine, this);
diff --git a/src/select.hpp b/src/select.hpp
index ae19fe1..c6c7f51 100644
--- a/src/select.hpp
+++ b/src/select.hpp
@@ -36,7 +36,7 @@
 
 #include "fd.hpp"
 #include "thread.hpp"
-#include "atomic_counter.hpp"
+#include "poller_base.hpp"
 
 namespace zmq
 {
@@ -44,7 +44,7 @@ namespace zmq
     //  Implements socket polling mechanism using POSIX.1-2001 select()
     //  function.
 
-    class select_t
+    class select_t : public poller_base_t
     {
     public:
 
@@ -62,7 +62,6 @@ namespace zmq
         void reset_pollout (handle_t handle_);
         void add_timer (int timeout_, struct i_poll_events *events_, int id_);
         void cancel_timer (struct i_poll_events *events_, int id_);
-        int get_load ();
         void start ();
         void stop ();
 
@@ -109,10 +108,6 @@ namespace zmq
         //  Handle of the physical thread doing the I/O work.
         thread_t worker;
 
-        //  Load of the poller. Currently number of file descriptors
-        //  registered with the poller.
-        atomic_counter_t load;
-
         select_t (const select_t&);
         void operator = (const select_t&);
     };