X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/9e2f30eca15bf2c12a75bf0d59ee82f7c4553193..872d65264714799b25eb231609c3f05bae3d03dc:/src/smpi/smpi_base.cpp diff --git a/src/smpi/smpi_base.cpp b/src/smpi/smpi_base.cpp index 687a6d83ed..eeaada79e5 100644 --- a/src/smpi/smpi_base.cpp +++ b/src/smpi/smpi_base.cpp @@ -4,6 +4,8 @@ /* 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 "private.h" #include "xbt/virtu.h" #include "mc/mc.h" @@ -15,6 +17,8 @@ #include "simgrid/sg_config.h" #include "colls/colls.h" +#include "src/simix/SynchroComm.hpp" + XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_base, smpi, "Logging specific to SMPI (base)"); static int match_recv(void* a, void* b, smx_synchro_t ignored) { @@ -68,13 +72,17 @@ typedef struct s_smpi_factor_multival { // FIXME: this should be merged (dedupli int nb_values; double values[4];//arbitrary set to 4 } s_smpi_factor_multival_t; + xbt_dynar_t smpi_os_values = NULL; xbt_dynar_t smpi_or_values = NULL; xbt_dynar_t smpi_ois_values = NULL; -double smpi_wtime_sleep = 0.0; -double smpi_iprobe_sleep = 1e-4; -double smpi_test_sleep = 1e-4; +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_iprobe_sleep( + "smpi/iprobe", "Minimum time to inject inside a call to MPI_Iprobe", 1e-4); +static simgrid::config::Flag smpi_test_sleep( + "smpi/test", "Minimum time to inject inside a call to MPI_Test", 1e-4); static int factor_cmp(const void *pa, const void *pb) { @@ -252,10 +260,12 @@ static MPI_Request build_request(void *buf, int count, MPI_Datatype datatype, in request->old_type = datatype; request->size = smpi_datatype_size(datatype) * count; + smpi_datatype_use(datatype); request->src = src; request->dst = dst; request->tag = tag; request->comm = comm; + smpi_comm_use(request->comm); request->action = NULL; request->flags = flags; request->detached = 0; @@ -272,7 +282,6 @@ static MPI_Request build_request(void *buf, int count, MPI_Datatype datatype, in request->op = MPI_REPLACE; request->send = 0; request->recv = 0; - // if (flags & SEND) smpi_datatype_unuse(datatype); return request; } @@ -334,7 +343,7 @@ void smpi_mpi_start(MPI_Request request) if (request->flags & RECV) { print_request("New recv", request); - int async_small_thresh = xbt_cfg_get_int("smpi/async_small_thresh"); + int async_small_thresh = xbt_cfg_get_int("smpi/async-small-thresh"); xbt_mutex_t mut = smpi_process_mailboxes_mutex(); if (async_small_thresh != 0 ||request->flags & RMA) @@ -382,8 +391,6 @@ void smpi_mpi_start(MPI_Request request) // we make a copy here, as the size is modified by simix, and we may reuse the request in another receive later request->real_size=request->size; - smpi_datatype_use(request->old_type); - smpi_comm_use(request->comm); request->action = simcall_comm_irecv(SIMIX_process_self(), mailbox, request->buf, &request->real_size, &match_recv, !smpi_process_get_replaying()? &smpi_comm_copy_buffer_callback : &smpi_comm_null_copy_buffer_callback, request, -1.0); @@ -412,7 +419,7 @@ void smpi_mpi_start(MPI_Request request) XBT_DEBUG("sending size of %zu : sleep %f ", request->size, smpi_os(request->size)); } - int async_small_thresh = xbt_cfg_get_int("smpi/async_small_thresh"); + int async_small_thresh = xbt_cfg_get_int("smpi/async-small-thresh"); xbt_mutex_t mut=smpi_process_remote_mailboxes_mutex(receiver); @@ -449,7 +456,7 @@ void smpi_mpi_start(MPI_Request request) void* buf = request->buf; if ( (! (request->flags & SSEND)) && ((request->flags & RMA) || - (static_cast(request->size) < xbt_cfg_get_int("smpi/send_is_detached_thresh")))) { + (static_cast(request->size) < xbt_cfg_get_int("smpi/send-is-detached-thresh")))) { void *oldbuf = NULL; request->detached = 1; XBT_DEBUG("Send request %p is detached", request); @@ -472,8 +479,6 @@ void smpi_mpi_start(MPI_Request request) // we make a copy here, as the size is modified by simix, and we may reuse the request in another receive later request->real_size=request->size; - smpi_datatype_use(request->old_type); - smpi_comm_use(request->comm); request->action = simcall_comm_isend(SIMIX_process_from_PID(request->src+1), mailbox, request->size, -1.0, buf, request->real_size, &match_send, &xbt_free_f, // how to free the userdata if a detached send fails @@ -508,6 +513,8 @@ void smpi_mpi_request_free(MPI_Request * request) if((*request)->refcount<0) xbt_die("wrong refcount"); if((*request)->refcount==0){ + smpi_datatype_unuse((*request)->old_type); + smpi_comm_unuse((*request)->comm); print_request("Destroying", (*request)); xbt_free(*request); *request = MPI_REQUEST_NULL; @@ -530,6 +537,7 @@ MPI_Request smpi_rma_send_init(void *buf, int count, MPI_Datatype datatype, int request = build_request(buf==MPI_BOTTOM ? (void*)0 : buf, count, datatype, src, dst, tag, comm, RMA | NON_PERSISTENT | ISEND | SEND | PREPARED | ACCUMULATE); request->op = op; +// smpi_datatype_use(datatype); } return request; } @@ -545,6 +553,7 @@ MPI_Request smpi_rma_recv_init(void *buf, int count, MPI_Datatype datatype, int request = build_request(buf==MPI_BOTTOM ? (void*)0 : buf, count, datatype, src, dst, tag, comm, RMA | NON_PERSISTENT | RECV | PREPARED | ACCUMULATE); request->op = op; +// smpi_datatype_use(datatype); } return request; } @@ -686,14 +695,13 @@ static void finish_wait(MPI_Request * request, MPI_Status * status) if(req->flags & RECV) subtype->unserialize(req->buf, req->old_buf, req->real_size/smpi_datatype_size(datatype) , datatype->substruct, req->op); - if(req->detached == 0) free(req->buf); + xbt_free(req->buf); }else if(req->flags & RECV){//apply op on contiguous buffer for accumulate int n =req->real_size/smpi_datatype_size(datatype); smpi_op_apply(req->op, req->buf, req->old_buf, &n, &datatype); + xbt_free(req->buf); } } - smpi_comm_unuse(req->comm); - smpi_datatype_unuse(req->old_type); } if (TRACE_smpi_view_internals()) { @@ -705,8 +713,6 @@ static void finish_wait(MPI_Request * request, MPI_Status * status) } if(req->detached_sender!=NULL){ - smpi_datatype_unuse(req->detached_sender->old_type); - smpi_comm_unuse(req->detached_sender->comm); smpi_mpi_request_free(&(req->detached_sender)); } if(req->flags & PERSISTENT) @@ -838,7 +844,7 @@ void smpi_mpi_iprobe(int source, int tag, MPI_Comm comm, int* flag, MPI_Status* print_request("New iprobe", request); // We have to test both mailboxes as we don't know if we will receive one one or another - if (xbt_cfg_get_int("smpi/async_small_thresh")>0){ + if (xbt_cfg_get_int("smpi/async-small-thresh")>0){ mailbox = smpi_process_mailbox_small(); XBT_DEBUG("trying to probe the perm recv mailbox"); request->action = simcall_comm_iprobe(mailbox, 0, request->src, request->tag, &match_recv, (void*)request); @@ -850,7 +856,8 @@ void smpi_mpi_iprobe(int source, int tag, MPI_Comm comm, int* flag, MPI_Status* } if (request->action){ - MPI_Request req = (MPI_Request)SIMIX_comm_get_src_data(request->action); + simgrid::simix::Comm *sync_comm = static_cast(request->action); + MPI_Request req = (MPI_Request)sync_comm->src_data; *flag = 1; if(status != MPI_STATUS_IGNORE && !(req->flags & PREPARED)) { status->MPI_SOURCE = smpi_group_rank(smpi_comm_group(comm), req->src); @@ -877,14 +884,10 @@ void smpi_mpi_wait(MPI_Request * request, MPI_Status * status) return; } - if ((*request)->action != NULL) { // this is not a detached send + if ((*request)->action != NULL) + // this is not a detached send simcall_comm_wait((*request)->action, -1.0); - if((MC_is_active() || MC_record_replay_is_active()) && (*request)->action) - (*request)->action->comm.dst_data = NULL; // dangling pointer : dst_data is freed with a wait, need to set it to - // NULL for system state comparison - } - finish_wait(request, status); if (*request != MPI_REQUEST_NULL && ((*request)->flags & NON_PERSISTENT)) *request = MPI_REQUEST_NULL; @@ -1308,10 +1311,7 @@ void smpi_mpi_reduce(void *sendbuf, void *recvbuf, int count, MPI_Datatype datat void **tmpbufs; char* sendtmpbuf = (char*) sendbuf; - if( sendbuf == MPI_IN_PLACE ) { - sendtmpbuf = (char *)smpi_get_tmp_sendbuffer(count*smpi_datatype_get_extent(datatype)); - smpi_datatype_copy(recvbuf, count, datatype,sendtmpbuf, count, datatype); - } + rank = smpi_comm_rank(comm); size = smpi_comm_size(comm); @@ -1320,6 +1320,11 @@ void smpi_mpi_reduce(void *sendbuf, void *recvbuf, int count, MPI_Datatype datat smpi_coll_tuned_reduce_ompi_basic_linear(sendtmpbuf, recvbuf, count, datatype, op, root, comm); return; } + + if( sendbuf == MPI_IN_PLACE ) { + sendtmpbuf = (char *)smpi_get_tmp_sendbuffer(count*smpi_datatype_get_extent(datatype)); + smpi_datatype_copy(recvbuf, count, datatype,sendtmpbuf, count, datatype); + } if(rank != root) { // Send buffer to root @@ -1367,9 +1372,9 @@ void smpi_mpi_reduce(void *sendbuf, void *recvbuf, int count, MPI_Datatype datat xbt_free(tmpbufs); xbt_free(requests); - if( sendbuf == MPI_IN_PLACE ) { - smpi_free_tmp_buffer(sendtmpbuf); - } + } + if( sendbuf == MPI_IN_PLACE ) { + smpi_free_tmp_buffer(sendtmpbuf); } }