Repository: pika/pika

Issue:
Question re. interrupting `BlockingChannel.consume` from another thread
In [Dramatiq][dramatiq] I've got an architecture where consumers live on their own thread and call `BlockingChannel`'s `consume` method with an inactivity timeout of `1s` by default. Whenever a consumer gets a message from Rabbit, it puts it on a shared in-memory work queue where it is grabbed and processed by any of `n` worker threads. Once the worker thread is done processing the message, it gets put on an "acks" queue that belongs to the consumer. When the consumer hits the inactivity timeout it acks any pending messages off its queue and then waits for new messages from RMQ.

Acking from the worker threads is not an option since `BlockingChannel`s are not thread safe.

When you set a QoS value the fact that `consume` can't be interrupted from another thread really starts to hurt throughput, at which point you are forced to either:

1. use a tiny `inactivity_timeout` (<100ms), which "fixes" the throughput issue but is essentially a spinlock so it's inefficient,
2. switch to `SelectConnection` and implement your own interrupt, this is probably what any sane person would do, but the problem is you'd end up implementing a lot of stuff that `BlockingChannel` already implements, or
3. hack your own interrupt by hooking into implementation details of `BlockingChannel`.

The latter is what I ended up doing ([here][c1], [here][c2] and [here][c3]) and my interrupt implementation is working well, but I don't feel too good about depending on `BlockingConnection`'s implementation details so my "issue" is more of a question than anything: is there a better way I could be doing this and, if not, would this be a good addition to `BlockingConnection` (I'd be happy to contribute)? 

[dramatiq]: http://dramatiq.io/
[c1]: https://github.com/Bogdanp/dramatiq/blob/master/dramatiq/brokers/rabbitmq.py#L359-L373
[c2]: https://github.com/Bogdanp/dramatiq/blob/master/dramatiq/brokers/rabbitmq.py#L422-L435
[c3]: https://github.com/Bogdanp/dramatiq/blob/master/dramatiq/brokers/rabbitmq.py#L439-L458

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 cc38c3f..ebfd179 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):
         """Add the callback 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 fdc456b..ba814bd 100644
--- a/pika/adapters/base_connection.py
+++ b/pika/adapters/base_connection.py
@@ -127,6 +127,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 2a1629f..3a845f8 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
@@ -158,7 +161,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,))
 
@@ -380,7 +383,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()
@@ -527,6 +530,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
 
@@ -622,9 +637,8 @@ class BlockingConnection(object):
 
         """
         if not callable(callback):
