WaitSet detail classes (#1989)

* Refs 11608. Required changes on Condition API.

Signed-off-by: Miguel Company <MiguelCompany@eprosima.com>

* Refs 11608. ConditionNotifier with empty implementation.

Signed-off-by: Miguel Company <MiguelCompany@eprosima.com>

* Refs 11608. WaitSetImpl with empty implementation.

Signed-off-by: Miguel Company <MiguelCompany@eprosima.com>

* Refs 11608. ConditionNotifier unit test.

Signed-off-by: Miguel Company <MiguelCompany@eprosima.com>

* Refs 11608. WaitSetImpl unit test.

Signed-off-by: Miguel Company <MiguelCompany@eprosima.com>

* Refs 11608. Added unordered_vector.

Signed-off-by: Miguel Company <MiguelCompany@eprosima.com>

* Refs 11608. ConditionNotifier implementation.

Signed-off-by: Miguel Company <MiguelCompany@eprosima.com>

* Refs 11608. Condition management on WaitSetImpl.

Signed-off-by: Miguel Company <MiguelCompany@eprosima.com>

* Refs 11608. Implenting wait and wake_up on WaitSetImpl.

Signed-off-by: Miguel Company <MiguelCompany@eprosima.com>

* Refs 11656. Linters.

Signed-off-by: Miguel Company <MiguelCompany@eprosima.com>

* Refs 11656. Solving link issues.

Signed-off-by: Miguel Company <MiguelCompany@eprosima.com>

* Refs 11656. Doxygen improvements.

Signed-off-by: Miguel Company <MiguelCompany@eprosima.com>

* Refs 11656. Check vector size on test.

Signed-off-by: Miguel Company <MiguelCompany@eprosima.com>

include/fastdds/dds/core/condition/Condition.hpp

@@ -41,15 +41,15 @@ class Condition
      * @brief Retrieves the trigger_value of the Condition
      * @return true if trigger_value is set to 'true', 'false' otherwise
      */
-    RTPS_DllAPI bool get_trigger_value() const
+    RTPS_DllAPI virtual bool get_trigger_value() const
     {
         logWarning(CONDITION, "get_trigger_value public member function not implemented");
         return false; // TODO return trigger value
     }
 
 };
 
-typedef std::vector<Condition> ConditionSeq;
+using ConditionSeq = std::vector<Condition*>;
 
 } // namespace dds
 } // namespace fastdds

include/fastdds/dds/core/condition/GuardCondition.hpp

@@ -59,8 +59,6 @@ class GuardCondition : public Condition
 
 };
 
-typedef std::vector<Condition> ConditionSeq;
-
 } // namespace dds
 } // namespace fastdds
 } // namespace eprosima

src/cpp/CMakeLists.txt

