Repository: pika/pika

Issue:
Backport #956 for 0.12

Discussion:

Patch:
diff --git a/docs/faq.rst b/docs/faq.rst
index 132b495..f70ef55 100644
--- a/docs/faq.rst
+++ b/docs/faq.rst
@@ -3,7 +3,7 @@ Frequently Asked Questions
 
 - Is Pika thread safe?
 
-    Pika does not have any notion of threading in the code. If you want to use Pika with threading, make sure you have a Pika connection per thread, created in that thread. It is not safe to share one Pika connection across threads.
+    Pika does not have any notion of threading in the code. If you want to use Pika with threading, make sure you have a Pika connection per thread, created in that thread. It is not safe to share one Pika connection across threads, with one exception: you may call the connection method `add_callback_threadsafe` from another thread to schedule a callback within an active pika connection.
 
 - How do I report a bug with Pika?
 
diff --git a/pika/adapters/asyncio_connection.py b/pika/adapters/asyncio_connection.py
index b2ac570..57cb2d4 100644
--- a/pika/adapters/asyncio_connection.py
+++ b/pika/adapters/asyncio_connection.py
@@ -20,6 +20,17 @@ class IOLoopAdapter:
         self.readers = set()
         self.writers = set()
 
+    def close(self):
+        """Release ioloop's resources.
+
+        This method is intended to be called by the application or test code
+        only after the ioloop's outermost `start()` call returns. After calling
+        `close()`, no other interaction with the closed instance of ioloop
+        should be performed.
+
+        """
+        self.loop.close()
+
     def add_timeout(self, deadline, callback_method):
         """Add the callback_method to the EventLoop timer to fire after deadline
         seconds. Returns a Handle to the timeout.
@@ -41,6 +52,28 @@ class IOLoopAdapter:
         """
         return handle.cancel()
 
+    def add_callback_threadsafe(self, callback):
+        """Requests a call to the given function as soon as possible in the
+        context of this IOLoop's thread.
+
+        NOTE: This is the only thread-safe method offered by the IOLoop adapter.
+         All other manipulations of the IOLoop adapter and its parent connection
+         must be performed from the connection's thread.
+
+        For example, a thread may request a call to the
+        `channel.basic_ack` method of a connection that is running in a
+        different thread via
+
+        ```
+        connection.add_callback_threadsafe(
+            functools.partial(channel.basic_ack, delivery_tag=...))
+        ```
+
+        :param method callback: The callback method; must be callable.
+
+        """
+        self.loop.call_soon_threadsafe(callback)
+
     def add_handler(self, fd, cb, event_state):
         """ Registers the given handler to receive the given events for ``fd``.
 
diff --git a/pika/adapters/base_connection.py b/pika/adapters/base_connection.py
index d6e8407..cc3836d 100644
--- a/pika/adapters/base_connection.py
+++ b/pika/adapters/base_connection.py
@@ -135,6 +135,32 @@ class BaseConnection(connection.Connection):
         """
         self.ioloop.remove_timeout(timeout_id)
 
+    def add_callback_threadsafe(self, callback):
+        """Requests a call to the given function as soon as possible in the
+        context of this connection's IOLoop thread.
+
+        NOTE: This is the only thread-safe method offered by the connection. All
+         other manipulations of the connection must be performed from the
+         connection's thread.
+
+        For example, a thread may request a call to the
+        `channel.basic_ack` method of a connection that is running in a
+        different thread via
+
+        ```
+        connection.add_callback_threadsafe(
+            functools.partial(channel.basic_ack, delivery_tag=...))
+        ```
+
+        :param method callback: The callback method; must be callable.
+
+        """
+        if not callable(callback):
+            raise TypeError(
+                'callback must be a callable, but got %r' % (callback,))
+
+        self.ioloop.add_callback_threadsafe(callback)
+
     def _adapter_connect(self):
         """Connect to the RabbitMQ broker, returning True if connected.
 
diff --git a/pika/adapters/blocking_connection.py b/pika/adapters/blocking_connection.py
index 4f7751f..29701c2 100644
--- a/pika/adapters/blocking_connection.py
+++ b/pika/adapters/blocking_connection.py
@@ -10,6 +10,9 @@ and the :class:`~pika.adapters.blocking_connection.BlockingChannel`
 classes.
 
 """
+# Suppress too-many-lines
+# pylint: disable=C0302
+
 # Disable "access to protected member warnings: this wrapper implementation is
 # a friend of those instances
 # pylint: disable=W0212
@@ -155,7 +158,7 @@ class _CallbackResult(object):
                                 with `append_element`
         """
         assert self._ready, '_CallbackResult was not set'
-        assert isinstance(self._values, list) and len(self._values) > 0, (
+        assert isinstance(self._values, list) and self._values, (
             '_CallbackResult value is incompatible with append_element: %r'
             % (self._values,))
 
@@ -378,7 +381,7 @@ class BlockingConnection(object):
 
     def _cleanup(self):
         """Clean up members that might inhibit garbage collection"""
-        self._impl.ioloop.deactivate_poller()
+        self._impl.ioloop.close()
         self._ready_events.clear()
         self._opened_result.reset()
         self._open_error_result.reset()
@@ -525,6 +528,18 @@ class BlockingConnection(object):
         """
         self._ready_events.append(evt)
 
+    def _on_threadsafe_callback(self, user_callback):
+        """Handle callback that was registered via `add_callback_threadsafe`.
+
+        :param user_callback: callback passed to `add_callback_threadsafe` by
+                              the application.
+
+        """
+        # Turn it into a 0-delay timeout to take advantage of our existing logic
+        # that deals with reentrancy
+        self.add_timeout(0, user_callback)
+
+
     def _on_connection_blocked(self, user_callback, method_frame):
         """Handle Connection.Blocked notification from RabbitMQ broker
 
@@ -632,6 +647,29 @@ class BlockingConnection(object):
 
         return timer_id
 
+    def add_callback_threadsafe(self, callback):
+        """Requests a call to the given function as soon as possible in the
+        context of this connection's thread.
+
+        NOTE: This is the only thread-safe method in `BlockingConnection`. All
+         other manipulations of `BlockingConnection` must be performed from the
+         connection's thread.
+
+        For example, a thread may request a call to the
+        `BlockingChannel.basic_ack` method of a `BlockingConnection` that is
+        running in a different thread via
+
+        ```
+        connection.add_callback_threadsafe(
+            functools.partial(channel.basic_ack, delivery_tag=...))
+        ```
+
+        :param method callback: The callback method; must be callable
+
+        """
+        self._impl.add_callback_threadsafe(
+            functools.partial(self._on_threadsafe_callback, callback))
+
     def remove_timeout(self, timeout_id):
         """Remove a timer if it's still in the timeout stack
 
@@ -874,7 +912,7 @@ class _ConsumerCancellationEvt(_ChannelPendingEvt):
     `Basic.Cancel`
     """
 
-    __slots__ = 'method_frame'
+    __slots__ = ('method_frame',)
 
     def __init__(self, method_frame):
         """
@@ -1798,7 +1836,8 @@ class BlockingChannel(object):
         """Blocking consumption of a queue instead of via a callback. This
         method is a generator that yields each message as a tuple of method,
         properties, and body. The active generator iterator terminates when the
-        consumer is cancelled by client or broker.
+        consumer is cancelled by client via `BlockingChannel.cancel()` or by
+        broker.
 
         Example:
 
@@ -2398,7 +2437,8 @@ class BlockingChannel(object):
         :param queue: The queue name
         :type queue: str or unicode; if empty string, the broker will create a
           unique queue name;
-        :param bool passive: Only check to see if the queue exists
+        :param bool passive: Only check to see if the queue exists and raise
+          `ChannelClosed` if it doesn't;
         :param bool durable: Survive reboots of the broker
         :param bool exclusive: Only allow access by the current connection
         :param bool auto_delete: Delete after consumer cancels or disconnects
diff --git a/pika/adapters/select_connection.py b/pika/adapters/select_connection.py
index d199139..f6970d5 100644
--- a/pika/adapters/select_connection.py
+++ b/pika/adapters/select_connection.py
@@ -3,16 +3,15 @@ platform pika is running on.
 
 """
 import abc
-import os
+import collections
+import errno
+import functools
+import heapq
 import logging
-import socket
 import select
-import errno
 import time
 import threading
 
-from collections import defaultdict
-
 import pika.compat
 
 from pika.adapters.base_connection import BaseConnection
@@ -111,6 +110,173 @@ class SelectConnection(BaseConnection):
         super(SelectConnection, self)._adapter_disconnect()
 
 
+@functools.total_ordering
+class _Timeout(object):
+    """Represents a timeout"""
+
+    __slots__ = ('deadline', 'callback',)
+
+    def __init__(self, deadline, callback):
+        """
+        :param float deadline: timer expiration as non-negative epoch number
+        :param callable callback: callback to call when timeout expires
+        :raises ValueError, TypeError:
+        """
+
+        if deadline < 0:
+            raise ValueError(
+                'deadline must be non-negative epoch number, but got %r' %
+                (deadline,))
+
+        if not callable(callback):
+            raise TypeError(
+                'callback must be a callable, but got %r' % (callback,))
+
+        self.deadline = deadline
+        self.callback = callback
+
+    def __eq__(self, other):
+        """NOTE: not supporting sort stability"""
+        return self.deadline == other.deadline
+
+    def __lt__(self, other):
+        """NOTE: not supporting sort stability"""
+        return self.deadline < other.deadline
+
+    def __le__(self, other):
+        """NOTE: not supporting sort stability"""
+        return self.deadline <= other.deadline
+
+
+class _Timer(object):
+    """Manage timeouts for use in ioloop"""
+
+    # Cancellation count threshold for triggering garbage collection of
+    # cancelled timers
+    _GC_CANCELLATION_THRESHOLD = 1024
+
+    def __init__(self):
+        self._timeout_heap = []
+
+        # Number of canceled timeouts on heap; for scheduling garbage
+        # collection of canceled timeouts
+        self._num_cancellations = 0
+
+    def close(self):
+        """Release resources. Don't use the `_Timer` instance after closing
+        it
+        """
+        # Eliminate potential reference cycles to aid garbage-collection
+        if self._timeout_heap is not None:
+            for timeout in self._timeout_heap:
+                timeout.callback = None
+            self._timeout_heap = None
+
+    def call_later(self, delay, callback):
+        """Schedule a one-shot timeout given delay seconds.
+
+        NOTE: you may cancel the timer before dispatch of the callback. Timer
+            Manager cancels the timer upon dispatch of the callback.
+
+        :param float delay: Non-negative number of seconds from now until
+                            expiration
+        :param method callback: The callback method, having the signature
+                                `callback()`
+
+        :rtype: _Timeout
+        :raises ValueError, TypeError
+
+        """
+        if delay < 0:
+            raise ValueError(
+                'call_later: delay must be non-negative, but got %r'
+                % (delay,))
+
+        now = time.time()
+
+        timeout = _Timeout(now + delay, callback)
+
+        heapq.heappush(self._timeout_heap, timeout)
+
+        LOGGER.debug('call_later: added timeout %r with deadline=%r and '
+                     'callback=%r; now=%s; delay=%s', timeout, timeout.deadline,
+                     timeout.callback, now, delay)
+
+        return timeout
+
+    def remove_timeout(self, timeout):
+        """Cancel the timeout
+
+        :param _Timeout timeout: The timer to cancel
+
+        """
+        # NOTE removing from the heap is difficult, so we just deactivate the
+        # timeout and garbage-collect it at a later time; see discussion
+        # in http://docs.python.org/library/heapq.html
+        if timeout.callback is None:
+            LOGGER.warning(
+                'remove_timeout: timeout was already removed or called %r',
+                timeout)
+        else:
+            LOGGER.debug('remove_timeout: removing timeout %r with deadline=%r '
+                         'and callback=%r', timeout, timeout.deadline,
+                         timeout.callback)
+            timeout.callback = None
+            self._num_cancellations += 1
+
+    def get_remaining_interval(self):
+        """Get the interval to the next timeout expiration
+
+        :returns: non-negative number of seconds until next timer expiration;
+                  None if there are no timers
+        :rtype: float
+
+        """
+        if self._timeout_heap:
+            interval = max(0, self._timeout_heap[0].deadline - time.time())
+        else:
+            interval = None
+
+        return interval
+
+    def process_timeouts(self):
+        """Process pending timeouts, invoking callbacks for those whose time has
+        come
+
+        """
+        if self._timeout_heap:
+            now = time.time()
+
+            # Remove ready timeouts from the heap now to prevent IO starvation
+            # from timeouts added during callback processing
+            ready_timeouts = []
+
+            while self._timeout_heap and self._timeout_heap[0].deadline <= now:
+                timeout = heapq.heappop(self._timeout_heap)
+                if timeout.callback is not None:
+                    ready_timeouts.append(timeout)
+                else:
+                    self._num_cancellations -= 1
+
+            # Invoke ready timeout callbacks
+            for timeout in ready_timeouts:
+                if timeout.callback is None:
+                    # Must have been canceled from a prior callback
+                    self._num_cancellations -= 1
+                    continue
+
+                timeout.callback()
+                timeout.callback = None
+
+            # Garbage-collect canceled timeouts if they exceed threshold
+            if (self._num_cancellations >= self._GC_CANCELLATION_THRESHOLD and
+                    self._num_cancellations > (len(self._timeout_heap) >> 1)):
+                self._num_cancellations = 0
+                self._timeout_heap = [t for t in self._timeout_heap
+                                      if t.callback is not None]
+                heapq.heapify(self._timeout_heap)
+
+
 class IOLoop(object):
     """Singleton wrapper that decides which type of poller to use, creates an
     instance of it in start_poller and keeps the invoking application in a
@@ -123,33 +289,67 @@ class IOLoop(object):
     """
 
     def __init__(self):
