Why does the queue size remain zero?

Question

I am using multiprocessing to process my records.

queue = Queue()

def produce(i, item):
    data = process(i, item)
    queue.put(data)

def process(i, item):
    data = do_processing(i, item)
    return data

if __name__ == '__main__':
    records = load_records()

    with ProcessPoolExecutor(max_workers=os.cpu_count()) as executor:
        print('produce items')
        for i, item in ennumerate(records.items()):
            executor.submit(produce, i, item)

    print('queue size:{}'.format(queue.qsize()))
    while not queue.empty():
        save(queue.get())

Here I put records in a queue from produce as that step is time-consuming. After processing the records, I save them. As the consume step is not time consuming, I don't bother to run it in a separate thread.

Here after I execute the code, the queue remains empty. What is going on here?

Answer 1

I think this is how to do what you want. As mentioned in a comment, each process runs in its own memory space so global variables like the queue can't be simply shared and you can't pass it as an argument to each process.

When using a ProcessPoolExecutor, you can effectively do what's needed—sharing the queue—by defining an initializer function that will be called at the beginning of each process that will create a global for that process and pass the queue as an argument to it.

Here's something that closely resembles your code and is actually runnable illustrating what I mean:

from concurrent.futures import ProcessPoolExecutor
from multiprocessing import Queue
import os


MAX_RECORDS = 10

def load_records():
    return dict.fromkeys(range(MAX_RECORDS), 0)

def do_processing(item):
    return item

def init_queue(queue):
    globals()['queue'] = queue  # Makes queue a global in each process.

def produce(i, item):
    data = process(i, item)
    queue.put(data)

def process(i, item):
    data = do_processing(item)
    return data


if __name__ == '__main__':
    records = load_records()

    queue = Queue()
    with ProcessPoolExecutor(max_workers=os.cpu_count(),
                             initializer=init_queue, initargs=(queue,)) as executor:
        print('producing items')
        for i, item in enumerate(records.items()):
            future = executor.submit(produce, i, item)
        print('done producing items')

    print('queue size: {}'.format(queue.qsize()))
    while not queue.empty():
        print(queue.get())

Output:

producing items
done producing items
queue size: 10
(0, 0)
(1, 0)
(2, 0)
(3, 0)
(4, 0)
(5, 0)
(6, 0)
(7, 0)
(8, 0)
(9, 0)

Answer 2

Use multiprocessing.Queue() for handling multiprocessing.

queue = multiprocessing.Queue()

def produce(item):
    data = process(item)
    queue.put(data)

def process(item):
    data = do_processing(item)
    return data

if __name__ == '__main__':
    records = load_records()

    with ProcessPoolExecutor(max_workers=os.cpu_count()) as executor:
        print('produce items')
        for item in records.items():
            executor.submit(produce, item)

    print('queue size:{}'.format(queue.qsize()))
    while not queue.empty():
        save(queue.get())

Answer 3

def process(item):
    data = do_processing(item)
    return data

queue = Queue() # not a multiprocessing queue

with multiprocessing.Pool(processes=os.cpu_count()) as pool:
    for result in pool.imap(process, records):
        queue.put(result)

while not queue.empty():
    save(queue.get())