X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/dff9e15c44ab6340d27215957c56fa72fad246a2..d30fcaa41c0aabda5afef028a793b9e295a4aadb:/src/smpi/smpi_base.c diff --git a/src/smpi/smpi_base.c b/src/smpi/smpi_base.c index 98992b99ae..e5af68d34c 100644 --- a/src/smpi/smpi_base.c +++ b/src/smpi/smpi_base.c @@ -1,4 +1,5 @@ #include "private.h" +#include "xbt/time.h" XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_base, smpi, "Logging specific to SMPI (base)"); @@ -12,6 +13,9 @@ XBT_LOG_EXTERNAL_CATEGORY(smpi_util); smpi_mpi_global_t smpi_mpi_global = NULL; +/** + * Operations of MPI_OP : implemented=land,sum,min,max + **/ void smpi_mpi_land_func(void *a, void *b, int *length, MPI_Datatype * datatype); @@ -27,164 +31,184 @@ void smpi_mpi_land_func(void *a, void *b, int *length, } } +/** + * sum two vectors element-wise + * + * @param a the first vectors + * @param b the second vectors + * @return the second vector is modified and contains the element-wise sums + **/ void smpi_mpi_sum_func(void *a, void *b, int *length, MPI_Datatype * datatype); void smpi_mpi_sum_func(void *a, void *b, int *length, MPI_Datatype * datatype) { int i; - if (*datatype == smpi_mpi_global->mpi_int) { + if (*datatype == smpi_mpi_global->mpi_byte) { + char *x = a, *y = b; + for (i = 0; i < *length; i++) { + y[i] = x[i] + y[i]; + } + } else if (*datatype == smpi_mpi_global->mpi_int) { int *x = a, *y = b; for (i = 0; i < *length; i++) { y[i] = x[i] + y[i]; } + } else if (*datatype == smpi_mpi_global->mpi_float) { + float *x = a, *y = b; + for (i = 0; i < *length; i++) { + y[i] = x[i] + y[i]; + } + } else if (*datatype == smpi_mpi_global->mpi_double) { + double *x = a, *y = b; + for (i = 0; i < *length; i++) { + y[i] = x[i] + y[i]; + } } } -int smpi_mpi_comm_rank(smpi_mpi_communicator_t comm) -{ - return comm->index_to_rank_map[smpi_host_index()]; -} +/** + * compute the min of two vectors element-wise + **/ +void smpi_mpi_min_func(void *a, void *b, int *length, + MPI_Datatype * datatype); -void smpi_mpi_init() +void smpi_mpi_min_func(void *a, void *b, int *length, MPI_Datatype * datatype) { - smx_host_t host; - smx_host_t *hosts; - int host_count; int i; - smpi_host_data_t hdata; - - SIMIX_mutex_lock(smpi_global->running_hosts_count_mutex); - smpi_global->running_hosts_count++; - SIMIX_mutex_unlock(smpi_global->running_hosts_count_mutex); - - // initialize some local variables - host = SIMIX_host_self(); - hosts = SIMIX_host_get_table(); - host_count = SIMIX_host_get_number(); - - hdata = xbt_new(s_smpi_host_data_t, 1); - - for (i = 0; i < host_count && host != hosts[i]; i++); - - hdata->index = i; - hdata->mutex = SIMIX_mutex_init(); - hdata->cond = SIMIX_cond_init(); - - SIMIX_host_set_data(host, hdata); - - // node 0 sets the globals - if (0 == i) { - - smpi_global->hosts = hosts; - smpi_global->host_count = host_count; - - smpi_mpi_global = xbt_new(s_smpi_mpi_global_t, 1); - - // global communicator - smpi_mpi_global->mpi_comm_world = xbt_new(s_smpi_mpi_communicator_t, 1); - smpi_mpi_global->mpi_comm_world->size = host_count; - 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(); - smpi_mpi_global->mpi_comm_world->rank_to_index_map = - xbt_new(int, host_count); - smpi_mpi_global->mpi_comm_world->index_to_rank_map = - xbt_new(int, host_count); - for (i = 0; i < host_count; i++) { - smpi_mpi_global->mpi_comm_world->rank_to_index_map[i] = i; - smpi_mpi_global->mpi_comm_world->index_to_rank_map[i] = i; + if (*datatype == smpi_mpi_global->mpi_byte) { + char *x = a, *y = b; + for (i = 0; i < *length; i++) { + y[i] = x[i] < y[i] ? x[i] : y[i]; } - - // mpi datatypes - 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(s_smpi_mpi_datatype_t, 1); - smpi_mpi_global->mpi_int->size = sizeof(int); - 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(s_smpi_mpi_op_t, 1); - smpi_mpi_global->mpi_land->func = smpi_mpi_land_func; - 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 - SIMIX_mutex_lock(smpi_global->start_stop_mutex); - smpi_global->root_ready = 1; - SIMIX_cond_broadcast(smpi_global->start_stop_cond); - SIMIX_mutex_unlock(smpi_global->start_stop_mutex); - } else { - - // make sure root is done before own initialization - SIMIX_mutex_lock(smpi_global->start_stop_mutex); - while (!smpi_global->root_ready) { - SIMIX_cond_wait(smpi_global->start_stop_cond, - smpi_global->start_stop_mutex); + if (*datatype == smpi_mpi_global->mpi_int) { + int *x = a, *y = b; + for (i = 0; i < *length; i++) { + y[i] = x[i] < y[i] ? x[i] : y[i]; + } + } else { + if (*datatype == smpi_mpi_global->mpi_float) { + float *x = a, *y = b; + for (i = 0; i < *length; i++) { + y[i] = x[i] < y[i] ? x[i] : y[i]; + } + } else { + if (*datatype == smpi_mpi_global->mpi_double) { + double *x = a, *y = b; + for (i = 0; i < *length; i++) { + y[i] = x[i] < y[i] ? x[i] : y[i]; + } + + } + } } - SIMIX_mutex_unlock(smpi_global->start_stop_mutex); - } - - // wait for all nodes to signal initializatin complete - SIMIX_mutex_lock(smpi_global->start_stop_mutex); - smpi_global->ready_process_count++; - if (smpi_global->ready_process_count >= 3 * host_count) { - SIMIX_cond_broadcast(smpi_global->start_stop_cond); - } - while (smpi_global->ready_process_count < 3 * host_count) { - SIMIX_cond_wait(smpi_global->start_stop_cond, - smpi_global->start_stop_mutex); - } - SIMIX_mutex_unlock(smpi_global->start_stop_mutex); - - return; } -void smpi_mpi_finalize() +/** + * compute the max of two vectors element-wise + **/ +void smpi_mpi_max_func(void *a, void *b, int *length, + MPI_Datatype * datatype); + +void smpi_mpi_max_func(void *a, void *b, int *length, MPI_Datatype * datatype) { int i; + if (*datatype == smpi_mpi_global->mpi_byte) { + char *x = a, *y = b; + for (i = 0; i < *length; i++) { + y[i] = x[i] > y[i] ? x[i] : y[i]; + } + } else if (*datatype == smpi_mpi_global->mpi_int) { + int *x = a, *y = b; + for (i = 0; i < *length; i++) { + y[i] = x[i] > y[i] ? x[i] : y[i]; + } + } else if (*datatype == smpi_mpi_global->mpi_float) { + float *x = a, *y = b; + for (i = 0; i < *length; i++) { + y[i] = x[i] > y[i] ? x[i] : y[i]; + } + } else if (*datatype == smpi_mpi_global->mpi_double) { + double *x = a, *y = b; + for (i = 0; i < *length; i++) { + y[i] = x[i] > y[i] ? x[i] : y[i]; + } - SIMIX_mutex_lock(smpi_global->running_hosts_count_mutex); - i = --smpi_global->running_hosts_count; - SIMIX_mutex_unlock(smpi_global->running_hosts_count_mutex); + } +} - SIMIX_mutex_lock(smpi_global->start_stop_mutex); - smpi_global->ready_process_count--; - SIMIX_mutex_unlock(smpi_global->start_stop_mutex); - SIMIX_mutex_destroy(smpi_host_mutex()); - SIMIX_cond_destroy(smpi_host_cond()); - if (0 >= i) { - // wake up senders/receivers - for (i = 0; i < smpi_global->host_count; i++) { - if (SIMIX_process_is_suspended(smpi_global->sender_processes[i])) { - SIMIX_process_resume(smpi_global->sender_processes[i]); - } - if (SIMIX_process_is_suspended(smpi_global->receiver_processes[i])) { - SIMIX_process_resume(smpi_global->receiver_processes[i]); - } - } +/** + * tell the MPI rank of the calling process (from its SIMIX process id) + **/ +int smpi_mpi_comm_rank(smpi_mpi_communicator_t comm) +{ + return comm->index_to_rank_map[smpi_process_index()]; +} - 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); +void smpi_process_init(int *argc, char ***argv) +{ + smpi_process_data_t pdata; - xbt_free(smpi_mpi_global->mpi_byte); - xbt_free(smpi_mpi_global->mpi_int); - xbt_free(smpi_mpi_global->mpi_double); + // initialize some local variables - xbt_free(smpi_mpi_global->mpi_land); - xbt_free(smpi_mpi_global->mpi_sum); + pdata = xbt_new(s_smpi_process_data_t, 1); + SIMIX_process_set_data(SIMIX_process_self(), pdata); - xbt_free(smpi_mpi_global); + /* get rank from command line, and remove it from argv */ + pdata->index = atoi((*argv)[1]); + DEBUG1("I'm rank %d", pdata->index); + if (*argc > 2) { + memmove((*argv)[1], (*argv)[2], sizeof(char *) * (*argc - 2)); + (*argv)[(*argc) - 1] = NULL; + } + (*argc)--; + + pdata->mutex = SIMIX_mutex_init(); + pdata->cond = SIMIX_cond_init(); + pdata->finalize = 0; + + pdata->pending_recv_request_queue = xbt_fifo_new(); + pdata->pending_send_request_queue = xbt_fifo_new(); + pdata->received_message_queue = xbt_fifo_new(); + + pdata->main = SIMIX_process_self(); + pdata->sender = SIMIX_process_create("smpi_sender", + smpi_sender, pdata, + SIMIX_host_get_name(SIMIX_host_self()), + 0, NULL, + /*props */ NULL); + pdata->receiver = SIMIX_process_create("smpi_receiver", + smpi_receiver, pdata, + SIMIX_host_get_name(SIMIX_host_self + ()), 0, NULL, + /*props */ NULL); + + smpi_global->main_processes[pdata->index] = SIMIX_process_self(); + return; +} + +void smpi_process_finalize() +{ + smpi_process_data_t pdata = SIMIX_process_get_data(SIMIX_process_self()); + pdata->finalize = 2; /* Tell sender and receiver to quit */ + SIMIX_process_resume(pdata->sender); + SIMIX_process_resume(pdata->receiver); + while (pdata->finalize > 0) { /* wait until it's done */ + SIMIX_cond_wait(pdata->cond, pdata->mutex); } + SIMIX_mutex_destroy(pdata->mutex); + SIMIX_cond_destroy(pdata->cond); + xbt_fifo_free(pdata->pending_recv_request_queue); + xbt_fifo_free(pdata->pending_send_request_queue); + xbt_fifo_free(pdata->received_message_queue); + xbt_free(pdata); } int smpi_mpi_barrier(smpi_mpi_communicator_t comm) @@ -207,20 +231,14 @@ int smpi_mpi_barrier(smpi_mpi_communicator_t comm) int smpi_mpi_isend(smpi_mpi_request_t request) { + smpi_process_data_t pdata = SIMIX_process_get_data(SIMIX_process_self()); int retval = MPI_SUCCESS; - int index = smpi_host_index(); if (NULL == request) { retval = MPI_ERR_INTERN; } else { - SIMIX_mutex_lock(smpi_global->pending_send_request_queues_mutexes[index]); - xbt_fifo_push(smpi_global->pending_send_request_queues[index], request); - SIMIX_mutex_unlock(smpi_global->pending_send_request_queues_mutexes - [index]); - - if (SIMIX_process_is_suspended(smpi_global->sender_processes[index])) { - SIMIX_process_resume(smpi_global->sender_processes[index]); - } + xbt_fifo_push(pdata->pending_send_request_queue, request); + SIMIX_process_resume(pdata->sender); } return retval; @@ -229,24 +247,30 @@ int smpi_mpi_isend(smpi_mpi_request_t request) int smpi_mpi_irecv(smpi_mpi_request_t request) { int retval = MPI_SUCCESS; - int index = smpi_host_index(); + smpi_process_data_t pdata = SIMIX_process_get_data(SIMIX_process_self()); if (NULL == request) { retval = MPI_ERR_INTERN; } else { - SIMIX_mutex_lock(smpi_global->pending_recv_request_queues_mutexes[index]); - xbt_fifo_push(smpi_global->pending_recv_request_queues[index], request); - SIMIX_mutex_unlock(smpi_global->pending_recv_request_queues_mutexes - [index]); + xbt_fifo_push(pdata->pending_recv_request_queue, request); - if (SIMIX_process_is_suspended(smpi_global->receiver_processes[index])) { - SIMIX_process_resume(smpi_global->receiver_processes[index]); + if (SIMIX_process_is_suspended(pdata->receiver)) { + SIMIX_process_resume(pdata->receiver); } } return retval; } +void print_req( smpi_mpi_request_t r ); +void print_req( smpi_mpi_request_t r ) { + printf("***req %p-> src=%d dst=%d tag=%d completed=0x%x consumed=0x%x\n",r,r->src,r->dst,r->tag,r->completed,r->consumed); +} + + +/** + * wait and friends ... + **/ int smpi_mpi_wait(smpi_mpi_request_t request, smpi_mpi_status_t * status) { int retval = MPI_SUCCESS; @@ -255,6 +279,10 @@ int smpi_mpi_wait(smpi_mpi_request_t request, smpi_mpi_status_t * status) retval = MPI_ERR_INTERN; } else { SIMIX_mutex_lock(request->mutex); + +#ifdef DEBUG_STEPH + print_req( request ); //@@ +#endif while (!request->completed) { SIMIX_cond_wait(request->cond, request->mutex); } @@ -268,3 +296,84 @@ int smpi_mpi_wait(smpi_mpi_request_t request, smpi_mpi_status_t * status) return retval; } + +/** + * waitall + **/ +int smpi_mpi_waitall(int count, smpi_mpi_request_t requests[], + smpi_mpi_status_t status[]) +{ + int cpt; + int index; + int retval; + smpi_mpi_status_t stat; + + for (cpt = 0; cpt < count; cpt++) { + retval = smpi_mpi_waitany(count, requests, &index, &stat); + if (retval != MPI_SUCCESS) + return retval; + if (MPI_STATUS_IGNORE != status) + memcpy(&(status[index]), &stat, sizeof(stat)); + } + return MPI_SUCCESS; +} + +/** + * waitany + **/ +int smpi_mpi_waitany(int count, smpi_mpi_request_t * requests, int *index, + smpi_mpi_status_t * status) +{ + int cpt; + + *index = MPI_UNDEFINED; + if (NULL == requests) { + return MPI_ERR_INTERN; + } + /* First check if one of them is already done */ + for (cpt = 0; cpt < count; cpt++) { + printf("...exam req[%d] of msg from [%d]\n",cpt,requests[cpt]->src); + if (requests[cpt]->completed && !requests[cpt]->consumed) { /* got ya */ + printf("...found match req[%d] of msg from [%d]\n",cpt,requests[cpt]->src); + *index = cpt; + goto found_request; + } + } + /* If none found, block */ + /* FIXME: should use a SIMIX_cond_waitany, when implemented. For now, block on the first one */ + while (1) { + for (cpt = 0; cpt < count; cpt++) { + +#ifdef DEBUG_STEPH + print_req( requests[cpt] ); +#endif + if (!requests[cpt]->completed) { /* this one is not done, wait on it */ + printf("... blocked waiting a msg %d->%d, tag=%d\n",requests[cpt]->src,requests[cpt]->dst,requests[cpt]->tag); + while (!requests[cpt]->completed) + SIMIX_cond_wait(requests[cpt]->cond, requests[cpt]->mutex); + + *index = cpt; + goto found_request; + } + } + if (cpt == count) /* they are all done. Damn user */ + return MPI_ERR_REQUEST; + } + +found_request: +#ifdef DEBUG_STEPH + print_req( requests[cpt] ); +#endif + requests[*index]->consumed = 1; +#ifdef DEBUG_STEPH + print_req( requests[cpt] ); +#endif + printf("...accessing *req[%d]->consumed\n",cpt); + if (NULL != status) { + status->MPI_SOURCE = requests[*index]->src; + status->MPI_TAG = requests[*index]->tag; + status->MPI_ERROR = MPI_SUCCESS; + } + return MPI_SUCCESS; + +}