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

remove the ylib component from the fcoll framework. It is not used, there are

Просмотр исходного кода 
no plans to use it. We can always recover it from svn if we would ever change
our minds.

This commit was SVN r28840.
Этот коммит содержится в:
Edgar Gabriel 2013-07-18 16:18:06 +00:00
родитель 92c6b806b9
Коммит 93cef82873
10 изменённых файлов: 0 добавлений и 1852 удалений
Показать статистику Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

51
ompi/mca/fcoll/ylib/Makefile.am
Просмотреть файл

@ -1,51 +0,0 @@
#
# Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
# University Research and Technology
# Corporation. All rights reserved.
# Copyright (c) 2004-2005 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) 2012 Cisco Systems, Inc. All rights reserved.
# $COPYRIGHT$
#
# Additional copyrights may follow
#
# $HEADER$
#
sources = \
fcoll_ylib.h \
fcoll_ylib_module.c \
fcoll_ylib_component.c \
fcoll_ylib_file_read_all.c \
fcoll_ylib_file_read_all_begin.c \
fcoll_ylib_file_read_all_end.c \
fcoll_ylib_file_write_all.c \
fcoll_ylib_file_write_all_begin.c \
fcoll_ylib_file_write_all_end.c
# Make the output library in this directory, and name it either
# mca_<type>_<name>.la (for DSO builds) or libmca_<type>_<name>.la
# (for static builds).
if MCA_BUILD_ompi_fcoll_ylib_DSO
component_noinst =
component_install = mca_fcoll_ylib.la
else
component_noinst = libmca_fcoll_ylib.la
component_install =
endif
mcacomponentdir = $(libdir)/openmpi
mcacomponent_LTLIBRARIES = $(component_install)
mca_fcoll_ylib_la_SOURCES = $(sources)
mca_fcoll_ylib_la_LDFLAGS = -module -avoid-version
noinst_LTLIBRARIES = $(component_noinst)
libmca_fcoll_ylib_la_SOURCES =$(sources)
libmca_fcoll_ylib_la_LDFLAGS = -module -avoid-version

86
ompi/mca/fcoll/ylib/fcoll_ylib.h
Просмотреть файл

@ -1,86 +0,0 @@
/*
* Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
* University Research and Technology
* Corporation. All rights reserved.
* Copyright (c) 2004-2006 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$
*
* Additional copyrights may follow
*
* $HEADER$
*/
#ifndef MCA_FCOLL_YLIB_EXPORT_H
#define MCA_FCOLL_YLIB_EXPORT_H
#include "ompi_config.h"
#include "mpi.h"
#include "opal/mca/mca.h"
#include "ompi/mca/fcoll/fcoll.h"
#include "ompi/mca/fcoll/base/base.h"
#include "ompi/mca/io/ompio/io_ompio.h"
BEGIN_C_DECLS
/* Globally exported variables */
extern int mca_fcoll_ylib_priority;
extern int mca_fcoll_ylib_num_io_procs;
extern int mca_fcoll_ylib_stripe_size;
extern int mca_fcoll_ylib_blocks_per_cycle;
OMPI_MODULE_DECLSPEC extern mca_fcoll_base_component_2_0_0_t mca_fcoll_ylib_component;
/* API functions */
int mca_fcoll_ylib_component_init_query(bool enable_progress_threads,
bool enable_mpi_threads);
struct mca_fcoll_base_module_1_0_0_t *
mca_fcoll_ylib_component_file_query (mca_io_ompio_file_t *fh, int *priority);
int mca_fcoll_ylib_component_file_unquery (mca_io_ompio_file_t *file);
int mca_fcoll_ylib_module_init (mca_io_ompio_file_t *file);
int mca_fcoll_ylib_module_finalize (mca_io_ompio_file_t *file);
int mca_fcoll_ylib_file_read_all (mca_io_ompio_file_t *fh,
void *buf,
int count,
struct ompi_datatype_t *datatype,
ompi_status_public_t * status);
int mca_fcoll_ylib_file_read_all_begin (mca_io_ompio_file_t *fh,
void *buf,
int count,
struct ompi_datatype_t *datatype);
int mca_fcoll_ylib_file_read_all_end (mca_io_ompio_file_t *fh,
void *buf,
ompi_status_public_t * status);
int mca_fcoll_ylib_file_write_all (mca_io_ompio_file_t *fh,
void *buf,
int count,
struct ompi_datatype_t *datatype,
ompi_status_public_t * status);
int mca_fcoll_ylib_file_write_all_begin (mca_io_ompio_file_t *fh,
void *buf,
int count,
struct ompi_datatype_t *datatype);
int mca_fcoll_ylib_file_write_all_end (mca_io_ompio_file_t *fh,
void *buf,
ompi_status_public_t * status);
END_C_DECLS
#endif /* MCA_FCOLL_YLIB_EXPORT_H */

115
ompi/mca/fcoll/ylib/fcoll_ylib_component.c
Просмотреть файл

