X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/d0443518bf57fa4111a6437a8f3322f12cd88f29..c7abe4a06a869040289677126a8c3e1b9a92216f:/src/smpi/smpi_base.c diff --git a/src/smpi/smpi_base.c b/src/smpi/smpi_base.c index d923982079..f3bfa38085 100644 --- a/src/smpi/smpi_base.c +++ b/src/smpi/smpi_base.c @@ -1,253 +1,422 @@ #include "private.h" - -XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_base, smpi, "Logging specific to SMPI (base)"); +#include "xbt/time.h" + +XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_base, smpi, + "Logging specific to SMPI (base)"); +XBT_LOG_EXTERNAL_CATEGORY(smpi_base); +XBT_LOG_EXTERNAL_CATEGORY(smpi_bench); +XBT_LOG_EXTERNAL_CATEGORY(smpi_kernel); +XBT_LOG_EXTERNAL_CATEGORY(smpi_mpi); +XBT_LOG_EXTERNAL_CATEGORY(smpi_mpi_dt); +XBT_LOG_EXTERNAL_CATEGORY(smpi_coll); +XBT_LOG_EXTERNAL_CATEGORY(smpi_receiver); +XBT_LOG_EXTERNAL_CATEGORY(smpi_sender); +XBT_LOG_EXTERNAL_CATEGORY(smpi_util); smpi_mpi_global_t smpi_mpi_global = NULL; -void smpi_mpi_land_func(void *a, void *b, int *length, MPI_Datatype *datatype); -void smpi_mpi_land_func(void *a, void *b, int *length, MPI_Datatype *datatype) +/** + * Operations of MPI_OP : implemented=land,sum,min,max + **/ +void smpi_mpi_land_func(void *a, void *b, int *length, + MPI_Datatype * datatype); + +void smpi_mpi_land_func(void *a, void *b, int *length, + MPI_Datatype * datatype) { - int i; - if (*datatype == smpi_mpi_global->mpi_int) { - int *x = a, *y = b; - for (i = 0; i < *length; i++) { - y[i] = x[i] && y[i]; - } - } + int i; + if (*datatype == smpi_mpi_global->mpi_int) { + int *x = a, *y = b; + for (i = 0; i < *length; i++) { + y[i] = x[i] && y[i]; + } + } } -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) +/** + * 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) { - int *x = a, *y = b; - for (i = 0; i < *length; i++) { - y[i] = x[i] + y[i]; - } - } + int i; + if ((*datatype == smpi_mpi_global->mpi_byte) || (*datatype == smpi_mpi_global->mpi_char)) { + 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) +/** + *i multiply 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 products + **/ +void smpi_mpi_prod_func(void *a, void *b, int *length, + MPI_Datatype * datatype); + +void smpi_mpi_prod_func(void *a, void *b, int *length, MPI_Datatype * datatype) { - return comm->index_to_rank_map[smpi_host_index()]; + int i; + if ((*datatype == smpi_mpi_global->mpi_byte) || (*datatype == smpi_mpi_global->mpi_char)) { + 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]; + } + } } +/** + * 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; - } - - // 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); - } - 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; + int i; + if ((*datatype == smpi_mpi_global->mpi_byte) || (*datatype == smpi_mpi_global->mpi_char)) { + 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]; + } + + } + } + } + } } -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; + int i; + if ((*datatype == smpi_mpi_global->mpi_byte) || (*datatype == smpi_mpi_global->mpi_char)) { + 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) { +/** + * 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()]; +} - // 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]); - } - } +void smpi_process_init(int *argc, char ***argv) +{ + smpi_process_data_t pdata; + + // initialize some local variables + + pdata = xbt_new(s_smpi_process_data_t, 1); + SIMIX_process_set_data(SIMIX_process_self(), pdata); + + /* 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); +} - 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); - xbt_free(smpi_mpi_global->mpi_byte); - xbt_free(smpi_mpi_global->mpi_int); - xbt_free(smpi_mpi_global->mpi_double); +/*int smpi_mpi_barrier(smpi_mpi_communicator_t comm) +{ - xbt_free(smpi_mpi_global->mpi_land); - xbt_free(smpi_mpi_global->mpi_sum); + SIMIX_mutex_lock(comm->barrier_mutex); + ++comm->barrier_count; + if (comm->barrier_count > comm->size) { // only happens on second barrier... + comm->barrier_count = 0; + } else if (comm->barrier_count == comm->size) { + SIMIX_cond_broadcast(comm->barrier_cond); + } + while (comm->barrier_count < comm->size) { + SIMIX_cond_wait(comm->barrier_cond, comm->barrier_mutex); + } + SIMIX_mutex_unlock(comm->barrier_mutex); + + return MPI_SUCCESS; +} +*/ - xbt_free(smpi_mpi_global); +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; - } + if (NULL == request) { + retval = MPI_ERR_INTERN; + } else { + xbt_fifo_push(pdata->pending_send_request_queue, request); + SIMIX_process_resume(pdata->sender); + } + return retval; } -int smpi_mpi_barrier(smpi_mpi_communicator_t comm) +int smpi_mpi_irecv(smpi_mpi_request_t request) { + int retval = MPI_SUCCESS; + smpi_process_data_t pdata = SIMIX_process_get_data(SIMIX_process_self()); + + if (NULL == request) { + retval = MPI_ERR_INTERN; + } else { + xbt_fifo_push(pdata->pending_recv_request_queue, request); - SIMIX_mutex_lock(comm->barrier_mutex); - ++comm->barrier_count; - if (comm->barrier_count > comm->size) { // only happens on second barrier... - comm->barrier_count = 0; - } else if (comm->barrier_count == comm->size) { - SIMIX_cond_broadcast(comm->barrier_cond); - } - while (comm->barrier_count < comm->size) { - SIMIX_cond_wait(comm->barrier_cond, comm->barrier_mutex); - } - SIMIX_mutex_unlock(comm->barrier_mutex); - - return MPI_SUCCESS; + if (SIMIX_process_is_suspended(pdata->receiver)) { + SIMIX_process_resume(pdata->receiver); + } + } + + return retval; } -int smpi_mpi_isend(smpi_mpi_request_t request) +void print_req( smpi_mpi_request_t r ); +void print_req( smpi_mpi_request_t r ) { + fprintf(stderr,"***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; - 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]); - } - } - - return retval; + int retval = MPI_SUCCESS; + + if (NULL == request) { + retval = MPI_ERR_INTERN; + } else { + + DEBUG3("entered smpi_mpi_wait() for req_src=%d,req_dst=%d,req_tag=%d", + request->src,request->dst,request->tag); + SIMIX_mutex_lock(request->mutex); +//#define DEBUG_STEPH +#ifdef DEBUG_STEPH + print_req( request ); //@@ +#endif + while (!request->completed) { + SIMIX_cond_wait(request->cond, request->mutex); + } + if (NULL != status) { + status->MPI_SOURCE = request->src; + status->MPI_TAG = request->tag; + status->MPI_ERROR = MPI_SUCCESS; + } + SIMIX_mutex_unlock(request->mutex); + } + + return retval; } -int smpi_mpi_irecv(smpi_mpi_request_t request) +/** + * waitall + **/ +int smpi_mpi_waitall(int count, smpi_mpi_request_t requests[], + smpi_mpi_status_t status[]) { - int retval = MPI_SUCCESS; - int index = smpi_host_index(); - - 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]); - - if (SIMIX_process_is_suspended(smpi_global->receiver_processes[index])) { - SIMIX_process_resume(smpi_global->receiver_processes[index]); - } - } - - return retval; + 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; } -int smpi_mpi_wait(smpi_mpi_request_t request, smpi_mpi_status_t *status) +/** + * waitany + **/ +int smpi_mpi_waitany(int count, smpi_mpi_request_t * requests, int *index, + smpi_mpi_status_t * status) { - int retval = MPI_SUCCESS; - - if (NULL == request) { - retval = MPI_ERR_INTERN; - } else { - SIMIX_mutex_lock(request->mutex); - while (!request->completed) { - SIMIX_cond_wait(request->cond, request->mutex); - } - if (NULL != status) { - status->MPI_SOURCE = request->src; - status->MPI_TAG = request->tag; - status->MPI_ERROR = MPI_SUCCESS; - } - SIMIX_mutex_unlock(request->mutex); - } - - return retval; + int cpt; + + DEBUG0("entering smpi_wait_any() ..."); + *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++) { + DEBUG2(" exam req[%d] of msg from <%d>",cpt,requests[cpt]->src); + if (requests[cpt]->completed && !requests[cpt]->consumed) { /* got ya */ + DEBUG2("smpi_wait_any() found match req[%d] of msg from <%d>",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 */ + DEBUG3("smpi_waitany() blocked waiting a msg <%d> -> <%d>, tag=%d",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 + DEBUG2("smpi_waitany() request %p unblocked ... mark *req[%d]->consumed",requests[*index],cpt); + if (NULL != status) { + status->MPI_SOURCE = requests[*index]->src; + status->MPI_TAG = requests[*index]->tag; + status->MPI_ERROR = MPI_SUCCESS; + } + return MPI_SUCCESS; + }