Fix a leak in AsyncHTTPClient shutdown.

Clients created without force_instance=True were never getting
GC'd because we were using the wrong reference for the async_clients
instance cache.

To guard against future errors, failure to remove an instance
from the cache when we expected to now raises an error instead of
failing silently.

Closes #1079.

tornado/httpclient.py

@@ -144,12 +144,21 @@ class AsyncHTTPClient(Configurable):
 
     def __new__(cls, io_loop=None, force_instance=False, **kwargs):
         io_loop = io_loop or IOLoop.current()
-        if io_loop in cls._async_clients() and not force_instance:
-            return cls._async_clients()[io_loop]
+        if force_instance:
+            instance_cache = None
+        else:
+            instance_cache = cls._async_clients()
+        if instance_cache is not None and io_loop in instance_cache:
+            return instance_cache[io_loop]
         instance = super(AsyncHTTPClient, cls).__new__(cls, io_loop=io_loop,
                                                        **kwargs)
-        if not force_instance:
-            cls._async_clients()[io_loop] = instance
+        # Make sure the instance knows which cache to remove itself from.
+        # It can't simply call _async_clients() because we may be in
+        # __new__(AsyncHTTPClient) but instance.__class__ may be
+        # SimpleAsyncHTTPClient.
+        instance._instance_cache = instance_cache
+        if instance_cache is not None:
+            instance_cache[instance.io_loop] = instance
         return instance
 
     def initialize(self, io_loop, defaults=None):
@@ -172,9 +181,13 @@ class AsyncHTTPClient(Configurable):
         ``close()``.
 
         """
+        if self._closed:
+            return
         self._closed = True
-        if self._async_clients().get(self.io_loop) is self:
-            del self._async_clients()[self.io_loop]
+        if self._instance_cache is not None:
+            if self._instance_cache.get(self.io_loop) is not self:
+                raise RuntimeError("inconsistent AsyncHTTPClient cache")
+            del self._instance_cache[self.io_loop]
 
     def fetch(self, request, callback=None, **kwargs):
         """Executes a request, asynchronously returning an `HTTPResponse`.