+++ /dev/null
-#include "private.h"
-
-XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_sender, smpi,
- "Logging specific to SMPI (sender)");
-
-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;
-
- int index;
-
- xbt_fifo_t request_queue;
-
- smpi_mpi_request_t request;
-
- smx_host_t dhost;
-
- smx_action_t action;
-
- e_surf_action_state_t state;
-
- smpi_received_message_t message;
-
- int dindex;
-
- self = SIMIX_process_self();
- shost = SIMIX_host_self();
-
- index = mydata->index;
-
- DEBUG0("Up and running");
-
- request_queue = mydata->pending_send_request_queue;
-
- while (1) {
- request = xbt_fifo_shift(request_queue);
-
- if (NULL != request) {
- message = xbt_mallocator_get(smpi_global->message_mallocator);
-
- SIMIX_mutex_lock(request->mutex);
-
- 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);
-
- 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]);
-
- 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;
-
- 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, "communication",
- request->datatype->size * request->count,
- -1.0);
-
- SIMIX_register_action_to_condition(action, request->cond);
-
- 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);
- }
-
- xbt_fifo_push(remote_process->received_message_queue, message);
-
- SIMIX_unregister_action_to_condition(action, request->cond);
- SIMIX_action_destroy(action);
-
- SIMIX_mutex_unlock(request->mutex);
-
- // wake up receiver if necessary
- SIMIX_process_resume(remote_process->receiver);
-
- } 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;
-}
