Merge pull request open-mpi#8559 from edgargabriel/topic/atomicity

Topic/atomicity

Author: edgargabriel

ompi/mca/fbtl/base/base.h

@@ -10,7 +10,7 @@
  * Copyright (c) 2004-2005 The Regents of the University of California.
  *                         All rights reserved.
  * Copyright (c) 2008      Cisco Systems, Inc.  All rights reserved.
- * Copyright (c) 2008-2011 University of Houston. All rights reserved.
+ * Copyright (c) 2008-2021 University of Houston. All rights reserved.
  * Copyright (c) 2018      Research Organization for Information Science
  *                         and Technology (RIST). All rights reserved.
  * $COPYRIGHT$
@@ -50,6 +50,8 @@ OMPI_DECLSPEC int mca_fbtl_base_find_available(bool enable_progress_threads,
 OMPI_DECLSPEC int mca_fbtl_base_init_file (struct ompio_file_t *file);
 
 OMPI_DECLSPEC int mca_fbtl_base_get_param (struct ompio_file_t *file, int keyval);
+
+OMPI_DECLSPEC bool mca_fbtl_base_check_atomicity (struct ompio_file_t *file);
 /*
  * Globals
  */

ompi/mca/fbtl/base/fbtl_base_file_select.c

@@ -9,7 +9,7 @@
  *                         University of Stuttgart.  All rights reserved.
  * Copyright (c) 2004-2005 The Regents of the University of California.
  *                         All rights reserved.
- * Copyright (c) 2008-2011 University of Houston. All rights reserved.
+ * Copyright (c) 2008-2021 University of Houston. All rights reserved.
  * Copyright (c) 2018      Research Organization for Information Science
  *                         and Technology (RIST). All rights reserved.
  * $COPYRIGHT$
@@ -259,3 +259,13 @@ int mca_fbtl_base_file_select (struct ompio_file_t *file,
 
     return err;
 }
+
+
+bool mca_fbtl_base_check_atomicity (struct ompio_file_t *file)
+{
+    /* by default, return false. An fbtl can overwrite this setting
+    ** if they have support for atomic operations through locks or other
+    ** measures.
+    */
+    return false;
+}

ompi/mca/fbtl/fbtl.h

@@ -10,7 +10,7 @@
  *                         University of Stuttgart.  All rights reserved.
  * Copyright (c) 2004-2005 The Regents of the University of California.
  *                         All rights reserved.
- * Copyright (c) 2008-2014 University of Houston. All rights reserved.
+ * Copyright (c) 2008-2021 University of Houston. All rights reserved.
  * Copyright (c) 2015      Los Alamos National Security, LLC. All rights
  *                         reserved.
  * Copyright (c) 2018      Research Organization for Information Science
@@ -135,6 +135,10 @@ typedef bool (*mca_fbtl_base_module_progress_fn_t)
 
 typedef void (*mca_fbtl_base_module_request_free_fn_t)
     ( struct mca_ompio_request_t *request);
+
+typedef bool (*mca_fbtl_base_module_check_atomicity_fn_t)
+    (struct ompio_file_t *file);
+
 /*
  * ***********************************************************************
  * ***************************  module structure *************************
@@ -150,12 +154,13 @@ struct mca_fbtl_base_module_1_0_0_t {
     mca_fbtl_base_module_finalize_1_0_0_fn_t fbtl_module_finalize;
 
     /* FBTL function pointers */
-    mca_fbtl_base_module_preadv_fn_t        fbtl_preadv;
-    mca_fbtl_base_module_ipreadv_fn_t       fbtl_ipreadv;
-    mca_fbtl_base_module_pwritev_fn_t       fbtl_pwritev;
-    mca_fbtl_base_module_ipwritev_fn_t      fbtl_ipwritev;
-    mca_fbtl_base_module_progress_fn_t      fbtl_progress;
-    mca_fbtl_base_module_request_free_fn_t  fbtl_request_free;
+    mca_fbtl_base_module_preadv_fn_t          fbtl_preadv;
+    mca_fbtl_base_module_ipreadv_fn_t         fbtl_ipreadv;
+    mca_fbtl_base_module_pwritev_fn_t         fbtl_pwritev;
+    mca_fbtl_base_module_ipwritev_fn_t        fbtl_ipwritev;
+    mca_fbtl_base_module_progress_fn_t        fbtl_progress;
+    mca_fbtl_base_module_request_free_fn_t    fbtl_request_free;
+    mca_fbtl_base_module_check_atomicity_fn_t fbtl_check_atomicity;
 };
 typedef struct mca_fbtl_base_module_1_0_0_t mca_fbtl_base_module_1_0_0_t;
 typedef mca_fbtl_base_module_1_0_0_t mca_fbtl_base_module_t;

ompi/mca/fbtl/ime/fbtl_ime.c

@@ -11,6 +11,7 @@
 #include "mpi.h"
 
 #include "ompi/mca/fbtl/fbtl.h"
+#include "ompi/mca/fbtl/base/base.h"
 #include "ompi/mca/fbtl/ime/fbtl_ime.h"
 
 /*
@@ -26,8 +27,8 @@ static mca_fbtl_base_module_1_0_0_t ime =  {
     mca_fbtl_ime_pwritev,         /* blocking write */
     mca_fbtl_ime_ipwritev,        /* non-blocking write */
     mca_fbtl_ime_progress,        /* module specific progress */
-    mca_fbtl_ime_request_free     /* free module specific data items on the request */
-};
+    mca_fbtl_ime_request_free,     /* free module specific data items on the request */
+    mca_fbtl_base_check_atomicity  /* check whether atomicity is supported on this fs */};
 /*
  * *******************************************************************
  * ************************* structure ends **************************
@@ -179,4 +180,4 @@ void mca_fbtl_ime_complete_cb  (struct ime_aiocb *aiocb, int err, ssize_t bytes)
 {
     ssize_t *req_status = (ssize_t *) aiocb->user_context;
     *req_status = err == 0 ? bytes : FBTL_IME_REQ_ERROR;
-}
+}

ompi/mca/fbtl/posix/fbtl_posix.c

@@ -58,12 +58,13 @@ static mca_fbtl_base_module_1_0_0_t posix =  {
 #if defined (FBTL_POSIX_HAVE_AIO)
     mca_fbtl_posix_ipwritev,        /* non-blocking write */
     mca_fbtl_posix_progress,        /* module specific progress */
-    mca_fbtl_posix_request_free     /* free module specific data items on the request */
+    mca_fbtl_posix_request_free,    /* free module specific data items on the request */
 #else
     NULL,                           /* non-blocking write */
     NULL,                           /* module specific progress */
-    NULL                            /* free module specific data items on the request */
+    NULL,                           /* free module specific data items on the request */
 #endif
