Python-Multiprocessing - Kann Nicht An Aktuellen Thread

Ich arbeite Segmentierung großen ctype-arrays und-Verarbeitung parallel. Ich erhalte den folgenden Fehler und glauben es, weil ein segment des Arrays ist die Verarbeitung beendet worden ist, bevor ein anderer. Ich habe versucht, mit Prozess.join() das erste set der Prozesse, die warten, aber das funktioniert nicht. Ideen?

Exception RuntimeError: RuntimeError('cannot join current thread',) in <Finalize object, dead> ignored

Mit:

    ....

        with closing(multiprocessing.Pool(initializer=init(array))) as p:
            del array #Since the array is now stored in a shared array destroy the array ref for memory reasons

            step = y // cores
            if step != 0:
                jobs =[]
                for i in range (0, y, step):
                    process = p.Process(target=stretch, args= (shared_arr,slice(i, i+step)),kwargs=options)
                    jobs.append(process)
                    process.start()

                for j in jobs:
                    j.join()

    del jobs
    del process

Update:

 #Create an ctypes array
        array = ArrayConvert.SharedMemArray(array)
        #Create a global of options
        init_options(options) #options is a dict
        with closing(multiprocessing.Pool(initializer=init(array))) as p:
            del array #Since the array is not stored in a shared array destroy the array ref for memory reasons


            step = y // cores
            if step != 0:
                for i in range (0, y, step):
                    #Package all the options into a global dictionary

                    p.map_async(stretch,[slice(i, i+step)])

                    #p.apply_async(stretch,args=(shared_arr,slice(i, i+step)),kwargs=options)

        p.join()        

def init_options(options_):
    global kwoptions
    kwoptions = options_

Die Funktion, dass ich mich übergeben map_async gespeichert ist, in einem anderen Modul, so dass ich kämpfen bin, um die Globale kwoptions übergeben, die Funktion. Es scheint nicht Recht zu übergeben globals um zwischen den Modulen wie folgt aus (unpythonic). Ist dies der Weg, um in der Lage sein, um pass kwargs durch map_async.

Sollte ich die Umarbeitung der multiprocessing mit etwas anderes (anwenden oder Prozess?)

  • sollten Sie das nicht schaffen, neue Prozesse, wenn Sie Pool. Siehe ein Beispiel
  • Danke für den link und für die Buchung, die Antwort - es hat geholfen. Ist es möglich, pass map_async eine Reihe von kwargs obwohl? Ich dachte, dass map_async single war arg nur( der Grund, warum ich ging mit Prozess). Ich denke, ich könnte wickeln Sie Sie mit einem lamdba und übergeben der Funktion als arg?
  • Erstellen Sie einen wrapper als eine benannte Globale Funktion pass-Schlüsselwort args zu map_async.
  • Sie können feststellen, dass bugs.python.org/issue15101 ist im Zusammenhang mit diesem und dass dieser behoben wird, was in Kürze 2.7.4.
InformationsquelleAutor Jzl5325 | 2012-08-07
  1. 1

    initializer argument für Pool() akzeptiert eine Funktion; ersetzen initializer=init(array) mit initializer=init, initargs=(array,)

    Übergeben keyword-Argumente zu Funktion f() verwendet mit pool.*map* Familie, die Sie erstellen einen wrapper mp_f():

    #!/usr/bin/env python
import logging
import multiprocessing as mp
from contextlib import closing

def init(shared_array_):
    # globals that should be available in worker processes should be
    # initialized here
    global shared_array
    shared_array = shared_array_

def f(interval, a=None, b=None):
    mp.get_logger().info("interval=%r, a=%r, b=%r" % (interval, a, b))
    shared_array[interval] = [a + interval.start]*b # fake computations

def mp_f(arg_kwargs):
    try:
        arg, kwargs = arg_kwargs
        return f(arg, **kwargs) # pass keyword args to f()
    except Exception:
        mp.get_logger().error("f%r failed" % (arg_kwargs,))

def main():
    mp.log_to_stderr().setLevel(logging.INFO)

    N = 10**6
    array = mp.RawArray('i', N) # create shared array

    # create workers pool; use all available CPU cores
    with closing(mp.Pool(initializer=init, initargs=(array,))) as p:
        options = dict(a=5, b=N//4) # dummy options
        step = options['b']
        args = ((slice(i, i+step), options) for i in range(0, N, step))
        for _ in p.imap_unordered(mp_f, args): # submit jobs
            pass
    p.join()
    mp.get_logger().info(array[::step])

if __name__=="__main__":
    mp.freeze_support() # for py2exe and the-like on Windows
    main()
    InformationsquelleAutor jfs
  2. 2

    Also habe ich diese Arbeit durch Nacharbeit der code und entfernen von pool (nach J. F. Sebastians Kommentar).

    In pseudo-code:

    initialize the shared array
determine step size
create an empty list of jobs
create the process, pass it the kwargs, and append it to the job list
start the jobs
join the jobs

    Hier ist der code, wenn das hilft keine googler:

    #Initialize the ctypes array
        init(array)
        #Remove the reference to the array (to preserve memory on multiple iterations.
        del array

        step = y // cores
        jobs = []
        if step != 0:
            for i in range(0,y,step):        
                p = multiprocessing.Process(target=stretch,args= (shared_arr,slice(i, i+step)),kwargs=options)
                jobs.append(p)

            for job in jobs:
                job.start()
            for job in jobs:
                job.join()
    • Es gibt keine Punkt-zu haben mehr Prozesse als CPUs zur Verfügung (für CPU-bound jobs). mp.Pool ermöglicht die separate Anzahl der worker-Prozesse und die Zahl der Arbeitsplätze also, die zum gleichen Arbeitsprozess in einem pool ausgeführt werden können mehrere Aufträge in der Reihenfolge.
    InformationsquelleAutor Jzl5325
Schreibe einen Kommentar