X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/c8304965b869fd1264284ea426f02dcd111fb62b..c7abe4a06a869040289677126a8c3e1b9a92216f:/src/smpi/smpi_sender.c diff --git a/src/smpi/smpi_sender.c b/src/smpi/smpi_sender.c index a7ea2cd15a..eb5e727d58 100644 --- a/src/smpi/smpi_sender.c +++ b/src/smpi/smpi_sender.c @@ -1,107 +1,110 @@ #include "private.h" -int smpi_sender(int argc, char **argv) -{ - smx_process_t self; - smx_host_t shost; - int rank; - - xbt_fifo_t request_queue; - smx_mutex_t request_queue_mutex; - int size; +XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_sender, smpi, + "Logging specific to SMPI (sender)"); - int running_hosts_count; - - smpi_mpi_request_t *request; +int smpi_sender(int argc, char *argv[]) +{ + smpi_process_data_t mydata = SIMIX_process_get_data(SIMIX_process_self()); + smx_process_t self; + smx_host_t shost; - smx_host_t dhost; + int index; - smx_action_t action; + xbt_fifo_t request_queue; - smpi_received_message_t *message; + smpi_mpi_request_t request; - int drank; + smx_host_t dhost; - smx_process_t receiver_process; + smx_action_t action; - self = SIMIX_process_self(); - shost = SIMIX_host_self(); - rank = smpi_mpi_comm_rank(smpi_mpi_global->mpi_comm_world, shost); + e_surf_action_state_t state; - // make sure root is done before own initialization - SIMIX_mutex_lock(smpi_global->start_stop_mutex); - if (!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); + smpi_received_message_t message; - 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); + int dindex; - smpi_global->sender_processes[rank] = self; + self = SIMIX_process_self(); + shost = SIMIX_host_self(); - // 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 * size) { - SIMIX_cond_wait(smpi_global->start_stop_cond, smpi_global->start_stop_mutex); - } else { - SIMIX_cond_broadcast(smpi_global->start_stop_cond); - } - SIMIX_mutex_unlock(smpi_global->start_stop_mutex); + index = mydata->index; - do { + DEBUG0("Up and running"); - SIMIX_mutex_lock(request_queue_mutex); - request = xbt_fifo_shift(request_queue); - SIMIX_mutex_unlock(request_queue_mutex); + request_queue = mydata->pending_send_request_queue; - if (NULL == request) { - SIMIX_process_suspend(self); - } else { + while (1) { + request = xbt_fifo_shift(request_queue); - message = xbt_mallocator_get(smpi_global->message_mallocator); + if (NULL != request) { + 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; - message->dst = request->dst; - message->tag = request->tag; - message->buf = xbt_malloc(request->datatype->size * request->count); - memcpy(message->buf, request->buf, request->datatype->size * request->count); + message->comm = request->comm; + message->src = request->comm->index_to_rank_map[index]; + message->tag = request->tag; + message->data = request->data; + 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]; - drank = smpi_mpi_comm_rank(smpi_mpi_global->mpi_comm_world, dhost); + dindex = request->comm->rank_to_index_map[request->dst]; + smpi_process_data_t remote_process = + SIMIX_process_get_data(smpi_global->main_processes[dindex]); + dhost = SIMIX_process_get_host(smpi_global->main_processes[dindex]); - SIMIX_mutex_lock(smpi_global->received_message_queues_mutexes[drank]); - xbt_fifo_push(smpi_global->received_message_queues[drank], message); - SIMIX_mutex_unlock(smpi_global->received_message_queues_mutexes[drank]); + DEBUG4("handle send request %p to %s (req_dst=%d,req_tag=%d)", + request,SIMIX_host_get_name(dhost),request->dst,message->tag); + message->forward = (request->forward - 1) / 2; + request->forward = request->forward / 2; - request->completed = 1; + if (0 < request->forward) { + request->dst = + (request->dst + message->forward + 1) % request->comm->size; + xbt_fifo_push(request_queue, request); + } else { + DEBUG4("DONE Handling send request %p to %s (req_dst=%d,req_tag=%d)", + request, SIMIX_host_get_name(dhost),request->dst,message->tag); + 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, "communication", + request->datatype->size * request->count, + -1.0); - 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_register_action_to_condition(action, request->cond); - SIMIX_mutex_unlock(request->simdata->mutex); + for (state = SIMIX_action_get_state(action); + state == SURF_ACTION_READY || + state == SURF_ACTION_RUNNING; + state = SIMIX_action_get_state(action) + ) { + SIMIX_cond_wait(request->cond, request->mutex); + } - // wake up receiver if necessary - receiver_process = smpi_global->receiver_processes[drank]; - if (SIMIX_process_is_suspended(receiver_process)) { - SIMIX_process_resume(receiver_process); - } + xbt_fifo_push(remote_process->received_message_queue, message); - } + SIMIX_unregister_action_to_condition(action, request->cond); + SIMIX_action_destroy(action); - SIMIX_mutex_lock(smpi_global->running_hosts_count_mutex); - running_hosts_count = smpi_global->running_hosts_count; - SIMIX_mutex_unlock(smpi_global->running_hosts_count_mutex); + SIMIX_mutex_unlock(request->mutex); - } while (0 < running_hosts_count); + // wake up receiver if necessary + SIMIX_process_resume(remote_process->receiver); - return 0; + } else if (mydata->finalize > 0) { /* main wants me to die and nothing to do */ + DEBUG0("===Main wants me to die and I'm done. Bye, guys.==="); + mydata->finalize--; + SIMIX_cond_signal(mydata->cond); + return 0; + } else { + DEBUG0("Nothing to do. Let's get a nap"); + SIMIX_process_suspend(self); + DEBUG0("===Uh? Someone called me?==="); + } + } + return 0; }