+    mca_fbtl_posix_check_atomicity  /* check whether atomicity is supported on this fs */
 };
 /*
  * *******************************************************************
@@ -144,34 +145,38 @@ bool mca_fbtl_posix_progress ( mca_ompio_request_t *req)
                     data->aio_req_status[i]        = EINPROGRESS;
                     start_offset = data->aio_reqs[i].aio_offset;
                     total_length = data->aio_reqs[i].aio_nbytes;
+                    /* release previous lock */
+                    mca_fbtl_posix_unlock ( &data->aio_lock, data->aio_fh, &data->aio_lock_counter );
+                    
                     if ( data->aio_req_type == FBTL_POSIX_WRITE ) {
-                        ret_code = mca_fbtl_posix_lock( &data->aio_lock, data->aio_fh, F_WRLCK, start_offset, total_length, OMPIO_LOCK_ENTIRE_REGION );
+                        ret_code = mca_fbtl_posix_lock( &data->aio_lock, data->aio_fh, F_WRLCK, start_offset, total_length,
+                                                        OMPIO_LOCK_ENTIRE_REGION, &data->aio_lock_counter );
                         if ( 0 < ret_code ) {
                             opal_output(1, "mca_fbtl_posix_progress: error in mca_fbtl_posix_lock() %d", ret_code);
                             /* Just in case some part of the lock actually succeeded. */
-                            mca_fbtl_posix_unlock ( &data->aio_lock, data->aio_fh );
+                            mca_fbtl_posix_unlock ( &data->aio_lock, data->aio_fh, &data->aio_lock_counter );
                             return OMPI_ERROR;
                         }
                         if (-1 == aio_write(&data->aio_reqs[i])) {
                             opal_output(1, "mca_fbtl_posix_progress: error in aio_write()");
-                            mca_fbtl_posix_unlock ( &data->aio_lock, data->aio_fh );
+                            mca_fbtl_posix_unlock ( &data->aio_lock, data->aio_fh, &data->aio_lock_counter );
                             return OMPI_ERROR;
                         }                        
                     }
                     else if (  data->aio_req_type == FBTL_POSIX_READ ) {
-                        ret_code = mca_fbtl_posix_lock( &data->aio_lock, data->aio_fh, F_RDLCK, start_offset, total_length, OMPIO_LOCK_ENTIRE_REGION );
+                        ret_code = mca_fbtl_posix_lock( &data->aio_lock, data->aio_fh, F_RDLCK, start_offset, total_length,
+                                                        OMPIO_LOCK_ENTIRE_REGION, &data->aio_lock_counter );
                         if ( 0 < ret_code ) {
                             opal_output(1, "mca_fbtl_posix_progress: error in mca_fbtl_posix_lock() %d", ret_code);
                             /* Just in case some part of the lock actually succeeded. */
-                            mca_fbtl_posix_unlock ( &data->aio_lock, data->aio_fh );
+                            mca_fbtl_posix_unlock ( &data->aio_lock, data->aio_fh, &data->aio_lock_counter );
                             return OMPI_ERROR;
                         }
                         if (-1 == aio_read(&data->aio_reqs[i])) {
                             opal_output(1, "mca_fbtl_posix_progress: error in aio_read()");
-                            mca_fbtl_posix_unlock ( &data->aio_lock, data->aio_fh );
+                            mca_fbtl_posix_unlock ( &data->aio_lock, data->aio_fh, &data->aio_lock_counter );
                             return OMPI_ERROR;
                         }
-                        mca_fbtl_posix_unlock ( &data->aio_lock, data->aio_fh );
                     }
                 }
 		else {
@@ -199,10 +204,9 @@ bool mca_fbtl_posix_progress ( mca_ompio_request_t *req)
 #if 0
     printf("lcount=%d open_reqs=%d\n", lcount, data->aio_open_reqs );
 #endif
-
     if ( (lcount == data->aio_req_chunks) && (0 != data->aio_open_reqs )) {
         /* release the lock of the previous operations */
-        mca_fbtl_posix_unlock ( &data->aio_lock, data->aio_fh );
+        mca_fbtl_posix_unlock ( &data->aio_lock, data->aio_fh, &data->aio_lock_counter );
 
 	/* post the next batch of operations */
 	data->aio_first_active_req = data->aio_last_active_req;
@@ -218,30 +222,32 @@ bool mca_fbtl_posix_progress ( mca_ompio_request_t *req)
         total_length = (end_offset - start_offset);
 
         if ( FBTL_POSIX_READ == data->aio_req_type ) {
-            ret_code = mca_fbtl_posix_lock( &data->aio_lock, data->aio_fh, F_RDLCK, start_offset, total_length, OMPIO_LOCK_ENTIRE_REGION );
+            ret_code = mca_fbtl_posix_lock( &data->aio_lock, data->aio_fh, F_RDLCK, start_offset, total_length,
+                                            OMPIO_LOCK_ENTIRE_REGION, &data->aio_lock_counter );
         }
         else if ( FBTL_POSIX_WRITE == data->aio_req_type ) {
-            ret_code = mca_fbtl_posix_lock( &data->aio_lock, data->aio_fh, F_WRLCK, start_offset, total_length, OMPIO_LOCK_ENTIRE_REGION );
+            ret_code = mca_fbtl_posix_lock( &data->aio_lock, data->aio_fh, F_WRLCK, start_offset, total_length,
+                                            OMPIO_LOCK_ENTIRE_REGION, &data->aio_lock_counter );
         }
         if ( 0 < ret_code ) {
             opal_output(1, "mca_fbtl_posix_progress: error in mca_fbtl_posix_lock() %d", ret_code);
             /* Just in case some part of the lock actually succeeded. */
-            mca_fbtl_posix_unlock ( &data->aio_lock, data->aio_fh );
+            mca_fbtl_posix_unlock ( &data->aio_lock, data->aio_fh, &data->aio_lock_counter );
             return OMPI_ERROR;
         }
 
 	for ( i=data->aio_first_active_req; i< data->aio_last_active_req; i++ ) {
 	    if ( FBTL_POSIX_READ == data->aio_req_type ) {
 		if (-1 == aio_read(&data->aio_reqs[i])) {
 		    opal_output(1, "mca_fbtl_posix_progress: error in aio_read()");
-                    mca_fbtl_posix_unlock ( &data->aio_lock, data->aio_fh );
+                    mca_fbtl_posix_unlock ( &data->aio_lock, data->aio_fh, &data->aio_lock_counter );
 		    return OMPI_ERROR;
 		}
 	    }
 	    else if ( FBTL_POSIX_WRITE == data->aio_req_type ) {
 		if (-1 == aio_write(&data->aio_reqs[i])) {
 		    opal_output(1, "mca_fbtl_posix_progress: error in aio_write()");
-                    mca_fbtl_posix_unlock ( &data->aio_lock, data->aio_fh );
+                    mca_fbtl_posix_unlock ( &data->aio_lock, data->aio_fh, &data->aio_lock_counter );
 		    return OMPI_ERROR;
 		}
 	    }
@@ -255,8 +261,13 @@ bool mca_fbtl_posix_progress ( mca_ompio_request_t *req)
 	/* all pending operations are finished for this request */
 	req->req_ompi.req_status.MPI_ERROR = OMPI_SUCCESS;
 	req->req_ompi.req_status._ucount = data->aio_total_len;
-        mca_fbtl_posix_unlock ( &data->aio_lock, data->aio_fh );
-	ret = true;
+        mca_fbtl_posix_unlock ( &data->aio_lock, data->aio_fh, &data->aio_lock_counter );
+
+        if ( data->aio_fh->f_atomicity ) {
+            mca_fbtl_posix_unlock ( &data->aio_lock, data->aio_fh, &data->aio_lock_counter );
+        }
+
+        ret = true;
     }
 #endif
     return ret;
@@ -268,8 +279,8 @@ void mca_fbtl_posix_request_free ( mca_ompio_request_t *req)
     /* Free the fbtl specific data structures */
     mca_fbtl_posix_request_data_t *data=(mca_fbtl_posix_request_data_t *)req->req_data;
     if (NULL != data ) {
-        mca_fbtl_posix_unlock ( &data->aio_lock, data->aio_fh );
-	if ( NULL != data->aio_reqs ) {
+            
+        if ( NULL != data->aio_reqs ) {
 	    free ( data->aio_reqs);
 	}
 	if ( NULL != data->aio_req_status ) {
@@ -281,3 +292,27 @@ void mca_fbtl_posix_request_free ( mca_ompio_request_t *req)
 #endif
   return;
 }
+
+bool mca_fbtl_posix_check_atomicity ( ompio_file_t *file)
+{    
+    struct flock lock;
+    
+    lock.l_type   = F_WRLCK;
+    lock.l_whence = SEEK_SET;
+    lock.l_start  = 0;
+    lock.l_len    = 0;
+    lock.l_pid    = 0;
+    
+    if (fcntl(file->fd, F_GETLK, &lock) < 0)
+    {
+#ifdef VERBOSE
+        printf("Failed to get lock info for '%s': %s\n", filename, strerror(errno));
+#endif
+        return false;
+    }
+
+#ifdef VERBOSE
+    printf("Lock would have worked, l_type=%d\n", (int)lock.l_type);
+#endif
+    return true;
+}

ompi/mca/fbtl/posix/fbtl_posix.h

@@ -9,7 +9,7 @@
  *                         University of Stuttgart.  All rights reserved.
  * Copyright (c) 2004-2005 The Regents of the University of California.
  *                         All rights reserved.
- * Copyright (c) 2008-2020 University of Houston. All rights reserved.
+ * Copyright (c) 2008-2021 University of Houston. All rights reserved.
  * Copyright (c) 2018      Research Organization for Information Science
  *                         and Technology (RIST). All rights reserved.
  * $COPYRIGHT$
@@ -64,10 +64,12 @@ ssize_t mca_fbtl_posix_ipwritev (ompio_file_t *file,
 
 bool mca_fbtl_posix_progress     ( mca_ompio_request_t *req);
 void mca_fbtl_posix_request_free ( mca_ompio_request_t *req);
+bool mca_fbtl_posix_check_atomicity ( ompio_file_t *file);
 
 int mca_fbtl_posix_lock ( struct flock *lock, ompio_file_t *fh, int op, 
-                          OMPI_MPI_OFFSET_TYPE iov_offset, off_t len, int flags);
-void  mca_fbtl_posix_unlock ( struct flock *lock, ompio_file_t *fh );
+                          OMPI_MPI_OFFSET_TYPE iov_offset, off_t len, int flags,
+                          int *lock_counter);
+void  mca_fbtl_posix_unlock ( struct flock *lock, ompio_file_t *fh, int *lock_counter );
 
 
 struct mca_fbtl_posix_request_data_t {
@@ -78,9 +80,10 @@ struct mca_fbtl_posix_request_data_t {
     int            aio_first_active_req; /* first active posted req */
     int            aio_last_active_req;  /* last currently active poted req */
     struct aiocb       *aio_reqs;       /* pointer array of req structures */
-    int          *aio_req_status;       /* array of statuses */
+    int           *aio_req_status;       /* array of statuses */
     ssize_t        aio_total_len;       /* total amount of data written */
     struct flock   aio_lock;            /* lock used for certain file systems */
+    int            aio_lock_counter;    /* to keep track of no. of lock calls */
     ompio_file_t  *aio_fh;       /* pointer back to the mca_io_ompio_fh structure */
 };
 typedef struct mca_fbtl_posix_request_data_t mca_fbtl_posix_request_data_t;
@@ -92,6 +95,22 @@ typedef struct mca_fbtl_posix_request_data_t mca_fbtl_posix_request_data_t;
 #define FBTL_POSIX_READ 1
 #define FBTL_POSIX_WRITE 2
 
+#define OMPIO_SET_ATOMICITY_LOCK(_fh, _lock, _lock_counter, _op) {     \
+     int32_t _orig_flags = _fh->f_flags;                               \
+     _fh->f_flags &= ~OMPIO_LOCK_NEVER;                                \
+     _fh->f_flags &= ~OMPIO_LOCK_NOT_THIS_OP;                          \
+     off_t _end_offset = (off_t)_fh->f_io_array[_fh->f_num_of_io_entries-1].offset +         \
+            (off_t)_fh->f_io_array[_fh->f_num_of_io_entries-1].length;                       \
+     off_t _len = _end_offset - (off_t)_fh->f_io_array[0].offset;                            \
+     int _ret = mca_fbtl_posix_lock ( &_lock, _fh, _op, (off_t)_fh->f_io_array[0].offset,    \
+                                    _len, OMPIO_LOCK_ENTIRE_REGION, &_lock_counter);         \
+     if ( _ret == -1 ) {                                                                     \
+          opal_output(1, "mca_fbtl_posix: error in mca_fbtl_posix_lock():%s",                \
+                      strerror(errno));                                                      \
+          return OMPI_ERROR;                                                                 \
+     }                                                                                       \
+     _fh->f_flags = _orig_flags; }
+
 
 /*
  * ******************************************************************

ompi/mca/fbtl/posix/fbtl_posix_ipreadv.c

@@ -70,8 +70,13 @@ ssize_t mca_fbtl_posix_ipreadv (ompio_file_t *fh,
         free(data);
         return 0;
     }
+    data->aio_lock_counter = 0;
     data->aio_fh = fh;
 
+    if ( fh->f_atomicity ) {
+        OMPIO_SET_ATOMICITY_LOCK(fh, data->aio_lock, data->aio_lock_counter, F_RDLCK);
+    }
+    
     for ( i=0; i<fh->f_num_of_io_entries; i++ ) {
         data->aio_reqs[i].aio_offset  = (OMPI_MPI_OFFSET_TYPE)(intptr_t)
             fh->f_io_array[i].offset;
@@ -94,10 +99,11 @@ ssize_t mca_fbtl_posix_ipreadv (ompio_file_t *fh,
     start_offset = data->aio_reqs[data->aio_first_active_req].aio_offset;
     end_offset   = data->aio_reqs[data->aio_last_active_req-1].aio_offset + data->aio_reqs[data->aio_last_active_req-1].aio_nbytes;
     total_length = (end_offset - start_offset);
-    ret = mca_fbtl_posix_lock( &data->aio_lock, data->aio_fh, F_RDLCK, start_offset, total_length, OMPIO_LOCK_ENTIRE_REGION );
+    ret = mca_fbtl_posix_lock( &data->aio_lock, data->aio_fh, F_RDLCK, start_offset, total_length,
+                               OMPIO_LOCK_ENTIRE_REGION, &data->aio_lock_counter );
     if ( 0 < ret ) {
         opal_output(1, "mca_fbtl_posix_ipreadv: error in mca_fbtl_posix_lock() error ret=%d  %s", ret, strerror(errno));
-        mca_fbtl_posix_unlock ( &data->aio_lock, data->aio_fh );            
+        mca_fbtl_posix_unlock ( &data->aio_lock, data->aio_fh, &data->aio_lock_counter );            
         free(data->aio_reqs);
         free(data->aio_req_status);
         free(data);
@@ -115,7 +121,7 @@ ssize_t mca_fbtl_posix_ipreadv (ompio_file_t *fh,
 	}
 	if ( MAX_ATTEMPTS == counter ) {
            opal_output(1, "mca_fbtl_posix_ipreadv: error in aio_read(): errno %d %s", errno, strerror(errno));
-	   mca_fbtl_posix_unlock ( &data->aio_lock, data->aio_fh );            
+	   mca_fbtl_posix_unlock ( &data->aio_lock, data->aio_fh, &data->aio_lock_counter );            
 	   free(data->aio_reqs);
 	   free(data->aio_req_status);
 	   free(data);