-            raise ValueError(
-                'callback parameter must be callable, but got %r'
-                % (callback,))
+            raise TypeError(
+                'callback must be a callable, but got %r' % (callback,))
 
         evt = _TimerEvt(callback=callback)
         timer_id = self._impl.add_timeout(
@@ -634,6 +648,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
 
@@ -877,7 +914,7 @@ class _ConsumerCancellationEvt(_ChannelPendingEvt):
     `Basic.Cancel`
     """
 
-    __slots__ = 'method_frame'
+    __slots__ = ('method_frame',)
 
     def __init__(self, method_frame):
         """
@@ -1802,7 +1839,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:
 
@@ -2397,7 +2435,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 3ae3eca..85922ce 100644
--- a/pika/adapters/select_connection.py
+++ b/pika/adapters/select_connection.py
@@ -3,6 +3,7 @@ platform pika is running on.
 
 """
 import abc
+import collections
 import errno
 import functools
 import heapq
@@ -11,8 +12,6 @@ import select
 import time
 import threading
 
-from collections import defaultdict
-
 import pika.compat
 
 from pika.adapters.base_connection import BaseConnection
@@ -164,6 +163,16 @@ class _Timer(object):
         # 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.
 
@@ -283,9 +292,28 @@ class IOLoop(object):
     def __init__(self):
         self._timer = _Timer()
 
-        self._poller = self._get_poller(self._timer.get_remaining_interval,
+        # 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(get_wait_seconds, process_timeouts):
         """Determine the best poller to use for this environment and instantiate
@@ -347,13 +375,57 @@ class IOLoop(object):
         """
         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 timeouts, invoking callbacks for those
+        """[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
 
@@ -386,20 +458,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.
+
+        To invoke `stop()` safely from a thread other than this IOLoop's thread,
+        call it via `add_callback_threadsafe`; e.g.,
+
+            `ioloop.add_callback_threadsafe(ioloop.stop)`
 
         """
+        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):
@@ -443,6 +526,10 @@ class _PollerBase(_AbstractBase):  # pylint: disable=R0902
         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()
 
@@ -456,14 +543,60 @@ class _PollerBase(_AbstractBase):  # pylint: disable=R0902
 
         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)
+
+    def close(self):
+        """Release poller's resources.
+
+        `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.
+
+        """
+        # 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.
+
+        """
+        with self._waking_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 interrupt to poller: %s", err)
+                raise
 
-        # ioloop-interrupt socket pair; initialized in start()
-        self._r_interrupt = None
-        self._w_interrupt = None
 
     def _get_max_wait(self):
         """Get the interval to the next timeout event, or a default interval
@@ -558,7 +691,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
@@ -580,22 +713,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)
@@ -610,46 +731,26 @@ class _PollerBase(_AbstractBase):  # pylint: disable=R0902
             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):
@@ -797,7 +898,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
@@ -856,9 +957,8 @@ class KQueuePoller(_PollerBase):
     def __init__(self, get_wait_seconds, process_timeouts):
         """Create an instance of the KQueuePoller
         """
-        super(KQueuePoller, self).__init__(get_wait_seconds, process_timeouts)
-
         self._kqueue = None
+        super(KQueuePoller, self).__init__(get_wait_seconds, process_timeouts)
 
     @staticmethod
     def _map_event(kevent):
@@ -894,7 +994,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)
 
@@ -908,8 +1008,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
@@ -1013,7 +1114,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
 
@@ -1027,10 +1128,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 3c3c736..c8369df 100644
--- a/pika/adapters/tornado_connection.py
+++ b/pika/adapters/tornado_connection.py
@@ -79,3 +79,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 72c7c2e..d48487a 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 60d40ca..03bfb5e 100644
--- a/pika/connection.py
+++ b/pika/connection.py
@@ -1244,8 +1244,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)


Test patch:
diff --git a/tests/acceptance/async_adapter_tests.py b/tests/acceptance/async_adapter_tests.py
index 00b87ee..d392d04 100644
--- a/tests/acceptance/async_adapter_tests.py
+++ b/tests/acceptance/async_adapter_tests.py
@@ -11,6 +11,7 @@
 # pylint: disable=W0613
 
 import functools
+import threading
 import time
 import uuid
 
@@ -37,6 +38,7 @@ class TestConfirmSelect(AsyncTestCase, AsyncAdapters):
         channel.confirm_delivery(ack_nack_callback=self.ack_nack_callback,
                                  callback=self.on_complete)
 
+    @staticmethod
     def ack_nack_callback(frame):
         pass
 
@@ -493,6 +495,114 @@ class TestBlockedConnectionUnblocks(AsyncTestCase, AsyncAdapters):  # pylint: di
                                                              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"
 
diff --git a/tests/acceptance/async_test_base.py b/tests/acceptance/async_test_base.py
index ac017f9..db01e48 100644
--- a/tests/acceptance/async_test_base.py
+++ b/tests/acceptance/async_test_base.py
@@ -52,34 +52,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.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"""
@@ -99,7 +120,7 @@ class AsyncTestCase(unittest.TestCase):
     def on_timeout(self):
         """called when stuck waiting for connection to close"""
         self.logger.error('%s timed out; on_timeout called at %s',
-            self, datetime.utcnow())
+                          self, datetime.utcnow())
         self.timeout = None  # the dispatcher should have removed it
         self._timed_out = True
         # initiate cleanup
@@ -108,12 +129,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.exchange,
@@ -150,20 +171,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
@@ -172,27 +204,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 287b486..d9db804 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,48 @@ 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):
 
     def test(self):
@@ -1495,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
@@ -1654,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)
 
@@ -1669,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(
+            q_name,
+            ackAndEnqueueMessageViaAnotherThread,
+            auto_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, auto_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
@@ -1939,7 +2174,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 69a80e5..f303023 100644
--- a/tests/unit/blocking_connection_tests.py
+++ b/tests/unit/blocking_connection_tests.py
@@ -125,8 +125,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,
@@ -151,8 +150,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,
@@ -177,8 +175,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 62bf3f1..3f40222 100644
--- a/tests/unit/select_connection_ioloop_tests.py
+++ b/tests/unit/select_connection_ioloop_tests.py
@@ -19,6 +19,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')
@@ -36,6 +42,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,
@@ -73,16 +80,79 @@ 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()
+        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
 
 
 @unittest.skipIf(not POLL_SUPPORTED, 'poll not supported')
@@ -438,6 +508,7 @@ class IOLoopEintrTestCaseSelect(IOLoopBaseTest):
         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)
@@ -493,6 +564,7 @@ class SelectPollerTestPollWithoutSockets(unittest.TestCase):
         poller = select_connection.SelectPoller(
             get_wait_seconds=timer.get_remaining_interval,
             process_timeouts=timer.process_timeouts)
+        self.addCleanup(poller.close)
 
         timer_call_container = []
         timer.call_later(0.00001, lambda: timer_call_container.append(1))
@@ -514,3 +586,46 @@ class SelectPollerTestPollWithoutSockets(unittest.TestCase):
                 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
index 1017e7e..72e3db5 100644
--- a/tests/unit/select_connection_timer_tests.py
+++ b/tests/unit/select_connection_timer_tests.py
@@ -91,6 +91,20 @@ class TimeoutClassTests(unittest.TestCase):
 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())