Das deutsche Python-Forum

@chris_adnap: @chris_adnap: Man sollte keine kryptischen Abkürzungen verwenden. Wenn man `thread` meint, nicht nur `th` schreiben.

Das kann man mit `any()` deutlich kompakter ausdrücken:

Aber das rettet einen leider nicht davor, dass hier eine „race condition“ vorliegt.

Erschwerend kommt dazu das die Aufrufe nicht mal im gleichen Process sein müssen, weil WSGI-Anwendungen auch auf mehreren Prozessen laufen können.

Das muss man beispielsweise über die Datenbank synchronisieren. Also in einer Transaktion prüfen ob der Wert für „Update läuft“ da drin steht, und falls nicht den reinschreiben und commiten, und nur wenn das erfolgreich war, den Thread starten.

Oder man schreibt sich einen eigenen Dienst für das Update.

Oder man verwendet eine Task Queue wie Celery dafür.

__deets__ hat geschrieben: Samstag 9. September 2023, 10:58 ... Installier dir Celery, und gut ist.

DeaD_EyE hat geschrieben: Samstag 9. September 2023, 12:07 Wie schon die Vorredner sagten, ist Nebenläufigkeit nicht so einfach.

Ein Ansatz wäre es, einen ThreadPoolExecutor zu verwenden und nur einen Task zuzulassen.
Wenn man mehrere Hintergrundprozesse will, dann wird es etwas komplexer, da man dann ja die Tasks irgendwas zuordnen muss (z.B. einem eingeloggtem User).

Die Routen würde ich aufteilen. Eine Route um die Berechnung zu starten und eine Route für das Ergebnis.

Sirius3 hat geschrieben: Samstag 9. September 2023, 13:28 @chris_adnap: um Threads zu koordinieren, benutzt man passende Datensturkturen, wie Queues, Locks oder Events:

__blackjack__ hat geschrieben: Samstag 9. September 2023, 14:32 Das mit den Prozessen hast Du richtig verstanden. Jeder Prozess hat seine eigenen Threads.

Mit Dienst war etwas wie Celery gemeint, nur eben speziell für diese Aufgabe statt einer allgemeinen Task Queue. Ein Prozess der nur das Update macht wenn er dazu beauftragt wird. Und der die Regel dass nur ein Update zur gleichen Zeit läuft sicherstellt.

@home_blueprint.route('/test', methods=['GET'])
def test():
    thread_name = 'force_update'
    if not TP.ForceUpdate().status_check()[0]:  # PRÜFE OB PROZESS GERADE LÄUFT
	    TP.ForceUpdate().status_set('start_force_update'):  # <-- FALLS NICHT, SETZE ICH GLEICH ALS ERSTES DEN WERT VON FALSE AUF TRUE

	    # start des Update in einem Thread. GANZ ZUM SCHLUSS SETZE ICH MIT ('stop_force_update')" den Wert wieder auf FALSE + setze datetime.now() einen Timestamp.
            threading.Thread(target=TP.ForceUpdate().start_force_update, name=thread_name, daemon=False).start()  
            return 'Update Prozess startet. Bitte warten', 200
            
    else:
        if not TP.ForceUpdate().status_check()[1]:
            return 'Update Prozess läuft gerade. Bitte warten.', 200
        else:
            return TP.ForceUpdate().status_check()[1], 200
            
 ....
 
def status_check(self):
    force_update_status = self.session.query(db.Settings).filter(db.Settings.option == 'force_update_running').one()
    if force_update_status.boolean:
        return True, None

__deets__ hat geschrieben: Montag 11. September 2023, 14:16 Ohne zu sehen, wie das mit der Datenbank geloest ist, kann man da nicht wirklich was zu sagen, ob es denn richtig funktioniert. Die hilft einem sicher dabei, weil sie einen Weg darstellt, einen verteilten Synchronisationsmechanismus aufzubauen. Ublicherweise durch select for update und Konsorten. Aber muss eben auch richtig gemacht werden.

Sirius3 hat geschrieben: Montag 11. September 2023, 14:51 @chris_adnap: Datenbank alleine löst Dein Problem nicht, da braucht es auch eine explizite Synchronisation der Prozesse, automatisch passiert da gar nichts. Für viele Datenbanksysteme gibt es da entsprechende Mechanismen, die man aber kennen muß.
Welche Datenbank benutzt Du und wie sieht Deine Lösung jetzt aus?

Aber das rettet einen leider nicht davor, dass hier eine „race condition“ vorliegt. Wenn die Funktion zweimal aufgerufen wird, kann es passieren das beide im ``if`` zum Schluss kommen, dass gerade kein Update-Thread läuft bevor sie ihn dann beide starten.

__deets__ hat geschrieben: Montag 11. September 2023, 22:15 Doch, das ist so falsch. Weil es immer noch die schon von __blackjack__ angesprochene Lücke zwischen Test und Update hat. Da ist noch genau gar nichts gewonnen.

Der richtige Weg beinhaltet vorher die Tabelle mit select for update zu sperren. Dann garantiert die DB, das ein weiterer Thread/Prozess/Client da nichts dran ändern kann. Dann prüft und setzt man den Status.

Und wenn dein System aus irgendwelchen Gründen abschmiert währen die Flagge gesetzt ist, ist es kaputt. Auch beim select for Update übrigens. Da muss man dann noch weiter gehen.

Ich bin immer noch der Meinung, das wäre inzwischen mit einem dafür entworfenen System wie celery gelöst. Aber du hast dich für den Weg der vielen Fehler entschieden

https://github.com/cameronmaske/celery-once

Sirius3 hat geschrieben: Montag 11. September 2023, 22:26 Das Thema korrekte Synchronisation ist schwierig und deshalb ist es nicht verwunderlich, dass Dein neuer Ansatz genauso fehlerhaft ist, wie die davor.

Die Lösung ist dagegen ganz einfach, mit with_for_update: https://docs.sqlalchemy.org/en/20/core/ ... for_update

Code: Alles auswählen

def set_is_running(self):
    force_update_status = self.session.query(db.Settings).filter_by(option='force_update_running').with_for_update().first()
    result = force_update_status.boolean
    if not result:
        force_update_status.boolean = True
    session.add(force_update_status)
    session.commit()
    return result

Die Methode hat natürlich noch den Schönheitsfehler, dass falls der Update-Prozess abstürzt, das Flag dauerhaft auf True steht und somit nie wieder ein Update gemacht werden kann. In Deinem Fall, wo Du weißt, dass der Updateprozess maximal ein paar Minuten dauern kann, wäre die einfachste Lösung, einen Timeout zu implementieren, dass man also die aktuelle Zeit speichert und prüft, wie lange der letzte Update schon her ist.

Sonstige Anmerkungen, TP ist eine kryptische Abkürzung, die zudem noch falsch geschrieben ist, weil TP eine Konstante wäre, was ja das Objekt offensichtlich nicht ist.
Es ist falsch, ständig eine neue Instanz von ForceUpdate zu erzeugen. Das läst befürchten, dass es sich gar nicht um eine richtige Klasse handelt.

thread_name = 'force_update'
force_update = tp.TPForceUpdate()
force_update_status = force_update.status_check()
if not force_update_status[0]:
    if force_update.status_set('start_force_update'):
        threading.Thread(target=force_update.start_force_update, name=thread_name, daemon=False).start()
        return 'Update Prozess startet. Bitte warten', 200
    else:
        return 'Fehler beim starten des Force Update Prozess.', 200
else:
    if not force_update_status[1]:
        return 'Update Prozess läuft noch. Bitte warten.', 200
    else:
        return force_update_status[1], 200