Parallelize Image Processing Using Numpy

Question

I'm trying to speed up a section of my code using parallel processing in python, but I'm having trouble getting it to work right, or even find examples that are relevant to me.

The code produces a low-polygon version of an image using Delaunay triangulation, and the part that's slowing me down is finding the mean values of each triangle.

I've been able to get a good speed increase by vectorizing my code, but hope to get more using parallelization:

The code I'm having trouble with is an extremely simple for loop:

for tri in tris:
        lopo[tridex==tri,:] = np.mean(hipo[tridex==tri,:],axis=0)

The variables referenced are as follows.

tris - a unique python list of all the indices of the triangles

lopo - a Numpy array of the final low-polygon version of the image

hipo - a Numpy array of the original image

tridex - a Numpy array the same size as the image. Each element represents a pixel and stores the triangle that the pixel lies within

I can't seem to find a good example that uses multiple numpy arrays as input, with one of them shared.

I've tried multiprocessing (with the above snippet wrapped in a function called colorImage):

p = Process(target=colorImage, args=(hipo,lopo,tridex,ppTris))
p.start()
p.join()

But I get a a broken pipe error immediately.

Answer 1

So the way that Python's multiprocessing works (for the most part) is that you have to designate the individual threads that you want to run. I made a brief introductory tutorial here: http://will-farmer.com/parallel-python.html

In your case, what I would recommend is split tris into a bunch of different parts, each equally sized, each that represents a "worker". You can split this list with numpy.split() (documentation here: http://docs.scipy.org/doc/numpy/reference/generated/numpy.split.html).

Then for each list in tri, we use the Threading and Queue modules to designate 8 workers.

import numpy as np
# split into 8 different lists
tri_lists = np.split(tris, 8)
# Queues are threadsafe
return_values = queue.Queue()
threads = []
def color_image(q, tris, hipo, tridex):
    """ This is the function we're parallelizing """
    for tri in tris:
        return_values.put(np.mean(hipo[tridex==tri,:], axis=0))
# Now we run the jobs
for i in range(8):
    threads.append(threading.Thread(
        target=color_image,
        args=(return_values, tri_lists[i], hipo, tridex)))
# Now we have to cleanup our results
# First get items from queue
results = [item for item in return_values.queue]
# Now set values in lopo
for i in range(len(results)):
    for t in tri_lists[i]:
        lopo[tridex==t, :] = results[i]

This isn't the cleanest way to do it, and I'm not sure if it works since I can't test it, but this is a decent way to do it. The parallelized part is now np.mean(), while setting the values is not parallelized.

If you want to also parallelize the setting of the values, you'll have to have a shared variable, either using the Queue, or with a global variable.

See this post for a shared global variable: Python Global Variable with thread