-        self._poller = self._get_poller()
+        self._timer = _Timer()
+
+        # Callbacks requested via `add_callback`
+        self._callbacks = collections.deque()
+
+        # Identity of this IOLoop's thread
+        self._thread_id = None
+
+        self._poller = self._get_poller(self._get_remaining_interval,
+                                        self.process_timeouts)
+
+    def close(self):
+        """Release IOLoop's resources.
+
+        `IOLoop.close` is intended to be called by the application or test code
+        only after `IOLoop.start()` returns. After calling `close()`, no other
+        interaction with the closed instance of `IOLoop` should be performed.
+
+        """
+        if self._callbacks is not None:
+            self._poller.close()
+            self._timer.close()
+            self._callbacks = None
 
     @staticmethod
-    def _get_poller():
-        """Determine the best poller to use for this enviroment."""
+    def _get_poller(get_wait_seconds, process_timeouts):
+        """Determine the best poller to use for this environment and instantiate
+        it.
+
+        :param get_wait_seconds: Function for getting the maximum number of
+                                 seconds to wait for IO for use by the poller
+        :param process_timeouts: Function for processing timeouts for use by the
+                                 poller
+
+        :returns: the instantiated poller instance supporting `_PollerBase` API
+        """
 
         poller = None
 
+        kwargs = dict(get_wait_seconds=get_wait_seconds,
+                      process_timeouts=process_timeouts)
+
         if hasattr(select, 'epoll'):
             if not SELECT_TYPE or SELECT_TYPE == 'epoll':
                 LOGGER.debug('Using EPollPoller')
-                poller = EPollPoller()
+                poller = EPollPoller(**kwargs)
 
         if not poller and hasattr(select, 'kqueue'):
             if not SELECT_TYPE or SELECT_TYPE == 'kqueue':
                 LOGGER.debug('Using KQueuePoller')
-                poller = KQueuePoller()
+                poller = KQueuePoller(**kwargs)
 
         if (not poller and hasattr(select, 'poll') and
                 hasattr(select.poll(), 'modify')):  # pylint: disable=E1101
             if not SELECT_TYPE or SELECT_TYPE == 'poll':
                 LOGGER.debug('Using PollPoller')
-                poller = PollPoller()
+                poller = PollPoller(**kwargs)
 
         if not poller:
             LOGGER.debug('Using SelectPoller')
-            poller = SelectPoller()
+            poller = SelectPoller(**kwargs)
 
         return poller
 
@@ -164,7 +364,7 @@ class IOLoop(object):
         :rtype: str
 
         """
-        return self._poller.add_timeout(deadline, callback_method)
+        return self._timer.call_later(deadline, callback_method)
 
     def remove_timeout(self, timeout_id):
         """[API] Remove a timeout
@@ -172,7 +372,58 @@ class IOLoop(object):
         :param str timeout_id: The timeout id to remove
 
         """
-        self._poller.remove_timeout(timeout_id)
+        self._timer.remove_timeout(timeout_id)
+
+    def add_callback_threadsafe(self, callback):
+        """Requests a call to the given function as soon as possible in the
+        context of this IOLoop's thread.
+
+        NOTE: This is the only thread-safe method in IOLoop. All other
+        manipulations of IOLoop must be performed from the IOLoop's thread.
+
+        For example, a thread may request a call to the `stop` method of an
+        ioloop that is running in a different thread via
+        `ioloop.add_callback_threadsafe(ioloop.stop)`
+
+        :param method callback: The callback method
+
+        """
+        if not callable(callback):
+            raise TypeError(
+                'callback must be a callable, but got %r' % (callback,))
+
+        # NOTE: `deque.append` is atomic
+        self._callbacks.append(callback)
+        if threading.current_thread().ident != self._thread_id:
+            # Wake up the IOLoop running in another thread
+            self._poller.wake_threadsafe()
+
+        LOGGER.debug('add_callback_threadsafe: added callback=%r', callback)
+
+    def process_timeouts(self):
+        """[Extension] Process pending callbacks and timeouts, invoking those
+        whose time has come. Internal use only.
+
+        """
+        # Avoid I/O starvation by postponing new callbacks to the next iteration
+        for _ in pika.compat.xrange(len(self._callbacks)):
+            self._callbacks.popleft()()
+
+        self._timer.process_timeouts()
+
+    def _get_remaining_interval(self):
+        """Get the remaining interval to the next callback or timeout
+        expiration.
+
+        :returns: non-negative number of seconds until next callback or timer
+                  expiration; None if there are no callbacks and timers
+        :rtype: float
+
+        """
+        if self._callbacks:
+            return 0
+
+        return self._timer.get_remaining_interval()
 
     def add_handler(self, fileno, handler, events):
         """[API] Add a new fileno to the set to be monitored
@@ -206,27 +457,31 @@ class IOLoop(object):
         exit. See `IOLoop.stop`.
 
         """
+        self._thread_id = threading.current_thread().ident
         self._poller.start()
 
     def stop(self):
         """[API] Request exit from the ioloop. The loop is NOT guaranteed to
-        stop before this method returns. This is the only method that may be
-        called from another thread.
+        stop before this method returns.
 
-        """
-        self._poller.stop()
+        To invoke `stop()` safely from a thread other than this IOLoop's thread,
+        call it via `add_callback_threadsafe`; e.g.,
 
-    def process_timeouts(self):
-        """[Extension] Process pending timeouts, invoking callbacks for those
-        whose time has come
+            `ioloop.add_callback_threadsafe(ioloop.stop)`
 
         """
-        self._poller.process_timeouts()
+        if (self._thread_id is not None and
+                threading.current_thread().ident != self._thread_id):
+            LOGGER.warning('Use add_callback_threadsafe to request '
+                           'ioloop.stop() from another thread')
+
+        self._poller.stop()
 
     def activate_poller(self):
         """[Extension] Activate the poller
 
         """
+        self._thread_id = threading.current_thread().ident
         self._poller.activate_poller()
 
     def deactivate_poller(self):
@@ -259,7 +514,21 @@ class _PollerBase(_AbstractBase):  # pylint: disable=R0902
     # if the poller uses MS override with 1000
     POLL_TIMEOUT_MULT = 1
 
-    def __init__(self):
+    def __init__(self, get_wait_seconds, process_timeouts):
+        """
+        :param get_wait_seconds: Function for getting the maximum number of
+                                 seconds to wait for IO for use by the poller
+        :param process_timeouts: Function for processing timeouts for use by the
+                                 poller
+
+        """
+        self._get_wait_seconds = get_wait_seconds
+        self._process_timeouts = process_timeouts
+
+        # We guard access to the waking file descriptors to avoid races from
+        # closing them while another thread is calling our `wake()` method.
+        self._waking_mutex = threading.Lock()
+
         # fd-to-handler function mappings
         self._fd_handlers = dict()
 
@@ -271,105 +540,77 @@ class _PollerBase(_AbstractBase):  # pylint: disable=R0902
         # Reentrancy tracker of the `start` method
         self._start_nesting_levels = 0
 
-        self._timeouts = {}
-        self._next_timeout = None
-
         self._stopping = False
 
-        # Mutex for controlling critical sections where ioloop-interrupt sockets
-        # are created, used, and destroyed. Needed in case `stop()` is called
-        # from a thread.
-        self._mutex = threading.Lock()
+        # Create ioloop-interrupt socket pair and register read handler.
+        self._r_interrupt, self._w_interrupt = self._get_interrupt_pair()
+        self.add_handler(self._r_interrupt.fileno(), self._read_interrupt, READ)
 
-        # ioloop-interrupt socket pair; initialized in start()
-        self._r_interrupt = None
-        self._w_interrupt = None
+    def close(self):
+        """Release poller's resources.
 
-    def add_timeout(self, deadline, callback_method):
-        """Add the callback_method to the IOLoop timer to fire after deadline
-        seconds. Returns a handle to the timeout. Do not confuse with
-        Tornado's timeout where you pass in the time you want to have your
-        callback called. Only pass in the seconds until it's to be called.
-
-        :param int deadline: The number of seconds to wait to call callback
-        :param method callback_method: The callback method
-        :rtype: str
+        `close()` is intended to be called after the poller's `start()` method
+        returns. After calling `close()`, no other interaction with the closed
+        poller instance should be performed.
 
         """
-        timeout_at = time.time() + deadline
-        value = {'deadline': timeout_at, 'callback': callback_method}
-        # TODO when timer resolution is low (e.g., windows), we get id collision
-        # when retrying failing connection with tiny (e.g., 0) retry interval
-        timeout_id = hash(frozenset(value.items()))
-        self._timeouts[timeout_id] = value
-
-        if not self._next_timeout or timeout_at < self._next_timeout:
-            self._next_timeout = timeout_at
-
-        LOGGER.debug('add_timeout: added timeout %s; deadline=%s at %s',
-                     timeout_id, deadline, timeout_at)
-        return timeout_id
-
-    def remove_timeout(self, timeout_id):
-        """Remove a timeout if it's still in the timeout stack
-
-        :param str timeout_id: The timeout id to remove
+        # Unregister and close ioloop-interrupt socket pair; mutual exclusion is
+        # necessary to avoid race condition with `wake_threadsafe` executing in
+        # another thread's context
+        assert self._start_nesting_levels == 0, \
+            'Cannot call close() before start() unwinds.'
+
+        with self._waking_mutex:
+            if self._w_interrupt is not None:
+                self.remove_handler(self._r_interrupt.fileno()) # pylint: disable=E1101
+                self._r_interrupt.close()
+                self._r_interrupt = None
+                self._w_interrupt.close()
+                self._w_interrupt = None
+
+        self.deactivate_poller()
+
+        self._fd_handlers = None
+        self._fd_events = None
+        self._processing_fd_event_map = None
+
+    def wake_threadsafe(self):
+        """Wake up the poller as soon as possible. As the name indicates, this
+        method is thread-safe.
 
         """
-        try:
-            timeout = self._timeouts.pop(timeout_id)
-        except KeyError:
-            LOGGER.warning('remove_timeout: %s not found', timeout_id)
-        else:
-            if timeout['deadline'] == self._next_timeout:
-                self._next_timeout = None
+        with self._waking_mutex:
+            if self._w_interrupt is None:
+                return
 
-            LOGGER.debug('remove_timeout: removed %s', timeout_id)
+            try:
+                # Send byte to interrupt the poll loop, use send() instead of
+                # os.write for Windows compatibility
+                self._w_interrupt.send(b'X')
+            except pika.compat.SOCKET_ERROR as err:
+                if err.errno != errno.EWOULDBLOCK:
+                    raise
+            except Exception as err:
+                # There's nothing sensible to do here, we'll exit the interrupt
+                # loop after POLL_TIMEOUT secs in worst case anyway.
+                LOGGER.warning("Failed to send interrupt to poller: %s", err)
+                raise
 
-    def _get_next_deadline(self):
-        """Get the interval to the next timeout event, or a default interval
 
-        """
-        if self._next_timeout:
-            timeout = max(self._next_timeout - time.time(), 0)
+    def _get_max_wait(self):
+        """Get the interval to the next timeout event, or a default interval
 
-        elif self._timeouts:
-            deadlines = [t['deadline'] for t in self._timeouts.values()]
-            self._next_timeout = min(deadlines)
-            timeout = max((self._next_timeout - time.time(), 0))
+        :returns: maximum number of self.POLL_TIMEOUT_MULT-scaled time units
+                  to wait for IO events
 
+        """
+        delay = self._get_wait_seconds()
+        if delay is None:
+            delay = self._MAX_POLL_TIMEOUT
         else:
-            timeout = self._MAX_POLL_TIMEOUT
-
-        timeout = min(timeout, self._MAX_POLL_TIMEOUT)
-        return timeout * self.POLL_TIMEOUT_MULT
+            delay = min(delay, self._MAX_POLL_TIMEOUT)
 
-    def process_timeouts(self):
-        """Process pending timeouts, invoking callbacks for those whose time has
-        come
-
-        """
-        now = time.time()
-        # Run the timeouts in order of deadlines. Although this shouldn't
-        # be strictly necessary it preserves old behaviour when timeouts
-        # were only run periodically.
-        to_run = sorted(
-            [(k, timer)
-             for (k, timer) in self._timeouts.items()
-             if timer['deadline'] <= now],
-            key=lambda item: item[1]['deadline'])
-
-        for k, timer in to_run:
-            if k not in self._timeouts:
-                # Previous invocation(s) should have deleted the timer.
-                continue
-            try:
-                timer['callback']()
-            finally:
-                # Don't do 'del self._timeout[k]' as the key might
-                # have been deleted just now.
-                if self._timeouts.pop(k, None) is not None:
-                    self._next_timeout = None
+        return delay * self.POLL_TIMEOUT_MULT
 
     def add_handler(self, fileno, handler, events):
         """Add a new fileno to the set to be monitored
@@ -449,7 +690,7 @@ class _PollerBase(_AbstractBase):  # pylint: disable=R0902
         """
         # Activate the underlying poller and register current events
         self._init_poller()
-        fd_to_events = defaultdict(int)
+        fd_to_events = collections.defaultdict(int)
         for event, file_descriptors in self._fd_events.items():
             for fileno in file_descriptors:
                 fd_to_events[fileno] |= event
@@ -471,22 +712,10 @@ class _PollerBase(_AbstractBase):  # pylint: disable=R0902
 
         if self._start_nesting_levels == 1:
             LOGGER.debug('Entering IOLoop')
-            self._stopping = False
 
             # Activate the underlying poller and register current events
             self.activate_poller()
 
-            # Create ioloop-interrupt socket pair and register read handler.
-            # NOTE: we defer their creation because some users (e.g.,
-            # BlockingConnection adapter) don't use the event loop and these
-            # sockets would get reported as leaks
-            with self._mutex:
-                assert self._r_interrupt is None
-                self._r_interrupt, self._w_interrupt = self._get_interrupt_pair(
-                )
-                self.add_handler(self._r_interrupt.fileno(),
-                                 self._read_interrupt, READ)
-
         else:
             LOGGER.debug('Reentering IOLoop at nesting level=%s',
                          self._start_nesting_levels)
@@ -495,52 +724,32 @@ class _PollerBase(_AbstractBase):  # pylint: disable=R0902
             # Run event loop
             while not self._stopping:
                 self.poll()
-                self.process_timeouts()
+                self._process_timeouts()
 
         finally:
             self._start_nesting_levels -= 1
 
             if self._start_nesting_levels == 0:
-                LOGGER.debug('Cleaning up IOLoop')
-                # Unregister and close ioloop-interrupt socket pair
-                with self._mutex:
-                    self.remove_handler(self._r_interrupt.fileno())
-                    self._r_interrupt.close()
-                    self._r_interrupt = None
-                    self._w_interrupt.close()
-                    self._w_interrupt = None
-
-                # Deactivate the underlying poller
-                self.deactivate_poller()
+                try:
+                    LOGGER.debug('Deactivating poller')
+
+                    # Deactivate the underlying poller
+                    self.deactivate_poller()
+                finally:
+                    self._stopping = False
             else:
                 LOGGER.debug('Leaving IOLoop with %s nesting levels remaining',
                              self._start_nesting_levels)
 
     def stop(self):
         """Request exit from the ioloop. The loop is NOT guaranteed to stop
-        before this method returns. This is the only method that may be called
-        from another thread.
+        before this method returns.
 
         """
         LOGGER.debug('Stopping IOLoop')
         self._stopping = True
 
-        with self._mutex:
-            if self._w_interrupt is None:
-                return
-
-            try:
-                # Send byte to interrupt the poll loop, use send() instead of
-                # os.write for Windows compatibility
-                self._w_interrupt.send(b'X')
-            except pika.compat.SOCKET_ERROR as err:
-                if err.errno != errno.EWOULDBLOCK:
-                    raise
-            except Exception as err:
-                # There's nothing sensible to do here, we'll exit the interrupt
-                # loop after POLL_TIMEOUT secs in worst case anyway.
-                LOGGER.warning("Failed to send ioloop interrupt: %s", err)
-                raise
+        self.wake_threadsafe()
 
     @abc.abstractmethod
     def poll(self):
@@ -633,7 +842,7 @@ class _PollerBase(_AbstractBase):  # pylint: disable=R0902
         so use a pair of simple TCP sockets instead. The sockets will be
         closed and garbage collected by python when the ioloop itself is.
         """
-        return pika.compat._nonblocking_socketpair()
+        return pika.compat._nonblocking_socketpair() # pylint: disable=W0212
 
     def _read_interrupt(self, interrupt_fd, events):  # pylint: disable=W0613
         """ Read the interrupt byte(s). We ignore the event mask as we can ony
@@ -644,7 +853,7 @@ class _PollerBase(_AbstractBase):  # pylint: disable=R0902
         """
         try:
             # NOTE Use recv instead of os.read for windows compatibility
-            self._r_interrupt.recv(512)
+            self._r_interrupt.recv(512) # pylint: disable=E1101
         except pika.compat.SOCKET_ERROR as err:
             if err.errno != errno.EAGAIN:
                 raise
@@ -659,12 +868,6 @@ class SelectPoller(_PollerBase):
     # if the poller uses MS specify 1000
     POLL_TIMEOUT_MULT = 1
 
-    def __init__(self):
-        """Create an instance of the SelectPoller
-
-        """
-        super(SelectPoller, self).__init__()
-
     def poll(self):
         """Wait for events of interest on registered file descriptors until an
         event of interest occurs or next timer deadline or _MAX_POLL_TIMEOUT,
@@ -677,12 +880,12 @@ class SelectPoller(_PollerBase):
                         self._fd_events[ERROR]):
                     read, write, error = select.select(
                         self._fd_events[READ], self._fd_events[WRITE],
-                        self._fd_events[ERROR], self._get_next_deadline())
+                        self._fd_events[ERROR], self._get_max_wait())
                 else:
                     # NOTE When called without any FDs, select fails on
                     # Windows with error 10022, 'An invalid argument was
                     # supplied'.
-                    time.sleep(self._get_next_deadline())
+                    time.sleep(self._get_max_wait())
                     read, write, error = [], [], []
 
                 break
@@ -694,7 +897,7 @@ class SelectPoller(_PollerBase):
 
         # Build an event bit mask for each fileno we've received an event for
 
-        fd_event_map = defaultdict(int)
+        fd_event_map = collections.defaultdict(int)
         for fd_set, evt in zip((read, write, error), (READ, WRITE, ERROR)):
             for fileno in fd_set:
                 fd_event_map[fileno] |= evt
@@ -750,17 +953,11 @@ class SelectPoller(_PollerBase):
 class KQueuePoller(_PollerBase):
     """KQueuePoller works on BSD based systems and is faster than select"""
 
-    def __init__(self):
+    def __init__(self, get_wait_seconds, process_timeouts):
         """Create an instance of the KQueuePoller
-
-        :param int fileno: The file descriptor to check events for
-        :param method handler: What is called when an event happens
-        :param int events: The events to look for
-
         """
-        super(KQueuePoller, self).__init__()
-
         self._kqueue = None
+        super(KQueuePoller, self).__init__(get_wait_seconds, process_timeouts)
 
     @staticmethod
     def _map_event(kevent):
@@ -776,6 +973,9 @@ class KQueuePoller(_PollerBase):
         elif kevent.flags & select.KQ_EV_ERROR:
             return ERROR
 
+        # Should never happen
+        return None
+
     def poll(self):
         """Wait for events of interest on registered file descriptors until an
         event of interest occurs or next timer deadline or _MAX_POLL_TIMEOUT,
@@ -785,7 +985,7 @@ class KQueuePoller(_PollerBase):
         while True:
             try:
                 kevents = self._kqueue.control(None, 1000,
-                                               self._get_next_deadline())
+                                               self._get_max_wait())
                 break
             except _SELECT_ERRORS as error:
                 if _is_resumable(error):
@@ -793,7 +993,7 @@ class KQueuePoller(_PollerBase):
                 else:
                     raise
 
-        fd_event_map = defaultdict(int)
+        fd_event_map = collections.defaultdict(int)
         for event in kevents:
             fd_event_map[event.ident] |= self._map_event(event)
 
@@ -807,8 +1007,9 @@ class KQueuePoller(_PollerBase):
 
     def _uninit_poller(self):
         """Notify the implementation to release the poller resource"""
-        self._kqueue.close()
-        self._kqueue = None
+        if self._kqueue is not None:
+            self._kqueue.close()
+            self._kqueue = None
 
     def _register_fd(self, fileno, events):
         """The base class invokes this method to notify the implementation to
@@ -882,16 +1083,12 @@ class PollPoller(_PollerBase):
     """
     POLL_TIMEOUT_MULT = 1000
 
