ports/openmpi
ports/openmpi
1
1
Форкнуть 0
Код Релизы Активность

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
Этот коммит содержится в:
Edgar Gabriel 2015-08-18 10:29:13 -05:00
родитель 145bac088d
Коммит 1b45712595
5 изменённых файлов: 307 добавлений и 102 удалений
Показать статистику Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

27
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 <sys/mman.h>
#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;
}

132
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;
}

127
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;
}

62
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 */

61
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 */