Питонні багатопроцесорні питання - python, process, queue, multiprocessing

Я маю кілька проблем, які виникають при використанні процесів і черг.

Коли я запускаю наступний код, цільова функція просто отримує елемент з головної черги і додає її до іншої черги, специфічної для цього процесу.

import sys
import multiprocessing
from Queue import Empty

# This is just taking a number from the queue
# and adding it to another queue
def my_callable(from_queue, to_queue):
while True:
try:
tmp = from_queue.get(0)
to_queue.put(tmp)
print to_queue
except Empty:
break

# Create a master queue and fill it with numbers
main_queue = multiprocessing.Queue()
for i in xrange(100):
main_queue.put(i)

all_queues = []
processes = []
# Create processes
for i in xrange(5):
# Each process gets a queue that it will put numbers into
queue = multiprocessing.Queue()
# Keep up with the queue we are creating so we can get it later
all_queues.append(queue)
# Pass in our master queue and the queue we are transferring data to
process = multiprocessing.Process(target=my_callable,
args=(main_queue, queue))
# Keep up with the processes
processes.append(process)

for thread in processes:
thread.start()

for thread in processes:
thread.join()

Коли цільова функція друкує чергу, що використовується, ви помітите, що одна черга використовується майже виключно.

Якщо ви візьмете вивід і роздрукуєте його, ви побачите, що більшість номерів закінчуються під одну чергу.

def queue_get_all(q):
items = []
maxItemsToRetreive = 100
for numOfItemsRetrieved in range(0, maxItemsToRetreive):
try:
if numOfItemsRetrieved == maxItemsToRetreive:
break
items.append(q.get_nowait())
except Empty, e:
break
return items

for tmp in all_queues:
print queue_get_all(tmp)

Що викликає це? Чи є в моєму коді щось, що я повинен робити, що вирівнюватиме роботу цих процесів?

ВИХІД

[0, 2, 3, 4, 5, 6, 7, 8]
[1, 9, 10]
[11, 14, 15, 16]
[12, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, 73, 74, 75, 76, 77, 78, 79, 80, 81, 82, 83, 84, 85, 86, 87, 88, 89, 90, 91, 92, 93, 94, 95, 96, 97, 98, 99]
[13]

Відповіді:

2 для відповіді № 1

Думаю, у вас є дві проблеми:

def my_callable(from_queue, to_queue):
while True:
try:
tmp = from_queue.get(0)
to_queue.put(tmp)
print to_queue
except Empty:
break

З документів для отримати:

Видалити та повернути елемент з черги. Якщо необов'язковий аргумент блоку є True (за замовчуванням) і тайм-аут - None (за замовчуванням), блокуються при необхідності, поки елемент доступний. Якщо тайм-аут є позитивним числом, він блокується на більшість секунд, що вичерпуються, і піднімає виняток Queue.Empty, якщо протягом цього часу не було доступно жодного елемента. В іншому випадку (блок False), повертає елемент, якщо він одразу доступний, інакше підніме виключення Queue.Empty (тайм-аут ігнорується в цьому випадку).

Так як ви проходите 0 як перший параметр, він еквівалентний get(False). Це робить його неблокуючим, що означає, якщо віннеможливо одразу отримати значення, піднявши порожній виняток, який завершить ваш робочий процес. Оскільки всі ваші "робочі" функції ідентичні і намагаються витягнути з головної черги одночасно, деякі не можуть отримати відразу ж значення і помре.

Надання .get() невеликий час очікування має вирішити цю проблему.

Друга проблема полягає в тому, що ваша функція "роботи" займає в основному нульовий час для завершення. Дайте йому невелику паузу sleep(.2) імітувати деяку нетривіальну роботу, і вона буде поширюватися по робочих:

def my_callable(from_queue, to_queue):
while True:
try:
tmp = from_queue.get(True, .1)
sleep(0.2)
to_queue.put(tmp)
except Empty:
break

EDIT:

Я забув сказати, що в цілому для цього типу проблем краще не покладатися на тайм-аут .get() для сигналізації про закінчення черги. Ви отримуєте більше контролю, якщо ви використовуєте певний тип об'єкта маркера "кінець черги", який ви переходите в чергу, яка повідомляє працівникам, що настав час вийти. Таким чином, ви можете мати всі блоки, чекаючи або нового введення, або команди "виходу".

---