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