root/ompi/mca/sharedfp/sm/sharedfp_sm_write.c

DEFINITIONS

1. mca_sharedfp_sm_write
2. mca_sharedfp_sm_write_ordered

   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-2017 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) 2013-2018 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 
  23 #include "ompi_config.h"
  24 #include "sharedfp_sm.h"
  25 
  26 #include "mpi.h"
  27 #include "ompi/constants.h"
  28 #include "ompi/mca/sharedfp/sharedfp.h"
  29 #include "ompi/mca/sharedfp/base/base.h"
  30 
  31 int mca_sharedfp_sm_write (ompio_file_t *fh,
  32                            const void *buf,
  33                            int count,
  34                            struct ompi_datatype_t *datatype,
  35                            ompi_status_public_t *status)
  36 {
  37     int ret = OMPI_SUCCESS;
  38     OMPI_MPI_OFFSET_TYPE offset = 0;
  39     long bytesRequested = 0;
  40     size_t numofBytes;
  41 
  42     if( NULL == fh->f_sharedfp_data ){
  43         opal_output(ompi_sharedfp_base_framework.framework_output,
  44                     "sharedfp_sm_write:  module not initialized\n");
  45         return OMPI_ERROR;
  46     }
  47 
  48     /* Calculate the number of bytes to write*/
  49     opal_datatype_type_size ( &datatype->super, &numofBytes);
  50     bytesRequested = count * numofBytes;
  51 
  52     /*Retrieve the shared file data struct*/
  53 
  54     if ( mca_sharedfp_sm_verbose ) {
  55         opal_output(ompi_sharedfp_base_framework.framework_output,
  56                     "sharedfp_sm_write: Requested is %ld\n",bytesRequested);
  57     }
  58 
  59     /*Request the offset to write bytesRequested bytes*/
  60     ret = mca_sharedfp_sm_request_position(fh, bytesRequested,&offset);
  61     offset /= fh->f_etype_size;
  62     if ( -1 != ret ) {
  63         if ( mca_sharedfp_sm_verbose ) {
  64             opal_output(ompi_sharedfp_base_framework.framework_output,
  65                         "sharedfp_sm_write: fset received is %lld\n",offset);
  66         }
  67 
  68         /* Write to the file*/
  69         ret = mca_common_ompio_file_write_at(fh,offset,buf,count,datatype,status);
  70     }
  71 
  72     return ret;
  73 }
  74 
  75 int mca_sharedfp_sm_write_ordered (ompio_file_t *fh,
  76                                    const void *buf,
  77                                    int count,
  78                                    struct ompi_datatype_t *datatype,
  79                                    ompi_status_public_t *status)
  80 {
  81     int ret = OMPI_SUCCESS;
  82     OMPI_MPI_OFFSET_TYPE offset = 0;
  83     long sendBuff = 0;
  84     long *buff=NULL;
  85     long offsetBuff;
  86     OMPI_MPI_OFFSET_TYPE offsetReceived = 0;
  87     long bytesRequested = 0;
  88     int recvcnt = 1, sendcnt = 1;
  89     size_t numofBytes;
  90     int i;
  91 
  92     if( NULL == fh->f_sharedfp_data){
  93         opal_output(ompi_sharedfp_base_framework.framework_output,
  94                     "sharedfp_sm_write_ordered: module not initialzed \n");
  95         return OMPI_ERROR;
  96     }
  97 
  98     /* Calculate the number of bytes to write*/
  99     opal_datatype_type_size ( &datatype->super, &numofBytes);
 100     sendBuff = count * numofBytes;
 101 
 102     if ( 0 == fh->f_rank ) {
 103         buff = (long*)malloc(sizeof(long) * fh->f_size);
 104         if ( NULL == buff )
 105             return OMPI_ERR_OUT_OF_RESOURCE;
 106     }
 107 
 108     ret = fh->f_comm->c_coll->coll_gather ( &sendBuff, sendcnt, OMPI_OFFSET_DATATYPE,
 109                                             buff, recvcnt, OMPI_OFFSET_DATATYPE, 0,
 110                                             fh->f_comm, fh->f_comm->c_coll->coll_gather_module );
 111     if ( OMPI_SUCCESS != ret ) {
 112         goto exit;
 113     }
 114 
 115     /* All the counts are present now in the recvBuff.
 116     ** The size of recvBuff is sizeof_newComm
 117     */
 118     if (  0 == fh->f_rank ) {
 119         for (i = 0; i < fh->f_size ; i ++) {
 120             bytesRequested += buff[i];
 121             if ( mca_sharedfp_sm_verbose ) {
 122                 opal_output(ompi_sharedfp_base_framework.framework_output,
 123                             "sharedfp_sm_write_ordered: Bytes requested are %ld\n",bytesRequested);
 124             }
 125         }
 126 
 127         /* Request the offset to write bytesRequested bytes
 128         ** only the root process needs to do the request,
 129         ** since the root process will then tell the other
 130         ** processes at what offset they should write their
 131         ** share of the data.
 132         */
 133         ret = mca_sharedfp_sm_request_position(fh,bytesRequested,&offsetReceived);
 134         if( OMPI_SUCCESS != ret){
 135             goto exit;
 136         }
 137         if ( mca_sharedfp_sm_verbose ) {
 138             opal_output(ompi_sharedfp_base_framework.framework_output,
 139                         "sharedfp_sm_write_ordered: Offset received is %lld\n",offsetReceived);
 140         }
 141         buff[0] += offsetReceived;
 142 
 143         for (i = 1 ; i < fh->f_size; i++) {
 144             buff[i] += buff[i-1];
 145         }
 146     }
 147 
 148     /* Scatter the results to the other processes*/
 149     ret = fh->f_comm->c_coll->coll_scatter ( buff, sendcnt, OMPI_OFFSET_DATATYPE,
 150                                              &offsetBuff, recvcnt, OMPI_OFFSET_DATATYPE, 0,
 151                                              fh->f_comm, fh->f_comm->c_coll->coll_scatter_module );
 152 
 153     if ( OMPI_SUCCESS != ret ) {
 154         goto exit;
 155     }
 156 
 157     /* Each process now has its own individual offset */
 158     offset = offsetBuff - sendBuff;
 159     offset /= fh->f_etype_size;
 160 
 161     if ( mca_sharedfp_sm_verbose ) {
 162         opal_output(ompi_sharedfp_base_framework.framework_output,
 163                     "sharedfp_sm_write_ordered: Offset returned is %lld\n",offset);
 164     }
 165     /* write to the file */
 166     ret = mca_common_ompio_file_write_at_all(fh,offset,buf,count,datatype,status);
 167 
 168 exit:
 169     if ( NULL != buff ) {
 170         free ( buff );
 171     }
 172 
 173     return ret;
 174 }