Mehrere parallele Prozesse (Container) mit Semaphor / Mutex

[Falc410]

Ich teste gerade, ob ich mehrere Container die ein Python Script starten um Daten aus einem NFS Mount vearbeiten, starten kann, damit diese parallel arbeiten.
Das Problem ist, dass ich natürlich nicht 2x die selben Daten verabeiten möchte - gleichzeitig schon gar nicht.

Meine Idee war jetzt die Liste mit Directories im Mount zu shufflen vor der Ausführung, so dass, theoretisch, jedesmal mit einem anderen Ordner gestartet wird.
Bevor ich dann die Daten verarbeite, schaue ich nach, ob es schon ein Output Verzeichnis gibt. Falls nicht, lege ich eins an und erstelle eine leere Datei darin und beginne mit dem Verarbeiten.
Falls es die Datei schon gibt, überspringe ich das Verzeichnis in der Liste und mache mit dem nächsten weiter, da ich davon ausgehe, dass der andere Prozess die schon bearbeitet (gerade jetzt oder schon fertig).

Ist halt nicht perfekt, da theoretisch wenn es doof läuft, die gleichzeitig schauen ob es die Datei gibt und dann gleichzeitig die Datei anlegen und weitermachen. Jemand einen Vorschlag wie ich das optimieren kann?

def check_duplicate_report(dir):
    # Test if new Output Folder already exists, skip if there is an Report Directory for today already
    if not os.path.exists(dir + os.sep + REPORT_DATE_TODAY + os.sep + TOOL_NAME):
        if not os.path.exists(dir + os.sep + REPORT_DATE_TODAY):
            if not DRY_RUN:
                os.mkdir(dir + os.sep + REPORT_DATE_TODAY)
        if not DRY_RUN:
            os.mkdir(dir + os.sep + REPORT_DATE_TODAY + os.sep + TOOL_NAME)           
            open(dir + os.sep + REPORT_DATE_TODAY + os.sep + TOOL_NAME + os.sep + 'started', mode="w").close()
        return False
    elif os.listdir(dir + os.sep + REPORT_DATE_TODAY + os.sep + TOOL_NAME):
        # this only checks if files exists in this folder
        logging.warning("Folder for %s already exists and is not empty! Report might have run previously. Skipping folder..." % (dir + os.sep + REPORT_DATE_TODAY + os.sep + TOOL_NAME))
        return True

 

[NPC]

Mein Vorschlag:

• Script "Manager: Liest Daten aus Mount aus und erstellt einen Job mit den zu verarbeitenden Daten (Liste mit absoluten Dateipfaden o. ä.; ich kenne deine Anforderung nicht). Dieser Job wird an den nächsten freien Worker aus dem Pool gesendet.
• Der "Pool" ist eine Liste mit verfügbaren Workers
• Script "Worker": Verarbeitet den Job. Kann nach Außen kommunizieren, ob er frei ist oder nicht ("idle"/"busy")

Ablauf:

1. Worker starten (1 - 999)
2. jeder Worker registriert sich im Pool
3. Manager in Endlosschleife starten:
   1. Scan nach neuen Files
   2. Wenn neue Files, dann neuen Job erstellen, sonst time.sleep(1)
   3. Wenn Worker frei, dann Job senden, sonst time.sleep(1)
   4. goto 1

Kommunikation zwischen den Scripts kann ganz einfach via Dateisystem oder z. B. SQLite gemacht werden. Wichtig ist nur, das es genau einen Manager gibt.

 

[Falc410]

Ja auf den Manager wollte ich eigentlich verzichten. Das wäre jetzt halt deutlich größerer Entwicklungsaufwand. Aktuell werden die Container in AWS automatisch gestartet und nach Ausführung wieder beendet.
Der Manager / Scheduler müsste jetzt 24x7 als Container laufen, wobei die Container schon von AWS gescheduled werden. Hatte auch überlegt eine Message Queue (Redis / RabbitMQ whatever) zu verwenden. Aber das ist halt overkill. Dazu kommt, dass die Container dann auch alle im selben "Service" laufen müssen, damit die untereinander kommunizieren müssen. Aktuell wird jeder Container seperat für sich gestartet, erledigt sein Script und das wars. Ursprünglich war auch nur ein Container geplant und nicht mehrere parallel.

 

[Iqra]

Das Problem ist der von Dir hinzugefügte Zufall.
Du brauchst eine Möglichkeit, vorm Zuweisen der Threads exakt zu sagen, was wo hin soll und das in Abhängigkeit der Threadanzahl, die du gleichzeitig laufen lassen willst.

Ein Pool/Manager wäre schon die richtige Idee, wenn dir das aber zu aufwendig ist dann könnte man auch einfach hergehen und

• eine nach Kriterium K sortierte Liste der Verzeichnisse erstellen
• diese Liste in so viele Partitionen aufteilen wie es Threads gibt (oder geben soll)
• jeden Thread mit seiner Teil-Liste füttern und ihn diese abarbeiten lassen.

Streng genommen Brauchst du dann noch nicht mal mutexes. Aber: Multithreading ist schon eher weniger deterministisch, da mußt du nicht mutwillig noch mehr Unklarheiten einbauen.

 

[Falc410]

Es ist ja kein Multi-Threading. Es sind wirklich verschiedene Container in denen ein Script liegt.
Theoretisch könnte ich versuchen, einen großen Container laufen zu lassen und dann wie du schon sagst, die Liste aufteilen und mehrere threads in Python zu starten. Das wäre die Alternative.

Aber langfristig werde ich wohl um einen Scheduler nicht drumherum kommen der dann die Container manuell startet und denen per Parameter die Elemente der Liste zuweist.

Gibt es denn vielleicht zufälligerweise schon ein Framework für so etwas? Ich bin ja sicher nicht der Erste, der so etwas machen möchte. Kubernetes oder so, ist ja eher dafür gedacht, dass etwas skaliert aber ich wüsste nicht, dass ich dann noch eine Logik drin habe, die z.B. die Liste aufteilt und jedem Container andere Parameter übergibt. Theoretisch hab noch mehr Skripte die ausgeführt werden sollen wenn die vorherigen Container fertig sind - d.h. ich müsste auch Abhängigkeiten definieren.

 

[marcOcram]

Ist halt nicht perfekt, da theoretisch wenn es doof läuft, die gleichzeitig schauen ob es die Datei gibt und dann gleichzeitig die Datei anlegen und weitermachen. Jemand einen Vorschlag wie ich das optimieren kann?

Bei Windows/NTFS kann man eine Datei exklusiv öffnen und das kann halt nur einer.
Das würde dann zu Folgendem führen:
Welcher Container die Datei anlegt ist dann irrelevant, da nur ein Container den exklusiven Zugriff auf diese Lock-Datei bekommen kann. Dieser Container muss halt den Zugriff solange exklusiv behalten, bis die Bearbeitung beendet wurde.

EDIT: https://github.com/drandreaskrueger/lockbydir das könnte doch was sein.

 

[Falc410]

Danke, so etwas ähnliches habe ich auch schon gesucht. Für NFS habe ich das hier gefunden
https://stackoverflow.com/questions/37633951/python-locking-text-file-on-nfs