Add ThreadPool.abort() to stop processing early.

tests/threading_utils_test.py

 #!/usr/bin/env python
 # Copyright 2013 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 # Lambda may not be necessary.
 # pylint: disable=W0108
 
 import functools
 import logging
@@ skipping 248 matching lines @@
         return x
 
       with lock:
         pool.add_task(0, wait_and_return, 'a')
         pool.add_task(2, return_x, 'b')
         pool.add_task(1, return_x, 'c')
 
       actual = pool.join()
     self.assertEqual(['a', 'c', 'b'], actual)
 
+  @timeout(2)
+  def test_abort(self):
+    # Trigger a ridiculous amount of tasks, and abort the remaining.
+    with threading_utils.ThreadPool(2, 2, 0) as pool:
+      # Allow 10 tasks to run initially.
+      sem = threading.Semaphore(10)
+
+      def grab_and_return(x):
+        sem.acquire()
+        return x
+
+      for i in range(100):
+        pool.add_task(0, grab_and_return, i)
+
+      # Running at 11 would hang.
+      results = [pool.get_one_result() for _ in xrange(10)]
+      # At that point, there's 10 completed tasks and 2 tasks hanging, 88
+      # pending.
+      self.assertEqual(88, pool.abort())
+      # Calling .join() before these 2 .release() would hang.
+      sem.release()
+      sem.release()
+      results.extend(pool.join())
+    # The results *may* be out of order. Even if the calls are processed
+    # strictly in FIFO mode, a thread may preempt another one when returning the
+    # values.
+    self.assertEqual(range(12), sorted(results))
+
 
 class AutoRetryThreadPoolTest(unittest.TestCase):
   def test_bad_class(self):
     exceptions = [AutoRetryThreadPoolTest]
     with self.assertRaises(AssertionError):
       threading_utils.AutoRetryThreadPool(exceptions, 1, 0, 1, 0)
 
   def test_no_exception(self):
     with self.assertRaises(AssertionError):
       threading_utils.AutoRetryThreadPool([], 1, 0, 1, 0)
@@ skipping 224 matching lines @@
         raise CustomError()
       tp.add_task(0, channel.wrap_task(task_func))
       with self.assertRaises(CustomError):
         channel.pull()
 
 
 if __name__ == '__main__':
   VERBOSE = '-v' in sys.argv
   logging.basicConfig(level=logging.DEBUG if VERBOSE else logging.ERROR)
   unittest.main()