From: Arnaud Giersch Date: Wed, 9 Feb 2011 09:19:27 +0000 (+0100) Subject: Define class for message queue. X-Git-Tag: v0.1~156 X-Git-Url: https://bilbo.iut-bm.univ-fcomte.fr/and/gitweb/loba.git/commitdiff_plain/892e36c1af3105e5f548a4ef8b9f777b95d47fc2 Define class for message queue. --- diff --git a/Makefile b/Makefile index a4ccf17..3990aa5 100644 --- a/Makefile +++ b/Makefile @@ -34,6 +34,7 @@ SRC.loba := main.cpp \ deployment.cpp \ hostdata.cpp \ $(wildcard loba_*.cpp) \ + messages.cpp \ misc.cpp \ neighbor.cpp \ options.cpp \ diff --git a/communicator.cpp b/communicator.cpp index 506f126..3516b68 100644 --- a/communicator.cpp +++ b/communicator.cpp @@ -1,7 +1,4 @@ #include -#include -#include -#include #include #include @@ -14,15 +11,6 @@ XBT_LOG_EXTERNAL_DEFAULT_CATEGORY(comm); #include "communicator.h" -std::string message::to_string() -{ - static const char* str[] = { "INFO", "CREDIT", "LOAD", - "CTRL_CLOSE", "DATA_CLOSE" }; - std::ostringstream oss; - oss << str[type] << ": " << amount; - return oss.str(); -} - communicator::communicator() : host(static_cast(MSG_host_get_data(MSG_host_self()))) { @@ -75,46 +63,6 @@ void communicator::send(const char* dest, message* msg) sent_comm.push_back(comm); } -bool communicator::recv(message*& msg, m_host_t& from, double timeout) -{ - if (timeout != 0) { - volatile double deadline = - timeout > 0 ? MSG_get_clock() + timeout : 0.0; - receiver_mutex.acquire(); - while (received.empty() && (!deadline || deadline > MSG_get_clock())) { - xbt_ex_t e; - XBT_DEBUG("waiting for a message to come"); - TRY { - if (deadline) - receiver_cond.timedwait(receiver_mutex, - deadline - MSG_get_clock()); - else - receiver_cond.wait(receiver_mutex); - } - CATCH (e) { - if (e.category != timeout_error) - RETHROW; - xbt_ex_free(e); - } - } - receiver_mutex.release(); - } - - if (received.empty()) - return false; - - m_task_t task = received.front(); - received.pop(); - msg = static_cast(MSG_task_get_data(task)); - from = MSG_task_get_source(task); - MSG_task_destroy(task); - - XBT_DEBUG("received %s from %s", - msg->to_string().c_str(), MSG_host_get_name(from)); - - return true; -} - void communicator::flush(bool wait) { sent_comm.remove_if(comm_test_n_destroy); @@ -129,6 +77,16 @@ void communicator::flush(bool wait) } } +bool communicator::recv(message*& msg, m_host_t& from, double timeout) +{ + XBT_DEBUG("waiting for a message to come"); + bool recvd = received.pop(msg, from, timeout); + if (recvd) + XBT_DEBUG("received %s from %s", + msg->to_string().c_str(), MSG_host_get_name(from)); + return recvd; +} + int communicator::receiver_wrapper(int, char* []) { communicator* comm; @@ -178,10 +136,7 @@ void communicator::receiver() comm_check_n_destroy(ch->comm); if (strcmp(MSG_task_get_name(ch->task), "finalize")) { XBT_DEBUG("received message on %s", ch->mbox); - receiver_mutex.acquire(); received.push(ch->task); - receiver_cond.signal(); - receiver_mutex.release(); ch->task = NULL; ch->comm = MSG_task_irecv(&ch->task, ch->mbox); xbt_dynar_set_as(comms, index, msg_comm_t, ch->comm); diff --git a/communicator.h b/communicator.h index b51f743..7cb1efc 100644 --- a/communicator.h +++ b/communicator.h @@ -4,28 +4,11 @@ #define COMMUNICATOR_H #include -#include -#include #include #include "hostdata.h" +#include "messages.h" #include "synchro.h" -class message { -public: - enum message_type { INFO, CREDIT, LOAD, CTRL_CLOSE, DATA_CLOSE }; - - message(message_type t, double a): type(t), amount(a) { } - - message_type get_type() const { return type; } - double get_amount() const { return amount; } - - std::string to_string(); - -private: - message_type type; - double amount; -}; - class communicator { public: communicator(); @@ -34,15 +17,15 @@ public: // Send a message to the "dest" mailbox void send(const char* dest, message* msg); + // Try to flush pending sending communications. + // If "wait" is true, blocks until success. + void flush(bool wait); + // Try to get a message. Returns true on success. // Parameter "timeout" may be 0 for non-blocking operation, -1 for // infinite waiting, or any positive timeout. bool recv(message*& msg, m_host_t& from, double timeout); - // Try to flush pending sending communications. - // If "wait" is true, blocks until success. - void flush(bool wait); - private: // Myself const hostdata* host; @@ -51,7 +34,7 @@ private: std::list sent_comm; // Queue of received messages - std::queue received; + message_queue received; // Handling of receiving thread mutex_t receiver_mutex; diff --git a/messages.cpp b/messages.cpp new file mode 100644 index 0000000..2fa4861 --- /dev/null +++ b/messages.cpp @@ -0,0 +1,55 @@ +#include + +#include "messages.h" + +std::string message::to_string() +{ + static const char* str[] = { "INFO", "CREDIT", "LOAD", + "CTRL_CLOSE", "DATA_CLOSE" }; + std::ostringstream oss; + oss << str[type] << ": " << amount; + return oss.str(); +} + +void message_queue::push(m_task_t task) +{ + mutex.acquire(); + queue.push(task); + cond.signal(); + mutex.release(); +} + +bool message_queue::pop(message*& msg, m_host_t& from, double timeout) +{ + if (timeout != 0) { + volatile double deadline = + timeout > 0 ? MSG_get_clock() + timeout : 0.0; + mutex.acquire(); + while (queue.empty() && (!deadline || deadline > MSG_get_clock())) { + xbt_ex_t e; + TRY { + if (deadline) + cond.timedwait(mutex, deadline - MSG_get_clock()); + else + cond.wait(mutex); + } + CATCH (e) { + if (e.category != timeout_error) + RETHROW; + xbt_ex_free(e); + } + } + mutex.release(); + } + + if (queue.empty()) + return false; + + m_task_t task = queue.front(); + queue.pop(); + msg = static_cast(MSG_task_get_data(task)); + from = MSG_task_get_source(task); + MSG_task_destroy(task); + + return true; +} diff --git a/messages.h b/messages.h new file mode 100644 index 0000000..34e9248 --- /dev/null +++ b/messages.h @@ -0,0 +1,48 @@ +#ifndef MESSAGES_H +#define MESSAGES_H + +#include +#include +#include +#include "synchro.h" + +class message { +public: + enum message_type { INFO, CREDIT, LOAD, CTRL_CLOSE, DATA_CLOSE }; + + message(message_type t, double a): type(t), amount(a) { } + + message_type get_type() const { return type; } + double get_amount() const { return amount; } + + std::string to_string(); + +private: + message_type type; + double amount; +}; + +class message_queue { +public: + // Push a message on queue + void push(m_task_t task); + + // Try to get a message. Returns true on success. + // Parameter "timeout" may be 0 for non-blocking operation, -1 for + // infinite waiting, or any positive timeout. + bool pop(message*& msg, m_host_t& from, double timeout); + + bool empty() const { return queue.empty(); } + size_t size() const { return queue.size(); } + +private: + mutex_t mutex; + condition_t cond; + std::queue queue; +}; + +#endif // !MESSAGES_H + +// Local variables: +// mode: c++ +// End: