X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/f1a8552a42735c41d109bb458835a17976baed9c..ad12e936954df10a0395da3b161e820e07e97f52:/src/smpi/smpi_base.c diff --git a/src/smpi/smpi_base.c b/src/smpi/smpi_base.c index 1972895d0d..ed9620b716 100644 --- a/src/smpi/smpi_base.c +++ b/src/smpi/smpi_base.c @@ -1,816 +1,613 @@ -#include -#include -#include +/* Copyright (c) 2007, 2008, 2009, 2010. The SimGrid Team. + * All rights reserved. */ -#include "private.h" - -SMPI_Global_t smpi_global = NULL; - -SMPI_MPI_Global_t smpi_mpi_global = NULL; +/* 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. */ -XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi,XBT_LOG_ROOT_CAT, "All SMPI categories (see \ref SMPI_API)"); - -int inline smpi_mpi_comm_size(smpi_mpi_communicator_t *comm) -{ - return comm->size; +#include "private.h" +#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); + +void smpi_process_init(int* argc, char*** argv) { + int index; + smpi_process_data_t data; + smx_process_t proc; + + proc = SIMIX_process_self(); + index = atoi((*argv)[1]); + data = smpi_process_remote_data(index); + SIMIX_process_set_data(proc, data); + if (*argc > 2) { + free((*argv)[1]); + memmove(&(*argv)[1], &(*argv)[2], sizeof(char *) * (*argc - 2)); + (*argv)[(*argc) - 1] = NULL; + } + (*argc)--; + DEBUG2("<%d> New process in the game: %p", index, proc); } -// FIXME: smarter algorithm? -int smpi_mpi_comm_rank(smpi_mpi_communicator_t *comm, smx_host_t host) -{ - int i; +void smpi_process_destroy(void) { + int index = smpi_process_index(); - for(i = comm->size - 1; i > 0 && host != comm->hosts[i]; i--); - - return i; + DEBUG1("<%d> Process left the game", index); } -int inline smpi_mpi_comm_rank_self(smpi_mpi_communicator_t *comm) -{ - return smpi_mpi_comm_rank(comm, SIMIX_host_self()); +static MPI_Request build_request(void* buf, int count, MPI_Datatype datatype, int src, int dst, int tag, MPI_Comm comm, unsigned flags) { + MPI_Request request; + + request = xbt_new(s_smpi_mpi_request_t, 1); + request->buf = buf; + request->size = smpi_datatype_size(datatype) * count; + request->src = src; + request->dst = dst; + request->tag = tag; + request->comm = comm; + request->rdv = NULL; + request->pair = NULL; + request->complete = 0; + request->match = MPI_REQUEST_NULL; + request->flags = flags; +#ifdef HAVE_TRACING + request->send = 0; + request->recv = 0; +#endif + return request; } -//int smpi_mpi_comm_world_rank_self() -//{ -// return smpi_mpi_comm_rank(smpi_mpi_global->mpi_comm_world, SIMIX_host_self()); -//} - -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; - - int running_hosts_count; - - smpi_mpi_request_t *request; - - smx_host_t dhost; - - smx_action_t communicate_action; - - smpi_received_message_t *message; - - int drank; - - smx_process_t receiver_process; - - 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); - 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); - - 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; - - // 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); - - do { - - SIMIX_mutex_lock(request_queue_mutex); - request = xbt_fifo_shift(request_queue); - SIMIX_mutex_unlock(request_queue_mutex); - - if (NULL == request) { - SIMIX_process_suspend(self); - } else { - - SIMIX_mutex_lock(request->mutex); - - // copy request to appropriate received queue - message = xbt_mallocator_get(smpi_global->message_mallocator); - 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); - - 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]); - xbt_fifo_push(smpi_global->received_message_queues[drank], message); - SIMIX_mutex_unlock(smpi_global->received_message_queues_mutexes[drank]); - - request->completed = 1; - - communicate_action = SIMIX_action_communicate(shost, dhost, - NULL, request->datatype->size * request->count * 1.0, -1.0); - - SIMIX_register_condition_to_action(communicate_action, request->cond); - SIMIX_register_action_to_condition(communicate_action, request->cond); - - SIMIX_cond_wait(request->cond, request->mutex); - - SIMIX_mutex_unlock(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); - } - - } - - 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); - - } while (0 < running_hosts_count); - - SIMIX_mutex_lock(smpi_global->start_stop_mutex); - smpi_global->ready_process_count--; - if (smpi_global->ready_process_count == 0) { - SIMIX_cond_broadcast(smpi_global->start_stop_cond); - } else if (smpi_global->ready_process_count < 0) { - // FIXME: can't happen! abort! - } - SIMIX_mutex_unlock(smpi_global->start_stop_mutex); - - return 0; -} - -int smpi_receiver(int argc, char **argv) -{ - smx_process_t self; - int rank; - - xbt_fifo_t request_queue; - smx_mutex_t request_queue_mutex; - xbt_fifo_t message_queue; - smx_mutex_t message_queue_mutex; - int size; - - int running_hosts_count; - - 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); - - // 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); - - 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; - - // 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); +/* MPI Low level calls */ +MPI_Request smpi_mpi_send_init(void* buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) { + MPI_Request request = build_request(buf, count, datatype, smpi_comm_rank(comm), dst, tag, comm, PERSISTENT | SEND); - do { - request = NULL; - message = NULL; - - // FIXME: better algorithm, maybe some kind of balanced tree? or a heap? - - // FIXME: not the best way to request multiple locks... - SIMIX_mutex_lock(request_queue_mutex); - SIMIX_mutex_lock(message_queue_mutex); - for (request_item = xbt_fifo_get_first_item(request_queue); - NULL != request_item; - request_item = xbt_fifo_get_next_item(request_item)) { - request = xbt_fifo_get_item_content(request_item); - for (message_item = xbt_fifo_get_first_item(message_queue); - NULL != message_item; - message_item = xbt_fifo_get_next_item(message_item)) { - message = xbt_fifo_get_item_content(message_item); - if (request->comm == message->comm && - (MPI_ANY_SOURCE == request->src || request->src == message->src) && - request->tag == message->tag) { - xbt_fifo_remove_item(request_queue, request_item); - xbt_fifo_remove_item(message_queue, message_item); - goto stopsearch; - } - } - } -stopsearch: - SIMIX_mutex_unlock(message_queue_mutex); - SIMIX_mutex_unlock(request_queue_mutex); - - if (NULL == request || NULL == message) { - SIMIX_process_suspend(self); - } else { - 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->cond); - - 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); - running_hosts_count = smpi_global->running_hosts_count; - SIMIX_mutex_unlock(smpi_global->running_hosts_count_mutex); - - } while (0 < running_hosts_count); - - SIMIX_mutex_lock(smpi_global->start_stop_mutex); - smpi_global->ready_process_count--; - if (smpi_global->ready_process_count == 0) { - SIMIX_cond_broadcast(smpi_global->start_stop_cond); - } else if (smpi_global->ready_process_count < 0) { - // FIXME: can't happen, abort! - } - SIMIX_mutex_unlock(smpi_global->start_stop_mutex); - - return 0; + return request; } -void *smpi_request_new() -{ - smpi_mpi_request_t *request = xbt_new(smpi_mpi_request_t, 1); - - request->completed = 0; - request->mutex = SIMIX_mutex_init(); - request->cond = SIMIX_cond_init(); +MPI_Request smpi_mpi_recv_init(void* buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm) { + MPI_Request request = build_request(buf, count, datatype, src, smpi_comm_rank(comm), tag, comm, PERSISTENT | RECV); - return request; + return request; } -void smpi_request_free(void *pointer) -{ - - smpi_mpi_request_t *request = pointer; +void smpi_mpi_start(MPI_Request request) { + xbt_assert0(request->complete == 0, "Cannot start a non-finished communication"); + if((request->flags & RECV) == RECV) { + smpi_process_post_recv(request); + print_request("New recv", request); + request->pair = SIMIX_network_irecv(request->rdv, request->buf, &request->size); + } else { + smpi_process_post_send(request->comm, request); // FIXME + print_request("New send", request); + request->pair = SIMIX_network_isend(request->rdv, request->size, -1.0, request->buf, request->size, NULL); + } +} - if (NULL != request) { - SIMIX_cond_destroy(request->cond); - SIMIX_mutex_destroy(request->mutex); - xbt_free(request); - } +void smpi_mpi_startall(int count, MPI_Request* requests) { + int i; - return; + for(i = 0; i < count; i++) { + smpi_mpi_start(requests[i]); + } } -void smpi_request_reset(void *pointer) -{ - return; +void smpi_mpi_request_free(MPI_Request* request) { + xbt_free(*request); + *request = MPI_REQUEST_NULL; } +MPI_Request smpi_isend_init(void* buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) { + MPI_Request request = build_request(buf, count, datatype, smpi_comm_rank(comm), dst, tag, comm, NON_PERSISTENT | SEND); -void *smpi_message_new() -{ - return xbt_new(smpi_received_message_t, 1); + return request; } -void smpi_message_free(void *pointer) -{ - if (NULL != pointer) { - xbt_free(pointer); - } +MPI_Request smpi_mpi_isend(void* buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) { + MPI_Request request = smpi_isend_init(buf, count, datatype, dst, tag, comm); - return; -} - -void smpi_message_reset(void *pointer) -{ - return; + smpi_mpi_start(request); + return request; } -void smpi_global_init() -{ - int i; +MPI_Request smpi_irecv_init(void* buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm) { + MPI_Request request = build_request(buf, count, datatype, src, smpi_comm_rank(comm), tag, comm, NON_PERSISTENT | RECV); - int size = SIMIX_host_get_number(); - - smpi_global = xbt_new(s_SMPI_Global_t, 1); - - // config variable - smpi_global->reference_speed = SMPI_DEFAULT_SPEED; + return request; +} - smpi_global->root_ready = 0; - smpi_global->ready_process_count = 0; +MPI_Request smpi_mpi_irecv(void* buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm) { + MPI_Request request = smpi_irecv_init(buf, count, datatype, src, tag, comm); - // start/stop - smpi_global->start_stop_mutex = SIMIX_mutex_init(); - smpi_global->start_stop_cond = SIMIX_cond_init(); + smpi_mpi_start(request); + return request; +} - // processes - smpi_global->sender_processes = xbt_new(smx_process_t, size); - smpi_global->receiver_processes = xbt_new(smx_process_t, size); +void smpi_mpi_recv(void* buf, int count, MPI_Datatype datatype, int src, int tag, MPI_Comm comm, MPI_Status* status) { + MPI_Request request; - // running hosts - smpi_global->running_hosts_count_mutex = SIMIX_mutex_init(); - smpi_global->running_hosts_count = 0; + request = smpi_mpi_irecv(buf, count, datatype, src, tag, comm); + smpi_mpi_wait(&request, status); +} - // mallocators - smpi_global->request_mallocator = xbt_mallocator_new(SMPI_REQUEST_MALLOCATOR_SIZE, - smpi_request_new, smpi_request_free, smpi_request_reset); - smpi_global->message_mallocator = xbt_mallocator_new(SMPI_MESSAGE_MALLOCATOR_SIZE, - smpi_message_new, smpi_message_free, smpi_message_reset); +void smpi_mpi_send(void* buf, int count, MPI_Datatype datatype, int dst, int tag, MPI_Comm comm) { + MPI_Request request; - // - smpi_global->pending_send_request_queues = xbt_new(xbt_fifo_t, size); - smpi_global->pending_send_request_queues_mutexes = xbt_new(smx_mutex_t, size); - smpi_global->pending_recv_request_queues = xbt_new(xbt_fifo_t, size); - smpi_global->pending_recv_request_queues_mutexes = xbt_new(smx_mutex_t, size); - smpi_global->received_message_queues = xbt_new(xbt_fifo_t, size); - smpi_global->received_message_queues_mutexes = xbt_new(smx_mutex_t, size); - smpi_global->timers = xbt_new(xbt_os_timer_t, size); - smpi_global->timers_mutexes = xbt_new(smx_mutex_t, size); + request = smpi_mpi_isend(buf, count, datatype, dst, tag, comm); + smpi_mpi_wait(&request, MPI_STATUS_IGNORE); +} - for(i = 0; i < size; i++) { - smpi_global->pending_send_request_queues[i] = xbt_fifo_new(); - smpi_global->pending_send_request_queues_mutexes[i] = SIMIX_mutex_init(); - smpi_global->pending_recv_request_queues[i] = xbt_fifo_new(); - smpi_global->pending_recv_request_queues_mutexes[i] = SIMIX_mutex_init(); - smpi_global->received_message_queues[i] = xbt_fifo_new(); - smpi_global->received_message_queues_mutexes[i] = SIMIX_mutex_init(); - smpi_global->timers[i] = xbt_os_timer_new(); - smpi_global->timers_mutexes[i] = SIMIX_mutex_init(); - } +void smpi_mpi_sendrecv(void* sendbuf, int sendcount, MPI_Datatype sendtype, int dst, int sendtag, void* recvbuf, int recvcount, MPI_Datatype recvtype, int src, int recvtag, MPI_Comm comm, MPI_Status* status) { + MPI_Request requests[2]; + MPI_Status stats[2]; + + requests[0] = smpi_isend_init(sendbuf, sendcount, sendtype, dst, sendtag, comm); + requests[1] = smpi_irecv_init(recvbuf, recvcount, recvtype, src, recvtag, comm); + smpi_mpi_startall(2, requests); + smpi_mpi_waitall(2, requests, stats); + if(status != MPI_STATUS_IGNORE) { + // Copy receive status + memcpy(status, &stats[1], sizeof(MPI_Status)); + } +} +static void finish_wait(MPI_Request* request, MPI_Status* status) { + if(status != MPI_STATUS_IGNORE) { + status->MPI_SOURCE = (*request)->src; + status->MPI_TAG = (*request)->tag; + status->MPI_ERROR = MPI_SUCCESS; + } + print_request("finishing wait", *request); + if((*request)->complete == 1) { + SIMIX_rdv_destroy((*request)->rdv); + } else { + (*request)->match->complete = 1; + (*request)->match->match = MPI_REQUEST_NULL; + } + if(((*request)->flags & NON_PERSISTENT) == NON_PERSISTENT) { + smpi_mpi_request_free(request); + } else { + (*request)->rdv = NULL; + (*request)->pair = NULL; + } } -void smpi_global_destroy() -{ - int i; +int smpi_mpi_test(MPI_Request* request, MPI_Status* status) { + int flag = (*request)->complete; - int size = SIMIX_host_get_number(); + if(flag) { + smpi_mpi_wait(request, status); + } + return flag; +} - // start/stop - SIMIX_mutex_destroy(smpi_global->start_stop_mutex); - SIMIX_cond_destroy(smpi_global->start_stop_cond); +int smpi_mpi_testany(int count, MPI_Request requests[], int* index, MPI_Status* status) { + int i, flag; + + *index = MPI_UNDEFINED; + flag = 0; + for(i = 0; i < count; i++) { + if(requests[i] != MPI_REQUEST_NULL && requests[i]->complete) { + smpi_mpi_wait(&requests[i], status); + *index = i; + flag = 1; + break; + } + } + return flag; +} - // processes - xbt_free(smpi_global->sender_processes); - xbt_free(smpi_global->receiver_processes); +void smpi_mpi_wait(MPI_Request* request, MPI_Status* status) { + print_request("wait", *request); + SIMIX_network_wait((*request)->pair, -1.0); + finish_wait(request, status); +} - // running hosts - SIMIX_mutex_destroy(smpi_global->running_hosts_count_mutex); +int smpi_mpi_waitany(int count, MPI_Request requests[], MPI_Status* status) { + xbt_dynar_t comms; + int i, size, index; + int* map; + + index = MPI_UNDEFINED; + if(count > 0) { + // First check for already completed requests + for(i = 0; i < count; i++) { + if(requests[i] != MPI_REQUEST_NULL && requests[i]->complete) { + index = i; + smpi_mpi_wait(&requests[index], status); + break; + } + } + if(index == MPI_UNDEFINED) { + // Otherwise, wait for a request to complete + comms = xbt_dynar_new(sizeof(smx_comm_t), NULL); + map = xbt_new(int, count); + size = 0; + DEBUG0("Wait for one of"); + for(i = 0; i < count; i++) { + if(requests[i] != MPI_REQUEST_NULL && requests[i]->complete == 0) { + print_request(" ", requests[i]); + xbt_dynar_push(comms, &requests[i]->pair); + map[size] = i; + size++; + } + } + if(size > 0) { + index = SIMIX_network_waitany(comms); + index = map[index]; + finish_wait(&requests[index], status); + } + xbt_free(map); + xbt_dynar_free(&comms); + } + } + return index; +} - // mallocators - xbt_mallocator_free(smpi_global->request_mallocator); - xbt_mallocator_free(smpi_global->message_mallocator); +void smpi_mpi_waitall(int count, MPI_Request requests[], MPI_Status status[]) { + int index; + MPI_Status stat; + + while(count > 0) { + index = smpi_mpi_waitany(count, requests, &stat); + if(index == MPI_UNDEFINED) { + break; + } + if(status != MPI_STATUS_IGNORE) { + memcpy(&status[index], &stat, sizeof(stat)); + } + // FIXME: check this -v + // Move the last request to the found position + requests[index] = requests[count - 1]; + requests[count - 1] = MPI_REQUEST_NULL; + count--; + } +} - for(i = 0; i < size; i++) { - xbt_fifo_free(smpi_global->pending_send_request_queues[i]); - SIMIX_mutex_destroy(smpi_global->pending_send_request_queues_mutexes[i]); - xbt_fifo_free(smpi_global->pending_recv_request_queues[i]); - SIMIX_mutex_destroy(smpi_global->pending_recv_request_queues_mutexes[i]); - xbt_fifo_free(smpi_global->received_message_queues[i]); - SIMIX_mutex_destroy(smpi_global->received_message_queues_mutexes[i]); - xbt_os_timer_free(smpi_global->timers[i]); - SIMIX_mutex_destroy(smpi_global->timers_mutexes[i]); - } +int smpi_mpi_waitsome(int incount, MPI_Request requests[], int* indices, MPI_Status status[]) { + int i, count; + + count = 0; + for(i = 0; i < incount; i++) { + if(requests[i] != MPI_REQUEST_NULL && requests[i]->complete) { + smpi_mpi_wait(&requests[i], status != MPI_STATUS_IGNORE ? &status[i] : MPI_STATUS_IGNORE); + indices[count] = i; + count++; + } + } + return count; +} - xbt_free(smpi_global->pending_send_request_queues); - xbt_free(smpi_global->pending_send_request_queues_mutexes); - xbt_free(smpi_global->pending_recv_request_queues); - xbt_free(smpi_global->pending_recv_request_queues_mutexes); - xbt_free(smpi_global->received_message_queues); - xbt_free(smpi_global->received_message_queues_mutexes); - xbt_free(smpi_global->timers); - xbt_free(smpi_global->timers_mutexes); +void smpi_mpi_bcast(void* buf, int count, MPI_Datatype datatype, int root, MPI_Comm comm) { + // arity=2: a binary tree, arity=4 seem to be a good setting (see P2P-MPI)) + nary_tree_bcast(buf, count, datatype, root, comm, 4); +} - xbt_free(smpi_global); +void smpi_mpi_barrier(MPI_Comm comm) { + // arity=2: a binary tree, arity=4 seem to be a good setting (see P2P-MPI)) + nary_tree_barrier(comm, 4); } -int smpi_run_simulation(int argc, char **argv) -{ - smx_cond_t cond = NULL; - smx_action_t action = NULL; +void smpi_mpi_gather(void* sendbuf, int sendcount, MPI_Datatype sendtype, void* recvbuf, int recvcount, MPI_Datatype recvtype, int root, MPI_Comm comm) { + int system_tag = 666; + int rank, size, src, index, sendsize, recvsize; + MPI_Request* requests; + + rank = smpi_comm_rank(comm); + size = smpi_comm_size(comm); + if(rank != root) { + // Send buffer to root + smpi_mpi_send(sendbuf, sendcount, sendtype, root, system_tag, comm); + } else { + sendsize = smpi_datatype_size(sendtype); + recvsize = smpi_datatype_size(recvtype); + // Local copy from root + memcpy(&((char*)recvbuf)[root * recvcount * recvsize], sendbuf, sendcount * sendsize * sizeof(char)); + // Receive buffers from senders + requests = xbt_new(MPI_Request, size - 1); + index = 0; + for(src = 0; src < size; src++) { + if(src != root) { + requests[index] = smpi_irecv_init(&((char*)recvbuf)[src * recvcount * recvsize], recvcount, recvtype, src, system_tag, comm); + index++; + } + } + // Wait for completion of irecv's. + smpi_mpi_startall(size - 1, requests); + smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE); + xbt_free(requests); + } +} - xbt_fifo_t actions_failed = xbt_fifo_new(); - xbt_fifo_t actions_done = xbt_fifo_new(); - - srand(SMPI_RAND_SEED); - - SIMIX_global_init(&argc, argv); - - SIMIX_function_register("smpi_simulated_main", smpi_simulated_main); - SIMIX_function_register("smpi_sender", smpi_sender); - SIMIX_function_register("smpi_receiver", smpi_receiver); - - // FIXME: ought to verify these files... - SIMIX_create_environment(argv[1]); - - // must initialize globals between creating environment and launching app.... - smpi_global_init(); - - SIMIX_launch_application(argv[2]); - - /* Prepare to display some more info when dying on Ctrl-C pressing */ - // FIXME: doesn't work - //signal(SIGINT, inthandler); - - /* Clean IO before the run */ - fflush(stdout); - fflush(stderr); - - while (SIMIX_solve(actions_done, actions_failed) != -1.0) { - while ((action = xbt_fifo_pop(actions_failed))) { - DEBUG1("** %s failed **", action->name); - while ((cond = xbt_fifo_pop(action->cond_list))) { - SIMIX_cond_broadcast(cond); - } - SIMIX_action_destroy(action); - } - while ((action = xbt_fifo_pop(actions_done))) { - DEBUG1("** %s done **",action->name); - while ((cond = xbt_fifo_pop(action->cond_list))) { - SIMIX_cond_broadcast(cond); - } - SIMIX_action_destroy(action); - } - } +void smpi_mpi_gatherv(void* sendbuf, int sendcount, MPI_Datatype sendtype, void* recvbuf, int* recvcounts, int* displs, MPI_Datatype recvtype, int root, MPI_Comm comm) { + int system_tag = 666; + int rank, size, src, index, sendsize; + MPI_Request* requests; + + rank = smpi_comm_rank(comm); + size = smpi_comm_size(comm); + if(rank != root) { + // Send buffer to root + smpi_mpi_send(sendbuf, sendcount, sendtype, root, system_tag, comm); + } else { + sendsize = smpi_datatype_size(sendtype); + // Local copy from root + memcpy(&((char*)recvbuf)[displs[root]], sendbuf, sendcount * sendsize * sizeof(char)); + // Receive buffers from senders + requests = xbt_new(MPI_Request, size - 1); + index = 0; + for(src = 0; src < size; src++) { + if(src != root) { + requests[index] = smpi_irecv_init(&((char*)recvbuf)[displs[src]], recvcounts[src], recvtype, src, system_tag, comm); + index++; + } + } + // Wait for completion of irecv's. + smpi_mpi_startall(size - 1, requests); + smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE); + xbt_free(requests); + } +} - xbt_fifo_free(actions_failed); - xbt_fifo_free(actions_done); +void smpi_mpi_allgather(void* sendbuf, int sendcount, MPI_Datatype sendtype, void* recvbuf, int recvcount, MPI_Datatype recvtype, MPI_Comm comm) { + int system_tag = 666; + int rank, size, other, index, sendsize, recvsize; + MPI_Request* requests; + + rank = smpi_comm_rank(comm); + size = smpi_comm_size(comm); + sendsize = smpi_datatype_size(sendtype); + recvsize = smpi_datatype_size(recvtype); + // Local copy from self + memcpy(&((char*)recvbuf)[rank * recvcount * recvsize], sendbuf, sendcount * sendsize * sizeof(char)); + // Send/Recv buffers to/from others; + requests = xbt_new(MPI_Request, 2 * (size - 1)); + index = 0; + for(other = 0; other < size; other++) { + if(other != rank) { + requests[index] = smpi_isend_init(sendbuf, sendcount, sendtype, other, system_tag, comm); + index++; + requests[index] = smpi_irecv_init(&((char*)recvbuf)[other * recvcount * recvsize], recvcount, recvtype, other, system_tag, comm); + index++; + } + } + // Wait for completion of all comms. + smpi_mpi_startall(2 * (size - 1), requests); + smpi_mpi_waitall(2 * (size - 1), requests, MPI_STATUS_IGNORE); + xbt_free(requests); +} - INFO1("simulation time %g", SIMIX_get_clock()); +void smpi_mpi_allgatherv(void* sendbuf, int sendcount, MPI_Datatype sendtype, void* recvbuf, int* recvcounts, int* displs, MPI_Datatype recvtype, MPI_Comm comm) { + int system_tag = 666; + int rank, size, other, index, sendsize, recvsize; + MPI_Request* requests; + + rank = smpi_comm_rank(comm); + size = smpi_comm_size(comm); + sendsize = smpi_datatype_size(sendtype); + recvsize = smpi_datatype_size(recvtype); + // Local copy from self + memcpy(&((char*)recvbuf)[displs[rank]], sendbuf, sendcount * sendsize * sizeof(char)); + // Send buffers to others; + requests = xbt_new(MPI_Request, 2 * (size - 1)); + index = 0; + for(other = 0; other < size; other++) { + if(other != rank) { + requests[index] = smpi_isend_init(sendbuf, sendcount, sendtype, other, system_tag, comm); + index++; + requests[index] = smpi_irecv_init(&((char*)recvbuf)[displs[other]], recvcounts[other], recvtype, other, system_tag, comm); + index++; + } + } + // Wait for completion of all comms. + smpi_mpi_startall(2 * (size - 1), requests); + smpi_mpi_waitall(2 * (size - 1), requests, MPI_STATUS_IGNORE); + xbt_free(requests); +} - smpi_global_destroy(); - - SIMIX_clean(); - - return 0; -} - -void smpi_mpi_land_func(void *x, void *y, void *z) -{ - *(int *)z = *(int *)x && *(int *)y; -} - -void smpi_mpi_sum_func(void *x, void *y, void *z) -{ - *(int *)z = *(int *)x + *(int *)y; -} - - -void smpi_mpi_init() -{ - smx_process_t process; - smx_host_t host; - smx_host_t *hosts; - int size; - - 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 - process = SIMIX_process_self(); - host = SIMIX_host_self(); - hosts = SIMIX_host_get_table(); - size = SIMIX_host_get_number(); - - // node 0 sets the globals - if (host == hosts[0]) { - - 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->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->hosts = hosts; - smpi_mpi_global->mpi_comm_world->processes = xbt_new(smx_process_t, size); - smpi_mpi_global->mpi_comm_world->processes[0] = process; - - // mpi datatypes - smpi_mpi_global->mpi_byte = xbt_new(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->size = sizeof(int); - smpi_mpi_global->mpi_double = xbt_new(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->func = smpi_mpi_land_func; - smpi_mpi_global->mpi_sum = xbt_new(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); - 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_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 - 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); +void smpi_mpi_scatter(void* sendbuf, int sendcount, MPI_Datatype sendtype, void* recvbuf, int recvcount, MPI_Datatype recvtype, int root, MPI_Comm comm) { + int system_tag = 666; + int rank, size, dst, index, sendsize, recvsize; + MPI_Request* requests; + + rank = smpi_comm_rank(comm); + size = smpi_comm_size(comm); + if(rank != root) { + // Recv buffer from root + smpi_mpi_recv(recvbuf, recvcount, recvtype, root, system_tag, comm, MPI_STATUS_IGNORE); + } else { + sendsize = smpi_datatype_size(sendtype); + recvsize = smpi_datatype_size(recvtype); + // Local copy from root + memcpy(recvbuf, &((char*)sendbuf)[root * sendcount * sendsize], recvcount * recvsize * sizeof(char)); + // Send buffers to receivers + requests = xbt_new(MPI_Request, size - 1); + index = 0; + for(dst = 0; dst < size; dst++) { + if(dst != root) { + requests[index] = smpi_isend_init(&((char*)sendbuf)[dst * sendcount * sendsize], sendcount, sendtype, dst, system_tag, comm); + index++; + } + } + // Wait for completion of isend's. + smpi_mpi_startall(size - 1, requests); + smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE); + xbt_free(requests); + } +} - return; +void smpi_mpi_scatterv(void* sendbuf, int* sendcounts, int* displs, MPI_Datatype sendtype, void* recvbuf, int recvcount, MPI_Datatype recvtype, int root, MPI_Comm comm) { + int system_tag = 666; + int rank, size, dst, index, sendsize, recvsize; + MPI_Request* requests; + + rank = smpi_comm_rank(comm); + size = smpi_comm_size(comm); + if(rank != root) { + // Recv buffer from root + smpi_mpi_recv(recvbuf, recvcount, recvtype, root, system_tag, comm, MPI_STATUS_IGNORE); + } else { + sendsize = smpi_datatype_size(sendtype); + recvsize = smpi_datatype_size(recvtype); + // Local copy from root + memcpy(recvbuf, &((char*)sendbuf)[displs[root]], recvcount * recvsize * sizeof(char)); + // Send buffers to receivers + requests = xbt_new(MPI_Request, size - 1); + index = 0; + for(dst = 0; dst < size; dst++) { + if(dst != root) { + requests[index] = smpi_isend_init(&((char*)sendbuf)[displs[dst]], sendcounts[dst], sendtype, dst, system_tag, comm); + index++; + } + } + // Wait for completion of isend's. + smpi_mpi_startall(size - 1, requests); + smpi_mpi_waitall(size - 1, requests, MPI_STATUS_IGNORE); + xbt_free(requests); + } } -void smpi_mpi_finalize() -{ - int i; +void smpi_mpi_reduce(void* sendbuf, void* recvbuf, int count, MPI_Datatype datatype, MPI_Op op, int root, MPI_Comm comm) { + int system_tag = 666; + int rank, size, src, index, datasize; + MPI_Request* requests; + void** tmpbufs; + + rank = smpi_comm_rank(comm); + size = smpi_comm_size(comm); + if(rank != root) { + // Send buffer to root + smpi_mpi_send(sendbuf, count, datatype, root, system_tag, comm); + } else { + datasize = smpi_datatype_size(datatype); + // Local copy from root + memcpy(recvbuf, sendbuf, count * datasize * sizeof(char)); + // Receive buffers from senders + //TODO: make a MPI_barrier here ? + requests = xbt_new(MPI_Request, size - 1); + tmpbufs = xbt_new(void*, size - 1); + index = 0; + for(src = 0; src < size; src++) { + if(src != root) { + tmpbufs[index] = xbt_malloc(count * datasize); + requests[index] = smpi_irecv_init(tmpbufs[index], count, datatype, src, system_tag, comm); + index++; + } + } + // Wait for completion of irecv's. + smpi_mpi_startall(size - 1, requests); + for(src = 0; src < size - 1; src++) { + index = smpi_mpi_waitany(size - 1, requests, MPI_STATUS_IGNORE); + if(index == MPI_UNDEFINED) { + break; + } + smpi_op_apply(op, tmpbufs[index], recvbuf, &count, &datatype); + } + for(index = 0; index < size - 1; index++) { + xbt_free(tmpbufs[index]); + } + xbt_free(tmpbufs); + xbt_free(requests); + } +} - SIMIX_mutex_lock(smpi_global->running_hosts_count_mutex); - i = --smpi_global->running_hosts_count; - SIMIX_mutex_unlock(smpi_global->running_hosts_count_mutex); +void smpi_mpi_allreduce(void* sendbuf, void* recvbuf, int count, MPI_Datatype datatype, MPI_Op op, MPI_Comm comm) { + smpi_mpi_reduce(sendbuf, recvbuf, count, datatype, op, 0, comm); + smpi_mpi_bcast(recvbuf, count, datatype, 0, comm); + +/* +FIXME: buggy implementation + + int system_tag = 666; + int rank, size, other, index, datasize; + MPI_Request* requests; + void** tmpbufs; + + rank = smpi_comm_rank(comm); + size = smpi_comm_size(comm); + datasize = smpi_datatype_size(datatype); + // Local copy from self + memcpy(recvbuf, sendbuf, count * datasize * sizeof(char)); + // Send/Recv buffers to/from others; + //TODO: make a MPI_barrier here ? + requests = xbt_new(MPI_Request, 2 * (size - 1)); + tmpbufs = xbt_new(void*, size - 1); + index = 0; + for(other = 0; other < size; other++) { + if(other != rank) { + tmpbufs[index / 2] = xbt_malloc(count * datasize); + requests[index] = smpi_mpi_isend(sendbuf, count, datatype, other, system_tag, comm); + requests[index + 1] = smpi_mpi_irecv(tmpbufs[index / 2], count, datatype, other, system_tag, comm); + index += 2; + } + } + // Wait for completion of all comms. + for(other = 0; other < 2 * (size - 1); other++) { + index = smpi_mpi_waitany(size - 1, requests, MPI_STATUS_IGNORE); + if(index == MPI_UNDEFINED) { + break; + } + if((index & 1) == 1) { + // Request is odd: it's a irecv + smpi_op_apply(op, tmpbufs[index / 2], recvbuf, &count, &datatype); + } + } + for(index = 0; index < size - 1; index++) { + xbt_free(tmpbufs[index]); + } + xbt_free(tmpbufs); + xbt_free(requests); +*/ +} - SIMIX_mutex_lock(smpi_global->start_stop_mutex); - smpi_global->ready_process_count--; - SIMIX_mutex_unlock(smpi_global->start_stop_mutex); - - if (0 >= i) { - - // wake up senders/receivers - for (i = 0; i < smpi_mpi_global->mpi_comm_world->size; 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]); - } - } - - // wait for senders/receivers to exit... - SIMIX_mutex_lock(smpi_global->start_stop_mutex); - if (smpi_global->ready_process_count > 0) { - SIMIX_cond_wait(smpi_global->start_stop_cond, smpi_global->start_stop_mutex); - } - SIMIX_mutex_unlock(smpi_global->start_stop_mutex); - - 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); - xbt_free(smpi_mpi_global->mpi_int); - xbt_free(smpi_mpi_global->mpi_double); - - xbt_free(smpi_mpi_global->mpi_land); - xbt_free(smpi_mpi_global->mpi_sum); - - xbt_free(smpi_mpi_global); - } - -} - -// FIXME: could cause trouble with multithreaded procs on same host... -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; -} - -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; - 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]); - SIMIX_mutex_unlock(smpi_global->timers_mutexes[rank]); - - host = smpi_mpi_global->mpi_comm_world->hosts[rank]; - compute_action = SIMIX_action_execute(host, NULL, duration * SMPI_DEFAULT_SPEED); - mutex = SIMIX_mutex_init(); - cond = SIMIX_cond_init(); - - SIMIX_register_condition_to_action(compute_action, cond); - SIMIX_register_action_to_condition(compute_action, cond); - SIMIX_mutex_lock(mutex); - SIMIX_cond_wait(cond, mutex); - SIMIX_mutex_unlock(mutex); - - SIMIX_mutex_destroy(mutex); - SIMIX_cond_destroy(cond); - - // FIXME: check for success/failure? - - return; -} - -void smpi_barrier(smpi_mpi_communicator_t *comm) -{ - - SIMIX_mutex_lock(comm->barrier_mutex); - if(++comm->barrier_count < comm->size) { - SIMIX_cond_wait(comm->barrier_cond, comm->barrier_mutex); - } else { - comm->barrier_count = 0; - SIMIX_cond_broadcast(comm->barrier_cond); - } - SIMIX_mutex_unlock(comm->barrier_mutex); - - return; -} - -// FIXME: smarter algorithm... -int smpi_comm_rank(smpi_mpi_communicator_t *comm, smx_host_t host) -{ - int i; - for(i = 0; i < comm->size && host != comm->hosts[i]; i++); - if (i >= comm->size) i = -1; - return i; -} - -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; - - *request = NULL; - - if (0 > count) { - retval = MPI_ERR_COUNT; - } else if (NULL == buf) { - retval = MPI_ERR_INTERN; - } else if (NULL == datatype) { - retval = MPI_ERR_TYPE; - } else if (NULL == comm) { - retval = MPI_ERR_COMM; - } else if (MPI_ANY_SOURCE != src && (0 > src || comm->size <= src)) { - retval = MPI_ERR_RANK; - } else if (0 > dst || comm->size <= dst) { - retval = MPI_ERR_RANK; - } else if (0 > tag) { - retval = MPI_ERR_TAG; - } else { - *request = xbt_mallocator_get(smpi_global->request_mallocator); - (*request)->comm = comm; - (*request)->src = src; - (*request)->dst = dst; - (*request)->tag = tag; - (*request)->buf = buf; - (*request)->count = count; - (*request)->datatype = datatype; - } - return retval; -} - -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); - - if (NULL != request) { - SIMIX_mutex_lock(smpi_global->pending_send_request_queues_mutexes[rank]); - xbt_fifo_push(smpi_global->pending_send_request_queues[rank], request); - SIMIX_mutex_unlock(smpi_global->pending_send_request_queues_mutexes[rank]); - } - - if (SIMIX_process_is_suspended(smpi_global->sender_processes[rank])) { - SIMIX_process_resume(smpi_global->sender_processes[rank]); - } - - return retval; -} - -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); - - if (NULL != request) { - SIMIX_mutex_lock(smpi_global->pending_recv_request_queues_mutexes[rank]); - xbt_fifo_push(smpi_global->pending_recv_request_queues[rank], request); - SIMIX_mutex_unlock(smpi_global->pending_recv_request_queues_mutexes[rank]); - } - - if (SIMIX_process_is_suspended(smpi_global->receiver_processes[rank])) { - SIMIX_process_resume(smpi_global->receiver_processes[rank]); - } - - return retval; -} - -void smpi_wait(smpi_mpi_request_t *request, smpi_mpi_status_t *status) -{ - if (NULL != request) { - SIMIX_mutex_lock(request->mutex); - if (!request->completed) { - SIMIX_cond_wait(request->cond, request->mutex); - } - if (NULL != status) { - status->MPI_SOURCE = request->src; - } - SIMIX_mutex_unlock(request->mutex); - } -} - -// FIXME: move into own file -int smpi_gettimeofday(struct timeval *tv, struct timezone *tz) -{ - double now; - int retval = 0; - smpi_bench_end(); - if (NULL == tv) { - retval = -1; - } else { - now = SIMIX_get_clock(); - tv->tv_sec = now; - tv->tv_usec = ((now - (double)tv->tv_sec) * 1000000.0); - } - smpi_bench_begin(); - return retval; -} - -unsigned int smpi_sleep(unsigned int seconds) -{ - smx_mutex_t mutex; - smx_cond_t cond; - smx_host_t host; - smx_action_t sleep_action; - - smpi_bench_end(); - host = SIMIX_host_self(); - sleep_action = SIMIX_action_sleep(host, seconds); - mutex = SIMIX_mutex_init(); - cond = SIMIX_cond_init(); - - SIMIX_register_condition_to_action(sleep_action, cond); - SIMIX_register_action_to_condition(sleep_action, cond); - SIMIX_mutex_lock(mutex); - SIMIX_cond_wait(cond, mutex); - SIMIX_mutex_unlock(mutex); - - SIMIX_mutex_destroy(mutex); - SIMIX_cond_destroy(cond); - - // FIXME: check for success/failure? - - smpi_bench_begin(); - return 0; -} - -void smpi_exit(int status) -{ - smpi_bench_end(); - SIMIX_mutex_lock(smpi_global->running_hosts_count_mutex); - smpi_global->running_hosts_count--; - SIMIX_mutex_unlock(smpi_global->running_hosts_count_mutex); - SIMIX_process_kill(SIMIX_process_self()); - return; +void smpi_mpi_scan(void* sendbuf, void* recvbuf, int count, MPI_Datatype datatype, MPI_Op op, MPI_Comm comm) { + int system_tag = 666; + int rank, size, other, index, datasize; + int total; + MPI_Request* requests; + void** tmpbufs; + + rank = smpi_comm_rank(comm); + size = smpi_comm_size(comm); + datasize = smpi_datatype_size(datatype); + // Local copy from self + memcpy(recvbuf, sendbuf, count * datasize * sizeof(char)); + // Send/Recv buffers to/from others; + total = rank + (size - (rank + 1)); + requests = xbt_new(MPI_Request, total); + tmpbufs = xbt_new(void*, rank); + index = 0; + for(other = 0; other < rank; other++) { + tmpbufs[index] = xbt_malloc(count * datasize); + requests[index] = smpi_irecv_init(tmpbufs[index], count, datatype, other, system_tag, comm); + index++; + } + for(other = rank + 1; other < size; other++) { + requests[index] = smpi_isend_init(sendbuf, count, datatype, other, system_tag, comm); + index++; + } + // Wait for completion of all comms. + smpi_mpi_startall(size - 1, requests); + for(other = 0; other < total; other++) { + index = smpi_mpi_waitany(size - 1, requests, MPI_STATUS_IGNORE); + if(index == MPI_UNDEFINED) { + break; + } + if(index < rank) { + // #Request is below rank: it's a irecv + smpi_op_apply(op, tmpbufs[index], recvbuf, &count, &datatype); + } + } + for(index = 0; index < size - 1; index++) { + xbt_free(tmpbufs[index]); + } + xbt_free(tmpbufs); + xbt_free(requests); }