From 1b457125957d2fd293fccfdc8725e4c4500a7b09 Mon Sep 17 00:00:00 2001 From: Edgar Gabriel Date: Tue, 18 Aug 2015 10:29:13 -0500 Subject: [PATCH] bring the addproc component up to date with support for split collectives. No pr required for this commit, since the addproc component is not part of v2.x --- .../addproc/sharedfp_addproc_file_open.c | 27 ++-- .../sharedfp/addproc/sharedfp_addproc_iread.c | 132 +++++++++++++++++- .../addproc/sharedfp_addproc_iwrite.c | 127 ++++++++++++++++- .../sharedfp/addproc/sharedfp_addproc_read.c | 62 +++----- .../sharedfp/addproc/sharedfp_addproc_write.c | 61 +++----- 5 files changed, 307 insertions(+), 102 deletions(-) diff --git a/ompi/mca/sharedfp/addproc/sharedfp_addproc_file_open.c b/ompi/mca/sharedfp/addproc/sharedfp_addproc_file_open.c index b51b605368..d7ecf4c248 100644 --- a/ompi/mca/sharedfp/addproc/sharedfp_addproc_file_open.c +++ b/ompi/mca/sharedfp/addproc/sharedfp_addproc_file_open.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) 2013 University of Houston. All rights reserved. + * Copyright (c) 2013-2015 University of Houston. All rights reserved. * $COPYRIGHT$ * * Additional copyrights may follow @@ -30,6 +30,7 @@ #include #include "ompi/mca/sharedfp/base/base.h" + int mca_sharedfp_addproc_file_open (struct ompi_communicator_t *comm, char* filename, int amode, @@ -49,17 +50,19 @@ int mca_sharedfp_addproc_file_open (struct ompi_communicator_t *comm, shfileHandle = (mca_io_ompio_file_t *)malloc(sizeof(mca_io_ompio_file_t)); ret = ompio_io_ompio_file_open(comm,filename,amode,info,shfileHandle,false); if ( OMPI_SUCCESS != ret) { - printf( "mca_sharedfp_addproc_file_open: Error during file open\n"); + opal_output(0, "mca_sharedfp_addproc_file_open: Error during file open\n"); return ret; } /*Memory is allocated here for the sh structure*/ if ( mca_sharedfp_addproc_verbose ) { - printf( "mca_sharedfp_addproc_file_open: malloc f_sharedfp_ptr struct\n"); + opal_output(ompi_sharedfp_base_framework.framework_output, + "mca_sharedfp_addproc_file_open: malloc f_sharedfp_ptr struct\n"); } sh = (struct mca_sharedfp_base_data_t*)malloc(sizeof(struct mca_sharedfp_base_data_t)); if ( NULL == sh ){ - printf( "mca_sharedfp_addproc_file_open: Error, unable to malloc f_sharedfp_ptr struct\n"); + opal_output(ompi_sharedfp_base_framework.framework_output, + "mca_sharedfp_addproc_file_open: Error, unable to malloc f_sharedfp_ptr struct\n"); return OMPI_ERR_OUT_OF_RESOURCE; } @@ -72,23 +75,24 @@ int mca_sharedfp_addproc_file_open (struct ompi_communicator_t *comm, rank = ompi_comm_rank ( sh->comm ); if ( mca_sharedfp_addproc_verbose ) { - printf( "mca_sharedfp_addproc_file_open: START spawn by rank=%d\n",rank); + opal_output(ompi_sharedfp_base_framework.framework_output, + "mca_sharedfp_addproc_file_open: START spawn by rank=%d\n",rank); } /*Spawn a new process which will maintain the offsets for this file open*/ ret = MPI_Comm_spawn("mca_sharedfp_addproc_control", MPI_ARGV_NULL, 1, MPI_INFO_NULL, 0, sh->comm, &newInterComm, &err); if ( OMPI_SUCCESS != ret ) { - printf( "mca_sharedfp_addproc_file_open: error spawning control process ret=%d\n", - ret); + opal_output(0, "mca_sharedfp_addproc_file_open: error spawning control process ret=%d\n", + ret); } /*If spawning successful*/ if (newInterComm) { addproc_data = (struct mca_sharedfp_addproc_data*)malloc(sizeof(struct mca_sharedfp_addproc_data)); if ( NULL == addproc_data ){ - printf( "mca_sharedfp_addproc_file_open: Error, unable to malloc addproc_data struct\n"); - return OMPI_ERR_OUT_OF_RESOURCE; + opal_output (0,"mca_sharedfp_addproc_file_open: Error, unable to malloc addproc_data struct\n"); + return OMPI_ERR_OUT_OF_RESOURCE; } /*Store the new Intercommunicator*/ @@ -100,7 +104,8 @@ int mca_sharedfp_addproc_file_open (struct ompi_communicator_t *comm, fh->f_sharedfp_data = sh; } else{ - printf( "mca_sharedfp_addproc_file_open: DONE spawn by rank=%d, errcode[success=%d, err=%d]=%d\n", + opal_output(ompi_sharedfp_base_framework.framework_output, + "mca_sharedfp_addproc_file_open: DONE spawn by rank=%d, errcode[success=%d, err=%d]=%d\n", rank, MPI_SUCCESS, MPI_ERR_SPAWN, ret); ret = OMPI_ERROR; } @@ -120,7 +125,7 @@ int mca_sharedfp_addproc_file_close (mca_io_ompio_file_t *fh) if ( NULL == fh->f_sharedfp_data){ /* Can happen with lazy initialization of the sharedfp structures */ if ( mca_sharedfp_addproc_verbose ) { - printf( "sharedfp_addproc_file_close - shared file pointer structure not initialized\n"); + opal_output(0, "sharedfp_addproc_file_close - shared file pointer structure not initialized\n"); } return OMPI_SUCCESS; } diff --git a/ompi/mca/sharedfp/addproc/sharedfp_addproc_iread.c b/ompi/mca/sharedfp/addproc/sharedfp_addproc_iread.c index 5b1ef44e16..9b56b6d538 100644 --- a/ompi/mca/sharedfp/addproc/sharedfp_addproc_iread.c +++ b/ompi/mca/sharedfp/addproc/sharedfp_addproc_iread.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) 2013 University of Houston. All rights reserved. + * Copyright (c) 2013-2015 University of Houston. All rights reserved. * $COPYRIGHT$ * * Additional copyrights may follow @@ -24,6 +24,9 @@ #include "mpi.h" #include "ompi/constants.h" #include "ompi/mca/sharedfp/sharedfp.h" +#include "ompi/mca/sharedfp/base/base.h" + + int mca_sharedfp_addproc_iread(mca_io_ompio_file_t *fh, void *buf, @@ -47,7 +50,8 @@ int mca_sharedfp_addproc_iread(mca_io_ompio_file_t *fh, bytesRequested = count * numofBytes; if ( mca_sharedfp_addproc_verbose ){ - printf("mca_sharedfp_addproc_iread: Bytes Requested is %ld\n",bytesRequested); + opal_output(ompi_sharedfp_base_framework.framework_output, + "mca_sharedfp_addproc_iread: Bytes Requested is %ld\n",bytesRequested); } /* Retrieve the shared file data struct */ sh = fh->f_sharedfp_data; @@ -56,7 +60,8 @@ int mca_sharedfp_addproc_iread(mca_io_ompio_file_t *fh, ret = mca_sharedfp_addproc_request_position(sh,bytesRequested,&offset); if( OMPI_SUCCESS == ret ){ if ( mca_sharedfp_addproc_verbose ){ - printf("mca_sharedfp_addproc_iread: Offset received is %lld\n",offset); + opal_output(ompi_sharedfp_base_framework.framework_output, + "mca_sharedfp_addproc_iread: Offset received is %lld\n",offset); } /* Read from the file */ ret = ompio_io_ompio_file_iread_at ( sh->sharedfh, offset, buf, count, datatype, request); @@ -69,8 +74,118 @@ int mca_sharedfp_addproc_read_ordered_begin(mca_io_ompio_file_t *fh, int count, struct ompi_datatype_t *datatype) { - opal_output(0,"mca_sharedfp_addproc_read_ordered_begin: NOT IMPLEMENTED\n"); - return OMPI_ERROR; + int ret = OMPI_SUCCESS; + OMPI_MPI_OFFSET_TYPE offset = 0, offsetReceived = 0; + long sendBuff = 0; + long *buff=NULL; + long offsetBuff, bytesRequested = 0; + size_t numofBytes; + int rank, size, i; + struct mca_sharedfp_base_data_t *sh = NULL; + + if(NULL == fh->f_sharedfp_data){ + opal_output(0, "sharedfp_addproc_read_ordered_begin: shared file pointer " + "structure not initialized correctly\n"); + return OMPI_ERROR; + } + + if ( true == fh->f_split_coll_in_use ) { + opal_output(0, "Only one split collective I/O operation allowed per " + "file handle at any given point in time!\n"); + return MPI_ERR_REQUEST; + } + + /*Retrieve the new communicator*/ + sh = fh->f_sharedfp_data; + + /* Calculate the number of bytes to read*/ + opal_datatype_type_size ( &datatype->super, &numofBytes); + sendBuff = count * numofBytes; + + /* Get the ranks in the communicator */ + rank = ompi_comm_rank ( sh->comm); + size = ompi_comm_size ( sh->comm); + + if ( 0 == rank ) { + buff = (long*)malloc(sizeof(OMPI_MPI_OFFSET_TYPE) * size); + if ( NULL == buff ) + return OMPI_ERR_OUT_OF_RESOURCE; + } + + ret = sh->comm->c_coll.coll_gather( &sendBuff, 1, OMPI_OFFSET_DATATYPE, + buff, 1, OMPI_OFFSET_DATATYPE, 0, sh->comm, + sh->comm->c_coll.coll_gather_module); + if ( OMPI_SUCCESS != ret ) { + goto exit; + } + + /* All the counts are present now in the recvBuff. + The size of recvBuff is sizeof_newComm + */ + if ( 0 == rank ) { + for (i = 0; i < size ; i ++) { + if ( mca_sharedfp_addproc_verbose ){ + opal_output(ompi_sharedfp_base_framework.framework_output, + "sharedfp_addproc_read_ordered_begin: Buff is %ld\n",buff[i]); + } + bytesRequested += buff[i]; + + if ( mca_sharedfp_addproc_verbose ){ + opal_output(ompi_sharedfp_base_framework.framework_output, + "sharedfp_addproc_read_ordered_begin: Bytes requested are %ld\n", + bytesRequested); + } + } + + /* Request the offset to read bytesRequested bytes + ** only the root process needs to do the request, + ** since the root process will then tell the other + ** processes at what offset they should read their + ** share of the data. + */ + ret = mca_sharedfp_addproc_request_position(sh,bytesRequested,&offsetReceived); + if( OMPI_SUCCESS != ret ){ + goto exit; + } + if ( mca_sharedfp_addproc_verbose ){ + opal_output(ompi_sharedfp_base_framework.framework_output, + "sharedfp_addproc_read_ordered_begin: Offset received is %lld\n", + offsetReceived); + } + buff[0] += offsetReceived; + + + for (i = 1 ; i < size; i++) { + buff[i] += buff[i-1]; + } + } + + /* Scatter the results to the other processes*/ + ret = sh->comm->c_coll.coll_scatter ( buff, 1, OMPI_OFFSET_DATATYPE, &offsetBuff, + 1, OMPI_OFFSET_DATATYPE, 0, sh->comm, + sh->comm->c_coll.coll_scatter_module ); + if ( OMPI_SUCCESS != ret ) { + goto exit; + } + + /*Each process now has its own individual offset in recvBUFF*/ + offset = offsetBuff - sendBuff; + + if ( mca_sharedfp_addproc_verbose ){ + opal_output(ompi_sharedfp_base_framework.framework_output, + "sharedfp_addproc_read_ordered_begin: Offset returned is %lld\n",offset); + } + + /* read from the file */ + ret = ompio_io_ompio_file_iread_at_all(sh->sharedfh,offset,buf,count,datatype,&fh->f_split_coll_req); + fh->f_split_coll_in_use = true; + +exit: + if ( NULL != buff ) { + free ( buff ); + } + + return ret; } @@ -79,7 +194,10 @@ int mca_sharedfp_addproc_read_ordered_end(mca_io_ompio_file_t *fh, void *buf, ompi_status_public_t *status) { - opal_output(0,"mca_sharedfp_addproc_read_ordered_end: NOT IMPLEMENTED\n"); - return OMPI_ERROR; + int ret = OMPI_SUCCESS; + ret = ompi_request_wait ( &fh->f_split_coll_req, status ); + /* remove the flag again */ + fh->f_split_coll_in_use = false; + return ret; } diff --git a/ompi/mca/sharedfp/addproc/sharedfp_addproc_iwrite.c b/ompi/mca/sharedfp/addproc/sharedfp_addproc_iwrite.c index b846652db1..7539c94e08 100644 --- a/ompi/mca/sharedfp/addproc/sharedfp_addproc_iwrite.c +++ b/ompi/mca/sharedfp/addproc/sharedfp_addproc_iwrite.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) 2013 University of Houston. All rights reserved. + * Copyright (c) 2013-2015 University of Houston. All rights reserved. * $COPYRIGHT$ * * Additional copyrights may follow @@ -24,6 +24,7 @@ #include "mpi.h" #include "ompi/constants.h" #include "ompi/mca/sharedfp/sharedfp.h" +#include "ompi/mca/sharedfp/base/base.h" int mca_sharedfp_addproc_iwrite(mca_io_ompio_file_t *fh, void *buf, @@ -50,13 +51,15 @@ int mca_sharedfp_addproc_iwrite(mca_io_ompio_file_t *fh, sh = fh->f_sharedfp_data; if ( mca_sharedfp_addproc_verbose ){ - printf("sharedfp_addproc_iwrite: Bytes Requested is %ld\n",bytesRequested); + opal_output(ompi_sharedfp_base_framework.framework_output, + "sharedfp_addproc_iwrite: Bytes Requested is %ld\n",bytesRequested); } /* Request the offset to write bytesRequested bytes */ ret = mca_sharedfp_addproc_request_position(sh,bytesRequested,&offset); if ( OMPI_SUCCESS == ret ) { if ( mca_sharedfp_addproc_verbose ){ - printf("sharedfp_addproc_iwrite: Offset received is %lld\n",offset); + opal_output(ompi_sharedfp_base_framework.framework_output, + "sharedfp_addproc_iwrite: Offset received is %lld\n",offset); } /* Write to the file */ ret = ompio_io_ompio_file_iwrite_at(sh->sharedfh,offset,buf,count,datatype,request); @@ -70,8 +73,114 @@ int mca_sharedfp_addproc_write_ordered_begin(mca_io_ompio_file_t *fh, int count, struct ompi_datatype_t *datatype) { - opal_output(0,"mca_sharedfp_addproc_write_ordered_begin: NOT IMPLEMENTED\n"); - return OMPI_ERROR; + int ret = OMPI_SUCCESS; + OMPI_MPI_OFFSET_TYPE offset = 0, offsetReceived = 0; + long sendBuff = 0; + long *buff=NULL; + long offsetBuff; + long bytesRequested = 0; + int recvcnt = 1, sendcnt = 1; + size_t numofBytes; + int rank, size, i; + struct mca_sharedfp_base_data_t *sh = NULL; + + if ( NULL == fh->f_sharedfp_data){ + opal_output(0, "sharedfp_addproc_write_ordered_begin:" + " shared file pointer structure not initialized correctly\n"); + return OMPI_ERROR; + } + + if ( true == fh->f_split_coll_in_use ) { + opal_output(0, "Only one split collective I/O operation allowed per file handle " + "at any given point in time!\n"); + return MPI_ERR_REQUEST; + } + + /*Retrieve the shared file pointer structure*/ + sh = fh->f_sharedfp_data; + + /* Calculate the number of bytes to write*/ + opal_datatype_type_size ( &datatype->super, &numofBytes); + sendBuff = count * numofBytes; + + /* Get the ranks in the communicator */ + rank = ompi_comm_rank ( sh->comm ); + size = ompi_comm_size ( sh->comm ); + + if ( 0 == rank ) { + buff = (long*)malloc(sizeof(OMPI_MPI_OFFSET_TYPE) * size); + if ( NULL == buff ) + return OMPI_ERR_OUT_OF_RESOURCE; + } + + ret = sh->comm->c_coll.coll_gather ( &sendBuff, sendcnt, OMPI_OFFSET_DATATYPE, buff, + recvcnt, OMPI_OFFSET_DATATYPE, 0, sh->comm, + sh->comm->c_coll.coll_gather_module); + if( OMPI_SUCCESS != ret ){ + goto exit; + } + + /* All the counts are present now in the recvBuff. + The size of recvBuff is sizeof_newComm + */ + if ( 0 == rank ) { + for (i = 0; i < size ; i ++) { + bytesRequested += buff[i]; + + if ( mca_sharedfp_addproc_verbose ){ + opal_output(ompi_sharedfp_base_framework.framework_output, + "sharedfp_addproc_write_ordered_begin: Bytes requested are %ld\n", + bytesRequested); + } + } + + /* Request the offset to write bytesRequested bytes + ** only the root process needs to do the request, + ** since the root process will then tell the other + ** processes at what offset they should write their + ** share of the data. + */ + ret = mca_sharedfp_addproc_request_position(sh,bytesRequested,&offsetReceived); + if( OMPI_SUCCESS != ret ){ + goto exit; + } + if ( mca_sharedfp_addproc_verbose ){ + opal_output(ompi_sharedfp_base_framework.framework_output, + "sharedfp_addproc_write_ordered_begin: Offset received is %lld\n", + offsetReceived); + } + buff[0] += offsetReceived; + + for (i = 1 ; i < size; i++) { + buff[i] += buff[i-1]; + } + } + + /* Scatter the results to the other processes*/ + ret = sh->comm->c_coll.coll_scatter ( buff, sendcnt, OMPI_OFFSET_DATATYPE, &offsetBuff, + recvcnt, OMPI_OFFSET_DATATYPE, 0, sh->comm, + sh->comm->c_coll.coll_scatter_module ); + if( OMPI_SUCCESS != ret ){ + goto exit; + } + + /*Each process now has its own individual offset in recvBUFF*/ + offset = offsetBuff - sendBuff; + + if ( mca_sharedfp_addproc_verbose ){ + opal_output(ompi_sharedfp_base_framework.framework_output, + "sharedfp_addproc_write_ordered: Offset returned is %lld\n",offset); + } + + /* write to the file */ + ret = ompio_io_ompio_file_iwrite_at_all(sh->sharedfh,offset,buf,count,datatype,&fh->f_split_coll_req); + fh->f_split_coll_in_use = true; + +exit: + if ( NULL != buff ) { + free ( buff ); + } + return ret; } @@ -79,6 +188,10 @@ int mca_sharedfp_addproc_write_ordered_end(mca_io_ompio_file_t *fh, void *buf, ompi_status_public_t *status) { - opal_output(0,"mca_sharedfp_addproc_write_ordered_end: NOT IMPLEMENTED\n"); - return OMPI_ERROR; + int ret = OMPI_SUCCESS; + ret = ompi_request_wait ( &fh->f_split_coll_req, status ); + + /* remove the flag again */ + fh->f_split_coll_in_use = false; + return ret; } diff --git a/ompi/mca/sharedfp/addproc/sharedfp_addproc_read.c b/ompi/mca/sharedfp/addproc/sharedfp_addproc_read.c index 1e4c836e14..c20c449a4b 100644 --- a/ompi/mca/sharedfp/addproc/sharedfp_addproc_read.c +++ b/ompi/mca/sharedfp/addproc/sharedfp_addproc_read.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) 2013 University of Houston. All rights reserved. + * Copyright (c) 2013-2015 University of Houston. All rights reserved. * $COPYRIGHT$ * * Additional copyrights may follow @@ -24,6 +24,7 @@ #include "mpi.h" #include "ompi/constants.h" #include "ompi/mca/sharedfp/sharedfp.h" +#include "ompi/mca/sharedfp/base/base.h" int mca_sharedfp_addproc_read ( mca_io_ompio_file_t *fh, void *buf, int count, MPI_Datatype datatype, MPI_Status *status) @@ -33,23 +34,11 @@ int mca_sharedfp_addproc_read ( mca_io_ompio_file_t *fh, long bytesRequested = 0; size_t numofBytes; struct mca_sharedfp_base_data_t *sh = NULL; - mca_sharedfp_base_module_t * shared_fp_base_module = NULL; if(NULL == fh->f_sharedfp_data){ - if ( mca_sharedfp_addproc_verbose ) { - printf("sharedfp_addproc_read: opening the shared file pointer file\n"); - } - shared_fp_base_module = fh->f_sharedfp; - - ret = shared_fp_base_module->sharedfp_file_open(fh->f_comm, - fh->f_filename, - fh->f_amode, - fh->f_info, - fh); - if ( OMPI_SUCCESS != ret ) { - opal_output(0,"sharedfp_addproc_read - error opening the shared file pointer\n"); - return ret; - } + opal_output(0, "sharedfp_addproc_read: shared file pointer " + "structure not initialized correctly\n"); + return OMPI_ERROR; } /* Calculate the number of bytes to write */ @@ -57,7 +46,8 @@ int mca_sharedfp_addproc_read ( mca_io_ompio_file_t *fh, bytesRequested = count * numofBytes; if ( mca_sharedfp_addproc_verbose ){ - printf("mca_sharedfp_addproc_read: Bytes Requested is %ld\n", bytesRequested); + opal_output(ompi_sharedfp_base_framework.framework_output, + "mca_sharedfp_addproc_read: Bytes Requested is %ld\n", bytesRequested); } /* Retrieve the shared file data struct */ sh = fh->f_sharedfp_data; @@ -66,7 +56,8 @@ int mca_sharedfp_addproc_read ( mca_io_ompio_file_t *fh, ret = mca_sharedfp_addproc_request_position(sh,bytesRequested,&offset); if( OMPI_SUCCESS == ret ){ if ( mca_sharedfp_addproc_verbose ){ - printf("mca_sharedfp_addproc_read: Offset received is %lld\n",offset); + opal_output(ompi_sharedfp_base_framework.framework_output, + "mca_sharedfp_addproc_read: Offset received is %lld\n",offset); } /* Read from the file */ ret = ompio_io_ompio_file_read_at(sh->sharedfh,offset,buf,count,datatype,status); @@ -89,26 +80,13 @@ int mca_sharedfp_addproc_read_ordered (mca_io_ompio_file_t *fh, size_t numofBytes; int rank, size, i; struct mca_sharedfp_base_data_t *sh = NULL; - mca_sharedfp_base_module_t * shared_fp_base_module = NULL; if(NULL == fh->f_sharedfp_data){ - if ( mca_sharedfp_addproc_verbose ) { - printf("sharedfp_addproc_read_ordered: opening the shared file pointer file\n"); - } - shared_fp_base_module = fh->f_sharedfp; - - ret = shared_fp_base_module->sharedfp_file_open(fh->f_comm, - fh->f_filename, - fh->f_amode, - fh->f_info, - fh); - if ( OMPI_SUCCESS != ret ) { - opal_output(0,"sharedfp_addproc_read_ordered - error opening the shared file pointer\n"); - return ret; - } + opal_output(0, "sharedfp_addproc_read_ordered: shared file pointer " + "structure not initialized correctly\n"); + return OMPI_ERROR; } - /*Retrieve the new communicator*/ sh = fh->f_sharedfp_data; @@ -121,7 +99,7 @@ int mca_sharedfp_addproc_read_ordered (mca_io_ompio_file_t *fh, size = ompi_comm_size ( sh->comm); if ( 0 == rank ) { - buff = (long*)malloc(sizeof(long) * size); + buff = (long*)malloc(sizeof(OMPI_MPI_OFFSET_TYPE) * size); if ( NULL == buff ) return OMPI_ERR_OUT_OF_RESOURCE; } @@ -139,12 +117,15 @@ int mca_sharedfp_addproc_read_ordered (mca_io_ompio_file_t *fh, if ( 0 == rank ) { for (i = 0; i < size ; i ++) { if ( mca_sharedfp_addproc_verbose ){ - printf("sharedfp_addproc_read_ordered: Buff is %ld\n",buff[i]); + opal_output(ompi_sharedfp_base_framework.framework_output, + "sharedfp_addproc_read_ordered: Buff is %ld\n",buff[i]); } bytesRequested += buff[i]; if ( mca_sharedfp_addproc_verbose ){ - printf("sharedfp_addproc_read_ordered: Bytes requested are %ld\n",bytesRequested); + opal_output(ompi_sharedfp_base_framework.framework_output, + "sharedfp_addproc_read_ordered: Bytes requested are %ld\n", + bytesRequested); } } @@ -159,7 +140,9 @@ int mca_sharedfp_addproc_read_ordered (mca_io_ompio_file_t *fh, goto exit; } if ( mca_sharedfp_addproc_verbose ){ - printf("sharedfp_addproc_read_ordered: Offset received is %lld\n",offsetReceived); + opal_output(ompi_sharedfp_base_framework.framework_output, + "sharedfp_addproc_read_ordered: Offset received is %lld\n", + offsetReceived); } buff[0] += offsetReceived; @@ -181,7 +164,8 @@ int mca_sharedfp_addproc_read_ordered (mca_io_ompio_file_t *fh, offset = offsetBuff - sendBuff; if ( mca_sharedfp_addproc_verbose ){ - printf("sharedfp_addproc_read_ordered: Offset returned is %lld\n",offset); + opal_output(ompi_sharedfp_base_framework.framework_output, + "sharedfp_addproc_read_ordered: Offset returned is %lld\n",offset); } /* read from the file */ diff --git a/ompi/mca/sharedfp/addproc/sharedfp_addproc_write.c b/ompi/mca/sharedfp/addproc/sharedfp_addproc_write.c index 5f751021b8..02a9fa4589 100644 --- a/ompi/mca/sharedfp/addproc/sharedfp_addproc_write.c +++ b/ompi/mca/sharedfp/addproc/sharedfp_addproc_write.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) 2013 University of Houston. All rights reserved. + * Copyright (c) 2013-2015 University of Houston. All rights reserved. * $COPYRIGHT$ * * Additional copyrights may follow @@ -24,6 +24,7 @@ #include "mpi.h" #include "ompi/constants.h" #include "ompi/mca/sharedfp/sharedfp.h" +#include "ompi/mca/sharedfp/base/base.h" int mca_sharedfp_addproc_write (mca_io_ompio_file_t *fh, void *buf, @@ -36,23 +37,10 @@ int mca_sharedfp_addproc_write (mca_io_ompio_file_t *fh, long bytesRequested = 0; size_t numofBytes; struct mca_sharedfp_base_data_t *sh = NULL; - mca_sharedfp_base_module_t * shared_fp_base_module = NULL; if(NULL == fh->f_sharedfp_data){ - if ( mca_sharedfp_addproc_verbose ) { - printf("sharedfp_addproc_write: opening the shared file pointer file\n"); - } - shared_fp_base_module = fh->f_sharedfp; - - ret = shared_fp_base_module->sharedfp_file_open(fh->f_comm, - fh->f_filename, - fh->f_amode, - fh->f_info, - fh); - if ( OMPI_SUCCESS != ret ) { - opal_output(0,"sharedfp_addproc_write - error opening the shared file pointer\n"); - return ret; - } + opal_output(0, "sharedfp_addproc_write: shared file pointer structure not initialized correctly\n"); + return OMPI_ERROR; } /* Calculate the number of bytes to write*/ @@ -63,14 +51,17 @@ int mca_sharedfp_addproc_write (mca_io_ompio_file_t *fh, sh = fh->f_sharedfp_data; if ( mca_sharedfp_addproc_verbose ){ - printf("sharedfp_addproc_write: sharedfp_addproc_write: Bytes Requested is %ld\n",bytesRequested); + opal_output(ompi_sharedfp_base_framework.framework_output, + "sharedfp_addproc_write: sharedfp_addproc_write: Bytes Requested is %ld\n", + bytesRequested); } /*Request the offset to write bytesRequested bytes*/ ret = mca_sharedfp_addproc_request_position( sh, bytesRequested, &offset); if ( OMPI_SUCCESS == ret ) { if ( mca_sharedfp_addproc_verbose ){ - printf("sharedfp_addproc_write: Offset received is %lld\n",offset); + opal_output(ompi_sharedfp_base_framework.framework_output, + "sharedfp_addproc_write: Offset received is %lld\n",offset); } /* Write to the file */ ret = ompio_io_ompio_file_write_at(sh->sharedfh,offset,buf,count,datatype,status); @@ -95,23 +86,11 @@ int mca_sharedfp_addproc_write_ordered (mca_io_ompio_file_t *fh, size_t numofBytes; int rank, size, i; struct mca_sharedfp_base_data_t *sh = NULL; - mca_sharedfp_base_module_t * shared_fp_base_module = NULL; - if ( NULL == fh->f_sharedfp_data){ - if ( mca_sharedfp_addproc_verbose ) { - printf("sharedfp_addproc_write_ordered: opening the shared file pointer\n"); - } - shared_fp_base_module = fh->f_sharedfp; - - ret = shared_fp_base_module->sharedfp_file_open(fh->f_comm, - fh->f_filename, - fh->f_amode, - fh->f_info, - fh); - if ( OMPI_SUCCESS != ret ) { - opal_output(0,"sharedfp_addproc_write_ordered - error opening the shared file pointer\n"); - return ret; - } + if(NULL == fh->f_sharedfp_data){ + opal_output(0, "sharedfp_addproc_write_ordered: shared file pointer " + "structure not initialized correctly\n"); + return OMPI_ERROR; } /*Retrieve the shared file pointer structure*/ @@ -126,7 +105,7 @@ int mca_sharedfp_addproc_write_ordered (mca_io_ompio_file_t *fh, size = ompi_comm_size ( sh->comm ); if ( 0 == rank ) { - buff = (long*)malloc(sizeof(long) * size); + buff = (long*)malloc(sizeof(OMPI_MPI_OFFSET_TYPE) * size); if ( NULL == buff ) return OMPI_ERR_OUT_OF_RESOURCE; } @@ -146,7 +125,9 @@ int mca_sharedfp_addproc_write_ordered (mca_io_ompio_file_t *fh, bytesRequested += buff[i]; if ( mca_sharedfp_addproc_verbose ){ - printf("sharedfp_addproc_write_ordered: Bytes requested are %ld\n",bytesRequested); + opal_output(ompi_sharedfp_base_framework.framework_output, + "sharedfp_addproc_write_ordered: Bytes requested are %ld\n", + bytesRequested); } } @@ -161,7 +142,9 @@ int mca_sharedfp_addproc_write_ordered (mca_io_ompio_file_t *fh, goto exit; } if ( mca_sharedfp_addproc_verbose ){ - printf("sharedfp_addproc_write_ordered: Offset received is %lld\n",offsetReceived); + opal_output(ompi_sharedfp_base_framework.framework_output, + "sharedfp_addproc_write_ordered: Offset received is %lld\n", + offsetReceived); } buff[0] += offsetReceived; @@ -182,7 +165,9 @@ int mca_sharedfp_addproc_write_ordered (mca_io_ompio_file_t *fh, offset = offsetBuff - sendBuff; if ( mca_sharedfp_addproc_verbose ){ - printf("sharedfp_addproc_write_ordered: Offset returned is %lld\n",offset); + opal_output(ompi_sharedfp_base_framework.framework_output, + "sharedfp_addproc_write_ordered: Offset returned is %lld\n", + offset); } /* write to the file */