barrière MPI ne bloque pas écriture de fichiers, chasse d'eau et os.fsync

Question

je le code de test qui effectue les opérations suivantes:

Ecrire un message de test à un fichier> Barrière> Lire le message de test> Affirmer égal> Répéter.

from __future__ import print_function 
import os 
from mpi4py import MPI 


comm = MPI.COMM_WORLD 
rank = comm.Get_rank() 
loop = True 


def main(): 
    global loop 
    txt_write = 'buhahaha' 

    with open('test', 'w') as f1: 
     if rank == 0: 
      f1.write(txt_write) 

     f1.flush() 
     os.fsync(f1.fileno()) 

    comm.barrier() 

    with open('test') as f2: 
     txt_read = f2.read() 

    try: 
     assert txt_read == txt_write 
    except: 
     print("Assertion error", txt_read, "!=", txt_write, 'rank=', rank) 
     loop = False 
    finally: 
     comm.barrier() 
     if rank == 0: 
      os.remove('test') 


if __name__ == '__main__': 
    i = 0 
    while loop: 
     main() 
     if i % 1000 == 0 and rank == 0: 
      print("Iterations:", i) 

     i += 1 

Il fonctionne pour quelques 100 ou 1000 itérations, mais à un moment donné, il lit un fichier vide et l'assertion échoue. D'autres réponses avaient recommandé l'utilisation de flush et os.fsync, mais cela ne semble pas aider - cela rend l'exécution plus lente. Une idée de comment résoudre ce problème?

Answer 1

Peut-être que vous pouvez essayer quelque chose comme ça, à la place:

if rank == 0: 
    with open('test', 'w') as f1: 
    f1.write(txt_write) 
    # as @jschultz410 correctly pointed out, 
    # we remove f1.flush() and f1.close() 

comm.barrier() 

with open('test') as f2: 
    txt_read = f2.read() 

Answer 2

Le code a donné lieu à un race condition où tous les processus s'ouvraient le même fichier en même temps. Merci à @ jschultz410 et @mko pour avoir identifié cette erreur logique.

Ma solution pour le code était d'utiliser un flux de mémoire au lieu d'un vrai fichier. Maintenant, la ouverte, écrire et lire parties du code Devient:

from io import StringIO 

f1 = StringIO() 
if rank == 0: 
    f1.write(txt_write) 

f1.flush() 
comm.barrier() 

txt_read = f1.getvalue()