Fix deadlock if futures submit to CancelOnShutdown during shutdown

Fixes #98

README.md

@@ -70,6 +70,8 @@ def fetch_urls(urls):
 
 ### v1.15.0
 
+- Fixed possible deadlock in CancelOnShutdownExecutor
+  ([#98](https://github.com/rohanpm/more-executors/issues/98))
 - Fixed `Executors.bind` with `functools.partial`
   ([#96](https://github.com/rohanpm/more-executors/issues/96))
 

more_executors/cancel_on_shutdown.py

@@ -42,12 +42,13 @@ def shutdown(self, wait=True):
             if self._shutdown:
                 return
             self._shutdown = True
+            futures = self._futures.copy()
 
-            for f in self._futures.copy():
-                cancel = f.cancel()
-                self._log.debug("Cancel %s: %s", f, cancel)
+        for f in futures:
+            cancel = f.cancel()
+            self._log.debug("Cancel %s: %s", f, cancel)
 
-            self._delegate.shutdown(wait)
+        self._delegate.shutdown(wait)
 
     def submit(self, fn, *args, **kwargs):
         with self._lock:

tests/test_cancel_on_shutdown.py

@@ -1,4 +1,5 @@
 from threading import Event
+import time
 
 from hamcrest import assert_that, equal_to, is_in, calling, raises, \
                      less_than_or_equal_to, greater_than_or_equal_to
@@ -82,3 +83,36 @@ def submit_more(f):
 
     # And the tests in submit_more should have run
     assert_that(submit_more_done[0], equal_to(True))
+
+
+def test_submit_during_shutdown_no_deadlock():
+    proceed = Event()
+    submit_more_done = [False]
+
+    executor = Executors.thread_pool(max_workers=2).with_cancel_on_shutdown()
+
+    def submit_more():
+        proceed.wait()
+        assert_that(calling(executor.submit).with_args(lambda: None),
+                    raises(RuntimeError, 'Cannot submit after shutdown'))
+        submit_more_done[0] = True
+
+    futures = [executor.submit(submit_more)
+               for _ in (1, 2, 3)]
+
+    # Let threads proceed only after shutdown has already started...
+    def set_soon():
+        time.sleep(0.10)
+        proceed.set()
+
+    proceed_f = Executors.thread_pool(max_workers=1).submit(set_soon)
+    executor.shutdown(wait=True)
+    proceed_f.result()
+
+    # It should have reached here without deadlocking
+
+    # All futures should have completed
+    assert all([f.done() for f in futures])
+
+    # And the tests in submit_more should have run
+    assert_that(submit_more_done[0], equal_to(True))