diff --git a/tornado/ioloop.py b/tornado/ioloop.py
index 839e7ee54ef0431ff961e360f7537a008856aed6..f6ec177bd8c009b1879a01d0da56468b47735fd9 100644
--- a/tornado/ioloop.py
+++ b/tornado/ioloop.py
@@ -47,7 +47,6 @@ import threading
 import time
 import traceback
 import math
-import weakref
 
 from tornado.concurrent import Future, is_future, chain_future, future_set_exc_info, future_add_done_callback  # noqa: E501
 from tornado.log import app_log, gen_log
@@ -185,7 +184,7 @@ class IOLoop(Configurable):
     _current = threading.local()
 
     # In Python 3, _ioloop_for_asyncio maps from asyncio loops to IOLoops.
-    _ioloop_for_asyncio = weakref.WeakKeyDictionary()
+    _ioloop_for_asyncio = dict()
 
     @classmethod
     def configure(cls, impl, **kwargs):
diff --git a/tornado/platform/asyncio.py b/tornado/platform/asyncio.py
index b2ad9fe6ef030193689ebef8e719e60495bb5839..b6a490afac5c76727d5dbb280177cc0604418d54 100644
--- a/tornado/platform/asyncio.py
+++ b/tornado/platform/asyncio.py
@@ -38,6 +38,20 @@ class BaseAsyncIOLoop(IOLoop):
         self.readers = set()
         self.writers = set()
         self.closing = False
+        # If an asyncio loop was closed through an asyncio interface
+        # instead of IOLoop.close(), we'd never hear about it and may
+        # have left a dangling reference in our map. In case an
+        # application (or, more likely, a test suite) creates and
+        # destroys a lot of event loops in this way, check here to
+        # ensure that we don't have a lot of dead loops building up in
+        # the map.
+        #
+        # TODO(bdarnell): consider making self.asyncio_loop a weakref
+        # for AsyncIOMainLoop and make _ioloop_for_asyncio a
+        # WeakKeyDictionary.
+        for loop in list(IOLoop._ioloop_for_asyncio):
+            if loop.is_closed():
+                del IOLoop._ioloop_for_asyncio[loop]
         IOLoop._ioloop_for_asyncio[asyncio_loop] = self
         super(BaseAsyncIOLoop, self).initialize(**kwargs)
 
@@ -49,6 +63,7 @@ class BaseAsyncIOLoop(IOLoop):
             if all_fds:
                 self.close_fd(fileobj)
         self.asyncio_loop.close()
+        del IOLoop._ioloop_for_asyncio[self.asyncio_loop]
 
     def add_handler(self, fd, handler, events):
         fd, fileobj = self.split_fd(fd)
diff --git a/tornado/test/asyncio_test.py b/tornado/test/asyncio_test.py
index 37f2f6e1b4d3e18efa3299c34f5a6dfd8b5af3f7..a7c7564964b2b97521413c3a15915d56293101ab 100644
--- a/tornado/test/asyncio_test.py
+++ b/tornado/test/asyncio_test.py
@@ -122,6 +122,44 @@ class AsyncIOLoopTest(AsyncTestCase):
             42)
 
 
+@unittest.skipIf(asyncio is None, "asyncio module not present")
+class LeakTest(unittest.TestCase):
+    def setUp(self):
+        # Trigger a cleanup of the mapping so we start with a clean slate.
+        AsyncIOLoop().close()
+        # If we don't clean up after ourselves other tests may fail on
+        # py34.
+        self.orig_policy = asyncio.get_event_loop_policy()
+        asyncio.set_event_loop_policy(asyncio.DefaultEventLoopPolicy())
+
+    def tearDown(self):
+        asyncio.get_event_loop().close()
+        asyncio.set_event_loop_policy(self.orig_policy)
+
+    def test_ioloop_close_leak(self):
+        orig_count = len(IOLoop._ioloop_for_asyncio)
+        for i in range(10):
+            # Create and close an AsyncIOLoop using Tornado interfaces.
+            loop = AsyncIOLoop()
+            loop.close()
+        new_count = len(IOLoop._ioloop_for_asyncio) - orig_count
+        self.assertEqual(new_count, 0)
+
+    def test_asyncio_close_leak(self):
+        orig_count = len(IOLoop._ioloop_for_asyncio)
+        for i in range(10):
+            # Create and close an AsyncIOMainLoop using asyncio interfaces.
+            loop = asyncio.new_event_loop()
+            loop.call_soon(IOLoop.current)
+            loop.call_soon(loop.stop)
+            loop.run_forever()
+            loop.close()
+        new_count = len(IOLoop._ioloop_for_asyncio) - orig_count
+        # Because the cleanup is run on new loop creation, we have one
+        # dangling entry in the map (but only one).
+        self.assertEqual(new_count, 1)
+
+
 @unittest.skipIf(asyncio is None, "asyncio module not present")
 class AnyThreadEventLoopPolicyTest(unittest.TestCase):
     def setUp(self):