common/ompio: implement pipelined file_iwrite and iread

Pipelined iread/iwrite operations require the notion of subrequests, i.e. a user level request
can contain multiple internal subrequests that all have to complete before the user level operation
is considered finished. This requires adjustments
to the internal ompio progress engine and data structures.

Note: this is purely just a pipelined algorithm, no overlap between different iterations.

Signed-off-by: Edgar Gabriel <edgar.gabriel@amd.com>

Author: edgargabriel

ompi/mca/common/ompio/common_ompio.h

@@ -14,6 +14,7 @@
  * Copyright (c) 2018      Research Organization for Information Science
  *                         and Technology (RIST). All rights reserved.
  * Copyright (c) 2018      DataDirect Networks. All rights reserved.
+ * Copyright (c) 2022      Advanced Micro Devices, Inc. All rights reserved.
  * $COPYRIGHT$
  *
  * Additional copyrights may follow
@@ -101,6 +102,8 @@
 #define OMPIO_PERM_NULL               -1
 #define OMPIO_IOVEC_INITIAL_SIZE      100
 
+extern opal_mutex_t mca_common_ompio_mutex;
+
 enum ompio_fs_type
 {
     NONE = 0,
@@ -274,7 +277,7 @@ OMPI_DECLSPEC int mca_common_ompio_file_iwrite_at_all (ompio_file_t *fp, OMPI_MP
 
 OMPI_DECLSPEC int mca_common_ompio_build_io_array ( ompio_file_t *fh, int index, int cycles,
                                                     size_t bytes_per_cycle, size_t max_data, uint32_t iov_count,
-                                                    struct iovec *decoded_iov, int *ii, int *jj, size_t *tbw,
+                                                    struct iovec *decoded_iov, int *ii, size_t *tbw,
                                                     size_t *spc, mca_common_ompio_io_array_t **io_array,
                                                     int *num_io_entries );
 

ompi/mca/common/ompio/common_ompio_buffer.h

@@ -31,14 +31,10 @@
         opal_output(1, "common_ompio: error allocating memory\n");      \
         return OMPI_ERR_OUT_OF_RESOURCE;                                \
     }                                                                   \
-    _decoded_iov = (struct iovec *) malloc ( sizeof ( struct iovec ));  \
-    if ( NULL == _decoded_iov ) {                                       \
-        opal_output(1, "common_ompio: could not allocate memory.\n");   \
-        return OMPI_ERR_OUT_OF_RESOURCE;                                \
-    }                                                                   \
-    _decoded_iov->iov_base = _tbuf;                                     \
-    _decoded_iov->iov_len  = _max_data;                                 \
-    _iov_count=1;}
+    if (NULL != _decoded_iov) {                                         \
+        ((struct iovec*)_decoded_iov)->iov_base = _tbuf;                \
+        ((struct iovec*)_decoded_iov)->iov_len  = _max_data;            \
+        _iov_count=1;}}
 
 #define OMPIO_PREPARE_READ_BUF(_fh,_buf,_count,_datatype,_tbuf,_convertor,_max_data,_tmp_buf_size,_decoded_iov,_iov_count){ \
         OBJ_CONSTRUCT( _convertor, opal_convertor_t);                                    \
@@ -49,14 +45,10 @@
         opal_output(1, "common_ompio: error allocating memory\n");      \
         return OMPI_ERR_OUT_OF_RESOURCE;                                \
     }                                                                   \
-    _decoded_iov = (struct iovec *) malloc ( sizeof ( struct iovec ));  \
-    if ( NULL == _decoded_iov ) {                                       \
-        opal_output(1, "common_ompio: could not allocate memory.\n");   \
-        return OMPI_ERR_OUT_OF_RESOURCE;                                \
-    }                                                                   \
-    _decoded_iov->iov_base = _tbuf;                                     \
-    _decoded_iov->iov_len  = _max_data;                                 \
-    _iov_count=1;}
+    if (NULL != _decoded_iov) {                                         \
+        ((struct iovec*)_decoded_iov)->iov_base = _tbuf;                \
+        ((struct iovec*)_decoded_iov)->iov_len  = _max_data;            \
+	_iov_count=1;}}
 
 void mca_common_ompio_check_gpu_buf ( ompio_file_t *fh, const void *buf, 
 				      int *is_gpu, int *is_managed);

ompi/mca/common/ompio/common_ompio_file_open.c

@@ -48,6 +48,12 @@
 static mca_common_ompio_generate_current_file_view_fn_t generate_current_file_view_fn;
 static mca_common_ompio_get_mca_parameter_value_fn_t get_mca_parameter_value_fn;
 
+/*
+ * Global, component-wide OMPIO mutex
+ */
+opal_mutex_t mca_common_ompio_mutex = {{0}};
+
+
 int mca_common_ompio_file_open (ompi_communicator_t *comm,
                                 const char *filename,
                                 int amode,