X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/86f9a6c7bece8591b61752ee74a005e14fe3af50..d2fd3818a5e681f28d2ee11aa2c20c2d7dbabc03:/src/smpi/smpi_global.cpp diff --git a/src/smpi/smpi_global.cpp b/src/smpi/smpi_global.cpp index a46a06860f..1ad2a306b9 100644 --- a/src/smpi/smpi_global.cpp +++ b/src/smpi/smpi_global.cpp @@ -7,6 +7,7 @@ #include "private.h" #include "private.hpp" #include "simgrid/s4u/Mailbox.hpp" +#include "smpi/smpi_shared_malloc.hpp" #include "simgrid/sg_config.h" #include "src/kernel/activity/SynchroComm.hpp" #include "src/mc/mc_record.h" @@ -15,6 +16,7 @@ #include "src/simix/smx_private.h" #include "surf/surf.h" #include "xbt/replay.hpp" +#include #include /* DBL_MAX */ #include @@ -41,7 +43,7 @@ struct papi_process_data { #endif std::unordered_map location2speedup; -Process **process_data = nullptr; +simgrid::smpi::Process **process_data = nullptr; int process_count = 0; int smpi_universe_size = 0; int* index_to_process_data = nullptr; @@ -51,6 +53,10 @@ MPI_Comm MPI_COMM_WORLD = MPI_COMM_UNINITIALIZED; MPI_Errhandler *MPI_ERRORS_RETURN = nullptr; MPI_Errhandler *MPI_ERRORS_ARE_FATAL = nullptr; MPI_Errhandler *MPI_ERRHANDLER_NULL = nullptr; +static simgrid::config::Flag smpi_wtime_sleep( + "smpi/wtime", "Minimum time to inject inside a call to MPI_Wtime", 0.0); +static simgrid::config::Flag smpi_init_sleep( + "smpi/init", "Time to inject inside a call to MPI_Init", 0.0); void (*smpi_comm_copy_data_callback) (smx_activity_t, void*, size_t) = &smpi_comm_copy_buffer_callback; @@ -61,13 +67,13 @@ int smpi_process_count() return process_count; } -Process* smpi_process() +simgrid::smpi::Process* smpi_process() { simgrid::MsgActorExt* msgExt = static_cast(SIMIX_process_self()->data); - return static_cast(msgExt->data); + return static_cast(msgExt->data); } -Process* smpi_process_remote(int index) +simgrid::smpi::Process* smpi_process_remote(int index) { return process_data[index_to_process_data[index]]; } @@ -77,7 +83,7 @@ MPI_Comm smpi_process_comm_self(){ } void smpi_process_init(int *argc, char ***argv){ - Process::init(argc, argv); + simgrid::smpi::Process::init(argc, argv); } int smpi_process_index(){ @@ -98,31 +104,66 @@ void smpi_comm_set_copy_data_callback(void (*callback) (smx_activity_t, void*, s smpi_comm_copy_data_callback = callback; } +void print(std::vector> vec) { + fprintf(stderr, "{"); + for(auto elt: vec) { + fprintf(stderr, "(0x%x, 0x%x),", elt.first, elt.second); + } + stderr, fprintf(stderr, "}\n"); +} +void memcpy_private(void *dest, const void *src, size_t n, std::vector> &private_blocks) { + for(auto block : private_blocks) { + memcpy((uint8_t*)dest+block.first, (uint8_t*)src+block.first, block.second-block.first); + } +} + void smpi_comm_copy_buffer_callback(smx_activity_t synchro, void *buff, size_t buff_size) { + simgrid::kernel::activity::Comm *comm = dynamic_cast(synchro); + int src_shared=0, dst_shared=0; + int src_offset, dst_offset; + std::vector> src_private_blocks; + std::vector> dst_private_blocks; XBT_DEBUG("Copy the data over"); + if(src_shared=smpi_is_shared(buff, src_private_blocks, &src_offset)) { + XBT_DEBUG("Sender %p is shared. Let's ignore it.", buff); + src_private_blocks = shift_private_blocks(src_private_blocks, src_offset); + } + else { + src_private_blocks.clear(); + src_private_blocks.push_back(std::make_pair(0, buff_size)); + } + if(dst_shared=smpi_is_shared((char*)comm->dst_buff, dst_private_blocks, &dst_offset)) { + XBT_DEBUG("Receiver %p is shared. Let's ignore it.", (char*)comm->dst_buff); + dst_private_blocks = shift_private_blocks(dst_private_blocks, dst_offset); + } + else { + dst_private_blocks.clear(); + dst_private_blocks.push_back(std::make_pair(0, buff_size)); + } + auto private_blocks = merge_private_blocks(src_private_blocks, dst_private_blocks); void* tmpbuff=buff; - simgrid::kernel::activity::Comm *comm = dynamic_cast(synchro); - if((smpi_privatize_global_variables) && (static_cast(buff) >= smpi_start_data_exe) && (static_cast(buff) < smpi_start_data_exe + smpi_size_data_exe ) ){ XBT_DEBUG("Privatization : We are copying from a zone inside global memory... Saving data to temp buffer !"); smpi_switch_data_segment( - (static_cast((static_cast(comm->src_proc->data)->data))->index())); + (static_cast((static_cast(comm->src_proc->data)->data))->index())); tmpbuff = static_cast(xbt_malloc(buff_size)); - memcpy(tmpbuff, buff, buff_size); + memcpy_private(tmpbuff, buff, buff_size, private_blocks); } if((smpi_privatize_global_variables) && ((char*)comm->dst_buff >= smpi_start_data_exe) && ((char*)comm->dst_buff < smpi_start_data_exe + smpi_size_data_exe )){ XBT_DEBUG("Privatization : We are copying to a zone inside global memory - Switch data segment"); smpi_switch_data_segment( - (static_cast((static_cast(comm->dst_proc->data)->data))->index())); + (static_cast((static_cast(comm->dst_proc->data)->data))->index())); } - memcpy(comm->dst_buff, tmpbuff, buff_size); + XBT_DEBUG("Copying %zu bytes from %p to %p", buff_size, tmpbuff,comm->dst_buff); + memcpy_private(comm->dst_buff, tmpbuff, buff_size, private_blocks); + if (comm->detached) { // if this is a detached send, the source buffer was duplicated by SMPI // sender to make the original buffer available to the application ASAP @@ -131,8 +172,8 @@ void smpi_comm_copy_buffer_callback(smx_activity_t synchro, void *buff, size_t b //xbt_free(comm->comm.src_data);// inside SMPI the request is kept inside the user data and should be free comm->src_buff = nullptr; } - if(tmpbuff!=buff)xbt_free(tmpbuff); + } void smpi_comm_null_copy_buffer_callback(smx_activity_t comm, void *buff, size_t buff_size) @@ -267,15 +308,15 @@ void smpi_global_init() smpirun=1; } smpi_universe_size = process_count; - process_data = new Process*[process_count]; + process_data = new simgrid::smpi::Process*[process_count]; for (i = 0; i < process_count; i++) { - process_data[i] = new Process(i); + process_data[i] = new simgrid::smpi::Process(i); } //if the process was launched through smpirun script we generate a global mpi_comm_world //if not, we let MPI_COMM_NULL, and the comm world will be private to each mpi instance if(smpirun){ - group = new Group(process_count); - MPI_COMM_WORLD = new Comm(group, nullptr); + group = new simgrid::smpi::Group(process_count); + MPI_COMM_WORLD = new simgrid::smpi::Comm(group, nullptr); MPI_Attr_put(MPI_COMM_WORLD, MPI_UNIVERSE_SIZE, reinterpret_cast(process_count)); msg_bar_t bar = MSG_barrier_init(process_count); @@ -291,6 +332,7 @@ void smpi_global_destroy() int count = smpi_process_count(); smpi_bench_destroy(); + smpi_shared_destroy(); if (MPI_COMM_WORLD != MPI_COMM_UNINITIALIZED){ delete MPI_COMM_WORLD->group(); MSG_barrier_destroy(process_data[0]->finalization_barrier()); @@ -299,10 +341,10 @@ void smpi_global_destroy() } for (int i = 0; i < count; i++) { if(process_data[i]->comm_self()!=MPI_COMM_NULL){ - Comm::destroy(process_data[i]->comm_self()); + simgrid::smpi::Comm::destroy(process_data[i]->comm_self()); } if(process_data[i]->comm_intra()!=MPI_COMM_NULL){ - Comm::destroy(process_data[i]->comm_intra()); + simgrid::smpi::Comm::destroy(process_data[i]->comm_intra()); } xbt_os_timer_free(process_data[i]->timer()); xbt_mutex_destroy(process_data[i]->mailboxes_mutex()); @@ -313,9 +355,9 @@ void smpi_global_destroy() if (MPI_COMM_WORLD != MPI_COMM_UNINITIALIZED){ MPI_COMM_WORLD->cleanup_smp(); - MPI_COMM_WORLD->cleanup_attr(); - if(Colls::smpi_coll_cleanup_callback!=nullptr) - Colls::smpi_coll_cleanup_callback(); + MPI_COMM_WORLD->cleanup_attr(); + if(simgrid::smpi::Colls::smpi_coll_cleanup_callback!=nullptr) + simgrid::smpi::Colls::smpi_coll_cleanup_callback(); delete MPI_COMM_WORLD; } @@ -342,7 +384,7 @@ void __attribute__ ((weak)) user_main_() int __attribute__ ((weak)) smpi_simulated_main_(int argc, char **argv) { - Process::init(&argc, &argv); + simgrid::smpi::Process::init(&argc, &argv); user_main_(); return 0; } @@ -370,7 +412,6 @@ static void smpi_init_logs(){ XBT_LOG_CONNECT(smpi); /* Keep this line as soon as possible in this function: xbt_log_appender_file.c depends on it DO NOT connect this in XBT or so, or it will be useless to xbt_log_appender_file.c */ XBT_LOG_CONNECT(instr_smpi); - XBT_LOG_CONNECT(smpi_base); XBT_LOG_CONNECT(smpi_bench); XBT_LOG_CONNECT(smpi_coll); XBT_LOG_CONNECT(smpi_colls); @@ -386,14 +427,15 @@ static void smpi_init_logs(){ XBT_LOG_CONNECT(smpi_request); XBT_LOG_CONNECT(smpi_replay); XBT_LOG_CONNECT(smpi_rma); + XBT_LOG_CONNECT(smpi_shared); XBT_LOG_CONNECT(smpi_utils); } } static void smpi_init_options(){ - Colls::set_collectives(); - Colls::smpi_coll_cleanup_callback=nullptr; + simgrid::smpi::Colls::set_collectives(); + simgrid::smpi::Colls::smpi_coll_cleanup_callback=nullptr; smpi_cpu_threshold = xbt_cfg_get_double("smpi/cpu-threshold"); smpi_host_speed = xbt_cfg_get_double("smpi/host-speed"); smpi_privatize_global_variables = xbt_cfg_get_boolean("smpi/privatize-global-variables"); @@ -502,3 +544,29 @@ void SMPI_init(){ void SMPI_finalize(){ smpi_global_destroy(); } + +void smpi_mpi_init() { + if(smpi_init_sleep > 0) + simcall_process_sleep(smpi_init_sleep); +} + +double smpi_mpi_wtime(){ + double time; + if (smpi_process()->initialized() != 0 && smpi_process()->finalized() == 0 && smpi_process()->sampling() == 0) { + smpi_bench_end(); + time = SIMIX_get_clock(); + // to avoid deadlocks if used as a break condition, such as + // while (MPI_Wtime(...) < time_limit) { + // .... + // } + // because the time will not normally advance when only calls to MPI_Wtime + // are made -> deadlock (MPI_Wtime never reaches the time limit) + if(smpi_wtime_sleep > 0) + simcall_process_sleep(smpi_wtime_sleep); + smpi_bench_begin(); + } else { + time = SIMIX_get_clock(); + } + return time; +} +