Eu estou tendo alguns problemas surgem ao usar processos e filas.
Quando executo o código a seguir, a função de destino simplesmente obtém um item de uma fila mestre e o adiciona a outra fila específica desse processo.
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()
Quando a função de destino imprimir a fila em uso, você notará que uma fila é usada quase exclusivamente.
Se você pegar a saída e imprimi-la, verá que a maioria dos números acaba em uma única fila.
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)
O quê está causando isto? Existe alguma coisa no meu código que eu deveria estar fazendo para nivelar o trabalho que esses processos estão fazendo?
SAÍDA
[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]
Respostas:
2 para resposta № 1Eu acho que você tem dois problemas aqui:
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
Dos documentos para obter:
Remova e retorne um item da fila. Se o bloco args opcional for True (o padrão) e o tempo limite for Nenhum (o padrão), bloqueie, se necessário, até que um item esteja disponível. Se o tempo limite for um número positivo, ele bloqueará no máximo dois segundos do tempo limite e gerará a exceção Queue.Empty se nenhum item estiver disponível dentro desse tempo. Caso contrário (o bloco é False), retorne um item se um estiver imediatamente disponível, caso contrário, a exceção Queue.Empty será levantada (o tempo limite será ignorado nesse caso).
Desde que você está passando 0 como o primeiro parâmetro, é equivalente a get(False). Isso faz com que seja sem bloqueio, o que significa que senão pode obter um valor imediatamente, ele irá disparar uma exceção vazia, o que encerrará seu processo de trabalho. Como todas as suas funções de "trabalho" são idênticas e tentam extrair da fila principal ao mesmo tempo, algumas podem não conseguir obtenha um valor imediatamente e morrerá.
Dando o .get() um pequeno tempo limite deve corrigir esse problema.
O segundo problema é que a função "trabalho" leva basicamente zero tempo para ser concluída. Dê uma pequena pausa com sleep(.2) para simular algum trabalho não-trival e ele distribuirá entre os trabalhadores:
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
EDITAR:
Eu esqueci de dizer, geralmente é melhor para este tipo de problema não depender do tempo limite de .get() para sinalizar o final da fila. Você obtém mais controle se usar algum tipo de objeto marcador "fim da fila" que você passa para a fila que informa aos funcionários que é hora de sair. Desta forma, você pode ter todos eles bloqueados, esperando por uma nova entrada ou um "comando" de saída.