common/ompio: add 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.

Extract the file view into a separate datastructure. This is required in
the next step since we need to cache file view and position of the file pointer on the request.

Replicate the file view and the file pointer position on the request.
This is required to correctly increment where to read/write data for every
subrequest, and handle the potential situation that the code changes the file
after posting an iread/iwrite but before the operation finishes.

Furthermore, when initiating an iread/iwrite we need to immidiatly move the position
of the handle to the end of the operation, such that subsequent read/write operations
use the correct position to start out with, and don't accidentaly interfere
with the already ongoing non-blocking operation.
Signed-off-by: Edgar Gabriel <[email protected]>

Signed-off-by: Edgar Gabriel <[email protected]>

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-2023 Advanced Micro Devices, Inc. All rights reserved.
  * $COPYRIGHT$
  *
  * Additional copyrights may follow
@@ -140,18 +141,32 @@ typedef int (*mca_common_ompio_generate_current_file_view_fn_t) (struct ompio_fi
    temporary buffer on aggregators from the fcoll modules */
 typedef int (*mca_common_ompio_get_mca_parameter_value_fn_t) ( char *mca_parameter_name, int name_length );
 
-
 struct mca_common_ompio_print_queue;
 
+/* File View parameters and position in file*/
+struct ompio_fview_t {
+    uint32_t               f_flags;
+    OMPI_MPI_OFFSET_TYPE   f_offset; /* byte offset of current position */
+    OMPI_MPI_OFFSET_TYPE   f_disp;   /* file_view displacement */
+    struct iovec          *f_decoded_iov;
+    uint32_t               f_iov_count;
+    size_t                 f_position_in_file_view; /* in bytes */
+    size_t                 f_total_bytes; /* total bytes read/written within 1 Fview*/
+    int                    f_index_in_file_view;
+    ptrdiff_t              f_view_extent;
+    size_t                 f_view_size;
+    size_t                 f_etype_size;
+};
+typedef struct ompio_fview_t ompio_fview_t;
+
 /**
  * Back-end structure for MPI_File
  */
 struct ompio_file_t {
     /* General parameters */
     int                    fd;
     struct ompi_file_t    *f_fh;     /* pointer back to the file_t structure */
-    OMPI_MPI_OFFSET_TYPE   f_offset; /* byte offset of current position */
-    OMPI_MPI_OFFSET_TYPE   f_disp;   /* file_view displacement */
+    uint32_t               f_flags;
     int                    f_rank;
     int                    f_size;
     int                    f_amode;
@@ -163,7 +178,6 @@ struct ompio_file_t {
     opal_convertor_t      *f_mem_convertor;
     opal_convertor_t      *f_file_convertor;
     opal_info_t           *f_info;
-    int32_t                f_flags;
     void                  *f_fs_ptr;
     int                    f_fs_block_size;
     int                    f_atomicity;
@@ -180,24 +194,16 @@ struct ompio_file_t {
     */
     void                  *f_sharedfp_data;
 
-
     /* File View parameters */
-    struct iovec     *f_decoded_iov;
-    uint32_t          f_iov_count;
-    ompi_datatype_t  *f_iov_type;
-    size_t            f_position_in_file_view; /* in bytes */
-    size_t            f_total_bytes; /* total bytes read/written within 1 Fview*/
-    int               f_index_in_file_view;
-    ptrdiff_t         f_view_extent;
-    size_t            f_view_size;
-    ompi_datatype_t  *f_etype;
-    ompi_datatype_t  *f_filetype;
-    ompi_datatype_t  *f_orig_filetype; /* the fileview passed by the user to us */
-    size_t            f_etype_size;
+    struct ompio_fview_t   f_fview;
+    ompi_datatype_t       *f_iov_type;
+    ompi_datatype_t       *f_etype;
+    ompi_datatype_t       *f_filetype;
+    ompi_datatype_t       *f_orig_filetype; /* the fileview passed by the user to us */
 
     /* contains IO requests that needs to be read/written */
     mca_common_ompio_io_array_t *f_io_array;
-    int                      f_num_of_io_entries;
+    int                    f_num_of_io_entries;
 
     /* Hooks for modules to hang things */
     mca_base_component_t *f_fs_component;
@@ -272,9 +278,9 @@ OMPI_DECLSPEC int mca_common_ompio_file_iwrite_all (ompio_file_t *fp, const void
 OMPI_DECLSPEC int mca_common_ompio_file_iwrite_at_all (ompio_file_t *fp, OMPI_MPI_OFFSET_TYPE offset, const void *buf,
                                                        int count, struct ompi_datatype_t *datatype, ompi_request_t **request);
 
-OMPI_DECLSPEC int mca_common_ompio_build_io_array ( ompio_file_t *fh, int index, int cycles,
+OMPI_DECLSPEC int mca_common_ompio_build_io_array ( ompio_fview_t *fview, 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 );
 
@@ -342,6 +348,8 @@ OMPI_DECLSPEC int mca_common_ompio_decode_datatype (struct ompio_file_t *fh,
                                                     struct iovec **iov,
                                                     uint32_t *iov_count);
 
+OMPI_DECLSPEC int mca_common_ompio_fview_duplicate (struct ompio_fview_t *outfv, struct ompio_fview_t *infv);
+
 OMPI_DECLSPEC int mca_common_ompio_set_callbacks(mca_common_ompio_generate_current_file_view_fn_t generate_current_file_view,
                                                  mca_common_ompio_get_mca_parameter_value_fn_t get_mca_parameter_value);
 #endif /* MCA_COMMON_OMPIO_H */

ompi/mca/common/ompio/common_ompio_aggregators.c

@@ -239,13 +239,13 @@ int mca_common_ompio_fview_based_grouping(ompio_file_t *fh,
     OMPI_MPI_OFFSET_TYPE *start_offsets_lens = NULL;
 
     //Store start offset,length and corresponding rank in an array
-    if(NULL == fh->f_decoded_iov){
+    if(NULL == fh->f_fview.f_decoded_iov){
       start_offset_len[0] = 0;
       start_offset_len[1] = 0;
     }
     else{
-       start_offset_len[0] = (OMPI_MPI_OFFSET_TYPE) fh->f_decoded_iov[0].iov_base;
-       start_offset_len[1] = fh->f_decoded_iov[0].iov_len;
+       start_offset_len[0] = (OMPI_MPI_OFFSET_TYPE) fh->f_fview.f_decoded_iov[0].iov_base;
+       start_offset_len[1] = fh->f_fview.f_decoded_iov[0].iov_len;
     }
     start_offset_len[2] = fh->f_rank;
 
@@ -1275,13 +1275,13 @@ int mca_common_ompio_prepare_to_group(ompio_file_t *fh,
     int ret=OMPI_SUCCESS;
 
     //Store start offset and length in an array //also add bytes per process
-    if(NULL == fh->f_decoded_iov){
+    if(NULL == fh->f_fview.f_decoded_iov){
          start_offset_len[0] = 0;
          start_offset_len[1] = 0;
     }
     else{
-         start_offset_len[0] = (OMPI_MPI_OFFSET_TYPE) fh->f_decoded_iov[0].iov_base;
-         start_offset_len[1] = fh->f_decoded_iov[0].iov_len;
+         start_offset_len[0] = (OMPI_MPI_OFFSET_TYPE) fh->f_fview.f_decoded_iov[0].iov_base;
+         start_offset_len[1] = fh->f_fview.f_decoded_iov[0].iov_len;
     }
     start_offset_len[2] = bytes_per_proc;
     start_offsets_lens_tmp = (OMPI_MPI_OFFSET_TYPE* )malloc (3 * fh->f_init_procs_per_group * sizeof(OMPI_MPI_OFFSET_TYPE));

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

@@ -99,8 +99,8 @@ int mca_common_ompio_file_open (ompi_communicator_t *comm,
     ompio_fh->f_info   = info;
 
     /* set some function pointers required for fcoll, fbtls and sharedfp modules*/
-    ompio_fh->f_generate_current_file_view=generate_current_file_view_fn;
-    ompio_fh->f_get_mca_parameter_value=get_mca_parameter_value_fn;
+    ompio_fh->f_generate_current_file_view = generate_current_file_view_fn;
+    ompio_fh->f_get_mca_parameter_value    = get_mca_parameter_value_fn;
 
     ompio_fh->f_filename = filename;
     if (opal_path_is_absolute(filename) ) {
@@ -343,9 +343,9 @@ int mca_common_ompio_file_close (ompio_file_t *ompio_fh)
         ompio_fh->f_procs_in_group = NULL;
     }
 
-    if (NULL != ompio_fh->f_decoded_iov) {
-        free (ompio_fh->f_decoded_iov);
-        ompio_fh->f_decoded_iov = NULL;
+    if (NULL != ompio_fh->f_fview.f_decoded_iov) {
+        free (ompio_fh->f_fview.f_decoded_iov);
+        ompio_fh->f_fview.f_decoded_iov = NULL;
     }
 
     if (NULL != ompio_fh->f_mem_convertor) {
@@ -415,21 +415,21 @@ int mca_common_ompio_file_get_position (ompio_file_t *fh,
 {
     OMPI_MPI_OFFSET_TYPE off;
 
-    if ( 0 == fh->f_view_extent ||
-         0 == fh->f_view_size   ||
-         0 == fh->f_etype_size ) {
+    if ( 0 == fh->f_fview.f_view_extent ||
+         0 == fh->f_fview.f_view_size   ||
+         0 == fh->f_fview.f_etype_size ) {
         /* not sure whether we should raise an error here */
         *offset = 0;
         return OMPI_SUCCESS;
     }
     /* No. of copies of the entire file view */
-    off = (fh->f_offset - fh->f_disp)/fh->f_view_extent;
+    off = (fh->f_fview.f_offset - fh->f_fview.f_disp)/fh->f_fview.f_view_extent;
 
     /* No. of elements per view */
-    off *= (fh->f_view_size / fh->f_etype_size);
+    off *= (fh->f_fview.f_view_size / fh->f_fview.f_etype_size);
 
     /* No of elements used in the current copy of the view */
-    off += fh->f_total_bytes / fh->f_etype_size;
+    off += fh->f_fview.f_total_bytes / fh->f_fview.f_etype_size;
 
     *offset = off;
     return OMPI_SUCCESS;
@@ -445,9 +445,9 @@ int mca_common_ompio_set_file_defaults (ompio_file_t *fh)
        ptrdiff_t d[2], base;
        int i, flag;
 
-       fh->f_io_array = NULL;
-       fh->f_perm = OMPIO_PERM_NULL;
-       fh->f_flags = 0;
+       fh->f_flags         = 0;
+       fh->f_perm          = OMPIO_PERM_NULL;
+       fh->f_io_array      = NULL;
 
        fh->f_bytes_per_agg = OMPIO_MCA_GET(fh, bytes_per_agg);
        opal_info_get (fh->f_info, "cb_buffer_size", &stripe_str, &flag);
@@ -458,43 +458,41 @@ int mca_common_ompio_set_file_defaults (ompio_file_t *fh)
            OBJ_RELEASE(stripe_str);
        }
 
-       fh->f_atomicity = 0;
        fh->f_fs_block_size = 4096;
+       fh->f_atomicity     = 0;
+       fh->f_stripe_size   = 0;
+       fh->f_stripe_count  = 0;
 
-       fh->f_offset = 0;
-       fh->f_disp = 0;
-       fh->f_position_in_file_view = 0;
-       fh->f_index_in_file_view = 0;
-       fh->f_total_bytes = 0;
+       /* File View */
+       fh->f_fview.f_flags                 = 0;
+       fh->f_fview.f_offset                = 0;
+       fh->f_fview.f_disp                  = 0;
+       fh->f_fview.f_position_in_file_view = 0;
+       fh->f_fview.f_index_in_file_view    = 0;
+       fh->f_fview.f_total_bytes           = 0;
+       fh->f_fview.f_decoded_iov           = NULL;
+
+       fh->f_iov_type      = MPI_DATATYPE_NULL;
+       fh->f_etype         = MPI_DATATYPE_NULL;
+       fh->f_filetype      = MPI_DATATYPE_NULL;
+       fh->f_orig_filetype = MPI_DATATYPE_NULL;
 
        fh->f_init_procs_per_group = -1;
-       fh->f_init_procs_in_group = NULL;
-       
-       fh->f_procs_per_group = -1;
-       fh->f_procs_in_group = NULL;
-       
-       fh->f_init_num_aggrs = -1;
-       fh->f_init_aggr_list = NULL;
-       
-       fh->f_num_aggrs = -1;
-       fh->f_aggr_list = NULL;
-       
-       /* Default file View */
-       fh->f_iov_type = MPI_DATATYPE_NULL;
-       fh->f_stripe_size = 0;
-       /*Decoded iovec of the file-view*/
-       fh->f_decoded_iov = NULL;
-       fh->f_etype = MPI_DATATYPE_NULL;
-       fh->f_filetype = MPI_DATATYPE_NULL;
-       fh->f_orig_filetype = MPI_DATATYPE_NULL;
-       fh->f_datarep = NULL;
+       fh->f_init_procs_in_group  = NULL;
+       fh->f_procs_per_group      = -1;
+       fh->f_procs_in_group       = NULL;
+       fh->f_init_num_aggrs       = -1;
+       fh->f_init_aggr_list       = NULL;
+       fh->f_num_aggrs            = -1;
+       fh->f_aggr_list            = NULL;
+       fh->f_datarep              = NULL;
 
        /*Create a derived datatype for the created iovec */
        types[0] = &ompi_mpi_long.dt;
        types[1] = &ompi_mpi_long.dt;
 
-       d[0] = (ptrdiff_t) fh->f_decoded_iov;
-       d[1] = (ptrdiff_t) &fh->f_decoded_iov[0].iov_len;
+       d[0] = (ptrdiff_t) fh->f_fview.f_decoded_iov;
+       d[1] = (ptrdiff_t) &fh->f_fview.f_decoded_iov[0].iov_len;
 
        base = d[0];
        for (i=0 ; i<2 ; i++) {