Merge pull request #3451 from jacquesg/devpoll

Restore /dev/poll support

README.md

@@ -68,6 +68,7 @@ case of a released version.
 
 OS and version | Architecture | Compiler and version | Build system | Last report | Remarks
 -------------- | ------------ | -------------------- | ------------ | ----------- | -------
+Solaris 10 | x86, amd64, sparc | GCC 8.1.0 | CMake | 2019/03/18 |
 
 
 ### Supported platforms without known active users

src/devpoll.cpp

@@ -47,8 +47,7 @@
 #include "i_poll_events.hpp"
 
 zmq::devpoll_t::devpoll_t (const zmq::thread_ctx_t &ctx_) :
-    ctx (ctx_),
-    stopping (false)
+    worker_poller_base_t (ctx_)
 {
     devpoll_fd = open ("/dev/poll", O_RDWR);
     errno_assert (devpoll_fd != -1);
@@ -56,7 +55,9 @@ zmq::devpoll_t::devpoll_t (const zmq::thread_ctx_t &ctx_) :
 
 zmq::devpoll_t::~devpoll_t ()
 {
-    worker.stop ();
+    //  Wait till the worker thread exits.
+    stop_worker ();
+
     close (devpoll_fd);
 }
 
@@ -70,6 +71,7 @@ void zmq::devpoll_t::devpoll_ctl (fd_t fd_, short events_)
 zmq::devpoll_t::handle_t zmq::devpoll_t::add_fd (fd_t fd_,
                                                  i_poll_events *reactor_)
 {
+    check_thread ();
     //  If the file descriptor table is too small expand it.
     fd_table_t::size_type sz = fd_table.size ();
     if (sz <= (fd_table_t::size_type) fd_) {
@@ -98,6 +100,7 @@ zmq::devpoll_t::handle_t zmq::devpoll_t::add_fd (fd_t fd_,
 
 void zmq::devpoll_t::rm_fd (handle_t handle_)
 {
+    check_thread ();
     zmq_assert (fd_table[handle_].valid);
 
     devpoll_ctl (handle_, POLLREMOVE);
@@ -109,6 +112,7 @@ void zmq::devpoll_t::rm_fd (handle_t handle_)
 
 void zmq::devpoll_t::set_pollin (handle_t handle_)
 {
+    check_thread ();
     devpoll_ctl (handle_, POLLREMOVE);
     fd_table[handle_].events |= POLLIN;
     devpoll_ctl (handle_, fd_table[handle_].events);
@@ -116,6 +120,7 @@ void zmq::devpoll_t::set_pollin (handle_t handle_)
 
 void zmq::devpoll_t::reset_pollin (handle_t handle_)
 {
+    check_thread ();
     devpoll_ctl (handle_, POLLREMOVE);
     fd_table[handle_].events &= ~((short) POLLIN);
     devpoll_ctl (handle_, fd_table[handle_].events);
@@ -123,6 +128,7 @@ void zmq::devpoll_t::reset_pollin (handle_t handle_)
 
 void zmq::devpoll_t::set_pollout (handle_t handle_)
 {
+    check_thread ();
     devpoll_ctl (handle_, POLLREMOVE);
     fd_table[handle_].events |= POLLOUT;
     devpoll_ctl (handle_, fd_table[handle_].events);
@@ -130,19 +136,15 @@ void zmq::devpoll_t::set_pollout (handle_t handle_)
 
 void zmq::devpoll_t::reset_pollout (handle_t handle_)
 {
+    check_thread ();
     devpoll_ctl (handle_, POLLREMOVE);
     fd_table[handle_].events &= ~((short) POLLOUT);
     devpoll_ctl (handle_, fd_table[handle_].events);
 }
 
-void zmq::devpoll_t::start ()
-{
-    ctx.start_thread (worker, worker_routine, this);
-}
-
 void zmq::devpoll_t::stop ()
 {
-    stopping = true;
+    check_thread ();
 }
 
 int zmq::devpoll_t::max_fds ()
@@ -152,7 +154,7 @@ int zmq::devpoll_t::max_fds ()
 
 void zmq::devpoll_t::loop ()
 {
-    while (!stopping) {
+    while (true) {
         struct pollfd ev_buf[max_io_events];
         struct dvpoll poll_req;
 
@@ -163,6 +165,14 @@ void zmq::devpoll_t::loop ()
         //  Execute any due timers.
         int timeout = (int) execute_timers ();
 
+        if (get_load () == 0) {
+            if (timeout == 0)
+                break;
+
+            // TODO sleep for timeout
+            continue;
+        }
+
         //  Wait for events.
         //  On Solaris, we can retrieve no more then (OPEN_MAX - 1) events.
         poll_req.dp_fds = &ev_buf[0];
@@ -195,9 +205,4 @@ void zmq::devpoll_t::loop ()
     }
 }
 
-void zmq::devpoll_t::worker_routine (void *arg_)
-{
-    ((devpoll_t *) arg_)->loop ();
-}
-
 #endif

src/devpoll.hpp

@@ -47,7 +47,7 @@ struct i_poll_events;
 
 //  Implements socket polling mechanism using the "/dev/poll" interface.
 
-class devpoll_t : public poller_base_t
+class devpoll_t : public worker_poller_base_t
 {
   public:
     typedef fd_t handle_t;
@@ -62,21 +62,14 @@ class devpoll_t : public poller_base_t
     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 ();
 
   private:
-    //  Main worker thread routine.
-    static void worker_routine (void *arg_);
-
     //  Main event loop.
     void loop ();
 
-    // Reference to ZMQ context.
-    const thread_ctx_t &ctx;
-
     //  File descriptor referring to "/dev/poll" pseudo-device.
     fd_t devpoll_fd;
 
@@ -97,12 +90,6 @@ class devpoll_t : public poller_base_t
     //  Pollset manipulation function.
     void devpoll_ctl (fd_t fd_, short events_);
 
-    //  If true, thread is in the process of shutting down.
-    bool stopping;
-
-    //  Handle of the physical thread doing the I/O work.
-    thread_t worker;
-
     devpoll_t (const devpoll_t &);
     const devpoll_t &operator= (const devpoll_t &);
 };

src/epoll.hpp

@@ -103,9 +103,6 @@ class epoll_t : public worker_poller_base_t
     typedef std::vector<poll_entry_t *> retired_t;
     retired_t _retired;
 
-    //  Handle of the physical thread doing the I/O work.
-    thread_t _worker;
-
     epoll_t (const epoll_t &);
     const epoll_t &operator= (const epoll_t &);
 };