Python parallelized code is taking much longer than it should

Question

EDIT: My goal with this post is to understand the source of the time-drain. I welcome other recommendations as well, but my main concern, and I want to learn about is why my code is not speeding up with parallelization? What is it that's causing the parallelization to slow down?

I previously asked a question about this, and I now realize that it wasn't a good one. I apologize for the poor post. So I'm re-asking having put more effort into solving it.

I have managed to implement a parallelized solution. However, the parallelized code is much much much slower than the serialized version.

EDIT: The foo() function below is rather simple, and can be put more concisely, but the real version of the function is a bit more complicated. The main problem is still the fact that at thousands of arrays, each of length ~70,000, the sheer number of comparisons is what's causing the slowness. So parallelization seems to be the best solution here. Of course, recommendations for making the steps more efficient are welcome, and I appreciate any such suggestions.

Problem

Consider a list of numpy arrays. I need to do pairwise comparisons on these arrays in the list. My real problem has thousands of arrays of length ~70,000, but the toy example below has much smaller numbers (can be adjusted with the listLen and arrayLen variables though)

Attempt

Here foo() is the comparison function that will be used. If you try playing around with arrayLen and listLen, you'll see that no matter what values you choose, the parallelized function do_calculations_mp is always slower than the non-parallelized version do_calculations_no_mp. From what I've read, multiprocesing.Process has less overhead than multiprocess.Pool, so it shouldn't be taking this long, right?

I'd really appreciate any help on this.

import numpy as np
from multiprocessing import Process
import itertools
import random
import sys
from datetime import datetime

def foo(arr1, arr2):
    matches = 0
    for i in range(len(arr1)):
        if arr1[i] == arr2[i]:
            matches += 1
    return(matches)


def do_calculations_mp(aList):
    flag_indices = []

    processes = []

    index_combns = list(itertools.combinations(range(len(aList)),2))
    for i,j in index_combns:
        p = Process(target = foo, args = (aList[i], aList[j]))
        processes.append(p)
        p.start()

    for procs in processes:
        procs.join()

    return(flag_indices)


def do_calculations_no_mp(aList):
    flag_indices = []
    index_combns = list(itertools.combinations(range(len(aList)),2))
    for i,j in index_combns:
        numMatches = foo(aList[i], aList[j])

    return(flag_indices)

if __name__ == '__main__':
    listLen = 50
    arrayLen = 300
    # Creates a list of listLen arrays, where each array has length arrayLen
    myList = [np.array([random.choice([0,1,2,5]) for i in range(arrayLen)]) for x in range(listLen)]

    print("Processing No MP:             " + str(datetime.now()))
    flagged = do_calculations_no_mp(myList)
    print("Done processing No MP:        " + str(datetime.now()))

    print("Processing MP:                " + str(datetime.now()))
    flagged_mp = do_calculations_mp(myList)
    print("Done processing MP:           " + str(datetime.now()))

Answer 1

A few general points:

1. You have a finite number of processors. Your code is creating a number of processes equal to the number of combinations that you have, which might (greatly) exceed the number of processors and that is rather pointless.
2. You cannot return results from a process with your return statement. You need another mechanism for doing so, such as passing a multiprocessing.Queue instance to your processes to which results are written (there are other ways).
3. Using a multiprocessing pool solves issues 1. and 2.
4. Keeping the arrays in shared memory will reduce the overhead of multiprocessing.
5. The more CPU-intensive foo is, the better multiprocessing will perform compared to serial processing. This means having long lists.

I have modified your demo incorporating these ideas to demonstrate that multiprocessing can outperform serial processing. Note that foo now accesses myList as a global variable, which is a list of shared memory arrays, and so it now only has to be passed list indices.

from multiprocessing import Pool, Array
import itertools
import random
import sys
from time import time_ns


listLen = 6
arrayLen = 5_000_000

def foo(i, j):
    arr1 = myList[i]
    arr2 = myList[j]
    matches = 0
    for index in range(arrayLen):
        if arr1[index] == arr2[index]:
            matches += 1
    return(matches)

def generate_args():
    index_combns = list(itertools.combinations(range(listLen), 2))
    for i, j in index_combns:
        yield (i, j)

