title: multiprocessing.Queue fails intermittently. Bug in Python?
body: <p>Python's <code>multiprocessing.Queue</code>fails intermittently, and I don't know why. Is this a bug in Python or my script?</p>
  
  <h1>Minimal failing script</h1>
  
  <pre><code>import multiprocessing
  import time
  import logging
  import multiprocessing.util
  multiprocessing.util.log_to_stderr(level=logging.DEBUG)
  
  queue = multiprocessing.Queue(maxsize=10)
  
  def worker(queue):
      queue.put('abcdefghijklmnop')
  
      # "Indicate that no more data will be put on this queue by the
      # current process." --Documentation
      # time.sleep(0.01)
      queue.close()
  
  proc = multiprocessing.Process(target=worker, args=(queue,))
  proc.start()
  
  # "Indicate that no more data will be put on this queue by the current
  # process." --Documentation
  # time.sleep(0.01)
  queue.close()
  
  proc.join()
  </code></pre>
  
  <p>I am testing this in CPython 3.6.6 in Debian. It also fails with docker <code>python:3.7.0-alpine</code>.</p>
  
  <pre><code>docker run --rm -v "${PWD}/test.py:/test.py" \
      python:3-alpine python3 /test.py
  </code></pre>
  
  <p>The above script <strong>sometimes</strong> fails with a BrokenPipeError.</p>
  
  <pre><code>Traceback (most recent call last):
    File "/usr/lib/python3.6/multiprocessing/queues.py", line 240, in _feed
      send_bytes(obj)
    File "/usr/lib/python3.6/multiprocessing/connection.py", line 200, in send_bytes
      self._send_bytes(m[offset:offset + size])
    File "/usr/lib/python3.6/multiprocessing/connection.py", line 404, in _send_bytes
      self._send(header + buf)
    File "/usr/lib/python3.6/multiprocessing/connection.py", line 368, in _send
      n = write(self._handle, buf)
  BrokenPipeError: [Errno 32] Broken pipe
  </code></pre>
  
  <h1>Test harness</h1>
  
  <p>Because this is intermittent, I wrote a shell script to call it many times and count the failures.</p>
  
  <pre><code>#!/bin/sh
  total=10
  
  successes=0
  for i in `seq ${total}`
  do
      if ! docker run --rm -v "${PWD}/test.py:/test.py" python:3-alpine \
           python3 test.py 2&gt;&amp;1 \
           | grep --silent BrokenPipeError
      then
          successes=$(expr ${successes} + 1)
      fi
  done
  python3 -c "print(${successes} / ${total})"
  </code></pre>
  
  <p>This usually shows some fraction, maybe 0.2 indicating intermittent failures.</p>
  
  <h1>Timing adjustments</h1>
  
  <p>If I insert <code>time.sleep(0.01)</code> before either <code>queue.close()</code>, it works consistently. I noticed in the <a href="https://github.com/python/cpython/blob/3.7/Lib/multiprocessing/queues.py#L160" rel="noreferrer">source code</a> that writing happens in its own thread. I think if the writing thread is still trying to write data and all of the other threads close the queue, then it causes the error.</p>
  
  <h1>Debug logs</h1>
  
  <p>By uncommenting the first few lines, I can trace the execution for failures and successes.</p>
  
  <p>Failure:</p>
  
  <pre><code>[DEBUG/MainProcess] created semlock with handle 140480257941504
  [DEBUG/MainProcess] created semlock with handle 140480257937408
  [DEBUG/MainProcess] created semlock with handle 140480257933312
  [DEBUG/MainProcess] Queue._after_fork()
  [DEBUG/Process-1] Queue._after_fork()
  [INFO/Process-1] child process calling self.run()
  [DEBUG/Process-1] Queue._start_thread()
  [DEBUG/Process-1] doing self._thread.start()
  [DEBUG/Process-1] starting thread to feed data to pipe
  [DEBUG/Process-1] ... done self._thread.start()
  [DEBUG/Process-1] telling queue thread to quit
  [INFO/Process-1] process shutting down
  [DEBUG/Process-1] running all "atexit" finalizers with priority &gt;= 0
  [DEBUG/Process-1] running the remaining "atexit" finalizers
  [DEBUG/Process-1] joining queue thread
  Traceback (most recent call last):
    File "/usr/lib/python3.7/multiprocessing/queues.py", line 242, in _feed
      send_bytes(obj)
    File "/usr/lib/python3.7/multiprocessing/connection.py", line 200, in send_bytes
      self._send_bytes(m[offset:offset + size])
    File "/usr/lib/python3.7/multiprocessing/connection.py", line 404, in _send_bytes
      self._send(header + buf)
    File "/usr/lib/python3.7/multiprocessing/connection.py", line 368, in _send
      n = write(self._handle, buf)
  BrokenPipeError: [Errno 32] Broken pipe
  [DEBUG/Process-1] feeder thread got sentinel -- exiting
  [DEBUG/Process-1] ... queue thread joined
  [INFO/Process-1] process exiting with exitcode 0
  [INFO/MainProcess] process shutting down
  [DEBUG/MainProcess] running all "atexit" finalizers with priority &gt;= 0
  [DEBUG/MainProcess] running the remaining "atexit" finalizers
  </code></pre>
  
  <p>"Success" (really silent failure, only able to replicate with Python 3.6):</p>
  
  <pre><code>[DEBUG/MainProcess] created semlock with handle 139710276231168
  [DEBUG/MainProcess] created semlock with handle 139710276227072
  [DEBUG/MainProcess] created semlock with handle 139710276222976
  [DEBUG/MainProcess] Queue._after_fork()
  [DEBUG/Process-1] Queue._after_fork()
  [INFO/Process-1] child process calling self.run()
  [DEBUG/Process-1] Queue._start_thread()
  [DEBUG/Process-1] doing self._thread.start()
  [DEBUG/Process-1] starting thread to feed data to pipe
  [DEBUG/Process-1] ... done self._thread.start()
  [DEBUG/Process-1] telling queue thread to quit
  [INFO/Process-1] process shutting down
  [INFO/Process-1] error in queue thread: [Errno 32] Broken pipe
  [DEBUG/Process-1] running all "atexit" finalizers with priority &gt;= 0
  [DEBUG/Process-1] running the remaining "atexit" finalizers
  [DEBUG/Process-1] joining queue thread
  [DEBUG/Process-1] ... queue thread joined
  [INFO/Process-1] process exiting with exitcode 0
  [INFO/MainProcess] process shutting down
  [DEBUG/MainProcess] running all "atexit" finalizers with priority &gt;= 0
  [DEBUG/MainProcess] running the remaining "atexit" finalizers
  </code></pre>
  
  <p>True success (using either <code>time.sleep(0.01)</code>):</p>
  
  <pre><code>[DEBUG/MainProcess] created semlock with handle 140283921616896
  [DEBUG/MainProcess] created semlock with handle 140283921612800
  [DEBUG/MainProcess] created semlock with handle 140283921608704
  [DEBUG/MainProcess] Queue._after_fork()
  [DEBUG/Process-1] Queue._after_fork()
  [INFO/Process-1] child process calling self.run()
  [DEBUG/Process-1] Queue._start_thread()
  [DEBUG/Process-1] doing self._thread.start()
  [DEBUG/Process-1] starting thread to feed data to pipe
  [DEBUG/Process-1] ... done self._thread.start()
  [DEBUG/Process-1] telling queue thread to quit
  [INFO/Process-1] process shutting down
  [DEBUG/Process-1] feeder thread got sentinel -- exiting
  [DEBUG/Process-1] running all "atexit" finalizers with priority &gt;= 0
  [DEBUG/Process-1] running the remaining "atexit" finalizers
  [DEBUG/Process-1] joining queue thread
  [DEBUG/Process-1] ... queue thread joined
  [INFO/Process-1] process exiting with exitcode 0
  [INFO/MainProcess] process shutting down
  [DEBUG/MainProcess] running all "atexit" finalizers with priority &gt;= 0
  [DEBUG/MainProcess] running the remaining "atexit" finalizers
  </code></pre>
  
  <p>The difference seems to be that in the truly successful case, the feeder receives the sentinel object before the <code>atexit</code> handlers.</p>