Многопроцессорность «OSError: [Errno 24] Слишком много открытых файлов»: Как очистить задания и очереди?

Question

Я делаю все возможное, чтобы закрыть и очистить очередь Queue, когда я закончил использовать их, чтобы собирать выходные данные процесса в многопроцессорном модуле python. Вот какой код, который умирает в какой-то момент из-за «слишком большого количества открытых файлов». Что еще я могу сделать, чтобы очистить полные задания/очереди, чтобы я мог делать столько, сколько мне нравится?

# The following [fails to] demonstrates how to clean up jobs and queues (the queues is key?) to avoid the OSError of too many files open. 
def dummy(inv,que): 
    que.put(inv) 
    return(0) 
from multiprocessing import Process, Queue, cpu_count 
nTest=2800 
queues=[None for ii in range(nTest)] 
for ii in range(nTest): 
    queues[ii]=Queue() 
    job=Process(target=dummy, args=[ii,queues[ii]]) 
    job.start() 
    print('Started job %d'%ii) 
    job.join() 
    print('Joined job %d'%ii) 
    job.terminate() 
    print('Terminated job %d'%ii) 
    queues[ii].close() 

Поскольку это OSError, в моем коде нет конкретной строки, которая вызывает проблему. Отчет выглядит следующим образом:

... 
Terminated job 1006 
Started job 1007 
Joined job 1007 
Terminated job 1007 
Started job 1008 
Joined job 1008 
Terminated job 1008 
Started job 1009 
Joined job 1009 
Terminated job 1009 
     --------------------------------------------------------------------------- 

OSError         Traceback (most recent call last) 
<ipython-input-2-5f057cd2fe88> in <module>() 
----> 1 breaktest() 

... in breaktest() 

/usr/lib64/python2.6/multiprocessing/__init__.pyc in Queue(maxsize) 

/usr/lib64/python2.6/multiprocessing/queues.pyc in __init__(self, maxsize) 

/usr/lib64/python2.6/multiprocessing/synchronize.pyc in __init__(self) 

/usr/lib64/python2.6/multiprocessing/synchronize.pyc in __init__(self, kind, value, maxvalue) 

OSError: [Errno 24] Too many open files 
> /usr/lib64/python2.6/multiprocessing/synchronize.py(49)__init__() 

Answer 1

Более общее решение для моей реальной проблемы (которое мотивировало игрушечный пример в моем сообщении) заключалось в том, чтобы избежать того, чтобы любые переменные цикла обращались непосредственно к очередям или к любому объекту, содержащему их. Это, в сочетании с установкой элемента списка None, когда я закончил с очередью, возможно, в сочетании с ручным вызовом gc.collect(), приводит к надлежащему уничтожению (сборке мусора) каждой очереди, когда я закончил с каждой из них.

@Finch_Powers решил проблему, поскольку я разместил ее в комментарии.

Я подозреваю, что опыт может отличаться в разных системах.

Фактический код, где это помогло функция runFunctionsInParallel в https://github.com/cpbl/cpblUtilities/blob/master/parallel.py

Answer 2

Ваш сценарий застревает после ~ 1000 задач, потому что это предел по умолчанию дескрипторов файлов для одного процесса.

Очереди реализованы с помощью Pipes, которые содержат дескрипторы файлов. Трубы должным образом удаляются через GC. Когда вы храните их в списке, они не собирают мусор и файловые дескрипторы протекают до тех пор, пока ваш процесс не выделит 1024 из них, а затем он выйдет из строя.

У вас есть необходимость в хранении очередей в списке?