multiprocessing.Queue is running background thread feeder. Feeder serialize and sends buffered data to pipe.

The issue is with exception handling, feeder is catching all exceptions but out of main loop, so after exception is handled feeder is not going back to loop - thread finish. If feeder thread is not running any Queue.put will execute without exceptions but message not gonna be delivered.

Solution is to move exception handling inside main loop. I will provide PR.

I have run performance tests (found: bpo-17025) and submitted patch do not affect performance.

Can you expand on which exceptions you are getting in the feeder thread?

Nevermind, I saw the PR and the test case.

New changeset bc50f03 by Antoine Pitrou (grzgrzgrz3) in branch 'master':
bpo-30414: multiprocessing.Queue._feed do not break from main loop on exc (bpo-1683)
bc50f03

New changeset 2783cc4 by Antoine Pitrou in branch '3.6':
[3.6] bpo-30414: multiprocessing.Queue._feed do not break from main loop on exc (GH-1683) (bpo-1815)
2783cc4

New changeset 89004d7 by Antoine Pitrou in branch '3.5':
[3.5] bpo-30414: multiprocessing.Queue._feed do not break from main loop on exc (GH-1683) (bpo-1816)
89004d7

New changeset bdd9647 by Antoine Pitrou in branch '2.7':
[2.7] bpo-30414: multiprocessing.Queue._feed do not break from main loop on exc (GH-1683) (bpo-1817)
bdd9647

This is committed and pushed, thank you!

This fix, while preventing the Queue to crash, does not give any way to programatically detect that the message was dropped. This is a problem as we can no longer assume that the Queue will not drop messages. For instance, we can no longer detect deadlocks in concurrent.futures.ProcessPoolExecutor as done in #1013 where the crashed QueueFeederThread was used to monitor the working state of the executor.

We could either:

• Put a flag highlighting the fact that some messages where dropped.
• Add an argument to the Queue to close on pickling errors.

I'd be happy to work on a PR to implement any solution that you think is reasonable.

Thomas, thanks for the heads up. I would suggest something like the following patch to multiprocessing.Pool:

$ git diff
diff --git a/Lib/multiprocessing/queues.py b/Lib/multiprocessing/queues.py
index 7f77837..ebbb360 100644
--- a/Lib/multiprocessing/queues.py
+++ b/Lib/multiprocessing/queues.py
@@ -260,8 +260,16 @@ class Queue(object):
                     info('error in queue thread: %s', e)
                     return
                 else:
-                    import traceback
-                    traceback.print_exc()
+                    self._on_queue_thread_error(e)
+
+    def _on_queue_thread_error(self, e):
+        """
+        Private API called when feeding data in the background thread
+        raises an exception.  For overriding by concurrent.futures.
+        """
+        import traceback
+        traceback.print_exc()
+
 
 _sentinel = object()
 

Then you can write your own Queue subclass in concurrent.futures to handle that error and clean up/restart whatever needs to be cleaned up or restarted. What do you think?

I think this is a good solution as it let the user define easily the behavior it needs in other situation too. I would recommend adding the object responsible for the failure to the _on_queue_thread_error callback. This would simplify the error handling.

@@ -260,8 +260,16 @@ class Queue(object):
                     info('error in queue thread: %s', e)
                     return
                 else:
-                    import traceback
-                    traceback.print_exc()
+                    self._on_queue_thread_error(e, obj)
+
+    def _on_queue_thread_error(self, e, obj):
+        """
+        Private API called when feeding data in the background thread
+        raises an exception.  For overriding by concurrent.futures.
+        """
+        import traceback
+        traceback.print_exc()
+

We started to get *random* failures of test_queue_feeder_donot_stop_onexc() of test_multiprocessing_spawn since a few days. I may be related to this change. Can you please take a look at bpo-30595?