From: markls Date: Wed, 29 Aug 2007 23:50:04 +0000 (+0000) Subject: moved some more data structures into private.h. Still having problem where X-Git-Tag: v3.3~1233 X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/commitdiff_plain/a6bb32631746cb8f7744196398e11d9d499f7d8b?hp=c8304965b869fd1264284ea426f02dcd111fb62b moved some more data structures into private.h. Still having problem where suspended tasks do not seem to want to resume... git-svn-id: svn+ssh://scm.gforge.inria.fr/svn/simgrid/simgrid/trunk@4127 48e7efb5-ca39-0410-a469-dd3cf9ba447f --- diff --git a/src/smpi/private.h b/src/smpi/private.h index 091fb009d7..5d0814142f 100644 --- a/src/smpi/private.h +++ b/src/smpi/private.h @@ -1,27 +1,41 @@ #ifndef SMPI_PRIVATE_H #define SMPI_PRIVATE_H -#include "simix/simix.h" #include "xbt/mallocator.h" #include "xbt/xbt_os_time.h" + +#include "simix/simix.h" + #include "smpi/smpi.h" #define SMPI_DEFAULT_SPEED 100 #define SMPI_REQUEST_MALLOCATOR_SIZE 100 #define SMPI_MESSAGE_MALLOCATOR_SIZE 100 -typedef struct smpi_mpi_communicator_simdata { +typedef struct smpi_mpi_communicator_t { + int size; smx_host_t *hosts; smx_process_t *processes; int barrier_count; smx_mutex_t barrier_mutex; smx_cond_t barrier_cond; -} s_smpi_mpi_communicator_simdata_t; +} s_smpi_mpi_communicator_t; + +typedef struct smpi_mpi_request_t { + smpi_mpi_communicator_t comm; + int src; + int dst; + int tag; + + void *buf; + smpi_mpi_datatype_t datatype; + int count; + + short int completed :1; -typedef struct smpi_mpi_request_simdata { smx_mutex_t mutex; smx_cond_t cond; -} s_smpi_mpi_request_simdata_t; +} s_smpi_mpi_request_t; typedef struct SMPI_Global { @@ -58,19 +72,19 @@ typedef struct SMPI_Global { } s_SMPI_Global_t, *SMPI_Global_t; extern SMPI_Global_t smpi_global; -struct smpi_received_message_t { - smpi_mpi_communicator_t *comm; +typedef struct smpi_received_message_t { + smpi_mpi_communicator_t comm; int src; int dst; int tag; void *buf; -}; -typedef struct smpi_received_message_t smpi_received_message_t; +} s_smpi_received_message_t; +typedef struct smpi_received_message_t *smpi_received_message_t; // function prototypes -int smpi_mpi_comm_size(smpi_mpi_communicator_t *comm); -int smpi_mpi_comm_rank(smpi_mpi_communicator_t *comm, smx_host_t host); -int smpi_mpi_comm_rank_self(smpi_mpi_communicator_t *comm); +int smpi_mpi_comm_size(smpi_mpi_communicator_t comm); +int smpi_mpi_comm_rank(smpi_mpi_communicator_t comm, smx_host_t host); +int smpi_mpi_comm_rank_self(smpi_mpi_communicator_t comm); int smpi_mpi_comm_world_rank_self(void); int smpi_sender(int argc, char **argv); int smpi_receiver(int argc, char **argv); @@ -89,11 +103,11 @@ void smpi_mpi_init(void); void smpi_mpi_finalize(void); void smpi_bench_begin(void); void smpi_bench_end(void); -void smpi_barrier(smpi_mpi_communicator_t *comm); -int smpi_create_request(void *buf, int count, smpi_mpi_datatype_t *datatype, - int src, int dst, int tag, smpi_mpi_communicator_t *comm, smpi_mpi_request_t **request); -int smpi_isend(smpi_mpi_request_t *request); -int smpi_irecv(smpi_mpi_request_t *request); -void smpi_wait(smpi_mpi_request_t *request, smpi_mpi_status_t *status); +void smpi_barrier(smpi_mpi_communicator_t comm); +int smpi_create_request(void *buf, int count, smpi_mpi_datatype_t datatype, + int src, int dst, int tag, smpi_mpi_communicator_t comm, smpi_mpi_request_t *request); +int smpi_isend(smpi_mpi_request_t request); +int smpi_irecv(smpi_mpi_request_t request); +int smpi_wait(smpi_mpi_request_t request, smpi_mpi_status_t *status); #endif diff --git a/src/smpi/smpi_base.c b/src/smpi/smpi_base.c index 87b1c35dac..7f0fd9d1a2 100644 --- a/src/smpi/smpi_base.c +++ b/src/smpi/smpi_base.c @@ -12,22 +12,22 @@ void smpi_mpi_sum_func(void *x, void *y, void *z) *(int *)z = *(int *)x + *(int *)y; } -int inline smpi_mpi_comm_size(smpi_mpi_communicator_t *comm) +int inline smpi_mpi_comm_size(smpi_mpi_communicator_t comm) { return comm->size; } // FIXME: smarter algorithm? -int smpi_mpi_comm_rank(smpi_mpi_communicator_t *comm, smx_host_t host) +int smpi_mpi_comm_rank(smpi_mpi_communicator_t comm, smx_host_t host) { int i; - for(i = comm->size - 1; i > 0 && host != comm->simdata->hosts[i]; i--); + for(i = comm->size - 1; i > 0 && host != comm->hosts[i]; i--); return i; } -int inline smpi_mpi_comm_rank_self(smpi_mpi_communicator_t *comm) +int inline smpi_mpi_comm_rank_self(smpi_mpi_communicator_t comm) { return smpi_mpi_comm_rank(comm, SIMIX_host_self()); } @@ -55,28 +55,27 @@ void smpi_mpi_init() smpi_mpi_global = xbt_new(s_SMPI_MPI_Global_t, 1); // global communicator - smpi_mpi_global->mpi_comm_world = xbt_new(smpi_mpi_communicator_t, 1); - smpi_mpi_global->mpi_comm_world->size = size; - smpi_mpi_global->mpi_comm_world->simdata = xbt_new(s_smpi_mpi_communicator_simdata_t, 1); - smpi_mpi_global->mpi_comm_world->simdata->barrier_count = 0; - smpi_mpi_global->mpi_comm_world->simdata->barrier_mutex = SIMIX_mutex_init(); - smpi_mpi_global->mpi_comm_world->simdata->barrier_cond = SIMIX_cond_init(); - smpi_mpi_global->mpi_comm_world->simdata->hosts = hosts; - smpi_mpi_global->mpi_comm_world->simdata->processes = xbt_new(smx_process_t, size); - smpi_mpi_global->mpi_comm_world->simdata->processes[0] = process; + smpi_mpi_global->mpi_comm_world = xbt_new(s_smpi_mpi_communicator_t, 1); + smpi_mpi_global->mpi_comm_world->size = size; + smpi_mpi_global->mpi_comm_world->hosts = hosts; + smpi_mpi_global->mpi_comm_world->processes = xbt_new(smx_process_t, size); + smpi_mpi_global->mpi_comm_world->processes[0] = process; + smpi_mpi_global->mpi_comm_world->barrier_count = 0; + smpi_mpi_global->mpi_comm_world->barrier_mutex = SIMIX_mutex_init(); + smpi_mpi_global->mpi_comm_world->barrier_cond = SIMIX_cond_init(); // mpi datatypes - smpi_mpi_global->mpi_byte = xbt_new(smpi_mpi_datatype_t, 1); + smpi_mpi_global->mpi_byte = xbt_new(s_smpi_mpi_datatype_t, 1); smpi_mpi_global->mpi_byte->size = (size_t)1; - smpi_mpi_global->mpi_int = xbt_new(smpi_mpi_datatype_t, 1); + smpi_mpi_global->mpi_int = xbt_new(s_smpi_mpi_datatype_t, 1); smpi_mpi_global->mpi_int->size = sizeof(int); - smpi_mpi_global->mpi_double = xbt_new(smpi_mpi_datatype_t, 1); + smpi_mpi_global->mpi_double = xbt_new(s_smpi_mpi_datatype_t, 1); smpi_mpi_global->mpi_double->size = sizeof(double); // mpi operations - smpi_mpi_global->mpi_land = xbt_new(smpi_mpi_op_t, 1); + smpi_mpi_global->mpi_land = xbt_new(s_smpi_mpi_op_t, 1); smpi_mpi_global->mpi_land->func = smpi_mpi_land_func; - smpi_mpi_global->mpi_sum = xbt_new(smpi_mpi_op_t, 1); + smpi_mpi_global->mpi_sum = xbt_new(s_smpi_mpi_op_t, 1); smpi_mpi_global->mpi_sum->func = smpi_mpi_sum_func; // signal all nodes to perform initialization @@ -94,7 +93,7 @@ void smpi_mpi_init() } SIMIX_mutex_unlock(smpi_global->start_stop_mutex); - smpi_mpi_global->mpi_comm_world->simdata->processes[smpi_mpi_comm_rank_self(smpi_mpi_global->mpi_comm_world)] = process; + smpi_mpi_global->mpi_comm_world->processes[smpi_mpi_comm_rank_self(smpi_mpi_global->mpi_comm_world)] = process; } // wait for all nodes to signal initializatin complete @@ -134,10 +133,9 @@ void smpi_mpi_finalize() } } - SIMIX_mutex_destroy(smpi_mpi_global->mpi_comm_world->simdata->barrier_mutex); - SIMIX_cond_destroy(smpi_mpi_global->mpi_comm_world->simdata->barrier_cond); - xbt_free(smpi_mpi_global->mpi_comm_world->simdata->processes); - xbt_free(smpi_mpi_global->mpi_comm_world->simdata); + SIMIX_mutex_destroy(smpi_mpi_global->mpi_comm_world->barrier_mutex); + SIMIX_cond_destroy(smpi_mpi_global->mpi_comm_world->barrier_cond); + xbt_free(smpi_mpi_global->mpi_comm_world->processes); xbt_free(smpi_mpi_global->mpi_comm_world); xbt_free(smpi_mpi_global->mpi_byte); @@ -153,23 +151,23 @@ void smpi_mpi_finalize() } -void smpi_barrier(smpi_mpi_communicator_t *comm) +void smpi_barrier(smpi_mpi_communicator_t comm) { - SIMIX_mutex_lock(comm->simdata->barrier_mutex); - if(++comm->simdata->barrier_count < comm->size) { - SIMIX_cond_wait(comm->simdata->barrier_cond, comm->simdata->barrier_mutex); + SIMIX_mutex_lock(comm->barrier_mutex); + if(++comm->barrier_count < comm->size) { + SIMIX_cond_wait(comm->barrier_cond, comm->barrier_mutex); } else { - comm->simdata->barrier_count = 0; - SIMIX_cond_broadcast(comm->simdata->barrier_cond); + comm->barrier_count = 0; + SIMIX_cond_broadcast(comm->barrier_cond); } - SIMIX_mutex_unlock(comm->simdata->barrier_mutex); + SIMIX_mutex_unlock(comm->barrier_mutex); return; } -int smpi_create_request(void *buf, int count, smpi_mpi_datatype_t *datatype, - int src, int dst, int tag, smpi_mpi_communicator_t *comm, smpi_mpi_request_t **request) +int smpi_create_request(void *buf, int count, smpi_mpi_datatype_t datatype, + int src, int dst, int tag, smpi_mpi_communicator_t comm, smpi_mpi_request_t *request) { int retval = MPI_SUCCESS; @@ -190,7 +188,7 @@ int smpi_create_request(void *buf, int count, smpi_mpi_datatype_t *datatype, } else if (0 > tag) { retval = MPI_ERR_TAG; } else { - *request = xbt_mallocator_get(smpi_global->request_mallocator); + *request = xbt_mallocator_get(smpi_global->request_mallocator); (*request)->comm = comm; (*request)->src = src; (*request)->dst = dst; @@ -202,7 +200,7 @@ int smpi_create_request(void *buf, int count, smpi_mpi_datatype_t *datatype, return retval; } -int smpi_isend(smpi_mpi_request_t *request) +int smpi_isend(smpi_mpi_request_t request) { int retval = MPI_SUCCESS; int rank = smpi_mpi_comm_rank_self(smpi_mpi_global->mpi_comm_world); @@ -220,7 +218,7 @@ int smpi_isend(smpi_mpi_request_t *request) return retval; } -int smpi_irecv(smpi_mpi_request_t *request) +int smpi_irecv(smpi_mpi_request_t request) { int retval = MPI_SUCCESS; int rank = smpi_mpi_comm_rank_self(smpi_mpi_global->mpi_comm_world); @@ -238,16 +236,18 @@ int smpi_irecv(smpi_mpi_request_t *request) return retval; } -void smpi_wait(smpi_mpi_request_t *request, smpi_mpi_status_t *status) +int smpi_wait(smpi_mpi_request_t request, smpi_mpi_status_t *status) { if (NULL != request) { - SIMIX_mutex_lock(request->simdata->mutex); + SIMIX_mutex_lock(request->mutex); if (!request->completed) { - // SIMIX_cond_wait(request->simdata->cond, request->simdata->mutex); + SIMIX_cond_wait(request->cond, request->mutex); } if (NULL != status) { status->MPI_SOURCE = request->src; } - SIMIX_mutex_unlock(request->simdata->mutex); + SIMIX_mutex_unlock(request->mutex); } + + return MPI_SUCCESS; } diff --git a/src/smpi/smpi_bench.c b/src/smpi/smpi_bench.c index b3e5f9c245..aec13c01c4 100644 --- a/src/smpi/smpi_bench.c +++ b/src/smpi/smpi_bench.c @@ -4,8 +4,11 @@ void smpi_bench_begin() { int rank = smpi_mpi_comm_rank_self(smpi_mpi_global->mpi_comm_world); + SIMIX_mutex_lock(smpi_global->timers_mutexes[rank]); + xbt_os_timer_start(smpi_global->timers[rank]); + return; } @@ -14,28 +17,31 @@ void smpi_bench_end() int rank = smpi_mpi_comm_rank_self(smpi_mpi_global->mpi_comm_world); double duration; smx_host_t host; - smx_action_t compute_action; + char compute[] = "compute"; + smx_action_t action; smx_mutex_t mutex; smx_cond_t cond; xbt_os_timer_stop(smpi_global->timers[rank]); - duration = xbt_os_timer_elapsed(smpi_global->timers[rank]); + duration = xbt_os_timer_elapsed(smpi_global->timers[rank]); + SIMIX_mutex_unlock(smpi_global->timers_mutexes[rank]); - host = smpi_mpi_global->mpi_comm_world->simdata->hosts[rank]; - compute_action = SIMIX_action_execute(host, NULL, duration * SMPI_DEFAULT_SPEED); - mutex = SIMIX_mutex_init(); - cond = SIMIX_cond_init(); + host = smpi_mpi_global->mpi_comm_world->hosts[rank]; + action = SIMIX_action_execute(host, compute, duration * SMPI_DEFAULT_SPEED); + mutex = SIMIX_mutex_init(); + cond = SIMIX_cond_init(); SIMIX_mutex_lock(mutex); - SIMIX_register_action_to_condition(compute_action, cond); + SIMIX_register_action_to_condition(action, cond); SIMIX_cond_wait(cond, mutex); - SIMIX_unregister_action_to_condition(compute_action, cond); + SIMIX_unregister_action_to_condition(action, cond); SIMIX_mutex_unlock(mutex); SIMIX_mutex_destroy(mutex); SIMIX_cond_destroy(cond); + SIMIX_action_destroy(action); // FIXME: check for success/failure? diff --git a/src/smpi/smpi_global.c b/src/smpi/smpi_global.c index 4aef5a2427..686012911a 100644 --- a/src/smpi/smpi_global.c +++ b/src/smpi/smpi_global.c @@ -1,18 +1,18 @@ #include + #include "private.h" -XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi,XBT_LOG_ROOT_CAT, "All SMPI categories (see \ref SMPI_API)"); +XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi,XBT_LOG_ROOT_CAT, "All SMPI categories"); SMPI_Global_t smpi_global = NULL; void *smpi_request_new() { - smpi_mpi_request_t *request = xbt_new(smpi_mpi_request_t, 1); + smpi_mpi_request_t request = xbt_new(s_smpi_mpi_request_t, 1); request->completed = 0; - request->simdata = xbt_new(s_smpi_mpi_request_simdata_t, 1); - request->simdata->mutex = SIMIX_mutex_init(); - request->simdata->cond = SIMIX_cond_init(); + request->mutex = SIMIX_mutex_init(); + request->cond = SIMIX_cond_init(); return request; } @@ -20,12 +20,11 @@ void *smpi_request_new() void smpi_request_free(void *pointer) { - smpi_mpi_request_t *request = pointer; + smpi_mpi_request_t request = pointer; if (NULL != request) { - SIMIX_cond_destroy(request->simdata->cond); - SIMIX_mutex_destroy(request->simdata->mutex); - xbt_free(request->simdata); + SIMIX_cond_destroy(request->cond); + SIMIX_mutex_destroy(request->mutex); xbt_free(request); } @@ -40,7 +39,7 @@ void smpi_request_reset(void *pointer) void *smpi_message_new() { - return xbt_new(smpi_received_message_t, 1); + return xbt_new(s_smpi_received_message_t, 1); } void smpi_message_free(void *pointer) @@ -63,7 +62,7 @@ void smpi_global_init() int size = SIMIX_host_get_number(); - smpi_global = xbt_new(s_SMPI_Global_t, 1); + smpi_global = xbt_new(s_SMPI_Global_t, 1); // config variable smpi_global->reference_speed = SMPI_DEFAULT_SPEED; @@ -156,6 +155,8 @@ void smpi_global_destroy() xbt_free(smpi_global->timers_mutexes); xbt_free(smpi_global); + + smpi_global = NULL; } int smpi_run_simulation(int argc, char **argv) diff --git a/src/smpi/smpi_mpi.c b/src/smpi/smpi_mpi.c index 988f2ee9df..fe412d3e1b 100644 --- a/src/smpi/smpi_mpi.c +++ b/src/smpi/smpi_mpi.c @@ -80,6 +80,7 @@ int MPI_Type_size(MPI_Datatype datatype, size_t *size) return retval; } +// FIXME: check comm value and barrier success... int MPI_Barrier(MPI_Comm comm) { smpi_bench_end(); @@ -95,8 +96,8 @@ int MPI_Irecv(void *buf, int count, MPI_Datatype datatype, int src, int tag, MPI smpi_bench_end(); dst = smpi_mpi_comm_rank_self(comm); retval = smpi_create_request(buf, count, datatype, src, dst, tag, comm, request); - if (NULL != *request) { - smpi_irecv(*request); + if (NULL != *request && MPI_SUCCESS == retval) { + retval = smpi_irecv(*request); } smpi_bench_begin(); return retval; @@ -106,15 +107,18 @@ int MPI_Recv(void *buf, int count, MPI_Datatype datatype, int src, int tag, MPI_ { int retval = MPI_SUCCESS; int dst; - smpi_mpi_request_t *request; + smpi_mpi_request_t request; + int rank; smpi_bench_end(); dst = smpi_mpi_comm_rank_self(comm); retval = smpi_create_request(buf, count, datatype, src, dst, tag, comm, &request); - if (NULL != request) { - smpi_irecv(request); - smpi_wait(request, status); - // FIXME: mallocator - //xbt_free(request); + if (NULL != request && MPI_SUCCESS == retval) { + retval = smpi_irecv(request); + if (MPI_SUCCESS == retval) { + retval = smpi_wait(request, status); + } + rank = smpi_mpi_comm_rank_self(smpi_mpi_global->mpi_comm_world); + xbt_mallocator_release(smpi_global->request_mallocator, request); } smpi_bench_begin(); return retval; @@ -127,8 +131,8 @@ int MPI_Isend(void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI smpi_bench_end(); src = smpi_mpi_comm_rank_self(comm); retval = smpi_create_request(buf, count, datatype, src, dst, tag, comm, request); - if (NULL != *request) { - smpi_isend(*request); + if (NULL != *request && MPI_SUCCESS == retval) { + retval = smpi_isend(*request); } smpi_bench_begin(); return retval; @@ -138,15 +142,18 @@ int MPI_Send(void *buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_ { int retval = MPI_SUCCESS; int src; - smpi_mpi_request_t *request; + smpi_mpi_request_t request; + int rank; smpi_bench_end(); src = smpi_mpi_comm_rank_self(comm); retval = smpi_create_request(buf, count, datatype, src, dst, tag, comm, &request); - if (NULL != request) { - smpi_isend(request); - smpi_wait(request, MPI_STATUS_IGNORE); - // FIXME: mallocator - //xbt_free(request) + if (NULL != request && MPI_SUCCESS == retval) { + retval = smpi_isend(request); + if (MPI_SUCCESS == retval) { + smpi_wait(request, MPI_STATUS_IGNORE); + } + rank = smpi_mpi_comm_rank_self(smpi_mpi_global->mpi_comm_world); + xbt_mallocator_release(smpi_global->request_mallocator, request); } smpi_bench_begin(); return retval; diff --git a/src/smpi/smpi_receiver.c b/src/smpi/smpi_receiver.c index eb4723fdcc..4acf7a1fec 100644 --- a/src/smpi/smpi_receiver.c +++ b/src/smpi/smpi_receiver.c @@ -13,14 +13,13 @@ int smpi_receiver(int argc, char **argv) int running_hosts_count; - smpi_mpi_request_t *request; - smpi_received_message_t *message; + smpi_mpi_request_t request; + smpi_received_message_t message; xbt_fifo_item_t request_item; xbt_fifo_item_t message_item; - self = SIMIX_process_self(); - rank = smpi_mpi_comm_rank_self(smpi_mpi_global->mpi_comm_world); + self = SIMIX_process_self(); // make sure root is done before own initialization SIMIX_mutex_lock(smpi_global->start_stop_mutex); @@ -29,11 +28,13 @@ int smpi_receiver(int argc, char **argv) } SIMIX_mutex_unlock(smpi_global->start_stop_mutex); + rank = smpi_mpi_comm_rank_self(smpi_mpi_global->mpi_comm_world); + size = smpi_mpi_comm_size(smpi_mpi_global->mpi_comm_world); + request_queue = smpi_global->pending_recv_request_queues[rank]; request_queue_mutex = smpi_global->pending_recv_request_queues_mutexes[rank]; message_queue = smpi_global->received_message_queues[rank]; message_queue_mutex = smpi_global->received_message_queues_mutexes[rank]; - size = smpi_mpi_comm_size(smpi_mpi_global->mpi_comm_world); smpi_global->receiver_processes[rank] = self; @@ -80,17 +81,19 @@ stopsearch: if (NULL == request || NULL == message) { SIMIX_process_suspend(self); } else { - SIMIX_mutex_lock(request->simdata->mutex); + + SIMIX_mutex_lock(request->mutex); memcpy(request->buf, message->buf, request->datatype->size * request->count); request->src = message->src; request->completed = 1; - SIMIX_cond_broadcast(request->simdata->cond); + SIMIX_cond_broadcast(request->cond); - SIMIX_mutex_unlock(request->simdata->mutex); + SIMIX_mutex_unlock(request->mutex); xbt_free(message->buf); xbt_mallocator_release(smpi_global->message_mallocator, message); + } SIMIX_mutex_lock(smpi_global->running_hosts_count_mutex); diff --git a/src/smpi/smpi_sender.c b/src/smpi/smpi_sender.c index a7ea2cd15a..41ff7a1bd2 100644 --- a/src/smpi/smpi_sender.c +++ b/src/smpi/smpi_sender.c @@ -12,13 +12,14 @@ int smpi_sender(int argc, char **argv) int running_hosts_count; - smpi_mpi_request_t *request; + smpi_mpi_request_t request; smx_host_t dhost; + char communicate[] = "communicate"; smx_action_t action; - smpi_received_message_t *message; + smpi_received_message_t message; int drank; @@ -26,7 +27,6 @@ int smpi_sender(int argc, char **argv) self = SIMIX_process_self(); shost = SIMIX_host_self(); - rank = smpi_mpi_comm_rank(smpi_mpi_global->mpi_comm_world, shost); // make sure root is done before own initialization SIMIX_mutex_lock(smpi_global->start_stop_mutex); @@ -35,9 +35,11 @@ int smpi_sender(int argc, char **argv) } SIMIX_mutex_unlock(smpi_global->start_stop_mutex); + rank = smpi_mpi_comm_rank(smpi_mpi_global->mpi_comm_world, shost); + size = smpi_mpi_comm_size(smpi_mpi_global->mpi_comm_world); + request_queue = smpi_global->pending_send_request_queues[rank]; request_queue_mutex = smpi_global->pending_send_request_queues_mutexes[rank]; - size = smpi_mpi_comm_size(smpi_mpi_global->mpi_comm_world); smpi_global->sender_processes[rank] = self; @@ -63,7 +65,7 @@ int smpi_sender(int argc, char **argv) message = xbt_mallocator_get(smpi_global->message_mallocator); - SIMIX_mutex_lock(request->simdata->mutex); + SIMIX_mutex_lock(request->mutex); message->comm = request->comm; message->src = request->src; @@ -72,7 +74,7 @@ int smpi_sender(int argc, char **argv) message->buf = xbt_malloc(request->datatype->size * request->count); memcpy(message->buf, request->buf, request->datatype->size * request->count); - dhost = request->comm->simdata->hosts[request->dst]; + dhost = request->comm->hosts[request->dst]; drank = smpi_mpi_comm_rank(smpi_mpi_global->mpi_comm_world, dhost); SIMIX_mutex_lock(smpi_global->received_message_queues_mutexes[drank]); @@ -81,13 +83,15 @@ int smpi_sender(int argc, char **argv) request->completed = 1; - action = SIMIX_action_communicate(shost, dhost, "communicate", request->datatype->size * request->count * 1.0, -1.0); + action = SIMIX_action_communicate(shost, dhost, communicate, request->datatype->size * request->count, -1.0); + + SIMIX_register_action_to_condition(action, request->cond); + SIMIX_cond_wait(request->cond, request->mutex); + SIMIX_unregister_action_to_condition(action, request->cond); - SIMIX_register_action_to_condition(action, request->simdata->cond); - SIMIX_cond_wait(request->simdata->cond, request->simdata->mutex); - SIMIX_unregister_action_to_condition(action, request->simdata->cond); + SIMIX_mutex_unlock(request->mutex); - SIMIX_mutex_unlock(request->simdata->mutex); + SIMIX_action_destroy(action); // wake up receiver if necessary receiver_process = smpi_global->receiver_processes[drank]; diff --git a/src/smpi/smpi_util.c b/src/smpi/smpi_util.c index b7df6a8a60..faae149560 100644 --- a/src/smpi/smpi_util.c +++ b/src/smpi/smpi_util.c @@ -21,22 +21,23 @@ unsigned int smpi_sleep(unsigned int seconds) smx_mutex_t mutex; smx_cond_t cond; smx_host_t host; - smx_action_t sleep_action; + smx_action_t action; smpi_bench_end(); - host = SIMIX_host_self(); - sleep_action = SIMIX_action_sleep(host, seconds); - mutex = SIMIX_mutex_init(); - cond = SIMIX_cond_init(); + host = SIMIX_host_self(); + action = SIMIX_action_sleep(host, seconds); + mutex = SIMIX_mutex_init(); + cond = SIMIX_cond_init(); SIMIX_mutex_lock(mutex); - SIMIX_register_action_to_condition(sleep_action, cond); + SIMIX_register_action_to_condition(action, cond); SIMIX_cond_wait(cond, mutex); - SIMIX_unregister_action_to_condition(sleep_action, cond); + SIMIX_unregister_action_to_condition(action, cond); SIMIX_mutex_unlock(mutex); SIMIX_mutex_destroy(mutex); SIMIX_cond_destroy(cond); + SIMIX_action_destroy(action); // FIXME: check for success/failure?