communicator.cpp

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