X-Git-Url: https://bilbo.iut-bm.univ-fcomte.fr/and/gitweb/loba.git/blobdiff_plain/1c3da8681b394ce3b14fb44e27cb682c045d5f92..20ee10ed13aee7d9fb9c844648d5159ed4a054d2:/messages.cpp diff --git a/messages.cpp b/messages.cpp index 025871a..70e2236 100644 --- a/messages.cpp +++ b/messages.cpp @@ -8,25 +8,50 @@ XBT_LOG_EXTERNAL_DEFAULT_CATEGORY(comm); #include "messages.h" -std::string message::to_string() +message::message(message_type t, double a, double c) + : type(t), amount(a) , credit(c) { - static const char* str[] = { "INFO", "CREDIT", "LOAD", - "CTRL_CLOSE", "DATA_CLOSE" }; - std::ostringstream oss; - oss << str[type] << ": " << amount; - return oss.str(); + // compute message size + // arbitrary: 8 for type, and 8 for each double + switch (type) { + case CTRL: + size = opt::bookkeeping ? 24 : 16; // type + amount + (credit)? + break; + case DATA: + size = 16 + opt::comm_cost(amount); // type + amount + data size + break; + default: + size = 8; // type + break; + } } -double message::get_size() const +std::string message::to_string() { - // arbitrary: 8 for type, and 8 for amount - double size = 16; - if (type == LOAD) - size += opt::comm_cost(amount); - return size; + std::ostringstream oss; + switch (type) { + case CTRL: + oss << "CTRL: " << amount << " (info)"; + if (opt::bookkeeping) + oss << "; " << credit << " (credit)"; + break; + case DATA: + oss << "DATA: " << amount << " (load)"; + break; + case CTRL_CLOSE: + oss << "CTRL_CLOSE"; + break; + case DATA_CLOSE: + oss << "DATA_CLOSE"; + break; + default: + oss << "UNKNOWN MESSAGE TYPE: " << type; + break; + } + return oss.str(); } -void message_queue::push(m_task_t task) +void message_queue::push(msg_task_t task) { if (queue.push(task)) { // list was empty, the push must be signaled @@ -36,38 +61,31 @@ void message_queue::push(m_task_t task) } } -bool message_queue::pop(message*& msg, m_host_t& from, double timeout) +bool message_queue::pop(message*& msg, msg_host_t& from, double timeout) { - m_task_t task; + msg_task_t task = nullptr; if (!queue.try_pop(task)) { if (timeout == 0.0) return false; mutex.acquire(); if (!queue.try_pop(task)) { - xbt_ex_t e; XBT_DEBUG("waiting for a message to come"); - TRY { - if (timeout > 0) - cond.timedwait(mutex, timeout); - else - cond.wait(mutex); - } - TRY_CLEANUP { - mutex.release(); - } - CATCH (e) { - if (e.category != timeout_error) - RETHROW; - xbt_ex_free(e); - return false; // got a timeout + bool hit_timeout; + if (timeout > 0) { + hit_timeout = !cond.timedwait(mutex, timeout); + } else { + cond.wait(mutex); + hit_timeout = false; } bool pop_was_successful = queue.try_pop(task); - xbt_assert(pop_was_successful); - } else { - mutex.release(); + xbt_assert(hit_timeout || pop_was_successful); } + mutex.release(); } + if (task == nullptr) + return false; + msg = static_cast(MSG_task_get_data(task)); from = MSG_task_get_source(task); MSG_task_destroy(task);