-    def __init__(self):
+    def __init__(self, get_wait_seconds, process_timeouts):
         """Create an instance of the KQueuePoller
 
-        :param int fileno: The file descriptor to check events for
-        :param method handler: What is called when an event happens
-        :param int events: The events to look for
-
         """
         self._poll = None
-        super(PollPoller, self).__init__()
+        super(PollPoller, self).__init__(get_wait_seconds, process_timeouts)
 
     @staticmethod
     def _create_poller():
@@ -908,7 +1105,7 @@ class PollPoller(_PollerBase):
         """
         while True:
             try:
-                events = self._poll.poll(self._get_next_deadline())
+                events = self._poll.poll(self._get_max_wait())
                 break
             except _SELECT_ERRORS as error:
                 if _is_resumable(error):
@@ -916,7 +1113,7 @@ class PollPoller(_PollerBase):
                 else:
                     raise
 
-        fd_event_map = defaultdict(int)
+        fd_event_map = collections.defaultdict(int)
         for fileno, event in events:
             fd_event_map[fileno] |= event
 
@@ -930,10 +1127,11 @@ class PollPoller(_PollerBase):
 
     def _uninit_poller(self):
         """Notify the implementation to release the poller resource"""
-        if hasattr(self._poll, "close"):
-            self._poll.close()
+        if self._poll is not None:
+            if hasattr(self._poll, "close"):
+                self._poll.close()
 
-        self._poll = None
+            self._poll = None
 
     def _register_fd(self, fileno, events):
         """The base class invokes this method to notify the implementation to
diff --git a/pika/adapters/tornado_connection.py b/pika/adapters/tornado_connection.py
index ce407d1..db34dfd 100644
--- a/pika/adapters/tornado_connection.py
+++ b/pika/adapters/tornado_connection.py
@@ -94,3 +94,29 @@ class TornadoConnection(base_connection.BaseConnection):
 
         """
         return self.ioloop.remove_timeout(timeout_id)
+
+    def add_callback_threadsafe(self, callback):
+        """Requests a call to the given function as soon as possible in the
+        context of this connection's IOLoop thread.
+
+        NOTE: This is the only thread-safe method offered by the connection. All
+         other manipulations of the connection must be performed from the
+         connection's thread.
+
+        For example, a thread may request a call to the
+        `channel.basic_ack` method of a connection that is running in a
+        different thread via
+
+        ```
+        connection.add_callback_threadsafe(
+            functools.partial(channel.basic_ack, delivery_tag=...))
+        ```
+
+        :param method callback: The callback method; must be callable.
+
+        """
+        if not callable(callback):
+            raise TypeError(
+                'callback must be a callable, but got %r' % (callback,))
+
+        self.ioloop.add_callback(callback)
diff --git a/pika/adapters/twisted_connection.py b/pika/adapters/twisted_connection.py
index 053ddf6..1dac51f 100644
--- a/pika/adapters/twisted_connection.py
+++ b/pika/adapters/twisted_connection.py
@@ -225,6 +225,28 @@ class IOLoopReactorAdapter(object):
         """
         call.cancel()
 
+    def add_callback_threadsafe(self, callback):
+        """Requests a call to the given function as soon as possible in the
+        context of this IOLoop's thread.
+
+        NOTE: This is the only thread-safe method offered by the IOLoop adapter.
+         All other manipulations of the IOLoop adapter and its parent connection
+         must be performed from the connection's thread.
+
+        For example, a thread may request a call to the
+        `channel.basic_ack` method of a connection that is running in a
+        different thread via
+
+        ```
+        connection.add_callback_threadsafe(
+            functools.partial(channel.basic_ack, delivery_tag=...))
+        ```
+
+        :param method callback: The callback method; must be callable.
+
+        """
+        self.reactor.callFromThread(callback)
+
     def stop(self):
         # Guard against stopping the reactor multiple times
         if not self.started:
diff --git a/pika/connection.py b/pika/connection.py
index e371a03..0c4e2a7 100644
--- a/pika/connection.py
+++ b/pika/connection.py
@@ -1238,8 +1238,11 @@ class Connection(object):
             LOGGER.warning('Suppressing close request on %s', self)
             return
 
+        # NOTE The connection is either in opening or open state
+
         # Initiate graceful closing of channels that are OPEN or OPENING
-        self._close_channels(reply_code, reply_text)
+        if self._channels:
+            self._close_channels(reply_code, reply_text)
 
         # Set our connection state
         self._set_connection_state(self.CONNECTION_CLOSING)
diff --git a/pylintrc b/pylintrc
index 2e516a7..4f96c79 100644
--- a/pylintrc
+++ b/pylintrc
@@ -70,7 +70,7 @@ confidence=
 # --enable=similarities". If you want to run only the classes checker, but have
 # no Warning level messages displayed, use"--disable=all --enable=classes
 # --disable=W"
-disable=
+disable=R1705
 
 
 [REPORTS]


Test patch:
diff --git a/tests/acceptance/async_adapter_tests.py b/tests/acceptance/async_adapter_tests.py
index e6f6e47..53004f6 100644
--- a/tests/acceptance/async_adapter_tests.py
+++ b/tests/acceptance/async_adapter_tests.py
@@ -10,6 +10,8 @@
 # Suppress pylint warning about unused argument
 # pylint: disable=W0613
 
+import functools
+import threading
 import time
 import uuid
 
@@ -458,3 +460,128 @@ class TestBlockedConnectionUnblocks(AsyncTestCase, AsyncAdapters):  # pylint: di
         super(TestBlockedConnectionUnblocks, self).on_closed(connection,
                                                              reply_code,
                                                              reply_text)
+
+
+class TestAddCallbackThreadsafeRequestBeforeIOLoopStarts(AsyncTestCase, AsyncAdapters):
+    DESCRIPTION = "Test add_callback_threadsafe request before ioloop starts."
+
+    def _run_ioloop(self, *args, **kwargs):  # pylint: disable=W0221
+        """We intercept this method from AsyncTestCase in order to call
+        add_callback_threadsafe before AsyncTestCase starts the ioloop.
+
+        """
+        self.my_start_time = time.time()
+        # Request a callback from our current (ioloop's) thread
+        self.connection.add_callback_threadsafe(self.on_requested_callback)
+
+        return super(
+            TestAddCallbackThreadsafeRequestBeforeIOLoopStarts, self)._run_ioloop(
+                *args, **kwargs)
+
+    def start(self, *args, **kwargs):  # pylint: disable=W0221
+        self.loop_thread_ident = threading.current_thread().ident
+        self.my_start_time = None
+        self.got_callback = False
+        super(TestAddCallbackThreadsafeRequestBeforeIOLoopStarts, self).start(*args, **kwargs)
+        self.assertTrue(self.got_callback)
+
+    def begin(self, channel):
+        self.stop()
+
+    def on_requested_callback(self):
+        self.assertEqual(threading.current_thread().ident,
+                         self.loop_thread_ident)
+        self.assertLess(time.time() - self.my_start_time, 0.25)
+        self.got_callback = True
+
+
+class TestAddCallbackThreadsafeFromIOLoopThread(AsyncTestCase, AsyncAdapters):
+    DESCRIPTION = "Test add_callback_threadsafe request from same thread."
+
+    def start(self, *args, **kwargs):
+        self.loop_thread_ident = threading.current_thread().ident
+        self.my_start_time = None
+        self.got_callback = False
+        super(TestAddCallbackThreadsafeFromIOLoopThread, self).start(*args, **kwargs)
+        self.assertTrue(self.got_callback)
+
+    def begin(self, channel):
+        self.my_start_time = time.time()
+        # Request a callback from our current (ioloop's) thread
+        channel.connection.add_callback_threadsafe(self.on_requested_callback)
+
+    def on_requested_callback(self):
+        self.assertEqual(threading.current_thread().ident,
+                         self.loop_thread_ident)
+        self.assertLess(time.time() - self.my_start_time, 0.25)
+        self.got_callback = True
+        self.stop()
+
+
+class TestAddCallbackThreadsafeFromAnotherThread(AsyncTestCase, AsyncAdapters):
+    DESCRIPTION = "Test add_callback_threadsafe request from another thread."
+
+    def start(self, *args, **kwargs):
+        self.loop_thread_ident = threading.current_thread().ident
+        self.my_start_time = None
+        self.got_callback = False
+        super(TestAddCallbackThreadsafeFromAnotherThread, self).start(*args, **kwargs)
+        self.assertTrue(self.got_callback)
+
+    def begin(self, channel):
+        self.my_start_time = time.time()
+        # Request a callback from ioloop while executing in another thread
+        timer = threading.Timer(
+            0,
+            lambda: channel.connection.add_callback_threadsafe(
+                self.on_requested_callback))
+        self.addCleanup(timer.cancel)
+        timer.start()
+
+    def on_requested_callback(self):
+        self.assertEqual(threading.current_thread().ident,
+                         self.loop_thread_ident)
+        self.assertLess(time.time() - self.my_start_time, 0.25)
+        self.got_callback = True
+        self.stop()
+
+
+class TestIOLoopStopBeforeIOLoopStarts(AsyncTestCase, AsyncAdapters):
+    DESCRIPTION = "Test ioloop.stop() before ioloop starts causes ioloop to exit quickly."
+
+    def _run_ioloop(self, *args, **kwargs):  # pylint: disable=W0221
+        """We intercept this method from AsyncTestCase in order to call
+        ioloop.stop() before AsyncTestCase starts the ioloop.
+        """
+        # Request ioloop to stop before it starts
+        self.my_start_time = time.time()
+        self.stop_ioloop_only()
+
+        return super(
+            TestIOLoopStopBeforeIOLoopStarts, self)._run_ioloop(*args, **kwargs)
+
+    def start(self, *args, **kwargs):  # pylint: disable=W0221
+        self.loop_thread_ident = threading.current_thread().ident
+        self.my_start_time = None
+        super(TestIOLoopStopBeforeIOLoopStarts, self).start(*args, **kwargs)
+        self.assertLess(time.time() - self.my_start_time, 0.25)
+
+    def begin(self, channel):
+        pass
+
+
+class TestViabilityOfMultipleTimeoutsWithSameDeadlineAndCallback(AsyncTestCase, AsyncAdapters):  # pylint: disable=C0103
+    DESCRIPTION = "Test viability of multiple timeouts with same deadline and callback"
+
+    def begin(self, channel):
+        timer1 = channel.connection.add_timeout(0, self.on_my_timer)
+        timer2 = channel.connection.add_timeout(0, self.on_my_timer)
+
+        self.assertNotEqual(timer1, timer2)
+
+        channel.connection.remove_timeout(timer1)
+
+        # Wait for timer2 to fire
+
+    def on_my_timer(self):
+        self.stop()
diff --git a/tests/acceptance/async_test_base.py b/tests/acceptance/async_test_base.py
index 33f5d3d..6a80b0c 100644
--- a/tests/acceptance/async_test_base.py
+++ b/tests/acceptance/async_test_base.py
@@ -68,34 +68,55 @@ class AsyncTestCase(unittest.TestCase):
         """Extend to start the actual tests on the channel"""
         self.fail("AsyncTestCase.begin_test not extended")
 
-    def start(self, adapter=None):
+    def start(self, adapter, ioloop_factory):
         self.logger.info('start at %s', datetime.datetime.utcnow())
         self.adapter = adapter or self.ADAPTER
 