@ -1,115 +0,0 @@
/*
* Copyright (c) 2004-2007 The Trustees of Indiana University and Indiana
* University Research and Technology
* Corporation. All rights reserved.
* Copyright (c) 2004-2005 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 Cisco Systems, Inc. All rights reserved.
* Copyright (c) 2008-2011 University of Houston. All rights reserved.
* $COPYRIGHT$
*
* Additional copyrights may follow
*
* $HEADER$
*
* These symbols are in a file by themselves to provide nice linker
* semantics. Since linkers generally pull in symbols by object
* files, keeping these symbols as the only symbols in this file
* prevents utility programs such as "ompi_info" from having to import
* entire components just to query their version and parameters.
*/
#include "ompi_config.h"
#include "fcoll_ylib.h"
#include "mpi.h"
#include "ompi/mca/fcoll/fcoll.h"
#include "fcoll_ylib.h"
/*
* Public string showing the fcoll ompi_ylib component version number
*/
const char *mca_fcoll_ylib_component_version_string =
"Open MPI ylib collective MCA component version " OMPI_VERSION;
/*
* Global variables
*/
int mca_fcoll_ylib_priority = 0;
int mca_fcoll_ylib_num_io_procs = 1;
int mca_fcoll_ylib_stripe_size = 1048576;
int mca_fcoll_ylib_blocks_per_cycle = 20;
/*
* Local function
*/
static int ylib_register(void);
/*
* Instantiate the public struct with all of our public information
* and pointers to our public functions in it
*/
mca_fcoll_base_component_2_0_0_t mca_fcoll_ylib_component = {
/* First, the mca_component_t struct containing meta information
* about the component itself */
{
MCA_FCOLL_BASE_VERSION_2_0_0,
/* Component name and version */
"ylib",
OMPI_MAJOR_VERSION,
OMPI_MINOR_VERSION,
OMPI_RELEASE_VERSION,
NULL,
NULL,
NULL,
ylib_register,
},
{
/* The component is checkpoint ready */
MCA_BASE_METADATA_PARAM_CHECKPOINT
},
mca_fcoll_ylib_component_init_query,
mca_fcoll_ylib_component_file_query,
mca_fcoll_ylib_component_file_unquery
};
static int
ylib_register(void)
{
mca_fcoll_ylib_priority = 0;
(void) mca_base_component_var_register(&mca_fcoll_ylib_component.fcollm_version,
"priority", "Priority of the ylib fcoll component",
MCA_BASE_VAR_TYPE_INT, NULL, 0, 0,
OPAL_INFO_LVL_9,
MCA_BASE_VAR_SCOPE_READONLY, &mca_fcoll_ylib_priority);
mca_fcoll_ylib_num_io_procs = 1;
(void) mca_base_component_var_register(&mca_fcoll_ylib_component.fcollm_version,
"num_io_procs", "Number of writers in the ylib fcoll component",
MCA_BASE_VAR_TYPE_INT, NULL, 0, 0,
OPAL_INFO_LVL_9,
MCA_BASE_VAR_SCOPE_READONLY, &mca_fcoll_ylib_num_io_procs);
mca_fcoll_ylib_stripe_size = 1048576;
(void) mca_base_component_var_register(&mca_fcoll_ylib_component.fcollm_version,
"stripe_size", "Stripe Size of the ylib fcoll component",
MCA_BASE_VAR_TYPE_INT, NULL, 0, 0,
OPAL_INFO_LVL_9,
MCA_BASE_VAR_SCOPE_READONLY, &mca_fcoll_ylib_stripe_size);
mca_fcoll_ylib_blocks_per_cycle = 20;
(void) mca_base_component_var_register(&mca_fcoll_ylib_component.fcollm_version,
"blocks_per_cycle", "Blocks to write per cycle of the ylib fcoll component",
MCA_BASE_VAR_TYPE_INT, NULL, 0, 0,
OPAL_INFO_LVL_9,
MCA_BASE_VAR_SCOPE_READONLY, &mca_fcoll_ylib_blocks_per_cycle);
return OMPI_SUCCESS;
}

662
ompi/mca/fcoll/ylib/fcoll_ylib_file_read_all.c
Просмотреть файл

