some code reorganization in preparation for non-blocking read and write
operations. This commit was SVN r32534.
Этот коммит содержится в:
родитель
a914c68356
Коммит
da1b6c2e87
@ -231,6 +231,11 @@ OMPI_DECLSPEC void mca_io_ompio_get_num_aggregators ( int *num_aggregators);
|
||||
OMPI_DECLSPEC void mca_io_ompio_get_bytes_per_agg ( int *bytes_per_agg);
|
||||
|
||||
|
||||
OMPI_DECLSPEC int mca_io_ompio_build_io_array ( mca_io_ompio_file_t *fh, int index, int cycles,
|
||||
size_t bpc, int max_data, uint32_t iov_count,
|
||||
struct iovec *decoded_iov, int *ii, int *jj,
|
||||
size_t *tbw );
|
||||
|
||||
OMPI_DECLSPEC int ompi_io_ompio_set_file_defaults (mca_io_ompio_file_t *fh);
|
||||
|
||||
OMPI_DECLSPEC int ompio_io_ompio_file_open (ompi_communicator_t *comm,
|
||||
@ -284,6 +289,12 @@ OMPI_DECLSPEC int ompio_io_ompio_file_iwrite (mca_io_ompio_file_t *fh,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_request_t **request);
|
||||
|
||||
OMPI_DECLSPEC int ompio_io_ompio_file_iread (mca_io_ompio_file_t *fh,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_request_t **request);
|
||||
|
||||
OMPI_DECLSPEC int ompio_io_ompio_file_read (mca_io_ompio_file_t *fh,
|
||||
void *buf,
|
||||
int count,
|
||||
|
@ -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-2013 University of Houston. All rights reserved.
|
||||
* Copyright (c) 2008-2014 University of Houston. All rights reserved.
|
||||
* $COPYRIGHT$
|
||||
*
|
||||
* Additional copyrights may follow
|
||||
@ -32,34 +32,45 @@
|
||||
#include "io_ompio.h"
|
||||
#include "math.h"
|
||||
|
||||
int
|
||||
mca_io_ompio_file_read (ompi_file_t *fp,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t *status)
|
||||
/* Read and write routines are split into two interfaces.
|
||||
** The
|
||||
** mca_io_ompio_file_read/write[_at]
|
||||
**
|
||||
** routines are the ones registered with the ompio modules.
|
||||
** The
|
||||
**
|
||||
** ompio_io_ompio_file_read/write[_at]
|
||||
**
|
||||
** routesin are used e.g. from the shared file pointer modules.
|
||||
** The main difference is, that the first one takes an ompi_file_t
|
||||
** as a file pointer argument, while the second uses the ompio internal
|
||||
** mca_io_ompio_file_t structure.
|
||||
*/
|
||||
|
||||
int mca_io_ompio_file_read (ompi_file_t *fp,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t *status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
|
||||
data = (mca_io_ompio_data_t *) fp->f_io_selected_data;
|
||||
|
||||
ret = ompio_io_ompio_file_read(&data->ompio_fh,buf,count,datatype,status);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
ompio_io_ompio_file_read (mca_io_ompio_file_t *fh,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t *status)
|
||||
int ompio_io_ompio_file_read (mca_io_ompio_file_t *fh,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t *status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
|
||||
size_t total_bytes_read = 0; /* total bytes that have been read*/
|
||||
size_t bytes_to_read_in_cycle = 0; /* left to be read in a cycle*/
|
||||
size_t bytes_per_cycle = 0; /* total read in each cycle by each process*/
|
||||
int index = 0;
|
||||
int cycles = 0;
|
||||
@ -70,9 +81,6 @@ ompio_io_ompio_file_read (mca_io_ompio_file_t *fh,
|
||||
size_t max_data = 0;
|
||||
int i = 0; /* index into the decoded iovec of the buffer */
|
||||
int j = 0; /* index into the file vie iovec */
|
||||
int k = 0; /* index into the io_array */
|
||||
size_t sum_previous_counts = 0;
|
||||
size_t sum_previous_length = 0;
|
||||
|
||||
if ( 0 == count ) {
|
||||
if ( MPI_STATUS_IGNORE != status ) {
|
||||
@ -102,120 +110,21 @@ ompio_io_ompio_file_read (mca_io_ompio_file_t *fh,
|
||||
printf ("Bytes per Cycle: %d Cycles: %d\n",bytes_per_cycle, cycles);
|
||||
#endif
|
||||
|
||||
sum_previous_length = fh->f_position_in_file_view;
|
||||
j = fh->f_index_in_file_view;
|
||||
|
||||
for (index = 0; index < cycles; index++) {
|
||||
OPAL_PTRDIFF_TYPE disp;
|
||||
int block = 1;
|
||||
k = 0;
|
||||
if ((index == cycles-1) && (max_data % bytes_per_cycle)) {
|
||||
bytes_to_read_in_cycle = max_data % bytes_per_cycle;
|
||||
}
|
||||
else {
|
||||
bytes_to_read_in_cycle = bytes_per_cycle;
|
||||
}
|
||||
|
||||
/*
|
||||
ompi_io_ompio_create_list (fh->f_decoded_iov, fh->f_iov_count,
|
||||
decoded_iov, iov_count,
|
||||
&total_bytes_read, &bytes_to_read_in_cycle,
|
||||
&sum_previous_counts, &sum_previous_length,
|
||||
&decoded_iov_index, &fview_iov_index,
|
||||
&fh->f_io_array, &fh->f_num_of_io_entries);
|
||||
*/
|
||||
|
||||
fh->f_io_array = (mca_io_ompio_io_array_t *)malloc
|
||||
(OMPIO_IOVEC_INITIAL_SIZE * sizeof (mca_io_ompio_io_array_t));
|
||||
if (NULL == fh->f_io_array) {
|
||||
opal_output(1, "OUT OF MEMORY\n");
|
||||
return OMPI_ERR_OUT_OF_RESOURCE;
|
||||
}
|
||||
|
||||
while (bytes_to_read_in_cycle) {
|
||||
/* reallocate if needed */
|
||||
if (OMPIO_IOVEC_INITIAL_SIZE*block <= k) {
|
||||
block ++;
|
||||
fh->f_io_array = (mca_io_ompio_io_array_t *)realloc
|
||||
(fh->f_io_array, OMPIO_IOVEC_INITIAL_SIZE *
|
||||
block * sizeof (mca_io_ompio_io_array_t));
|
||||
if (NULL == fh->f_io_array) {
|
||||
opal_output(1, "OUT OF MEMORY\n");
|
||||
return OMPI_ERR_OUT_OF_RESOURCE;
|
||||
}
|
||||
}
|
||||
|
||||
if (decoded_iov[i].iov_len -
|
||||
(total_bytes_read - sum_previous_counts) <= 0) {
|
||||
sum_previous_counts += decoded_iov[i].iov_len;
|
||||
i = i + 1;
|
||||
}
|
||||
|
||||
disp = (OPAL_PTRDIFF_TYPE)decoded_iov[i].iov_base +
|
||||
(total_bytes_read - sum_previous_counts);
|
||||
fh->f_io_array[k].memory_address = (IOVBASE_TYPE *)disp;
|
||||
|
||||
if (decoded_iov[i].iov_len -
|
||||
(total_bytes_read - sum_previous_counts) >=
|
||||
bytes_to_read_in_cycle) {
|
||||
fh->f_io_array[k].length = bytes_to_read_in_cycle;
|
||||
}
|
||||
else {
|
||||
fh->f_io_array[k].length = decoded_iov[i].iov_len -
|
||||
(total_bytes_read - sum_previous_counts);
|
||||
}
|
||||
if (! (fh->f_flags & OMPIO_CONTIGUOUS_FVIEW)) {
|
||||
if (fh->f_decoded_iov[j].iov_len -
|
||||
(fh->f_total_bytes - sum_previous_length) <= 0) {
|
||||
sum_previous_length += fh->f_decoded_iov[j].iov_len;
|
||||
j = j + 1;
|
||||
if (j == (int)fh->f_iov_count) {
|
||||
j = 0;
|
||||
sum_previous_length = 0;
|
||||
fh->f_offset += fh->f_view_extent;
|
||||
fh->f_position_in_file_view = sum_previous_length;
|
||||
fh->f_index_in_file_view = j;
|
||||
fh->f_total_bytes = 0;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
disp = (OPAL_PTRDIFF_TYPE)fh->f_decoded_iov[j].iov_base +
|
||||
(fh->f_total_bytes - sum_previous_length);
|
||||
fh->f_io_array[k].offset = (IOVBASE_TYPE *)(intptr_t)(disp + fh->f_offset);
|
||||
|
||||
if (! (fh->f_flags & OMPIO_CONTIGUOUS_FVIEW)) {
|
||||
if (fh->f_decoded_iov[j].iov_len -
|
||||
(fh->f_total_bytes - sum_previous_length)
|
||||
< fh->f_io_array[k].length) {
|
||||
fh->f_io_array[k].length = fh->f_decoded_iov[j].iov_len -
|
||||
(fh->f_total_bytes - sum_previous_length);
|
||||
}
|
||||
}
|
||||
|
||||
total_bytes_read += fh->f_io_array[k].length;
|
||||
fh->f_total_bytes += fh->f_io_array[k].length;
|
||||
bytes_to_read_in_cycle -= fh->f_io_array[k].length;
|
||||
k = k + 1;
|
||||
}
|
||||
fh->f_position_in_file_view = sum_previous_length;
|
||||
fh->f_index_in_file_view = j;
|
||||
fh->f_num_of_io_entries = k;
|
||||
|
||||
#if 0
|
||||
if (fh->f_rank == 0) {
|
||||
int i;
|
||||
printf("*************************** %d\n", fh->f_num_of_io_entries);
|
||||
|
||||
for (i=0 ; i<fh->f_num_of_io_entries ; i++) {
|
||||
printf(" ADDRESS: %p OFFSET: %p LENGTH: %d\n",
|
||||
fh->f_io_array[i].memory_address,
|
||||
fh->f_io_array[i].offset,
|
||||
fh->f_io_array[i].length);
|
||||
}
|
||||
}
|
||||
#endif
|
||||
|
||||
mca_io_ompio_build_io_array ( fh,
|
||||
index,
|
||||
cycles,
|
||||
bytes_per_cycle,
|
||||
max_data,
|
||||
iov_count,
|
||||
decoded_iov,
|
||||
&i,
|
||||
&j,
|
||||
&total_bytes_read);
|
||||
|
||||
if (fh->f_num_of_io_entries) {
|
||||
fh->f_fbtl->fbtl_preadv (fh, NULL);
|
||||
}
|
||||
@ -239,50 +148,174 @@ ompio_io_ompio_file_read (mca_io_ompio_file_t *fh,
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
mca_io_ompio_file_read_at (ompi_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t * status)
|
||||
int mca_io_ompio_file_read_at (ompi_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t * status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
|
||||
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
||||
|
||||
ret = ompio_io_ompio_file_read_at(&data->ompio_fh, offset,buf,count,datatype,status);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
ompio_io_ompio_file_read_at (mca_io_ompio_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t * status)
|
||||
int ompio_io_ompio_file_read_at (mca_io_ompio_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t * status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
|
||||
ompi_io_ompio_set_explicit_offset (fh, offset);
|
||||
|
||||
ompio_io_ompio_file_read (fh,
|
||||
buf,
|
||||
count,
|
||||
datatype,
|
||||
status);
|
||||
ret = ompio_io_ompio_file_read (fh,
|
||||
buf,
|
||||
count,
|
||||
datatype,
|
||||
status);
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
mca_io_ompio_file_read_all (ompi_file_t *fh,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t * status)
|
||||
|
||||
int mca_io_ompio_file_iread (ompi_file_t *fh,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_request_t **request)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
|
||||
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
||||
ret = ompio_io_ompio_file_iread(&data->ompio_fh,buf,count,datatype,request);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
|
||||
int ompio_io_ompio_file_iread (mca_io_ompio_file_t *fh,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_request_t **request)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
|
||||
if ( NULL != fh->f_fbtl->fbtl_ipreadv ) {
|
||||
// This fbtl has support for non-blocking operations
|
||||
|
||||
size_t total_bytes_read = 0; /* total bytes that have been read*/
|
||||
uint32_t iov_count = 0;
|
||||
struct iovec *decoded_iov = NULL;
|
||||
|
||||
size_t max_data = 0;
|
||||
int i = 0; /* index into the decoded iovec of the buffer */
|
||||
int j = 0; /* index into the file vie iovec */
|
||||
|
||||
if ( 0 == count ) {
|
||||
// if ( MPI_STATUS_IGNORE != status ) {
|
||||
// status->_ucount = 0;
|
||||
// }
|
||||
// return ret;
|
||||
}
|
||||
|
||||
ompi_io_ompio_decode_datatype (fh,
|
||||
datatype,
|
||||
count,
|
||||
buf,
|
||||
&max_data,
|
||||
&decoded_iov,
|
||||
&iov_count);
|
||||
|
||||
// Non-blocking operations have to occur in a single cycle
|
||||
j = fh->f_index_in_file_view;
|
||||
|
||||
mca_io_ompio_build_io_array ( fh,
|
||||
0, // index
|
||||
1, // no. of cyces
|
||||
max_data, // setting bytes per cycle to match data
|
||||
max_data,
|
||||
iov_count,
|
||||
decoded_iov,
|
||||
&i,
|
||||
&j,
|
||||
&total_bytes_read);
|
||||
|
||||
if (fh->f_num_of_io_entries) {
|
||||
fh->f_fbtl->fbtl_ipreadv (fh, NULL, request);
|
||||
}
|
||||
|
||||
fh->f_num_of_io_entries = 0;
|
||||
if (NULL != fh->f_io_array) {
|
||||
free (fh->f_io_array);
|
||||
fh->f_io_array = NULL;
|
||||
}
|
||||
|
||||
if (NULL != decoded_iov) {
|
||||
free (decoded_iov);
|
||||
decoded_iov = NULL;
|
||||
}
|
||||
|
||||
// if ( MPI_STATUS_IGNORE != status ) {
|
||||
// status->_ucount = max_data;
|
||||
// }
|
||||
}
|
||||
else {
|
||||
// This fbtl does not support non-blocking operations
|
||||
ompio_io_ompio_file_read (fh, buf, count, datatype, MPI_STATUS_IGNORE);
|
||||
}
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int mca_io_ompio_file_iread_at (ompi_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_request_t **request)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
|
||||
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
||||
ret = ompio_io_ompio_file_iread_at(&data->ompio_fh,offset,buf,count,datatype,request);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int ompio_io_ompio_file_iread_at (mca_io_ompio_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_request_t **request)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
|
||||
ompi_io_ompio_set_explicit_offset (fh, offset);
|
||||
ret = ompio_io_ompio_file_iread (fh,
|
||||
buf,
|
||||
count,
|
||||
datatype,
|
||||
request);
|
||||
return ret;
|
||||
}
|
||||
|
||||
|
||||
/* Infrastructure for collective operations */
|
||||
/******************************************************/
|
||||
int mca_io_ompio_file_read_all (ompi_file_t *fh,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t * status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
@ -305,74 +338,33 @@ mca_io_ompio_file_read_all (ompi_file_t *fh,
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
mca_io_ompio_file_read_all_begin (ompi_file_t *fh,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype)
|
||||
|
||||
int mca_io_ompio_file_read_at_all (ompi_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t * status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
|
||||
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
||||
|
||||
ret = data->ompio_fh.
|
||||
f_fcoll->fcoll_file_read_all_begin (&data->ompio_fh,
|
||||
buf,
|
||||
count,
|
||||
datatype);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
mca_io_ompio_file_read_all_end (ompi_file_t *fh,
|
||||
void *buf,
|
||||
ompi_status_public_t * status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
|
||||
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
||||
|
||||
ret = data->ompio_fh.
|
||||
f_fcoll->fcoll_file_read_all_end (&data->ompio_fh,
|
||||
buf,
|
||||
status);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
mca_io_ompio_file_read_at_all (ompi_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t * status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
|
||||
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
||||
|
||||
ret = ompio_io_ompio_file_read_at_all(&data->ompio_fh,offset,buf,count,datatype,status);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
ompio_io_ompio_file_read_at_all (mca_io_ompio_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t * status)
|
||||
int ompio_io_ompio_file_read_at_all (mca_io_ompio_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t * status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
|
||||
ompi_io_ompio_set_explicit_offset (fh, offset);
|
||||
|
||||
ret = fh->f_fcoll->fcoll_file_read_all (fh,
|
||||
buf,
|
||||
count,
|
||||
@ -380,116 +372,16 @@ ompio_io_ompio_file_read_at_all (mca_io_ompio_file_t *fh,
|
||||
status);
|
||||
return ret;
|
||||
}
|
||||
int
|
||||
mca_io_ompio_file_read_at_all_begin (ompi_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
|
||||
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
||||
|
||||
ret = ompio_io_ompio_file_read_at_all_begin(&data->ompio_fh,offset,buf,count,datatype);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
ompio_io_ompio_file_read_at_all_begin (mca_io_ompio_file_t *ompio_fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
|
||||
ompi_io_ompio_set_explicit_offset (ompio_fh, offset);
|
||||
|
||||
ret = ompio_fh->f_fcoll->fcoll_file_read_all_begin (ompio_fh,
|
||||
buf,
|
||||
count,
|
||||
datatype);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
mca_io_ompio_file_read_at_all_end (ompi_file_t *fh,
|
||||
void *buf,
|
||||
ompi_status_public_t * status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
|
||||
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
||||
|
||||
ret = ompio_io_ompio_file_read_at_all_end(&data->ompio_fh,buf,status);
|
||||
|
||||
return ret;
|
||||
}
|
||||
int
|
||||
ompio_io_ompio_file_read_at_all_end (mca_io_ompio_file_t *ompio_fh,
|
||||
void *buf,
|
||||
ompi_status_public_t * status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
|
||||
ret = ompio_fh->f_fcoll->fcoll_file_read_all_end (ompio_fh,
|
||||
buf,
|
||||
status);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
mca_io_ompio_file_iread (ompi_file_t *fh,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_request_t **request)
|
||||
{
|
||||
int ret = MPI_ERR_OTHER;
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
mca_io_ompio_file_iread_at (ompi_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_request_t **request)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
|
||||
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
||||
ret = ompio_io_ompio_file_iread_at(&data->ompio_fh,offset,buf,count,datatype,request);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
ompio_io_ompio_file_iread_at (mca_io_ompio_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_request_t **request)
|
||||
{
|
||||
int ret = MPI_ERR_OTHER;
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
mca_io_ompio_file_read_shared (ompi_file_t *fp,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t * status)
|
||||
/* Infrastructure for shared file pointer operations
|
||||
** (individual and ordered)*/
|
||||
/******************************************************/
|
||||
int mca_io_ompio_file_read_shared (ompi_file_t *fp,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t * status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
@ -506,12 +398,11 @@ mca_io_ompio_file_read_shared (ompi_file_t *fp,
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
mca_io_ompio_file_iread_shared (ompi_file_t *fh,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_request_t **request)
|
||||
int mca_io_ompio_file_iread_shared (ompi_file_t *fh,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_request_t **request)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
@ -528,12 +419,11 @@ mca_io_ompio_file_iread_shared (ompi_file_t *fh,
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
mca_io_ompio_file_read_ordered (ompi_file_t *fh,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t * status)
|
||||
int mca_io_ompio_file_read_ordered (ompi_file_t *fh,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t * status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
@ -550,11 +440,10 @@ mca_io_ompio_file_read_ordered (ompi_file_t *fh,
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
mca_io_ompio_file_read_ordered_begin (ompi_file_t *fh,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype)
|
||||
int mca_io_ompio_file_read_ordered_begin (ompi_file_t *fh,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
@ -571,10 +460,9 @@ mca_io_ompio_file_read_ordered_begin (ompi_file_t *fh,
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
mca_io_ompio_file_read_ordered_end (ompi_file_t *fh,
|
||||
void *buf,
|
||||
ompi_status_public_t * status)
|
||||
int mca_io_ompio_file_read_ordered_end (ompi_file_t *fh,
|
||||
void *buf,
|
||||
ompi_status_public_t * status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
@ -590,3 +478,100 @@ mca_io_ompio_file_read_ordered_end (ompi_file_t *fh,
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
|
||||
/* Split collectives . Not really used but infrastructure is in place */
|
||||
/**********************************************************************/
|
||||
int mca_io_ompio_file_read_all_begin (ompi_file_t *fh,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
|
||||
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
||||
|
||||
ret = data->ompio_fh.
|
||||
f_fcoll->fcoll_file_read_all_begin (&data->ompio_fh,
|
||||
buf,
|
||||
count,
|
||||
datatype);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int mca_io_ompio_file_read_all_end (ompi_file_t *fh,
|
||||
void *buf,
|
||||
ompi_status_public_t * status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
|
||||
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
||||
|
||||
ret = data->ompio_fh.
|
||||
f_fcoll->fcoll_file_read_all_end (&data->ompio_fh,
|
||||
buf,
|
||||
status);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int mca_io_ompio_file_read_at_all_begin (ompi_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
|
||||
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
||||
ret = ompio_io_ompio_file_read_at_all_begin(&data->ompio_fh,offset,buf,count,datatype);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int ompio_io_ompio_file_read_at_all_begin (mca_io_ompio_file_t *ompio_fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
|
||||
ompi_io_ompio_set_explicit_offset (ompio_fh, offset);
|
||||
ret = ompio_fh->f_fcoll->fcoll_file_read_all_begin (ompio_fh,
|
||||
buf,
|
||||
count,
|
||||
datatype);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int mca_io_ompio_file_read_at_all_end (ompi_file_t *fh,
|
||||
void *buf,
|
||||
ompi_status_public_t * status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
|
||||
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
||||
ret = ompio_io_ompio_file_read_at_all_end(&data->ompio_fh,buf,status);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int ompio_io_ompio_file_read_at_all_end (mca_io_ompio_file_t *ompio_fh,
|
||||
void *buf,
|
||||
ompi_status_public_t * status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
|
||||
ret = ompio_fh->f_fcoll->fcoll_file_read_all_end (ompio_fh,
|
||||
buf,
|
||||
status);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
@ -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-2013 University of Houston. All rights reserved.
|
||||
* Copyright (c) 2008-2014 University of Houston. All rights reserved.
|
||||
* $COPYRIGHT$
|
||||
*
|
||||
* Additional copyrights may follow
|
||||
@ -35,12 +35,27 @@
|
||||
#include "math.h"
|
||||
#include <unistd.h>
|
||||
|
||||
int
|
||||
mca_io_ompio_file_write (ompi_file_t *fp,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t *status)
|
||||
/* Read and write routines are split into two interfaces.
|
||||
** The
|
||||
** mca_io_ompio_file_read/write[_at]
|
||||
**
|
||||
** routines are the ones registered with the ompio modules.
|
||||
** The
|
||||
**
|
||||
** ompio_io_ompio_file_read/write[_at]
|
||||
**
|
||||
** routesin are used e.g. from the shared file pointer modules.
|
||||
** The main difference is, that the first one takes an ompi_file_t
|
||||
** as a file pointer argument, while the second uses the ompio internal
|
||||
** mca_io_ompio_file_t structure.
|
||||
*/
|
||||
|
||||
|
||||
int mca_io_ompio_file_write (ompi_file_t *fp,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t *status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
@ -53,30 +68,23 @@ mca_io_ompio_file_write (ompi_file_t *fp,
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
ompio_io_ompio_file_write (mca_io_ompio_file_t *fh,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t *status)
|
||||
int ompio_io_ompio_file_write (mca_io_ompio_file_t *fh,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t *status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
|
||||
size_t total_bytes_written = 0; /* total bytes that have been written*/
|
||||
size_t bytes_to_write_in_cycle = 0; /* left to be written in a cycle*/
|
||||
size_t bytes_per_cycle = 0; /* total written in each cycle by each process*/
|
||||
int index = 0;
|
||||
int cycles = 0;
|
||||
|
||||
uint32_t iov_count = 0;
|
||||
struct iovec *decoded_iov = NULL;
|
||||
|
||||
size_t bytes_per_cycle=0;
|
||||
size_t total_bytes_written = 0;
|
||||
size_t max_data = 0;
|
||||
int i = 0; /* index into the decoded iovec of the buffer */
|
||||
int j = 0; /* index into the file vie iovec */
|
||||
int k = 0; /* index into the io_array */
|
||||
size_t sum_previous_counts = 0;
|
||||
size_t sum_previous_length = 0;
|
||||
int j = 0; /* index into the file view iovec */
|
||||
|
||||
if ( 0 == count ) {
|
||||
if ( MPI_STATUS_IGNORE != status ) {
|
||||
@ -97,115 +105,21 @@ ompio_io_ompio_file_write (mca_io_ompio_file_t *fh,
|
||||
cycles = ceil((float)max_data/bytes_per_cycle);
|
||||
|
||||
#if 0
|
||||
printf ("Bytes per Cycle: %d Cycles: %d\n",bytes_per_cycle, cycles);
|
||||
printf ("Bytes per Cycle: %d Cycles: %d\n", bytes_per_cycle, cycles);
|
||||
#endif
|
||||
|
||||
sum_previous_length = fh->f_position_in_file_view;
|
||||
j = fh->f_index_in_file_view;
|
||||
|
||||
for (index = 0; index < cycles; index++) {
|
||||
OPAL_PTRDIFF_TYPE disp;
|
||||
int block = 1;
|
||||
|
||||
k = 0;
|
||||
if ((index == cycles-1) && (max_data % bytes_per_cycle)) {
|
||||
bytes_to_write_in_cycle = max_data % bytes_per_cycle;
|
||||
}
|
||||
else {
|
||||
bytes_to_write_in_cycle = bytes_per_cycle;
|
||||
}
|
||||
|
||||
fh->f_io_array = (mca_io_ompio_io_array_t *)malloc
|
||||
(OMPIO_IOVEC_INITIAL_SIZE * sizeof (mca_io_ompio_io_array_t));
|
||||
if (NULL == fh->f_io_array) {
|
||||
opal_output(1, "OUT OF MEMORY\n");
|
||||
return OMPI_ERR_OUT_OF_RESOURCE;
|
||||
}
|
||||
|
||||
while (bytes_to_write_in_cycle) {
|
||||
/* reallocate if needed */
|
||||
if (OMPIO_IOVEC_INITIAL_SIZE*block <= k) {
|
||||
block ++;
|
||||
fh->f_io_array = (mca_io_ompio_io_array_t *)realloc
|
||||
(fh->f_io_array, OMPIO_IOVEC_INITIAL_SIZE *
|
||||
block * sizeof (mca_io_ompio_io_array_t));
|
||||
if (NULL == fh->f_io_array) {
|
||||
opal_output(1, "OUT OF MEMORY\n");
|
||||
return OMPI_ERR_OUT_OF_RESOURCE;
|
||||
}
|
||||
}
|
||||
|
||||
if (decoded_iov[i].iov_len -
|
||||
(total_bytes_written - sum_previous_counts) <= 0) {
|
||||
sum_previous_counts += decoded_iov[i].iov_len;
|
||||
i = i + 1;
|
||||
}
|
||||
|
||||
disp = (OPAL_PTRDIFF_TYPE)decoded_iov[i].iov_base +
|
||||
(total_bytes_written - sum_previous_counts);
|
||||
fh->f_io_array[k].memory_address = (IOVBASE_TYPE *)disp;
|
||||
|
||||
if (decoded_iov[i].iov_len -
|
||||
(total_bytes_written - sum_previous_counts) >=
|
||||
bytes_to_write_in_cycle) {
|
||||
fh->f_io_array[k].length = bytes_to_write_in_cycle;
|
||||
}
|
||||
else {
|
||||
fh->f_io_array[k].length = decoded_iov[i].iov_len -
|
||||
(total_bytes_written - sum_previous_counts);
|
||||
}
|
||||
|
||||
if (! (fh->f_flags & OMPIO_CONTIGUOUS_FVIEW)) {
|
||||
if (fh->f_decoded_iov[j].iov_len -
|
||||
(fh->f_total_bytes - sum_previous_length) <= 0) {
|
||||
sum_previous_length += fh->f_decoded_iov[j].iov_len;
|
||||
j = j + 1;
|
||||
if (j == (int)fh->f_iov_count) {
|
||||
j = 0;
|
||||
sum_previous_length = 0;
|
||||
fh->f_offset += fh->f_view_extent;
|
||||
fh->f_position_in_file_view = sum_previous_length;
|
||||
fh->f_index_in_file_view = j;
|
||||
fh->f_total_bytes = 0;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
disp = (OPAL_PTRDIFF_TYPE)fh->f_decoded_iov[j].iov_base +
|
||||
(fh->f_total_bytes - sum_previous_length);
|
||||
fh->f_io_array[k].offset = (IOVBASE_TYPE *)(intptr_t)(disp + fh->f_offset);
|
||||
|
||||
if (! (fh->f_flags & OMPIO_CONTIGUOUS_FVIEW)) {
|
||||
if (fh->f_decoded_iov[j].iov_len -
|
||||
(fh->f_total_bytes - sum_previous_length)
|
||||
< fh->f_io_array[k].length) {
|
||||
fh->f_io_array[k].length = fh->f_decoded_iov[j].iov_len -
|
||||
(fh->f_total_bytes - sum_previous_length);
|
||||
}
|
||||
}
|
||||
|
||||
total_bytes_written += fh->f_io_array[k].length;
|
||||
fh->f_total_bytes += fh->f_io_array[k].length;
|
||||
bytes_to_write_in_cycle -= fh->f_io_array[k].length;
|
||||
k = k + 1;
|
||||
}
|
||||
fh->f_position_in_file_view = sum_previous_length;
|
||||
fh->f_index_in_file_view = j;
|
||||
fh->f_num_of_io_entries = k;
|
||||
|
||||
#if 0
|
||||
if (fh->f_rank == 0) {
|
||||
int d;
|
||||
printf("*************************** %d\n", fh->f_num_of_io_entries);
|
||||
|
||||
for (d=0 ; d<fh->f_num_of_io_entries ; d++) {
|
||||
printf(" ADDRESS: %p OFFSET: %p LENGTH: %d\n",
|
||||
fh->f_io_array[d].memory_address,
|
||||
fh->f_io_array[d].offset,
|
||||
fh->f_io_array[d].length);
|
||||
}
|
||||
}
|
||||
#endif
|
||||
mca_io_ompio_build_io_array ( fh,
|
||||
index,
|
||||
cycles,
|
||||
bytes_per_cycle,
|
||||
max_data,
|
||||
iov_count,
|
||||
decoded_iov,
|
||||
&i,
|
||||
&j,
|
||||
&total_bytes_written);
|
||||
|
||||
if (fh->f_num_of_io_entries) {
|
||||
fh->f_fbtl->fbtl_pwritev (fh, NULL);
|
||||
@ -230,34 +144,32 @@ ompio_io_ompio_file_write (mca_io_ompio_file_t *fh,
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
mca_io_ompio_file_write_at (ompi_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t *status)
|
||||
int mca_io_ompio_file_write_at (ompi_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t *status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
|
||||
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
||||
|
||||
ret = ompio_io_ompio_file_write_at (&data->ompio_fh, offset,buf,count,datatype,status);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
ompio_io_ompio_file_write_at (mca_io_ompio_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t *status)
|
||||
int ompio_io_ompio_file_write_at (mca_io_ompio_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t *status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
ompi_io_ompio_set_explicit_offset (fh, offset);
|
||||
|
||||
ompi_io_ompio_set_explicit_offset (fh, offset);
|
||||
ret = ompio_io_ompio_file_write (fh,
|
||||
buf,
|
||||
count,
|
||||
@ -267,12 +179,260 @@ ompio_io_ompio_file_write_at (mca_io_ompio_file_t *fh,
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
mca_io_ompio_file_write_all (ompi_file_t *fh,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t *status)
|
||||
int mca_io_ompio_file_iwrite (ompi_file_t *fp,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_request_t **request)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
|
||||
data = (mca_io_ompio_data_t *) fp->f_io_selected_data;
|
||||
ret = ompio_io_ompio_file_iwrite(&data->ompio_fh,buf,count,datatype,request);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int ompio_io_ompio_file_iwrite (mca_io_ompio_file_t *fh,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_request_t **request)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
|
||||
if ( NULL != fh->f_fbtl->fbtl_ipwritev ) {
|
||||
// This fbtl has support for non-blocking operations
|
||||
|
||||
uint32_t iov_count = 0;
|
||||
struct iovec *decoded_iov = NULL;
|
||||
size_t max_data = 0;
|
||||
size_t total_bytes_written =0;
|
||||
int i = 0; /* index into the decoded iovec of the buffer */
|
||||
int j = 0; /* index into the file vie iovec */
|
||||
|
||||
if ( 0 == count ) {
|
||||
// if ( MPI_STATUS_IGNORE != status ) {
|
||||
// status->_ucount = 0;
|
||||
// }
|
||||
return ret;
|
||||
}
|
||||
|
||||
ompi_io_ompio_decode_datatype (fh,
|
||||
datatype,
|
||||
count,
|
||||
buf,
|
||||
&max_data,
|
||||
&decoded_iov,
|
||||
&iov_count);
|
||||
j = fh->f_index_in_file_view;
|
||||
|
||||
/* Non blocking operations have to occur in a single cycle */
|
||||
mca_io_ompio_build_io_array ( fh,
|
||||
0, // index of current cycle iteration
|
||||
1, // number of cycles
|
||||
max_data, // setting bytes_per_cycle to max_data
|
||||
max_data,
|
||||
iov_count,
|
||||
decoded_iov,
|
||||
&i,
|
||||
&j,
|
||||
&total_bytes_written);
|
||||
|
||||
if (fh->f_num_of_io_entries) {
|
||||
fh->f_fbtl->fbtl_ipwritev (fh, NULL, request);
|
||||
}
|
||||
|
||||
fh->f_num_of_io_entries = 0;
|
||||
if (NULL != fh->f_io_array) {
|
||||
free (fh->f_io_array);
|
||||
fh->f_io_array = NULL;
|
||||
}
|
||||
if (NULL != decoded_iov) {
|
||||
free (decoded_iov);
|
||||
decoded_iov = NULL;
|
||||
}
|
||||
|
||||
//if ( MPI_STATUS_IGNORE != status ) {
|
||||
// status->_ucount = max_data;
|
||||
// }
|
||||
}
|
||||
else {
|
||||
// This fbtl does not support non-blocking write operations
|
||||
ret = ompio_io_ompio_file_write(fh,buf,count,datatype,MPI_STATUS_IGNORE);
|
||||
|
||||
}
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int mca_io_ompio_file_iwrite_at (ompi_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_request_t **request)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
|
||||
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
||||
ret = ompio_io_ompio_file_iwrite_at(&data->ompio_fh,offset,buf,count,datatype,request);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int ompio_io_ompio_file_iwrite_at (mca_io_ompio_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_request_t **request)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
|
||||
ompi_io_ompio_set_explicit_offset (fh, offset);
|
||||
ret = ompio_io_ompio_file_iwrite (fh,
|
||||
buf,
|
||||
count,
|
||||
datatype,
|
||||
request);
|
||||
return ret;
|
||||
}
|
||||
|
||||
/* Helper function used by both read and write operations */
|
||||
/**************************************************************/
|
||||
|
||||
int mca_io_ompio_build_io_array ( mca_io_ompio_file_t *fh, int index, int cycles,
|
||||
size_t bytes_per_cycle, int max_data, uint32_t iov_count,
|
||||
struct iovec *decoded_iov, int *ii, int *jj, size_t *tbw )
|
||||
{
|
||||
OPAL_PTRDIFF_TYPE disp;
|
||||
int block = 1;
|
||||
size_t total_bytes_written = *tbw; /* total bytes that have been written*/
|
||||
size_t bytes_to_write_in_cycle = 0; /* left to be written in a cycle*/
|
||||
size_t sum_previous_counts = 0;
|
||||
size_t sum_previous_length = 0;
|
||||
int k = 0; /* index into the io_array */
|
||||
int i = *ii;
|
||||
int j = *jj;
|
||||
|
||||
sum_previous_length = fh->f_position_in_file_view;
|
||||
|
||||
if ((index == cycles-1) && (max_data % bytes_per_cycle)) {
|
||||
bytes_to_write_in_cycle = max_data % bytes_per_cycle;
|
||||
}
|
||||
else {
|
||||
bytes_to_write_in_cycle = bytes_per_cycle;
|
||||
}
|
||||
|
||||
fh->f_io_array = (mca_io_ompio_io_array_t *)malloc
|
||||
(OMPIO_IOVEC_INITIAL_SIZE * sizeof (mca_io_ompio_io_array_t));
|
||||
if (NULL == fh->f_io_array) {
|
||||
opal_output(1, "OUT OF MEMORY\n");
|
||||
return OMPI_ERR_OUT_OF_RESOURCE;
|
||||
}
|
||||
|
||||
while (bytes_to_write_in_cycle) {
|
||||
/* reallocate if needed */
|
||||
if (OMPIO_IOVEC_INITIAL_SIZE*block <= k) {
|
||||
block ++;
|
||||
fh->f_io_array = (mca_io_ompio_io_array_t *)realloc
|
||||
(fh->f_io_array, OMPIO_IOVEC_INITIAL_SIZE *
|
||||
block * sizeof (mca_io_ompio_io_array_t));
|
||||
if (NULL == fh->f_io_array) {
|
||||
opal_output(1, "OUT OF MEMORY\n");
|
||||
return OMPI_ERR_OUT_OF_RESOURCE;
|
||||
}
|
||||
}
|
||||
|
||||
if (decoded_iov[i].iov_len -
|
||||
(total_bytes_written - sum_previous_counts) <= 0) {
|
||||
sum_previous_counts += decoded_iov[i].iov_len;
|
||||
i = i + 1;
|
||||
}
|
||||
|
||||
disp = (OPAL_PTRDIFF_TYPE)decoded_iov[i].iov_base +
|
||||
(total_bytes_written - sum_previous_counts);
|
||||
fh->f_io_array[k].memory_address = (IOVBASE_TYPE *)disp;
|
||||
|
||||
if (decoded_iov[i].iov_len -
|
||||
(total_bytes_written - sum_previous_counts) >=
|
||||
bytes_to_write_in_cycle) {
|
||||
fh->f_io_array[k].length = bytes_to_write_in_cycle;
|
||||
}
|
||||
else {
|
||||
fh->f_io_array[k].length = decoded_iov[i].iov_len -
|
||||
(total_bytes_written - sum_previous_counts);
|
||||
}
|
||||
|
||||
if (! (fh->f_flags & OMPIO_CONTIGUOUS_FVIEW)) {
|
||||
if (fh->f_decoded_iov[j].iov_len -
|
||||
(fh->f_total_bytes - sum_previous_length) <= 0) {
|
||||
sum_previous_length += fh->f_decoded_iov[j].iov_len;
|
||||
j = j + 1;
|
||||
if (j == (int)fh->f_iov_count) {
|
||||
j = 0;
|
||||
sum_previous_length = 0;
|
||||
fh->f_offset += fh->f_view_extent;
|
||||
fh->f_position_in_file_view = sum_previous_length;
|
||||
fh->f_index_in_file_view = j;
|
||||
fh->f_total_bytes = 0;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
disp = (OPAL_PTRDIFF_TYPE)fh->f_decoded_iov[j].iov_base +
|
||||
(fh->f_total_bytes - sum_previous_length);
|
||||
fh->f_io_array[k].offset = (IOVBASE_TYPE *)(intptr_t)(disp + fh->f_offset);
|
||||
|
||||
if (! (fh->f_flags & OMPIO_CONTIGUOUS_FVIEW)) {
|
||||
if (fh->f_decoded_iov[j].iov_len -
|
||||
(fh->f_total_bytes - sum_previous_length)
|
||||
< fh->f_io_array[k].length) {
|
||||
fh->f_io_array[k].length = fh->f_decoded_iov[j].iov_len -
|
||||
(fh->f_total_bytes - sum_previous_length);
|
||||
}
|
||||
}
|
||||
|
||||
total_bytes_written += fh->f_io_array[k].length;
|
||||
fh->f_total_bytes += fh->f_io_array[k].length;
|
||||
bytes_to_write_in_cycle -= fh->f_io_array[k].length;
|
||||
k = k + 1;
|
||||
}
|
||||
fh->f_position_in_file_view = sum_previous_length;
|
||||
fh->f_index_in_file_view = j;
|
||||
fh->f_num_of_io_entries = k;
|
||||
|
||||
#if 0
|
||||
if (fh->f_rank == 0) {
|
||||
int d;
|
||||
printf("*************************** %d\n", fh->f_num_of_io_entries);
|
||||
|
||||
for (d=0 ; d<fh->f_num_of_io_entries ; d++) {
|
||||
printf(" ADDRESS: %p OFFSET: %p LENGTH: %d\n",
|
||||
fh->f_io_array[d].memory_address,
|
||||
fh->f_io_array[d].offset,
|
||||
fh->f_io_array[d].length);
|
||||
}
|
||||
}
|
||||
#endif
|
||||
*ii = i;
|
||||
*jj = j;
|
||||
*tbw = total_bytes_written;
|
||||
|
||||
return OMPI_SUCCESS;
|
||||
}
|
||||
|
||||
/* Collective operations */
|
||||
/******************************************************************/
|
||||
|
||||
int mca_io_ompio_file_write_all (ompi_file_t *fh,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t *status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
@ -296,51 +456,12 @@ mca_io_ompio_file_write_all (ompi_file_t *fh,
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
mca_io_ompio_file_write_all_begin (ompi_file_t *fh,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
|
||||
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
||||
|
||||
ret = data->ompio_fh.
|
||||
f_fcoll->fcoll_file_write_all_begin (&data->ompio_fh,
|
||||
buf,
|
||||
count,
|
||||
datatype);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
mca_io_ompio_file_write_all_end (ompi_file_t *fh,
|
||||
void *buf,
|
||||
ompi_status_public_t *status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
|
||||
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
||||
|
||||
ret = data->ompio_fh.
|
||||
f_fcoll->fcoll_file_write_all_end (&data->ompio_fh,
|
||||
buf,
|
||||
status);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
mca_io_ompio_file_write_at_all (ompi_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t *status)
|
||||
int mca_io_ompio_file_write_at_all (ompi_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t *status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
@ -351,18 +472,16 @@ mca_io_ompio_file_write_at_all (ompi_file_t *fh,
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
ompio_io_ompio_file_write_at_all (mca_io_ompio_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t *status)
|
||||
int ompio_io_ompio_file_write_at_all (mca_io_ompio_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t *status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
|
||||
ompi_io_ompio_set_explicit_offset (fh, offset);
|
||||
|
||||
ret = fh->f_fcoll->fcoll_file_write_all (fh,
|
||||
buf,
|
||||
count,
|
||||
@ -371,140 +490,16 @@ ompio_io_ompio_file_write_at_all (mca_io_ompio_file_t *fh,
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
mca_io_ompio_file_write_at_all_begin (ompi_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
|
||||
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
||||
ret = ompio_io_ompio_file_write_at_all_begin(&data->ompio_fh,offset,buf,count,datatype);
|
||||
/* Infrastructure for shared file pointer operations */
|
||||
/* (Individual and collective */
|
||||
/******************************************************/
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
ompio_io_ompio_file_write_at_all_begin (mca_io_ompio_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
|
||||
ompi_io_ompio_set_explicit_offset (fh, offset);
|
||||
|
||||
ret = fh->f_fcoll->fcoll_file_write_all_begin (fh,
|
||||
buf,
|
||||
count,
|
||||
datatype);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
mca_io_ompio_file_write_at_all_end (ompi_file_t *fh,
|
||||
void *buf,
|
||||
ompi_status_public_t * status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
|
||||
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
||||
ret = ompio_io_ompio_file_write_at_all_end(&data->ompio_fh,buf,status);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
ompio_io_ompio_file_write_at_all_end (mca_io_ompio_file_t *fh,
|
||||
void *buf,
|
||||
ompi_status_public_t * status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
|
||||
ret = fh->f_fcoll->fcoll_file_write_all_end (fh,
|
||||
buf,
|
||||
status);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
mca_io_ompio_file_iwrite (ompi_file_t *fp,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_request_t **request)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
|
||||
data = (mca_io_ompio_data_t *) fp->f_io_selected_data;
|
||||
ret = ompio_io_ompio_file_iwrite(&data->ompio_fh,buf,count,datatype,request);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
ompio_io_ompio_file_iwrite (mca_io_ompio_file_t *fh,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_request_t **request)
|
||||
{
|
||||
|
||||
int ret = MPI_ERR_OTHER;
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
mca_io_ompio_file_iwrite_at (ompi_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_request_t **request)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
|
||||
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
||||
ret = ompio_io_ompio_file_iwrite_at(&data->ompio_fh,offset,buf,count,datatype,request);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
ompio_io_ompio_file_iwrite_at (mca_io_ompio_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_request_t **request)
|
||||
{
|
||||
|
||||
int ret = OMPI_SUCCESS;
|
||||
|
||||
ompi_io_ompio_set_explicit_offset (fh, offset);
|
||||
ret = ompio_io_ompio_file_iwrite (fh,
|
||||
buf,
|
||||
count,
|
||||
datatype,
|
||||
request);
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
mca_io_ompio_file_write_shared (ompi_file_t *fp,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t * status)
|
||||
int mca_io_ompio_file_write_shared (ompi_file_t *fp,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t * status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
@ -521,12 +516,11 @@ mca_io_ompio_file_write_shared (ompi_file_t *fp,
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
mca_io_ompio_file_iwrite_shared (ompi_file_t *fp,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_request_t **request)
|
||||
int mca_io_ompio_file_iwrite_shared (ompi_file_t *fp,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_request_t **request)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
@ -543,12 +537,11 @@ mca_io_ompio_file_iwrite_shared (ompi_file_t *fp,
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
mca_io_ompio_file_write_ordered (ompi_file_t *fp,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t * status)
|
||||
int mca_io_ompio_file_write_ordered (ompi_file_t *fp,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype,
|
||||
ompi_status_public_t * status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
@ -565,11 +558,10 @@ mca_io_ompio_file_write_ordered (ompi_file_t *fp,
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
mca_io_ompio_file_write_ordered_begin (ompi_file_t *fp,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype)
|
||||
int mca_io_ompio_file_write_ordered_begin (ompi_file_t *fp,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
@ -586,10 +578,9 @@ mca_io_ompio_file_write_ordered_begin (ompi_file_t *fp,
|
||||
return ret;
|
||||
}
|
||||
|
||||
int
|
||||
mca_io_ompio_file_write_ordered_end (ompi_file_t *fp,
|
||||
void *buf,
|
||||
ompi_status_public_t *status)
|
||||
int mca_io_ompio_file_write_ordered_end (ompi_file_t *fp,
|
||||
void *buf,
|
||||
ompi_status_public_t *status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
@ -605,3 +596,99 @@ mca_io_ompio_file_write_ordered_end (ompi_file_t *fp,
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
|
||||
/* Split collectives . Not really used but infrastructure is in place */
|
||||
/**********************************************************************/
|
||||
int mca_io_ompio_file_write_at_all_begin (ompi_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
|
||||
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
||||
ret = ompio_io_ompio_file_write_at_all_begin(&data->ompio_fh,offset,buf,count,datatype);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int ompio_io_ompio_file_write_at_all_begin (mca_io_ompio_file_t *fh,
|
||||
OMPI_MPI_OFFSET_TYPE offset,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
|
||||
ompi_io_ompio_set_explicit_offset (fh, offset);
|
||||
ret = fh->f_fcoll->fcoll_file_write_all_begin (fh,
|
||||
buf,
|
||||
count,
|
||||
datatype);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int mca_io_ompio_file_write_at_all_end (ompi_file_t *fh,
|
||||
void *buf,
|
||||
ompi_status_public_t * status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
|
||||
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
||||
ret = ompio_io_ompio_file_write_at_all_end(&data->ompio_fh,buf,status);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int ompio_io_ompio_file_write_at_all_end (mca_io_ompio_file_t *fh,
|
||||
void *buf,
|
||||
ompi_status_public_t * status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
|
||||
ret = fh->f_fcoll->fcoll_file_write_all_end (fh,
|
||||
buf,
|
||||
status);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int mca_io_ompio_file_write_all_begin (ompi_file_t *fh,
|
||||
void *buf,
|
||||
int count,
|
||||
struct ompi_datatype_t *datatype)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
|
||||
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
||||
ret = data->ompio_fh.
|
||||
f_fcoll->fcoll_file_write_all_begin (&data->ompio_fh,
|
||||
buf,
|
||||
count,
|
||||
datatype);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
int mca_io_ompio_file_write_all_end (ompi_file_t *fh,
|
||||
void *buf,
|
||||
ompi_status_public_t *status)
|
||||
{
|
||||
int ret = OMPI_SUCCESS;
|
||||
mca_io_ompio_data_t *data;
|
||||
|
||||
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
||||
ret = data->ompio_fh.
|
||||
f_fcoll->fcoll_file_write_all_end (&data->ompio_fh,
|
||||
buf,
|
||||
status);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
|
Загрузка…
x
Ссылка в новой задаче
Block a user