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 <[email protected]>

Author: edgargabriel

ompi/mca/common/ompio/common_ompio.h

@@ -274,7 +274,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_file_read.c

@@ -124,7 +124,6 @@ int mca_common_ompio_file_read_default (ompio_file_t *fh, void *buf,
     size_t spc=0;
     ssize_t ret_code=0;
     int i = 0; /* index into the decoded iovec of the buffer */
-    int j = 0; /* index into the file via iovec */
 
     mca_common_ompio_decode_datatype (fh, datatype, count, buf,
                                       &max_data, fh->f_mem_convertor,
@@ -138,11 +137,10 @@ int mca_common_ompio_file_read_default (ompio_file_t *fh, void *buf,
             cycles, max_data);
 #endif
 
-    j = fh->f_index_in_file_view;
     for (index = 0; index < cycles; index++) {
         mca_common_ompio_build_io_array (fh, index, cycles, bytes_per_cycle,
                                          max_data, iov_count, decoded_iov,
-                                         &i, &j, &total_bytes_read, &spc,
+                                         &i, &total_bytes_read, &spc,
                                          &fh->f_io_array, &fh->f_num_of_io_entries);
         if (fh->f_num_of_io_entries == 0) {
 	    ret_code = 0;
@@ -188,7 +186,6 @@ int mca_common_ompio_file_read_pipelined (ompio_file_t *fh, void *buf,
     size_t spc=0;
     ssize_t ret_code=0;
     int i = 0; /* index into the decoded iovec of the buffer */
-    int j = 0; /* index into the file via iovec */
 
     char *tbuf1=NULL, *tbuf2=NULL;
     char *unpackbuf=NULL, *readbuf=NULL;
@@ -236,7 +233,6 @@ int mca_common_ompio_file_read_pipelined (ompio_file_t *fh, void *buf,
     **    - unpack buffer i
     */
 
-    j = fh->f_index_in_file_view;
     if (can_overlap) {
 	mca_common_ompio_register_progress ();	    
     }
@@ -255,7 +251,7 @@ int mca_common_ompio_file_read_pipelined (ompio_file_t *fh, void *buf,
 
             mca_common_ompio_build_io_array (fh, index, cycles, bytes_per_cycle,
                                              bytes_this_cycle, iov_count,
-                                             decoded_iov, &i, &j, &tbr, &spc,
+                                             decoded_iov, &i, &tbr, &spc,
                                              &fh->f_io_array, &fh->f_num_of_io_entries);
 	    if (fh->f_num_of_io_entries == 0) {
 		ret_code = 0;
@@ -352,53 +348,103 @@ int mca_common_ompio_file_read_at (ompio_file_t *fh,
     return ret;
 }
 
+static void mca_common_ompio_post_next_read_subreq(struct mca_ompio_request_t *req, int index)
+{
+    uint32_t iov_count = 1;
+    size_t pos = 0, spc = 0, tbw = 0;
+    int i = 0;
+    mca_ompio_request_t *ompio_subreq=NULL;
+    size_t bytes_per_cycle = OMPIO_MCA_GET(req->req_fh, pipeline_buffer_size);
+
+    struct iovec *decoded_iov = (struct iovec *) malloc (sizeof (struct iovec));
+    if ( NULL == decoded_iov ) {
+        opal_output(1, "common_ompio: could not allocate memory.\n");
+        return;
+    }
+
+    /* Step 1: finish index-1 unpack operation */
+    if (index - 1 >= 0) {
+        size_t num_bytes = bytes_per_cycle;
+        /**
+         *  should really be 'req_num_subreqs -1 == index -1'
+         *  which is the same as below.
+         */
+        if (req->req_num_subreqs == index) {
+            num_bytes = req->req_max_data - (index-1)* bytes_per_cycle;
+        }
+        decoded_iov->iov_base = req->req_tbuf;
+        decoded_iov->iov_len  = num_bytes;
+        opal_convertor_unpack (&req->req_convertor, decoded_iov, &iov_count, &pos);
+    }
+
+    /* Step 2: post next iread subrequest */
+    if (req->req_num_subreqs == index) {
+        /* all done */
+        free (decoded_iov);
+        return;
+    }
+
+    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, index, req->req_num_subreqs,
+                                     bytes_per_cycle, decoded_iov->iov_len,
+                                     iov_count, decoded_iov,
+                                     &i, &tbw, &spc,
+                                     &req->req_fh->f_io_array,
+                                     &req->req_fh->f_num_of_io_entries);
+
+    mca_common_ompio_request_alloc ( &ompio_subreq, MCA_OMPIO_REQUEST_READ);
+    ompio_subreq->req_parent = req;
+    req->req_fh->f_fbtl->fbtl_ipreadv (req->req_fh, (ompi_request_t *)ompio_subreq);
+
+    free(req->req_fh->f_io_array);
+    req->req_fh->f_io_array = NULL;
+    req->req_fh->f_num_of_io_entries = 0;
+
+    free (decoded_iov);
+}
 
 int mca_common_ompio_file_iread (ompio_file_t *fh,
-			       void *buf,
-			       int count,
-			       struct ompi_datatype_t *datatype,
-			       ompi_request_t **request)
+                               void *buf,
+                               int count,
+                               struct ompi_datatype_t *datatype,
+                               ompi_request_t **request)
 {
     int ret = OMPI_SUCCESS;
     mca_ompio_request_t *ompio_req=NULL;
-    size_t spc=0;
+    struct iovec *decoded_iov = NULL;
 
     if (fh->f_amode & MPI_MODE_WRONLY){
-//      opal_output(10, "Improper use of FILE Mode, Using WRONLY for Read!\n");
         ret = MPI_ERR_ACCESS;
-      return ret;
+        return ret;
     }
 
     mca_common_ompio_request_alloc ( &ompio_req, MCA_OMPIO_REQUEST_READ);
 
-    if ( 0 == count ) {
+    if (0 == count || 0 == fh->f_iov_count) {
         ompio_req->req_ompi.req_status.MPI_ERROR = OMPI_SUCCESS;
         ompio_req->req_ompi.req_status._ucount = 0;
         ompi_request_complete (&ompio_req->req_ompi, false);
         *request = (ompi_request_t *) ompio_req;
-        
+
         return OMPI_SUCCESS;
     }
 
     if ( NULL != fh->f_fbtl->fbtl_ipreadv ) {
         // This fbtl has support for non-blocking operations
-
-        size_t total_bytes_read = 0;       /* total bytes that have been read*/
         uint32_t iov_count = 0;
-        struct iovec *decoded_iov = NULL;
-        
         size_t max_data = 0;
-        int i = 0; /* index into the decoded iovec of the buffer */
-        int j = 0; /* index into the file via iovec */
-        
         bool need_to_copy = false;    
-    
         int is_gpu, is_managed;
-        mca_common_ompio_check_gpu_buf ( fh, buf, &is_gpu, &is_managed);
-        if ( is_gpu && !is_managed ) {
+
+        mca_common_ompio_check_gpu_buf (fh, buf, &is_gpu, &is_managed);
+        if (is_gpu && !is_managed) {
             need_to_copy = true;
         }
 
+        mca_common_ompio_register_progress ();
+
         if ( !( fh->f_flags & OMPIO_DATAREP_NATIVE ) &&
              !(datatype == &ompi_mpi_byte.dt  ||
                datatype == &ompi_mpi_char.dt   )) {
@@ -411,88 +457,61 @@ int mca_common_ompio_file_iread (ompio_file_t *fh,
             need_to_copy = true;
         }         
 
-        if ( need_to_copy ) {
-            char *tbuf=NULL;
+        if (need_to_copy) {
+            size_t pipeline_buf_size = OMPIO_MCA_GET(fh, pipeline_buffer_size);
 
-            OMPIO_PREPARE_READ_BUF(fh, buf, count, datatype, tbuf, &ompio_req->req_convertor,
-                                   max_data, 0, decoded_iov, iov_count);
-            
-            ompio_req->req_tbuf = tbuf;
-            ompio_req->req_size = max_data;
+            OMPIO_PREPARE_READ_BUF(fh, buf, count, datatype, ompio_req->req_tbuf,
+                                   &ompio_req->req_convertor, max_data,
+                                   pipeline_buf_size, decoded_iov, iov_count);
+
+            ompio_req->req_num_subreqs = ceil((double)max_data/pipeline_buf_size);
+            ompio_req->req_size        = pipeline_buf_size;
+            ompio_req->req_max_data    = max_data;
+            ompio_req->req_post_next_subreq = mca_common_ompio_post_next_read_subreq;
+            ompio_req->req_fh          = fh;
+            ompio_req->req_ompi.req_status.MPI_ERROR = MPI_SUCCESS;
+
+            mca_common_ompio_post_next_read_subreq (ompio_req, 0);
         }
         else {
-            mca_common_ompio_decode_datatype (fh,
-                                              datatype,
-                                              count,
-                                              buf,
-                                              &max_data,
-                                              fh->f_mem_convertor,
-                                              &decoded_iov,
-                                              &iov_count);
-        }
-    
-        if ( 0 < max_data && 0 == fh->f_iov_count  ) {
-            ompio_req->req_ompi.req_status.MPI_ERROR = OMPI_SUCCESS;
-            ompio_req->req_ompi.req_status._ucount = 0;
-            ompi_request_complete (&ompio_req->req_ompi, false);
-            *request = (ompi_request_t *) ompio_req;
-            if (NULL != decoded_iov) {
-                free (decoded_iov);
-                decoded_iov = NULL;
-            }
-
-            return OMPI_SUCCESS;
-        }
-
-        // Non-blocking operations have to occur in a single cycle
-        j = fh->f_index_in_file_view;
-        
-        mca_common_ompio_build_io_array ( fh,
-                                          0,         // index
-                                          1,         // no. of cyces
-                                          max_data,  // setting bytes per cycle to match data
-                                          max_data,
-                                          iov_count,
-                                          decoded_iov,
-                                          &i,
-                                          &j,
-                                          &total_bytes_read, 
-                                          &spc,
-                                          &fh->f_io_array,
-                                          &fh->f_num_of_io_entries);
-
-	if (fh->f_num_of_io_entries) {
-	  fh->f_fbtl->fbtl_ipreadv (fh, (ompi_request_t *) ompio_req);
-	}
-
-        mca_common_ompio_register_progress ();
-
-	fh->f_num_of_io_entries = 0;
-	if (NULL != fh->f_io_array) {
-	    free (fh->f_io_array);
-	    fh->f_io_array = NULL;
-	}
+	    int i = 0;
+	    size_t spc = 0, tbr = 0;
+	    mca_common_ompio_decode_datatype (fh, datatype, count, buf,
+                                              &max_data, fh->f_mem_convertor,
+                                              &decoded_iov, &iov_count);
+
+            /**
+             * Non-blocking operations have to occur in a single cycle
+             * If the f_io_array is too long, the fbtl will chunk it up
+             * internally.
+             */
+            mca_common_ompio_build_io_array (fh, 0, 1, max_data, max_data,
+                                             iov_count, decoded_iov, &i,
+                                             &tbr, &spc,
+                                             &fh->f_io_array, &fh->f_num_of_io_entries);
 
-	if (NULL != decoded_iov) {
-	    free (decoded_iov);
-	    decoded_iov = NULL;
-	}
+            fh->f_fbtl->fbtl_ipreadv (fh, (ompi_request_t *) ompio_req);
+        }
     }
     else {
-	// This fbtl does not  support non-blocking operations
-	ompi_status_public_t status;
-	ret = mca_common_ompio_file_read (fh, buf, count, datatype, &status);
+        // This fbtl does not  support non-blocking operations
+        ompi_status_public_t status;
+        ret = mca_common_ompio_file_read (fh, buf, count, datatype, &status);
 
-	ompio_req->req_ompi.req_status.MPI_ERROR = ret;
-	ompio_req->req_ompi.req_status._ucount = status._ucount;
-	ompi_request_complete (&ompio_req->req_ompi, false);
+        ompio_req->req_ompi.req_status.MPI_ERROR = ret;
+        ompio_req->req_ompi.req_status._ucount = status._ucount;
+        ompi_request_complete (&ompio_req->req_ompi, false);
     }
 
+    fh->f_num_of_io_entries = 0;
+    free (fh->f_io_array);
+    fh->f_io_array = NULL;
+    free (decoded_iov);
+
     *request = (ompi_request_t *) ompio_req;
     return ret;
 }
 
-
 int mca_common_ompio_file_iread_at (ompio_file_t *fh,
 				  OMPI_MPI_OFFSET_TYPE offset,
 				  void *buf,