root/ompi/mca/io/ompio/io_ompio_file_write.c

DEFINITIONS

1. mca_io_ompio_file_write
2. mca_io_ompio_file_write_at
3. mca_io_ompio_file_iwrite
4. mca_io_ompio_file_iwrite_at
5. mca_io_ompio_file_write_all
6. mca_io_ompio_file_write_at_all
7. mca_io_ompio_file_iwrite_all
8. mca_io_ompio_file_iwrite_at_all
9. mca_io_ompio_file_write_shared
10. mca_io_ompio_file_iwrite_shared
11. mca_io_ompio_file_write_ordered
12. mca_io_ompio_file_write_ordered_begin
13. mca_io_ompio_file_write_ordered_end
14. mca_io_ompio_file_write_all_begin
15. mca_io_ompio_file_write_all_end
16. mca_io_ompio_file_write_at_all_begin
17. mca_io_ompio_file_write_at_all_end

   1 /*
   2  * Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
   3  *                         University Research and Technology
   4  *                         Corporation.  All rights reserved.
   5  * Copyright (c) 2004-2016 The University of Tennessee and The University
   6  *                         of Tennessee Research Foundation.  All rights
   7  *                         reserved.
   8  * Copyright (c) 2004-2005 High Performance Computing Center Stuttgart,
   9  *                         University of Stuttgart.  All rights reserved.
  10  * Copyright (c) 2004-2005 The Regents of the University of California.
  11  *                         All rights reserved.
  12  * Copyright (c) 2008-2019 University of Houston. All rights reserved.
  13  * Copyright (c) 2015-2018 Research Organization for Information Science
  14  *                         and Technology (RIST). All rights reserved.
  15  * $COPYRIGHT$
  16  *
  17  * Additional copyrights may follow
  18  *
  19  * $HEADER$
  20  */
  21 
  22 #include "ompi_config.h"
  23 
  24 #include "ompi/communicator/communicator.h"
  25 #include "ompi/info/info.h"
  26 #include "ompi/file/file.h"
  27 #include "ompi/mca/fs/fs.h"
  28 #include "ompi/mca/fs/base/base.h"
  29 #include "ompi/mca/fcoll/fcoll.h"
  30 #include "ompi/mca/fcoll/base/base.h"
  31 #include "ompi/mca/fbtl/fbtl.h"
  32 #include "ompi/mca/fbtl/base/base.h"
  33 #include "ompi/mca/sharedfp/sharedfp.h"
  34 #include "ompi/mca/sharedfp/base/base.h"
  35 
  36 #include "io_ompio.h"
  37 #include "ompi/mca/common/ompio/common_ompio_request.h"
  38 #include "math.h"
  39 #include <unistd.h>
  40 
  41 /* Read and write routines are split into two interfaces.
  42 **   The
  43 **   mca_io_ompio_file_read/write[_at]
  44 **
  45 ** routines are the ones registered with the ompio modules.
  46 ** The
  47 **
  48 ** mca_common_ompio_file_read/write[_at]
  49 **
  50 ** routesin are used e.g. from the shared file pointer modules.
  51 ** The main difference is, that the first one takes an ompi_file_t
  52 ** as a file pointer argument, while the second uses the ompio internal
  53 ** ompio_file_t structure.
  54 */
  55 
  56 
  57 int mca_io_ompio_file_write (ompi_file_t *fp,
  58                              const void *buf,
  59                              int count,
  60                              struct ompi_datatype_t *datatype,
  61                              ompi_status_public_t *status)
  62 {
  63     int ret = OMPI_SUCCESS;
  64     mca_common_ompio_data_t *data;
  65     ompio_file_t *fh;
  66 
  67     data = (mca_common_ompio_data_t *) fp->f_io_selected_data;
  68     fh = &data->ompio_fh;
  69     OPAL_THREAD_LOCK(&fp->f_lock);
  70     ret = mca_common_ompio_file_write(fh,buf,count,datatype,status);
  71     OPAL_THREAD_UNLOCK(&fp->f_lock);
  72     return ret;
  73 }
  74 
  75 
  76 int mca_io_ompio_file_write_at (ompi_file_t *fh,
  77                                 OMPI_MPI_OFFSET_TYPE offset,
  78                                 const void *buf,
  79                                 int count,
  80                                 struct ompi_datatype_t *datatype,
  81                                 ompi_status_public_t *status)
  82 {
  83     int ret = OMPI_SUCCESS;
  84     mca_common_ompio_data_t *data;
  85 
  86     data = (mca_common_ompio_data_t *) fh->f_io_selected_data;
  87     OPAL_THREAD_LOCK(&fh->f_lock);
  88     ret = mca_common_ompio_file_write_at (&data->ompio_fh, offset,buf,count,datatype,status);
  89     OPAL_THREAD_UNLOCK(&fh->f_lock);
  90 
  91     return ret;
  92 }
  93 
  94 int mca_io_ompio_file_iwrite (ompi_file_t *fp,
  95                               const void *buf,
  96                               int count,
  97                               struct ompi_datatype_t *datatype,
  98                               ompi_request_t **request)
  99 {
 100     int ret = OMPI_SUCCESS;
 101     mca_common_ompio_data_t *data;
 102 
 103     data = (mca_common_ompio_data_t *) fp->f_io_selected_data;
 104     OPAL_THREAD_LOCK(&fp->f_lock);
 105     ret = mca_common_ompio_file_iwrite(&data->ompio_fh,buf,count,datatype,request);
 106     OPAL_THREAD_UNLOCK(&fp->f_lock);
 107 
 108     return ret;
 109 }
 110 
 111 
 112 int mca_io_ompio_file_iwrite_at (ompi_file_t *fh,
 113                                  OMPI_MPI_OFFSET_TYPE offset,
 114                                  const void *buf,
 115                                  int count,
 116                                  struct ompi_datatype_t *datatype,
 117                                  ompi_request_t **request)
 118 {
 119     int ret = OMPI_SUCCESS;
 120     mca_common_ompio_data_t *data;
 121 
 122     data = (mca_common_ompio_data_t *) fh->f_io_selected_data;
 123     OPAL_THREAD_LOCK(&fh->f_lock);
 124     ret = mca_common_ompio_file_iwrite_at(&data->ompio_fh,offset,buf,count,datatype,request);
 125     OPAL_THREAD_UNLOCK(&fh->f_lock);
 126 
 127     return ret;
 128 }
 129 
 130 
 131 /* Collective operations                                          */
 132 /******************************************************************/
 133 
 134 int mca_io_ompio_file_write_all (ompi_file_t *fh,
 135                                  const void *buf,
 136                                  int count,
 137                                  struct ompi_datatype_t *datatype,
 138                                  ompi_status_public_t *status)
 139 {
 140     int ret = OMPI_SUCCESS;
 141     mca_common_ompio_data_t *data;
 142 
 143     data = (mca_common_ompio_data_t *) fh->f_io_selected_data;
 144 
 145     OPAL_THREAD_LOCK(&fh->f_lock);
 146     ret = mca_common_ompio_file_write_all (&data->ompio_fh,
 147                                            buf,
 148                                            count,
 149                                            datatype,
 150                                            status);
 151     OPAL_THREAD_UNLOCK(&fh->f_lock);
 152     if ( MPI_STATUS_IGNORE != status ) {
 153         size_t size;
 154 
 155         opal_datatype_type_size (&datatype->super, &size);
 156         status->_ucount = count * size;
 157     }
 158 
 159     return ret;
 160 }
 161 
 162 int mca_io_ompio_file_write_at_all (ompi_file_t *fh,
 163                                     OMPI_MPI_OFFSET_TYPE offset,
 164                                     const void *buf,
 165                                     int count,
 166                                     struct ompi_datatype_t *datatype,
 167                                     ompi_status_public_t *status)
 168 {
 169     int ret = OMPI_SUCCESS;
 170     mca_common_ompio_data_t *data;
 171 
 172     data = (mca_common_ompio_data_t *) fh->f_io_selected_data;
 173     OPAL_THREAD_LOCK(&fh->f_lock);
 174     ret = mca_common_ompio_file_write_at_all(&data->ompio_fh,offset,buf,count,datatype,status);
 175     OPAL_THREAD_UNLOCK(&fh->f_lock);
 176 
 177     return ret;
 178 }
 179 
 180 int mca_io_ompio_file_iwrite_all (ompi_file_t *fh,
 181                                   const void *buf,
 182                                   int count,
 183                                   struct ompi_datatype_t *datatype,
 184                                   ompi_request_t **request)
 185 {
 186     int ret = OMPI_SUCCESS;
 187     mca_common_ompio_data_t *data=NULL;
 188     ompio_file_t *fp=NULL;
 189 
 190     data = (mca_common_ompio_data_t *) fh->f_io_selected_data;
 191     fp = &data->ompio_fh;
 192 
 193     OPAL_THREAD_LOCK(&fh->f_lock);
 194     ret = mca_common_ompio_file_iwrite_all (&data->ompio_fh,
 195                                             buf,
 196                                             count,
 197                                             datatype,
 198                                             request);
 199     OPAL_THREAD_UNLOCK(&fh->f_lock);
 200 
 201     return ret;
 202 }
 203 
 204 
 205 int mca_io_ompio_file_iwrite_at_all (ompi_file_t *fh,
 206                                      OMPI_MPI_OFFSET_TYPE offset,
 207                                      const void *buf,
 208                                      int count,
 209                                      struct ompi_datatype_t *datatype,
 210                                      ompi_request_t **request)
 211 {
 212     int ret = OMPI_SUCCESS;
 213     mca_common_ompio_data_t *data;
 214 
 215     data = (mca_common_ompio_data_t *) fh->f_io_selected_data;
 216     OPAL_THREAD_LOCK(&fh->f_lock);
 217     ret = mca_common_ompio_file_iwrite_at_all ( &data->ompio_fh, offset, buf, count, datatype, request );
 218     OPAL_THREAD_UNLOCK(&fh->f_lock);
 219     return ret;
 220 }
 221 
 222 
 223 /* Infrastructure for shared file pointer operations */
 224 /* (Individual and collective */
 225 /******************************************************/
 226 
 227 int mca_io_ompio_file_write_shared (ompi_file_t *fp,
 228                                     const void *buf,
 229                                     int count,
 230                                     struct ompi_datatype_t *datatype,
 231                                     ompi_status_public_t * status)
 232 {
 233     int ret = OMPI_SUCCESS;
 234     mca_common_ompio_data_t *data;
 235     ompio_file_t *fh;
 236     mca_sharedfp_base_module_t * shared_fp_base_module;
 237 
 238     data = (mca_common_ompio_data_t *) fp->f_io_selected_data;
 239     fh = &data->ompio_fh;
 240 
 241     /*get the shared fp module associated with this file*/
 242     shared_fp_base_module = fh->f_sharedfp;
 243     if ( NULL == shared_fp_base_module ){
 244         opal_output(0, "No shared file pointer component found for this communicator. Can not execute\n");
 245         return OMPI_ERROR;
 246     }
 247     OPAL_THREAD_LOCK(&fp->f_lock);
 248     ret = shared_fp_base_module->sharedfp_write(fh,buf,count,datatype,status);
 249     OPAL_THREAD_UNLOCK(&fp->f_lock);
 250 
 251     return ret;
 252 }
 253 
 254 int mca_io_ompio_file_iwrite_shared (ompi_file_t *fp,
 255                                      const void *buf,
 256                                      int count,
 257                                      struct ompi_datatype_t *datatype,
 258                                      ompi_request_t **request)
 259 {
 260     int ret = OMPI_SUCCESS;
 261     mca_common_ompio_data_t *data;
 262     ompio_file_t *fh;
 263     mca_sharedfp_base_module_t * shared_fp_base_module;
 264 
 265     data = (mca_common_ompio_data_t *) fp->f_io_selected_data;
 266     fh = &data->ompio_fh;
 267 
 268     /*get the shared fp module associated with this file*/
 269     shared_fp_base_module = fh->f_sharedfp;
 270     if ( NULL == shared_fp_base_module ){
 271         opal_output(0, "No shared file pointer component found for this communicator. Can not execute\n");
 272         return OMPI_ERROR;
 273     }
 274     OPAL_THREAD_LOCK(&fp->f_lock);
 275     ret = shared_fp_base_module->sharedfp_iwrite(fh,buf,count,datatype,request);
 276     OPAL_THREAD_UNLOCK(&fp->f_lock);
 277 
 278     return ret;
 279 }
 280 
 281 int mca_io_ompio_file_write_ordered (ompi_file_t *fp,
 282                                      const void *buf,
 283                                      int count,
 284                                      struct ompi_datatype_t *datatype,
 285                                      ompi_status_public_t * status)
 286 {
 287     int ret = OMPI_SUCCESS;
 288     mca_common_ompio_data_t *data;
 289     ompio_file_t *fh;
 290     mca_sharedfp_base_module_t * shared_fp_base_module;
 291 
 292     data = (mca_common_ompio_data_t *) fp->f_io_selected_data;
 293     fh = &data->ompio_fh;
 294 
 295     /*get the shared fp module associated with this file*/
 296     shared_fp_base_module = fh->f_sharedfp;
 297     if ( NULL == shared_fp_base_module ){
 298         opal_output(0,"No shared file pointer component found for this communicator. Can not execute\n");
 299         return OMPI_ERROR;
 300     }
 301     OPAL_THREAD_LOCK(&fp->f_lock);
 302     ret = shared_fp_base_module->sharedfp_write_ordered(fh,buf,count,datatype,status);
 303     OPAL_THREAD_UNLOCK(&fp->f_lock);
 304 
 305     return ret;
 306 }
 307 
 308 int mca_io_ompio_file_write_ordered_begin (ompi_file_t *fp,
 309                                            const void *buf,
 310                                            int count,
 311                                            struct ompi_datatype_t *datatype)
 312 {
 313     int ret = OMPI_SUCCESS;
 314     mca_common_ompio_data_t *data;
 315     ompio_file_t *fh;
 316     mca_sharedfp_base_module_t * shared_fp_base_module;
 317 
 318     data = (mca_common_ompio_data_t *) fp->f_io_selected_data;
 319     fh = &data->ompio_fh;
 320 
 321     /*get the shared fp module associated with this file*/
 322     shared_fp_base_module = fh->f_sharedfp;
 323     if ( NULL == shared_fp_base_module ){
 324         opal_output(0, "No shared file pointer component found for this communicator. Can not execute\n");
 325         return OMPI_ERROR;
 326     }
 327     OPAL_THREAD_LOCK(&fp->f_lock);
 328     ret = shared_fp_base_module->sharedfp_write_ordered_begin(fh,buf,count,datatype);
 329     OPAL_THREAD_UNLOCK(&fp->f_lock);
 330 
 331     return ret;
 332 }
 333 
 334 int mca_io_ompio_file_write_ordered_end (ompi_file_t *fp,
 335                                          const void *buf,
 336                                          ompi_status_public_t *status)
 337 {
 338     int ret = OMPI_SUCCESS;
 339     mca_common_ompio_data_t *data;
 340     ompio_file_t *fh;
 341     mca_sharedfp_base_module_t * shared_fp_base_module;
 342 
 343     data = (mca_common_ompio_data_t *) fp->f_io_selected_data;
 344     fh = &data->ompio_fh;
 345 
 346     /*get the shared fp module associated with this file*/
 347     shared_fp_base_module = fh->f_sharedfp;
 348     if ( NULL == shared_fp_base_module ){
 349         opal_output(0, "No shared file pointer component found for this communicator. Can not execute\n");
 350         return OMPI_ERROR;
 351     }
 352     OPAL_THREAD_LOCK(&fp->f_lock);
 353     ret = shared_fp_base_module->sharedfp_write_ordered_end(fh,buf,status);
 354     OPAL_THREAD_UNLOCK(&fp->f_lock);
 355 
 356     return ret;
 357 }
 358 
 359 
 360 /* Split collectives . Not really used but infrastructure is in place */
 361 /**********************************************************************/
 362 int mca_io_ompio_file_write_all_begin (ompi_file_t *fh,
 363                                        const void *buf,
 364                                        int count,
 365                                        struct ompi_datatype_t *datatype)
 366 {
 367     int ret = OMPI_SUCCESS;
 368     ompio_file_t *fp;
 369     mca_common_ompio_data_t *data;
 370 
 371     data = (mca_common_ompio_data_t *) fh->f_io_selected_data;
 372     fp = &data->ompio_fh;
 373     if ( true == fp->f_split_coll_in_use ) {
 374         printf("Only one split collective I/O operation allowed per file handle at any given point in time!\n");
 375         return MPI_ERR_OTHER;
 376     }
 377     /* No need for locking fh->f_lock, that is done in file_iwrite_all */
 378     ret = mca_io_ompio_file_iwrite_all ( fh, buf, count, datatype, &fp->f_split_coll_req );
 379     fp->f_split_coll_in_use = true;
 380 
 381     return ret;
 382 }
 383 
 384 int mca_io_ompio_file_write_all_end (ompi_file_t *fh,
 385                                      const void *buf,
 386                                      ompi_status_public_t *status)
 387 {
 388     int ret = OMPI_SUCCESS;
 389     ompio_file_t *fp;
 390     mca_common_ompio_data_t *data;
 391 
 392     data = (mca_common_ompio_data_t *) fh->f_io_selected_data;
 393     fp = &data->ompio_fh;
 394     ret = ompi_request_wait ( &fp->f_split_coll_req, status );
 395 
 396     /* remove the flag again */
 397     fp->f_split_coll_in_use = false;
 398     return ret;
 399 }
 400 
 401 
 402 int mca_io_ompio_file_write_at_all_begin (ompi_file_t *fh,
 403                                           OMPI_MPI_OFFSET_TYPE offset,
 404                                           const void *buf,
 405                                           int count,
 406                                           struct ompi_datatype_t *datatype)
 407 {
 408     int ret = OMPI_SUCCESS;
 409     mca_common_ompio_data_t *data=NULL;
 410     ompio_file_t *fp=NULL;
 411 
 412     data = (mca_common_ompio_data_t *) fh->f_io_selected_data;
 413     fp = &data->ompio_fh;
 414 
 415     if ( true == fp->f_split_coll_in_use ) {
 416         printf("Only one split collective I/O operation allowed per file handle at any given point in time!\n");
 417         return MPI_ERR_REQUEST;
 418     }
 419     OPAL_THREAD_LOCK(&fh->f_lock);
 420     ret = mca_common_ompio_file_iwrite_at_all ( fp, offset, buf, count, datatype, &fp->f_split_coll_req );
 421     OPAL_THREAD_UNLOCK(&fh->f_lock);
 422     fp->f_split_coll_in_use = true;
 423 
 424     return ret;
 425 }
 426 
 427 
 428 int mca_io_ompio_file_write_at_all_end (ompi_file_t *fh,
 429                                         const void *buf,
 430                                         ompi_status_public_t * status)
 431 {
 432     int ret = OMPI_SUCCESS;
 433     mca_common_ompio_data_t *data;
 434     ompio_file_t *fp=NULL;
 435 
 436     data = (mca_common_ompio_data_t *) fh->f_io_selected_data;
 437     fp = &data->ompio_fh;
 438     ret = ompi_request_wait ( &fp->f_split_coll_req, status );
 439 
 440     /* remove the flag again */
 441     fp->f_split_coll_in_use = false;
 442 
 443     return ret;
 444 }