Dado um processo multithread de leitura / gravação bastante padrão, com uma fila de leitura e uma fila de gravação:

8 vezesworker done é impresso, mas a instrução join () nunca é passada. Mas se eu substituirqueue_out.put(r) por `queue_out.put (1) ele funciona.

Isso está derretendo meu cérebro, provavelmente algo realmente estúpido. Devo fazer uma cópia do meu dicionário e colocá-la na fila de retorno? Cometi um erro estúpido em algum lugar?

Função de processo

def reader(queue_in, queue_out, funktion):
    # Read from the queue
    while True:
        r = queue_in.get()
        if r == 'DONE':
            return
        funktion(r) # funktion adds additional keys to the dictionary
        queue_out.put(r) # <---- replacing r by 1 does let me join()
    print "worker done" # <----- this happens

Preencher a fila de entrada

def writer(generator, queue):
    # Write to the queue
    for r in enumerate(generator):
        # r is a complex dictionary
        queue.put(r)    
    print "writer done"
    for _ in range(0, WORKERS):
        queue.put((-1, "DONE"))

O resto

WORKERS = 8

# init Queues
queue_in = Queue()
queue_out = Queue()

# Start processes, with input and output quests
readers = []
for _ in range(0, WORKERS):
    p = Process(target=reader, args=(queue_in, queue_out, funktion))
    p.daemon = True
    p.start()
    readers.append(p)

writer(generator, queue_in)

for p in readers:
    p.join()

print "joined"  # <---- this never happens

queue_in.close()

while not queue_out.empty():
    print queue_out.get()
queue_out.close()