Multi-Processing

The multiprocessing library defines various classes, most of which operate in the same way as similar classes in the threading and related modules. Whereas in using threading you also imported resources from other modules, the multiprocessing module tries to put all necessary resources into one convenient place, simplifying imports. But you will easily recognize the program style from your recent work on multi-threading.

Our first multiprocessing example is marked up below as though we were editing the first thread.py example. This shows how similar the two environments are. Create process.py as shown:

"""
process.py: demonstrate creation and parallel execution of processes.
"""

import multiprocessing
import time
import sys

def run(i, name):
    """Sleep for a given number of seconds, report and terminate."""
    time.sleep(i)
    print(name, "finished after", i, "seconds")
    sys.stdout.flush()

if __name__ == "__main__":
    for i in range(6):
        t = multiprocessing.Process(target=run, args=(i, "P"+str(i)))
        t.start()
    print("Processes started")

Note that this program has been correctly written as a module, so that the action of starting six processes is only performed by the process that runs this code, and not in any processes that may try to import the module. This is very important, because the subprocesses have to get their description of the work to be done from somewhere, and they do that by importing the main module. So in this case the subprocesses will import the process module (so the test __name__ == "__main__" is false) to access the run() function.

The output should not be at all surprising:

Processes started
P0 finished after 0 seconds
P1 finished after 1 seconds
P2 finished after 2 seconds
P3 finished after 3 seconds
P4 finished after 4 seconds
P5 finished after 5 seconds

The following listing shows the code for the multiprocessor version alongside the equivalent threading-based code. The differences are small enough to be negligible, and to allow anyone who understood the threaded code to also understand the multi-process version.

"""
output.py: The output process for the miniature framework.
"""
identity = lambda x: x

import multiprocessing
import sys

class OutThread(multiprocessing.Process):

    def __init__(self, N, q, sorting=True, *args, **kw):
        """Initialize process and save queue reference."""
        multiprocessing.Process.__init__(self, *args, **kw)
        self.queue = q
        self.workers = N
        self.sorting = sorting
        self.output = []

    def run(self):
        """Extract items and print until all done."""
        while self.workers:
            p = self.queue.get()
            if p is None:
                self.workers -= 1
            else:
                # This is a real output packet
                self.output.append(p)
        print("".join(c for (i, c) in (sorted if self.sorting else identity)(self.output)))
        print ("Output process terminating")
        sys.stdout.flush()

The main difference between the two pieces of code is the use of multiprocessing.process in place of threading.Thread, and associated changes to a couple of comments. It is also necessary to flush the process's standard output stream to make sure that it is captured before the process terminates—otherwise you will see a confusing lack of output! (Feel free to try running the program with the flush() call commented out to verify this).

The next listing shows the differences in the worker code when processes are being used instead of threads.

"""
worker.py: a sample worker process that receives input
       through one queue and routes output through another.
"""

from multiprocessing import Process
import sys

class WorkerThread(Process):

    def __init__(self, iq, oq, *args, **kw):
        """Initialize process and save Queue references."""
        Process.__init__(self, *args, **kw)
        self.iq, self.oq = iq, oq

    def run(self):
        while True:
            work = self.iq.get()
            if work is None:
                self.oq.put(None)
                print("Worker", self.name, "done")
                self.iq.task_done()
                break
            i, c = work
            result = (i, self.process(c)) # this is the "work"
            self.oq.put(result)
            self.iq.task_done()
        sys.stdout.flush()

    def process(self, s):
        """This defines how the string is processed to produce a result."""
        return s.upper()

Again the only change is to use Process from multiprocessing instead of Thread from threading. (Two of the differences are again in comments.)

The control process again needs very little change: queue objects come from the multiprocessing module rather than the queue module, and in that module if you are going to join() a queue then you must use a JoinableQueue. The rest of the logic is exactly the same, with the exception that the code must now be guarded so that it isn't executed when the module is imported by the multiprocessing module. This means you have to indent the majority of the logic. This is easy in Eclipse: just highlight all the lines of code (making sure you are selecting whole lines) and then press Tab once.

"""
control.py: Creates queues, starts output and worker processes,
        and pushes inputs into the input queue.
"""
from multiprocessing import Queue, JoinableQueue
from output import OutThread
from worker import WorkerThread

if __name__ == '__main__':
    WORKERS = 10

    inq = JoinableQueue(maxsize=int(WORKERS*1.5))
    outq = Queue(maxsize=int(WORKERS*1.5))

    ot = OutThread(WORKERS, outq, sorting=True)
    ot.start()

    for i in range(WORKERS):
        w = WorkerThread(inq, outq)
        w.start()
    instring = input("Words of wisdom: ")
    # feed the process pool with work units
    for work in enumerate(instring):
        inq.put(work)
    # terminate the process pool
    for i in range(WORKERS):
        inq.put(None)
    inq.join()
    print("Control process terminating")