Python multiprocessing.Process behaves non deterministic

Viewed 356

The following code shows a simple multiprocessing.Process pipeline with a shared dictionary of lists and a task queue for different consumers:

import multiprocessing

class Consumer(multiprocessing.Process):

    def __init__(self, task_queue, result_dict):
        multiprocessing.Process.__init__(self)
        self.task_queue = task_queue
        self.result_dict = result_dict

    def run(self):
        proc_name = self.name
        while True:
            next_task = self.task_queue.get()
            if next_task is None:
                # Poison pill means shutdown
                print('%s: Exiting' % proc_name)
                self.task_queue.task_done()
                break
            print('%s: %s' % (proc_name, next_task))

            # Do something with the next_task
            l = self.result_dict[5]
            l.append(3)
            self.result_dict[5] = l
            # alternative, but same problem
            #self.result_dict[5] += [3]

            self.task_queue.task_done()
        return

def provide_tasks(tasks, num_worker):
    low = [ 
        ['w1', 'w2'],
        ['w3'],
        ['w4', 'w5']
    ]
    for el in low:
        tasks.put(el)
    # Add a poison pill for each worker
    for i in range(num_worker):
        tasks.put(None)

if __name__ == '__main__':
    num_worker = 3
    tasks = multiprocessing.JoinableQueue()
    manager = multiprocessing.Manager()

    results = manager.dict()
    lists = [manager.list() for i in range(1, 11)]
    for i in range(1, 11):
        results[i] = lists[i - 1]

    worker = [Consumer(tasks, results) for i in range(num_worker)]
    for w in worker:
        w.start()

    p = multiprocessing.Process(target=provide_tasks, args=(tasks,num_worker))
    p.start()
    # Wait for all of the tasks to finish
    p.join()
    print(results)

When you run this example with Python3.x you will receive different outputs for the results dict. I actually expect the results dict to look like

{1: [], 2: [], 3: [], 4: [], 5: [3, 3, 3], 6: [], 7: [], 8: [], 9: [], 10: []}

But for some executions it looks like this:

{1: [], 2: [], 3: [], 4: [], 5: [3, 3], 6: [], 7: [], 8: [], 9: [], 10: []}

Can someone explain me this behavior? Why is somethings a number missing?

Updated solution approach according to the proposed answer:

if next_task is None:
    with lock:
        self.result_dict.update(self.local_dict)
        [...]

Where lock is a manager.Lock() and self.local_dict is a defaultdict(list).

Moved lock according to the answer comment. Also added a version that doesn't work with the lock.

# Works
with lock:
    l = self.result_dict[x]
    l.append(3)
    self.result_dict[x] = l
self.task_queue.task_done()

# Doesn't work. Even if I move the lock out of the loop. 
for x in range(1, 10):
    with lock:
        l = self.result_dict[x]
        l.append(3)
        self.result_dict[x] = l

In order to get the second example to work, we need to call join on all worker too.

1 Answers
Related