Das deutsche Python-Forum

Hallo liebe Python-Freunde,

ich bin relativ neu und hoffe keine komplett doofe Frage zu stellen:

    def multiprocessing(self, func, args, workers):
        with ProcessPoolExecutor(workers) as ex:
            ex.map(func, args)

Da ich knapp 400 Zeilen Code habe, erspare ich mir mal den Rest.

Im großen und ganzen wird hiermit eine Funktion aufgerufen, die verschachtelte Funktionen ausführt. Die Funktionen haben kein Rückgabewert.
Das Programm läuft auch durch wie es soll. Allerdings beendet sich das Programm längere Zeit nicht und spuckt dann den Fehlercode:

Code: Alles auswählen

Error in atexit._run_exitfuncs:
Traceback (most recent call last):
  File "/Users/louismueller/opt/anaconda3/lib/python3.7/concurrent/futures/process.py", line 102, in _python_exit
    thread_wakeup.wakeup()
  File "/Users/louismueller/opt/anaconda3/lib/python3.7/concurrent/futures/process.py", line 90, in wakeup
    self._writer.send_bytes(b"")
  File "/Users/louismueller/opt/anaconda3/lib/python3.7/multiprocessing/connection.py", line 183, in send_bytes
    self._check_closed()
  File "/Users/louismueller/opt/anaconda3/lib/python3.7/multiprocessing/connection.py", line 136, in _check_closed
    raise OSError("handle is closed")
OSError: handle is closed

Hat jemand eine Idee? Ich glaube, ich muss der Funktion ein "Ende" vorgeben, weiß aber nicht genau wie ich das anstellen soll. Wollte es erst mit .join() probieren, bin aber leider gescheitert.

Genau der ersparte Rest ist der entscheidende. Denn die Frage ist, was deine worker machen, damit man weiss, warum sie damit nicht aufhoeren.

Wenn die Funktionen keinen Rückgabewert haben ist IMHO `map()` die falsche Wahl, denn das sammelt ja Rückgabewerte in einer Liste.

Danke erstmal für eure Antworten.

Also dann versuche ich meine Funktion in gekürzter Form zu veröffentlichen:

Code: Alles auswählen

def optimumstart(self, City):
	k=1
        try:
            self.safe = True
            datalist=open("mypath"+City+".txt","r")
            filedaten=datalist.read()
            datalist.close()
            list = ast.literal_eval(filedaten)
            scooter = self.get_scooters(list[0][0], list[0][1], self.token)
            while k < len(list):
                scooter2 = self.get_scooters(list[k][0], list[k][1], self.token)
                scooter = scooter + scooter2
                if(len(scooter)<10):
                    k+=1000
                    print("Unfortunately there are no data for " + City +  " city yet.")
                    self.safe = False
                k+=1
	#duplikateliminierung
        except:
            self.safe = False
            print("Unfortunately there are no data for " + City +  " city yet.")
            scooter = []
        if(self.safe==True):
                    self.savedata(City, scooter)
 def savedata(self, City, data):

        curr_datetime = time.strftime('%Y%m%d-%H%M%S')
        try:
            dir_path = os.path.join(
                self.BASE_DIR,
                'data',
                self.operator, 
                City,
                time.strftime('%Y'), 
                time.strftime('%m'), 
                time.strftime('%d'), 
                time.strftime('%H')
            )
        except:
            dir_path = os.path.join(
                self.BASE_DIR,
                'data',
                self.operator, 
                time.strftime('%Y'), 
                time.strftime('%m'), 
                time.strftime('%d'), 
                time.strftime('%H')
            )
        # create output directory if not exists
        if not os.path.exists(dir_path):
            os.makedirs(dir_path)
        
        # build file path
        self.name = self.operator + "_" + City
        file_path = os.path.join(dir_path, '{0}_{1}.json'.format(self.name, curr_datetime))
        # save data
        with open(file_path, 'w') as f:
            json.dump(data, f)
            
        return not 'error' in data

Übergeben wird eine Liste an Städten, passend dazu holt er sich die Koordinaten aus einer entsprechenden File.
Ohne Multiprocessing läuft alles einwandfrei durch, dauert aber ewig.