@@ -158,8 +158,10 @@ set(${PROJECT_NAME}_source_files
     dynamic-types/DynamicDataHelper.cpp
 
     fastrtps_deprecated/attributes/TopicAttributes.cpp
+    fastdds/core/condition/ConditionNotifier.cpp
     fastdds/core/condition/StatusCondition.cpp
     fastdds/core/condition/WaitSet.cpp
+    fastdds/core/condition/WaitSetImpl.cpp
     fastdds/core/policy/ParameterList.cpp
     fastdds/core/policy/QosPolicyUtils.cpp
     fastdds/publisher/qos/WriterQos.cpp

src/cpp/fastdds/core/condition/ConditionNotifier.cpp

@@ -0,0 +1,75 @@
+// Copyright 2021 Proyectos y Sistemas de Mantenimiento SL (eProsima).
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+/**
+ * @file ConditionNotifier.cpp
+ */
+
+#include "ConditionNotifier.hpp"
+
+#include <mutex>
+
+#include <fastdds/dds/core/condition/Condition.hpp>
+
+#include <fastdds/core/condition/WaitSetImpl.hpp>
+
+namespace eprosima {
+namespace fastdds {
+namespace dds {
+namespace detail {
+
+void ConditionNotifier::attach_to (
+        WaitSetImpl* wait_set)
+{
+    if (nullptr != wait_set)
+    {
+        std::lock_guard<std::mutex> guard(mutex_);
+        entries_.remove(wait_set);
+        entries_.emplace_back(wait_set);
+    }
+}
+
+void ConditionNotifier::detach_from (
+        WaitSetImpl* wait_set)
+{
+    if (nullptr != wait_set)
+    {
+        std::lock_guard<std::mutex> guard(mutex_);
+        entries_.remove(wait_set);
+    }
+}
+
+void ConditionNotifier::notify ()
+{
+    std::lock_guard<std::mutex> guard(mutex_);
+    for (WaitSetImpl* wait_set : entries_)
+    {
+        wait_set->wake_up();
+    }
+}
+
+void ConditionNotifier::will_be_deleted (
+        const Condition& condition)
+{
+    std::lock_guard<std::mutex> guard(mutex_);
+    for (WaitSetImpl* wait_set : entries_)
+    {
+        wait_set->will_be_deleted(condition);
+    }
+}
+
+}  // namespace detail
+}  // namespace dds
+}  // namespace fastdds
+}  // namespace eprosima

src/cpp/fastdds/core/condition/ConditionNotifier.hpp

@@ -0,0 +1,78 @@
+// Copyright 2021 Proyectos y Sistemas de Mantenimiento SL (eProsima).
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+/**
+ * @file ConditionNotifier.hpp
+ */
+
+#ifndef _FASTDDS_CORE_CONDITION_CONDITIONNOTIFIER_HPP_
+#define _FASTDDS_CORE_CONDITION_CONDITIONNOTIFIER_HPP_
+
+#include <mutex>
+
+#include <fastdds/dds/core/condition/Condition.hpp>
+
+#include <utils/collections/unordered_vector.hpp>
+
+namespace eprosima {
+namespace fastdds {
+namespace dds {
+namespace detail {
+
+struct WaitSetImpl;
+
+struct ConditionNotifier
+{
+    /**
+     * Add a WaitSet implementation to the list of attached entries.
+     * Does nothing if wait_set was already attached to this notifier.
+     * @param wait_set WaitSet implementation to add to the list.
+     */
+    void attach_to (
+            WaitSetImpl* wait_set);
+
+
+    /**
+     * Remove a WaitSet implementation from the list of attached entries.
+     * Does nothing if wait_set was not attached to this notifier.
+     * @param wait_set WaitSet implementation to remove from the list.
+     */
+    void detach_from (
+            WaitSetImpl* wait_set);
+
+    /**
+     * Wake up all the WaitSet implementations attached to this notifier.
+     */
+    void notify ();
+
+    /**
+     * Inform all the WaitSet implementations attached to this notifier that
+     * a condition is going to be deleted.
+     * @param condition The Condition being deleted.
+     */
+    void will_be_deleted (
+            const Condition& condition);
+
+private:
+
+    std::mutex mutex_;
+    eprosima::utilities::collections::unordered_vector<WaitSetImpl*> entries_;
+};
+
+}  // namespace detail
+}  // namespace dds
+}  // namespace fastdds
+}  // namespace eprosima
+
+#endif // _FASTDDS_CORE_CONDITION_CONDITIONNOTIFIER_HPP_

src/cpp/fastdds/core/condition/WaitSetImpl.cpp

@@ -0,0 +1,130 @@
+// Copyright 2021 Proyectos y Sistemas de Mantenimiento SL (eProsima).
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+/**
+ * @file WaitSetImpl.cpp
+ */
+
+#include "WaitSetImpl.hpp"
+
+#include <condition_variable>
+#include <mutex>
+
+#include <fastdds/dds/core/condition/Condition.hpp>
+#include <fastdds/rtps/common/Time_t.h>
+#include <fastrtps/types/TypesBase.h>
+
+using eprosima::fastrtps::types::ReturnCode_t;
+
+namespace eprosima {
+namespace fastdds {
+namespace dds {
+namespace detail {
+
+ReturnCode_t WaitSetImpl::attach_condition(
+        const Condition& condition)
+{
+    std::lock_guard<std::mutex> guard(mutex_);
+    bool was_there = entries_.remove(&condition);
+    entries_.emplace_back(&condition);
+
+    // Should wake_up when adding a new triggered condition
+    if (is_waiting_ && !was_there && condition.get_trigger_value())
+    {
+        wake_up();
+    }
+
+    return ReturnCode_t::RETCODE_OK;
+}
+
+ReturnCode_t WaitSetImpl::detach_condition(
+        const Condition& condition)
+{
+    std::lock_guard<std::mutex> guard(mutex_);
+    bool was_there = entries_.remove(&condition);
+    return was_there ? ReturnCode_t::RETCODE_OK : ReturnCode_t::RETCODE_PRECONDITION_NOT_MET;
+}
+
+ReturnCode_t WaitSetImpl::wait(
+        ConditionSeq& active_conditions,
+        const fastrtps::Duration_t& timeout)
+{
+    std::unique_lock<std::mutex> lock(mutex_);
+
+    if (is_waiting_)
+    {
+        return ReturnCode_t::RETCODE_PRECONDITION_NOT_MET;
+    }
+
+    auto fill_active_conditions = [&]()
+            {
+                bool ret_val = entries_.empty();
+                active_conditions.clear();
+                for (const Condition* c : entries_)
+                {
+                    if (c->get_trigger_value())
+                    {
+                        ret_val = true;
+                        active_conditions.push_back(const_cast<Condition*>(c));
+                    }
+                }
+                return ret_val;
+            };
+
+    bool condition_value = false;
+    is_waiting_ = true;
+    if (fastrtps::c_TimeInfinite == timeout)
+    {
+        cond_.wait(lock, fill_active_conditions);
+        condition_value = true;
+    }
+    else
+    {
+        auto ns = timeout.to_ns();
+        condition_value = cond_.wait_for(lock, std::chrono::nanoseconds(ns), fill_active_conditions);
+    }
+    is_waiting_ = false;
+
+    return condition_value ? ReturnCode_t::RETCODE_OK : ReturnCode_t::RETCODE_TIMEOUT;
+}
+
+ReturnCode_t WaitSetImpl::get_conditions(
+        ConditionSeq& attached_conditions) const
+{
+    std::lock_guard<std::mutex> guard(mutex_);
+    attached_conditions.reserve(entries_.size());
+    attached_conditions.clear();
+    for (const Condition* c : entries_)
+    {
+        attached_conditions.push_back(const_cast<Condition*>(c));
+    }
+    return ReturnCode_t::RETCODE_OK;
+}
+
+void WaitSetImpl::wake_up()
+{
+    cond_.notify_one();
+}
+
+void WaitSetImpl::will_be_deleted (
+        const Condition& condition)
+{
+    std::lock_guard<std::mutex> guard(mutex_);
+    entries_.remove(&condition);
+}
+
+}  // namespace detail
+}  // namespace dds
+}  // namespace fastdds
+}  // namespace eprosima