Skip to content

Instantly share code, notes, and snippets.

@roolebo
Created May 29, 2016 10:36
Show Gist options
  • Save roolebo/3bad473c89ff6d133ea37228bc79c5db to your computer and use it in GitHub Desktop.
Save roolebo/3bad473c89ff6d133ea37228bc79c5db to your computer and use it in GitHub Desktop.
Python 3.4-3.5 pipe leak in multiprocessing.Process
import unittest
from multiprocessing import Process
import time
import os
import fcntl
def count_leaks(process_count):
def p_start(i):
print(i)
procs = []
for i in range(process_count):
p = Process(target=p_start, args=(str(i),))
p.start()
procs.append(p)
for p in procs:
p.join()
leaks = 0
for p in procs:
try:
fcntl.fcntl(p.sentinel, fcntl.F_GETFD)
except:
pass
else:
os.close(p.sentinel)
leaks += 1
return leaks
class MpPipeLeakTest(unittest.TestCase):
def test(self):
self.assertEqual(count_leaks(10), 0)
if __name__ == '__main__':
unittest.main()
___________________________________ summary ____________________________________
py26: commands succeeded
py27: commands succeeded
ERROR: py34: commands failed
ERROR: py35: commands failed
[tox]
envlist = py26, py27, py34, py35
skipsdist=True
[testenv]
commands = python leak.py {posargs}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment