io_thread_t merged with poller_base_t

The relationship of these two classes was 1:1.
Thus one of them was obsolete.

Signed-off-by: Martin Sustrik <sustrik@250bpm.com>

58 files changed, 185 insertions, 319 deletions

diff --git a/src/Makefile.am b/src/Makefile.am
index c2c6f65..7d3ee8d 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -23,7 +23,6 @@ libxs_la_SOURCES = \
     fd.hpp \
     fq.hpp \
     io_object.hpp \
-    io_thread.hpp \
     ip.hpp \
     ipc_address.hpp \
     ipc_connecter.hpp \
@@ -86,7 +85,6 @@ libxs_la_SOURCES = \
     err.cpp \
     fq.cpp \
     io_object.cpp \
-    io_thread.cpp \
     ip.cpp \
     ipc_address.cpp \
     ipc_connecter.cpp \
diff --git a/src/ctx.cpp b/src/ctx.cpp
index a8df950..8b27b31 100644
--- a/src/ctx.cpp
+++ b/src/ctx.cpp
@@ -31,7 +31,6 @@
 
 #include "ctx.hpp"
 #include "socket_base.hpp"
-#include "io_thread.hpp"
 #include "monitor.hpp"
 #include "reaper.hpp"
 #include "pipe.hpp"
@@ -198,8 +197,8 @@ xs::socket_base_t *xs::ctx_t::create_socket (int type_)
 
         //  Create I/O thread objects and launch them.
         for (uint32_t i = 2; i != io_thread_count + 2; i++) {
-            io_thread_t *io_thread = new (std::nothrow) io_thread_t (this, i);
-            alloc_assert (io_thread);
+            poller_base_t *io_thread = poller_base_t::create (this, i);
+            errno_assert (io_thread);
             io_threads.push_back (io_thread);
             slots [i] = io_thread->get_mailbox ();
             io_thread->start ();
@@ -227,7 +226,7 @@ xs::socket_base_t *xs::ctx_t::create_socket (int type_)
     #endif
 
         //  Create the monitor object.
-        io_thread_t *io_thread = choose_io_thread (0);
+        poller_base_t *io_thread = choose_io_thread (0);
         xs_assert (io_thread);
         monitor = new (std::nothrow) monitor_t (io_thread);
         alloc_assert (monitor);
@@ -302,7 +301,7 @@ void xs::ctx_t::send_command (uint32_t tid_, const command_t &command_)
     slots [tid_]->send (command_);
 }
 