-        self.connection = self.adapter(self.parameters, self.on_open,
-                                       self.on_open_error, self.on_closed)
-        self.timeout = self.connection.add_timeout(self.TIMEOUT,
-                                                   self.on_timeout)
-        self.connection.ioloop.start()
-        self.assertFalse(self._timed_out)
+        self.connection = self.adapter(self.parameters,
+                                       self.on_open,
+                                       self.on_open_error,
+                                       self.on_closed,
+                                       custom_ioloop=ioloop_factory())
+        try:
+            self.timeout = self.connection.add_timeout(self.TIMEOUT,
+                                                       self.on_timeout)
+            self._run_ioloop()
+            self.assertFalse(self._timed_out)
+        finally:
+            self.connection.ioloop.close()
+            self.connection = None
+
+    def stop_ioloop_only(self):
+        """Request stopping of the connection's ioloop to end the test without
+        closing the connection
+        """
+        self._safe_remove_test_timeout()
+        self.connection.ioloop.stop()
 
     def stop(self):
         """close the connection and stop the ioloop"""
         self.logger.info("Stopping test")
-        if self.timeout is not None:
-            self.connection.remove_timeout(self.timeout)
-            self.timeout = None
-        self.connection.close()
+        self._safe_remove_test_timeout()
+        self.connection.close() # NOTE: on_closed() will stop the ioloop
+
+    def _run_ioloop(self):
+        """Some tests need to subclass this in order to bootstrap their test
+        logic after we instantiate the connection and assign it to
+        `self.connection`, but before we run the ioloop
+        """
+        self.connection.ioloop.start()
 
-    def _stop(self):
+    def _safe_remove_test_timeout(self):
         if hasattr(self, 'timeout') and self.timeout is not None:
             self.logger.info("Removing timeout")
             self.connection.remove_timeout(self.timeout)
             self.timeout = None
-        if hasattr(self, 'connection') and self.connection:
+
+    def _stop(self):
+        self._safe_remove_test_timeout()
+        if hasattr(self, 'connection') and self.connection is not None:
             self.logger.info("Stopping ioloop")
             self.connection.ioloop.stop()
-            self.connection = None
 
     def on_closed(self, connection, reply_code, reply_text):
         """called when the connection has finished closing"""
@@ -124,12 +145,12 @@ class AsyncTestCase(unittest.TestCase):
 
 class BoundQueueTestCase(AsyncTestCase):
 
-    def start(self, adapter=None):
+    def start(self, adapter, ioloop_factory):
         # PY3 compat encoding
         self.exchange = 'e-' + self.__class__.__name__ + ':' + uuid.uuid1().hex
         self.queue = 'q-' + self.__class__.__name__ + ':' + uuid.uuid1().hex
         self.routing_key = self.__class__.__name__
-        super(BoundQueueTestCase, self).start(adapter)
+        super(BoundQueueTestCase, self).start(adapter, ioloop_factory)
 
     def begin(self, channel):
         self.channel.exchange_declare(self.on_exchange_declared, self.exchange,
@@ -164,20 +185,31 @@ class BoundQueueTestCase(AsyncTestCase):
 
 class AsyncAdapters(object):
 
-    def start(self, adapter_class):
+    def start(self, adapter_class, ioloop_factory):
+        """
+
+        :param adapter_class: pika connection adapter class to test.
+        :param ioloop_factory: to be called without args to instantiate a
+           non-shared ioloop to be passed as the `custom_ioloop` arg to the
+           `adapter_class` constructor. This is needed because some of the
+           adapters default to using a singleton ioloop, which results in
+           tests errors after prior tests close the ioloop to release resources,
+           in order to eliminate ResourceWarning warnings concerning unclosed
+           sockets from our adapters.
+        :return:
+        """
         raise NotImplementedError
 
     def select_default_test(self):
         """SelectConnection:DefaultPoller"""
-
         with mock.patch.multiple(select_connection, SELECT_TYPE=None):
-            self.start(adapters.SelectConnection)
+            self.start(adapters.SelectConnection, select_connection.IOLoop)
 
     def select_select_test(self):
         """SelectConnection:select"""
 
         with mock.patch.multiple(select_connection, SELECT_TYPE='select'):
-            self.start(adapters.SelectConnection)
+            self.start(adapters.SelectConnection, select_connection.IOLoop)
 
     @unittest.skipIf(
         not hasattr(select, 'poll') or
@@ -186,27 +218,36 @@ class AsyncAdapters(object):
         """SelectConnection:poll"""
 
         with mock.patch.multiple(select_connection, SELECT_TYPE='poll'):
-            self.start(adapters.SelectConnection)
+            self.start(adapters.SelectConnection, select_connection.IOLoop)
 
     @unittest.skipIf(not hasattr(select, 'epoll'), "epoll not supported")
     def select_epoll_test(self):
         """SelectConnection:epoll"""
 
         with mock.patch.multiple(select_connection, SELECT_TYPE='epoll'):
-            self.start(adapters.SelectConnection)
+            self.start(adapters.SelectConnection, select_connection.IOLoop)
 
     @unittest.skipIf(not hasattr(select, 'kqueue'), "kqueue not supported")
     def select_kqueue_test(self):
         """SelectConnection:kqueue"""
 
         with mock.patch.multiple(select_connection, SELECT_TYPE='kqueue'):
-            self.start(adapters.SelectConnection)
+            self.start(adapters.SelectConnection, select_connection.IOLoop)
 
     def tornado_test(self):
         """TornadoConnection"""
-        self.start(adapters.TornadoConnection)
+        ioloop_factory = None
+        if adapters.TornadoConnection is not None:
+            import tornado.ioloop
+            ioloop_factory = tornado.ioloop.IOLoop
+        self.start(adapters.TornadoConnection, ioloop_factory)
 
     @unittest.skipIf(sys.version_info < (3, 4), "Asyncio available for Python 3.4+")
     def asyncio_test(self):
         """AsyncioConnection"""
-        self.start(adapters.AsyncioConnection)
+        ioloop_factory = None
+        if adapters.AsyncioConnection is not None:
+            import asyncio
+            ioloop_factory = asyncio.new_event_loop
+
+        self.start(adapters.AsyncioConnection, ioloop_factory)
diff --git a/tests/acceptance/blocking_adapter_test.py b/tests/acceptance/blocking_adapter_test.py
index e8643f1..c27be2e 100644
--- a/tests/acceptance/blocking_adapter_test.py
+++ b/tests/acceptance/blocking_adapter_test.py
@@ -1,7 +1,9 @@
 """blocking adapter test"""
 from datetime import datetime
+import functools
 import logging
 import socket
+import threading
 import time
 import unittest
 import uuid
@@ -449,6 +451,46 @@ class TestBlockedConnectionTimeout(BlockingTestCaseBase):
              'Blocked connection timeout expired'))
 
 
+class TestAddCallbackThreadsafeFromSameThread(BlockingTestCaseBase):
+
+    def test(self):
+        """BlockingConnection.add_callback_threadsafe from same thread"""
+        connection = self._connect()
+
+        # Test timer completion
+        start_time = time.time()
+        rx_callback = []
+        connection.add_callback_threadsafe(
+            lambda: rx_callback.append(time.time()))
+        while not rx_callback:
+            connection.process_data_events(time_limit=None)
+
+        self.assertEqual(len(rx_callback), 1)
+        elapsed = time.time() - start_time
+        self.assertLess(elapsed, 0.25)
+
+
+class TestAddCallbackThreadsafeFromAnotherThread(BlockingTestCaseBase):
+
+    def test(self):
+        """BlockingConnection.add_callback_threadsafe from another thread"""
+        connection = self._connect()
+
+        # Test timer completion
+        start_time = time.time()
+        rx_callback = []
+        timer = threading.Timer(
+            0,
+            functools.partial(connection.add_callback_threadsafe,
+                              lambda: rx_callback.append(time.time())))
+        self.addCleanup(timer.cancel)
+        timer.start()
+        while not rx_callback:
+            connection.process_data_events(time_limit=None)
+
+        self.assertEqual(len(rx_callback), 1)
+        elapsed = time.time() - start_time
+        self.assertLess(elapsed, 0.25)
 
 
 class TestAddTimeoutRemoveTimeout(BlockingTestCaseBase):
@@ -488,6 +530,32 @@ class TestAddTimeoutRemoveTimeout(BlockingTestCaseBase):
         repr(evt)
 
 
+class TestViabilityOfMultipleTimeoutsWithSameDeadlineAndCallback(BlockingTestCaseBase):
+
+    def test(self):
+        """BlockingConnection viability of multiple timeouts with same deadline and callback"""
+        connection = self._connect()
+
+        rx_callback = []
+
+        def callback():
+            rx_callback.append(1)
+
+        timer1 = connection.add_timeout(0, callback)
+        timer2 = connection.add_timeout(0, callback)
+
+        self.assertNotEqual(timer1, timer2)
+
+        connection.remove_timeout(timer1)
+
+        # Wait for second timer to fire
+        start_wait_time = time.time()
+        while not rx_callback and time.time() - start_wait_time < 0.25:
+            connection.process_data_events(time_limit=0.001)
+
+        self.assertListEqual(rx_callback, [1])
+
+
 class TestRemoveTimeoutFromTimeoutCallback(BlockingTestCaseBase):
 
     def test(self):
@@ -508,7 +576,7 @@ class TestRemoveTimeoutFromTimeoutCallback(BlockingTestCaseBase):
         while not rx_timer2:
             connection.process_data_events(time_limit=None)
 
-        self.assertNotIn(timer_id1, connection._impl.ioloop._poller._timeouts)
+        self.assertIsNone(timer_id1.callback)
         self.assertFalse(connection._ready_events)
 
 
@@ -1471,7 +1539,7 @@ class TestBasicPublishDeliveredWhenPendingUnroutable(BlockingTestCaseBase):
                                mandatory=True)
         self.assertEqual(res, True)
 
-        # Flush channel to force Basic.Return
+        # Flush connection to force Basic.Return
         connection.channel().close()
 
         # Deposit a routable message in the queue
@@ -1630,7 +1698,7 @@ class TestPublishAndConsumeWithPubacksAndQosOfOne(BlockingTestCaseBase):
                                                       queue=q_name,
                                                       expected_count=0)
 
-        # Attempt to cosume again with a short timeout
+        # Attempt to consume again with a short timeout
         connection.process_data_events(time_limit=0.005)
         self.assertEqual(len(rx_messages), 2)
 
@@ -1645,6 +1713,197 @@ class TestPublishAndConsumeWithPubacksAndQosOfOne(BlockingTestCaseBase):
         self.assertEqual(frame.method.consumer_tag, consumer_tag)
 
 
