Benachrichtigung, wenn eine neue Aufgabe in ipython parallel gestartet wird

Question

Ein paar Fragen zu Best Practices mit ipyparallel. Ich versuche es zu verwenden, um ein Monte-Carlo-Framework für ein Modell zu implementieren, das ~ 15 zum Ausführen benötigt. Die Idee besteht darin, N-Engines (über SLURM) auszuführen und einen "Master" -Prozess zu haben, der alle benötigten Tasks asynchron abfängt und busy-wartet auf den Abschluss, wobei eine sqlite db mit dem Status jedes Laufs aktualisiert wird.

Ich würde gerne wissen, wenn eine Aufgabe einer Engine zugewiesen wurde, so dass ich ihren Status in meiner Datenbank verfolgen kann. Ich versuchte, die AsyncResult-Instanzen zu verwenden, um die msg_id abzurufen und die Task-Datenbank abzufragen, aber das "gestartete" Feld wird nicht aktualisiert, bis die Task abgeschlossen ist.

Es scheint, es sollte eine Möglichkeit geben, diese Benachrichtigung zu erhalten, oder zumindest den Hub abzufragen, während die Engine arbeitet.

Muss ich auch etwas unternehmen, um den Heartbeat-Timeout des Motors während einer lang andauernden Aufgabe zu vermeiden? Ist das der Zweck von client.spin_thread()?

Danke!

Answer 1

Ich habe einen Teil meiner eigenen Frage mit publish_data beantwortet. Die Idee ist, dass ich, anstatt einfach nur meine Hauptfunktion "worker" in jeder Engine aufzurufen, publish_data() vor und nach der Haupt-Worker-Funktion anrufe, um einen Status festzulegen, den der Client sehen kann. Zum Beispiel:

def wrapper(run_id, argDict): 
    from ipyparallel.engine.datapub import publish_data 

    publish_data({run_id : 'running'}) 
    status = runMonteCarloTrial(argDict) # runs for ~15 minutes 
    publish_data({run_id : status}) 
    return status 

Der „Master“ Aufgabe ruft:

ar = client.map_async(wrapper, listOfArgDicts) 

ich dann Schleife über ar bis alle AsyncResults abgeschlossen sind, ar.data Prüfung der veröffentlichten Daten zu lesen Studien zu identifizieren laufen und speichern Studienergebnisse zu einer SQLite3-Datenbank.

Dieser allgemeine Ansatz funktioniert für einen einfachen Testfall. Ich habe die Timeout-Frage für lang andauernde Funktionsaufrufe noch untersucht.