def init_pool_processes(aList):
    global myList
    myList = aList

def do_calculations_mp():
    numMatches = 0
    with Pool(initializer=init_pool_processes, initargs=(myList,)) as pool:
        for matches in pool.starmap(foo, generate_args()):
            numMatches += matches
        pool.close()
        pool.join()
    return numMatches

def do_calculations_no_mp():
    numMatches = 0
    for i, j in generate_args():
        numMatches += foo(i, j)
    return numMatches

if __name__ == '__main__':
    # Creates a list of listLen arrays, where each array has length arrayLen
    myList = [Array('i', [random.choice([0,1,2,5]) for i in range(arrayLen)], lock=False) for x in range(listLen)]

    start = time_ns()
    numMatches = do_calculations_no_mp()
    elapsed = (time_ns() - start) / 1_000_000_000
    print(f"Done processing No MP, numMatches = {numMatches}, elapsed time = {elapsed}")

    start = time_ns()
    numMatches = do_calculations_mp()
    elapsed = (time_ns() - start) / 1_000_000_000
    print(f"Done processing MP, numMatches = {numMatches}, elapsed time = {elapsed}")

Prints:

Done processing No MP, numMatches = 18748321, elapsed time = 10.0353228
Done processing MP, numMatches = 18748321, elapsed time = 2.7033887

With listLen defined as 70_000, however, which is approximately the length of your lists, we get:

Done processing No MP, numMatches = 261721, elapsed time = 0.1350159
Done processing MP, numMatches = 261721, elapsed time = 0.2939995

So the issue is that until your lists become longer, foo at the current complexity, is not sufficiently CPU-intensive to be more performant with multiprocessing. Of course, if your actual foo is far more complex, then ...

As the number of combinations become very large, you might want to consider using method multiprocessing.pool.imap_unordered with a suitable chunksize argument. foo would then need to be redefined to take a single tuple argument.

Answer 2

If performance matters, let's start from principles

Python Interpreter is interpreting instructions, use dis.dis( foo ) to see them below.

Python Interpreter is expensive if going to spawn processes (as detailed in your first question)

Python Interpreter is known to be awfully bad (read slow) if using serial-iterators (read for-loops, generator-based task-farming et al)

Python Interpreter in distributing "lightweight" tasks for scales of about 18M tasks, combined from data from 6E3 list-of-items, each item being about 1E6 long np.array-instance - as you described in your (now deleted) second question, will almost for sure kill its own capabilities (performance-wise), if all above are used at once inside multiprocessing.

def foo(arr1, arr2):
    matches = 0
    for i in range(len(arr1)):
        if arr1[i] == arr2[i]:
           matches += 1
return(matches)

was demonstrated as the MCVE-representation of the work-unit ( a task )

>>> dis.dis( foo )
  2           0 LOAD_CONST               1 (0)
              3 STORE_FAST               2 (matches)

  3           6 SETUP_LOOP              59 (to 68)
              9 LOAD_GLOBAL              0 (range)
             12 LOAD_GLOBAL              1 (len)
             15 LOAD_FAST                0 (arr1)
             18 CALL_FUNCTION            1
             21 CALL_FUNCTION            1
             24 GET_ITER            
        >>   25 FOR_ITER                39 (to 67)
             28 STORE_FAST               3 (i)

  4          31 LOAD_FAST                0 (arr1)
             34 LOAD_FAST                3 (i)
             37 BINARY_SUBSCR       
             38 LOAD_FAST                1 (arr2)
             41 LOAD_FAST                3 (i)
             44 BINARY_SUBSCR       
             45 COMPARE_OP               2 (==)
             48 POP_JUMP_IF_FALSE       25

  5          51 LOAD_FAST                2 (matches)
             54 LOAD_CONST               2 (1)
             57 INPLACE_ADD         
             58 STORE_FAST               2 (matches)
             61 JUMP_ABSOLUTE           25
             64 JUMP_ABSOLUTE           25
        >>   67 POP_BLOCK           

  6     >>   68 LOAD_FAST                2 (matches)
             71 RETURN_VALUE        

