diff -ur cpython.ori/Lib/multiprocessing/pool.py cpython/Lib/multiprocessing/pool.py
--- cpython.ori/Lib/multiprocessing/pool.py	2011-04-12 12:43:31.000000000 +0000
+++ cpython/Lib/multiprocessing/pool.py	2011-04-12 15:36:26.000000000 +0000
@@ -319,7 +319,12 @@
 
     @staticmethod
     def _handle_workers(pool):
-        while pool._worker_handler._state == RUN and pool._state == RUN:
+        thread = threading.current_thread()
+
+        # if the pool is closed, we must continue to maintain workers to drain
+        # the cache. if it's terminated, exit right away 
+        while thread._state == RUN or (pool._cache and
+                                       thread._state != TERMINATE):
             pool._maintain_pool()
             time.sleep(0.1)
         # send sentinel to stop workers
diff -ur cpython.ori/Lib/test/test_multiprocessing.py cpython/Lib/test/test_multiprocessing.py
--- cpython.ori/Lib/test/test_multiprocessing.py	2011-04-12 16:08:52.000000000 +0000
+++ cpython/Lib/test/test_multiprocessing.py	2011-04-12 16:18:02.000000000 +0000
@@ -1196,6 +1196,19 @@
         p.close()
         p.join()
 
+    def test_pool_worker_lifetime_async_close(self):
+        p = multiprocessing.Pool(3, maxtasksperchild=1)
+        results = []
+        for i in range(20):
+            results.append(p.apply_async(sqr, (i, )))
+        # close the pool before the tasks have completed - the join should
+        # block until completion
+        p.close()
+        p.join()
+        # check the results
+        for (j, res) in enumerate(results):
+            self.assertEqual(res.get(), sqr(j))
+
 #
 # Test that manager has expected number of shared objects left
 #