Request/response tests excluded and README typos corrected

Signed-off-by: zesk1999 <zesk1999@gmail.com>

Author: zesk1999

README.md

@@ -44,8 +44,8 @@ AI developers from all experience levels can make use of the framework through i
 
 ### Project Architecture
 
-The *SustainML Framework* is composed of different Software Modules, each one related to specific task, which are specialized in solving the different parts of the machine learning problem architecture definition, starting from the user’s problem description.
-Each of the modules conforms a Node.
+The *SustainML Framework* is composed of different Software Modules, each one related to a specific task, which are specialized in solving the different parts of the machine learning problem architecture definition, starting from the user’s problem description.
+Each of the modules forms a Node.
 These steps are basically:
 
 1. Encode the problem and constraints defined by the user
@@ -64,7 +64,7 @@ The exchanged information between the modules is over DDS.
 ### Framework
 
 The Framework includes a *Graphical User Interface* (GUI) in which user interacts and introduces the ML problem definition.
-That GUI implements also the **Orchestrator** node, a key node that feds the remain modules with the information provided by the user, retrieves all the results, and display them to the user though this GUI.
+That GUI implements also the **Orchestrator** node, a key node that feeds the remaining modules with the information provided by the user, retrieves all the results, and display them to the user through this GUI.
 
 This process can be iterative.
 So, based on a previous solution and the user's feedback, the framework provides new ML solutions.
