common/ompio: replicate fview on request

edgargabriel · edgargabriel · commit 2d4b2803f798 · 2023-02-06T19:51:30.000Z
need to 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 &lt;edgar.gabriel@amd.com&gt;
diff --git a/ompi/mca/common/ompio/common_ompio.h b/ompi/mca/common/ompio/common_ompio.h
@@ -350,6 +350,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 */
diff --git a/ompi/mca/common/ompio/common_ompio_file_read.c b/ompi/mca/common/ompio/common_ompio_file_read.c
@@ -379,7 +379,7 @@ static void mca_common_ompio_post_next_read_subreq(struct mca_ompio_request_t *r
     decoded_iov.iov_base = req->req_tbuf;
     decoded_iov.iov_len  = (req->req_num_subreqs-1 == index) ?
 	req->req_max_data - (index* bytes_per_cycle) : req->req_size;
-    mca_common_ompio_build_io_array (&(req->req_fh->f_fview), index, req->req_num_subreqs,
+    mca_common_ompio_build_io_array (req->req_fview, index, req->req_num_subreqs,
                                      bytes_per_cycle, decoded_iov.iov_len,
                                      iov_count, &decoded_iov,
                                      &i, &tbw, &spc,
@@ -454,6 +454,7 @@ int mca_common_ompio_file_iread (ompio_file_t *fh,
         }         
         
         if (need_to_copy) {
+            OMPI_MPI_OFFSET_TYPE prev_offset;
             size_t pipeline_buf_size = OMPIO_MCA_GET(fh, pipeline_buffer_size);
             
             OMPIO_PREPARE_READ_BUF(fh, buf, count, datatype, ompio_req->req_tbuf,
@@ -467,7 +468,26 @@ int mca_common_ompio_file_iread (ompio_file_t *fh,
             ompio_req->req_fh          = fh;
             ompio_req->req_ompi.req_status.MPI_ERROR = MPI_SUCCESS;
 
+	    ompio_req->req_fview = (struct ompio_fview_t *) malloc(sizeof(struct ompio_fview_t));
+	    if (NULL == ompio_req->req_fview) {
+		opal_output(1, "common_ompio: error allocating memory\n");
+		return OMPI_ERR_OUT_OF_RESOURCE;
+	    }
+            ret = mca_common_ompio_fview_duplicate(ompio_req->req_fview, &fh->f_fview);
+            if (OMPI_SUCCESS != ret) {
+                return ret;
+            }
+            mca_common_ompio_file_get_position (fh, &prev_offset );
             mca_common_ompio_post_next_read_subreq (ompio_req, 0);
+
+            /* Move file pointer to the end of the operation.
+             * Otherwise posting another I/O operation will start of
+             * from the wrong file position. The request will update
+             * the position where to write the next chunk of data
+             * using its internal copy of the file view and file pointer
+             * position.
+             */
+            mca_common_ompio_set_explicit_offset (fh, prev_offset+max_data);
         }
         else {
 	    int i = 0;
diff --git a/ompi/mca/common/ompio/common_ompio_file_view.c b/ompi/mca/common/ompio/common_ompio_file_view.c
@@ -13,6 +13,7 @@
  * Copyright (c) 2017-2018 Research Organization for Information Science
  *                         and Technology (RIST). All rights reserved.
  * Copyright (c) 2017      IBM Corporation. All rights reserved.
+ * Copyright (c) 2023      Advanced Micro Devices, Inc. All rights reserved.
  * $COPYRIGHT$
  *
  * Additional copyrights may follow
@@ -414,4 +415,32 @@ OMPI_MPI_OFFSET_TYPE get_contiguous_chunk_size (ompio_file_t *fh, int flag)
     return global_avg[0];
 }
 
+int mca_common_ompio_fview_duplicate (struct ompio_fview_t *outfv, struct ompio_fview_t *infv)
+{
+    uint32_t i;
+
+    memset(outfv, 0, sizeof(struct ompio_fview_t));
+    outfv->f_flags     = infv->f_flags;
+    outfv->f_offset    = infv->f_offset;
+    outfv->f_disp      = infv->f_disp;
+    outfv->f_iov_count = infv->f_iov_count;
 
+    outfv->f_decoded_iov = (struct iovec*) malloc (outfv->f_iov_count * sizeof(struct iovec));
+    if (NULL == outfv->f_decoded_iov) {
+        opal_output(1, "common_ompio_duplicate_fview: could not allocate memory\n");
+        return OMPI_ERR_OUT_OF_RESOURCE;
+    }
+    for (i=0; i < outfv->f_iov_count; i++) {
+        outfv->f_decoded_iov[i].iov_base = infv->f_decoded_iov[i].iov_base;
+        outfv->f_decoded_iov[i].iov_len  = infv->f_decoded_iov[i].iov_len ;
+    }
+
+    outfv->f_position_in_file_view = infv->f_position_in_file_view;
+    outfv->f_total_bytes           = infv->f_total_bytes;
+    outfv->f_index_in_file_view    = infv->f_index_in_file_view;
+    outfv->f_view_extent           = infv->f_view_extent;
+    outfv->f_view_size             = infv->f_view_size;
+    outfv->f_etype_size            = infv->f_etype_size;
+
+    return OMPI_SUCCESS;
+}
diff --git a/ompi/mca/common/ompio/common_ompio_file_write.c b/ompi/mca/common/ompio/common_ompio_file_write.c
@@ -12,7 +12,7 @@
  * Copyright (c) 2008-2019 University of Houston. All rights reserved.
  * Copyright (c) 2015-2018 Research Organization for Information Science
  *                         and Technology (RIST). All rights reserved.
- * Copyright (c) 2022      Advanced Micro Devices, Inc. All rights reserved.
+ * Copyright (c) 2022-2023 Advanced Micro Devices, Inc. All rights reserved.
  * $COPYRIGHT$
  *
  * Additional copyrights may follow
@@ -327,7 +327,7 @@ static void mca_common_ompio_post_next_write_subreq(struct mca_ompio_request_t *
     decoded_iov.iov_base = req->req_tbuf;
     decoded_iov.iov_len  = req->req_size;
     opal_convertor_pack (&req->req_convertor, &decoded_iov, &iov_count, &pos);
-    mca_common_ompio_build_io_array (&(req->req_fh->f_fview), index, req->req_num_subreqs,
+    mca_common_ompio_build_io_array (req->req_fview, index, req->req_num_subreqs,
                                      bytes_per_cycle, pos,
                                      iov_count, &decoded_iov,
                                      &i, &tbw, &spc,
@@ -398,7 +398,9 @@ int mca_common_ompio_file_iwrite (ompio_file_t *fh,
         }         
         
         if (need_to_copy) {
+            OMPI_MPI_OFFSET_TYPE prev_offset;
             size_t pipeline_buf_size = OMPIO_MCA_GET(fh, pipeline_buffer_size);
+
             OMPIO_PREPARE_BUF (fh, buf, count, datatype, ompio_req->req_tbuf,
                                &ompio_req->req_convertor, max_data,
                                pipeline_buf_size, NULL, iov_count);
@@ -410,7 +412,26 @@ int mca_common_ompio_file_iwrite (ompio_file_t *fh,
             ompio_req->req_fh          = fh;
             ompio_req->req_ompi.req_status.MPI_ERROR = MPI_SUCCESS;
 
+	    ompio_req->req_fview = (struct ompio_fview_t *) malloc(sizeof(struct ompio_fview_t));
+	    if (NULL == ompio_req->req_fview) {
+		opal_output(1, "common_ompio: error allocating memory\n");
+		return OMPI_ERR_OUT_OF_RESOURCE;
+	    }
+            ret = mca_common_ompio_fview_duplicate(ompio_req->req_fview, &fh->f_fview);
+            if (OMPI_SUCCESS != ret) {
+                return ret;
+            }
+            mca_common_ompio_file_get_position (fh, &prev_offset );
             mca_common_ompio_post_next_write_subreq (ompio_req, 0);
+
+            /* Move file pointer to the end of the operation.
+             * Otherwise posting another I/O operation will start of
+             * from the wrong file position. The request will update
+             * the position where to write the next chunk of data
+             * using its internal copy of the file view and file pointer
+             * position.
+             */
+            mca_common_ompio_set_explicit_offset (fh, prev_offset+max_data);
         }
         else {
             mca_common_ompio_decode_datatype (fh, datatype, count,
diff --git a/ompi/mca/common/ompio/common_ompio_request.c b/ompi/mca/common/ompio/common_ompio_request.c
@@ -82,6 +82,7 @@ void mca_common_ompio_request_construct(mca_ompio_request_t* req)
     req->req_num_subreqs       = 0;
     req->req_subreqs_completed = 0;
     req->req_fh                = NULL;
+    req->req_fview             = NULL;
     req->req_post_followup     = false;
 
     OBJ_CONSTRUCT(&req->req_item, opal_list_item_t);
@@ -96,7 +97,10 @@ void mca_common_ompio_request_destruct(mca_ompio_request_t* req)
     if ( NULL != req->req_data ) {
         free (req->req_data);
     }
-
+    if (NULL != req->req_fview) {
+	free (req->req_fview->f_decoded_iov);
+	free (req->req_fview);
+    }
     return;
 }
 
diff --git a/ompi/mca/common/ompio/common_ompio_request.h b/ompi/mca/common/ompio/common_ompio_request.h
@@ -66,6 +66,7 @@ struct mca_ompio_request_t {
     int                                     req_num_subreqs;
     int                               req_subreqs_completed;
     ompio_file_t                                    *req_fh;
+    ompio_fview_t                                *req_fview;
     bool                                  req_post_followup;
 };
 typedef struct mca_ompio_request_t mca_ompio_request_t;
