title: Multiprocessing Pool Hangs After Processing (on close or join)
body: <p>My main method (which is properly protected from being run when importing) looks like this: </p>
  
  <pre><code>def main():
      json_file, csv_out = get_user_input()
      sessions = get_sessions(json_file)
  
      if sessions:
          pool = multiprocessing.Pool()
          pool.map_async(process_one_item, sessions, 1)
          pool.close()
          pool.join()
  
      write_csv(csv_out, json_file, sessions)
  </code></pre>
  
  <p>It processes every single session (of over 1,000), which is just a list of timestamps that get processed before being written to disk, but when it finishes the final one, it just hangs. It never gets to "write_csv," which reads what has been written to disk.</p>
  
  <p>What am I missing?</p>
  
  <p>Update 1:</p>
  
  <pre><code>def main():
      json_file, csv_out = get_user_input()
      sessions = get_sessions(json_file)
  
      pool = multiprocessing.Pool()
  
      for results in pool.imap_unordered(func=process_one_item, 
                                         iterable=sessions, chunksize=1):
          store_results(results)
  
      pool.terminate()
      pool.join()
  
      write_csv(csv_out, json_file, sessions)
  </code></pre>
  
  <p>This also hung.</p>
  
  <p>The program can run if given less than 200 items AND finish. But when I go over 1,000, it hangs...</p>
  
  <p>Update 2:</p>
  
  <p>I added some print statements so I could better tell where it hangs.</p>
  
  <pre><code>def main():
      json_file, csv_out = get_user_input()
      sessions = get_sessions(json_file)
  
      pool = multiprocessing.Pool()
  
      for results in pool.imap_unordered(func=process_one_item, 
                                         iterable=sessions):
          store_results(results)
          print('...now back to main')
  
      print('Terminate NOW.')
      pool.terminate()
      print('Join NOW.')
      pool.join()
  
  write_csv(csv_out, json_file, sessions)
  </code></pre>
  
  <p>"...now back to main" always prints.</p>
  
  <p>"Terminate NOW." never prints.</p>
  
  <p>Update 3:</p>
  
  <p>Here's the stacktrace from when it's hanging and I kill the run.</p>
  
  <pre><code>Process ForkPoolWorker-1:
  Process ForkPoolWorker-13:
  Process ForkPoolWorker-23:
  Process ForkPoolWorker-29:
  Process ForkPoolWorker-27:
  Process ForkPoolWorker-30:
  Process ForkPoolWorker-31:
  Process ForkPoolWorker-26:
  Traceback (most recent call last):
    File , line 759, in next
      item = self._items.popleft()
  IndexError: pop from an empty deque
  During handling of the above exception, another exception occurred:
  Traceback (most recent call last):
    File "main.py", line 227, in &lt;module&gt;
      main()
    File "main.py", line 214, in main
      for results in pool.imap_unordered(func=process_one_item, iterable=sessions):
    File , line 763, in next
  Traceback (most recent call last):
  Traceback (most recent call last):
  Traceback (most recent call last):
    File , line 297, in _bootstrap
      self.run()
    File , line 297, in _bootstrap
      self.run()
    File , line 297, in _bootstrap
      self.run()
    File , line 99, in run
      self._target(*self._args, **self._kwargs)
    File , line 99, in run
      self._target(*self._args, **self._kwargs)
    File , line 110, in worker
      task = get()
    File , line 99, in run
      self._target(*self._args, **self._kwargs)
    File , line 351, in get
      with self._rlock:
    File , line 110, in worker
      task = get()
    File , line 110, in worker
      task = get()
    File , line 351, in get
      with self._rlock:
    File ", line 95, in __enter__
      return self._semlock.__enter__()
    File , line 95, in __enter__
      return self._semlock.__enter__()
  KeyboardInterrupt
  KeyboardInterrupt
  Traceback (most recent call last):
    File , line 297, in _bootstrap
      self.run()
    File , line 99, in run
      self._target(*self._args, **self._kwargs)
    File , line 110, in worker
      task = get()
    File , line 351, in get
      with self._rlock:
    File , line 95, in __enter__
      return self._semlock.__enter__()
  KeyboardInterrupt
    File , line 351, in get
      with self._rlock:
    File , line 95, in __enter__
      return self._semlock.__enter__()
  KeyboardInterrupt
  Traceback (most recent call last):
    File , line 297, in _bootstrap
      self.run()
    File , line 99, in run
      self._target(*self._args, **self._kwargs)
    File , line 110, in worker
      task = get()
    File , line 351, in get
      with self._rlock:
    File , line 95, in __enter__
      return self._semlock.__enter__()
  KeyboardInterrupt
  Traceback (most recent call last):
    File , line 297, in _bootstrap
      self.run()
    File , line 99, in run
      self._target(*self._args, **self._kwargs)
    File , line 110, in worker
      task = get()
    File , line 352, in get
      res = self._reader.recv_bytes()
    File , line 216, in recv_bytes
      buf = self._recv_bytes(maxlength)
    File , line 407, in _recv_bytes
      buf = self._recv(4)
    File , line 379, in _recv
      chunk = read(handle, remaining)
  KeyboardInterrupt
  Traceback (most recent call last):
    File , line 297, in _bootstrap
      self.run()
    File , line 99, in run
      self._target(*self._args, **self._kwargs)
    File , line 110, in worker
      task = get()
    File , line 351, in get
      with self._rlock:
    File , line 95, in __enter__
      return self._semlock.__enter__()
  KeyboardInterrupt
      self._cond.wait(timeout)
    File , line 296, in wait
  Traceback (most recent call last):
    File , line 297, in _bootstrap
      self.run()
    File , line 99, in run
      self._target(*self._args, **self._kwargs)
    File , line 110, in worker
      task = get()
    File , line 351, in get
      with self._rlock:
    File , line 95, in __enter__
      return self._semlock.__enter__()
  KeyboardInterrupt
      waiter.acquire()
  KeyboardInterrupt
  Process finished with exit code 1
  </code></pre>
  
  <p>I've removed the filenames to protect the innocent (user name in the path).</p>
  
  <p>Update 4:</p>
  
  <p>I added faulthandling and set a timeout.</p>
  
  <pre><code>if __name__ == '__main__':
      faulthandler.enable()
      faulthandler.dump_traceback_later(timeout=60, repeat=True, exit=True)
      start_time = time.perf_counter()
      main()
      print('\nDone!\n\nThis code ran in {:.1f} seconds'.format(time.perf_counter()-start_time))
  </code></pre>
  
  <p>When I did this I got a timeout and a stack trace of: </p>
  
  <pre><code>Thread 0x00001094 (most recent call first):  File "C:\ProgramData\Anaconda3\lib\multiprocessing\connection.py", line 306 in _recv_bytes  File "C:\ProgramData\Anaconda3\lib\multiprocessing\connection.py", line 250 in recv  File "C:\ProgramData\Anaconda3\lib\multiprocessing\pool.py", line 489 in _handle_results  File "C:\ProgramData\Anaconda3\lib\threading.py", line 864 in run  File "C:\ProgramData\Anaconda3\lib\threading.py", line 916 in _bootstrap_inner  File "C:\ProgramData\Anaconda3\lib\threading.py", line 884 in _bootstrapThread 0x0000476c (most recent call first):  File "C:\ProgramData\Anaconda3\lib\multiprocessing\connection.py", line 284 in _send_bytes  File "C:\ProgramData\Anaconda3\lib\multiprocessing\connection.py", line 206 in send  File "C:\ProgramData\Anaconda3\lib\multiprocessing\pool.py", line 450 in _handle_tasks  File "C:\ProgramData\Anaconda3\lib\threading.py", line 864 in run  File "C:\ProgramData\Anaconda3\lib\threading.py", line 916 in _bootstrap_inner  File "C:\ProgramData\Anaconda3\lib\threading.py", line 884 in _bootstrapThread 0x00002d80 (most recent call first):  File "C:\ProgramData\Anaconda3\lib\multiprocessing\pool.py", line 432 in _handle_workers  File "C:\ProgramData\Anaconda3\lib\threading.py", line 864 in run  File "C:\ProgramData\Anaconda3\lib\threading.py", line 916 in _bootstrap_inner  File "C:\ProgramData\Anaconda3\lib\threading.py", line 884 in _bootstrapThread 0x000038a8 (most recent call first):  File "C:\ProgramData\Anaconda3\lib\threading.py", line 295 in wait  File "C:\ProgramData\Anaconda3\lib\multiprocessing\pool.py", line 750 in next  File "C:/Users/mtanner/IdeaProjects/cti_trending_analysis/main.py", line 215 in main  File "C:/Users/mtanner/IdeaProjects/cti_trending_analysis/main.py", line 231 in &lt;module&gt;Process SpawnPoolWorker-36:Traceback (most recent call last):  File "C:\ProgramData\Anaconda3\lib\multiprocessing\pool.py", line 125, in worker    put((job, i, result))
    File "C:\ProgramData\Anaconda3\lib\multiprocessing\queues.py", line 344, in put    self._writer.send_bytes(obj)
    File "C:\ProgramData\Anaconda3\lib\multiprocessing\connection.py", line 200, in send_bytes    self._send_bytes(m[offset:offset + size])
    File "C:\ProgramData\Anaconda3\lib\multiprocessing\connection.py", line 280, in _send_bytes    ov, err = _winapi.WriteFile(self._handle, buf, overlapped=True)
  BrokenPipeError: [WinError 232] The pipe is being closedDuring handling of the above exception, another exception occurred:Traceback (most recent call last):  File "C:\ProgramData\Anaconda3\lib\multiprocessing\process.py", line 258, in _bootstrap    self.run()
    File "C:\ProgramData\Anaconda3\lib\multiprocessing\process.py", line 93, in run    self._target(*self._args, **self._kwargs)
    File "C:\ProgramData\Anaconda3\lib\multiprocessing\pool.py", line 130, in worker    put((job, i, (False, wrapped)))
    File "C:\ProgramData\Anaconda3\lib\multiprocessing\queues.py", line 344, in put    self._writer.send_bytes(obj)
    File "C:\ProgramData\Anaconda3\lib\multiprocessing\connection.py", line 200, in send_bytes    self._send_bytes(m[offset:offset + size])
    File "C:\ProgramData\Anaconda3\lib\multiprocessing\connection.py", line 280, in _send_bytes    ov, err = _winapi.WriteFile(self._handle, buf, overlapped=True)
  BrokenPipeError: [WinError 232] The pipe is being closedProcess SpawnPoolWorker-28:Traceback (most recent call last):  File "C:\ProgramData\Anaconda3\lib\multiprocessing\pool.py", line 125, in worker    put((job, i, result))
    File "C:\ProgramData\Anaconda3\lib\multiprocessing\queues.py", line 344, in put    self._writer.send_bytes(obj)
    File "C:\ProgramData\Anaconda3\lib\multiprocessing\connection.py", line 200, in send_bytes    self._send_bytes(m[offset:offset + size])
    File "C:\ProgramData\Anaconda3\lib\multiprocessing\connection.py", line 280, in _send_bytes    ov, err = _winapi.WriteFile(self._handle, buf, overlapped=True)
  BrokenPipeError: [WinError 232] The pipe is being closedDuring handling of the above exception, another exception occurred:Traceback (most recent call last):  File "C:\ProgramData\Anaconda3\lib\multiprocessing\process.py", line 258, in _bootstrap    self.run()
    File "C:\ProgramData\Anaconda3\lib\multiprocessing\process.py", line 93, in run    self._target(*self._args, **self._kwargs)
    File "C:\ProgramData\Anaconda3\lib\multiprocessing\pool.py", line 130, in worker    put((job, i, (False, wrapped)))
    File "C:\ProgramData\Anaconda3\lib\multiprocessing\queues.py", line 344, in put    self._writer.send_bytes(obj)
    File "C:\ProgramData\Anaconda3\lib\multiprocessing\connection.py", line 200, in send_bytes    self._send_bytes(m[offset:offset + size])
    File "C:\ProgramData\Anaconda3\lib\multiprocessing\connection.py", line 280, in _send_bytes    ov, err = _winapi.WriteFile(self._handle, buf, overlapped=True)
  BrokenPipeError: [WinError 232] The pipe is being closedProcess SpawnPoolWorker-26:Traceback (most recent call last):  File "C:\ProgramData\Anaconda3\lib\multiprocessing\pool.py", line 125, in worker    put((job, i, result))
    File "C:\ProgramData\Anaconda3\lib\multiprocessing\queues.py", line 344, in put    self._writer.send_bytes(obj)
    File "C:\ProgramData\Anaconda3\lib\multiprocessing\connection.py", line 200, in send_bytes    self._send_bytes(m[offset:offset + size])
    File "C:\ProgramData\Anaconda3\lib\multiprocessing\connection.py", line 280, in _send_bytes    ov, err = _winapi.WriteFile(self._handle, buf, overlapped=True)
  BrokenPipeError: [WinError 232] The pipe is being closedDuring handling of the above exception, another exception occurred:Traceback (most recent call last):  File "C:\ProgramData\Anaconda3\lib\multiprocessing\process.py", line 258, in _bootstrap    self.run()
    File "C:\ProgramData\Anaconda3\lib\multiprocessing\process.py", line 93, in run    self._target(*self._args, **self._kwargs)
    File "C:\ProgramData\Anaconda3\lib\multiprocessing\pool.py", line 130, in worker    put((job, i, (False, wrapped)))
    File "C:\ProgramData\Anaconda3\lib\multiprocessing\queues.py", line 344, in put    self._writer.send_bytes(obj)
    File "C:\ProgramData\Anaconda3\lib\multiprocessing\connection.py", line 200, in send_bytes    self._send_bytes(m[offset:offset + size])
    File "C:\ProgramData\Anaconda3\lib\multiprocessing\connection.py", line 280, in _send_bytes    ov, err = _winapi.WriteFile(self._handle, buf, overlapped=True)
  BrokenPipeError: [WinError 232] The pipe is being closed
  </code></pre>
  
  <p>Does this make sense to anyone?</p>
  
  <p>Update 5:</p>
  
  <p>In my function <code>process_one_item</code> function, I used the <code>sys.exit(-1)</code> call, attempting to end the run when something code-breaking happened. (this program was originally not multiprocessed). I think the multiprocessing was hiding this effect. I removed this call and the program now runs to completion regardless of session count. </p>
  
  <p>If someone can explain all this, I'm happy to count that as the solution!</p>