+class TestBasicConsumeWithAckFromAnotherThread(BlockingTestCaseBase):
+
+    def test(self):  # pylint: disable=R0914,R0915
+        """BlockingChannel.basic_consume with ack from another thread and \
+        requesting basic_ack via add_callback_threadsafe
+        """
+        # This test simulates processing of a message on another thread and
+        # then requesting an ACK to be dispatched on the connection's thread
+        # via BlockingConnection.add_callback_threadsafe
+
+        connection = self._connect()
+
+        ch = connection.channel()
+
+        q_name = 'TestBasicConsumeWithAckFromAnotherThread_q' + uuid.uuid1().hex
+        exg_name = ('TestBasicConsumeWithAckFromAnotherThread_exg' +
+                    uuid.uuid1().hex)
+        routing_key = 'TestBasicConsumeWithAckFromAnotherThread'
+
+        # Place channel in publisher-acknowledgments mode so that publishing
+        # with mandatory=True will be synchronous (for convenience)
+        res = ch.confirm_delivery()
+        self.assertIsNone(res)
+
+        # Declare a new exchange
+        ch.exchange_declare(exg_name, exchange_type='direct')
+        self.addCleanup(connection.channel().exchange_delete, exg_name)
+
+        # Declare a new queue
+        ch.queue_declare(q_name, auto_delete=True)
+        self.addCleanup(self._connect().channel().queue_delete, q_name)
+
+        # Bind the queue to the exchange using routing key
+        ch.queue_bind(q_name, exchange=exg_name, routing_key=routing_key)
+
+        # Publish 2 messages with mandatory=True for synchronous processing
+        ch.publish(exg_name, routing_key, body='msg1', mandatory=True)
+        ch.publish(exg_name, routing_key, body='last-msg', mandatory=True)
+
+        # Configure QoS for one message so that the 2nd message will be
+        # delivered only after the 1st one is ACKed
+        ch.basic_qos(prefetch_size=0, prefetch_count=1, all_channels=False)
+
+        # Create a consumer
+        rx_messages = []
+        def ackAndEnqueueMessageViaAnotherThread(rx_ch,
+                                                 rx_method,
+                                                 rx_properties, # pylint: disable=W0613
+                                                 rx_body):
+            LOGGER.debug(
+                '%s: Got message body=%r; delivery-tag=%r',
+                datetime.now(), rx_body, rx_method.delivery_tag)
+
+            # Request ACK dispatch via add_callback_threadsafe from other
+            # thread; if last message, cancel consumer so that start_consuming
+            # can return
+
+            def processOnConnectionThread():
+                LOGGER.debug('%s: ACKing message body=%r; delivery-tag=%r',
+                             datetime.now(),
+                             rx_body,
+                             rx_method.delivery_tag)
+                ch.basic_ack(delivery_tag=rx_method.delivery_tag,
+                             multiple=False)
+                rx_messages.append(rx_body)
+
+                # NOTE on python3, `b'last-msg' != 'last-msg'`
+                if rx_body == b'last-msg':
+                    LOGGER.debug('%s: Canceling consumer consumer-tag=%r',
+                                 datetime.now(),
+                                 rx_method.consumer_tag)
+                    rx_ch.basic_cancel(rx_method.consumer_tag)
+
+            # Spawn a thread to initiate ACKing
+            timer = threading.Timer(0,
+                                    lambda: connection.add_callback_threadsafe(
+                                        processOnConnectionThread))
+            self.addCleanup(timer.cancel)
+            timer.start()
+
+        consumer_tag = ch.basic_consume(
+            ackAndEnqueueMessageViaAnotherThread,
+            q_name,
+            no_ack=False,
+            exclusive=False,
+            arguments=None)
+
+        # Wait for both messages
+        LOGGER.debug('%s: calling start_consuming(); consumer tag=%r',
+                     datetime.now(),
+                     consumer_tag)
+        ch.start_consuming()
+        LOGGER.debug('%s: Returned from start_consuming(); consumer tag=%r',
+                     datetime.now(),
+                     consumer_tag)
+
+        self.assertEqual(len(rx_messages), 2)
+        self.assertEqual(rx_messages[0], b'msg1')
+        self.assertEqual(rx_messages[1], b'last-msg')
+
+
+class TestConsumeGeneratorWithAckFromAnotherThread(BlockingTestCaseBase):
+
+    def test(self):  # pylint: disable=R0914,R0915
+        """BlockingChannel.consume and requesting basic_ack from another \
+        thread via add_callback_threadsafe
+        """
+        connection = self._connect()
+
+        ch = connection.channel()
+
+        q_name = ('TestConsumeGeneratorWithAckFromAnotherThread_q' +
+                  uuid.uuid1().hex)
+        exg_name = ('TestConsumeGeneratorWithAckFromAnotherThread_exg' +
+                    uuid.uuid1().hex)
+        routing_key = 'TestConsumeGeneratorWithAckFromAnotherThread'
+
+        # Place channel in publisher-acknowledgments mode so that publishing
+        # with mandatory=True will be synchronous (for convenience)
+        res = ch.confirm_delivery()
+        self.assertIsNone(res)
+
+        # Declare a new exchange
+        ch.exchange_declare(exg_name, exchange_type='direct')
+        self.addCleanup(connection.channel().exchange_delete, exg_name)
+
+        # Declare a new queue
+        ch.queue_declare(q_name, auto_delete=True)
+        self.addCleanup(self._connect().channel().queue_delete, q_name)
+
+        # Bind the queue to the exchange using routing key
+        ch.queue_bind(q_name, exchange=exg_name, routing_key=routing_key)
+
+        # Publish 2 messages with mandatory=True for synchronous processing
+        ch.publish(exg_name, routing_key, body='msg1', mandatory=True)
+        ch.publish(exg_name, routing_key, body='last-msg', mandatory=True)
+
+        # Configure QoS for one message so that the 2nd message will be
+        # delivered only after the 1st one is ACKed
+        ch.basic_qos(prefetch_size=0, prefetch_count=1, all_channels=False)
+
+        # Create a consumer
+        rx_messages = []
+        def ackAndEnqueueMessageViaAnotherThread(rx_ch,
+                                                 rx_method,
+                                                 rx_properties, # pylint: disable=W0613
+                                                 rx_body):
+            LOGGER.debug(
+                '%s: Got message body=%r; delivery-tag=%r',
+                datetime.now(), rx_body, rx_method.delivery_tag)
+
+            # Request ACK dispatch via add_callback_threadsafe from other
+            # thread; if last message, cancel consumer so that consumer
+            # generator completes
+
+            def processOnConnectionThread():
+                LOGGER.debug('%s: ACKing message body=%r; delivery-tag=%r',
+                             datetime.now(),
+                             rx_body,
+                             rx_method.delivery_tag)
+                ch.basic_ack(delivery_tag=rx_method.delivery_tag,
+                             multiple=False)
+                rx_messages.append(rx_body)
+
+                # NOTE on python3, `b'last-msg' != 'last-msg'`
+                if rx_body == b'last-msg':
+                    LOGGER.debug('%s: Canceling consumer consumer-tag=%r',
+                                 datetime.now(),
+                                 rx_method.consumer_tag)
+                    # NOTE Need to use cancel() for the consumer generator
+                    # instead of basic_cancel()
+                    rx_ch.cancel()
+
+            # Spawn a thread to initiate ACKing
+            timer = threading.Timer(0,
+                                    lambda: connection.add_callback_threadsafe(
+                                        processOnConnectionThread))
+            self.addCleanup(timer.cancel)
+            timer.start()
+
+        for method, properties, body in ch.consume(q_name, no_ack=False):
+            ackAndEnqueueMessageViaAnotherThread(rx_ch=ch,
+                                                 rx_method=method,
+                                                 rx_properties=properties,
+                                                 rx_body=body)
+
+        self.assertEqual(len(rx_messages), 2)
+        self.assertEqual(rx_messages[0], b'msg1')
+        self.assertEqual(rx_messages[1], b'last-msg')
+
+
 class TestTwoBasicConsumersOnSameChannel(BlockingTestCaseBase):
 
     def test(self):  # pylint: disable=R0914
@@ -1914,7 +2173,7 @@ class TestBasicPublishWithoutPubacks(BlockingTestCaseBase):
                                                       queue=q_name,
                                                       expected_count=0)
 
-        # Attempt to cosume again with a short timeout
+        # Attempt to consume again with a short timeout
         connection.process_data_events(time_limit=0.005)
         self.assertEqual(len(rx_messages), 2)
 
diff --git a/tests/unit/blocking_connection_tests.py b/tests/unit/blocking_connection_tests.py
index 118aa2d..ab046a0 100644
--- a/tests/unit/blocking_connection_tests.py
+++ b/tests/unit/blocking_connection_tests.py
@@ -126,8 +126,7 @@ class BlockingConnectionTests(unittest.TestCase):
         connection._flush_output(lambda: False, lambda: True)
 
         self.assertEqual(connection._impl.ioloop.activate_poller.call_count, 1)
-        self.assertEqual(connection._impl.ioloop.deactivate_poller.call_count,
-                         1)
+        self.assertEqual(connection._impl.ioloop.close.call_count, 1)
 
     @patch.object(
         blocking_connection,
@@ -152,8 +151,7 @@ class BlockingConnectionTests(unittest.TestCase):
         self.assertSequenceEqual(cm.exception.args, (404, 'not found'))
 
         self.assertEqual(connection._impl.ioloop.activate_poller.call_count, 1)
-        self.assertEqual(connection._impl.ioloop.deactivate_poller.call_count,
-                         1)
+        self.assertEqual(connection._impl.ioloop.close.call_count, 1)
 
     @patch.object(
         blocking_connection,
@@ -178,8 +176,7 @@ class BlockingConnectionTests(unittest.TestCase):
         self.assertSequenceEqual(cm.exception.args, (200, 'ok'))
 
         self.assertEqual(connection._impl.ioloop.activate_poller.call_count, 1)
-        self.assertEqual(connection._impl.ioloop.deactivate_poller.call_count,
-                         1)
+        self.assertEqual(connection._impl.ioloop.close.call_count, 1)
 
     @patch.object(
         blocking_connection,
diff --git a/tests/unit/connection_timeout_tests.py b/tests/unit/connection_timeout_tests.py
index b44161b..3118fab 100644
--- a/tests/unit/connection_timeout_tests.py
+++ b/tests/unit/connection_timeout_tests.py
@@ -49,8 +49,13 @@ class ConnectionTests(unittest.TestCase):
                             connect=mock.Mock(side_effect=mock_timeout))
                 ) as create_sock_mock:
                     params = pika.ConnectionParameters(socket_timeout=2.0)
-                    conn = asyncio_connection.AsyncioConnection(params)
+                    ioloop = asyncio_connection.asyncio.new_event_loop()
+                    self.addCleanup(ioloop.close)
+                    conn = asyncio_connection.AsyncioConnection(
+                        params,
+                        custom_ioloop=ioloop)
                     conn._on_connect_timer()
+
         create_sock_mock.return_value.settimeout.assert_called_with(2.0)
         self.assertIn('timeout', str(err_ctx.exception))
 
@@ -99,6 +104,7 @@ class ConnectionTests(unittest.TestCase):
                             side_effect=mock_timeout))) as create_sock_mock:
                 params = pika.ConnectionParameters(socket_timeout=2.0)
                 conn = select_connection.SelectConnection(params)
+                self.addCleanup(conn.ioloop.close)
                 conn._on_connect_timer()
         create_sock_mock.return_value.settimeout.assert_called_with(2.0)
         self.assertIn('timeout', str(err_ctx.exception))
@@ -113,7 +119,11 @@ class ConnectionTests(unittest.TestCase):
                         connect=mock.Mock(
                             side_effect=mock_timeout))) as create_sock_mock:
                 params = pika.ConnectionParameters(socket_timeout=2.0)
-                conn = tornado_connection.TornadoConnection(params)
+                ioloop = tornado_connection.ioloop.IOLoop()
+                self.addCleanup(ioloop.close)
+                conn = tornado_connection.TornadoConnection(
+                    params,
+                    custom_ioloop=ioloop)
                 conn._on_connect_timer()
         create_sock_mock.return_value.settimeout.assert_called_with(2.0)
         self.assertIn('timeout', str(err_ctx.exception))
diff --git a/tests/unit/select_connection_ioloop_tests.py b/tests/unit/select_connection_ioloop_tests.py
index ffc34ea..d5c9cf1 100644
--- a/tests/unit/select_connection_ioloop_tests.py
+++ b/tests/unit/select_connection_ioloop_tests.py
@@ -20,6 +20,12 @@ import pika
 from pika import compat
 from pika.adapters import select_connection
 
