X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/aaa9acf3fdd6714a158a161cac64c82992d52899..c04b15f5346d042218b4fb8357cace04585e067b:/src/smpi/smpi_global.cpp diff --git a/src/smpi/smpi_global.cpp b/src/smpi/smpi_global.cpp index 0acc0d4d59..63f2a6cdbe 100644 --- a/src/smpi/smpi_global.cpp +++ b/src/smpi/smpi_global.cpp @@ -4,32 +4,43 @@ * under the terms of the license (GNU LGPL) which comes with this package. */ #include "mc/mc.h" -#include "private.h" -#include "private.hpp" #include "simgrid/s4u/Mailbox.hpp" -#include "simgrid/sg_config.h" -#include "src/kernel/activity/SynchroComm.hpp" -#include "src/mc/mc_record.h" -#include "src/mc/mc_replay.h" +#include "simgrid/s4u/Host.hpp" #include "src/msg/msg_private.h" #include "src/simix/smx_private.h" -#include "surf/surf.h" -#include "xbt/replay.hpp" -#include - +#include "src/surf/surf_interface.hpp" +#include "src/smpi/SmpiHost.hpp" +#include "xbt/config.hpp" +#include "src/smpi/private.h" +#include "smpi/smpi_shared_malloc.hpp" +#include "src/smpi/smpi_coll.hpp" +#include "src/smpi/smpi_comm.hpp" +#include "src/smpi/smpi_group.hpp" +#include "src/smpi/smpi_info.hpp" +#include "src/smpi/smpi_process.hpp" + +#include +#include +#include #include /* DBL_MAX */ #include -#include -#include -#include -#include -#include -#include + +#if HAVE_SENDFILE +#include +#endif XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_kernel, smpi, "Logging specific to SMPI (kernel)"); #include #include /* trim_right / trim_left */ +#ifndef RTLD_DEEPBIND +/* RTLD_DEEPBIND is a bad idea of GNU ld that obviously does not exist on other platforms + * See https://www.akkadia.org/drepper/dsohowto.pdf + * and https://lists.freebsd.org/pipermail/freebsd-current/2016-March/060284.html +*/ +#define RTLD_DEEPBIND 0 +#endif + #if HAVE_PAPI #include "papi.h" const char* papi_default_config_name = "default"; @@ -59,8 +70,6 @@ static simgrid::config::Flag smpi_init_sleep( void (*smpi_comm_copy_data_callback) (smx_activity_t, void*, size_t) = &smpi_comm_copy_buffer_callback; - - int smpi_process_count() { return process_count; @@ -68,7 +77,10 @@ int smpi_process_count() simgrid::smpi::Process* smpi_process() { - simgrid::MsgActorExt* msgExt = static_cast(SIMIX_process_self()->data); + smx_actor_t me = SIMIX_process_self(); + if (me == nullptr) // This happens sometimes (eg, when linking against NS3 because it pulls openMPI...) + return nullptr; + simgrid::MsgActorExt* msgExt = static_cast(me->data); return static_cast(msgExt->data); } @@ -89,6 +101,14 @@ int smpi_process_index(){ return smpi_process()->index(); } +void * smpi_process_get_user_data(){ + return smpi_process()->get_user_data(); +} + +void smpi_process_set_user_data(void *data){ + return smpi_process()->set_user_data(data); +} + int smpi_global_size() { @@ -103,49 +123,88 @@ void smpi_comm_set_copy_data_callback(void (*callback) (smx_activity_t, void*, s smpi_comm_copy_data_callback = callback; } -void smpi_comm_copy_buffer_callback(smx_activity_t synchro, void *buff, size_t buff_size) +static void print(std::vector> vec) { + std::fprintf(stderr, "{"); + for (auto elt : vec) { + std::fprintf(stderr, "(0x%zx, 0x%zx),", elt.first, elt.second); + } + std::fprintf(stderr, "}\n"); +} +static void memcpy_private(void* dest, const void* src, std::vector>& private_blocks) { + for(auto block : private_blocks) { + memcpy((uint8_t*)dest+block.first, (uint8_t*)src+block.first, block.second-block.first); + } +} - simgrid::kernel::activity::Comm *comm = dynamic_cast(synchro); +static void check_blocks(std::vector> &private_blocks, size_t buff_size) { + for(auto block : private_blocks) { + xbt_assert(block.first <= block.second && block.second <= buff_size, "Oops, bug in shared malloc."); + } +} +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; + int dst_shared = 0; + size_t src_offset = 0; + size_t dst_offset = 0; + std::vector> src_private_blocks; + std::vector> dst_private_blocks; XBT_DEBUG("Copy the data over"); - if(smpi_is_shared(buff)){ + if((src_shared=smpi_is_shared(buff, src_private_blocks, &src_offset))) { XBT_DEBUG("Sender %p is shared. Let's ignore it.", buff); - }else if(smpi_is_shared((char*)comm->dst_buff)){ + src_private_blocks = shift_and_frame_private_blocks(src_private_blocks, src_offset, buff_size); + } + 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); - }else{ - void* tmpbuff=buff; - 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())); - tmpbuff = static_cast(xbt_malloc(buff_size)); - memcpy(tmpbuff, buff, buff_size); - } - - 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())); - } - - XBT_DEBUG("Copying %zu bytes from %p to %p", buff_size, tmpbuff,comm->dst_buff); - memcpy(comm->dst_buff, tmpbuff, buff_size); + dst_private_blocks = shift_and_frame_private_blocks(dst_private_blocks, dst_offset, buff_size); + } + else { + dst_private_blocks.clear(); + dst_private_blocks.push_back(std::make_pair(0, buff_size)); + } + check_blocks(src_private_blocks, buff_size); + check_blocks(dst_private_blocks, buff_size); + auto private_blocks = merge_private_blocks(src_private_blocks, dst_private_blocks); + check_blocks(private_blocks, buff_size); + void* tmpbuff=buff; + if((smpi_privatize_global_variables == SMPI_PRIVATIZE_MMAP) && (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()); + tmpbuff = static_cast(xbt_malloc(buff_size)); + memcpy_private(tmpbuff, buff, 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 - xbt_free(buff); - //It seems that the request is used after the call there this should be free somewhere else but where??? - //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); + if((smpi_privatize_global_variables == SMPI_PRIVATIZE_MMAP) && ((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()); + } + XBT_DEBUG("Copying %zu bytes from %p to %p", buff_size, tmpbuff,comm->dst_buff); + memcpy_private(comm->dst_buff, tmpbuff, 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 + xbt_free(buff); + //It seems that the request is used after the call there this should be free somewhere else but where??? + //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); } @@ -177,9 +236,7 @@ int smpi_enabled() { void smpi_global_init() { - int i; MPI_Group group; - int smpirun=0; if (!MC_is_active()) { global_timer = xbt_os_timer_new(); @@ -276,27 +333,28 @@ void smpi_global_init() } } #endif + + int smpirun = 0; + msg_bar_t finalization_barrier = nullptr; if (process_count == 0){ process_count = SIMIX_process_count(); smpirun=1; + finalization_barrier = MSG_barrier_init(process_count); } smpi_universe_size = process_count; process_data = new simgrid::smpi::Process*[process_count]; - for (i = 0; i < process_count; i++) { - process_data[i] = new simgrid::smpi::Process(i); + for (int i = 0; i < process_count; i++) { + process_data[i] = new simgrid::smpi::Process(i, finalization_barrier); } //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){ + if (smpirun) { 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); - for (i = 0; i < process_count; i++) { + for (int i = 0; i < process_count; i++) group->set_mapping(i, i); - process_data[i]->set_finalization_barrier(bar); - } } } @@ -341,43 +399,13 @@ void smpi_global_destroy() } xbt_free(index_to_process_data); - if(smpi_privatize_global_variables) + if(smpi_privatize_global_variables == SMPI_PRIVATIZE_MMAP) smpi_destroy_global_memory_segments(); smpi_free_static(); } extern "C" { -#ifndef WIN32 - -void __attribute__ ((weak)) user_main_() -{ - xbt_die("Should not be in this smpi_simulated_main"); -} - -int __attribute__ ((weak)) smpi_simulated_main_(int argc, char **argv) -{ - simgrid::smpi::Process::init(&argc, &argv); - user_main_(); - return 0; -} - -inline static int smpi_main_wrapper(int argc, char **argv){ - int ret = smpi_simulated_main_(argc,argv); - if(ret !=0){ - XBT_WARN("SMPI process did not return 0. Return value : %d", ret); - smpi_process()->set_return_value(ret); - } - return 0; -} - -int __attribute__ ((weak)) main(int argc, char **argv) -{ - return smpi_main(smpi_main_wrapper, argc, argv); -} - -#endif - static void smpi_init_logs(){ /* Connect log categories. See xbt/log.c */ @@ -406,12 +434,39 @@ static void smpi_init_logs(){ } static void smpi_init_options(){ - + //return if already called + if (smpi_cpu_threshold > -1) + return; 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"); + const char* smpi_privatize_option = xbt_cfg_get_string("smpi/privatization"); + if (std::strcmp(smpi_privatize_option, "no") == 0) + smpi_privatize_global_variables = SMPI_PRIVATIZE_NONE; + else if (std::strcmp(smpi_privatize_option, "yes") == 0) + smpi_privatize_global_variables = SMPI_PRIVATIZE_DEFAULT; + else if (std::strcmp(smpi_privatize_option, "mmap") == 0) + smpi_privatize_global_variables = SMPI_PRIVATIZE_MMAP; + else if (std::strcmp(smpi_privatize_option, "dlopen") == 0) + smpi_privatize_global_variables = SMPI_PRIVATIZE_DLOPEN; + + // Some compatibility stuff: + else if (std::strcmp(smpi_privatize_option, "1") == 0) + smpi_privatize_global_variables = SMPI_PRIVATIZE_DEFAULT; + else if (std::strcmp(smpi_privatize_option, "0") == 0) + smpi_privatize_global_variables = SMPI_PRIVATIZE_NONE; + + else + xbt_die("Invalid value for smpi/privatization: '%s'", smpi_privatize_option); + +#if defined(__FreeBSD__) + if (smpi_privatize_global_variables == SMPI_PRIVATIZE_MMAP) { + XBT_INFO("Mixing mmap privatization is broken on FreeBSD, switching to dlopen privatization instead."); + smpi_privatize_global_variables = SMPI_PRIVATIZE_DLOPEN; + } +#endif + if (smpi_cpu_threshold < 0) smpi_cpu_threshold = DBL_MAX; @@ -428,7 +483,47 @@ static void smpi_init_options(){ } } -int smpi_main(int (*realmain) (int argc, char *argv[]), int argc, char *argv[]) +typedef std::function smpi_entry_point_type; +typedef int (* smpi_c_entry_point_type)(int argc, char **argv); +typedef void (*smpi_fortran_entry_point_type)(); + +static int smpi_run_entry_point(smpi_entry_point_type entry_point, std::vector args) +{ + const int argc = args.size(); + std::unique_ptr argv(new char*[argc + 1]); + for (int i = 0; i != argc; ++i) + argv[i] = args[i].empty() ? '\0' : &args[i].front(); + argv[argc] = nullptr; + + int res = entry_point(argc, argv.get()); + if (res != 0){ + XBT_WARN("SMPI process did not return 0. Return value : %d", res); + smpi_process()->set_return_value(res); + } + return 0; +} + +// TODO, remove the number of functions involved here +static smpi_entry_point_type smpi_resolve_function(void* handle) +{ + smpi_fortran_entry_point_type entry_point_fortran = (smpi_fortran_entry_point_type)dlsym(handle, "user_main_"); + if (entry_point_fortran != nullptr) { + return [entry_point_fortran](int argc, char** argv) { + smpi_process_init(&argc, &argv); + entry_point_fortran(); + return 0; + }; + } + + smpi_c_entry_point_type entry_point = (smpi_c_entry_point_type)dlsym(handle, "main"); + if (entry_point != nullptr) { + return entry_point; + } + + return smpi_entry_point_type(); +} + +int smpi_main(const char* executable, int argc, char *argv[]) { srand(SMPI_RAND_SEED); @@ -437,31 +532,115 @@ int smpi_main(int (*realmain) (int argc, char *argv[]), int argc, char *argv[]) * configuration tools */ return 0; } - smpi_init_logs(); TRACE_global_init(&argc, argv); - TRACE_add_start_function(TRACE_smpi_alloc); - TRACE_add_end_function(TRACE_smpi_release); SIMIX_global_init(&argc, argv); MSG_init(&argc,argv); SMPI_switch_data_segment = &smpi_switch_data_segment; - smpi_init_options(); + simgrid::s4u::Host::onCreation.connect([](simgrid::s4u::Host& host) { + host.extension_set(new simgrid::smpi::SmpiHost(&host)); + }); // parse the platform file: get the host list SIMIX_create_environment(argv[1]); - SIMIX_comm_set_copy_data_callback(smpi_comm_copy_data_callback); - SIMIX_function_register_default(realmain); - SIMIX_launch_application(argv[2]); + SIMIX_comm_set_copy_data_callback(smpi_comm_copy_buffer_callback); - smpi_global_init(); + static std::size_t rank = 0; - smpi_check_options(); + smpi_init_options(); - if(smpi_privatize_global_variables) - smpi_initialize_global_memory_segments(); + if (smpi_privatize_global_variables == SMPI_PRIVATIZE_DLOPEN) { + + std::string executable_copy = executable; + + // Prepare the copy of the binary (get its size) + struct stat fdin_stat; + stat(executable_copy.c_str(), &fdin_stat); + off_t fdin_size = fdin_stat.st_size; + + simix_global->default_function = [executable_copy, fdin_size](std::vector args) { + return std::function([executable_copy, fdin_size, args] { + + // Copy the dynamic library: + std::string target_executable = executable_copy + + "_" + std::to_string(getpid()) + + "_" + std::to_string(rank++) + ".so"; + + int fdin = open(executable_copy.c_str(), O_RDONLY); + xbt_assert(fdin >= 0, "Cannot read from %s", executable_copy.c_str()); + int fdout = open(target_executable.c_str(), O_CREAT | O_RDWR, S_IRWXU); + xbt_assert(fdout >= 0, "Cannot write into %s", target_executable.c_str()); + +#if HAVE_SENDFILE + ssize_t sent_size = sendfile(fdout, fdin, NULL, fdin_size); + xbt_assert(sent_size == fdin_size, + "Error while copying %s: only %zd bytes copied instead of %ld (errno: %d -- %s)", + target_executable.c_str(), sent_size, fdin_size, errno, strerror(errno)); +#else + XBT_VERB("Copy %d bytes into %s", static_cast(fdin_size), target_executable.c_str()); + const int bufsize = 1024 * 1024 * 4; + char buf[bufsize]; + while (int got = read(fdin, buf, bufsize)) { + if (got == -1) { + xbt_assert(errno == EINTR, "Cannot read from %s", executable_copy.c_str()); + } else { + char* p = buf; + int todo = got; + while (int done = write(fdout, p, todo)) { + if (done == -1) { + xbt_assert(errno == EINTR, "Cannot write into %s", target_executable.c_str()); + } else { + p += done; + todo -= done; + } + } + } + } +#endif + close(fdin); + close(fdout); + + // Load the copy and resolve the entry point: + void* handle = dlopen(target_executable.c_str(), RTLD_LAZY | RTLD_LOCAL | RTLD_DEEPBIND); + if (xbt_cfg_get_boolean("smpi/keep-temps") == false) + unlink(target_executable.c_str()); + if (handle == nullptr) + xbt_die("dlopen failed: %s (errno: %d -- %s)", dlerror(), errno, strerror(errno)); + smpi_entry_point_type entry_point = smpi_resolve_function(handle); + if (!entry_point) + xbt_die("Could not resolve entry point"); + + smpi_run_entry_point(entry_point, args); + }); + }; + + } + else { + + // Load the dynamic library and resolve the entry point: + void* handle = dlopen(executable, RTLD_LAZY | RTLD_LOCAL | RTLD_DEEPBIND); + if (handle == nullptr) + xbt_die("dlopen failed for %s: %s (errno: %d -- %s)", executable, dlerror(), errno, strerror(errno)); + smpi_entry_point_type entry_point = smpi_resolve_function(handle); + if (!entry_point) + xbt_die("main not found in %s", executable); + // TODO, register the executable for SMPI privatization + + // Execute the same entry point for each simulated process: + simix_global->default_function = [entry_point](std::vector args) { + return std::function([entry_point, args] { + smpi_run_entry_point(entry_point, args); + }); + }; + + } + + SIMIX_launch_application(argv[2]); + + SMPI_init(); /* Clean IO before the run */ fflush(stdout); @@ -487,8 +666,8 @@ int smpi_main(int (*realmain) (int argc, char *argv[]), int argc, char *argv[]) } } int count = smpi_process_count(); - int i, ret=0; - for (i = 0; i < count; i++) { + int ret = 0; + for (int i = 0; i < count; i++) { if(process_data[i]->return_value()!=0){ ret=process_data[i]->return_value();//return first non 0 value break; @@ -501,16 +680,15 @@ int smpi_main(int (*realmain) (int argc, char *argv[]), int argc, char *argv[]) return ret; } -// This function can be called from extern file, to initialize logs, options, and processes of smpi -// without the need of smpirun +// Called either directly from the user code, or from the code called by smpirun void SMPI_init(){ smpi_init_logs(); smpi_init_options(); smpi_global_init(); smpi_check_options(); - if (TRACE_is_enabled() && TRACE_is_configured()) - TRACE_smpi_alloc(); - if(smpi_privatize_global_variables) + TRACE_smpi_alloc(); + simgrid::surf::surfExitCallbacks.connect(TRACE_smpi_release); + if(smpi_privatize_global_variables == SMPI_PRIVATIZE_MMAP) smpi_initialize_global_memory_segments(); }