Multiprocessing in Python - Concurrence

Question

I have a simple class ImageData, to create some Images using Shepard Interpolation. I have to generate more then 50 images, so I though I could parallelized this. I've tried this (I've never worked with multiprocessing before, so, sorry if I made something stupid), already applying what @Alfe taught me:

class ImageData(object):

    def __init__(self, width, height, range_min=-1, range_max=1):
        """
        The ImageData constructor
        """
        self.width = width
        self.height = height
        #The values range each pixel can assume
        self.range_min = range_min
        self.range_max = range_max
        self.data = []
        for i in range(width):
            self.data.append([0] * height)

    def shepard_interpolation(self, queue, seeds=10):
        """
        Perform a Shepard shepard_interpolation
        :param queue
        :param seeds
        """
        points = []
        f = []
        for s in range(seeds):
            # Generate a pixel position
            pos_x = random.randrange(self.width)
            pos_y = random.randrange(self.height)

            # Save the f(x,y) data
            x = Utils.translate_range(pos_x, 0, self.width, self.range_min, self.range_max)
            y = Utils.translate_range(pos_y, 0, self.height, self.range_min, self.range_max)
            z = Utils.function(x, y)
            points.append([x, y])
            f.append(z)

        for i in range(self.width):
            xt = (Utils.translate_range(i, 0, self.width, self.range_min, self.range_max))
            for j in range(self.height):
                yt = (Utils.translate_range(j, 0, self.height, self.range_min, self.range_max))
                self.data[i][j] = Utils.shepard_euclidian(points, f, [xt, yt], 3)
        queue.put(self)


if __name__ == '__main__':
    q = Queue()
    processes = [Process(target=ImageData.shepard_interpolation, args=(ImageData(50, 50), q,))    for _ in range(2)]
    for process in processes:
        process.start()
    for process in processes:
        process.join()

    print "Finish"

The problem is 'cause when I call passing range(2), everything works, but when I try with range(3), it never ends (my code never reaches the 'Finish' print) and I don't know why. And, in my case, I have to generate more then 50 images, and I don't know how I could achieve this. I have a processor Core 2 Duo.

EDIT:

I tried to comment the queue.put(self) and it works. But I have to receive the result, and this is the only way that I know so far to achieve this. I also don't get why with two processes it works. Any idea about what I could do to solve this? I guess the erros is in shepard_euclidian method. But I couldn't find it until now. This is this function:

def shepard_euclidian(x, z, p, u):
    n = len(x)
    d = [0.0] * n
    for i in range(n-1):
        pi = x[i]
        d[i] = math.pow(math.hypot(pi[0]-p[0], pi[1]-p[1]), u)
    w = [0.0] * n
    sw = 0.0
    for i in range(n-1):
        w[i] = 1.0
        for k in range(n-1):
            if i != k:
                w[i] *= d[k]
        sw += w[i]
    for i in range(len(w)-1):
        if sw != 0.0:
            w[i] /= sw
        else:
            w[i] = 0.0
    c = 0.0
    for i in range(n):
        c += (w[i] * z[i])
    return c

When I try this:

for i in range(self.width):
        xt = (Uts.Utils.translate_range(i, 0, self.width, self.range_min, self.range_max))
        for j in range(self.height):
            yt = (Uts.Utils.translate_range(j, 0, self.height, self.range_min, self.range_max))
            data = Uts.Utils.shepard_euclidian(points, f, [xt, yt], 3)
    queue.put(data)

It works. But this is not what I want.

And if I try this:

   aux = ImageData(50, 50)
   for i in range(self.width):
        xt = (Uts.Utils.translate_range(i, 0, self.width, self.range_min, self.range_max))
        for j in range(self.height):
            yt = (Uts.Utils.translate_range(j, 0, self.height, self.range_min, self.range_max))
            aux[x][y] = Uts.Utils.shepard_euclidian(points, f, [xt, yt], 3)
   self.data = aux
   queue.put(self.data)

It doesn't work. I really don't know what to do.

Any help would be appreciated. Thank you in advance.

Answer 1

You want to memorize all processes (ugly called k in your look) and join all of them in a second loop. Maybe this solves your issues at hand.

And please use a better name:

processes = [
  Process(target=ImageData.shepard_interpolation, args=(ImageData(50, 50), q,))
  for _ in range(3) ]
for process in processes:
    process.start()
for process in processes:
    process.join()

EDIT:

I've got a minimal version of your code which works without problems on my machine:

import os
from multiprocessing import *

class ImageData(object):
    def __init__(self, a, b):
        pass

    def shepard_interpolation(self, queue, seeds=10):
        self.pid = os.getpid()
        print self.pid, "queue put"
        queue.put(self)

if __name__ == '__main__':
    q = Queue()
    processes = [ Process(
      target=ImageData.shepard_interpolation, args=(ImageData(50, 50), q))
        for _ in range(10) ]
    for process in processes:
        process.start()
    results = []
    for process in processes:  # just to have the correct amount of results
        results.append(q.get())
    print '---------Out--------'
    for process in processes:
        process.join()
    print [ result.pid for result in results ]