Using the amount of pseudo-instructions as a simplified measure of how much work has to be done ( where all of CPU-clocks + RAM-allocation-costs + RAM-I/O + O/S-system-management time spent must count and counts the actually accrued costs ), we start to see the relative-costs of all these ( unavoidable ) add-on costs, compared to the actual useful task ( i.e. how many pseudo-instructions are finally spent on what we want to compute --The-useful-WORK--, contrasting the amounts of already burnt overhead-costs and per-task add-on costs, that were needed to make this happen and run )

This fraction is cardinal if fighting for both performance & efficiency (of resources usage patterns).

For cases, where add-on overhead costs dominate, these cases are straight a sin of keeping anti-patterns stacked to their worst consequences

For cases, where instantiation + per-task add-on overhead costs make less than 0.01 % of the useful work, we still may result in pretty unsatisfactory low speedups (see the simulator and related details, as posted yesterday and see the devastated speedup line - i.e. from a day-long runtimes today, you will receive about a day-long runtimes, if keeping all these add-on costs as high as now )

For cases, where the scope of the useful work ( the actual foo(), yet performance-boosted to strip-off most of its performance antipatterns ) strongly dominates, i.e. diminishes all add-on + per-task overhead costs, there we still see the Amdahl's Law ceiling - called so, so self-explanatory - a "Law of diminishing returns" ( since adding more resources ceases to improve the performance, even if we add infinitely many CPU-cores and likes )

Let's test the foo()-speeds above ( as-was presented ) and below ( as-refactored ) :

from zmq import Stopwatch; aClk = Stopwatch()       # a [us]-resolution clock
import numpy as np

def test( aSize = 1E6 ):
    TheSIZE = int( min( 1E7, aSize ) )              # fused RAM-footprint < GB
    A = np.random.random_integers( 0, 9, TheSIZE )  # pre-allocate, fill
    B = np.random.random_integers( 0, 9, TheSIZE )  # pre-allocate, fill
    aClk.start() #----------------------------------# CRITICAL SECTION.start
    _ = np.where( A==B, 1, 0 ).sum()                #    smart compute "count"  
    t = aClk.stop() #-------------------------------# CRITICAL SECTION.end
    MASK = ( "test( aSize = {0:>20d} ) "
           + "took {1: >20d} [us] "
           + " {2:} matches found in random filled A, B "
           + " ~ {3:} [us/loop]"
             )
    print MASK.format( TheSIZE, t, _, float( t ) / TheSIZE )

use the same to measure how long does the foo( A, B ) last in your serial-iterator code

a "naked"-foo( A, B ) took about 800 [ns] (!!) per array-cell
( to that we mast add all SER/xfer/DES costs we pay for moving ~ 20[MB] of parameters' data, per task, which may take and costs hundreds of thousands of CPU-clocks, just to move the data into a remote process (worker), before it starts a few pseudo-instructions dis.dis( foo ) shown above, ouch, that hurts ... )
while
np.where()-code took under about 19 [ns] 42 X faster per array-cell ( 25 [ns] if all 4-cores on my localhost were background-loaded with desktop system running 3 TV-streams live, so less free CPU-cores to use numpy-multicore code tricks on-the-fly )

>>> test( 1E34 )
took 239712 [us]  999665 matches found in random filled A, B  ~ 0.0239712 [us/loop]
took 220671 [us] 1000246 matches found in random filled A, B  ~ 0.0220671 [us/loop]
took 227004 [us] 1001805 matches found in random filled A, B  ~ 0.0227004 [us/loop]
took 210587 [us]  999267 matches found in random filled A, B  ~ 0.0210587 [us/loop]
took 195863 [us]  998218 matches found in random filled A, B  ~ 0.0195863 [us/loop]
took 241407 [us] 1000017 matches found in random filled A, B  ~ 0.0241407 [us/loop]
took 193243 [us] 1001084 matches found in random filled A, B  ~ 0.0193243 [us/loop]

From the perspective of this comparison, start to try to properly refactor the process - the goal is to crawl all ~ 18M-pair-wise combinations from a 6E3-long list of ~10[MB]-large-np.arrays for all pairwise comparisons.

The faster the better, right?