communicator.cpp

   1 #include <algorithm>
   2 #include <functional>
   3 #include <msg/msg.h>
   4 #include <xbt/log.h>
   5
   6 XBT_LOG_EXTERNAL_DEFAULT_CATEGORY(comm);
   7
   8 #include "misc.h"
   9 #include "simgrid_features.h"
  10 #include "tracing.h"
  11
  12 #include "communicator.h"
  13
  14 namespace {
  15
  16     void check_for_lost_messages(size_t size, const char* descr)
  17     {
  18         if (size)
  19             XBT_WARN("lost %zu %s message%s!", size, descr, ESSE(size));
  20     }
  21
  22 }
  23
  24 communicator::communicator()
  25     : host(static_cast<hostdata*>(MSG_host_get_data(MSG_host_self())))
  26 {
  27     receiver_thread = new_msg_thread("receiver",
  28                                      std::bind(&communicator::receiver, this));
  29     receiver_thread->start();
  30 }
  31
  32 communicator::~communicator()
  33 {
  34     m_task_t task;
  35
  36     XBT_DEBUG("send finalize to receiver/ctrl");
  37     task = MSG_task_create("finalize", 0.0, 0, NULL);
  38     MSG_task_send(task, host->get_ctrl_mbox());
  39
  40     XBT_DEBUG("send finalize to receiver/data");
  41     task = MSG_task_create("finalize", 0.0, 0, NULL);
  42     MSG_task_send(task, host->get_data_mbox());
  43
  44     receiver_thread->wait();
  45     delete receiver_thread;
  46
  47     check_for_lost_messages(ctrl_received.size(), "received ctrl");
  48     check_for_lost_messages(data_received.size(), "received data");
  49     check_for_lost_messages(ctrl_sent.size(), "sent ctrl");
  50     check_for_lost_messages(data_sent.size(), "sent data");
  51 }
  52
  53 msg_comm_t communicator::real_send(const char* dest, message* msg)
  54 {
  55     XBT_DEBUG("send %s to %s", msg->to_string().c_str(), dest);
  56     m_task_t task = MSG_task_create("message", 0.0, msg->get_size(), msg);
  57     MSG_task_set_category(task,
  58                           msg->get_type() == message::DATA ?
  59                           TRACE_CAT_DATA : TRACE_CAT_CTRL);
  60     return MSG_task_isend(task, dest);
  61 }
  62
  63 void communicator::real_flush(sent_comm_type& sent_comm, bool wait)
  64 {
  65     sent_comm_type::iterator bound =
  66         std::remove_if(sent_comm.begin(), sent_comm.end(),
  67                        comm_test_n_destroy);
  68     sent_comm.erase(bound, sent_comm.end());
  69     if (wait && !sent_comm.empty()) {
  70         size_t size = sent_comm.size();
  71         msg_comm_t* comms = new msg_comm_t[size];
  72         std::copy(sent_comm.begin(), sent_comm.end(), comms);
  73         sent_comm.clear();
  74         MSG_comm_waitall(comms, size, -1.0);
  75         std::for_each(comms, comms + size, comm_check_n_destroy);
  76         delete[] comms;
  77     }
  78 }
  79
  80 void communicator::receiver()
  81 {
  82     xbt_dynar_t comms = xbt_dynar_new(sizeof(msg_comm_t), NULL);
  83     struct channel {
  84         msg_comm_t comm;
  85         m_task_t task;
  86         const char* mbox;
  87         message_queue& received;
  88     };
  89     channel chan[] = { { NULL, NULL, host->get_ctrl_mbox(), ctrl_received },
  90                        { NULL, NULL, host->get_data_mbox(), data_received } };
  91     const int chan_size = (sizeof chan) / (sizeof chan[0]);
  92
  93     for (int i = 0 ; i < chan_size ; ++i) {
  94         chan[i].comm = MSG_task_irecv(&chan[i].task, chan[i].mbox);
  95         xbt_dynar_push(comms, &chan[i].comm);
  96     }
  97
  98     while (!xbt_dynar_is_empty(comms)) {
  99
 100         int index = MSG_comm_waitany(comms);
 101         msg_comm_t finished_comm = xbt_dynar_get_as(comms, index, msg_comm_t);
 102         channel* ch;
 103
 104         for (ch = chan ; ch->comm != finished_comm ; ++ch)
 105             /* nop */;
 106
 107         comm_check_n_destroy(ch->comm);
 108         if (strcmp(MSG_task_get_name(ch->task), "finalize")) {
 109             XBT_DEBUG("received message on %s", ch->mbox);
 110             ch->received.push(ch->task);
 111             ch->task = NULL;
 112             ch->comm = MSG_task_irecv(&ch->task, ch->mbox);
 113             xbt_dynar_set_as(comms, index, msg_comm_t, ch->comm);
 114         } else {
 115             XBT_DEBUG("received finalize on %s", ch->mbox);
 116             MSG_task_destroy(ch->task);
 117             ch->task = NULL;
 118             ch->comm = NULL;
 119             xbt_dynar_remove_at(comms, index, NULL);
 120         }
 121
 122     }
 123     xbt_dynar_free(&comms);
 124 }
 125
 126 void communicator::comm_check_n_destroy(msg_comm_t comm)
 127 {
 128     xbt_assert(MSG_comm_get_status(comm) == MSG_OK);
 129     MSG_comm_destroy(comm);
 130 }
 131
 132 bool communicator::comm_test_n_destroy(msg_comm_t comm)
 133 {
 134     if (MSG_comm_test(comm)) {
 135         comm_check_n_destroy(comm);
 136         return true;
 137     } else
 138         return false;
 139 }
 140
 141 // Local variables:
 142 // mode: c++
 143 // End: