Support for UnifyFS with MPI_File_sync (#1801)

* Initial implementation for supporting UnifyFS in HDF5 with MPI_File_sync after write

* Committing clang-format changes

* Fix format

* Fix env variable and return value check

* Fix flag retrieve

* Fix issues with getting/setting the flag

* Fix merge conflicts

* Update

* Committing clang-format changes

* Update based on suggestions

* Committing clang-format changes

Co-authored-by: github-actions <41898282+github-actions[bot]@users.noreply.github.com>

1 files changed, 34 insertions, 10 deletions

diff --git a/src/H5FDmpio.c b/src/H5FDmpio.c
index ce32bc7..f442644 100644
--- a/src/H5FDmpio.c
+++ b/src/H5FDmpio.c
@@ -60,16 +60,17 @@ static char H5FD_mpi_native_g[] = "native";
  * driver doesn't bother to keep it updated since it's an expensive operation.
  */
 typedef struct H5FD_mpio_t {
-    H5FD_t   pub;       /* Public stuff, must be first                  */
-    MPI_File f;         /* MPIO file handle                             */
-    MPI_Comm comm;      /* MPI Communicator                             */
-    MPI_Info info;      /* MPI info object                              */
-    int      mpi_rank;  /* This process's rank                          */
-    int      mpi_size;  /* Total number of processes                    */
-    haddr_t  eof;       /* End-of-file marker                           */
-    haddr_t  eoa;       /* End-of-address marker                        */
-    haddr_t  last_eoa;  /* Last known end-of-address marker             */
-    haddr_t  local_eof; /* Local end-of-file address for each process   */
+    H5FD_t   pub;                    /* Public stuff, must be first                  */
+    MPI_File f;                      /* MPIO file handle                             */
+    MPI_Comm comm;                   /* MPI Communicator                             */
+    MPI_Info info;                   /* MPI info object                              */
+    int      mpi_rank;               /* This process's rank                          */
+    int      mpi_size;               /* Total number of processes                    */
+    haddr_t  eof;                    /* End-of-file marker                           */
+    haddr_t  eoa;                    /* End-of-address marker                        */
+    haddr_t  last_eoa;               /* Last known end-of-address marker             */
+    haddr_t  local_eof;              /* Local end-of-file address for each process   */
+    hbool_t  mpi_file_sync_required; /* Whether the ROMIO driver requires MPI_File_sync after write */
 } H5FD_mpio_t;
 
 /* Private Prototypes */
@@ -964,6 +965,10 @@ H5FD__mpio_open(const char *name, unsigned flags, hid_t fapl_id, haddr_t H5_ATTR
     file->mpi_rank = mpi_rank;
     file->mpi_size = mpi_size;
 
+    /* Retrieve the flag indicating whether MPI_File_sync is needed after each write */
+    if (H5_mpio_get_file_sync_required(fh, &file->mpi_file_sync_required) < 0)
+        HGOTO_ERROR(H5E_VFL, H5E_CANTGET, NULL, "unable to get mpi_file_sync_required hint")
+
     /* Only processor p0 will get the filesize and broadcast it. */
     if (mpi_rank == 0) {
         /* If MPI_File_get_size fails, broadcast file size as -1 to signal error */
@@ -1629,6 +1634,12 @@ H5FD__mpio_write(H5FD_t *_file, H5FD_mem_t type, hid_t H5_ATTR_UNUSED dxpl_id, h
             if (MPI_SUCCESS !=
                 (mpi_code = MPI_File_write_at_all(file->f, mpi_off, buf, size_i, buf_type, &mpi_stat)))
                 HMPI_GOTO_ERROR(FAIL, "MPI_File_write_at_all failed", mpi_code)
+
+            /* Do MPI_File_sync when needed by underlying ROMIO driver */
+            if (file->mpi_file_sync_required) {
+                if (MPI_SUCCESS != (mpi_code = MPI_File_sync(file->f)))
+                    HMPI_GOTO_ERROR(FAIL, "MPI_File_sync failed", mpi_code)
+            }
         } /* end if */
         else {
             if (type != H5FD_MEM_DRAW)
@@ -2638,6 +2649,12 @@ H5FD__mpio_write_vector(H5FD_t *_file, hid_t H5_ATTR_UNUSED dxpl_id, uint32_t co
             if (MPI_SUCCESS != (mpi_code = MPI_File_write_at_all(file->f, mpi_off, mpi_bufs_base, size_i,
                                                                  buf_type, &mpi_stat)))
                 HMPI_GOTO_ERROR(FAIL, "MPI_File_write_at_all failed", mpi_code)
+
+            /* Do MPI_File_sync when needed by underlying ROMIO driver */
+            if (file->mpi_file_sync_required) {
+                if (MPI_SUCCESS != (mpi_code = MPI_File_sync(file->f)))
+                    HMPI_GOTO_ERROR(FAIL, "MPI_File_sync failed", mpi_code)
+            }
         } /* end if */
         else if (size_i > 0) {
 #ifdef H5FDmpio_DEBUG
@@ -3020,6 +3037,7 @@ done:
  *                  H5FD_CTL_GET_MPI_COMMUNICATOR_OPCODE
  *                  H5FD_CTL_GET_MPI_RANK_OPCODE
  *                  H5FD_CTL_GET_MPI_SIZE_OPCODE
+ *                  H5FD_CTL_GET_MPI_FILE_SYNC_OPCODE
  *
  *              Note that these opcodes must be supported by all VFDs that
  *              support MPI.
@@ -3063,6 +3081,12 @@ H5FD__mpio_ctl(H5FD_t *_file, uint64_t op_code, uint64_t flags, const void H5_AT
             **((int **)output) = file->mpi_size;
             break;
 
+        case H5FD_CTL_GET_MPI_FILE_SYNC_OPCODE:
+            HDassert(output);
+            HDassert(*output);
+            **((hbool_t **)output) = file->mpi_file_sync_required;
+            break;
+
         default: /* unknown op code */
             if (flags & H5FD_CTL_FAIL_IF_UNKNOWN_FLAG) {