From 892e36c1af3105e5f548a4ef8b9f777b95d47fc2 Mon Sep 17 00:00:00 2001
From: Arnaud Giersch <arnaud.giersch@iut-bm.univ-fcomte.fr>
Date: Wed, 9 Feb 2011 10:19:27 +0100
Subject: [PATCH] Define class for message queue.
---
Makefile | 1 +
communicator.cpp | 65 ++++++++----------------------------------------
communicator.h | 29 +++++----------------
messages.cpp | 55 ++++++++++++++++++++++++++++++++++++++++
messages.h | 48 +++++++++++++++++++++++++++++++++++
5 files changed, 120 insertions(+), 78 deletions(-)
create mode 100644 messages.cpp
create mode 100644 messages.h
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 <algorithm>
-#include <cstring>
-#include <tr1/functional>
-#include <sstream>
#include <msg/msg.h>
#include <xbt/log.h>
@@ -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<hostdata*>(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<message*>(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 <list>
-#include <queue>
-#include <string>
#include <msg/msg.h>
#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<msg_comm_t> sent_comm;
// Queue of received messages
- std::queue<m_task_t> 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 <sstream>
+
+#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<message*>(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 <queue>
+#include <string>
+#include <msg/msg.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 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<m_task_t> queue;
+};
+
+#endif // !MESSAGES_H
+
+// Local variables:
+// mode: c++
+// End:
--
2.39.5