X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/9c9f17b4ac25fe6fcf026724a7d3cc4978b775ed..16bda2500ecf72bb920f4131e431b9f802ffc15a:/src/smpi/smpi_global.cpp diff --git a/src/smpi/smpi_global.cpp b/src/smpi/smpi_global.cpp index db497bfdbf..f11aae6f66 100644 --- a/src/smpi/smpi_global.cpp +++ b/src/smpi/smpi_global.cpp @@ -3,6 +3,14 @@ /* This program is free software; you can redistribute it and/or modify it * under the terms of the license (GNU LGPL) which comes with this package. */ +#include +#include +#include +#include +#include +#include +#include + #include "mc/mc.h" #include "private.h" #include "private.hpp" @@ -13,6 +21,7 @@ #include "src/mc/mc_replay.h" #include "src/msg/msg_private.h" #include "src/simix/smx_private.h" +#include "src/surf/surf_interface.hpp" #include "surf/surf.h" #include "xbt/replay.hpp" #include @@ -24,12 +33,25 @@ #include #include #include +#include #include +#include 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 + +/* Mac OSX does not have any header file providing that definition so we have to duplicate it here. Bummers. */ +extern char** environ; /* we use it in posix_spawnp below */ + #if HAVE_PAPI #include "papi.h" const char* papi_default_config_name = "default"; @@ -105,40 +127,48 @@ void smpi_comm_set_copy_data_callback(void (*callback) (smx_activity_t, void*, s void smpi_comm_copy_buffer_callback(smx_activity_t synchro, void *buff, size_t buff_size) { - XBT_DEBUG("Copy the data over"); - 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 !"); + XBT_DEBUG("Copy the data over"); + if(smpi_is_shared(buff)){ + XBT_DEBUG("Sender %p is shared. Let's ignore it.", buff); + }else if(smpi_is_shared((char*)comm->dst_buff)){ + XBT_DEBUG("Receiver %p is shared. Let's ignore it.", (char*)comm->dst_buff); + }else{ + 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(tmpbuff, buff, buff_size); + } - 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 == 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())); + } - 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); - XBT_DEBUG("Copying %zu bytes from %p to %p", buff_size, tmpbuff,comm->dst_buff); - memcpy(comm->dst_buff, tmpbuff, buff_size); - 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 (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(tmpbuff!=buff)xbt_free(tmpbuff); } void smpi_comm_null_copy_buffer_callback(smx_activity_t comm, void *buff, size_t buff_size) @@ -169,9 +199,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(); @@ -268,27 +296,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); - } } } @@ -297,6 +326,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()); @@ -332,43 +362,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 */ @@ -391,6 +391,7 @@ 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); } } @@ -401,7 +402,26 @@ static void smpi_init_options(){ 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/privatize-global-variables"); + 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/privatize-global-variables: %s", + smpi_privatize_option); + if (smpi_cpu_threshold < 0) smpi_cpu_threshold = DBL_MAX; @@ -418,7 +438,61 @@ static void smpi_init_options(){ } } -int smpi_main(int (*realmain) (int argc, char *argv[]), int argc, char *argv[]) +static int execute_command(const char * const argv[]) +{ + pid_t pid; + int status; + if (posix_spawnp(&pid, argv[0], nullptr, nullptr, (char* const*) argv, environ) != 0) + return 127; + if (waitpid(pid, &status, 0) != pid) + return 127; + return status; +} + +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)(void); + +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() ? const_cast(""): &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_point2 = + (smpi_fortran_entry_point_type) dlsym(handle, "user_main_"); + if (entry_point2 != nullptr) { + // fprintf(stderr, "EP user_main_=%p\n", entry_point2); + return [entry_point2](int argc, char** argv) { + smpi_process_init(&argc, &argv); + entry_point2(); + return 0; + }; + } + + smpi_c_entry_point_type entry_point = (smpi_c_entry_point_type) dlsym(handle, "main"); + if (entry_point != nullptr) { + // fprintf(stderr, "EP main=%p\n", entry_point); + return entry_point; + } + + return smpi_entry_point_type(); +} + +int smpi_main(const char* executable, int argc, char *argv[]) { srand(SMPI_RAND_SEED); @@ -427,31 +501,83 @@ 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(); - // 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(); + if (smpi_privatize_global_variables == SMPI_PRIVATIZE_DLOPEN) { - if(smpi_privatize_global_variables) - smpi_initialize_global_memory_segments(); + std::string executable_copy = executable; + + // Prepare the copy of the binary (open the file and get its size) + // (fdin will remain open for the whole process execution. That's a sort of leak but we can live with it) + int fdin = open(executable_copy.c_str(), O_RDONLY); + xbt_assert(fdin >= 0, "Cannot read from %s", executable_copy.c_str()); + struct stat fdin_stat; + fstat(fdin, &fdin_stat); + off_t fdin_size = fdin_stat.st_size; + + simix_global->default_function = [executable_copy, fdin, fdin_size](std::vector args) { + return std::function([executable_copy, fdin, fdin_size, args] { + + // Copy the dynamic library: + std::string target_executable = executable_copy + + "_" + std::to_string(getpid()) + + "_" + std::to_string(rank++) + ".so"; + + int fdout = open(target_executable.c_str(), O_WRONLY); + xbt_assert(fdout >= 0, "Cannot write into %s", target_executable.c_str()); + + sendfile(fdout, fdin, NULL, fdin_size); + close(fdout); + + // Load the copy and resolve the entry point: + void* handle = dlopen(target_executable.c_str(), RTLD_LAZY | RTLD_LOCAL | RTLD_DEEPBIND); + unlink(target_executable.c_str()); + if (handle == nullptr) + xbt_die("dlopen failed"); + 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", executable); + 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); @@ -491,16 +617,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(); }