How to specify the arguments of a multi-variable function when using multiprocessing Pool in python?

Question

I want to find some of the solutions of the Pythagorean relation using multiprocessing module in python in the following snippet. To speed up the process of finding the answers, I am trying to parallelize the process:

import itertools
from multiprocessing import Pool, Array
import numpy as np

MAXIMUM_INT: int = 10
answers = {'A': [], 'B': [], 'C': []}

base_range = np.arange(1, MAXIMUM_INT + 1)

A_range = B_range = C_range = base_range.tolist()

combinatorics = [A_range, B_range, C_range]
iteration = list(itertools.product(*combinatorics))

dim1 = dim2 = dim3 = MAXIMUM_INT


def init(A: int, B: int, C: int):
    global test1, test2, test3
    test1, test2, test3 = A, B, C


def conditionalAssert(answerDict: dict, A: int, B: int, C: int):
    t1 = np.frombuffer(test1.get_obj())
    t2 = np.frombuffer(test2.get_obj())
    t3 = np.frombuffer(test3.get_obj())

    if A**2 + B**2 == C**2:
        answerDict['A'].append(t1)
        answerDict['B'].append(t2)
        answerDict['C'].append(t3)


if __name__ == '__main__':
    relevantArray = Array('i', dim1 * dim2 * dim3)
    A = B = C = relevantArray
    pool = Pool(processes=7, initializer=init, initargs=(A, B, C))
    pool.starmap(conditionalAssert, iteration)
    print(answers)

However, running the script "as is" creates an error that I don't know how to solve. I would appreciate any help:

multiprocessing.pool.RemoteTraceback: 
"""
Traceback (most recent call last):
  File "C:\Users\username\anaconda3\Lib\multiprocessing\pool.py", line 125, in worker
    result = (True, func(*args, **kwds))
                    ^^^^^^^^^^^^^^^^^^^
  File "C:\Users\username\anaconda3\Lib\multiprocessing\pool.py", line 51, in starmapstar
    return list(itertools.starmap(args[0], args[1]))
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
TypeError: conditionalAssert() missing 1 required positional argument: 'C'
"""

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "c:\Users\username\OneDrive\Desktop\Projects\pythagorean.py", line 38, in <module>
    pool.starmap(conditionalAssert, iteration)
  File "C:\Users\username\anaconda3\Lib\multiprocessing\pool.py", line 375, in starmap
    return self._map_async(func, iterable, starmapstar, chunksize).get()
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\username\anaconda3\Lib\multiprocessing\pool.py", line 774, in get
    raise self._value
TypeError: conditionalAssert() missing 1 required positional argument: 'C'

Can you explain what are you trying to do? I'm seeing that you're doing A**2 + B**2 == C**2, i'm not sure if multiprocessing is the right approach here. — Andrej Kesely
– Andrej Kesely, Commented Jul 23, 2023 at 22:47
This is just a MWE for something else I am trying to solve. I know it's not the right choice but I am trying to experiment with multiprocessing. — Rebel
– Rebel, Commented Jul 23, 2023 at 22:48
iteration generates tuple of 3 elements, but your function conditionalAssert expect 4 parameters. Thus you get the error that the last parameter C is missing. — Andrej Kesely
– Andrej Kesely, Commented Jul 23, 2023 at 22:50
I realized that but I couldn't find a solution to resolve that issue. — Rebel
– Rebel, Commented Jul 23, 2023 at 22:55
You could remove answerDict: dict from the parameter list and instead conditionalAssert should return an object. In the __name__ == '__main__' you get all these objects from .starmap and construct the final one. Another approach is using docs.python.org/3/library/multiprocessing.shared_memory.html — Andrej Kesely
– Andrej Kesely, Commented Jul 23, 2023 at 22:57

Rebel · Accepted Answer · 2023-07-24 02:53:03Z

With the suggestions from Andrej Kesely, I was able to modify my original script to account for the issues that were raised as part of my initial draft. Here is my solution to the problem. (Again, thank you! and please let me know if it can be done better):

import itertools
from multiprocessing import Pool, Array, RawArray
import numpy as np

MAXIMUM_INT: int = 10
answers = {'A': [], 'B': [], 'C': []}

base_range = np.arange(1, MAXIMUM_INT + 1)

A_range = B_range = C_range = base_range.tolist()

combinatorics = [A_range, B_range, C_range]
iteration = list(itertools.product(*combinatorics))

dim1 = dim2 = dim3 = MAXIMUM_INT

def init(A: int, B: int, C: int):
    global test1, test2, test3
    test1, test2, test3 = A, B, C


def conditionalAssert(A: int, B: int, C: int):
    test1 = RawArray("i", dim1)
    test2 = RawArray("i", dim2)
    test3 = RawArray("i", dim3)

    t1 = np.frombuffer(test1, dtype=int)
    t2 = np.frombuffer(test2, dtype=int)
    t3 = np.frombuffer(test3, dtype=int)

    if int(A**2 + B**2) == int(C**2):
        answers['A'].append(A)
        answers['B'].append(B)
        answers['C'].append(C)
        return A, B, C
    return None

if __name__ == '__main__':
    relevantArray = Array('i', dim1 * dim2 * dim3)
    A = B = C = relevantArray
    pool = Pool(processes=7, initializer=init, initargs=(A, B, C))
    for object_returned_from_conditionalAssert in pool.starmap(conditionalAssert, iteration):
        if object_returned_from_conditionalAssert:
            e1, e2, e3 = object_returned_from_conditionalAssert
            answers['A'].append(e1)
            answers['B'].append(e2)
            answers['C'].append(e3)

    print(answers)

Collectives™ on Stack Overflow

How to specify the arguments of a multi-variable function when using multiprocessing Pool in python?

1 Answer 1

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Your Answer

Sign up or log in

Post as a guest

Related