Also, habe es jetzt hinbekommen, falls das Problem für zukünftige Coder auftreten sollte, ich habe es wie folgt gelöst:

Code: Alles auswählen

p = multiprocessing.Pool(multiprocessing.cpu_count())
p.map(self.optimumstart, self.cities)
p.close()
p.join()

Außer der nicht gezeigten Methode `get_scooters` ist da nichts, was lange dauern würde.

Was das Format in mypathCity.txt? Das ist doch hoffentlich nicht wirklich Python-Code, den Du hier versuchst mit literal_ast zu parsen? Ein sinnvolles Datenformat wäre json.
`self.safe` sieht aus, als ob es nur lokal verwendet wird, falls nicht, ist das ein grober Designfehler.
Bei `scooter` und `scooter2` ist der Aufruf identisch, so dass der nur einmal im Code vorkommen sollte. Statt einer zufälligen Zahl 1000 auf k zu addieren, um die Liste zu unterbrechen, gibt es `break`.
Die Klammern um if sind unnötig. Die while-Schleife sollte eine for-Schleife über die Einträge der Liste sein, die am besten nicht `list` heißt, weil das der Name des eingebauten Typs ist.
Niemals nackte `except´ benutzen, sondern nur die Fehler, die man auch erwartet, abfangen und verarbeiten.
Was soll den im try-Block in `savedata` schief gehen, wo doch im `except`-Block fast das selbe gemacht wird? Außer der City-Variable, die kein String sein könnte, dann geht aber an anderer Stelle schon etwas schief. Das aktuelle Datum sollte nur EINMAL statt vier mal ermittelt werden, sonst bist Du im Zweifel ein Jahr in die Vergangenheit gereist. `os.makedirs` hat ein Argument, dass es bei existierenden Verzeichnissen keinen Fehler gibt. `self.name` scheint wieder nur ein lokaler Name zu sein.
Mit dem Rückgabewert von `savedata` wird gar nichts gemacht.
So könnte das aussehen (ungetestet):

Code: Alles auswählen

def optimumstart(self, City):
    with open(f"mycity{City}.txt", encoding="utf-8") as data:
        scooter_data = ast.literal_eval(data.read())
    scooters = None
    for scooter in scooter_data:
        scooter2 = self.get_scooters(scooter[0], scooter[1], self.token)
        if scooters is None:
            scooters = scooter2
        else:
            scooters += scooter2
            if len(scooters) < 10:
                print(f"Unfortunately there are no data for {City} city yet.")
                return
    self.savedata(City, scooters)

def savedata(self, City, scooters):
    current_datetime = datetime.datetime.now()
    file_path = os.path.join(
        self.BASE_DIR,
        'data',
        self.operator, 
        City,
        current_datetime.strftime('%Y'), 
        current_datetime.strftime('%m'), 
        current_datetime.strftime('%d'), 
        current_datetime.strftime('%H'),
        f"{self.operator}_{City}_{current_datetime:%Y%m%d-%H%M%S}.json"
    )
    os.makedirs(os.path.dirname(file_path), exist_ok=True)
    with open(file_path, 'w', encoding="utf-8") as f:
        json.dump(scooters, f)
    return 'error' not in scooters

Danke Sirius3 für deine Antwort,

wie gesagt, bin blutiger Anfänger, werde aber alleine deine angedachten Änderungen überprüfen und dann abändern.

Beispielsweise das except mit dem fast ähnlichen Ablauf liegt daran, dass je nach Operator noch eine Stadt mit übergeben wird und mal nicht. Deshalb brauchte ich eine Variante mit und eine ohne City.

Den Fehler mit .txt werde ich schleunigst beheben, dann könnte ich die einzelnen Elemente auch deutlich leichter aufrufen.

Danke für deine Hilfe!

Das deutsche Python-Forum

Error in atexit._run_exitfuncs:

Error in atexit._run_exitfuncs:

Re: Error in atexit._run_exitfuncs:

Re: Error in atexit._run_exitfuncs:

Re: Error in atexit._run_exitfuncs:

Re: Error in atexit._run_exitfuncs:

Re: Error in atexit._run_exitfuncs:

Re: Error in atexit._run_exitfuncs: