ROM-IO thread safety

ompi/mca/io/romio321/romio/adio/include/mpipr.h

@@ -10,28 +10,30 @@
 
 #ifndef USE_MPI_VERSIONS
 
+#include "mpi-io/glue/openmpi/mpio_wrappers.h"
+
 #undef MPI_Abort
 #define MPI_Abort PMPI_Abort
 #undef MPI_Allgather
-#define MPI_Allgather PMPI_Allgather
+#define MPI_Allgather MPIW_Allgather
 #undef MPI_Allgatherv
 #define MPI_Allgatherv PMPI_Allgatherv
 #undef MPI_Allreduce
-#define MPI_Allreduce PMPI_Allreduce
+#define MPI_Allreduce MPIW_Allreduce
 #undef MPI_Alltoall
-#define MPI_Alltoall PMPI_Alltoall
+#define MPI_Alltoall MPIW_Alltoall
 #undef MPI_Alltoallv
 #define MPI_Alltoallv PMPI_Alltoallv
 #undef MPI_Alltoallw
-#define MPI_Alltoallw PMPI_Alltoallw
+#define MPI_Alltoallw MPIW_Alltoallw
 #undef MPI_Attr_delete
 #define MPI_Attr_delete PMPI_Attr_delete
 #undef MPI_Attr_get
 #define MPI_Attr_get PMPI_Attr_get
 #undef MPI_Barrier
-#define MPI_Barrier PMPI_Barrier
+#define MPI_Barrier MPIW_Barrier
 #undef MPI_Bcast
-#define MPI_Bcast PMPI_Bcast
+#define MPI_Bcast MPIW_Bcast
 #undef MPI_Bsend
 #define MPI_Bsend PMPI_Bsend
 #undef MPI_Bsend_init
@@ -105,9 +107,9 @@
 #undef MPI_Finalize
 #define MPI_Finalize PMPI_Finalize
 #undef MPI_Gather
-#define MPI_Gather PMPI_Gather
+#define MPI_Gather MPIW_Gather
 #undef MPI_Gatherv
-#define MPI_Gatherv PMPI_Gatherv
+#define MPI_Gatherv MPIW_Gatherv
 #undef MPI_Get_address
 #define MPI_Get_address PMPI_Get_address
 #undef MPI_Get_count
@@ -191,7 +193,7 @@
 #undef MPI_Probe
 #define MPI_Probe PMPI_Probe
 #undef MPI_Recv
-#define MPI_Recv PMPI_Recv
+#define MPI_Recv MPIW_Recv
 #undef MPI_Recv_init
 #define MPI_Recv_init PMPI_Recv_init
 #undef MPI_Reduce
@@ -213,7 +215,7 @@
 #undef MPI_Scatterv
 #define MPI_Scatterv PMPI_Scatterv
 #undef MPI_Send
-#define MPI_Send PMPI_Send
+#define MPI_Send MPIW_Send
 #undef MPI_Send_init
 #define MPI_Send_init PMPI_Send_init
 #undef MPI_Sendrecv
@@ -281,11 +283,11 @@
 #undef MPI_Unpack
 #define MPI_Unpack PMPI_Unpack
 #undef MPI_Wait
-#define MPI_Wait PMPI_Wait
+#define MPI_Wait MPIW_Wait
 #undef MPI_Waitall
-#define MPI_Waitall PMPI_Waitall
+#define MPI_Waitall MPIW_Waitall
 #undef MPI_Waitany
-#define MPI_Waitany PMPI_Waitany
+#define MPI_Waitany MPIW_Waitany
 #undef MPI_Waitsome
 #define MPI_Waitsome PMPI_Waitsome
 #undef MPI_Wtick

ompi/mca/io/romio321/romio/mpi-io/glue/openmpi/Makefile.mk

@@ -8,6 +8,10 @@
 if MPIO_GLUE_OPENMPI
 romio_other_sources +=              \
     mpi-io/glue/openmpi/mpio_file.c \
-    mpi-io/glue/openmpi/mpio_err.c
+    mpi-io/glue/openmpi/mpio_err.c \
+    mpi-io/glue/openmpi/mpio_wrappers.c
+
+noinst_HEADERS += mpi-io/glue/openmpi/mpio_wrappers.h
+
 endif MPIO_GLUE_OPENMPI
 

ompi/mca/io/romio321/romio/mpi-io/glue/openmpi/mpio_wrappers.c

@@ -0,0 +1,147 @@
+#include "mpi-io/mpioimpl.h"
+
+#include "mpi-io/glue/openmpi/mpio_wrappers.h"
+
+int MPIW_Allreduce(const void *sendbuf, void *recvbuf, int count,
+                   MPI_Datatype datatype, MPI_Op op, MPI_Comm comm)
+{
+    int res;
+    ROMIO_THREAD_CS_EXIT();
+    res = PMPI_Allreduce(sendbuf, recvbuf, count, datatype, op, comm);
+    ROMIO_THREAD_CS_ENTER();
+    return res;
+}
+
+int MPIW_Allgather(const void *sendbuf, int sendcount, MPI_Datatype sendtype,
+                   void *recvbuf, int recvcount, MPI_Datatype recvtype,
+                   MPI_Comm comm)
+{
+    int res;
+    ROMIO_THREAD_CS_EXIT();
+    res = PMPI_Allgather(sendbuf, sendcount, sendtype,
+                         recvbuf, recvcount, recvtype,
+                         comm);
+    ROMIO_THREAD_CS_ENTER();
+    return res;
+}
+
+int MPIW_Alltoall(const void *sendbuf, int sendcount, MPI_Datatype sendtype,
+                  void *recvbuf, int recvcount, MPI_Datatype recvtype,
+                  MPI_Comm comm)
+{
+    int res;
+    ROMIO_THREAD_CS_EXIT();
+    res = PMPI_Alltoall(sendbuf, sendcount, sendtype,
+                        recvbuf, recvcount, recvtype,
+                        comm);
+    ROMIO_THREAD_CS_ENTER();
+    return res;
+}
+
+int MPIW_Alltoallw(const void *sendbuf, const int sendcounts[], const int sdispls[], const MPI_Datatype sendtypes[],
+                   void *recvbuf, const int recvcounts[], const int rdispls[], const MPI_Datatype recvtypes[],
+                   MPI_Comm comm)
+{
+    int res;
+    ROMIO_THREAD_CS_EXIT();
+    res = PMPI_Alltoallw(sendbuf, sendcounts, sdispls, sendtypes,
+                        recvbuf, recvcounts, rdispls, recvtypes,
+                        comm);
+    ROMIO_THREAD_CS_ENTER();
+    return res;
+}
+
+int MPIW_Barrier(MPI_Comm comm)
+{
+    int res;
+    ROMIO_THREAD_CS_EXIT();
+    res = PMPI_Barrier(comm);
+    ROMIO_THREAD_CS_ENTER();
+    return res;
+}
+
+int MPIW_Bcast(void *buffer, int count, MPI_Datatype datatype,
+               int root, MPI_Comm comm)
+{
+    int res;
+    ROMIO_THREAD_CS_EXIT();
+    res = PMPI_Bcast(buffer, count, datatype, root, comm);
+    ROMIO_THREAD_CS_ENTER();
+    return res;
+}
+
+int MPIW_Gather(const void *sendbuf, int sendcount, MPI_Datatype sendtype,
+                void *recvbuf, int recvcount, MPI_Datatype recvtype,
+                int root, MPI_Comm comm)
+{
+    int res;
+    ROMIO_THREAD_CS_EXIT();
+    res = PMPI_Gather(sendbuf, sendcount, sendtype,
+                      recvbuf, recvcount, recvtype,
+                      root, comm);
+    ROMIO_THREAD_CS_ENTER();
+    return res;
+}
+
+int MPIW_Gatherv(const void *sendbuf, int sendcount, MPI_Datatype sendtype,
+                 void *recvbuf, const int recvcounts[], const int displs[],
+                 MPI_Datatype recvtype, int root, MPI_Comm comm)
+{
+    int res;
+    ROMIO_THREAD_CS_EXIT();
+    res = PMPI_Gatherv(sendbuf, sendcount, sendtype,
+                       recvbuf, recvcounts, displs, recvtype,
+                      root, comm);
+    ROMIO_THREAD_CS_ENTER();
+    return res;
+}
+
+int MPIW_Recv(void *buf, int count, MPI_Datatype datatype, int source,
+              int tag, MPI_Comm comm, MPI_Status *status)
+{
+    int res;
+    ROMIO_THREAD_CS_EXIT();
+    res = PMPI_Recv(buf, count, datatype, source, tag, comm, status);
+    ROMIO_THREAD_CS_ENTER();
+    return res;
+}
+
+
+int MPIW_Send(const void *buf, int count, MPI_Datatype datatype, int dest,
+              int tag, MPI_Comm comm)
+{
+    int res;
+    ROMIO_THREAD_CS_EXIT();
+    res = PMPI_Send(buf, count, datatype, dest, tag, comm);
+    ROMIO_THREAD_CS_ENTER();
+    return res;
+}
+
+int MPIW_Waitall(int count, MPI_Request array_of_requests[],
+                 MPI_Status array_of_statuses[])
+{
+    int res;
+    ROMIO_THREAD_CS_EXIT();
+    res = PMPI_Waitall(count, array_of_requests, array_of_statuses);
+    ROMIO_THREAD_CS_ENTER();
+    return res;
+}
+
+int MPIW_Waitany(int count, MPI_Request array_of_requests[],
+                 int *index, MPI_Status *status)
+{
+    int res;
+    ROMIO_THREAD_CS_EXIT();
+    res = PMPI_Waitany(count, array_of_requests, index, status);
+    ROMIO_THREAD_CS_ENTER();
+    return res;
+}
+
+int MPIW_Wait(MPI_Request *request, MPI_Status *status)
+{
+    int res;
+    ROMIO_THREAD_CS_EXIT();
+    res = PMPI_Wait(request, status);
+    ROMIO_THREAD_CS_ENTER();
+    return res;
+}

ompi/mca/io/romio321/romio/mpi-io/glue/openmpi/mpio_wrappers.h

@@ -0,0 +1,43 @@
+#include <mpi.h>
+
+int MPIW_Allreduce(const void *sendbuf, void *recvbuf, int count,
+                   MPI_Datatype datatype, MPI_Op op, MPI_Comm comm);
+
+int MPIW_Allgather(const void *sendbuf, int sendcount, MPI_Datatype sendtype,
+                   void *recvbuf, int recvcount, MPI_Datatype recvtype,
+                   MPI_Comm comm);
+
+int MPIW_Alltoall(const void *sendbuf, int sendcount, MPI_Datatype sendtype,
+                  void *recvbuf, int recvcount, MPI_Datatype recvtype,
+                   MPI_Comm comm);
+
+int MPIW_Alltoallw(const void *sendbuf, const int sendcounts[], const int sdispls[], const MPI_Datatype sendtypes[],
+                   void *recvbuf, const int recvcounts[], const int rdispls[], const MPI_Datatype recvtypes[],
+                   MPI_Comm comm);
+
+int MPIW_Barrier(MPI_Comm comm);
+
+int MPIW_Bcast(void *buffer, int count, MPI_Datatype datatype,
+               int root, MPI_Comm comm);
+
+int MPIW_Gather(const void *sendbuf, int sendcount, MPI_Datatype sendtype,
+                void *recvbuf, int recvcount, MPI_Datatype recvtype,
+                int root, MPI_Comm comm);
+
+int MPIW_Gatherv(const void *sendbuf, int sendcount, MPI_Datatype sendtype,
+                 void *recvbuf, const int recvcounts[], const int displs[],
+                 MPI_Datatype recvtype, int root, MPI_Comm comm);
+
+int MPIW_Recv(void *buf, int count, MPI_Datatype datatype, int source,
+              int tag, MPI_Comm comm, MPI_Status *status);
+
+int MPIW_Send(const void *buf, int count, MPI_Datatype datatype, int dest,
+              int tag, MPI_Comm comm);
+
+int MPIW_Waitall(int count, MPI_Request array_of_requests[],
+                 MPI_Status array_of_statuses[]);
+
+int MPIW_Waitany(int count, MPI_Request array_of_requests[],
+                 int *index, MPI_Status *status);
+
+int MPIW_Wait(MPI_Request *request, MPI_Status *status);

ompi/mca/io/romio321/romio/mpi-io/mpioimpl.h

@@ -35,8 +35,10 @@
    error reporting features provided by MPICH must implement these 
    four functions.  Defining these as empty should not change the behavior 
    of correct programs */
-#define ROMIO_THREAD_CS_ENTER()
-#define ROMIO_THREAD_CS_EXIT()
+extern void mca_io_romio321_lock(void);
+extern void mca_io_romio321_unlock(void);
+#define ROMIO_THREAD_CS_ENTER() mca_io_romio321_lock()
+#define ROMIO_THREAD_CS_EXIT() mca_io_romio321_unlock()
 #define ROMIO_THREAD_CS_YIELD()
 /* The MPI_DATATYPE_ISCOMMITTED macro now always sets err_=0.
    This is an optimistic approach for Open MPI, but it is likely other

ompi/mca/io/romio321/src/io_romio321.h

@@ -293,6 +293,10 @@ int mca_io_romio321_file_get_atomicity (struct ompi_file_t *fh,
                                      int *flag);
 int mca_io_romio321_file_sync (struct ompi_file_t *fh);
 
+void mca_io_romio321_lock(void);
+
+void mca_io_romio321_unlock(void);
+
 /* End Prototypes */
 
 END_C_DECLS

ompi/mca/io/romio321/src/io_romio321_component.c

@@ -13,8 +13,8 @@
  * Copyright (c) 2008-2015 Cisco Systems, Inc.  All rights reserved.
  * Copyright (c) 2015      Los Alamos National Security, LLC. All rights
  *                         reserved.
- * Copyright (c) 2015-2017 Research Organization for Information Science
- *                         and Technology (RIST). All rights reserved.
+ * Copyright (c) 2015-2019 Research Organization for Information Science
+ *                         and Technology (RIST).  All rights reserved.
  * Copyright (c) 2016-2017 IBM Corporation. All rights reserved.
  * $COPYRIGHT$
  *
@@ -249,9 +249,7 @@ static int delete_select(const char *filename, struct opal_info_t *info,
     opal_info_t *opal_info = &(ompi_info->super);
     opal_info_dup (info, &opal_info);
 
-    OPAL_THREAD_LOCK (&mca_io_romio321_mutex);
     ret = ROMIO_PREFIX(MPI_File_delete)(filename, ompi_info);
-    OPAL_THREAD_UNLOCK (&mca_io_romio321_mutex);
 
     ompi_info_free(&ompi_info);
     return ret;
@@ -267,10 +265,18 @@ register_datarep(const char * datarep,
 {
     int ret;
 
-    OPAL_THREAD_LOCK(&mca_io_romio321_mutex);
     ret = ROMIO_PREFIX(MPI_Register_datarep(datarep, read_fn, write_fn,
                                             extent_fn, state));
-    OPAL_THREAD_UNLOCK(&mca_io_romio321_mutex);
 
     return ret;
 }
+
+void mca_io_romio321_lock()
+{
+    OPAL_THREAD_LOCK(&mca_io_romio321_mutex);
+}
+
+void mca_io_romio321_unlock()
+{
+    OPAL_THREAD_UNLOCK(&mca_io_romio321_mutex);
+}