Блог

У меня есть многопроцессорная программа, в которой один процесс помещает множество элементов в очередь, а затем устанавливает событие после того, как он закончил помещать элементы в очередь. Другой процесс ожидает этого события, а затем опустошает очередь в цикле while, как в приведенном ниже коде, проверяя с помощью вызова empty (). Проблема в том, что время от времени (возможно, 1 раз из 10) функция empty() возвращает значение True, когда в очереди все еще есть элементы. Во время цикла while ни один другой процесс или поток не касается очереди. Проблема может быть воспроизведена с помощью приведенного ниже кода в одном процессе. Большую часть времени выход составляет:

 The size of my_queue initially is 400
The size of my_queue is 0
 

Но время от времени я буду получать что-то вроде:

 The size of my_queue initially is 400
The size of my_queue is 377
 

Вот код:

 from multiprocessing import Queue as MPQueue

my_list = []
my_queue = MPQueue()
for i in range(0, 400):
    my_queue.put(f"The value of i is {i}")

print(f"The size of my_queue initially is {my_queue.qsize()}")
while not my_queue.empty():
    my_list.append(my_queue.get())
print(f"The size of my_queue is {my_queue.qsize()}")
 

Является ли это ошибкой в многопроцессорной обработке.Модуль очереди или я что-то упускаю? Я также попытался использовать get_nowait (), а затем поймать очередь.Пустое исключение, но я получаю тот же результат. Любая помощь была бы очень признательна. Спасибо.

В какой-то момент я сталкиваюсь с той же проблемой. docs.python.org

многопроцессорная обработка класса.Очередь([максимальный размер])

qsize():

Верните приблизительный размер очереди. Из-за семантики > многопоточности/многопроцессорности это число >ненадежно.

Обратите внимание, что это может вызвать ошибку NotImplementedError на платформах Unix, таких как Mac OS X>, где не реализовано значение sem_getvalue ().