Parallel nested for loops in python functions

Question

I have to make several comparisons between strings (all vs all). for this reason I create a nested for loops like this one below:

avg_score = []
strings = get_strings()
for i in range(n)
    count = 0
    score = 0
    for j in range(n)
        if i == j:
            continue
        score += compare(strings[i], strings[j])
    avg_score[i] = score/coount
    do_other_stuff

strings is a list that contains all the text strings to be compared. avg_score is the average score between string i vs all the others. Compare is an external function that us NLP algorithm to compare strings and return a score value (approx 0.25 sec per comparison)

I would like to do this comparison in parallel since it should be faster.

I tried the parfor library, joblib but I do not understand how to process these for loops. probably I have to make the j loop in parallel instead of boths but I have problems when I have to pass argument i and strings to the function.

Answer 1

Your code will generate each pair twice as each for loop will go through every item of the list.

To reduce comparisons use you can use itertools

import itertools
for a, b in itertools.combinations(strings , 2):
    score += compare(a, b)

Or you can modify your loop to the following:

for i in range(n)
    count = 0
    score = 0
    for j in range(i+1,n)
        score += compare(strings[i], strings[j])

This way your total computation time should become half as there are no duplicate iterations here.

Answer 2

One simple approach would be something like:

from multiprocessing import Pool

def comp_score(i):
    s = strings[i]
    return sum(compare(s, r) for r in strings[:i] + strings[i+1:]) / (len(strings) - 1)

if __name__ == '__main__':
    strings = get_strings()
    with Pool(8) as p:
        avg_scores = list(p.map(comp_score, range(len(strings))))

If that works for you depends a bit on what do_other_stuff entails.

EDIT: Regarding your comments. Here's a version of how it could look like. But it doesn't include the right computations, since I can't see how count is actually computed, and I also don't know how the dictionary looks like. So it's more a "that's how you could do it in principle" answer.

from multiprocessing import Pool

def comp_score(i):
    avg_score, count = 1, 2  # Placeholder for calculation of avg_score and count
    return avg_score, count

if __name__ == '__main__':
    strings = get_strings()
    results = {'avg_scores': [], 'counts': []}
    with Pool(8) as p:
        for score, count in p.map(comp_score, range(len(strings))): 
            results['avg_scores'].append(score)
            results['counts'].append(count)

Answer 3

Here an example using multi-threading. Inside your run function you can actually do whatever you want in a multi-threading fashion.

import threading

class my_thread(threading.Thread):
    def __init__(self, strings, indexes, thread_id=0):
        super().__init__()
        self.strings = strings
        self.indexes = indexes
        self.avg_score = []
        self.id = thread_id
    
    def run(self):
        print(f'Thread {self.id} working: from {self.indexes[0]} to {self.indexes[-1]}')
        for i in self.indexes:
            if i > len(self.strings):
                break
            count = 0
            score = 0
            for j in range(len(self.strings)):
                if i == j:
                    continue
                score += compare(self.strings[i], self.strings[j])
                count += 1
            self.avg_score.append(score/count)
            # do other stuff
        print(f'Thread {self.id} done')

avg_score = []
strings = get_strings()
n = len(strings)

# Set the number of threads
num_threads = 10

threads = []
for ii in range(num_threads):
    threads.append(my_thread(strings, range(n//num_threads*ii, n//num_threads*(ii+1)), thread_id=ii))
    threads[-1].start()

for t in threads:
    t.join()
    avg_score.extend(t.avg_score)

Here a reproducible example:

import threading
import time
import random
import string

class my_thread(threading.Thread):
    def __init__(self, strings, indexes, thread_id=0):
        super().__init__()
        self.strings = strings
        self.indexes = indexes
        self.avg_score = []
        self.id = thread_id
    
    def run(self):
        print(f'Thread {self.id} working: from {self.indexes[0]} to {self.indexes[-1]}')
        for i in self.indexes:
            if i > len(self.strings):
                break
            count = 0
            score = 0
            for j in range(len(self.strings)):
                if i == j:
                    continue
                score += compare(self.strings[i], self.strings[j])
                count += 1
            self.avg_score.append(score/count)
            # do other stuff
        print(f'Thread {self.id} done')


def compare(str1, str2):
    time.sleep(0.01)
    return random.random()


strings = [''.join(random.choices(string.ascii_uppercase + string.digits, k=10)) for _ in range(100)]
n = len(strings)


# --- MULTI-THREADING ---

avg_score = []

# Set the number of threads
num_threads = 10

start_time = time.time()

threads = []
for ii in range(num_threads):
    threads.append(my_thread(strings, range(n//num_threads*ii, n//num_threads*(ii+1)), thread_id=ii))
    threads[-1].start()

for t in threads:
    t.join()
    avg_score.extend(t.avg_score)

print(f'Elapsed time with multi-threading: {time.time() - start_time} s')


# --- SINGLE THREAD ---

avg_score = []

start_time = time.time()

for i in range(n):
    count = 0
    score = 0
    for j in range(n):
        if i == j:
            continue
        score += compare(strings[i], strings[j])
        count += 1
    avg_score.append(score / count)
    # do other stuff

print(f'Elapsed time with no threads: {time.time() - start_time} s')