-xs::io_thread_t *xs::ctx_t::choose_io_thread (uint64_t affinity_)
+xs::poller_base_t *xs::ctx_t::choose_io_thread (uint64_t affinity_)
 {
     if (io_threads.empty ())
         return NULL;
diff --git a/src/ctx.hpp b/src/ctx.hpp
index e912443..c12fffc 100644
--- a/src/ctx.hpp
+++ b/src/ctx.hpp
@@ -40,7 +40,7 @@ namespace xs
 
     class object_t;
     class monitor_t;
-    class io_thread_t;
+    class poller_base_t;
     class socket_base_t;
     class reaper_t;
 
@@ -86,7 +86,7 @@ namespace xs
         //  Returns the I/O thread that is the least busy at the moment.
         //  Affinity specifies which I/O threads are eligible (0 = all).
         //  Returns NULL is no I/O thread is available.
-        xs::io_thread_t *choose_io_thread (uint64_t affinity_);
+        xs::poller_base_t *choose_io_thread (uint64_t affinity_);
 
         //  Returns reaper thread object.
         xs::object_t *get_reaper ();
@@ -142,7 +142,7 @@ namespace xs
         xs::reaper_t *reaper;
 
         //  I/O threads.
-        typedef std::vector <xs::io_thread_t*> io_threads_t;
+        typedef std::vector <xs::poller_base_t*> io_threads_t;
         io_threads_t io_threads;
 
         //  Array of pointers to mailboxes for both application and I/O threads.
diff --git a/src/epoll.cpp b/src/epoll.cpp
index 7208a57..7ca4608 100644
--- a/src/epoll.cpp
+++ b/src/epoll.cpp
@@ -34,7 +34,8 @@
 #include "config.hpp"
 #include "err.hpp"
 
-xs::epoll_t::epoll_t () :
+xs::epoll_t::epoll_t (xs::ctx_t *ctx_, uint32_t tid_) :
+    poller_base_t (ctx_, tid_),
     stopping (false)
 {
     epoll_fd = epoll_create (1);
diff --git a/src/epoll.hpp b/src/epoll.hpp
index bac555b..232d049 100644
--- a/src/epoll.hpp
+++ b/src/epoll.hpp
@@ -36,6 +36,7 @@
 namespace xs
 {
 
+    class ctx_t;
     struct i_poll_events;
 
     //  This class implements socket polling mechanism using the Linux-specific
@@ -45,7 +46,7 @@ namespace xs
     {
     public:
 
-        epoll_t ();
+        epoll_t (xs::ctx_t *ctx_, uint32_t tid_);
         ~epoll_t ();
 
         //  "poller" concept.
diff --git a/src/i_engine.hpp b/src/i_engine.hpp
index fc03f0e..9106411 100644
--- a/src/i_engine.hpp
+++ b/src/i_engine.hpp
@@ -25,7 +25,7 @@
 namespace xs
 {
 
-    class io_thread_t;
+    class poller_base_t;
 
     //  Abstract interface to be implemented by various engines.
 
@@ -34,7 +34,7 @@ namespace xs
         virtual ~i_engine () {}
 
         //  Plug the engine to the session.
-        virtual void plug (xs::io_thread_t *io_thread_,
+        virtual void plug (xs::poller_base_t *io_thread_,
             class session_base_t *session_) = 0;
 
         //  Unplug the engine from the session.
diff --git a/src/io_object.cpp b/src/io_object.cpp
index abc8204..0c73571 100644
--- a/src/io_object.cpp
+++ b/src/io_object.cpp
@@ -20,10 +20,10 @@
 */
 
 #include "io_object.hpp"
-#include "io_thread.hpp"
+#include "poller_base.hpp"
 #include "err.hpp"
 
-xs::io_object_t::io_object_t (io_thread_t *io_thread_) :
+xs::io_object_t::io_object_t (poller_base_t *io_thread_) :
     poller (NULL)
 {
     if (io_thread_)
@@ -34,13 +34,13 @@ xs::io_object_t::~io_object_t ()
 {
 }
 
-void xs::io_object_t::plug (io_thread_t *io_thread_)
+void xs::io_object_t::plug (poller_base_t *io_thread_)
 {
     xs_assert (io_thread_);
     xs_assert (!poller);
 
     //  Retrieve the poller from the thread we are running in.
-    poller = io_thread_->get_poller ();
+    poller = io_thread_;
 }
 
 void xs::io_object_t::unplug ()
diff --git a/src/io_object.hpp b/src/io_object.hpp
index 0abd457..0749ceb 100644
--- a/src/io_object.hpp
+++ b/src/io_object.hpp
@@ -30,8 +30,6 @@
 namespace xs
 {
 
-    class io_thread_t;
-
     //  Simple base class for objects that live in I/O threads.
     //  It makes communication with the poller object easier and
     //  makes defining unneeded event handlers unnecessary.
@@ -40,12 +38,12 @@ namespace xs
     {
     public:
 
-        io_object_t (xs::io_thread_t *io_thread_ = NULL);
+        io_object_t (xs::poller_base_t *io_thread_ = NULL);
         ~io_object_t ();
 
         //  When migrating an object from one I/O thread to another, first
         //  unplug it, then migrate it, then plug it to the new thread.
-        void plug (xs::io_thread_t *io_thread_);
+        void plug (xs::poller_base_t *io_thread_);
         void unplug ();
 
     protected:
diff --git a/src/io_thread.cpp b/src/io_thread.cpp
deleted file mode 100644
index 77cebfe..0000000
--- a/src/io_thread.cpp
+++ /dev/null
@@ -1,108 +0,0 @@
-/*
-    Copyright (c) 2009-2012 250bpm s.r.o.
-    Copyright (c) 2007-2009 iMatix Corporation
-    Copyright (c) 2007-2011 Other contributors as noted in the AUTHORS file
-
-    This file is part of Crossroads project.
-
-    Crossroads is free software; you can redistribute it and/or modify it under
-    the terms of the GNU Lesser General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    Crossroads 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
-    GNU Lesser General Public License for more details.
-
-    You should have received a copy of the GNU Lesser General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#include <new>
-
-#include "io_thread.hpp"
-#include "platform.hpp"
-#include "err.hpp"
-#include "ctx.hpp"
-
-xs::io_thread_t::io_thread_t (ctx_t *ctx_, uint32_t tid_) :
-    object_t (ctx_, tid_)
-{
-    poller = poller_base_t::create ();
-    xs_assert (poller);
-
-    mailbox_handle = poller->add_fd (mailbox.get_fd (), this);
-    poller->set_pollin (mailbox_handle);
-}
-
-xs::io_thread_t::~io_thread_t ()
-{
-    delete poller;
-}
-
-void xs::io_thread_t::start ()
-{
-    //  Start the underlying I/O thread.
-    poller->start ();
-}
-
-void xs::io_thread_t::stop ()
-{
-    send_stop ();
-}
-
-xs::mailbox_t *xs::io_thread_t::get_mailbox ()
-{
-    return &mailbox;
-}
-
-int xs::io_thread_t::get_load ()
-{
-    return poller->get_load ();
-}
-
-void xs::io_thread_t::in_event (fd_t fd_)
-{
-    //  TODO: Do we want to limit number of commands I/O thread can
-    //  process in a single go?
-
-    while (true) {
-
-        //  Get the next command. If there is none, exit.
-        command_t cmd;
-        int rc = mailbox.recv (&cmd, 0);
-        if (rc != 0 && errno == EINTR)
-            continue;
-        if (rc != 0 && errno == EAGAIN)
-            break;
-        errno_assert (rc == 0);
-
-        //  Process the command.
-        cmd.destination->process_command (cmd);
-    }
-}
-
-void xs::io_thread_t::out_event (fd_t fd_)
-{
-    //  We are never polling for POLLOUT here. This function is never called.
-    xs_assert (false);
-}
-
-void xs::io_thread_t::timer_event (handle_t handle_)
-{
-    //  No timers here. This function is never called.
-    xs_assert (false);
-}
-
-xs::poller_base_t *xs::io_thread_t::get_poller ()
-{
-    xs_assert (poller);
-    return poller;
-}
-
-void xs::io_thread_t::process_stop ()
-{
-    poller->rm_fd (mailbox_handle);
-    poller->stop ();
-}
diff --git a/src/io_thread.hpp b/src/io_thread.hpp
deleted file mode 100644
index 379f420..0000000
--- a/src/io_thread.hpp
+++ /dev/null
@@ -1,90 +0,0 @@
-/*
-    Copyright (c) 2009-2012 250bpm s.r.o.
-    Copyright (c) 2007-2009 iMatix Corporation
-    Copyright (c) 2007-2011 Other contributors as noted in the AUTHORS file
-
-    This file is part of Crossroads project.
-
-    Crossroads is free software; you can redistribute it and/or modify it under
-    the terms of the GNU Lesser General Public License as published by
-    the Free Software Foundation; either version 3 of the License, or
-    (at your option) any later version.
-
-    Crossroads 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
-    GNU Lesser General Public License for more details.
-
-    You should have received a copy of the GNU Lesser General Public License
-    along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#ifndef __XS_IO_THREAD_HPP_INCLUDED__
-#define __XS_IO_THREAD_HPP_INCLUDED__
-
-#include <vector>
-
-#include "stdint.hpp"
-#include "object.hpp"
-#include "poller_base.hpp"
-#include "mailbox.hpp"
-
-namespace xs
-{
-
-    class ctx_t;
-
-    //  Generic part of the I/O thread. Polling-mechanism-specific features
-    //  are implemented in separate "polling objects".
-
-    class io_thread_t : public object_t, public i_poll_events
-    {
-    public:
-
-        io_thread_t (xs::ctx_t *ctx_, uint32_t tid_);
-
-        //  Clean-up. If the thread was started, it's neccessary to call 'stop'
-        //  before invoking destructor. Otherwise the destructor would hang up.
-        ~io_thread_t ();
-
-        //  Launch the physical thread.
-        void start ();
-
-        //  Ask underlying thread to stop.
-        void stop ();
-
-        //  Returns mailbox associated with this I/O thread.
-        mailbox_t *get_mailbox ();
-
-        //  i_poll_events implementation.
-        void in_event (fd_t fd_);
-        void out_event (fd_t fd_);
-        void timer_event (handle_t handle_);
-
-        //  Used by io_objects to retrieve the assciated poller object.
-        poller_base_t *get_poller ();
-
-        //  Command handlers.
-        void process_stop ();
-
-        //  Returns load experienced by the I/O thread.
-        int get_load ();
-
-    private:
-
-        //  I/O thread accesses incoming commands via this mailbox.
-        mailbox_t mailbox;
-
-        //  Handle associated with mailbox' file descriptor.
-        handle_t mailbox_handle;
-
-        //  I/O multiplexing is performed using a poller object.
-        poller_base_t *poller;
-
-        io_thread_t (const io_thread_t&);
-        const io_thread_t &operator = (const io_thread_t&);
-    };
-
-}
-
-#endif
diff --git a/src/ipc_connecter.cpp b/src/ipc_connecter.cpp
index f6a01c3..b65887a 100644
--- a/src/ipc_connecter.cpp
+++ b/src/ipc_connecter.cpp
@@ -26,7 +26,7 @@
 #include <string>
 
 #include "stream_engine.hpp"
-#include "io_thread.hpp"
+#include "poller_base.hpp"
 #include "platform.hpp"
 #include "random.hpp"
 #include "err.hpp"
@@ -37,7 +37,7 @@
 #include <sys/socket.h>
 #include <sys/un.h>
 
-xs::ipc_connecter_t::ipc_connecter_t (class io_thread_t *io_thread_,
+xs::ipc_connecter_t::ipc_connecter_t (class poller_base_t *io_thread_,
       class session_base_t *session_, const options_t &options_,
       const char *address_, bool wait_) :
     own_t (io_thread_, options_),
diff --git a/src/ipc_connecter.hpp b/src/ipc_connecter.hpp
index 2a8e8e4..46e260c 100644
--- a/src/ipc_connecter.hpp
+++ b/src/ipc_connecter.hpp
@@ -34,7 +34,7 @@
 namespace xs
 {
 
-    class io_thread_t;
+    class poller_base_t;
     class session_base_t;
 
     class ipc_connecter_t : public own_t, public io_object_t
@@ -43,7 +43,7 @@ namespace xs
 
         //  If 'delay' is true connecter first waits for a while, then starts
         //  connection process.
-        ipc_connecter_t (xs::io_thread_t *io_thread_,
+        ipc_connecter_t (xs::poller_base_t *io_thread_,
             xs::session_base_t *session_, const options_t &options_,
             const char *address_, bool delay_);
         ~ipc_connecter_t ();
diff --git a/src/ipc_listener.cpp b/src/ipc_listener.cpp
index db49528..bd55cc7 100644
--- a/src/ipc_listener.cpp
+++ b/src/ipc_listener.cpp
@@ -28,7 +28,7 @@
 
 #include "stream_engine.hpp"
 #include "ipc_address.hpp"
-#include "io_thread.hpp"
+#include "poller_base.hpp"
 #include "session_base.hpp"
 #include "config.hpp"
 #include "err.hpp"
@@ -39,7 +39,7 @@
 #include <fcntl.h>
 #include <sys/un.h>
 
-xs::ipc_listener_t::ipc_listener_t (io_thread_t *io_thread_,
+xs::ipc_listener_t::ipc_listener_t (poller_base_t *io_thread_,
       socket_base_t *socket_, const options_t &options_) :
     own_t (io_thread_, options_),
     io_object_t (io_thread_),
@@ -83,7 +83,7 @@ void xs::ipc_listener_t::in_event (fd_t fd_)
 
     //  Choose I/O thread to run connecter in. Given that we are already
     //  running in an I/O thread, there must be at least one available.
-    io_thread_t *io_thread = choose_io_thread (options.affinity);
+    poller_base_t *io_thread = choose_io_thread (options.affinity);
     xs_assert (io_thread);
 
     //  Create and launch a session object. 
diff --git a/src/ipc_listener.hpp b/src/ipc_listener.hpp
index b599bff..24b96fb 100644
--- a/src/ipc_listener.hpp
+++ b/src/ipc_listener.hpp
@@ -35,14 +35,14 @@
 namespace xs
 {
 
-    class io_thread_t;
+    class poller_base_t;
     class socket_base_t;
 
     class ipc_listener_t : public own_t, public io_object_t
     {
     public:
 
-        ipc_listener_t (xs::io_thread_t *io_thread_,
+        ipc_listener_t (xs::poller_base_t *io_thread_,
             xs::socket_base_t *socket_, const options_t &options_);
         ~ipc_listener_t ();
 
diff --git a/src/monitor.cpp b/src/monitor.cpp
index b29c874..14a31f2 100644
--- a/src/monitor.cpp
+++ b/src/monitor.cpp
@@ -19,12 +19,12 @@
 */
 
 #include "monitor.hpp"
-#include "io_thread.hpp"
+#include "poller_base.hpp"
 #include "options.hpp"
 #include "random.hpp"
 #include "err.hpp"
 
-xs::monitor_t::monitor_t (xs::io_thread_t *io_thread_) :
+xs::monitor_t::monitor_t (xs::poller_base_t *io_thread_) :
     own_t (io_thread_, options_t ()),
     io_object_t (io_thread_),
     timer (NULL)
diff --git a/src/monitor.hpp b/src/monitor.hpp
index 73bc31c..91d441a 100644
--- a/src/monitor.hpp
+++ b/src/monitor.hpp
@@ -30,14 +30,14 @@
 namespace xs
 {
 
-    class io_thread_t;
+    class poller_base_t;
     class socket_base_t;
 
     class monitor_t : public own_t, public io_object_t
     {
     public:
 
-        monitor_t (xs::io_thread_t *io_thread_);
+        monitor_t (xs::poller_base_t *io_thread_);
         ~monitor_t ();
 
         void start ();
diff --git a/src/object.cpp b/src/object.cpp
index 2d1f5aa..df279f5 100644
--- a/src/object.cpp
+++ b/src/object.cpp
@@ -26,7 +26,7 @@
 #include "ctx.hpp"
 #include "err.hpp"
 #include "pipe.hpp"
-#include "io_thread.hpp"
+#include "poller_base.hpp"
 #include "session_base.hpp"
 #include "socket_base.hpp"
 
@@ -149,7 +149,7 @@ void xs::object_t::destroy_socket (socket_base_t *socket_)
     ctx->destroy_socket (socket_);
 }
 
-xs::io_thread_t *xs::object_t::choose_io_thread (uint64_t affinity_)
+xs::poller_base_t *xs::object_t::choose_io_thread (uint64_t affinity_)
 {
     return ctx->choose_io_thread (affinity_);
 }
diff --git a/src/object.hpp b/src/object.hpp
index b1bc0c3..14dad49 100644
--- a/src/object.hpp
+++ b/src/object.hpp
@@ -34,7 +34,7 @@ namespace xs
     class pipe_t;
     class socket_base_t;
     class session_base_t;
-    class io_thread_t;
+    class poller_base_t;
     class own_t;
 
     //  Base class for all objects that participate in inter-thread
@@ -62,7 +62,7 @@ namespace xs
         void destroy_socket (xs::socket_base_t *socket_);
 
         //  Chooses least loaded I/O thread.
-        xs::io_thread_t *choose_io_thread (uint64_t affinity_);
+        xs::poller_base_t *choose_io_thread (uint64_t affinity_);
 
         //  Logging related functions.
         void log (int sid_, const char *text_);
diff --git a/src/own.cpp b/src/own.cpp
index a60d9d0..3047d7b 100644
--- a/src/own.cpp
+++ b/src/own.cpp
@@ -20,7 +20,7 @@
 
 #include "own.hpp"
 #include "err.hpp"
-#include "io_thread.hpp"
+#include "poller_base.hpp"
 
 xs::own_t::own_t (class ctx_t *parent_, uint32_t tid_) :
     object_t (parent_, tid_),
@@ -32,7 +32,7 @@ xs::own_t::own_t (class ctx_t *parent_, uint32_t tid_) :
 {
 }
 
-xs::own_t::own_t (io_thread_t *io_thread_, const options_t &options_) :
+xs::own_t::own_t (poller_base_t *io_thread_, const options_t &options_) :
     object_t (io_thread_),
     options (options_),
     terminating (false),
diff --git a/src/own.hpp b/src/own.hpp
index d0afd62..de0320a 100644
--- a/src/own.hpp
+++ b/src/own.hpp
@@ -33,7 +33,7 @@ namespace xs
 {
 
     class ctx_t;
-    class io_thread_t;
+    class poller_base_t;
 
     //  Base class for objects forming a part of ownership hierarchy.
     //  It handles initialisation and destruction of such objects.
@@ -50,7 +50,7 @@ namespace xs
         own_t (xs::ctx_t *parent_, uint32_t tid_);
 
         //  The object is living within I/O thread.
-        own_t (xs::io_thread_t *io_thread_, const options_t &options_);
+        own_t (xs::poller_base_t *io_thread_, const options_t &options_);
 
         //  When another owned object wants to send command to this object
         //  it calls this function to let it know it should not shut down
diff --git a/src/pair.cpp b/src/pair.cpp
index dd14d8d..c35084f 100644
--- a/src/pair.cpp
+++ b/src/pair.cpp
@@ -117,7 +117,7 @@ bool xs::pair_t::xhas_out ()
     return result;
 }
 
-xs::pair_session_t::pair_session_t (io_thread_t *io_thread_, bool connect_,
+xs::pair_session_t::pair_session_t (poller_base_t *io_thread_, bool connect_,
       socket_base_t *socket_, const options_t &options_,
       const char *protocol_, const char *address_) :
     session_base_t (io_thread_, connect_, socket_, options_, protocol_,
diff --git a/src/pair.hpp b/src/pair.hpp
index 8ae3acf..658e8e9 100644
--- a/src/pair.hpp
+++ b/src/pair.hpp
@@ -63,7 +63,7 @@ namespace xs
     {
     public:
 
-        pair_session_t (xs::io_thread_t *io_thread_, bool connect_,
+        pair_session_t (xs::poller_base_t *io_thread_, bool connect_,
             socket_base_t *socket_, const options_t &options_,
             const char *protocol_, const char *address_);
         ~pair_session_t ();
diff --git a/src/pgm_receiver.cpp b/src/pgm_receiver.cpp
index b2cc513..3b29197 100644
--- a/src/pgm_receiver.cpp
+++ b/src/pgm_receiver.cpp
@@ -37,7 +37,7 @@
 #include "wire.hpp"
 #include "err.hpp"
 
-xs::pgm_receiver_t::pgm_receiver_t (class io_thread_t *parent_, 
+xs::pgm_receiver_t::pgm_receiver_t (class poller_base_t *parent_, 
       const options_t &options_) :
     io_object_t (parent_),
     pgm_socket (true, options_),
@@ -60,7 +60,7 @@ int xs::pgm_receiver_t::init (bool udp_encapsulation_, const char *network_)
     return pgm_socket.init (udp_encapsulation_, network_);
 }
 
-void xs::pgm_receiver_t::plug (io_thread_t *io_thread_,
+void xs::pgm_receiver_t::plug (poller_base_t *io_thread_,
     session_base_t *session_)
 {
     //  Retrieve PGM fds and start polling.
diff --git a/src/pgm_receiver.hpp b/src/pgm_receiver.hpp
index b8390c1..952265e 100644
--- a/src/pgm_receiver.hpp
+++ b/src/pgm_receiver.hpp
@@ -43,7 +43,7 @@
 namespace xs
 {
 
-    class io_thread_t;
+    class poller_base_t;
     class session_base_t;
 
     class pgm_receiver_t : public io_object_t, public i_engine
@@ -51,13 +51,13 @@ namespace xs
     
     public:
 
-        pgm_receiver_t (xs::io_thread_t *parent_, const options_t &options_);
+        pgm_receiver_t (xs::poller_base_t *parent_, const options_t &options_);
         ~pgm_receiver_t ();
 
         int init (bool udp_encapsulation_, const char *network_);
 
         //  i_engine interface implementation.
-        void plug (xs::io_thread_t *io_thread_,
+        void plug (xs::poller_base_t *io_thread_,
             xs::session_base_t *session_);
         void unplug ();
         void terminate ();
diff --git a/src/pgm_sender.cpp b/src/pgm_sender.cpp
index dde6284..20d3dba 100644
--- a/src/pgm_sender.cpp
+++ b/src/pgm_sender.cpp
@@ -31,14 +31,13 @@
 
 #include <stdlib.h>
 
-#include "io_thread.hpp"
 #include "pgm_sender.hpp"
 #include "session_base.hpp"
 #include "err.hpp"
 #include "wire.hpp"
 #include "stdint.hpp"
 
-xs::pgm_sender_t::pgm_sender_t (io_thread_t *parent_, 
+xs::pgm_sender_t::pgm_sender_t (poller_base_t *parent_, 
       const options_t &options_) :
     io_object_t (parent_),
     encoder (0),
@@ -65,7 +64,8 @@ int xs::pgm_sender_t::init (bool udp_encapsulation_, const char *network_)
     return rc;
 }
 
-void xs::pgm_sender_t::plug (io_thread_t *io_thread_, session_base_t *session_)
+void xs::pgm_sender_t::plug (poller_base_t *io_thread_,
+    session_base_t *session_)
 {
     //  Alocate 2 fds for PGM socket.
     fd_t downlink_socket_fd = retired_fd;
diff --git a/src/pgm_sender.hpp b/src/pgm_sender.hpp
index 000f5ba..be6476b 100644
--- a/src/pgm_sender.hpp
+++ b/src/pgm_sender.hpp
@@ -41,7 +41,7 @@
 namespace xs
 {
 
-    class io_thread_t;
+    class poller_base_t;
     class session_base_t;
 
     class pgm_sender_t : public io_object_t, public i_engine
@@ -49,13 +49,13 @@ namespace xs
 
     public:
 
-        pgm_sender_t (xs::io_thread_t *parent_, const options_t &options_);
+        pgm_sender_t (xs::poller_base_t *parent_, const options_t &options_);
         ~pgm_sender_t ();
 
         int init (bool udp_encapsulation_, const char *network_);
 
         //  i_engine interface implementation.
-        void plug (xs::io_thread_t *io_thread_,
+        void plug (xs::poller_base_t *io_thread_,
             xs::session_base_t *session_);
         void unplug ();
         void terminate ();
diff --git a/src/poller_base.cpp b/src/poller_base.cpp
index 83e0301..913a8d7 100644
--- a/src/poller_base.cpp
+++ b/src/poller_base.cpp
@@ -27,44 +27,57 @@
 #include "devpoll.hpp"
 #include "kqueue.hpp"
 
-xs::poller_base_t *xs::poller_base_t::create ()
+xs::poller_base_t *xs::poller_base_t::create (xs::ctx_t *ctx_, uint32_t tid_)
 {
     poller_base_t *result;
 #if defined XS_HAVE_SELECT
-    result = new (std::nothrow) select_t;
+    result = new (std::nothrow) select_t (ctx_, tid_);
 #elif defined XS_HAVE_POLL
-    result = new (std::nothrow) poll_t;
+    result = new (std::nothrow) poll_t (ctx_, tid_);
 #elif defined XS_HAVE_EPOLL
-    result = new (std::nothrow) epoll_t;
+    result = new (std::nothrow) epoll_t (ctx_, tid_);
 #elif defined XS_HAVE_DEVPOLL
-    result = new (std::nothrow) devpoll_t;
+    result = new (std::nothrow) devpoll_t (ctx_, tid_);
 #elif defined XS_HAVE_KQUEUE
-    result = new (std::nothrow) kqueue_t;
+    result = new (std::nothrow) kqueue_t (ctx_, tid_);
 #endif
     alloc_assert (result);
     return result;
 }
 
-xs::poller_base_t::poller_base_t ()
+xs::poller_base_t::poller_base_t (xs::ctx_t *ctx_, uint32_t tid_) :
+    object_t (ctx_, tid_)
 {
 }
 
 xs::poller_base_t::~poller_base_t ()
 {
-    //  Make sure there is no more load on the shutdown.
-    xs_assert (get_load () == 0);
 }
 
 void xs::poller_base_t::start ()
 {
+    mailbox_handle = add_fd (mailbox.get_fd (), this);
+    set_pollin (mailbox_handle);
     xstart ();
 }
 
 void xs::poller_base_t::stop ()
 {
+    //  Ask the I/O thread to stop.
+    send_stop ();
+}
+
+void xs::poller_base_t::process_stop ()
+{
+    rm_fd (mailbox_handle);
     xstop ();
 }
 
+xs::mailbox_t *xs::poller_base_t::get_mailbox ()
+{
+    return &mailbox;
+}
+
 int xs::poller_base_t::get_load ()
 {
     return load.get ();
@@ -126,3 +139,36 @@ uint64_t xs::poller_base_t::execute_timers ()
     //  There are no more timers.
     return 0;
 }
+
+void xs::poller_base_t::in_event (fd_t fd_)
+{
+    //  TODO: Do we want to limit number of commands I/O thread can
+    //  process in a single go?
+
+    while (true) {
+
+        //  Get the next command. If there is none, exit.
+        command_t cmd;
+        int rc = mailbox.recv (&cmd, 0);
+        if (rc != 0 && errno == EINTR)
+            continue;
+        if (rc != 0 && errno == EAGAIN)
+            break;
+        errno_assert (rc == 0);
+
+        //  Process the command.
+        cmd.destination->process_command (cmd);
+    }
+}
+
+void xs::poller_base_t::out_event (fd_t fd_)
+{
+    //  We are never polling for POLLOUT here. This function is never called.
+    xs_assert (false);
+}
+
+void xs::poller_base_t::timer_event (handle_t handle_)
+{
+    //  No timers here. This function is never called.
+    xs_assert (false);
+}
diff --git a/src/poller_base.hpp b/src/poller_base.hpp
index e63f203..2388fad 100644
--- a/src/poller_base.hpp
+++ b/src/poller_base.hpp
@@ -25,11 +25,15 @@
 
 #include "fd.hpp"
 #include "clock.hpp"
+#include "object.hpp"
+#include "mailbox.hpp"
 #include "atomic_counter.hpp"
 
 namespace xs
 {
 
+    class ctx_t;
+
     //  Handle of a file descriptor within a pollset.
     typedef void* handle_t;
 
@@ -50,12 +54,12 @@ namespace xs
         virtual void timer_event (handle_t handle_) = 0;
     };
 
-    class poller_base_t
+    class poller_base_t : public object_t, public i_poll_events
     {
     public:
 
         //  Create optimal poller mechanism for this environment.
-        static poller_base_t *create ();
+        static poller_base_t *create (xs::ctx_t *ctx_, uint32_t tid_);
 
         virtual ~poller_base_t ();
 
@@ -66,6 +70,9 @@ namespace xs
         void start ();
         void stop ();
 
+        //  Returns mailbox associated with this I/O poller.
+        mailbox_t *get_mailbox ();
+
         virtual handle_t add_fd (fd_t fd_, xs::i_poll_events *events_) = 0;
         virtual void rm_fd (handle_t handle_) = 0;
         virtual void set_pollin (handle_t handle_) = 0;
@@ -82,9 +89,14 @@ namespace xs
         //  Cancel the timer identified by the handle.
         void rm_timer (handle_t handle_);
 
+        //  i_poll_events implementation.
+        void in_event (fd_t fd_);
+        void out_event (fd_t fd_);
+        void timer_event (handle_t handle_);
+
     protected:
 
-        poller_base_t ();
+        poller_base_t (xs::ctx_t *ctx_, uint32_t tid_);
 
         //  Called by individual poller implementations to manage the load.
         void adjust_load (int amount_);
@@ -95,6 +107,8 @@ namespace xs
 
     private:
 
+        void process_stop ();
+
         //  Clock instance private to this I/O thread.
         clock_t clock;
 
@@ -111,6 +125,12 @@ namespace xs
         //  registered.
         atomic_counter_t load;
 
+        //  I/O thread accesses incoming commands via this mailbox.
+        mailbox_t mailbox;
+
+        //  Handle associated with mailbox' file descriptor.
+        handle_t mailbox_handle;
+
         poller_base_t (const poller_base_t&);
         const poller_base_t &operator = (const poller_base_t&);
     };
diff --git a/src/pub.cpp b/src/pub.cpp
index aa0ed8e..db285b8 100644
--- a/src/pub.cpp
+++ b/src/pub.cpp
@@ -44,7 +44,7 @@ bool xs::pub_t::xhas_in ()
     return false;
 }
 
-xs::pub_session_t::pub_session_t (io_thread_t *io_thread_, bool connect_,
+xs::pub_session_t::pub_session_t (poller_base_t *io_thread_, bool connect_,
       socket_base_t *socket_, const options_t &options_,
       const char *protocol_, const char *address_) :
     xpub_session_t (io_thread_, connect_, socket_, options_, protocol_,
diff --git a/src/pub.hpp b/src/pub.hpp
index 49aecc0..45b510e 100644
--- a/src/pub.hpp
+++ b/src/pub.hpp
@@ -28,7 +28,7 @@ namespace xs
 {
 
     class ctx_t;
-    class io_thread_t;
+    class poller_base_t;
     class socket_base_t;
     class msg_t;
 
@@ -53,7 +53,7 @@ namespace xs
     {
     public:
 
-        pub_session_t (xs::io_thread_t *io_thread_, bool connect_,
+        pub_session_t (xs::poller_base_t *io_thread_, bool connect_,
             xs::socket_base_t *socket_, const options_t &options_,
             const char *protocol_, const char *address_);
         ~pub_session_t ();
diff --git a/src/pull.cpp b/src/pull.cpp
index e168c03..923c85b 100644
--- a/src/pull.cpp
+++ b/src/pull.cpp
@@ -60,7 +60,7 @@ bool xs::pull_t::xhas_in ()
     return fq.has_in ();
 }
 
-xs::pull_session_t::pull_session_t (io_thread_t *io_thread_, bool connect_,
+xs::pull_session_t::pull_session_t (poller_base_t *io_thread_, bool connect_,
       socket_base_t *socket_, const options_t &options_,
       const char *protocol_, const char *address_) :
     session_base_t (io_thread_, connect_, socket_, options_, protocol_,
diff --git a/src/pull.hpp b/src/pull.hpp
index 29ca4d9..a02ef2d 100644
--- a/src/pull.hpp
+++ b/src/pull.hpp
@@ -32,7 +32,7 @@ namespace xs
     class ctx_t;
     class pipe_t;
     class msg_t;
-    class io_thread_t;
+    class poller_base_t;
 
     class pull_t :
         public socket_base_t
@@ -65,7 +65,7 @@ namespace xs
     {
     public:
 
-        pull_session_t (xs::io_thread_t *io_thread_, bool connect_,
+        pull_session_t (xs::poller_base_t *io_thread_, bool connect_,
             socket_base_t *socket_, const options_t &options_,
             const char *protocol_, const char *address_);
         ~pull_session_t ();
diff --git a/src/push.cpp b/src/push.cpp
index 45f0c62..ffc2d6e 100644
--- a/src/push.cpp
+++ b/src/push.cpp
@@ -60,7 +60,7 @@ bool xs::push_t::xhas_out ()
     return lb.has_out ();
 }
 
-xs::push_session_t::push_session_t (io_thread_t *io_thread_, bool connect_,
+xs::push_session_t::push_session_t (poller_base_t *io_thread_, bool connect_,
       socket_base_t *socket_, const options_t &options_,
       const char *protocol_, const char *address_) :
     session_base_t (io_thread_, connect_, socket_, options_, protocol_,
diff --git a/src/push.hpp b/src/push.hpp
index 85c2279..f6102c6 100644
--- a/src/push.hpp
+++ b/src/push.hpp
@@ -32,7 +32,7 @@ namespace xs
     class ctx_t;
     class pipe_t;
     class msg_t;
-    class io_thread_t;
+    class poller_base_t;
 
     class push_t :
         public socket_base_t
@@ -64,7 +64,7 @@ namespace xs
     {
     public:
 
-        push_session_t (xs::io_thread_t *io_thread_, bool connect_,
+        push_session_t (xs::poller_base_t *io_thread_, bool connect_,
             socket_base_t *socket_, const options_t &options_,
             const char *protocol_, const char *address_);
         ~push_session_t ();
diff --git a/src/reaper.cpp b/src/reaper.cpp
index 8dc2da6..ffce0a9 100644
--- a/src/reaper.cpp
+++ b/src/reaper.cpp
@@ -27,7 +27,7 @@ xs::reaper_t::reaper_t (class ctx_t *ctx_, uint32_t tid_) :
     sockets (0),
     terminating (false)
 {
-    poller = poller_base_t::create ();
+    poller = poller_base_t::create (ctx_, tid_);
     xs_assert (poller);
 
     mailbox_handle = poller->add_fd (mailbox.get_fd (), this);
diff --git a/src/rep.cpp b/src/rep.cpp
index 6454b57..5e5e415 100644
--- a/src/rep.cpp
+++ b/src/rep.cpp
@@ -112,7 +112,7 @@ bool xs::rep_t::xhas_out ()
     return xrep_t::xhas_out ();
 }
 
-xs::rep_session_t::rep_session_t (io_thread_t *io_thread_, bool connect_,
+xs::rep_session_t::rep_session_t (poller_base_t *io_thread_, bool connect_,
       socket_base_t *socket_, const options_t &options_,
       const char *protocol_, const char *address_) :
     xrep_session_t (io_thread_, connect_, socket_, options_, protocol_,
diff --git a/src/rep.hpp b/src/rep.hpp
index 0b608ee..2cd3931 100644
--- a/src/rep.hpp
+++ b/src/rep.hpp
@@ -29,7 +29,7 @@ namespace xs
 
     class ctx_t;
     class msg_t;
-    class io_thread_t;
+    class poller_base_t;
     class socket_base_t;
 
     class rep_t : public xrep_t
@@ -64,7 +64,7 @@ namespace xs
     {
     public:
 
-        rep_session_t (xs::io_thread_t *io_thread_, bool connect_,
+        rep_session_t (xs::poller_base_t *io_thread_, bool connect_,
             xs::socket_base_t *socket_, const options_t &options_,
             const char *protocol_, const char *address_);
         ~rep_session_t ();
diff --git a/src/req.cpp b/src/req.cpp
index af6c8cd..6e9867c 100644
--- a/src/req.cpp
+++ b/src/req.cpp
@@ -137,7 +137,7 @@ bool xs::req_t::xhas_out ()
     return xreq_t::xhas_out ();
 }
 
-xs::req_session_t::req_session_t (io_thread_t *io_thread_, bool connect_,
+xs::req_session_t::req_session_t (poller_base_t *io_thread_, bool connect_,
       socket_base_t *socket_, const options_t &options_,
       const char *protocol_, const char *address_) :
     xreq_session_t (io_thread_, connect_, socket_, options_, protocol_,
diff --git a/src/req.hpp b/src/req.hpp
index 46015e8..8da789f 100644
--- a/src/req.hpp
+++ b/src/req.hpp
@@ -31,7 +31,7 @@ namespace xs
 
     class ctx_t;
     class msg_t;
-    class io_thread_t;
+    class poller_base_t;
     class socket_base_t;
 
     class req_t : public xreq_t
@@ -65,7 +65,7 @@ namespace xs
     {
     public:
 
-        req_session_t (xs::io_thread_t *io_thread_, bool connect_,
+        req_session_t (xs::poller_base_t *io_thread_, bool connect_,
             xs::socket_base_t *socket_, const options_t &options_,
             const char *protocol_, const char *address_);
         ~req_session_t ();
diff --git a/src/session_base.cpp b/src/session_base.cpp
index efc4f8b..168501c 100644
--- a/src/session_base.cpp
+++ b/src/session_base.cpp
@@ -43,7 +43,7 @@
 #include "pull.hpp"
 #include "pair.hpp"
 
-xs::session_base_t *xs::session_base_t::create (class io_thread_t *io_thread_,
+xs::session_base_t *xs::session_base_t::create (class poller_base_t *io_thread_,
     bool connect_, class socket_base_t *socket_, const options_t &options_,
     const char *protocol_, const char *address_)
 {
@@ -100,7 +100,7 @@ xs::session_base_t *xs::session_base_t::create (class io_thread_t *io_thread_,
     return s;
 }
 
-xs::session_base_t::session_base_t (class io_thread_t *io_thread_,
+xs::session_base_t::session_base_t (class poller_base_t *io_thread_,
       bool connect_, class socket_base_t *socket_, const options_t &options_,
       const char *protocol_, const char *address_) :
     own_t (io_thread_, options_),
@@ -387,7 +387,7 @@ void xs::session_base_t::start_connecting (bool wait_)
 
     //  Choose I/O thread to run connecter in. Given that we are already
     //  running in an I/O thread, there must be at least one available.
-    io_thread_t *io_thread = choose_io_thread (options.affinity);
+    poller_base_t *io_thread = choose_io_thread (options.affinity);
     xs_assert (io_thread);
 
     //  Create the connecter object.
diff --git a/src/session_base.hpp b/src/session_base.hpp
index 45ad8f8..6a461d8 100644
--- a/src/session_base.hpp
+++ b/src/session_base.hpp
@@ -33,7 +33,7 @@ namespace xs
 {
 
     class pipe_t;
-    class io_thread_t;
+    class poller_base_t;
     class socket_base_t;
     struct i_engine;
 
@@ -45,7 +45,7 @@ namespace xs
     public:
 
         //  Create a session of the particular type.
-        static session_base_t *create (xs::io_thread_t *io_thread_,
+        static session_base_t *create (xs::poller_base_t *io_thread_,
             bool connect_, xs::socket_base_t *socket_,
             const options_t &options_, const char *protocol_,
             const char *address_);
@@ -67,7 +67,7 @@ namespace xs
 
     protected:
 
-        session_base_t (xs::io_thread_t *io_thread_, bool connect_,
+        session_base_t (xs::poller_base_t *io_thread_, bool connect_,
             xs::socket_base_t *socket_, const options_t &options_,
             const char *protocol_, const char *address_);
         ~session_base_t ();
@@ -116,7 +116,7 @@ namespace xs
 
         //  I/O thread the session is living in. It will be used to plug in
         //  the engines into the same thread.
-        xs::io_thread_t *io_thread;
+        xs::poller_base_t *io_thread;
 
         //  If true, identity is to be sent/recvd from the network.
         bool send_identity;
diff --git a/src/socket_base.cpp b/src/socket_base.cpp
index 97ac4b4..492d027 100644
--- a/src/socket_base.cpp
+++ b/src/socket_base.cpp
@@ -39,7 +39,7 @@
 #include "tcp_listener.hpp"
 #include "ipc_listener.hpp"
 #include "tcp_connecter.hpp"
-#include "io_thread.hpp"
+#include "poller_base.hpp"
 #include "session_base.hpp"
 #include "config.hpp"
 #include "clock.hpp"
@@ -338,7 +338,7 @@ int xs::socket_base_t::bind (const char *addr_)
 
     //  Remaining trasnports require to be run in an I/O thread, so at this
     //  point we'll choose one.
-    io_thread_t *io_thread = choose_io_thread (options.affinity);
+    poller_base_t *io_thread = choose_io_thread (options.affinity);
     if (!io_thread) {
         errno = EMTHREAD;
         return -1;
@@ -449,7 +449,7 @@ int xs::socket_base_t::connect (const char *addr_)
     }
 
     //  Choose the I/O thread to run the session in.
-    io_thread_t *io_thread = choose_io_thread (options.affinity);
+    poller_base_t *io_thread = choose_io_thread (options.affinity);
     if (!io_thread) {
         errno = EMTHREAD;
         return -1;
diff --git a/src/stream_engine.cpp b/src/stream_engine.cpp
index fb49152..7d69848 100644
--- a/src/stream_engine.cpp
+++ b/src/stream_engine.cpp
@@ -36,7 +36,7 @@
 #include <new>
 
 #include "stream_engine.hpp"
-#include "io_thread.hpp"
+#include "poller_base.hpp"
 #include "session_base.hpp"
 #include "config.hpp"
 #include "err.hpp"
@@ -103,7 +103,7 @@ xs::stream_engine_t::~stream_engine_t ()
     }
 }
 
-void xs::stream_engine_t::plug (io_thread_t *io_thread_,
+void xs::stream_engine_t::plug (poller_base_t *io_thread_,
     session_base_t *session_)
 {
     xs_assert (!plugged);
diff --git a/src/stream_engine.hpp b/src/stream_engine.hpp
index f246552..9060cd7 100644
--- a/src/stream_engine.hpp
+++ b/src/stream_engine.hpp
@@ -34,7 +34,7 @@
 namespace xs
 {
 
-    class io_thread_t;
+    class poller_base_t;
     class session_base_t;
 
     //  This engine handles any socket with SOCK_STREAM semantics,
@@ -48,7 +48,7 @@ namespace xs
         ~stream_engine_t ();
 
         //  i_engine interface implementation.
-        void plug (xs::io_thread_t *io_thread_,
+        void plug (xs::poller_base_t *io_thread_,
            xs::session_base_t *session_);
         void unplug ();
         void terminate ();
diff --git a/src/sub.cpp b/src/sub.cpp
index ad55ea3..08ce368 100644
--- a/src/sub.cpp
+++ b/src/sub.cpp
@@ -80,7 +80,7 @@ bool xs::sub_t::xhas_out ()
     return false;
 }
 
-xs::sub_session_t::sub_session_t (io_thread_t *io_thread_, bool connect_,
+xs::sub_session_t::sub_session_t (poller_base_t *io_thread_, bool connect_,
       socket_base_t *socket_, const options_t &options_,
       const char *protocol_, const char *address_) :
     xsub_session_t (io_thread_, connect_, socket_, options_, protocol_,
diff --git a/src/sub.hpp b/src/sub.hpp
index 70ade88..9b38822 100644
--- a/src/sub.hpp
+++ b/src/sub.hpp
@@ -29,7 +29,7 @@ namespace xs
 
     class ctx_t;
     class msg_t;
-    class io_thread_t;
+    class poller_base_t;
     class socket_base_t;
 
     class sub_t : public xsub_t
@@ -55,7 +55,7 @@ namespace xs
     {
     public:
 
-        sub_session_t (xs::io_thread_t *io_thread_, bool connect_,
+        sub_session_t (xs::poller_base_t *io_thread_, bool connect_,
             xs::socket_base_t *socket_, const options_t &options_,
             const char *protocol_, const char *address_);
         ~sub_session_t ();
diff --git a/src/tcp_connecter.cpp b/src/tcp_connecter.cpp
index 5007570..6816e64 100644
--- a/src/tcp_connecter.cpp
+++ b/src/tcp_connecter.cpp
@@ -24,7 +24,7 @@
 
 #include "tcp_connecter.hpp"
 #include "stream_engine.hpp"
-#include "io_thread.hpp"
+#include "poller_base.hpp"
 #include "platform.hpp"
 #include "random.hpp"
 #include "err.hpp"
@@ -46,7 +46,7 @@
 #endif
 #endif
 
-xs::tcp_connecter_t::tcp_connecter_t (class io_thread_t *io_thread_,
+xs::tcp_connecter_t::tcp_connecter_t (class poller_base_t *io_thread_,
       class session_base_t *session_, const options_t &options_,
       const char *address_, bool wait_) :
     own_t (io_thread_, options_),
diff --git a/src/tcp_connecter.hpp b/src/tcp_connecter.hpp
index 7164fa2..2b946a1 100644
--- a/src/tcp_connecter.hpp
+++ b/src/tcp_connecter.hpp
@@ -31,7 +31,7 @@
 namespace xs
 {
 
-    class io_thread_t;
+    class poller_base_t;
     class session_base_t;
 
     class tcp_connecter_t : public own_t, public io_object_t
@@ -40,7 +40,7 @@ namespace xs
 
         //  If 'delay' is true connecter first waits for a while, then starts
         //  connection process.
-        tcp_connecter_t (xs::io_thread_t *io_thread_,
+        tcp_connecter_t (xs::poller_base_t *io_thread_,
             xs::session_base_t *session_, const options_t &options_,
             const char *address_, bool delay_);
         ~tcp_connecter_t ();
diff --git a/src/tcp_listener.cpp b/src/tcp_listener.cpp
index 31a9149..e8900fc 100644
--- a/src/tcp_listener.cpp
+++ b/src/tcp_listener.cpp
@@ -26,7 +26,7 @@
 #include "platform.hpp"
 #include "tcp_listener.hpp"
 #include "stream_engine.hpp"
-#include "io_thread.hpp"
+#include "poller_base.hpp"
 #include "session_base.hpp"
 #include "config.hpp"
 #include "err.hpp"
@@ -48,7 +48,7 @@
 #include <ioctl.h>
 #endif
 
-xs::tcp_listener_t::tcp_listener_t (io_thread_t *io_thread_,
+xs::tcp_listener_t::tcp_listener_t (poller_base_t *io_thread_,
       socket_base_t *socket_, const options_t &options_) :
     own_t (io_thread_, options_),
     io_object_t (io_thread_),
@@ -94,7 +94,7 @@ void xs::tcp_listener_t::in_event (fd_t fd_)
 
     //  Choose I/O thread to run connecter in. Given that we are already
     //  running in an I/O thread, there must be at least one available.
-    io_thread_t *io_thread = choose_io_thread (options.affinity);
+    poller_base_t *io_thread = choose_io_thread (options.affinity);
     xs_assert (io_thread);
 
     //  Create and launch a session object. 
diff --git a/src/tcp_listener.hpp b/src/tcp_listener.hpp
index 153f771..22c000f 100644
--- a/src/tcp_listener.hpp
+++ b/src/tcp_listener.hpp
@@ -31,14 +31,14 @@
 namespace xs
 {
 
-    class io_thread_t;
+    class poller_base_t;
     class socket_base_t;
 
     class tcp_listener_t : public own_t, public io_object_t
     {
     public:
 
-        tcp_listener_t (xs::io_thread_t *io_thread_,
+        tcp_listener_t (xs::poller_base_t *io_thread_,
             xs::socket_base_t *socket_, const options_t &options_);
         ~tcp_listener_t ();
 
diff --git a/src/xpub.cpp b/src/xpub.cpp
index d27337e..6c0f75b 100644
--- a/src/xpub.cpp
+++ b/src/xpub.cpp
@@ -182,7 +182,7 @@ void xs::xpub_t::send_unsubscription (unsigned char *data_, size_t size_,
     }
 }
 
-xs::xpub_session_t::xpub_session_t (io_thread_t *io_thread_, bool connect_,
+xs::xpub_session_t::xpub_session_t (poller_base_t *io_thread_, bool connect_,
       socket_base_t *socket_, const options_t &options_,
       const char *protocol_, const char *address_) :
     session_base_t (io_thread_, connect_, socket_, options_, protocol_,
diff --git a/src/xpub.hpp b/src/xpub.hpp
index a26076e..2c61c63 100644
--- a/src/xpub.hpp
+++ b/src/xpub.hpp
@@ -37,7 +37,7 @@ namespace xs
     class ctx_t;
     class msg_t;
     class pipe_t;
-    class io_thread_t;
+    class poller_base_t;
 
     class xpub_t :
         public socket_base_t
@@ -89,7 +89,7 @@ namespace xs
     {
     public:
 
-        xpub_session_t (xs::io_thread_t *io_thread_, bool connect_,
+        xpub_session_t (xs::poller_base_t *io_thread_, bool connect_,
             socket_base_t *socket_, const options_t &options_,
             const char *protocol_, const char *address_);
         ~xpub_session_t ();
diff --git a/src/xrep.cpp b/src/xrep.cpp
index 051159d..4bae2d9 100644
--- a/src/xrep.cpp
+++ b/src/xrep.cpp
@@ -307,7 +307,7 @@ bool xs::xrep_t::xhas_out ()
     return true;
 }
 
-xs::xrep_session_t::xrep_session_t (io_thread_t *io_thread_, bool connect_,
+xs::xrep_session_t::xrep_session_t (poller_base_t *io_thread_, bool connect_,
       socket_base_t *socket_, const options_t &options_,
       const char *protocol_, const char *address_) :
     session_base_t (io_thread_, connect_, socket_, options_, protocol_,
diff --git a/src/xrep.hpp b/src/xrep.hpp
index d8723db..e5ab895 100644
--- a/src/xrep.hpp
+++ b/src/xrep.hpp
@@ -37,6 +37,7 @@ namespace xs
 
     class ctx_t;
     class pipe_t;
+    class poller_base_t;
 
     //  TODO: This class uses O(n) scheduling. Rewrite it to use O(1) algorithm.
     class xrep_t :
@@ -108,7 +109,7 @@ namespace xs
     {
     public:
 
-        xrep_session_t (xs::io_thread_t *io_thread_, bool connect_,
+        xrep_session_t (xs::poller_base_t *io_thread_, bool connect_,
             socket_base_t *socket_, const options_t &options_,
             const char *protocol_, const char *address_);
         ~xrep_session_t ();
diff --git a/src/xreq.cpp b/src/xreq.cpp
index d16f60e..948ff22 100644
--- a/src/xreq.cpp
+++ b/src/xreq.cpp
@@ -115,7 +115,7 @@ void xs::xreq_t::xterminated (pipe_t *pipe_)
     lb.terminated (pipe_);
 }
 
-xs::xreq_session_t::xreq_session_t (io_thread_t *io_thread_, bool connect_,
+xs::xreq_session_t::xreq_session_t (poller_base_t *io_thread_, bool connect_,
       socket_base_t *socket_, const options_t &options_,
       const char *protocol_, const char *address_) :
     session_base_t (io_thread_, connect_, socket_, options_, protocol_,
diff --git a/src/xreq.hpp b/src/xreq.hpp
index ab3a360..358b10a 100644
--- a/src/xreq.hpp
+++ b/src/xreq.hpp
@@ -32,7 +32,7 @@ namespace xs
     class ctx_t;
     class msg_t;
     class pipe_t;
-    class io_thread_t;
+    class poller_base_t;
     class socket_base_t;
 
     class xreq_t :
@@ -76,7 +76,7 @@ namespace xs
     {
     public:
 
-        xreq_session_t (xs::io_thread_t *io_thread_, bool connect_,
+        xreq_session_t (xs::poller_base_t *io_thread_, bool connect_,
             xs::socket_base_t *socket_, const options_t &options_,
             const char *protocol_, const char *address_);
         ~xreq_session_t ();
diff --git a/src/xsub.cpp b/src/xsub.cpp
index 80cc205..ee53b2c 100644
--- a/src/xsub.cpp
+++ b/src/xsub.cpp
@@ -221,7 +221,7 @@ void xs::xsub_t::send_subscription (unsigned char *data_, size_t size_,
     xs_assert (sent);
 }
 
-xs::xsub_session_t::xsub_session_t (io_thread_t *io_thread_, bool connect_,
+xs::xsub_session_t::xsub_session_t (poller_base_t *io_thread_, bool connect_,
       socket_base_t *socket_, const options_t &options_,
       const char *protocol_, const char *address_) :
     session_base_t (io_thread_, connect_, socket_, options_, protocol_,
diff --git a/src/xsub.hpp b/src/xsub.hpp
index f9e5210..8769706 100644
--- a/src/xsub.hpp
+++ b/src/xsub.hpp
@@ -32,7 +32,7 @@ namespace xs
 
     class ctx_t;
     class pipe_t;
-    class io_thread_t;
+    class poller_base_t;
 
     class xsub_t :
         public socket_base_t
@@ -91,7 +91,7 @@ namespace xs
     {
     public:
 
-        xsub_session_t (class io_thread_t *io_thread_, bool connect_,
+        xsub_session_t (class poller_base_t *io_thread_, bool connect_,
             socket_base_t *socket_, const options_t &options_,
             const char *protocol_, const char *address_);
         ~xsub_session_t ();