communicator.cpp

   1 #include <algorithm>
   2 #include <cstring>
   3 #include <tr1/functional>
   4 #include <sstream>
   5 #include <msg/msg.h>
   6 #include <xbt/log.h>
   7
   8 XBT_LOG_EXTERNAL_DEFAULT_CATEGORY(comm);
   9
  10 #include "misc.h"
  11 #include "options.h"
  12 #include "simgrid_features.h"
  13 #include "tracing.h"
  14
  15 #include "communicator.h"
  16
  17 std::string message::to_string()
  18 {
  19     static const char* str[] = { "INFO", "CREDIT", "LOAD",
  20                                  "CTRL_CLOSE", "DATA_CLOSE" };
  21     std::ostringstream oss;
  22     oss << str[type] << ": " << amount;
  23     return oss.str();
  24 }
  25
  26 communicator::communicator()
  27     : host(static_cast<hostdata*>(MSG_host_get_data(MSG_host_self())))
  28     , receiver_mutex(xbt_mutex_init())
  29     , receiver_cond(xbt_cond_init())
  30 {
  31     xbt_mutex_acquire(receiver_mutex);
  32     receiver_thread =
  33         MSG_process_create("receiver", communicator::receiver_wrapper,
  34                            this, MSG_host_self());
  35     // wait for the receiver to be ready
  36     xbt_cond_wait(receiver_cond, receiver_mutex);
  37     xbt_mutex_release(receiver_mutex);
  38 }
  39
  40 communicator::~communicator()
  41 {
  42     m_task_t task;
  43
  44     XBT_DEBUG("send finalize to receiver/ctrl");
  45     task = MSG_task_create("finalize", 0.0, 0, NULL);
  46     MSG_task_send(task, host->get_ctrl_mbox());
  47
  48     XBT_DEBUG("send finalize to receiver/data");
  49     task = MSG_task_create("finalize", 0.0, 0, NULL);
  50     MSG_task_send(task, host->get_data_mbox());
  51
  52     xbt_mutex_acquire(receiver_mutex);
  53     while (receiver_thread) {
  54         XBT_DEBUG("waiting for receiver to terminate");
  55         xbt_cond_wait(receiver_cond, receiver_mutex);
  56     }
  57     xbt_mutex_release(receiver_mutex);
  58
  59     if (!received.empty())
  60         XBT_WARN("lost %zu received message%s!",
  61                  received.size(), ESSE(received.size()));
  62     if (!sent_comm.empty())
  63         XBT_WARN("lost %zu sent message%s!",
  64                  sent_comm.size(), ESSE(sent_comm.size()));
  65
  66     xbt_cond_destroy(receiver_cond);
  67     xbt_mutex_destroy(receiver_mutex);
  68 }
  69
  70 void communicator::send(const char* dest, message* msg)
  71 {
  72     XBT_DEBUG("send %s to %s", msg->to_string().c_str(), dest);
  73     double msg_size = sizeof *msg;
  74     if (msg->get_type() == message::LOAD)
  75         msg_size += opt::comm_cost(msg->get_amount());
  76     m_task_t task = MSG_task_create("message", 0.0, msg_size, msg);
  77     TRACE_msg_set_task_category(task,
  78                                 msg->get_type() == message::LOAD ?
  79                                 TRACE_CAT_DATA : TRACE_CAT_CTRL);
  80     msg_comm_t comm = MSG_task_isend(task, dest);
  81     sent_comm.push_back(comm);
  82 }
  83
  84 bool communicator::recv(message*& msg, m_host_t& from, double timeout)
  85 {
  86     if (timeout != 0) {
  87         volatile double deadline =
  88             timeout > 0 ? MSG_get_clock() + timeout : 0.0;
  89         xbt_mutex_acquire(receiver_mutex);
  90         while (received.empty() && (!deadline || deadline > MSG_get_clock())) {
  91             xbt_ex_t e;
  92             XBT_DEBUG("waiting for a message to come");
  93             TRY {
  94                 if (deadline)
  95                     xbt_cond_timedwait(receiver_cond, receiver_mutex,
  96                                        deadline - MSG_get_clock());
  97                 else
  98                     xbt_cond_wait(receiver_cond, receiver_mutex);
  99             }
 100             CATCH (e) {
 101                 if (e.category != timeout_error)
 102                     RETHROW;
 103                 xbt_ex_free(e);
 104             }
 105         }
 106         xbt_mutex_release(receiver_mutex);
 107     }
 108
 109     if (received.empty())
 110         return false;
 111
 112     m_task_t task = received.front();
 113     received.pop();
 114     msg = static_cast<message*>(MSG_task_get_data(task));
 115     from = MSG_task_get_source(task);
 116     MSG_task_destroy(task);
 117
 118     XBT_DEBUG("received %s from %s",
 119            msg->to_string().c_str(), MSG_host_get_name(from));
 120
 121     return true;
 122 }
 123
 124 void communicator::flush(bool wait)
 125 {
 126     sent_comm.remove_if(comm_test_n_destroy);
 127     if (wait && !sent_comm.empty()) {
 128         msg_comm_t comms[sent_comm.size()];
 129         std::copy(sent_comm.begin(), sent_comm.end(), comms);
 130         MSG_comm_waitall(comms, sent_comm.size(), -1.0);
 131         if (!MSG_WAIT_DESTROYS_COMMS)
 132             std::for_each(sent_comm.begin(), sent_comm.end(),
 133                           comm_check_n_destroy);
 134         sent_comm.clear();
 135     }
 136 }
 137
 138 int communicator::receiver_wrapper(int, char* [])
 139 {
 140     communicator* comm;
 141     comm = static_cast<communicator*>(MSG_process_get_data(MSG_process_self()));
 142     comm->receiver();
 143
 144     XBT_DEBUG("terminate");
 145     xbt_mutex_acquire(comm->receiver_mutex);
 146     comm->receiver_thread = NULL;
 147     xbt_cond_signal(comm->receiver_cond);
 148     xbt_mutex_release(comm->receiver_mutex);
 149
 150     return 0;
 151 }
 152
 153 void communicator::receiver()
 154 {
 155     xbt_dynar_t comms = xbt_dynar_new(sizeof(msg_comm_t), NULL);
 156     struct channel {
 157         msg_comm_t comm;
 158         m_task_t task;
 159         const char* mbox;
 160     };
 161     channel chan[] = { { NULL, NULL, host->get_ctrl_mbox() },
 162                        { NULL, NULL, host->get_data_mbox() } };
 163     const int chan_size = (sizeof chan) / (sizeof chan[0]);
 164
 165     for (int i = 0 ; i < chan_size ; ++i) {
 166         chan[i].comm = MSG_task_irecv(&chan[i].task, chan[i].mbox);
 167         xbt_dynar_push(comms, &chan[i].comm);
 168     }
 169
 170     XBT_DEBUG("receiver ready");
 171     xbt_mutex_acquire(receiver_mutex);
 172     xbt_cond_signal(receiver_cond); // signal master that we are ready
 173     xbt_mutex_release(receiver_mutex);
 174
 175     while (!xbt_dynar_is_empty(comms)) {
 176
 177         int index = MSG_comm_waitany(comms);
 178         msg_comm_t finished_comm = xbt_dynar_get_as(comms, index, msg_comm_t);
 179         channel* ch;
 180
 181         for (ch = chan ; ch->comm != finished_comm ; ++ch)
 182             /* nop */;
 183
 184         comm_check_n_destroy(ch->comm);
 185         if (strcmp(MSG_task_get_name(ch->task), "finalize")) {
 186             XBT_DEBUG("received message on %s", ch->mbox);
 187             xbt_mutex_acquire(receiver_mutex);
 188             received.push(ch->task);
 189             xbt_cond_signal(receiver_cond);
 190             xbt_mutex_release(receiver_mutex);
 191             ch->task = NULL;
 192             ch->comm = MSG_task_irecv(&ch->task, ch->mbox);
 193             xbt_dynar_set_as(comms, index, msg_comm_t, ch->comm);
 194         } else {
 195             XBT_DEBUG("received finalize on %s", ch->mbox);
 196             MSG_task_destroy(ch->task);
 197             ch->task = NULL;
 198             ch->comm = NULL;
 199             xbt_dynar_remove_at(comms, index, NULL);
 200         }
 201
 202     }
 203     xbt_dynar_free(&comms);
 204 }
 205
 206 void communicator::comm_check_n_destroy(msg_comm_t comm)
 207 {
 208     xbt_assert(MSG_comm_get_status(comm) == MSG_OK);
 209     MSG_comm_destroy(comm);
 210 }
 211
 212 bool communicator::comm_test_n_destroy(msg_comm_t comm)
 213 {
 214     if (MSG_comm_test(comm)) {
 215         comm_check_n_destroy(comm);
 216         return true;
 217     } else
 218         return false;
 219 }
 220
 221 // Local variables:
 222 // mode: c++
 223 // End: