Multiprocessing to return large data sets in Python

user_10997344 0
Viewed 820

I have 2 functions in a Python 3.7 script that search 2 separate network nodes and returns very large data sets of strings in a list. The smaller data set length is ~300K entries, while the larger one is ~1.5M. This script takes almost an hour to execute because of how it has to compile the data sets as well as having the second data set be significantly larger. I don't have a way to shorten the run time by changing how the compilation happens, there's no easier way for me to get the data from the network nodes. But I can cut almost 10 minutes if I can run them simultaneously, so I'm trying to shorten the run time by using multiprocessing so I can run both of them at once.

I do not need them to necessarily start within the same second or finish at the same second, just want them to run at the same time.

Here's a breakdown of first attempt at coding for multiprocessing:

def p_func(arg1, arg2, pval):
     ## Do Stuff
    return pval

def s_func(arg1, sval):
    ## Do Stuff
    return sval

# Creating variables to get return values that multiprocessing can handle
pval = multiprocessing.Value(list)
sval = multiprocessing.Value(list)

# setting up multiprocessing Processes for each function and passing arguments
p1 = multiprocessing.Process(target=p_func, args=(arg1, arg2, pval))
s2 = multiprocessing.Process(target=s_func, args=(arg3, sval))
p1.start()
s1.start()
p1.join()
s1.join()

print("Number of values in pval: ", len(pval))
print("Number of values in sval: ", len(sval))

I believe I have solved my list concerns, so....

Based on comments I've updated my code as follows:

#! python3
import multiprocessing as mp
def p_func(arg1, arg2, pval):
    # takes arg1 and arg2 and queries network node to return list of ~300K
    # values and assigns that list to pval for return to main()
    return pval

def s_func(arg1, sval):
    # takes arg1 and queries network node to return list of ~1.5M 
    # values and assigns that list to sval for return to main()
    return sval

# Creating variables to get return values that multiprocessing can handle in 
# main()
with mp.Manager() as mgr
    pval = mgr.list()
    sval = mgr.list()

    # setting up multiprocessing Processes for each function and passing 
    # arguments
    p1 = mp.Process(target=p_func, args=(arg1, arg2, pval))
    s1 = mp.Process(target=s_func, args=(arg3, sval))
    p1.start()
    s1.start()
    p1.join()
    s1.join()

# out of with block
print("Number of values in pval: ", len(pval))
print("Number of values in sval: ", len(sval))

Now getting a TypeError: can't pickle _thread.lock objects on the p1.start() invocation. I'm guessing that one of the variables I have passed in the p1 declaration is causing a problem with multiprocessing, but I'm not sure how to read the error or resolve the problem.

python-multiprocessing python-3.x
1 Answers

Use a Manager.list() instead:

import multiprocessing as mp

def p_func(pval):
    pval.extend(list(range(300000)))

def s_func(sval):
    sval.extend(list(range(1500000)))

if __name__ == '__main__':
    # Creating variables to get return values that mp can handle
    with mp.Manager() as mgr:
        pval = mgr.list()
        sval = mgr.list()

        # setting up mp Processes for each function and passing arguments
        p1 = mp.Process(target=p_func, args=(pval,))
        s2 = mp.Process(target=s_func, args=(sval,))
        p1.start()
        s2.start()
        p1.join()
        s2.join()

        print("Number of values in pval: ", len(pval))
        print("Number of values in sval: ", len(sval))

Output:

Number of values in pval:  300000
Number of values in sval:  1500000

Manager objects are slower than shared memory but more flexible. Shared memory is faster, so if you know an upper limit for your arrays, you could use a fixed-sized shared memory Array and a shared value indicating the used size instead, such as:

#!python3
import multiprocessing as mp

def p_func(parr,psize):
    for i in range(10):
        parr[i] = i
    psize.value = 10

def s_func(sarr,ssize):
    for i in range(5):
        sarr[i] = i
    ssize.value = 5

if __name__ == '__main__':
    # Creating variables to get return values that mp can handle
    parr = mp.Array('i',2<<20) # 2M
    sarr = mp.Array('i',2<<20)
    psize = mp.Value('i',0)
    ssize = mp.Value('i',0)

    # setting up mp Processes for each function and passing arguments
    p1 = mp.Process(target=p_func, args=(parr,psize))
    s2 = mp.Process(target=s_func, args=(sarr,ssize))
    p1.start()
    s2.start()
    p1.join()
    s2.join()

    print("parr: ", parr[:psize.value])
    print("sarr: ", sarr[:ssize.value])

Output:

parr:  [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
sarr:  [0, 1, 2, 3, 4]
user_2356980
Related