Notification lorsqu'une nouvelle tâche démarre en parallèle ipython

Question

Quelques questions relatives aux meilleures pratiques avec ipyparallel. J'essaye de l'employer pour implémenter un cadre de Monte Carlo pour un modèle qui prend ~ 15 pour fonctionner. L'idée est d'exécuter N moteurs (via SLURM) et avoir un processus "maître" qui met en file d'attente toutes les tâches requises de façon asynchrone et occupée-attend l'achèvement, mettant à jour un db sqlite avec le statut de chaque exécution.

Je voudrais savoir quand une tâche a été assignée à un moteur afin que je puisse suivre son statut dans ma base de données. J'ai essayé d'utiliser les instances AsyncResult pour obtenir le msg_id et interroger la base de données de tâches, mais le champ "démarré" n'est pas mis à jour tant que la tâche n'est pas terminée.

Il semble qu'il devrait y avoir un moyen de recevoir cette notification, ou au moins d'interroger le concentrateur pendant que le moteur fonctionne.

De même, dois-je faire quelque chose pour éviter le timeout du battement du moteur pendant une tâche de longue durée? Est-ce le but de client.spin_thread()?

Merci!

Answer 1

J'ai répondu à une partie de ma propre question en utilisant publish_data. L'idée est que, au lieu de simplement appeler ma fonction "worker" principale dans chaque moteur, j'appelle publish_data() avant et après la fonction de travail principale pour définir un statut que le client peut voir. Par exemple:

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 

La tâche « maître » appelle:

ar = client.map_async(wrapper, listOfArgDicts) 

Je boucle puis sur ar jusqu'à ce que tous AsyncResults soient complets, en examinant ar.data lire les données publiées pour identifier les essais en cours d'exécution et l'enregistrement des résultats d'essais à une base de données sqlite3.

Cette approche générale fonctionne pour un cas de test simple. Je n'ai pas encore exploré la question du délai d'attente pour les appels de fonction de longue durée.