Multiprocessing - map - ich hab Kopfweh O_o

[patmaster]

Hi,

Ich verwende Python 2.7.

Ich muss eine Funktion für jedes Element eines generators ausführen. Nachdem die Funktion recht lange braucht, wollte ich das Ganze multiprocessen.
Dadurch wurde das Ding zwar wirklich erheblich schneller, nur wenn der generator 50 mal returned steh ich mit 50 Prozessen da und das ist nicht so toll.

Ich habe also nach einem Weg gegoogelt um die Anzahl an Prozessen zu limitieren und bin dabei auf Pool gestoßen, das dort so vorgeführt wird:

Code: Alles auswählen

import multiprocessing

def do_calculation(data):
    return data * 2

def start_process():
    print 'Starting', multiprocessing.current_process().name

if __name__ == '__main__':
    inputs = list(range(10))
    print 'Input   :', inputs
    
    builtin_outputs = map(do_calculation, inputs)
    print 'Built-in:', builtin_outputs
    
    pool_size = multiprocessing.cpu_count() * 2
    pool = multiprocessing.Pool(processes=pool_size,
                                initializer=start_process,
                                )
    pool_outputs = pool.map(do_calculation, inputs)
    pool.close() # no more tasks
    pool.join()  # wrap up current tasks

Sieht ja mal einfach aus, nur leider bekommt meine Funktion, also mein Pendant zu do_calculation() nicht nur 1 Argument sondern 4, wobei 3 davon keine Listen sind. Sprich der 1 Parameter sollte jeweils das "aktuelle" Element der Liste (hier inputs) sein und der Rest andere Werte, dich ich als Variablen übergeben würde.

Ich habe google bemüht und man findet da auch einiges in stackoverfow aber ich krieg das einfach nicht gebacken.

Könnte mir jemand anhand des Beispiels erklären wie man das machen würde ?

Hier noch die Funktion die aufgerufen werden soll:

Code: Alles auswählen

def linkxmlfiles(xmlfile, verbose, data_newgeslinks, internal_linking, anz):    
    try:            
        if verbose:
            print xmlfile
        infd = open(xmlfile)
        intext = infd.read()
        outfd = open(os.path.join(os.path.normpath(data_newgeslinks),
                                    os.path.basename(xmlfile)),
                        'w')
        outtext = insert_lawlinks(intext, internal_linking)
        try:
            outfd.write(outtext)
        except:
            #Tritt ein Fehler auf, schreib einfach das Eingangsfile...sicher ist sicher!
            print "ERROR: Musste Eingangsfile schreiben!!!"
            print(traceback.format_exc())               
            outfd.write(intext)
        infd.close()
        outfd.close()
    except:
        print(traceback.format_exc())
    return

Vielen Dank schon mal im Vorraus.

[senft]

Ab Python 3.3 gibt es pool.starmap(), wo die Elemente des Iterables wieder Iterables sind (die dann "unpacked" werden).

Code: Alles auswählen

#!/usr/bin/env python
# encoding: utf-8

import multiprocessing


def do_calculation(data, data2):
    return (data * 2, data2 * 3)


def start_process():
    print('Starting', multiprocessing.current_process().name)


if __name__ == '__main__':
    inputs = list(zip(range(10), range(10)))
    print('Input   :', inputs)

    pool_size = multiprocessing.cpu_count() * 2
    pool = multiprocessing.Pool(processes=pool_size,
                                initializer=start_process,
                                )
    pool_outputs = pool.starmap(do_calculation, inputs)
    pool.close()  # no more tasks
    pool.join()  # wrap up current tasks

Ansonsten, könntest du es mit functools.partial versuchen.

EDIT: Sorry.. zu spät gesehen, dass du Python 2.7 verwendest. Wobei das ja nicht sagt, dass du zwingend daran gebunden bist...

[patmaster]

Bin leider an 2.7 gebunden (in diesem Fall).
partial() hat's gebracht. Ich bin darauf zwar schon über stackoverflow gestoßen, aber ne gezieltere Suche hat einfachere Beispiele ergeben...danke