@ -1,662 +0,0 @@
/*
* Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
* University Research and Technology
* Corporation. All rights reserved.
* Copyright (c) 2004-2005 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$
*
* Additional copyrights may follow
*
* $HEADER$
*/
#include "ompi_config.h"
#include "fcoll_ylib.h"
#include "mpi.h"
#include "ompi/constants.h"
#include "ompi/mca/fcoll/fcoll.h"
#include "ompi/mca/io/ompio/io_ompio.h"
#include "ompi/mca/io/io.h"
#include "math.h"
#include "ompi/mca/pml/pml.h"
#include <unistd.h>
int
mca_fcoll_ylib_file_read_all (mca_io_ompio_file_t *fh,
void *buf,
int count,
struct ompi_datatype_t *datatype,
ompi_status_public_t *status)
{
size_t total_bytes_read = 0; /* total bytes that have been read*/
size_t total_bytes = 0; /* total bytes to be read */
size_t total_bytes_global = 0;
size_t bytes_per_cycle = 0; /* total read in each cycle by each process*/
size_t bytes_to_read_in_cycle = 0; /* left to be read in a cycle*/
size_t current_position = 0;
size_t max_data = 0;
size_t bytes_remaining = 0;
size_t bytes_rem = 0;
size_t prev_bytes_rem = 0;
int index = 0;
int current_index = 0;
int current = 0;
int previous = 0;
int cycles = 0;
int i=0, j=0, x=0, n=0;
int blocks = 0;
int bytes_left = 0;
/* array that contains the sorted indices of the global_iov */
int *sorted = NULL;
int *displs = NULL;
int *bytes_per_process = NULL;
int *bytes_received = NULL;
/* iovec structure and count of the buffer passed in */
uint32_t iov_count = 0;
struct iovec *decoded_iov = NULL;
int iov_index = 0;
char *receive_buf = NULL;
int global_fview_count = 0;
struct iovec *global_fview = NULL;
int local_count = 0;
struct iovec *iov = NULL;
int broken_count = 0;
struct iovec *broken_iovec = NULL;
int *fview_count = NULL;
int global_count = 0;
char *global_buf = NULL;
if (opal_datatype_is_contiguous_memory_layout(&datatype->super,1)) {
fh->f_flags |= OMPIO_CONTIGUOUS_MEMORY;
}
/**************************************************************************
** In case the data is not contigous in memory, decode it into an iovec **
**************************************************************************/
if (! (fh->f_flags & OMPIO_CONTIGUOUS_MEMORY)) {
ompi_io_ompio_decode_datatype (fh,
datatype,
count,
buf,
&max_data,
&decoded_iov,
&iov_count);
}
else {
max_data = count * datatype->super.size;
}
if (! (fh->f_flags & OMPIO_AGGREGATOR_IS_SET)) {
ompi_io_ompio_set_aggregator_props (fh,
mca_fcoll_ylib_num_io_procs,
max_data);
mca_fcoll_ylib_num_io_procs =
ceil((float)fh->f_size/fh->f_procs_per_group);
fh->f_aggregator_index =
ceil((float)fh->f_size/mca_fcoll_ylib_num_io_procs);
}
/*********************************************************************
*** Generate the File offsets/lengths corresponding to this read ***
********************************************************************/
ompi_io_ompio_generate_current_file_view (fh,
max_data,
&iov,
&local_count);
/*
for (i=0 ; i<local_count ; i++) {
printf("%d: OFFSET: %d LENGTH: %d\n",
fh->f_rank,
iov[i].iov_base,
iov[i].iov_len);
}
*/
/*************************************************************
* Breakdown the file view at each process per OST then send *
* each portion of the file view t0 the corresp aggregator *
*************************************************************/
ompi_io_ompio_break_file_view (fh,
iov,
local_count,
mca_fcoll_ylib_num_io_procs,
mca_fcoll_ylib_stripe_size,
&broken_iovec,
&broken_count);
/*
for (i=0 ; i<broken_count ; i++) {
printf("%d: OFFSET: %d LENGTH: %d\n",
fh->f_rank,
broken_iovec[i].iov_base,
broken_iovec[i].iov_len);
}
*/
if (NULL != iov) {
free (iov);
iov = NULL;
}
ompi_io_ompio_distribute_file_view (fh,
broken_iovec,
broken_count,
mca_fcoll_ylib_num_io_procs,
mca_fcoll_ylib_stripe_size,
&fview_count,
&global_fview,
&global_fview_count);
/*
for (i=0 ; i<global_fview_count ; i++) {
printf("%d: OFFSET: %d LENGTH: %d\n",
fh->f_rank,
global_fview[i].iov_base,
global_fview[i].iov_len);
}
*/
if (0 == fh->f_rank%fh->f_aggregator_index) {
if (global_fview_count) {
for (i=0 ; i<global_fview_count ; i++) {
total_bytes += global_fview[i].iov_len;
}
/* sort it */
sorted = (int *)malloc (global_fview_count * sizeof(int));
if (NULL == sorted) {
opal_output (1, "OUT OF MEMORY\n");
return OMPI_ERR_OUT_OF_RESOURCE;
}
ompi_io_ompio_sort_iovec (global_fview, global_fview_count, sorted);
}
bytes_per_process = (int *) malloc (fh->f_size * sizeof (int));
if (NULL == bytes_per_process) {
opal_output (1, "OUT OF MEMORY\n");
return OMPI_ERR_OUT_OF_RESOURCE;
}
displs = (int *) malloc (fh->f_size * sizeof (int));
if (NULL == displs) {
opal_output (1, "OUT OF MEMORY\n");
return OMPI_ERR_OUT_OF_RESOURCE;
}
}
fh->f_comm->c_coll.coll_allreduce (&total_bytes,
&total_bytes_global,
1,
MPI_DOUBLE,
MPI_MAX,
fh->f_comm,
fh->f_comm->c_coll.coll_allreduce_module);
bytes_received = (int *)malloc (mca_fcoll_ylib_num_io_procs * sizeof (int));
if (NULL == bytes_received) {
opal_output (1, "OUT OF MEMORY\n");
return OMPI_ERR_OUT_OF_RESOURCE;
}
bytes_per_cycle = mca_fcoll_ylib_stripe_size * mca_fcoll_ylib_blocks_per_cycle;
cycles = ceil ((float)total_bytes_global/bytes_per_cycle);
for (index = 0; index < cycles; index++) {
int k = 0;
size_t total_bytes_recv = 0;
size_t temp = 0;
global_count = 0;
memset(bytes_received, 0x0, mca_fcoll_ylib_num_io_procs*sizeof(int));
if (0 == fh->f_rank%fh->f_aggregator_index) {
memset(displs, 0x0, fh->f_size*sizeof(int));
memset(bytes_per_process, 0x0, fh->f_size*sizeof(int));
}
if (0 == fh->f_rank%fh->f_aggregator_index) {
if (total_bytes > bytes_per_cycle) {
bytes_to_read_in_cycle = bytes_per_cycle;
}
else {
bytes_to_read_in_cycle = total_bytes;
}
}
/*
printf ("****%d: Total_bytes: %d CYCLE %d Bytes %d OFFSET %d******\n",
fh->f_rank,
total_bytes,
index,
bytes_to_read_in_cycle,
fh->f_offset);
sleep(1);
*/
/**********************************************************
**Gather the Data from all the processes at the readers **
*********************************************************/
/* Calculate how much data will be contributed in this cycle
by each process*/
previous = current;
prev_bytes_rem = bytes_rem;
temp = bytes_per_cycle * mca_fcoll_ylib_num_io_procs * (index+1)
+ fh->f_offset;
while (current < broken_count) {
if (temp >=
(size_t)((OPAL_PTRDIFF_TYPE)broken_iovec[current].iov_base +
broken_iovec[current].iov_len)) {
k = ((OPAL_PTRDIFF_TYPE)broken_iovec[current].iov_base /
mca_fcoll_ylib_stripe_size) % mca_fcoll_ylib_num_io_procs;
if (bytes_rem) {
bytes_received[k] += bytes_rem;
total_bytes_recv += bytes_rem;
bytes_rem = 0;
}
else {
bytes_received[k] += broken_iovec[current].iov_len;
total_bytes_recv += broken_iovec[current].iov_len;
}
current ++;
}
else {
k = ((OPAL_PTRDIFF_TYPE)broken_iovec[current].iov_base /
mca_fcoll_ylib_stripe_size) % mca_fcoll_ylib_num_io_procs;
if (bytes_rem) {
bytes_received[k] += temp -
((broken_iovec[current].iov_len - bytes_rem) +
(OPAL_PTRDIFF_TYPE)broken_iovec[current].iov_base);
total_bytes_recv += temp -
((broken_iovec[current].iov_len - bytes_rem) +
(OPAL_PTRDIFF_TYPE)broken_iovec[current].iov_base);
bytes_rem -= temp -
((broken_iovec[current].iov_len - bytes_rem) +
(OPAL_PTRDIFF_TYPE)broken_iovec[current].iov_base);
break;
}
else {
if (temp > (size_t)broken_iovec[current].iov_base) {
bytes_received[k] += temp -
(OPAL_PTRDIFF_TYPE)broken_iovec[current].iov_base;
total_bytes_recv += temp -
(OPAL_PTRDIFF_TYPE)broken_iovec[current].iov_base;
bytes_rem = broken_iovec[current].iov_len -
(temp -
(OPAL_PTRDIFF_TYPE)broken_iovec[current].iov_base);
break;
}
else {
break;
}
}
}
}
/*
for (i=0 ; i<mca_fcoll_ylib_num_io_procs ; i++) {
printf ("%d sending %d to %d\n",fh->f_rank,
bytes_received[i], i);
}
sleep(1);
*/
if (0 == fh->f_rank%fh->f_aggregator_index && bytes_to_read_in_cycle) {
/* Calculate how much data will be recieved this cycle
by each aggregator*/
while (bytes_to_read_in_cycle) {
blocks = fview_count[0];
for (j=0 ; j<fh->f_size ; j++) {
if (sorted[current_index] < blocks) {
n = j;
break;
}
else {
blocks += fview_count[j+1];
}
}
if (bytes_remaining) {
if (bytes_remaining <= bytes_to_read_in_cycle) {
bytes_per_process[n] += bytes_remaining;
current_index ++;
bytes_to_read_in_cycle -= bytes_remaining;
bytes_remaining = 0;
continue;
}
else {
bytes_per_process[n] += bytes_to_read_in_cycle;
bytes_remaining -= bytes_to_read_in_cycle;
bytes_to_read_in_cycle = 0;
break;
}
}
else {
if (bytes_to_read_in_cycle <
global_fview[sorted[current_index]].iov_len) {
bytes_per_process[n] += bytes_to_read_in_cycle;
bytes_remaining =
global_fview[sorted[current_index]].iov_len -
bytes_to_read_in_cycle;
bytes_to_read_in_cycle = 0;
break;
}
else {
bytes_per_process[n] +=
global_fview[sorted[current_index]].iov_len;
bytes_to_read_in_cycle -=
global_fview[sorted[current_index]].iov_len;
current_index ++;
continue;
}
}
}
/*
for (i=0 ; i<fh->f_size ; i++) {
printf ("%d --> expecting %d from %d\n",fh->f_rank,
bytes_per_process[i], i);
}
*/
/* Calculate the displacement on where to put the data and allocate
the recieve buffer (global_buf) */
displs[0] = 0;
global_count = bytes_per_process[0];
for (i=1 ; i<fh->f_size ; i++) {
global_count += bytes_per_process[i];
displs[i] = displs[i-1] + bytes_per_process[i-1];
}
/*
for (i=0 ; i<fh->f_size ; i++) {
printf ("Proc %d sending %d at %d\n",
i,
bytes_per_process[i],
displs[i]);
}
*/
if (0 != global_count) {
global_buf = malloc (global_count);
if (NULL == global_buf) {
opal_output (1, "OUT OF MEMORY\n");
return OMPI_ERR_OUT_OF_RESOURCE;
}
}
}
/**********************************************************
******* Create the io array, and pass it to fbtl *********
*********************************************************/
if (0 == fh->f_rank%fh->f_aggregator_index && global_count) {
int bytes_to_read = global_count;
int *temp = NULL;
int block = 1;
k = 0;
temp = (int *)malloc (sizeof(int) * fh->f_size);
if (NULL == temp) {
opal_output(1, "OUT OF MEMORY\n");
return OMPI_ERR_OUT_OF_RESOURCE;
}
memset(temp, 0x0, fh->f_size*sizeof(int));
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) {
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");
return OMPI_ERR_OUT_OF_RESOURCE;
}
}
blocks = fview_count[0];
for (j=0 ; j<fh->f_size ; j++) {
if (sorted[x] < blocks) {
n = j;
break;
}
else {
blocks += fview_count[j+1];
}
}
for (j=0 ; j<n ; j++) {
start += bytes_per_process[j];
}
if (bytes_left) {
if (bytes_left <= bytes_to_read) {
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[n]];
temp[n] += (int)fh->f_io_array[k].length;
bytes_to_read -= 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_read;
fh->f_io_array[k].memory_address =
&global_buf[start+temp[n]];
temp[n] += (int)fh->f_io_array[k].length;
bytes_left -= bytes_to_read;
bytes_to_read = 0;;
k ++;
break;
}
}
else {
if (bytes_to_read < (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_read;
fh->f_io_array[k].memory_address =
&global_buf[start+temp[n]];
bytes_left =
global_fview[sorted[x]].iov_len - bytes_to_read;
bytes_to_read = 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[n]];
temp[n] += (int)fh->f_io_array[k].length;
bytes_to_read -= global_fview[sorted[x]].iov_len;
k ++;
x ++;
continue;
}
}
}
fh->f_num_of_io_entries = k;
/*
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);
}
*/
if (fh->f_num_of_io_entries) {
if (OMPI_SUCCESS != fh->f_fbtl->fbtl_preadv (fh, NULL)) {
opal_output (1, "READ FAILED\n");
return OMPI_ERROR;
}
}
if (NULL != temp) {
free (temp);
temp = NULL;
}
}
/**********************************************************
******************** DONE READING ************************
*********************************************************/
/**********************************************************
********* Scatter the Data from the readers **************
*********************************************************/
if (fh->f_flags & OMPIO_CONTIGUOUS_MEMORY) {
receive_buf = &((char*)buf)[total_bytes_read];
}
else if (total_bytes_recv) {
/* allocate a send buffer and copy the data that needs
to be sent into it in case the data is non-contigous
in memory */
receive_buf = malloc (total_bytes_recv);
if (NULL == receive_buf) {
opal_output (1, "OUT OF MEMORY\n");
return OMPI_ERR_OUT_OF_RESOURCE;
}
}
/* distribute the data to its corresponding processes */
ompi_io_ompio_scatter_data (fh,
receive_buf,
total_bytes_recv,
bytes_received,
broken_iovec,
previous,
prev_bytes_rem,
global_buf,
bytes_per_process,
displs,
mca_fcoll_ylib_num_io_procs,
mca_fcoll_ylib_stripe_size);
/*
if (0 == fh->f_rank%fh->f_aggregator_index) {
for (k=0 ; k<global_count/4 ; k++) {
printf ("%d: RECV %d \n",fh->f_rank,
((int *)global_buf)[k]);
}
}
*/
/* If data is not contigous in memory, copy the data from the
receive buffer into the buffer passed in */
if (!(fh->f_flags & OMPIO_CONTIGUOUS_MEMORY)) {
OPAL_PTRDIFF_TYPE mem_address;
size_t remaining = 0;
size_t temp_position = 0;
remaining = total_bytes_recv;
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;
}
}
total_bytes_read += total_bytes_recv;
total_bytes -= global_count;
/**********************************************************
**************** DONE GATHERING OF DATA ******************
*********************************************************/
if (0 == fh->f_rank%fh->f_aggregator_index) {
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 != global_buf) {
free (global_buf);
global_buf = NULL;
}
}
}
if (NULL != sorted) {
free (sorted);
sorted = NULL;
}
if (NULL != broken_iovec) {
free (broken_iovec);
broken_iovec = NULL;
}
if (NULL != global_fview) {
free (global_fview);
global_fview = NULL;
}
if (NULL != fview_count) {
free (fview_count);
fview_count = NULL;
}
if (NULL != decoded_iov) {
free (decoded_iov);
decoded_iov = NULL;
}
if (NULL != bytes_per_process) {
free (bytes_per_process);
bytes_per_process = NULL;
}
if (NULL != bytes_received) {
free (bytes_received);
bytes_received = NULL;
}
if (NULL != displs) {
free (displs);
displs = NULL;
}
/*
if (NULL != total_bytes_per_process) {
free (total_bytes_per_process);
total_bytes_per_process = NULL;
}
*/
return OMPI_SUCCESS;
}

