552 строки
14 KiB
C
552 строки
14 KiB
C
/* -*- Mode: C; c-basic-offset:4 ; -*- */
|
|
/*
|
|
* Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
|
|
* University Research and Technology
|
|
* Corporation. All rights reserved.
|
|
* Copyright (c) 2004-2007 The University of Tennessee and The University
|
|
* of Tennessee Research Foundation. All rights
|
|
* reserved.
|
|
* Copyright (c) 2004-2005 High Performance Computing Center Stuttgart,
|
|
* University of Stuttgart. All rights reserved.
|
|
* Copyright (c) 2004-2005 The Regents of the University of California.
|
|
* All rights reserved.
|
|
* Copyright (c) 2008-2011 University of Houston. All rights reserved.
|
|
* Copyright (c) 2015 Research Organization for Information Science
|
|
* and Technology (RIST). All rights reserved.
|
|
* $COPYRIGHT$
|
|
*
|
|
* Additional copyrights may follow
|
|
*
|
|
* $HEADER$
|
|
*/
|
|
|
|
#include "ompi_config.h"
|
|
|
|
#include "ompi/runtime/params.h"
|
|
#include "ompi/communicator/communicator.h"
|
|
#include "opal/datatype/opal_convertor.h"
|
|
#include "opal/datatype/opal_datatype.h"
|
|
#include "ompi/datatype/ompi_datatype.h"
|
|
#include "ompi/info/info.h"
|
|
#include "ompi/request/request.h"
|
|
|
|
#include <math.h>
|
|
#include <unistd.h>
|
|
#include "io_ompio_nbc.h"
|
|
|
|
|
|
|
|
int mca_io_ompio_get_f_aggregator_index (ompi_file_t *fh)
|
|
{
|
|
mca_io_ompio_data_t *data;
|
|
mca_io_ompio_file_t *file;
|
|
|
|
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
|
file = &data->ompio_fh;
|
|
|
|
return file->f_aggregator_index;
|
|
}
|
|
|
|
int mca_io_ompio_get_f_num_of_io_entries(ompi_file_t *fh)
|
|
{
|
|
|
|
mca_io_ompio_data_t *data;
|
|
mca_io_ompio_file_t *file;
|
|
|
|
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
|
file = &data->ompio_fh;
|
|
|
|
return file->f_num_of_io_entries;
|
|
}
|
|
|
|
int mca_io_ompio_get_fcoll_dynamic_num_io_procs (int *num_procs)
|
|
{
|
|
int param;
|
|
|
|
param = mca_base_var_find("ompi", "fcoll", "dynamic", "num_io_procs");
|
|
if (param >= 0){
|
|
const int *value = NULL;
|
|
mca_base_var_get_value(param, &value, NULL, NULL);
|
|
*num_procs = value[0];
|
|
/* printf("num procs : %d\n", num_procs);*/
|
|
return OMPI_SUCCESS;
|
|
}
|
|
else
|
|
return -1;
|
|
|
|
}
|
|
|
|
int mca_io_ompio_get_fcoll_dynamic_constant_cbs (int *constant_cbs)
|
|
{
|
|
int param;
|
|
|
|
param = mca_base_var_find("ompi", "fcoll", "dynamic", "constant_cbs");
|
|
if (param >= 0){
|
|
const int *value = NULL;
|
|
mca_base_var_get_value(param, &value, NULL, NULL);
|
|
*constant_cbs = value[0];
|
|
/* printf ("constant_cbs: %d\n", constant_cbs);*/
|
|
return OMPI_SUCCESS;
|
|
}
|
|
else{
|
|
constant_cbs[0] = -1;
|
|
return OMPI_SUCCESS;
|
|
}
|
|
|
|
}
|
|
|
|
int mca_io_ompio_get_fcoll_dynamic_cycle_buffer_size (int *cycle_buffer_size)
|
|
{
|
|
|
|
int param;
|
|
|
|
param = mca_base_var_find("ompi", "fcoll", "dynamic", "cycle_buffer_size");
|
|
if (param >= 0){
|
|
const int *value = NULL;
|
|
mca_base_var_get_value(param, &value, NULL, NULL);
|
|
*cycle_buffer_size = value[0];
|
|
/* printf ("cycle_buffer_size : %d\n", *cycle_buffer_size);*/
|
|
return OMPI_SUCCESS;
|
|
}
|
|
else
|
|
return -1;
|
|
|
|
}
|
|
|
|
int mca_io_ompio_get_f_io_array(ompi_file_t *fh,
|
|
mca_io_ompio_io_array_t **f_io_array)
|
|
{
|
|
|
|
mca_io_ompio_data_t *data;
|
|
mca_io_ompio_file_t *file;
|
|
|
|
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
|
file = &data->ompio_fh;
|
|
*f_io_array = file->f_io_array;
|
|
return OMPI_SUCCESS;
|
|
}
|
|
|
|
int mca_io_ompio_get_f_comm(ompi_file_t *fh, ompi_communicator_t **value)
|
|
{
|
|
mca_io_ompio_data_t *data;
|
|
mca_io_ompio_file_t *file;
|
|
|
|
data = (mca_io_ompio_data_t *)fh->f_io_selected_data;
|
|
file = &(data->ompio_fh);
|
|
*value = file->f_comm;
|
|
|
|
return OMPI_SUCCESS;
|
|
}
|
|
|
|
int mca_io_ompio_get_iov_type(ompi_file_t *fh, ompi_datatype_t **value)
|
|
{
|
|
mca_io_ompio_data_t *data;
|
|
mca_io_ompio_file_t *file;
|
|
|
|
data = (mca_io_ompio_data_t *)fh->f_io_selected_data;
|
|
file = &data->ompio_fh;
|
|
|
|
*value = file->f_iov_type;
|
|
return OMPI_SUCCESS;
|
|
}
|
|
|
|
int mca_io_ompio_get_f_procs_in_group(ompi_file_t *fh, int **value)
|
|
{
|
|
mca_io_ompio_data_t *data;
|
|
mca_io_ompio_file_t *file;
|
|
|
|
data = (mca_io_ompio_data_t *)fh->f_io_selected_data;
|
|
file = &data->ompio_fh;
|
|
|
|
*value = file->f_procs_in_group;
|
|
return OMPI_SUCCESS;
|
|
}
|
|
|
|
int mca_io_ompio_get_f_procs_per_group(ompi_file_t *fh)
|
|
{
|
|
mca_io_ompio_data_t *data;
|
|
mca_io_ompio_file_t *file;
|
|
|
|
data = (mca_io_ompio_data_t *)fh->f_io_selected_data;
|
|
file = &data->ompio_fh;
|
|
|
|
return file->f_procs_per_group;
|
|
}
|
|
|
|
signed int mca_io_ompio_get_f_flags(ompi_file_t *fh)
|
|
{
|
|
mca_io_ompio_data_t *data;
|
|
mca_io_ompio_file_t *file;
|
|
|
|
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
|
file = &data->ompio_fh;
|
|
|
|
return file->f_flags;
|
|
}
|
|
|
|
int mca_io_ompio_get_fd(ompi_file_t *fh)
|
|
{
|
|
mca_io_ompio_data_t *data;
|
|
mca_io_ompio_file_t *file;
|
|
|
|
data = (mca_io_ompio_data_t *) fh->f_io_selected_data;
|
|
file = &data->ompio_fh;
|
|
|
|
return file->fd;
|
|
}
|
|
|
|
int mca_io_ompio_generate_io_array(ompi_file_t *file,
|
|
struct iovec *global_fview,
|
|
int *tglobal_count,
|
|
int *fview_count,
|
|
int *bytes_per_process,
|
|
char *global_buf,
|
|
int *tblocks,
|
|
int *sorted,
|
|
int *nvalue,
|
|
int *bytes_left_ptr,
|
|
int *sorted_index)
|
|
{
|
|
mca_io_ompio_data_t *data;
|
|
mca_io_ompio_file_t *fh;
|
|
int k, j, x=sorted_index[0];
|
|
int blocks = *tblocks;
|
|
int bytes_left = bytes_left_ptr[0];
|
|
|
|
|
|
data = (mca_io_ompio_data_t *) file->f_io_selected_data;
|
|
fh = &data->ompio_fh;
|
|
|
|
|
|
if (fh->f_procs_in_group[fh->f_aggregator_index] == fh->f_rank) {
|
|
int global_count = *tglobal_count;
|
|
int bytes_to_write = global_count;
|
|
int *temp = NULL;
|
|
int block = 1;
|
|
k = 0;
|
|
temp = (int *)malloc (sizeof(int) * fh->f_procs_per_group);
|
|
if (NULL == temp) {
|
|
opal_output(1, "OUT OF MEMORY\n");
|
|
return OMPI_ERR_OUT_OF_RESOURCE;
|
|
}
|
|
memset(temp, 0x0, fh->f_procs_per_group*sizeof(int));
|
|
if (NULL != fh->f_io_array){
|
|
fh->f_num_of_io_entries = 0;
|
|
free (fh->f_io_array);
|
|
fh->f_io_array = NULL;
|
|
}
|
|
|
|
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");
|
|
free(temp);
|
|
return OMPI_ERR_OUT_OF_RESOURCE;
|
|
}
|
|
while (bytes_to_write) {
|
|
int start = 0;
|
|
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");
|
|
free(temp);
|
|
return OMPI_ERR_OUT_OF_RESOURCE;
|
|
}
|
|
}
|
|
blocks= fview_count[0];
|
|
for (j=0 ; j<fh->f_procs_per_group ; j++) {
|
|
if (sorted[x] < blocks) {
|
|
nvalue[0] = j;
|
|
break;
|
|
}
|
|
else {
|
|
blocks += fview_count[j+1];
|
|
}
|
|
}
|
|
for (j=0 ; j<nvalue[0] ; j++) {
|
|
start += bytes_per_process[j];
|
|
}
|
|
if (bytes_left) {
|
|
if (bytes_left <= bytes_to_write) {
|
|
fh->f_io_array[k].offset = (IOVBASE_TYPE *)
|
|
((OPAL_PTRDIFF_TYPE)global_fview[sorted[x]].iov_base +
|
|
(global_fview[sorted[x]].iov_len - bytes_left));
|
|
|
|
fh->f_io_array[k].length = bytes_left;
|
|
fh->f_io_array[k].memory_address = &global_buf[start+temp[nvalue[0]]];
|
|
/* printf("global_buf[%d] : %d\n",
|
|
(start+temp[nvalue[0]]),(int)global_buf[start+temp[nvalue[0]]]);*/
|
|
|
|
temp[nvalue[0]] += (int)fh->f_io_array[k].length;
|
|
bytes_to_write -= bytes_left;
|
|
bytes_left = 0;
|
|
k ++;
|
|
x ++;
|
|
continue;
|
|
}
|
|
else {
|
|
fh->f_io_array[k].offset = (IOVBASE_TYPE *)
|
|
((OPAL_PTRDIFF_TYPE)global_fview[sorted[x]].iov_base + (global_fview[sorted[x]].iov_len - bytes_left));
|
|
|
|
fh->f_io_array[k].length = bytes_to_write;
|
|
fh->f_io_array[k].memory_address =
|
|
&global_buf[start+temp[nvalue[0]]];
|
|
/* printf("global_buf[%d] : %d\n",
|
|
(start+temp[nvalue[0]]),(int)global_buf[start+temp[nvalue[0]]]);*/
|
|
|
|
temp[nvalue[0]] += (int)fh->f_io_array[k].length;
|
|
bytes_left -= bytes_to_write;
|
|
bytes_to_write = 0;;
|
|
k ++;
|
|
break;
|
|
}
|
|
}
|
|
else {
|
|
if (bytes_to_write < (int)global_fview[sorted[x]].iov_len) {
|
|
fh->f_io_array[k].offset = global_fview[sorted[x]].iov_base;
|
|
|
|
fh->f_io_array[k].length = bytes_to_write;
|
|
fh->f_io_array[k].memory_address = &global_buf[start+temp[nvalue[0]]];
|
|
/* printf("global_buf[%d] : %d\n",
|
|
(start+temp[nvalue[0]]),(int)global_buf[start+temp[nvalue[0]]]);*/
|
|
|
|
bytes_left =
|
|
global_fview[sorted[x]].iov_len - bytes_to_write;
|
|
bytes_to_write = 0;
|
|
k ++;
|
|
break;
|
|
}
|
|
else {
|
|
fh->f_io_array[k].offset = global_fview[sorted[x]].iov_base;
|
|
|
|
fh->f_io_array[k].length = global_fview[sorted[x]].iov_len;
|
|
fh->f_io_array[k].memory_address = &global_buf[start+temp[nvalue[0]]];
|
|
temp[nvalue[0]] += (int)fh->f_io_array[k].length;
|
|
/* printf("global_buf[%d] : %d\n",
|
|
(start+temp[nvalue[0]]),(int)global_buf[start+temp[nvalue[0]]]);*/
|
|
|
|
bytes_to_write -= global_fview[sorted[x]].iov_len;
|
|
k ++;
|
|
x ++;
|
|
continue;
|
|
}
|
|
}
|
|
}
|
|
fh->f_num_of_io_entries = k;
|
|
/* for (i=0 ; i<fh->f_num_of_io_entries ; i++) {
|
|
printf("OFFSET: %lu LENGTH: %d\n",
|
|
fh->f_io_array[i].offset,
|
|
fh->f_io_array[i].length);
|
|
}*/
|
|
|
|
|
|
bytes_left_ptr[0] = bytes_left;
|
|
sorted_index[0] = x;
|
|
|
|
if (NULL != temp) {
|
|
free (temp);
|
|
temp = NULL;
|
|
}
|
|
|
|
}
|
|
return OMPI_SUCCESS;
|
|
}
|
|
|
|
int mca_io_ompio_non_contiguous_create_receive_buf(int *bytes_received,
|
|
struct iovec *decoded_iov,
|
|
char *receive_buf)
|
|
{
|
|
|
|
OPAL_PTRDIFF_TYPE mem_address;
|
|
size_t remaining = 0;
|
|
size_t temp_position = 0;
|
|
int current_position = 0, iov_index = 0;
|
|
|
|
remaining = *bytes_received;
|
|
|
|
while (remaining) {
|
|
mem_address = (OPAL_PTRDIFF_TYPE)
|
|
(decoded_iov[iov_index].iov_base) + current_position;
|
|
|
|
if (remaining >=
|
|
(decoded_iov[iov_index].iov_len - current_position)) {
|
|
memcpy ((IOVBASE_TYPE *) mem_address,
|
|
receive_buf+temp_position,
|
|
decoded_iov[iov_index].iov_len - current_position);
|
|
remaining = remaining -
|
|
(decoded_iov[iov_index].iov_len - current_position);
|
|
temp_position = temp_position +
|
|
(decoded_iov[iov_index].iov_len - current_position);
|
|
iov_index = iov_index + 1;
|
|
current_position = 0;
|
|
}
|
|
else {
|
|
memcpy ((IOVBASE_TYPE *) mem_address,
|
|
receive_buf+temp_position,
|
|
remaining);
|
|
current_position = current_position + remaining;
|
|
remaining = 0;
|
|
}
|
|
}
|
|
if (NULL != receive_buf) {
|
|
free (receive_buf);
|
|
receive_buf = NULL;
|
|
}
|
|
return OMPI_SUCCESS;
|
|
}
|
|
|
|
|
|
int mca_io_ompio_non_contiguous_create_send_buf(int *bytes_sent,
|
|
struct iovec *decoded_iov,
|
|
char *send_buf)
|
|
{
|
|
|
|
OPAL_PTRDIFF_TYPE mem_address;
|
|
size_t remaining = 0;
|
|
size_t temp_position = 0;
|
|
int current_position = 0, iov_index = 0;
|
|
|
|
remaining = *bytes_sent;
|
|
while (remaining) {
|
|
mem_address = (OPAL_PTRDIFF_TYPE)
|
|
(decoded_iov[iov_index].iov_base) + current_position;
|
|
if (remaining >=
|
|
(decoded_iov[iov_index].iov_len - current_position)) {
|
|
memcpy (send_buf+temp_position,
|
|
(IOVBASE_TYPE *)mem_address,
|
|
decoded_iov[iov_index].iov_len - current_position);
|
|
remaining = remaining -
|
|
(decoded_iov[iov_index].iov_len - current_position);
|
|
temp_position = temp_position +
|
|
(decoded_iov[iov_index].iov_len - current_position);
|
|
iov_index = iov_index + 1;
|
|
current_position = 0;
|
|
}
|
|
else {
|
|
memcpy (send_buf+temp_position, (IOVBASE_TYPE *) mem_address,
|
|
remaining);
|
|
current_position = current_position + remaining;
|
|
remaining = 0;
|
|
}
|
|
}
|
|
return OMPI_SUCCESS;
|
|
}
|
|
|
|
|
|
|
|
int mca_io_ompio_get_datatype_size (ompi_datatype_t * datatype)
|
|
{
|
|
return datatype->super.size;
|
|
}
|
|
|
|
int mca_io_ompio_decode_datatype_external (ompi_file_t *fp,
|
|
ompi_datatype_t *datatype,
|
|
int count,
|
|
void *buf,
|
|
size_t *max_data,
|
|
struct iovec **iov,
|
|
uint32_t *iovec_count)
|
|
{
|
|
|
|
int res;
|
|
mca_io_ompio_data_t *data;
|
|
mca_io_ompio_file_t *fh;
|
|
|
|
data = (mca_io_ompio_data_t *) fp->f_io_selected_data;
|
|
fh = &data->ompio_fh;
|
|
res = ompi_io_ompio_decode_datatype (fh,
|
|
datatype,
|
|
count,
|
|
buf,
|
|
max_data,
|
|
iov,
|
|
iovec_count);
|
|
if(res != OMPI_SUCCESS){
|
|
printf("Error in ompio decode datatype\n");
|
|
return res;
|
|
}
|
|
return OMPI_SUCCESS;
|
|
|
|
}
|
|
|
|
int mca_io_ompio_datatype_is_contiguous(ompi_datatype_t *datatype,
|
|
ompi_file_t *fp)
|
|
{
|
|
mca_io_ompio_data_t *data;
|
|
mca_io_ompio_file_t *fh;
|
|
|
|
data = (mca_io_ompio_data_t *) fp->f_io_selected_data;
|
|
fh = &data->ompio_fh;
|
|
|
|
if (opal_datatype_is_contiguous_memory_layout(&datatype->super,1)){
|
|
fh->f_flags |= OMPIO_CONTIGUOUS_MEMORY;
|
|
return 1;
|
|
}
|
|
else
|
|
return 0;
|
|
}
|
|
|
|
|
|
int mca_io_ompio_set_aggregator_props (ompi_file_t *fp,
|
|
int num_aggregators,
|
|
size_t bytes_per_proc)
|
|
{
|
|
int res;
|
|
mca_io_ompio_data_t *data;
|
|
mca_io_ompio_file_t *fh;
|
|
|
|
data = (mca_io_ompio_data_t *) fp->f_io_selected_data;
|
|
fh = &data->ompio_fh;
|
|
res = ompi_io_ompio_set_aggregator_props (fh,
|
|
num_aggregators,
|
|
bytes_per_proc);
|
|
if(res != OMPI_SUCCESS){
|
|
printf("Error in aggregator props external\n");
|
|
return res;
|
|
}
|
|
|
|
return OMPI_SUCCESS;
|
|
}
|
|
|
|
int mca_io_ompio_generate_current_file_view (ompi_file_t *fp,
|
|
size_t max_data,
|
|
struct iovec **f_iov,
|
|
int *iov_count)
|
|
{
|
|
int res;
|
|
mca_io_ompio_data_t *data;
|
|
mca_io_ompio_file_t *fh;
|
|
|
|
data = (mca_io_ompio_data_t *) fp->f_io_selected_data;
|
|
fh = &data->ompio_fh;
|
|
res = ompi_io_ompio_generate_current_file_view (fh,
|
|
max_data,
|
|
f_iov,
|
|
iov_count);
|
|
if(res != OMPI_SUCCESS){
|
|
printf("Error in ompi_io_generate_current_file_view\n");
|
|
return res;
|
|
}
|
|
|
|
return OMPI_SUCCESS;
|
|
}
|
|
|
|
int mca_io_ompio_free_f_io_array (ompi_file_t *fp){
|
|
|
|
mca_io_ompio_data_t *data;
|
|
mca_io_ompio_file_t *fh;
|
|
|
|
data = (mca_io_ompio_data_t *) fp->f_io_selected_data;
|
|
fh = &data->ompio_fh;
|
|
|
|
if (NULL != fh->f_io_array) {
|
|
free (fh->f_io_array);
|
|
fh->f_io_array = NULL;
|
|
}
|
|
|
|
return OMPI_SUCCESS;
|
|
}
|