Issues with the multiprocessing module

Question

I am trying to write a script where an asset is to be manupulated twice with the second manupulation building upon the first one. Assuming that the first one takes significantly longer than the second one, I was thinking that the best way to do it would be to have multiple workers on the first and a single on the second. I am having issues coding that however. I setup an example to demostrate.

In this example, the asset is a list of strings. The first manupulation is calculating the hash values of the strings and the second one is summing the digits of the calculated hashses. The runtimes are artificially tweaked to create the described effect with time.sleep. To make the manupulations clear, here are the steps for a single string:

>>> s = 'foobar'
>>> h = hash(s)
>>> h
5857481616689290475
>>> n = sum(int(v) for v in str(abs(h)))
>>> n
101

I have managed to get it to work with 1 process handling each manupulation (see below) but I want to have multiple ones on the first.

import multiprocessing as mp
from time import sleep
from random import random


def hasher(q, l, words):
    for word in words:
        l.acquire()
        h = hash(word)
        print('hash of {} is {}'.format(word, h))
        l.release()
        sleep(1.0 * random())
        q.put(h)
    q.put('END')


def summer(q, l):
    while True:
        data = q.get()
        sleep(0.1 * random())
        if data == 'END':
            break
        else:
            l.acquire()
            print('sum of {} is {}'.format(data, sum(int(x) for x in str(abs(data)))))
            l.release()


if __name__ == '__main__':
    queue = mp.Queue()
    lock = mp.Lock()
    words = ['fwiimaafqa', 'nuwivfmgdc', 'foymwgcbut', 'sefmayofio', 'crbgzpihpa',
             'xsioddsfyw', 'zbefmckkyi', 'vkxymewyvt', 'ryrvrfkjqf', 'zobdvstxfh']

    bots = []
    for _ in range(1):
        bot = mp.Process(target=hasher, args=(queue, lock, words))
        bot.start()
        bots.append(bot)


    bot2 = mp.Process(target=summer, args=(queue, lock))
    bot2.start()

    for bot in bots:
        bot.join()
    bot2.join()

can someone help me start more hashers in parallel?

P.S I also tried with Pool but there, I could not get the summer to work in parallel..

Answer 1

If you intend to do it linearly, meaning first hash'em then sum'em, then a Pool will reduce and simplify much of your code. For starters, there is no need for queues and locks. Using the map method, you get in return a list of results so you can create a chain of hashing and summing:

from multiprocessing import Pool
from time import sleep

def hasher(word):
    h = hash(word)
    print('hash of {} is {}'.format(word, h))
    return h

def summer(h):
    s = sum(int(x) for x in str(abs(h)))
    print('sum of {} is {}'.format(h, s))
    return s

if __name__ == '__main__':
    words = ['fwiimaafqa', 'nuwivfmgdc', 'foymwgcbut', 'sefmayofio', 'crbgzpihpa',
             'xsioddsfyw', 'zbefmckkyi', 'vkxymewyvt', 'ryrvrfkjqf', 'zobdvstxfh']

    with Pool() as pool:
        hashes = pool.map(hasher, words)
        sums = pool.map(summer, hashes)

• You can customize the amount of processes in the Pool with the processes argument which defaults to cpu_count().

• The map divides the input list to chunks which can be controlled using the chunks argument. Note that this is an approximation and I would leave it to the Pool to do what it sees is best.

Answer 2

This solution uses a process pool but overlaps the summer processing with the hasher processing so that as soon as a hash becomes available it is submitted to a summer process for summation rather than waiting for all of the hashing to completed. I have modified the summer worker to return the hash argument and sum result back to the main process for printing.

import multiprocessing as mp
from time import sleep
from random import random


def hasher(word):
    h = hash(word)
    print('hash of {} is {}'.format(word, h))
    sleep(1.0 * random())
    return h


def summer(h):
    sleep(0.1 * random())
    s = sum(int(x) for x in str(abs(h)))
    return h, s



def compute_chunksize(number_of_jobs, pool_size):
    chunksize, remainder = divmod(number_of_jobs, pool_size * 4)
    if remainder != 0:
        chunksize += 1
    return chunksize


if __name__ == '__main__':
    N_CPU = mp.cpu_count()
    words = ['fwiimaafqa', 'nuwivfmgdc', 'foymwgcbut', 'sefmayofio', 'crbgzpihpa',
        'xsioddsfyw', 'zbefmckkyi', 'vkxymewyvt', 'ryrvrfkjqf', 'zobdvstxfh']
    with mp.Pool(N_CPU) as pool:
        chunksize = compute_chunksize(len(words), N_CPU)
        results = [pool.apply_async(summer, (h,)) for h in pool.imap(hasher, words, chunksize)]
        for result in results:
            h, s = result.get()
            print('sum of {} is {}'.format(h, s))

Prints:

hash of fwiimaafqa is 2628677472122653395
hash of nuwivfmgdc is 9063591039750236764
hash of foymwgcbut is -6974737365015421151
hash of sefmayofio is 5524859369098572411
hash of crbgzpihpa is -3068552401416065138
hash of xsioddsfyw is 1721137744256960524
hash of zbefmckkyi is -3131190058570978937
hash of vkxymewyvt is -3144743710838890191
hash of ryrvrfkjqf is -5704469091812886058
hash of zobdvstxfh is 5337176127259296686
sum of 2628677472122653395 is 87
sum of 9063591039750236764 is 85
sum of -6974737365015421151 is 77
sum of 5524859369098572411 is 93
sum of -3068552401416065138 is 68
sum of 1721137744256960524 is 76
sum of -3131190058570978937 is 86
sum of -3144743710838890191 is 81
sum of -5704469091812886058 is 91
sum of 5337176127259296686 is 95

Explanation

The call imap takes an iterable (in this case, words) and submit each item of the iterable to the pool as a job. It returns an iterable that when iterated is the next result. imap_unordered is the same except the results are returned in arbitrary order, not necessarily in the order of submission.

Since the hash values are being returned one by one when we iterate the call to imap_unordered, we then use method apply_async to submit the jobs to compute the hash sum. This is a non-blocking job submission request that submits job one-by-one and returns an AsyncResult class instance whose blocking get method can be called to get the result of the job submission. So we are essentially using a list comprehension to submit all the summer jobs and create a list of AsyncResult objects from which we get the results from in submission order). We could have also specified with apply_async a callback function that would be passed the result as soon as it becomes available.

Answer 3

Expanding on the idea of chunking the asset by @Wuerfelfreak, here is my take:

import multiprocessing as mp
from time import sleep
from random import random
from os import getpid
from collections import defaultdict


def chop(iterable, n):
    chunks = defaultdict(list)
    for i, item in enumerate(iterable):
        chunks[(i % n)].append(item)
    return chunks.values()
    

def hasher(q, l, words):
    for word in words:
        if word is not None:
            h = hash(word)
            pid = getpid()
            l.acquire()
            # lock is only to prevent interlaced printing; there should be no computations done here
            print('hash of {} is {} (performed by {})'.format(word, h, pid))
            l.release()
            sleep(1.0 * random())
            q.put(h)


def summer(q, l):
    while True:
        data = q.get()
        if data == 'END': # signal to terminate
            return
        sleep(0.1 * random())
        s = sum(int(x) for x in str(abs(data)))
        pid = getpid()
        # lock is only to prevent interlaced printing; there should be no computations done here
        l.acquire()
        print('sum of {} is {} (performed by {})'.format(data, s, pid))
        l.release()


if __name__ == '__main__':
    queue = mp.Queue()
    lock = mp.Lock()
    words = ['fwiimaafqa', 'nuwivfmgdc', 'foymwgcbut', 'sefmayofio', 'crbgzpihpa',
             'xsioddsfyw', 'zbefmckkyi', 'vkxymewyvt', 'ryrvrfkjqf', 'zobdvstxfh']

    bots = []
    n = 3
    for chunk in chop(words, n):
        bot = mp.Process(target=hasher, args=(queue, lock, chunk))
        bot.start()
        bots.append(bot)

    bot2 = mp.Process(target=summer, args=(queue, lock))
    bot2.start()

    for bot in bots:
        bot.join()
    queue.put('END')  # signal bot2 to terminate
    bot2.join()

This fulfills all requirements since, there are multiple hashers working in parallel and the summing does not wait for all the hashing to be concluded. This is also evident by the print statements:

hash of fwiimaafqa is 873046927385382714 (performed by 22484)
hash of crbgzpihpa is 1605709352436415502 (performed by 4068)
hash of ryrvrfkjqf is -1887526129446447819 (performed by 16488)
hash of zobdvstxfh is 1051323868469477168 (performed by 16488)
sum of -1887526129446447819 is 96 (performed by 21424)
hash of xsioddsfyw is -2508948279346433147 (performed by 4068)
hash of nuwivfmgdc is 2212724786813710734 (performed by 22484)
sum of 873046927385382714 is 87 (performed by 21424)
sum of 1605709352436415502 is 68 (performed by 21424)
sum of 1051323868469477168 is 89 (performed by 21424)
hash of zbefmckkyi is -1045565728264456870 (performed by 4068)
hash of vkxymewyvt is -2286350865213111794 (performed by 4068)
sum of -2508948279346433147 is 89 (performed by 21424)
sum of -1045565728264456870 is 85 (performed by 21424)
sum of -2286350865213111794 is 74 (performed by 21424)
hash of foymwgcbut is -7897655598357124743 (performed by 22484)
sum of 2212724786813710734 is 75 (performed by 21424)
hash of sefmayofio is -8074280815242974134 (performed by 22484)
sum of -7897655598357124743 is 105 (performed by 21424)
sum of -8074280815242974134 is 79 (performed by 21424)

Note: The use of the lock is only to prevent the interleaving of printed output from multiple processes printing in parallel and is not required if you are not printing from these worker functions.

Answer 4

bots = []
n = 4
for _id in range(n):
    bot = mp.Process(target=hasher, args=(queue, lock, words, n, _id))
    bot.start()
    bots.append(bot)

and

def hasher(q, l, words, n, i):
    for word in words[i::n]: # so that every process only works on every n-th word
        ...