Define class for message queue.

diff --git a/Makefile b/Makefile
index a4ccf170f78059dc61db066acae0dfc79099e334..3990aa5b0d3db0ed986843f3bb05eddea3d9e11d 100644 (file)
--- a/Makefile
+++ b/Makefile
@@ -34,6 +34,7 @@ SRC.loba := main.cpp          \
        deployment.cpp          \
        hostdata.cpp            \
        $(wildcard loba_*.cpp)  \
        deployment.cpp          \
        hostdata.cpp            \
        $(wildcard loba_*.cpp)  \

+       messages.cpp            \

        misc.cpp                \
        neighbor.cpp            \
        options.cpp             \
        misc.cpp                \
        neighbor.cpp            \
        options.cpp             \

diff --git a/communicator.cpp b/communicator.cpp
index 506f126ce14b37975b4e61dfee62302a1b89f4be..3516b68146ee8844a3c8ec8c33fad1b3c19befc6 100644 (file)
--- a/communicator.cpp
+++ b/communicator.cpp
@@ -1,7 +1,4 @@
 #include <algorithm>
 #include <algorithm>

-#include <cstring>
-#include <tr1/functional>
-#include <sstream>

 #include <msg/msg.h>
 #include <xbt/log.h>
 
 #include <msg/msg.h>
 #include <xbt/log.h>
 


@@ -14,15 +11,6 @@ XBT_LOG_EXTERNAL_DEFAULT_CATEGORY(comm);
 
 #include "communicator.h"
 
 
 #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())))
 {
 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);
 }
 
     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);
 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;
 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);
         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);
             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);
             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 b51f743d8aedb4ed5ceffcc39c6a84c1db854893..7cb1efcb0883374084f9aeb872b5a2fad1bd96b6 100644 (file)
--- a/communicator.h
+++ b/communicator.h
@@ -4,28 +4,11 @@
 #define COMMUNICATOR_H
 
 #include <list>
 #define COMMUNICATOR_H
 
 #include <list>

-#include <queue>
-#include <string>

 #include <msg/msg.h>
 #include "hostdata.h"
 #include <msg/msg.h>
 #include "hostdata.h"

+#include "messages.h"

 #include "synchro.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();
 class communicator {
 public:
     communicator();


@@ -34,15 +17,15 @@ public:
     // Send a message to the "dest" mailbox
     void send(const char* dest, message* msg);
 
     // 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 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;
 private:
     // Myself
     const hostdata* host;


@@ -51,7 +34,7 @@ private:
     std::list<msg_comm_t> sent_comm;
 
     // Queue of received messages
     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;
 
     // Handling of receiving thread
     mutex_t receiver_mutex;

diff --git a/messages.cpp b/messages.cpp
new file mode 100644 (file)
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 (file)
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: