Implemted more of the MQ stuff

Author: henricasanova

include/wrench/simgrid_S4U_util/S4U_PendingCommunication.h

@@ -19,8 +19,6 @@
 
 namespace wrench {
 
-    //    class SimulationMessage;
-
     /*******************/
     /** \cond INTERNAL */
     /*******************/
@@ -57,15 +55,17 @@ namespace wrench {
 
         //        ~S4U_PendingCommunication() default;
 
-        /** @brief The SimGrid communication handle */
-        simgrid::s4u::CommPtr comm_ptr;
-
         /** @brief The message */
         std::unique_ptr<SimulationMessage> simulation_message;
         /** @brief The CommPort */
         S4U_CommPort *commport;
         /** @brief The operation type */
         OperationType operation_type;
+
+        /** @brief The SimGrid Mailbox communication handle */
+        simgrid::s4u::CommPtr comm_ptr;
+        /** @brief The SimGrid MessageQueue communication handle */
+        simgrid::s4u::MessPtr mess_ptr;
     };
 
     /*******************/

src/wrench/simgrid_S4U_util/S4U_CommPort.cpp

@@ -143,8 +143,6 @@ namespace wrench {
         } else if (finished_recv == mq_comm) {
             mb_comm->cancel();
             mq_comm->wait();
-        } else {
-            std::cerr << "WTF\n";
         }
 
 #ifdef MESSAGE_MANAGER
@@ -266,24 +264,24 @@ namespace wrench {
             throw std::invalid_argument("S4U_CommPort::igetMessage(): Cannot be called with NULL_COMMPORT");
         }
 
-        simgrid::s4u::CommPtr comm_ptr;
-
         WRENCH_DEBUG("Igetting a message from commport '%s'", this->s4u_mb->get_cname());
 
         std::shared_ptr<S4U_PendingCommunication> pending_communication = std::make_shared<S4U_PendingCommunication>(
                 this, S4U_PendingCommunication::OperationType::RECEIVING);
 
+
         try {
-            comm_ptr = this->s4u_mb->get_async<void>((void **) (&(pending_communication->simulation_message)));
+            auto comm_ptr = this->s4u_mb->get_async<void>((void **) (&(pending_communication->simulation_message)));
+            pending_communication->comm_ptr = comm_ptr;
         } catch (simgrid::NetworkFailureException &e) {
             throw ExecutionException(std::make_shared<NetworkError>(
                     NetworkError::RECEIVING, NetworkError::FAILURE, this->s4u_mb->get_name(), ""));
         }
-        pending_communication->comm_ptr = comm_ptr;
+        simgrid::s4u::MessPtr mess_ptr = this->s4u_mq->get_async<void>((void **) (&(pending_communication->simulation_message)));
+        pending_communication->mess_ptr = mess_ptr;
         return pending_communication;
     }
 
-
     /**
     * @brief Generate a unique sequence number
     *

src/wrench/simgrid_S4U_util/S4U_PendingCommunication.cpp

@@ -10,6 +10,8 @@
 #include <memory>
 #include <iostream>
 #include <wrench/failure_causes/NetworkError.h>
+#include "wrench/failure_causes/FatalFailure.h"
+
 
 #ifdef MESSAGE_MANAGER
 #include <wrench/util/MessageManager.h>
@@ -27,7 +29,6 @@ WRENCH_LOG_CATEGORY(wrench_core_pending_communication, "Log category for Pending
 
 namespace wrench {
 
-
     /**
      * @brief Wait for the pending communication to complete
      *
@@ -48,7 +49,10 @@ namespace wrench {
      * @throw std::shared_ptr<NetworkError>
      */
     std::unique_ptr<SimulationMessage> S4U_PendingCommunication::wait(double timeout) {
+
+#if 0
         try {
+            // IS THIS NECESSARY?
             if (this->comm_ptr->get_state() != simgrid::s4u::Activity::State::FINISHED) {
                 this->comm_ptr->wait_for(timeout);
             }
@@ -73,6 +77,53 @@ namespace wrench {
         MessageManager::removeReceivedMessage(this->commport, this->simulation_message.get());
 #endif
         return std::move(this->simulation_message);
+#endif
+
+
+//        if (log) WRENCH_DEBUG("Getting a message from commport '%s' with timeout %lf sec", this->comm_ptr->get_cname(), timeout);
+        SimulationMessage *msg;
+
+        simgrid::s4u::ActivitySet pending_receives;
+        pending_receives.push(this->comm_ptr);
+        pending_receives.push(this->mess_ptr);
+
+        simgrid::s4u::ActivityPtr finished_recv;
+        try {
+            // Wait for one activity to complete
+            finished_recv = pending_receives.wait_any_for(timeout);
+        } catch (simgrid::TimeoutException &e) {
+            mess_ptr->cancel();
+            comm_ptr->cancel();
+            throw ExecutionException(std::make_shared<NetworkError>(NetworkError::RECEIVING, NetworkError::TIMEOUT, this->commport->get_name(), ""));
+        } catch (simgrid::Exception &e) {
+            auto failed_recv = pending_receives.get_failed_activity();
+            if (failed_recv == comm_ptr) {
+                mess_ptr->cancel();
+                throw ExecutionException(std::make_shared<NetworkError>(
+                        NetworkError::RECEIVING, NetworkError::FAILURE, this->comm_ptr->get_name(), ""));
+            } else {
+                comm_ptr->cancel();
+                throw ExecutionException(std::make_shared<wrench::FatalFailure>("A communication on a MQ should never fail"));
+            }
+        }
+
+        WRENCH_DEBUG("Got the message\n");
+
+        if (finished_recv == comm_ptr) {
+            mess_ptr->cancel();
+            comm_ptr->wait();
+        } else if (finished_recv == mess_ptr) {
+            comm_ptr->cancel();
+            mess_ptr->wait();
+        }
+
+#ifdef MESSAGE_MANAGER
+        MessageManager::removeReceivedMessage(this, msg);
+#endif
+
+        WRENCH_DEBUG("Received a '%s' message from commport '%s'", msg->getName().c_str(), this->commport->get_cname());
+
+        return std::unique_ptr<SimulationMessage>(msg);
     }
 
     /**

test/services/compute_services/bare_metal_compound_jobs/BareMetalComputeServiceOneActionTests.cpp

@@ -317,10 +317,10 @@ void BareMetalComputeServiceOneActionTest::do_Noop_test() {
     // Create and initialize a simulation
     auto simulation = wrench::Simulation::createSimulation();
 
-    int argc = 1;
+    int argc = 2;
     auto argv = (char **) calloc(argc, sizeof(char *));
     argv[0] = strdup("one_action_test");
-//            argv[1] = strdup("--wrench-full-log");
+            argv[1] = strdup("--wrench-full-log");
 
     ASSERT_NO_THROW(simulation->init(&argc, argv));