multi processing

This is about Python multiprocessing. The below scenario has an in_queue where items are put into. There are multipe worker processes that get items from the in_queue. When processed, a item is put into the out_queue. All workers share the same in_queue and out_queue. Queue is process safe.

There is another writer process, which reads from the out_queue, as soon as it gets an item, it simply prints it out.

When a process is trying to get an item from a Queue, if no item is available (here it sets timeout to 1 second to demonstrate), it simply waits a bit, and try again.

import queue, time

import multiprocessing

#note, running this from IDE like spyder wont show the prints, something weird with stdout.

#however, run a command line 'python script.py' will show all the prints              

def worker(in_queue, out_queue):

    print('**worker {0} started'.format(os.getpid()))

    while True:

        try:

            item = in_queue.get(timeout=1)

            if item == 'DONE':

                print('**worker {0} quited'.format(os.getpid()))

                break

            else:

                out_queue.put('processed item: {1} '.format(os.getpid(), item))

                print('**worker {0} processed {1} '.format(os.getpid(), item))

        except queue.Empty:

            print('**worker {0} waiting'.format(os.getpid()))

            time.sleep(1)            

            continue 

def writer(out_queue):

    print('--writer {0} started'.format(os.getpid()))

    while True:

        try:

            item = out_queue.get(timeout=1)

            if item == 'DONE':

                print('--writer {0} quited'.format(os.getpid()))

                break

            else:

               print('--writer {0} wrote {1} '.format(os.getpid(), item)) 

        except queue.Empty:

            print('--writer {0} waiting'.format(os.getpid()))

            time.sleep(1)            

            continue     

if __name__ == '__main__':

    in_queue = multiprocessing.Queue()

    out_queue = multiprocessing.Queue()

    workers = 2

    #create writer

    writer_process = multiprocessing.Process(target=writer, args=(out_queue,))

    writer_process.daemon = True

    writer_process.start()

    print('started')

    #create workers

    worker_processes = []

    for i in range(workers):

        p = multiprocessing.Process(target=worker, args= (in_queue, out_queue))

        p.daemon = True

        p.start()

        worker_processes.append(p)

    #send items to in_queue for worker to process

    for i in range(20):

        in_queue.put('something-{0}'.format(i))

        time.sleep(1)

    #send signals for workers to quit

    for i in range(workers):

        in_queue.put('DONE')

    #wait until the workers finalize and quit

    for p in worker_processes:

        p.join()

    #signal the writer process to quit    

    out_queue.put('DONE')

    #wait for the writer process to quit

    writer_process.join   

    #wait a bit for the writer and workers to print messages

    #otherwise, it finishes too quickly and the messages dont have a chance to print

    time.sleep(5)

    print('exited')   

The result is something like this: