Come sommare un loop in parallelo usando la multiprocessing in Python

Question

Ho difficoltà a capire come utilizzare il modulo di multiprocessing di Python.

Ho una somma 1-n dove n=10^10, che è troppo grande per entrare in un elenco, che sembra essere la spinta dei tanti esempi online utilizzando multiprocessing.

C'è un modo per "dividere" l'intervallo in segmenti di una certa dimensione e quindi eseguire la somma per ogni segmento?

Per esempio

def sum_nums(low,high): 
    result = 0 
    for i in range(low,high+1): 
     result += i 
    return result 

E voglio calcolare sum_nums(1,10**10) rompendo in su in molti sum_nums(1,1000) + sum_nums(1001,2000) + sum_nums(2001,3000)... e così via. So che c'è una forma stretta n(n+1)/2 ma fingiamo di non saperlo.

Ecco che cosa ho provato

import multiprocessing 

def sum_nums(low,high): 
    result = 0 
    for i in range(low,high+1): 
     result += i 
    return result 

if __name__ == "__main__": 
    n = 1000 
    procs = 2 

    sizeSegment = n/procs 

    jobs = [] 
    for i in range(0, procs): 
     process = multiprocessing.Process(target=sum_nums, args=(i*sizeSegment+1, (i+1)*sizeSegment)) 
     jobs.append(process) 

    for j in jobs: 
     j.start() 
    for j in jobs: 
     j.join() 

    #where is the result? 

Answer 1

In primo luogo, il modo migliore per aggirare il problema di memoria è quello di utilizzare un iteratore/generatore invece di una lista:

def sum_nums(low, high): 
    result = 0 
    for i in xrange(low, high+1): 
     result += 1 
    return result 

in python3, range() produce un iteratore, quindi è necessario solo in python2

Ora, quando il multiprocessing entra è quando si vuole dividere il elaborazione in diversi processi o core della CPU. Se non è necessario controllare i singoli lavoratori, il metodo più semplice consiste nell'utilizzare un pool di processi. Questo ti permetterà di mappare una funzione al pool e ottenere l'output. In alternativa è possibile utilizzare apply_async per applicare i lavori alla piscina uno alla volta e ottenere un risultato in ritardo, che si può ottenere con .get():

import multiprocessing 
from multiprocessing import Pool 
from time import time 

def sum_nums(low, high): 
    result = 0 
    for i in xrange(low, high+1): 
     result += i 
    return result 

# map requires a function to handle a single argument 
def sn((low,high)): 
    return sum_nums(low, high) 

if __name__ == '__main__': 
    #t = time() 
    # takes forever 
    #print sum_nums(1,10**10) 
    #print '{} s'.format(time() -t) 
    p = Pool(4) 

    n = int(1e8) 
    r = range(0,10**10+1,n) 
    results = [] 

    # using apply_async 
    t = time() 
    for arg in zip([x+1 for x in r],r[1:]): 
     results.append(p.apply_async(sum_nums, arg)) 

    # wait for results 
    print sum(res.get() for res in results) 
    print '{} s'.format(time() -t) 

    # using process pool 
    t = time() 
    print sum(p.map(sn, zip([x+1 for x in r], r[1:]))) 
    print '{} s'.format(time() -t) 

Sulla mia macchina, basta chiamare sum_nums con 10 ** 10 vogliono quasi nove minuti , ma utilizzando un Pool(8) e n=int(1e8) si riduce a poco più di un minuto.

Answer 2

trovo l'utilizzo di multiprocess.Pool e mappa() molto più semplice

Utilizzando il codice:

from multiprocessing import Pool 

def sum_nums(args): 
    low = int(args[0]) 
    high = int(args[1]) 
    return sum(range(low,high+1)) 

if __name__ == "__main__": 
    n = 1000 
    procs = 2 

    sizeSegment = n/procs 

    # Create size segments list 
    jobs = [] 
    for i in range(0, procs): 
     jobs.append((i*sizeSegment+1, (i+1)*sizeSegment)) 

    pool = Pool(procs).map(sum_nums, jobs) 
    result = sum(pool) 

    >>> print result 
    >>> 500500 

Answer 3

È possibile farlo somma senza multiprocessing a tutti, ed è probabilmente più semplice, se non più veloce, usare solo generatori.

# prepare a generator of generators each at 1000 point intervals 
>>> xr = (xrange(1000*i+1,i*1000+1001) for i in xrange(10000000)) 
>>> list(xr)[:3] 
[xrange(1, 1001), xrange(1001, 2001), xrange(2001, 3001)] 
# sum, using two map functions 
>>> xr = (xrange(1000*i+1,i*1000+1001) for i in xrange(10000000)) 
>>> sum(map(sum, map(lambda x:x, xr))) 
50000000005000000000L 

Tuttavia, se si desidera utilizzare multiprocessing, si può anche fare anche questo. Sto usando un fork di multiprocessing che è meglio in serializzazione (ma per il resto, non proprio diverso).

>>> xr = (xrange(1000*i+1,i*1000+1001) for i in xrange(10000000)) 
>>> import pathos 
>>> mmap = pathos.multiprocessing.ProcessingPool().map 
>>> tmap = pathos.multiprocessing.ThreadingPool().map 
>>> sum(tmap(sum, mmap(lambda x:x, xr))) 
50000000005000000000L 

La versione w/o multiprocessing è più veloce e richiede circa un minuto sul mio portatile. La versione multiprocessing richiede alcuni minuti a causa del sovraccarico di generazione di più processi python.

Se siete interessati, ottenere pathos qui: https://github.com/uqfoundation