36
ompi/mca/fcoll/ylib/fcoll_ylib_file_read_all_begin.c
Просмотреть файл

@ -1,36 +0,0 @@
/*
* Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
* University Research and Technology
* Corporation. All rights reserved.
* Copyright (c) 2004-2005 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$
*
* Additional copyrights may follow
*
* $HEADER$
*/
#include "ompi_config.h"
#include "fcoll_ylib.h"
#include "mpi.h"
#include "ompi/constants.h"
#include "ompi/mca/fcoll/fcoll.h"
int
mca_fcoll_ylib_file_read_all_begin (mca_io_ompio_file_t *fh,
void *buf,
int count,
struct ompi_datatype_t *datatype)
{
printf ("DYNAMIC READ ALL BEGIN\n");
return OMPI_SUCCESS;
}

35
ompi/mca/fcoll/ylib/fcoll_ylib_file_read_all_end.c
Просмотреть файл

@ -1,35 +0,0 @@
/*
* Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
* University Research and Technology
* Corporation. All rights reserved.
* Copyright (c) 2004-2005 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$
*
* Additional copyrights may follow
*
* $HEADER$
*/
#include "ompi_config.h"
#include "fcoll_ylib.h"
#include "mpi.h"
#include "ompi/constants.h"
#include "ompi/mca/fcoll/fcoll.h"
int
mca_fcoll_ylib_file_read_all_end (mca_io_ompio_file_t *fh,
void *buf,
ompi_status_public_t *status)
{
printf ("DYNAMIC READ ALL END\n");
return OMPI_SUCCESS;
}

706
ompi/mca/fcoll/ylib/fcoll_ylib_file_write_all.c
Просмотреть файл

@ -1,706 +0,0 @@
/*
* Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
* University Research and Technology
* Corporation. All rights reserved.
* Copyright (c) 2004-2005 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$
*
* Additional copyrights may follow
*
* $HEADER$
*/
#include "ompi_config.h"
#include "fcoll_ylib.h"
#include "mpi.h"
#include "ompi/constants.h"
#include "ompi/mca/fcoll/fcoll.h"
#include "ompi/mca/io/ompio/io_ompio.h"
#include "ompi/mca/io/io.h"
#include "math.h"
#include "ompi/mca/pml/pml.h"
#include <unistd.h>
#define TIME_BREAKDOWN 0
int
mca_fcoll_ylib_file_write_all (mca_io_ompio_file_t *fh,
void *buf,
int count,
struct ompi_datatype_t *datatype,
ompi_status_public_t *status)
{
size_t total_bytes_written = 0; /* total bytes that have been written*/
size_t total_bytes = 0; /* total bytes to be written */
size_t total_bytes_global = 0;
size_t bytes_per_cycle = 0; /* total written in each cycle by each process*/
size_t bytes_to_write_in_cycle = 0; /* left to be written in a cycle*/
size_t current_position = 0;
size_t max_data = 0;
size_t bytes_remaining = 0;
size_t bytes_rem = 0;
size_t prev_bytes_rem = 0;
int index = 0;
int current_index = 0;
int current = 0;
int previous = 0;
int cycles = 0;
int i=0, j=0, x=0, n=0;
int blocks = 0;
int bytes_left = 0;
/* array that contains the sorted indices of the global_iov */
int *sorted = NULL;
int *displs = NULL;
int *bytes_per_process = NULL;
int *bytes_sent = NULL;
/* iovec structure and count of the buffer passed in */
uint32_t iov_count = 0;
struct iovec *decoded_iov = NULL;
int iov_index = 0;
char *send_buf = NULL;
int global_fview_count = 0;
struct iovec *global_fview = NULL;
int local_count = 0;
struct iovec *iov = NULL;
int broken_count = 0;
struct iovec *broken_iovec = NULL;
int *fview_count = NULL;
int global_count = 0;
char *global_buf = NULL;
#if TIME_BREAKDOWN
double start_time=0, end_time=0, start_time2=0, end_time2=0;
double total=0 , total_io=0;
#endif
#if TIME_BREAKDOWN
if (0 == fh->f_rank%fh->f_aggregator_index) {
start_time = MPI_Wtime();
}
#endif
if (opal_datatype_is_contiguous_memory_layout(&datatype->super,1)) {
fh->f_flags |= OMPIO_CONTIGUOUS_MEMORY;
}
/**************************************************************************
** In case the data is not contigous in memory, decode it into an iovec **
**************************************************************************/
if (! (fh->f_flags & OMPIO_CONTIGUOUS_MEMORY)) {
ompi_io_ompio_decode_datatype (fh,
datatype,
count,
buf,
&max_data,
&decoded_iov,
&iov_count);
}
else {
max_data = count * datatype->super.size;
}
if (! (fh->f_flags & OMPIO_AGGREGATOR_IS_SET)) {
ompi_io_ompio_set_aggregator_props (fh,
mca_fcoll_ylib_num_io_procs,
max_data);
mca_fcoll_ylib_num_io_procs =
ceil((float)fh->f_size/fh->f_procs_per_group);
fh->f_aggregator_index =
ceil((float)fh->f_size/mca_fcoll_ylib_num_io_procs);
}
/*********************************************************************
*** Generate the File offsets/lengths corresponding to this write ***
********************************************************************/
ompi_io_ompio_generate_current_file_view (fh,
max_data,
&iov,
&local_count);
/*
for (i=0 ; i<local_count ; i++) {
printf("%d: OFFSET: %d LENGTH: %d\n",
fh->f_rank,
iov[i].iov_base,
iov[i].iov_len);
}
*/
/*************************************************************
* Breakdown the file view at each process per OST then send *
* each portion of the file view to the corresp aggregator *
*************************************************************/
ompi_io_ompio_break_file_view (fh,
iov,
local_count,
mca_fcoll_ylib_num_io_procs,
mca_fcoll_ylib_stripe_size,
&broken_iovec,
&broken_count);
/*
for (i=0 ; i<broken_count ; i++) {
printf("%d: OFFSET: %d LENGTH: %d\n",
fh->f_rank,
broken_iovec[i].iov_base,
broken_iovec[i].iov_len);
}
*/
if (NULL != iov) {
free (iov);
iov = NULL;
}
ompi_io_ompio_distribute_file_view (fh,
broken_iovec,
broken_count,
mca_fcoll_ylib_num_io_procs,
mca_fcoll_ylib_stripe_size,
&fview_count,
&global_fview,
&global_fview_count);
/*
for (i=0 ; i<global_fview_count ; i++) {
printf("%d: OFFSET: %d LENGTH: %d\n",
fh->f_rank,
global_fview[i].iov_base,
global_fview[i].iov_len);
}
*/
if (0 == fh->f_rank%fh->f_aggregator_index) {
if (global_fview_count) {
for (i=0 ; i<global_fview_count ; i++) {
total_bytes += global_fview[i].iov_len;
}
/* sort it */
sorted = (int *)malloc (global_fview_count * sizeof(int));
if (NULL == sorted) {
opal_output (1, "OUT OF MEMORY\n");
return OMPI_ERR_OUT_OF_RESOURCE;
}
ompi_io_ompio_sort_iovec (global_fview, global_fview_count, sorted);
}
bytes_per_process = (int *) malloc (fh->f_size * sizeof (int));
if (NULL == bytes_per_process) {
opal_output (1, "OUT OF MEMORY\n");
return OMPI_ERR_OUT_OF_RESOURCE;
}
displs = (int *) malloc (fh->f_size * sizeof (int));
if (NULL == displs) {
opal_output (1, "OUT OF MEMORY\n");
return OMPI_ERR_OUT_OF_RESOURCE;
}
}
fh->f_comm->c_coll.coll_allreduce (&total_bytes,
&total_bytes_global,
1,
MPI_DOUBLE,
MPI_MAX,
fh->f_comm,
fh->f_comm->c_coll.coll_allreduce_module);
bytes_sent = (int *)malloc (mca_fcoll_ylib_num_io_procs * sizeof (int));
if (NULL == bytes_sent) {
opal_output (1, "OUT OF MEMORY\n");
return OMPI_ERR_OUT_OF_RESOURCE;
}
bytes_per_cycle = mca_fcoll_ylib_stripe_size * mca_fcoll_ylib_blocks_per_cycle;
cycles = ceil ((float)total_bytes_global/bytes_per_cycle);
#if TIME_BREAKDOWN
if (0 == fh->f_rank%fh->f_aggregator_index) {
end_time = MPI_Wtime();
total = end_time-start_time;
printf ("%d: Preprocessing --- %f\n", fh->f_rank, total);
total = 0;
}
#endif
for (index = 0; index < cycles; index++) {
int k = 0;
size_t total_bytes_sent = 0;
size_t temp = 0;
global_count = 0;
#if TIME_BREAKDOWN
if (0 == fh->f_rank%fh->f_aggregator_index) {
start_time = MPI_Wtime();
}
#endif
memset(bytes_sent, 0x0, mca_fcoll_ylib_num_io_procs*sizeof(int));
if (0 == fh->f_rank%fh->f_aggregator_index) {
memset(displs, 0x0, fh->f_size*sizeof(int));
memset(bytes_per_process, 0x0, fh->f_size*sizeof(int));
if (total_bytes > bytes_per_cycle) {
bytes_to_write_in_cycle = bytes_per_cycle;
}
else {
bytes_to_write_in_cycle = total_bytes;
}
}
/*
printf ("****%d: Total_bytes: %d CYCLE %d Bytes %d OFFSET %d******\n",
fh->f_rank,
total_bytes,
index,
bytes_to_write_in_cycle,
fh->f_offset);
sleep(1);
*/
/**********************************************************
**Gather the Data from all the processes at the writers **
*********************************************************/
/* Calculate how much data will be contributed in this cycle
by each process*/
previous = current;
prev_bytes_rem = bytes_rem;
temp = bytes_per_cycle * mca_fcoll_ylib_num_io_procs * (index+1)
+ fh->f_offset;
while (current < broken_count) {
if (temp >=
(size_t)((OPAL_PTRDIFF_TYPE)broken_iovec[current].iov_base +
broken_iovec[current].iov_len)) {
k = ((OPAL_PTRDIFF_TYPE)broken_iovec[current].iov_base /
mca_fcoll_ylib_stripe_size) % mca_fcoll_ylib_num_io_procs;
if (bytes_rem) {
bytes_sent[k] += bytes_rem;
total_bytes_sent += bytes_rem;
bytes_rem = 0;
}
else {
bytes_sent[k] += broken_iovec[current].iov_len;
total_bytes_sent += broken_iovec[current].iov_len;
}
current ++;
}
else {
k = ((OPAL_PTRDIFF_TYPE)broken_iovec[current].iov_base /
mca_fcoll_ylib_stripe_size) % mca_fcoll_ylib_num_io_procs;
if (bytes_rem) {
bytes_sent[k] += temp -
((broken_iovec[current].iov_len - bytes_rem) +
(OPAL_PTRDIFF_TYPE)broken_iovec[current].iov_base);
total_bytes_sent += temp -
((broken_iovec[current].iov_len - bytes_rem) +
(OPAL_PTRDIFF_TYPE)broken_iovec[current].iov_base);
bytes_rem -= temp -
((broken_iovec[current].iov_len - bytes_rem) +
(OPAL_PTRDIFF_TYPE)broken_iovec[current].iov_base);
break;
}
else {
if (temp > (size_t)broken_iovec[current].iov_base) {
bytes_sent[k] += temp -
(OPAL_PTRDIFF_TYPE)broken_iovec[current].iov_base;
total_bytes_sent += temp -
(OPAL_PTRDIFF_TYPE)broken_iovec[current].iov_base;
bytes_rem = broken_iovec[current].iov_len -
(temp -
(OPAL_PTRDIFF_TYPE)broken_iovec[current].iov_base);
break;
}
else {
break;
}
}
}
}
/*
for (i=0 ; i<mca_fcoll_ylib_num_io_procs ; i++) {
if (bytes_sent[i]) {
printf ("%d sending %d to %d\n",fh->f_rank,
bytes_sent[i], i);
}
}
sleep(3);
*/
if (0 == fh->f_rank%fh->f_aggregator_index && bytes_to_write_in_cycle) {
/* Calculate how much data will be recieved this cycle
by each aggregator*/
while (bytes_to_write_in_cycle) {
blocks = fview_count[0];
for (j=0 ; j<fh->f_size ; j++) {
if (sorted[current_index] < blocks) {
n = j;
break;
}
else {
blocks += fview_count[j+1];
}
}
if (bytes_remaining) {
if (bytes_remaining <= bytes_to_write_in_cycle) {
bytes_per_process[n] += bytes_remaining;
current_index ++;
bytes_to_write_in_cycle -= bytes_remaining;
bytes_remaining = 0;
continue;
}
else {
bytes_per_process[n] += bytes_to_write_in_cycle;
bytes_remaining -= bytes_to_write_in_cycle;
bytes_to_write_in_cycle = 0;
break;
}
}
else {
if (bytes_to_write_in_cycle <
global_fview[sorted[current_index]].iov_len) {
bytes_per_process[n] += bytes_to_write_in_cycle;
bytes_remaining =
global_fview[sorted[current_index]].iov_len -
bytes_to_write_in_cycle;
bytes_to_write_in_cycle = 0;
break;
}
else {
bytes_per_process[n] +=
global_fview[sorted[current_index]].iov_len;
bytes_to_write_in_cycle -=
global_fview[sorted[current_index]].iov_len;
current_index ++;
continue;
}
}
}
/*
for (i=0 ; i<fh->f_size ; i++) {
printf ("%d --> expecting %d from %d\n",fh->f_rank,
bytes_per_process[i], i);
}
*/
/* Calculate the displacement on where to put the data and allocate
the recieve buffer (global_buf) */
displs[0] = 0;
global_count = bytes_per_process[0];
for (i=1 ; i<fh->f_size ; i++) {
global_count += bytes_per_process[i];
displs[i] = displs[i-1] + bytes_per_process[i-1];
}
/*
for (i=0 ; i<fh->f_size ; i++) {
printf ("Proc %d sending %d at %d\n",
i,
bytes_per_process[i],
displs[i]);
}
*/
if (0 != global_count) {
global_buf = malloc (global_count);
if (NULL == global_buf) {
opal_output (1, "OUT OF MEMORY\n");
return OMPI_ERR_OUT_OF_RESOURCE;
}
}
}
if (fh->f_flags & OMPIO_CONTIGUOUS_MEMORY) {
send_buf = &((char*)buf)[total_bytes_written];
}
else if (total_bytes_sent) {
/* allocate a send buffer and copy the data that needs
to be sent into it in case the data is non-contigous
in memory */
OPAL_PTRDIFF_TYPE mem_address;
size_t remaining = 0;
size_t temp_position = 0;
send_buf = malloc (total_bytes_sent);
if (NULL == send_buf) {
opal_output (1, "OUT OF MEMORY\n");
return OMPI_ERR_OUT_OF_RESOURCE;
}
remaining = total_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;
}
}
}
/* distribute the data to its corresponding aggregator */
ompi_io_ompio_gather_data (fh,
send_buf,
total_bytes_sent,
bytes_sent,
broken_iovec,
previous,
prev_bytes_rem,
global_buf,
bytes_per_process,
displs,
mca_fcoll_ylib_num_io_procs,
mca_fcoll_ylib_stripe_size);
/*
if (0 == fh->f_rank%fh->f_aggregator_index) {
for (k=0 ; k<global_count/4 ; k++) {
printf ("%d: RECV %d \n",fh->f_rank,
((int *)global_buf)[k]);
}
}
*/
if (! (fh->f_flags & OMPIO_CONTIGUOUS_MEMORY)) {
if (NULL != send_buf) {
free (send_buf);
send_buf = NULL;
}
}
total_bytes_written += total_bytes_sent;
total_bytes -= global_count;
/**********************************************************
**************** DONE GATHERING OF DATA ******************
*********************************************************/
/**********************************************************
******* Create the io array, and pass it to fbtl *********
*********************************************************/
if (0 == fh->f_rank%fh->f_aggregator_index && global_count) {
int bytes_to_write = global_count;
int *temp = NULL;
int block = 1;
k = 0;
temp = (int *)malloc (sizeof(int) * fh->f_size);
if (NULL == temp) {
opal_output(1, "OUT OF MEMORY\n");
return OMPI_ERR_OUT_OF_RESOURCE;
}
memset(temp, 0x0, fh->f_size*sizeof(int));
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) {
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");
return OMPI_ERR_OUT_OF_RESOURCE;
}
}
blocks = fview_count[0];
for (j=0 ; j<fh->f_size ; j++) {
if (sorted[x] < blocks) {
n = j;
break;
}
else {
blocks += fview_count[j+1];
}
}
for (j=0 ; j<n ; 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[n]];
temp[n] += (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[n]];
temp[n] += (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[n]];
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[n]];
temp[n] += (int)fh->f_io_array[k].length;
bytes_to_write -= global_fview[sorted[x]].iov_len;
k ++;
x ++;
continue;
}
}
}
fh->f_num_of_io_entries = k;
/*
printf("%d: *************************** %d\n", fh->f_rank, fh->f_num_of_io_entries);
for (i=0 ; i<fh->f_num_of_io_entries ; i++) {
printf(" ADDRESS: %p OFFSET: %d LENGTH: %d\n",
fh->f_io_array[i].memory_address,
fh->f_io_array[i].offset,
fh->f_io_array[i].length);
}
*/
#if TIME_BREAKDOWN
if (0 == fh->f_rank%fh->f_aggregator_index) {
start_time2 = MPI_Wtime();
}
#endif
if (fh->f_num_of_io_entries) {
if (OMPI_SUCCESS != fh->f_fbtl->fbtl_pwritev (fh, NULL)) {
opal_output (1, "WRITE FAILED\n");
return OMPI_ERROR;
}
}
#if TIME_BREAKDOWN
if (0 == fh->f_rank%fh->f_aggregator_index) {
end_time2 = MPI_Wtime();
total_io += end_time2-start_time2;
}
#endif
if (NULL != temp) {
free (temp);
temp = NULL;
}
}
/**********************************************************
******************** DONE WRITING ************************
*********************************************************/
if (0 == fh->f_rank%fh->f_aggregator_index) {
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 != global_buf) {
free (global_buf);
global_buf = NULL;
}
}
#if TIME_BREAKDOWN
if (0 == fh->f_rank%fh->f_aggregator_index) {
end_time = MPI_Wtime();
total += end_time-start_time;
}
#endif
}
#if TIME_BREAKDOWN
if (0 == fh->f_rank%fh->f_aggregator_index) {
printf ("%d: Total --- %f I/O ---- %f\n", fh->f_rank, total, total_io);
}
#endif
if (NULL != sorted) {
free (sorted);
sorted = NULL;
}
if (NULL != broken_iovec) {
free (broken_iovec);
broken_iovec = NULL;
}
if (NULL != global_fview) {
free (global_fview);
global_fview = NULL;
}
if (NULL != fview_count) {
free (fview_count);
fview_count = NULL;
}
if (NULL != decoded_iov) {
free (decoded_iov);
decoded_iov = NULL;
}
if (NULL != bytes_per_process) {
free (bytes_per_process);
bytes_per_process = NULL;
}
if (NULL != bytes_sent) {
free (bytes_sent);
bytes_sent = NULL;
}
if (NULL != displs) {
free (displs);
displs = NULL;
}
/*
if (NULL != total_bytes_per_process) {
free (total_bytes_per_process);
total_bytes_per_process = NULL;
}
*/
return OMPI_SUCCESS;
}

36
ompi/mca/fcoll/ylib/fcoll_ylib_file_write_all_begin.c
Просмотреть файл

@ -1,36 +0,0 @@
/*
* Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
* University Research and Technology
* Corporation. All rights reserved.
* Copyright (c) 2004-2005 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$
*
* Additional copyrights may follow
*
* $HEADER$
*/
#include "ompi_config.h"
#include "fcoll_ylib.h"
#include "mpi.h"
#include "ompi/constants.h"
#include "ompi/mca/fcoll/fcoll.h"
int
mca_fcoll_ylib_file_write_all_begin (mca_io_ompio_file_t *fh,
void *buf,
int count,
struct ompi_datatype_t *datatype)
{
printf ("DYNAMIC WRITE ALL BEGIN\n");
return OMPI_SUCCESS;
}

35
ompi/mca/fcoll/ylib/fcoll_ylib_file_write_all_end.c
Просмотреть файл

@ -1,35 +0,0 @@
/*
* Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
* University Research and Technology
* Corporation. All rights reserved.
* Copyright (c) 2004-2005 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$
*
* Additional copyrights may follow
*
* $HEADER$
*/
#include "ompi_config.h"
#include "fcoll_ylib.h"
#include "mpi.h"
#include "ompi/constants.h"
#include "ompi/mca/fcoll/fcoll.h"
int
mca_fcoll_ylib_file_write_all_end (mca_io_ompio_file_t *fh,
void *buf,
ompi_status_public_t *status)
{
printf ("DYNAMIC WRITE ALL END\n");
return OMPI_SUCCESS;
}

90
ompi/mca/fcoll/ylib/fcoll_ylib_module.c
Просмотреть файл

@ -1,90 +0,0 @@
/*
* Copyright (c) 2004-2007 The Trustees of Indiana University and Indiana
* University Research and Technology
* Corporation. All rights reserved.
* Copyright (c) 2004-2006 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$
*
* Additional copyrights may follow
*
* $HEADER$
*/
#include "ompi_config.h"
#include "fcoll_ylib.h"
#include <stdio.h>
#include "mpi.h"
#include "ompi/mca/fcoll/fcoll.h"
#include "ompi/mca/fcoll/base/base.h"
/*
* *******************************************************************
* ************************ actions structure ************************
* *******************************************************************
*/
static mca_fcoll_base_module_1_0_0_t ylib = {
mca_fcoll_ylib_module_init,
mca_fcoll_ylib_module_finalize,
mca_fcoll_ylib_file_read_all,
mca_fcoll_ylib_file_read_all_begin,
mca_fcoll_ylib_file_read_all_end,
mca_fcoll_ylib_file_write_all,
mca_fcoll_ylib_file_write_all_begin,
mca_fcoll_ylib_file_write_all_end
};
int
mca_fcoll_ylib_component_init_query(bool enable_progress_threads,
bool enable_mpi_threads)
{
/* Nothing to do */
return OMPI_SUCCESS;
}
mca_fcoll_base_module_1_0_0_t *
mca_fcoll_ylib_component_file_query (mca_io_ompio_file_t *fh, int *priority)
{
*priority = mca_fcoll_ylib_priority;
if (0 >= mca_fcoll_ylib_priority) {
return NULL;
}
if (mca_fcoll_base_query_table (fh, "ylib")) {
if (*priority < 50) {
*priority = 50;
}
}
return &ylib;
}
int mca_fcoll_ylib_component_file_unquery (mca_io_ompio_file_t *file)
{
/* This function might be needed for some purposes later. for now it
* does not have anything to do since there are no steps which need
* to be undone if this module is not selected */
return OMPI_SUCCESS;
}
int mca_fcoll_ylib_module_init (mca_io_ompio_file_t *file)
{
return OMPI_SUCCESS;
}
int mca_fcoll_ylib_module_finalize (mca_io_ompio_file_t *file)
{
return OMPI_SUCCESS;
}