#include <algorithm>
-#include <tr1/functional>
-#include <msg/msg.h>
+#include <simgrid/msg.h>
#include <xbt/log.h>
XBT_LOG_EXTERNAL_DEFAULT_CATEGORY(comm);
#include "misc.h"
-#include "options.h"
-#include "simgrid_features.h"
#include "tracing.h"
#include "communicator.h"
communicator::communicator()
: host(static_cast<hostdata*>(MSG_host_get_data(MSG_host_self())))
{
- using std::tr1::bind;
- receiver_thread = new_msg_thread("receiver",
- bind(&communicator::receiver, this));
+ receiver_thread = new_msg_thread("receiver", [this]() { this->receiver(); });
receiver_thread->start();
}
communicator::~communicator()
{
- m_task_t task;
+ msg_task_t task;
XBT_DEBUG("send finalize to receiver/ctrl");
- task = MSG_task_create("finalize", 0.0, 0, NULL);
+ task = MSG_task_create("finalize", 0.0, 0, nullptr);
MSG_task_send(task, host->get_ctrl_mbox());
XBT_DEBUG("send finalize to receiver/data");
- task = MSG_task_create("finalize", 0.0, 0, NULL);
+ task = MSG_task_create("finalize", 0.0, 0, nullptr);
MSG_task_send(task, host->get_data_mbox());
receiver_thread->wait();
msg_comm_t communicator::real_send(const char* dest, message* msg)
{
XBT_DEBUG("send %s to %s", msg->to_string().c_str(), dest);
- double msg_size = sizeof *msg;
- if (msg->get_type() == message::LOAD)
- msg_size += opt::comm_cost(msg->get_amount());
- m_task_t task = MSG_task_create("message", 0.0, msg_size, msg);
- TRACE_msg_set_task_category(task,
- msg->get_type() == message::LOAD ?
- TRACE_CAT_DATA : TRACE_CAT_CTRL);
+ msg_task_t task = MSG_task_create("message", 0.0, msg->get_size(), msg);
+ // MSG_task_set_category(task,
+ // msg->get_type() == message::DATA ?
+ // TRACE_CAT_DATA : TRACE_CAT_CTRL);
return MSG_task_isend(task, dest);
}
void communicator::real_flush(sent_comm_type& sent_comm, bool wait)
{
- sent_comm_type::iterator bound;
- bound = std::remove_if(sent_comm.begin(), sent_comm.end(),
- comm_test_n_destroy);
+ sent_comm_type::iterator bound =
+ std::remove_if(sent_comm.begin(), sent_comm.end(),
+ comm_test_n_destroy);
sent_comm.erase(bound, sent_comm.end());
if (wait && !sent_comm.empty()) {
size_t size = sent_comm.size();
std::copy(sent_comm.begin(), sent_comm.end(), comms);
sent_comm.clear();
MSG_comm_waitall(comms, size, -1.0);
- if (!MSG_WAIT_DESTROYS_COMMS)
- std::for_each(comms, comms + size, comm_check_n_destroy);
+ std::for_each(comms, comms + size, comm_check_n_destroy);
delete[] comms;
}
}
void communicator::receiver()
{
- xbt_dynar_t comms = xbt_dynar_new(sizeof(msg_comm_t), NULL);
+ xbt_dynar_t comms = xbt_dynar_new(sizeof(msg_comm_t), nullptr);
struct channel {
msg_comm_t comm;
- m_task_t task;
+ msg_task_t task;
const char* mbox;
message_queue& received;
};
- channel chan[] = { { NULL, NULL, host->get_ctrl_mbox(), ctrl_received },
- { NULL, NULL, host->get_data_mbox(), data_received } };
+ channel chan[] = { { nullptr, nullptr, host->get_ctrl_mbox(), ctrl_received },
+ { nullptr, nullptr, host->get_data_mbox(), data_received } };
const int chan_size = (sizeof chan) / (sizeof chan[0]);
for (int i = 0 ; i < chan_size ; ++i) {
if (strcmp(MSG_task_get_name(ch->task), "finalize")) {
XBT_DEBUG("received message on %s", ch->mbox);
ch->received.push(ch->task);
- ch->task = NULL;
+ ch->task = nullptr;
ch->comm = MSG_task_irecv(&ch->task, ch->mbox);
xbt_dynar_set_as(comms, index, msg_comm_t, ch->comm);
} else {
XBT_DEBUG("received finalize on %s", ch->mbox);
MSG_task_destroy(ch->task);
- ch->task = NULL;
- ch->comm = NULL;
- xbt_dynar_remove_at(comms, index, NULL);
+ ch->task = nullptr;
+ ch->comm = nullptr;
+ xbt_dynar_remove_at(comms, index, nullptr);
}
}