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
  13 #include "communicator.h"
  14
  15 std::string message::to_string()
  16 {
  17     static const char* str[] = { "INFO", "CREDIT", "LOAD",
  18                                  "CTRL_CLOSE", "DATA_CLOSE" };
  19     std::ostringstream oss;
  20     oss << str[type] << ": " << amount;
  21     return oss.str();
  22 }
  23
  24 int communicator::send_count_before_flush = 4;
  25
  26 communicator::communicator()
  27     : host((hostdata* )MSG_host_get_data(MSG_host_self()))
  28     , mutex(xbt_mutex_init())
  29     , cond(xbt_cond_init())
  30     , send_counter(0)
  31     , ctrl_task(NULL)
  32     , ctrl_comm(NULL)
  33     , data_task(NULL)
  34     , data_comm(NULL)
  35 {
  36     xbt_mutex_acquire(mutex);
  37     receiver_process =
  38         MSG_process_create("receiver", communicator::receiver_wrapper,
  39                            this, MSG_host_self());
  40     xbt_mutex_release(mutex);
  41 }
  42
  43 communicator::~communicator()
  44 {
  45     m_task_t task;
  46
  47     DEBUG0("send finalize to receiver/ctrl");
  48     task = MSG_task_create("finalize", 0.0, 0, NULL);
  49     MSG_task_send(task, get_ctrl_mbox());
  50
  51     DEBUG0("send finalize to receiver/data");
  52     task = MSG_task_create("finalize", 0.0, 0, NULL);
  53     MSG_task_send(task, get_data_mbox());
  54
  55     xbt_mutex_acquire(mutex);
  56     while (receiver_process) {
  57         DEBUG0("waiting for receiver to terminate");
  58         xbt_cond_wait(cond, mutex);
  59     }
  60     xbt_mutex_release(mutex);
  61
  62     if (ctrl_comm)
  63         WARN0("ctrl_comm is pending!");
  64     if (data_comm)
  65         WARN0("data_comm is pending!");
  66     if (!received.empty())
  67         WARN2("lost %lu received message%s!",
  68               (unsigned long )received.size(), ESSE(received.size()));
  69     if (!sent_comm.empty())
  70         WARN2("lost %lu sent message%s!",
  71               (unsigned long )sent_comm.size(), ESSE(sent_comm.size()));
  72
  73     xbt_cond_destroy(cond);
  74     xbt_mutex_destroy(mutex);
  75 }
  76
  77 void communicator::send(const char* dest, message* msg)
  78 {
  79     DEBUG2("send %s to %s", msg->to_string().c_str(), dest);
  80     double msg_size = sizeof *msg;
  81     if (msg->get_type() == message::LOAD)
  82         msg_size += opt::comm_cost(msg->get_amount());
  83     m_task_t task = MSG_task_create("message", 0.0, msg_size, msg);
  84     msg_comm_t comm = MSG_task_isend(task, dest);
  85     sent_comm.push_back(comm);
  86
  87     if (++send_counter >= send_count_before_flush) {
  88         flush(false);
  89         send_counter = 0;
  90     }
  91 }
  92
  93 bool communicator::recv(message*& msg, m_host_t& from, bool wait)
  94 {
  95     if (wait) {
  96         xbt_mutex_acquire(mutex);
  97         while (received.empty()) {
  98             DEBUG0("waiting for a message to come");
  99             xbt_cond_wait(cond, mutex);
 100         }
 101         xbt_mutex_release(mutex);
 102     }
 103
 104     if (received.empty())
 105         return false;
 106
 107     m_task_t task = received.front();
 108     received.pop();
 109     msg = (message* )MSG_task_get_data(task);
 110     from = MSG_task_get_source(task);
 111     MSG_task_destroy(task);
 112
 113     DEBUG2("received %s from %s",
 114            msg->to_string().c_str(), MSG_host_get_name(from));
 115
 116     return true;
 117 }
 118
 119 void communicator::flush(bool wait)
 120 {
 121     using std::tr1::bind;
 122     using std::tr1::placeholders::_1;
 123
 124     sent_comm.remove_if(comm_test_n_destroy);
 125     if (wait && !sent_comm.empty()) {
 126         xbt_dynar_t comms = xbt_dynar_new(sizeof(msg_comm_t), NULL);
 127         while (!sent_comm.empty()) {
 128             std::for_each(sent_comm.begin(), sent_comm.end(),
 129                           bind(xbt_dynar_push,
 130                                comms, bind(misc::address<msg_comm_t>(), _1)));
 131             MSG_comm_waitany(comms);
 132             xbt_dynar_reset(comms);
 133             sent_comm.remove_if(comm_test_n_destroy);
 134         }
 135         xbt_dynar_free(&comms);
 136     }
 137 }
 138
 139 bool communicator::comm_test_n_destroy(msg_comm_t comm)
 140 {
 141     if (MSG_comm_test(comm)) {
 142         MSG_comm_destroy(comm);
 143         return true;
 144     } else
 145         return false;
 146 }
 147
 148 int communicator::receiver_wrapper(int, char* [])
 149 {
 150     communicator* comm;
 151     comm = (communicator* )MSG_process_get_data(MSG_process_self());
 152     int result = comm->receiver();
 153
 154     DEBUG0("terminate");
 155     xbt_mutex_acquire(comm->mutex);
 156     comm->receiver_process = NULL;
 157     xbt_cond_signal(comm->cond);
 158     xbt_mutex_release(comm->mutex);
 159
 160     return result;
 161 }
 162
 163 int communicator::receiver()
 164 {
 165     ctrl_comm = MSG_task_irecv(&ctrl_task, get_ctrl_mbox());
 166     data_comm = MSG_task_irecv(&data_task, get_data_mbox());
 167     xbt_dynar_t comms = xbt_dynar_new(sizeof(msg_comm_t), NULL);
 168     while (ctrl_comm || data_comm) {
 169
 170         if (ctrl_comm)
 171             xbt_dynar_push(comms, &ctrl_comm);
 172         if (data_comm)
 173             xbt_dynar_push(comms, &data_comm);
 174         MSG_comm_waitany(comms);
 175         xbt_dynar_reset(comms);
 176
 177         if (ctrl_comm && comm_test_n_destroy(ctrl_comm)) {
 178             if (strcmp(MSG_task_get_name(ctrl_task), "finalize")) {
 179                 DEBUG0("received message from ctrl");
 180                 received.push(ctrl_task);
 181                 ctrl_task = NULL;
 182                 ctrl_comm = MSG_task_irecv(&ctrl_task, get_ctrl_mbox());
 183             } else {
 184                 DEBUG0("received finalize from ctrl");
 185                 MSG_task_destroy(ctrl_task);
 186                 ctrl_task = NULL;
 187                 ctrl_comm = NULL;
 188             }
 189         }
 190
 191         if (data_comm && comm_test_n_destroy(data_comm)) {
 192             if (strcmp(MSG_task_get_name(data_task), "finalize")) {
 193                 DEBUG0("received message from data");
 194                 received.push(data_task);
 195                 data_task = NULL;
 196                 data_comm = MSG_task_irecv(&data_task, get_data_mbox());
 197             } else {
 198                 DEBUG0("received finalize from data");
 199                 MSG_task_destroy(data_task);
 200                 data_task = NULL;
 201                 data_comm = NULL;
 202             }
 203         }
 204         xbt_mutex_acquire(mutex);
 205         xbt_cond_signal(cond);
 206         xbt_mutex_release(mutex);
 207     }
 208     xbt_dynar_free(&comms);
 209     return 0;
 210 }
 211
 212 // Local variables:
 213 // mode: c++
 214 // End: