diff options
Diffstat (limited to 'src/Panda/CSDIO_Shared.C')
-rw-r--r-- | src/Panda/CSDIO_Shared.C | 241 |
1 files changed, 241 insertions, 0 deletions
diff --git a/src/Panda/CSDIO_Shared.C b/src/Panda/CSDIO_Shared.C new file mode 100644 index 0000000..35e864d --- /dev/null +++ b/src/Panda/CSDIO_Shared.C @@ -0,0 +1,241 @@ +#include "definitions.h" +#include "ArrayGroup.h" +#include "MPIFS.h" +#include "Chunk.h" +#include "App_Info.h" +#include "Array.h" +#include "message.h" +#include "CSDIO_Shared.h" + +/* we could have made this class multiply inherit from CSDIO and CSDIO_Shared, but + * we would have to use virtual inheritance and depending on the compiler used, + * there could be a performance penalty (though it would still be dwarfed by the + * cost of message-passing and disk i/o) + */ + +extern MPIFS* MPIFS_global_obj; +extern int SUBCHUNK_SIZE; + +CSDIO_Shared::CSDIO_Shared(int *schema_string, int schema_size, int world_rank, + int comp_app_num,int comp_app_size , App_Info *app_info) +: CSDIO(schema_string, schema_size, world_rank, comp_app_num, + comp_app_size, app_info) +{ + + compute_chunk_ = new Chunk(); + current_chunk_ = new Chunk(); + subchunk_ = new Chunk(); + current_array_id_ = -1; + if ((op_type_ == RESTART)||(op_type_ == GENERAL_READ)|| + (op_type_ == READ_TIMESTEP)) + read_op_ = YES; + else + read_op_ = NO; + + /* Send the IO app info to the compute nodes */ + receive_io_app_info(); + + /* We need to set the following variables so that continue_io()* + * would start the I/O of the first subchunk automatically */ + contiguous_ = NO; + current_array_id_ = -1; + current_chunk_id_ = 0; + num_of_chunks_ = -1; /* This will cause get_next_chunk() to fail */ + current_subchunk_id_ = 0; + num_of_subchunks_ = -1; /* Causes get_next_subchunk() to fail */ + status_flag_ = START; + continue_io(); +} + +CSDIO_Shared::~CSDIO_Shared() +{ + if (subchunk_) delete subchunk_; + if (compute_chunk_) delete compute_chunk_; + subchunk_ = compute_chunk_ = NULL; +} + +Boolean CSDIO_Shared::get_next_array(){ + current_array_id_++; + if (current_array_id_ < num_of_arrays_){ + make_subchunks_ = -1; + current_array_ = find_array(current_array_id_); + nat_chunked_ = current_array_->nat_chunked(); + sub_chunked_ = current_array_->sub_chunked(); + array_rank_ = current_array_->rank(); + + if (array_rank_ > max_rank_){ + realloc_schema_bufs(array_rank_); + } + num_of_chunks_ = current_array_->layout(IO_NODE)->total_elements(); + current_chunk_id_ = -1; + if (nat_chunked_ && !sub_chunked_) + contiguous_ = YES; /* No need to use derived datatypes */ + else + contiguous_ = NO; /* Have to use derived datatypes */ + + bytes_to_go_ = 0; + current_subchunk_id_ = -1; + return YES; + } else + return NO; +} + + +Boolean CSDIO_Shared::get_next_chunk() +{ + int *ptr; + + if (!current_array_) return NO; + current_chunk_id_ = current_array_->get_next_index(current_chunk_id_, + my_io_rank_, + num_io_nodes_); + if (current_chunk_id_ < num_of_chunks_){ + current_chunk_->set_data_ptr(NULL); + current_chunk_->init(current_array_, current_chunk_id_, + IO_NODE, NO_ALLOC); + tag_ = current_chunk_id_*1000; + if (contiguous_){ + bytes_to_go_ = current_chunk_->total_size_in_bytes(); + current_chunk_->set_data_ptr(mem_buf_); + ptr = schema_bufs_[0]; + *ptr++ = current_array_id_; + *ptr++ = current_chunk_id_; + *ptr++ = (int) nat_chunked_; + *ptr++ = (int) contiguous_; + *ptr++ = op_type_; + *ptr++ = 0; + *ptr++ = 0; + compute_chunk_overlaps(current_array_, current_chunk_); + } + else { + if (!sub_chunked_ && (make_subchunks_ == -1)){ + current_array_->make_sub_chunks(current_chunk_); + make_subchunks_ = 1; + } + num_of_subchunks_ = current_array_->layout(SUB_CHUNK)->total_elements(); + current_subchunk_id_ = -1; + } + return YES; + } + else + return NO; +} + + +/* This should not be called for the contiguous_ case */ +Boolean CSDIO_Shared::get_next_subchunk() +{ + current_subchunk_id_++; + if (current_subchunk_id_ < num_of_subchunks_){ + subchunk_->set_data_ptr(NULL); + subchunk_->init(current_chunk_, current_subchunk_id_, NO_ALLOC); + bytes_to_go_ = subchunk_->total_size_in_bytes(); + + if (bytes_to_go_ < mem_buf_size_) + realloc_mem_bufs(bytes_to_go_); + + subchunk_->set_data_ptr(mem_buf_); + return YES; + } + else + return NO; +} + + +void CSDIO_Shared::start_subchunk_io() +{ + int *ptr; + + if (contiguous_){ + ptr = schema_bufs_[0]; + ptr[6] = min(SUBCHUNK_SIZE, bytes_to_go_); + if (read_op_) read_data(mem_buf_, ptr[6]); + if (read_op_) + nb_send_message((void *)mem_buf_, ptr[6], MPI_CHAR, dest_ids_[0], + tag_+CHUNK_DATA_FROM_IO, MPI_COMM_WORLD, &requests_[0]); + else + nb_receive_message((void *)mem_buf_, ptr[6], MPI_CHAR, dest_ids_[0], + tag_+CHUNK_DATA_TO_IO, MPI_COMM_WORLD, &requests_[0]); + ptr[5] += ptr[6]; /* Offset of the next subchunk */ + bytes_to_go_ -= ptr[6]; + status_flag_ = WAITING; + tag_ += 10; + + } else { + compute_chunk_overlaps(current_array_, subchunk_); + + compute_schemas(current_array_, subchunk_, compute_chunk_, current_array_id_); + if (read_op_){ + read_data(subchunk_); + send_data_to_compute_nodes(subchunk_, tag_); + } + else + receive_data_from_compute_nodes(subchunk_, tag_); + status_flag_ = WAITING; + tag_ += 10; + } +} + + +Boolean CSDIO_Shared::test_subchunk_io() +{ + int flag; + MPI_Testall(num_overlaps_, requests_, &flag, statuses_); + if (flag) { + status_flag_ = START; + if (!read_op_) + if (contiguous_) + write_data(mem_buf_, schema_bufs_[0][6], 1); + else + write_data(subchunk_); + + if (!contiguous_) free_datatypes(); + return YES; + } + return NO; +} + + +/* Return YES, if I/O is complete */ +Boolean CSDIO_Shared::continue_io() +{ + if (status_flag_ == START){ + if (!start_next_subchunk_io()) return YES; /* IO completed */ + } else if (status_flag_ == WAITING){ + if (test_subchunk_io()) + if (!start_next_subchunk_io()) return YES; /* IO done */ + } else { + printf("Error - Invalid status_flag value \n"); + exit(11); + } + return NO; +} + +/* Return yes if you can start the io of another subchunk */ +Boolean CSDIO_Shared::start_next_subchunk_io() +{ + if (contiguous_){ + if (bytes_to_go_ <= 0){ + while(!get_next_chunk()){ + if (!get_next_array()) return NO; + } + /* Since we might be looking at another array */ + if (!contiguous_) get_next_subchunk(); + } + + start_subchunk_io(); + } else { + + if (!get_next_subchunk()){ + /* We have finished this chunk */ + while(!get_next_chunk()){ + if (!get_next_array()) return NO; + } + if (!contiguous_) get_next_subchunk(); + } + + start_subchunk_io(); + } + return YES; +} + |