src/smpi/smpi_receiver.c

   1 #include "private.h"
   2
   3 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_receiver, smpi,
   4                                 "Logging specific to SMPI (receiver)");
   5
   6 int smpi_receiver(int argc, char **argv)
   7 {
   8   smx_process_t self;
   9   int index;
  10
  11   xbt_fifo_t request_queue;
  12   smx_mutex_t request_queue_mutex;
  13   xbt_fifo_t message_queue;
  14   smx_mutex_t message_queue_mutex;
  15
  16   int running_hosts_count;
  17
  18   smpi_mpi_request_t request;
  19   smpi_received_message_t message;
  20
  21   xbt_fifo_item_t request_item;
  22   xbt_fifo_item_t message_item;
  23
  24   self = SIMIX_process_self();
  25
  26   // make sure root is done before own initialization
  27   SIMIX_mutex_lock(smpi_global->start_stop_mutex);
  28   while (!smpi_global->root_ready) {
  29     SIMIX_cond_wait(smpi_global->start_stop_cond,
  30                     smpi_global->start_stop_mutex);
  31   }
  32   SIMIX_mutex_unlock(smpi_global->start_stop_mutex);
  33
  34   index = smpi_host_index();
  35
  36   request_queue = smpi_global->pending_recv_request_queues[index];
  37   request_queue_mutex =
  38     smpi_global->pending_recv_request_queues_mutexes[index];
  39   message_queue = smpi_global->received_message_queues[index];
  40   message_queue_mutex = smpi_global->received_message_queues_mutexes[index];
  41
  42   smpi_global->receiver_processes[index] = self;
  43
  44   // wait for all nodes to signal initializatin complete
  45   SIMIX_mutex_lock(smpi_global->start_stop_mutex);
  46   smpi_global->ready_process_count++;
  47   if (smpi_global->ready_process_count >= 3 * smpi_global->host_count) {
  48     SIMIX_cond_broadcast(smpi_global->start_stop_cond);
  49   }
  50   while (smpi_global->ready_process_count < 3 * smpi_global->host_count) {
  51     SIMIX_cond_wait(smpi_global->start_stop_cond,
  52                     smpi_global->start_stop_mutex);
  53   }
  54   SIMIX_mutex_unlock(smpi_global->start_stop_mutex);
  55
  56   do {
  57
  58     // FIXME: better algorithm, maybe some kind of balanced tree? or a heap?
  59
  60     // FIXME: not the best way to request multiple locks...
  61     SIMIX_mutex_lock(request_queue_mutex);
  62     SIMIX_mutex_lock(message_queue_mutex);
  63     for (request_item = xbt_fifo_get_first_item(request_queue);
  64          NULL != request_item;
  65          request_item = xbt_fifo_get_next_item(request_item)) {
  66       request = xbt_fifo_get_item_content(request_item);
  67       for (message_item = xbt_fifo_get_first_item(message_queue);
  68            NULL != message_item;
  69            message_item = xbt_fifo_get_next_item(message_item)) {
  70         message = xbt_fifo_get_item_content(message_item);
  71         if (request->comm == message->comm &&
  72             (MPI_ANY_SOURCE == request->src || request->src == message->src)
  73             && (MPI_ANY_TAG == request->tag || request->tag == message->tag)) {
  74           xbt_fifo_remove_item(request_queue, request_item);
  75           xbt_fifo_free_item(request_item);
  76           xbt_fifo_remove_item(message_queue, message_item);
  77           xbt_fifo_free_item(message_item);
  78           goto stopsearch;
  79         }
  80       }
  81     }
  82
  83     request = NULL;
  84     message = NULL;
  85
  86   stopsearch:
  87     SIMIX_mutex_unlock(message_queue_mutex);
  88     SIMIX_mutex_unlock(request_queue_mutex);
  89
  90     if (NULL == request || NULL == message) {
  91       SIMIX_process_suspend(self);
  92     } else {
  93
  94       SIMIX_mutex_lock(request->mutex);
  95       memcpy(request->buf, message->buf,
  96              request->datatype->size * request->count);
  97       request->src = message->src;
  98       request->data = message->data;
  99       request->forward = message->forward;
 100
 101       if (0 == request->forward) {
 102         request->completed = 1;
 103         SIMIX_cond_broadcast(request->cond);
 104       } else {
 105         request->src = request->comm->index_to_rank_map[index];
 106         request->dst = (request->src + 1) % request->comm->size;
 107         smpi_mpi_isend(request);
 108       }
 109
 110       SIMIX_mutex_unlock(request->mutex);
 111
 112       xbt_free(message->buf);
 113       xbt_mallocator_release(smpi_global->message_mallocator, message);
 114
 115     }
 116
 117     SIMIX_mutex_lock(smpi_global->running_hosts_count_mutex);
 118     running_hosts_count = smpi_global->running_hosts_count;
 119     SIMIX_mutex_unlock(smpi_global->running_hosts_count_mutex);
 120
 121   } while (0 < running_hosts_count);
 122
 123   return 0;
 124 }