+# protected-access
+# pylint: disable=W0212
+# missing-docstring
+# pylint: disable=C0111
+
+
 EPOLL_SUPPORTED = hasattr(select, 'epoll')
 POLL_SUPPORTED = hasattr(select, 'poll') and hasattr(select.poll(), 'modify')
 KQUEUE_SUPPORTED = hasattr(select, 'kqueue')
@@ -37,6 +43,7 @@ class IOLoopBaseTest(unittest.TestCase):
 
         self.ioloop = select_connection.IOLoop()
         self.addCleanup(setattr, self, 'ioloop', None)
+        self.addCleanup(self.ioloop.close)
 
         activate_poller_patch = mock.patch.object(
             self.ioloop._poller,
@@ -74,13 +81,76 @@ class IOLoopBaseTest(unittest.TestCase):
         self.fail('Test timed out')
 
 
+class IOLoopCloseClosesSubordinateObjectsTestSelect(IOLoopBaseTest):
+    """ Test ioloop being closed """
+    SELECT_POLLER = 'select'
+
+    def start_test(self):
+        with mock.patch.multiple(self.ioloop,
+                                 _timer=mock.DEFAULT,
+                                 _poller=mock.DEFAULT,
+                                 _callbacks=mock.DEFAULT) as mocks:
+            self.ioloop.close()
+            mocks['_timer'].close.assert_called_once()
+            mocks['_poller'].close.assert_called_once()
+            self.assertIsNone(self.ioloop._callbacks)
+
+
+class IOLoopCloseAfterStartReturnsTestSelect(IOLoopBaseTest):
+    """ Test IOLoop.close() after normal return from start(). """
+    SELECT_POLLER = 'select'
+
+    def start_test(self):
+        self.ioloop.stop() # so start will terminate quickly
+        self.start()
+        self.ioloop.close()
+        self.assertIsNone(self.ioloop._callbacks)
+
+
+class IOLoopCloseBeforeStartReturnsTestSelect(IOLoopBaseTest):
+    """ Test calling IOLoop.close() before return from start() raises exception. """
+    SELECT_POLLER = 'select'
+
+    def start_test(self):
+        callback_completed = []
+
+        def call_close_from_callback():
+            with self.assertRaises(AssertionError) as cm:
+                self.ioloop.close()
+
+            self.assertEqual(cm.exception.args[0],
+                             'Cannot call close() before start() unwinds.')
+            self.ioloop.stop()
+            callback_completed.append(1)
+
+        self.ioloop.add_callback_threadsafe(call_close_from_callback)
+        self.start()
+        self.assertEqual(callback_completed, [1])
+
+
 class IOLoopThreadStopTestSelect(IOLoopBaseTest):
     """ Test ioloop being stopped by another Thread. """
     SELECT_POLLER = 'select'
 
     def start_test(self):
         """Starts a thread that stops ioloop after a while and start polling"""
-        timer = threading.Timer(0.1, self.ioloop.stop)
+        timer = threading.Timer(
+            0.1,
+            lambda: self.ioloop.add_callback_threadsafe(self.ioloop.stop))
+        self.addCleanup(timer.cancel)
+        timer.start()
+        self.start() # NOTE: Normal return from `start()` constitutes success
+
+
+class IOLoopThreadStopTestSelect(IOLoopBaseTest):
+    """ Test ioloop being stopped by another Thread. """
+    SELECT_POLLER = 'select'
+
+    def start_test(self):
+        """Starts a thread that stops ioloop after a while and start polling"""
+        timer = threading.Timer(
+            0.1,
+            lambda: self.ioloop.add_callback_threadsafe(self.ioloop.stop))
         self.addCleanup(timer.cancel)
         timer.start()
         self.start() # NOTE: Normal return from `start()` constitutes success
@@ -164,7 +234,7 @@ class IOLoopTimerTestSelect(IOLoopBaseTest):
         """A timeout handler that tries to remove itself."""
         self.assertEqual(self.handle, handle_holder.pop())
         # This removal here should not raise exception by itself nor
-        # in the caller SelectPoller.process_timeouts().
+        # in the caller SelectPoller._process_timeouts().
         self.timer_got_called = True
         self.ioloop.remove_timeout(self.handle)
         self.ioloop.stop()
@@ -202,7 +272,7 @@ class IOLoopTimerTestSelect(IOLoopBaseTest):
             """
             self.concluded = True
             self.assertTrue(self.deleted_another_timer)
-            self.assertNotIn(target_timer, self.ioloop._poller._timeouts)
+            self.assertIsNone(target_timer.callback)
             self.ioloop.stop()
 
         self.ioloop.add_timeout(0.01, _on_timer_conclude)
@@ -438,7 +508,10 @@ class IOLoopEintrTestCaseSelect(IOLoopBaseTest):
            implementation of polling mechanism and another."""
         is_resumable_mock.side_effect = is_resumable_raw
 
-        self.poller = self.ioloop._get_poller()
+        timer = select_connection._Timer()
+        self.poller = self.ioloop._get_poller(timer.get_remaining_interval,
+                                              timer.process_timeouts)
+        self.addCleanup(self.poller.close)
 
         sockpair = self.poller._get_interrupt_pair()
         self.addCleanup(sockpair[0].close)
@@ -448,7 +521,7 @@ class IOLoopEintrTestCaseSelect(IOLoopBaseTest):
         self.poller.add_handler(sockpair[0].fileno(), self._eintr_read_handler,
                                 select_connection.READ)
 
-        self.poller.add_timeout(self.TIMEOUT, self._eintr_test_fail)
+        self.ioloop.add_timeout(self.TIMEOUT, self._eintr_test_fail)
 
         original_signal_handler = \
             signal.signal(signal.SIGUSR1, self.signal_handler)
@@ -490,23 +563,72 @@ class IOLoopEintrTestCaseKqueue(IOLoopEintrTestCaseSelect):
 
 class SelectPollerTestPollWithoutSockets(unittest.TestCase):
     def start_test(self):
-        poller = select_connection.SelectPoller()
+        timer = select_connection._Timer()
+        poller = select_connection.SelectPoller(
+            get_wait_seconds=timer.get_remaining_interval,
+            process_timeouts=timer.process_timeouts)
+        self.addCleanup(poller.close)
 
         timer_call_container = []
-        poller.add_timeout(0.00001, lambda: timer_call_container.append(1))
+        timer.call_later(0.00001, lambda: timer_call_container.append(1))
         poller.poll()
 
-        deadline = poller._next_timeout
+        delay = poller._get_wait_seconds()
+        self.assertIsNotNone(delay)
+        deadline = time.time() + delay
 
         while True:
-            poller.process_timeouts()
+            poller._process_timeouts()
 
             if time.time() < deadline:
                 self.assertEqual(timer_call_container, [])
             else:
                 # One last time in case deadline reached after previous
                 # processing cycle
-                poller.process_timeouts()
+                poller._process_timeouts()
                 break
 
         self.assertEqual(timer_call_container, [1])
+
+
+class PollerTestCaseSelect(unittest.TestCase):
+    SELECT_POLLER = 'select'
+
+    def setUp(self):
+        select_type_patch = mock.patch.multiple(
+            select_connection, SELECT_TYPE=self.SELECT_POLLER)
+        select_type_patch.start()
+        self.addCleanup(select_type_patch.stop)
+
+        timer = select_connection._Timer()
+        self.addCleanup(timer.close)
+        self.poller = select_connection.IOLoop._get_poller(
+            timer.get_remaining_interval,
+            timer.process_timeouts)
+        self.addCleanup(self.poller.close)
+
+    def test_poller_close(self):
+        self.poller.close()
+        self.assertIsNone(self.poller._r_interrupt)
+        self.assertIsNone(self.poller._w_interrupt)
+        self.assertIsNone(self.poller._fd_handlers)
+        self.assertIsNone(self.poller._fd_events)
+        self.assertIsNone(self.poller._processing_fd_event_map)
+
+
+@unittest.skipIf(not POLL_SUPPORTED, 'poll not supported')
+class PollerTestCasePoll(PollerTestCaseSelect):
+    """Same as PollerTestCaseSelect but uses poll syscall"""
+    SELECT_POLLER = 'poll'
+
+
+@unittest.skipIf(not EPOLL_SUPPORTED, 'epoll not supported')
+class PollerTestCaseEPoll(PollerTestCaseSelect):
+    """Same as PollerTestCaseSelect but uses epoll syscall"""
+    SELECT_POLLER = 'epoll'
+
+
+@unittest.skipIf(not KQUEUE_SUPPORTED, 'kqueue not supported')
+class PollerTestCaseKqueue(PollerTestCaseSelect):
+    """Same as PollerTestCaseSelect but uses kqueue syscall"""
+    SELECT_POLLER = 'kqueue'
diff --git a/tests/unit/select_connection_timer_tests.py b/tests/unit/select_connection_timer_tests.py
new file mode 100644
index 0000000..72e3db5
--- /dev/null
+++ b/tests/unit/select_connection_timer_tests.py
@@ -0,0 +1,353 @@
+# -*- coding: utf-8 -*-
+"""
+Tests for SelectConnection _Timer and _Timeout classes
+
+"""
+
+import time
+import unittest
+
+import mock
+
+from pika.adapters import select_connection
+
+
+# Suppress protected-access
+# pylint: disable=W0212
+
+# Suppress missing-docstring
+# pylint: disable=C0111
+
+# Suppress invalid-name
+# pylint: disable=C0103
+
+class TimeoutClassTests(unittest.TestCase):
+    """Test select_connection._Timeout class"""
+
+    def test_properties(self):
+        now = time.time()
+        cb = lambda: None
+        timeout = select_connection._Timeout(now + 5.3, cb)
+        self.assertIs(timeout.callback, cb)
+        self.assertEqual(timeout.deadline, now + 5.3)
+
+    def test_non_negative_deadline(self):
+        select_connection._Timeout(0, lambda: None)
+        select_connection._Timeout(5, lambda: None)
+
+        with self.assertRaises(ValueError) as cm:
+            select_connection._Timeout(-1, lambda: None)
+
+        self.assertIn('deadline must be non-negative epoch number',
+                      cm.exception.args[0])
+
+    def test_non_callable_callback_raises(self):
+        with self.assertRaises(TypeError) as cm:
+            select_connection._Timeout(5, None)
+
+        self.assertIn('callback must be a callable, but got',
+                      cm.exception.args[0])
+
+        with self.assertRaises(TypeError) as cm:
+            select_connection._Timeout(5, dict())
+
+        self.assertIn('callback must be a callable, but got',
+                      cm.exception.args[0])
+
+    def test_eq_operator(self):
+        # Comparison should be by deadline only
+        t1 = select_connection._Timeout(5, lambda: None)
+        t2 = select_connection._Timeout(5, lambda: 5)
+        self.assertEqual(t1, t2)
+
+        t2 = select_connection._Timeout(10, lambda: 5)
+        self.assertNotEqual(t1, t2)
+
+    def test_lt_operator(self):
+        # Comparison should be by deadline only
+        t1 = select_connection._Timeout(4, lambda: None)
+        t2 = select_connection._Timeout(5, lambda: 5)
+        self.assertLess(t1, t2)
+
+        t2 = select_connection._Timeout(4, lambda: 5)
+        self.assertFalse(t1 < t2)
+
+        t2 = select_connection._Timeout(3, lambda: 5)
+        self.assertFalse(t1 < t2)
+
+    def test_le_operator(self):
+        # Comparison should be by deadline only
+        t1 = select_connection._Timeout(4, lambda: None)
+        t2 = select_connection._Timeout(4, lambda: 5)
+        self.assertLessEqual(t1, t2)
+
+        t2 = select_connection._Timeout(5, lambda: 5)
+        self.assertLessEqual(t1, t2)
+
+        t2 = select_connection._Timeout(3, lambda: 5)
+        self.assertFalse(t1 <= t2)
+
+
+class TimerClassTests(unittest.TestCase):
+    """Test select_connection._Timer class"""
+
+    def test_close_empty(self):
+        timer = select_connection._Timer()
+        timer.close()
+        self.assertIsNone(timer._timeout_heap)
+
+    def test_close_non_empty(self):
+        timer = select_connection._Timer()
+        t1 = timer.call_later(10, lambda: 10)
+        t2 = timer.call_later(20, lambda: 20)
+        timer.close()
+        self.assertIsNone(timer._timeout_heap)
+        self.assertIsNone(t1.callback)
+        self.assertIsNone(t2.callback)
+
+    def test_no_timeouts_remaining_interval_is_none(self):
+        timer = select_connection._Timer()
+        self.assertIsNone(timer.get_remaining_interval())
+
+    def test_call_later_non_negative_delay_check(self):
+        now = time.time()
+
+        # 0 delay is okay
+        with mock.patch('time.time', return_value=now):
+            timer = select_connection._Timer()
+            timer.call_later(0, lambda: None)
+            self.assertEqual(timer._timeout_heap[0].deadline, now)
+            self.assertEqual(timer.get_remaining_interval(), 0)
+
+        # Positive delay is okay
+        with mock.patch('time.time', return_value=now):
+            timer = select_connection._Timer()
+            timer.call_later(0.5, lambda: None)
+            self.assertEqual(timer._timeout_heap[0].deadline, now + 0.5)
+            self.assertEqual(timer.get_remaining_interval(), 0.5)
+
+        # Negative delay raises ValueError
+        timer = select_connection._Timer()
+        with self.assertRaises(ValueError) as cm:
+            timer.call_later(-5, lambda: None)
+        self.assertIn('call_later: delay must be non-negative, but got',
+                      cm.exception.args[0])
+
+    def test_call_later_single_timer_expires(self):
+        now = time.time()
+
+        with mock.patch('time.time', return_value=now):
+            bucket = []
+            timer = select_connection._Timer()
+            timer.call_later(5, lambda: bucket.append(1))
+
+            # Nothing is ready to expire
+            timer.process_timeouts()
+            self.assertEqual(bucket, [])
+            self.assertEqual(timer.get_remaining_interval(), 5)
+
+        # Advance time by 5 seconds and expect the timer to expire
+        with mock.patch('time.time', return_value=now + 5):
+            self.assertEqual(timer.get_remaining_interval(), 0)
+            timer.process_timeouts()
+            self.assertEqual(bucket, [1])
+            self.assertEqual(len(timer._timeout_heap), 0)
+            self.assertIsNone(timer.get_remaining_interval())
+
+    def test_call_later_multiple_timers(self):
+        now = time.time()
+
+        bucket = []
+        timer = select_connection._Timer()
+
+        with mock.patch('time.time', return_value=now):
+            timer.call_later(5, lambda: bucket.append(1))
+            timer.call_later(5, lambda: bucket.append(2))
+            timer.call_later(10, lambda: bucket.append(3))
+
+            # Nothing is ready to fire yet
+            self.assertEqual(timer.get_remaining_interval(), 5)
+            timer.process_timeouts()
+            self.assertEqual(bucket, [])
+            self.assertEqual(timer.get_remaining_interval(), 5)
+
+        # Advance time by 6 seconds and expect first two timers to expire
+        with mock.patch('time.time', return_value=now + 6):
+            self.assertEqual(timer.get_remaining_interval(), 0)
+            timer.process_timeouts()
+            self.assertEqual(bucket, [1, 2])
+            self.assertEqual(len(timer._timeout_heap), 1)
+            self.assertEqual(timer.get_remaining_interval(), 4)
+
+        # Advance time by 10 seconds and expect the 3rd timeout to expire
+        with mock.patch('time.time', return_value=now + 10):
+            self.assertEqual(timer.get_remaining_interval(), 0)
+            timer.process_timeouts()
+            self.assertEqual(bucket, [1, 2, 3])
+            self.assertEqual(len(timer._timeout_heap), 0)
+            self.assertIsNone(timer.get_remaining_interval())
+
+    def test_add_and_remove_timeout(self):
+        now = time.time()
+
+        bucket = []
+        timer = select_connection._Timer()
+
+        with mock.patch('time.time', return_value=now):
+            timer.call_later(10, lambda: bucket.append(3)) # t3
+            t2 = timer.call_later(6, lambda: bucket.append(2))
+            t1 = timer.call_later(5, lambda: bucket.append(1))
+
+            # Nothing is ready to fire yet
+            self.assertEqual(timer.get_remaining_interval(), 5)
+            timer.process_timeouts()
+            self.assertEqual(bucket, [])
+            self.assertEqual(timer.get_remaining_interval(), 5)
+
+            # Cancel t1 and t2 that haven't expired yet
+            timer.remove_timeout(t1)
+            self.assertIsNone(t1.callback)
+            self.assertEqual(timer._num_cancellations, 1)
+            timer.remove_timeout(t2)
+            self.assertIsNone(t2.callback)
+            self.assertEqual(timer._num_cancellations, 2)
+            self.assertEqual(timer.get_remaining_interval(), 5)
+            timer.process_timeouts()
+            self.assertEqual(bucket, [])
+            self.assertEqual(timer._num_cancellations, 2)
+            self.assertEqual(timer.get_remaining_interval(), 5)
+            self.assertEqual(len(timer._timeout_heap), 3)
+
+        # Advance time by 6 seconds to expire t1 and t2 and verify they don't
+        # fire
+        with mock.patch('time.time', return_value=now + 6):
+            self.assertEqual(timer.get_remaining_interval(), 0)
+            timer.process_timeouts()
+            self.assertEqual(bucket, [])
+            self.assertEqual(timer._num_cancellations, 0)
+            self.assertEqual(len(timer._timeout_heap), 1)
+            self.assertEqual(timer.get_remaining_interval(), 4)
+
+        # Advance time by 10 seconds to expire t3 and verify it fires
+        with mock.patch('time.time', return_value=now + 10):
+            self.assertEqual(timer.get_remaining_interval(), 0)
+            timer.process_timeouts()
+            self.assertEqual(bucket, [3])
+            self.assertEqual(len(timer._timeout_heap), 0)
+            self.assertIsNone(timer.get_remaining_interval())
+
+    def test_gc_of_unexpired_timeouts(self):
+        now = time.time()
+        bucket = []
+        timer = select_connection._Timer()
+
+        with mock.patch.multiple(select_connection._Timer,
+                                 _GC_CANCELLATION_THRESHOLD=1):
+            with mock.patch('time.time', return_value=now):
+                t3 = timer.call_later(10, lambda: bucket.append(3))
+                t2 = timer.call_later(6, lambda: bucket.append(2))
+                t1 = timer.call_later(5, lambda: bucket.append(1))
+
+                # Cancel t1 and check that it doesn't trigger GC because it's
+                # not greater than half the timeouts
+                timer.remove_timeout(t1)
+                self.assertEqual(timer._num_cancellations, 1)
+                timer.process_timeouts()
+                self.assertEqual(timer._num_cancellations, 1)
+                self.assertEqual(bucket, [])
+                self.assertEqual(len(timer._timeout_heap), 3)
+                self.assertEqual(timer.get_remaining_interval(), 5)
+
+                # Cancel t3 and verify GC since it's now greater than half of
+                # total timeouts
+                timer.remove_timeout(t3)
+                self.assertEqual(timer._num_cancellations, 2)
+                timer.process_timeouts()
+                self.assertEqual(bucket, [])
+                self.assertEqual(len(timer._timeout_heap), 1)
+                self.assertIs(t2, timer._timeout_heap[0])
+                self.assertEqual(timer.get_remaining_interval(), 6)
+                self.assertEqual(timer._num_cancellations, 0)
+
+    def test_add_timeout_from_another_timeout(self):
+        now = time.time()
+        bucket = []
+        timer = select_connection._Timer()
+
+        with mock.patch('time.time', return_value=now):
+            t1 = timer.call_later(
+                5,
+                lambda: bucket.append(
+                    timer.call_later(0, lambda: bucket.append(2))))
+
+        # Advance time by 10 seconds and verify that t1 fires and creates t2,
+        # but timer manager defers firing of t2 to next `process_timeouts` in
+        # order to avoid IO starvation
+        with mock.patch('time.time', return_value=now + 10):
+            timer.process_timeouts()
+            t2 = bucket.pop()
+            self.assertIsInstance(t2, select_connection._Timeout)
+            self.assertIsNot(t2, t1)
+            self.assertEqual(bucket, [])
+            self.assertEqual(len(timer._timeout_heap), 1)
+            self.assertIs(t2, timer._timeout_heap[0])
+            self.assertEqual(timer.get_remaining_interval(), 0)
+
+            # t2 should now fire
+            timer.process_timeouts()
+            self.assertEqual(bucket, [2])
+            self.assertEqual(timer.get_remaining_interval(), None)
+
+    def test_cancel_unexpired_timeout_from_another_timeout(self):
+        now = time.time()
+        bucket = []
+        timer = select_connection._Timer()
+
+        with mock.patch('time.time', return_value=now):
+            t2 = timer.call_later(10, lambda: bucket.append(2))
+            t1 = timer.call_later(5, lambda: timer.remove_timeout(t2))
+
+            self.assertIs(t1, timer._timeout_heap[0])
+
+        # Advance time by 6 seconds and check that t2 is cancelled, but not
+        # removed from timeout heap
+        with mock.patch('time.time', return_value=now + 6):
+            timer.process_timeouts()
+            self.assertIsNone(t2.callback)
+            self.assertEqual(timer.get_remaining_interval(), 4)
+            self.assertIs(t2, timer._timeout_heap[0])
+            self.assertEqual(timer._num_cancellations, 1)
+
+        # Advance time by 10 seconds and verify that t2 is removed without
+        # firing
+        with mock.patch('time.time', return_value=now + 10):
+            timer.process_timeouts()
+            self.assertEqual(bucket, [])
+            self.assertIsNone(timer.get_remaining_interval())
+            self.assertEqual(len(timer._timeout_heap), 0)
+            self.assertEqual(timer._num_cancellations, 0)
+
+
+    def test_cancel_expired_timeout_from_another_timeout(self):
+        now = time.time()
+        bucket = []
+        timer = select_connection._Timer()
+
+        with mock.patch('time.time', return_value=now):
+            t2 = timer.call_later(10, lambda: bucket.append(2))
+            t1 = timer.call_later(
+                5,
+                lambda: (self.assertEqual(timer._num_cancellations, 0),
+                         timer.remove_timeout(t2)))
+
+            self.assertIs(t1, timer._timeout_heap[0])
+
+        # Advance time by 10 seconds and check that t2 is cancelled and
+        # removed from timeout heap
+        with mock.patch('time.time', return_value=now + 10):
+            timer.process_timeouts()
+            self.assertEqual(bucket, [])
+            self.assertIsNone(t2.callback)
+            self.assertIsNone(timer.get_remaining_interval())
+            self.assertEqual(len(timer._timeout_heap), 0)
+            self.assertEqual(timer._num_cancellations, 0)