@@ -77,9 +77,9 @@ This repository is divided in sub-packages with different targets:
 
 * `sustainml_cpp`: Main definition and implementation of the project logic library. C++ API provided.
 * `sustainml_docs`: ReadTheDocs documentation project ([available here](https://sustainml.readthedocs.io/en/latest/))
-* `sustainml_modules`: Set of piped modules that use the Python API.
-* `sustainml_py`: Wrap of the project logic library. Python API.
-* `sustainml_swig`: Binding from the C++ API `sustainml_cpp` to the `sustainml_py` Python API.
+* `sustainml_modules`: Set of pipeline modules that use the Python API.
+* `sustainml_py`: Wrapper of the project logic library. Python API.
+* `sustainml_swig`: Bindings from the C++ API `sustainml_cpp` to the `sustainml_py` Python API.
 
 ## Getting Help
 

sustainml_cpp/src/cpp/core/GenericServiceNodeImpl.hpp

@@ -56,6 +56,11 @@ class GenericServiceNodeImpl : public ServerBase
             const eprosima::fastdds::dds::rpc::RpcRequest& /*info*/,
             const std::string& configuration) override
     {
+        if (node_.shutting_down())
+        {
+            throw ::InternalError("update_configuration: node is shutting down");
+        }
+
         std::cout << "[RPC SERVER/" << tag_ << "] update_configuration cfg='"
                   << configuration << "'\n";
 

sustainml_cpp/src/cpp/core/NodeImpl.cpp

@@ -118,24 +118,26 @@ NodeImpl::NodeImpl(
 
 NodeImpl::~NodeImpl()
 {
+    shutting_down_.store(true, std::memory_order_release);
+
     EPROSIMA_LOG_INFO(NODE, "Destroying Node");
 
-    // Stop RPC server before destroying the participant
     if (rpc_server_)
     {
         rpc_server_->stop();
-    }
 
-    if (rpc_server_thread_.joinable())
-    {
-        rpc_server_thread_.join();
+        if (rpc_server_thread_.joinable())
+        {
+            rpc_server_thread_.join();
+        }
+        rpc_server_.reset();
     }
 
-    if (nullptr != participant_)
+    if (participant_)
     {
         participant_->delete_contained_entities();
-        auto dpf = DomainParticipantFactory::get_instance();
-        dpf->delete_participant(participant_);
+        DomainParticipantFactory::get_instance()->delete_participant(participant_);
+        participant_ = nullptr;
     }
 
     dispatcher_->stop();

sustainml_cpp/src/cpp/core/NodeImpl.hpp

@@ -127,6 +127,11 @@ class NodeImpl
         return req_res_listener_;
     }
 
+    bool shutting_down() const noexcept
+    {
+        return shutting_down_.load(std::memory_order_acquire);
+    }
+
 protected:
 
     /**
@@ -253,6 +258,8 @@ class NodeImpl
 
     }
     control_listener_;
+
+    std::atomic<bool> shutting_down_{false};
 };
 
 } // namespace core

sustainml_cpp/src/cpp/orchestrator/OrchestratorNode.cpp

@@ -62,7 +62,8 @@ template <typename ClientT>
 bool rpc_update_configuration(
         ClientT& client,
         const std::string& configuration,
-        std::string& out_cfg)
+        std::string& out_cfg,
+        const std::atomic<bool>& terminate_flag)
 {
     auto future = client.update_configuration(configuration);
 
@@ -73,6 +74,12 @@ bool rpc_update_configuration(
 
     while (true)
     {
+        if (terminate_flag.load(std::memory_order_acquire))
+        {
+            std::cerr << "[INFO] RPC aborted due to shutdown\n";
+            return false;
+        }
+
         // Wait a bit, but don’t block forever
         auto status = future.wait_for(step);
 
@@ -148,7 +155,7 @@ void OrchestratorNode::OrchestratorParticipantListener::on_participant_discovery
                 });
     }
 
-    // create the proxy for this node
+    // Create the proxy for this node
     NodeID node_id = common::get_node_id_from_name(participant_name);
 
     std::cout << "[DEBUG Orchestrator] Participant discovered: name='"
@@ -159,8 +166,9 @@ void OrchestratorNode::OrchestratorParticipantListener::on_participant_discovery
 
     std::lock_guard<std::mutex> lock(orchestrator_->proxies_mtx_);
 
-    // check if the node has been terminated
-    if (!orchestrator_->terminated_.load())
+    // Check if the node has been terminated
+    if (!orchestrator_->terminate_.load(std::memory_order_acquire) &&
+            !orchestrator_->terminated_.load(std::memory_order_acquire))
     {
         if (reason == eprosima::fastdds::rtps::ParticipantDiscoveryStatus::DISCOVERED_PARTICIPANT &&
                 orchestrator_->node_proxies_[static_cast<uint32_t>(node_id)] == nullptr)
@@ -223,6 +231,8 @@ OrchestratorNode::~OrchestratorNode()
 
 void OrchestratorNode::destroy()
 {
+    terminate_.store(true, std::memory_order_release);
+
     if (!terminated_.load())
     {
         {
@@ -379,7 +389,7 @@ bool OrchestratorNode::init()
 
     holder->app_requirements_client = create_AppRequirementsServiceClient(
         *participant_,
-        "AppRequirementsService",       // must match server side
+        "AppRequirementsService",       // Must match server side
         rqos);
     std::cout << "[DEBUG Orchestrator] Created AppRequirementsServiceClient service='AppRequirementsService' ok="
               << (holder->app_requirements_client ? "true" : "false")
@@ -481,7 +491,7 @@ std::pair<types::TaskId, types::UserInput*> OrchestratorNode::prepare_new_iterat
         std::lock_guard<std::mutex> lock(task_db_->get_mutex());
         task_db_->prepare_new_entry_nts(new_task_id, true);
         // Copy the UserInput from the previous iteration
-        // it also updates the iteration_id in the data
+        // It also updates the iteration_id in the data
         task_db_->copy_data_nts(old_task_id, new_task_id, {NodeID::ID_ORCHESTRATOR});
         task_db_->get_task_data_nts(new_task_id, output.second);
     }
@@ -715,7 +725,7 @@ types::ResponseType OrchestratorNode::configuration_request (
             {
                 std::cout << "[RPC CLIENT] calling AppRequirementsService.update_configuration tx="
                           << req.transaction_id() << std::endl;
-                if (!rpc_update_configuration(*holder->app_requirements_client, req.configuration(), cfg))
+                if (!rpc_update_configuration(*holder->app_requirements_client, req.configuration(), cfg, terminate_))
                 {
                     return res; // Timeout or not ready
                 }
@@ -725,7 +735,7 @@ types::ResponseType OrchestratorNode::configuration_request (
             {
                 std::cout << "[RPC CLIENT] calling HWConstraintsService.update_configuration tx="
                           << req.transaction_id() << std::endl;
-                if (!rpc_update_configuration(*holder->hw_constraints_client, req.configuration(), cfg))
+                if (!rpc_update_configuration(*holder->hw_constraints_client, req.configuration(), cfg, terminate_))
                 {
                     return res;
                 }
@@ -736,7 +746,7 @@ types::ResponseType OrchestratorNode::configuration_request (
                 std::cout << "[RPC CLIENT] calling HWResourcesService.update_configuration tx="
                           << req.transaction_id() << std::endl;
 
-                if (!rpc_update_configuration(*holder->hw_resources_client, req.configuration(), cfg))
+                if (!rpc_update_configuration(*holder->hw_resources_client, req.configuration(), cfg, terminate_))
                 {
                     return res;
                 }
@@ -746,7 +756,7 @@ types::ResponseType OrchestratorNode::configuration_request (
             {
                 std::cout << "[RPC CLIENT] calling CarbonFootprintService.update_configuration tx="
                           << req.transaction_id() << std::endl;
-                if (!rpc_update_configuration(*holder->carbon_footprint_client, req.configuration(), cfg))
+                if (!rpc_update_configuration(*holder->carbon_footprint_client, req.configuration(), cfg, terminate_))
                 {
                     return res;
                 }
@@ -756,7 +766,7 @@ types::ResponseType OrchestratorNode::configuration_request (
             {
                 std::cout << "[RPC CLIENT] calling MLModelMetadataService.update_configuration tx="
                           << req.transaction_id() << std::endl;
-                if (!rpc_update_configuration(*holder->ml_model_metadata_client, req.configuration(), cfg))
+                if (!rpc_update_configuration(*holder->ml_model_metadata_client, req.configuration(), cfg, terminate_))
                 {
                     return res;
                 }
@@ -766,7 +776,7 @@ types::ResponseType OrchestratorNode::configuration_request (
             {
                 std::cout << "[RPC CLIENT] calling MLModelService.update_configuration tx="
                           << req.transaction_id() << std::endl;
-                if (!rpc_update_configuration(*holder->ml_model_client, req.configuration(), cfg))
+                if (!rpc_update_configuration(*holder->ml_model_client, req.configuration(), cfg, terminate_))
                 {
                     return res;
                 }

sustainml_cpp/test/blackbox/CMakeLists.txt

@@ -14,6 +14,7 @@
 
 
 file(GLOB BLACKBOXTESTS_TEST_SOURCE "common/BlackboxTests*.cpp")
+list(FILTER BLACKBOXTESTS_TEST_SOURCE EXCLUDE REGEX ".*ResponseNodes.*\\.cpp$") # Old behavior tests
 set(BLACKBOXTESTS_SOURCE ${BLACKBOXTESTS_TEST_SOURCE}
     ${PROJECT_SOURCE_DIR}/src/cpp/types/typesImplTypeObjectSupport.cxx
     ${PROJECT_SOURCE_DIR}/src/cpp/types/typesImplPubSubTypes.cxx