Refactor disposbles (#483)

src/rpp/rpp/disposables/composite_disposable.hpp

@@ -27,14 +27,11 @@ namespace rpp
  *
  * @ingroup disposables
  */
-template<rpp::constraint::decayed_type Container>
+template<details::disposables::constraint::disposable_container Container>
 class composite_disposable_impl : public interface_composite_disposable
 {
 public:
-    composite_disposable_impl()
-        requires details::disposables::constraint::disposable_container<Container>
-    = default;
-
+    composite_disposable_impl()= default;
     composite_disposable_impl(const composite_disposable_impl&)           = delete;
     composite_disposable_impl(composite_disposable_impl&& other) noexcept = delete;
 
@@ -99,6 +96,33 @@ class composite_disposable_impl : public interface_composite_disposable
         }
     }
 
+    void remove(const disposable_wrapper& disposable) override
+    {
+        while (true)
+        {
+            State expected{State::None};
+            // need to acquire possible disposables state changing from other `add` or `remove`
+            if (m_current_state.compare_exchange_strong(expected, State::Edit, std::memory_order::acquire, std::memory_order::relaxed))
+            {
+                try
+                {
+                    m_disposables.remove(disposable);
+                }
+                catch(...)
+                {
+                    m_current_state.store(State::None, std::memory_order::release);
+                    throw;
+                }
+                // need to propogate disposables state changing to others
+                m_current_state.store(State::None, std::memory_order::release);
+                return;
+            }
+
+            if (expected == State::Disposed)
+                return;
+        }
+    }
+
 protected:
     virtual void dispose_impl() noexcept {}
 
@@ -113,4 +137,6 @@ class composite_disposable_impl : public interface_composite_disposable
     Container          m_disposables{};
     std::atomic<State> m_current_state{};
 };
+
+class composite_disposable : public composite_disposable_impl<rpp::details::disposables::dynamic_disposables_container<0>>{};
 } // namespace rpp

src/rpp/rpp/disposables/details/container.hpp

@@ -13,6 +13,7 @@
 #include <rpp/utils/exceptions.hpp>
 
 #include <array>
+#include <algorithm>
 #include <vector>
 
 namespace rpp::details::disposables
@@ -35,6 +36,11 @@ class dynamic_disposables_container_base
         m_data.push_back(std::move(d));
     }
 
+    void remove(const rpp::disposable_wrapper& d)
+    {
+        m_data.erase(std::remove(m_data.begin(), m_data.end(), d), m_data.end());
+    }
+
     void dispose() const
     {
         for (auto& d : m_data) {
@@ -80,6 +86,15 @@ class static_disposables_container
         m_data[m_size++] = std::move(d);
     }
 
+    void remove(const rpp::disposable_wrapper& d)
+    {
+        auto itr = std::remove(m_data.begin(), m_data.end(), d);
+        while(itr != m_data.end()) {
+            (*itr++) = disposable_wrapper{};
+            --m_size;
+        }
+    }
+
     void dispose() const
     {
         for (size_t i =0; i < m_size; ++i) {
@@ -105,7 +120,8 @@ struct none_disposables_container
         throw rpp::utils::more_disposables_than_expected{"none_disposables_container expected none disposables but obtained one"};
     }
 
-    static void dispose() {};
-    static void clear() {};
+    static void remove(const rpp::disposable_wrapper&) {}
+    static void dispose() {}
+    static void clear() {}
 };
 }

src/rpp/rpp/disposables/disposable_wrapper.hpp

@@ -66,6 +66,11 @@ class disposable_wrapper_impl
         return disposable_wrapper_impl{weak_tag{}, std::move(disposable)};
     }
 
+    bool operator==(const disposable_wrapper_impl& other) const
+    {
+        return get_original() == other.get_original();
+    }
+
     bool is_disposed() const noexcept
     {
         if (const auto locked = get_original())
@@ -88,6 +93,13 @@ class disposable_wrapper_impl
             other.dispose();
     }
 
+    void remove(const disposable_wrapper& other) const
+        requires std::derived_from<TDisposable, interface_composite_disposable>
+    {
+        if (const auto locked = get_original())
+            locked->remove(other);
+    }
+
     std::shared_ptr<TDisposable> get_original() const noexcept
     {
         if (const auto ptr_ptr = std::get_if<std::shared_ptr<TDisposable>>(&m_disposable))

src/rpp/rpp/disposables/fwd.hpp

@@ -50,10 +50,7 @@ namespace constraint
 
 namespace rpp
 {
-template<rpp::constraint::decayed_type Container>
-class composite_disposable_impl;
-
-using composite_disposable = composite_disposable_impl<rpp::details::disposables::dynamic_disposables_container<0>>;
+class composite_disposable;
 
 template<rpp::constraint::is_nothrow_invocable Fn>
 class callback_disposable;

src/rpp/rpp/disposables/interface_composite_disposable.hpp

@@ -23,9 +23,13 @@ struct interface_composite_disposable : public interface_disposable
     virtual void add(disposable_wrapper disposable) = 0;
 
     template<rpp::constraint::is_nothrow_invocable Fn>
-    void add(Fn&& invocable)
+    disposable_wrapper add(Fn&& invocable)
     {
-        add(make_callback_disposable(std::forward<Fn>(invocable)));
+        auto d = make_callback_disposable(std::forward<Fn>(invocable));
+        add(d);
+        return d;
     }
+
+    virtual void remove(const disposable_wrapper& d) = 0;
 };
 }

src/rpp/rpp/disposables/refcount_disposable.hpp

@@ -18,110 +18,76 @@
 
 #include <atomic>
 #include <memory>
+#include <limits>
 
 namespace rpp::details
 {
-struct refocunt_disposable_state_t final : public composite_disposable 
+class refocunt_disposable_state_t final : public rpp::composite_disposable 
 {
-    refocunt_disposable_state_t() = default;
-    refocunt_disposable_state_t(const refocunt_disposable_state_t&) = delete;
-    refocunt_disposable_state_t(refocunt_disposable_state_t&&) noexcept = delete;
-
-private:
-    using composite_disposable::dispose;
-
-    void dispose_impl() noexcept override
-    {
-        m_refcount.store(0, std::memory_order::relaxed);
-    }
-
 public:
-    void try_dispose()
+    void dispose_impl() noexcept override
     {
-        // just need atomicity, not guarding anything
-        if (m_refcount.fetch_sub(1, std::memory_order::relaxed) == 1)
-            dispose();
+        m_refcount.store(s_disposed, std::memory_order::relaxed);
     }
 
-    bool is_disposed_by_any() const noexcept
+    void release()
     {
-        return m_refcount.load(std::memory_order::relaxed) == 0 || is_disposed();
+        auto current_value = m_refcount.load(std::memory_order::relaxed);
+        while (true)
+        {
+            if (current_value == s_disposed)
+                return;
+
+            const size_t new_value = current_value == 1 ? s_disposed : current_value - 1;
+            if (!m_refcount.compare_exchange_strong(current_value, new_value, std::memory_order::relaxed, std::memory_order::relaxed))
+                continue;
+
+            if (new_value == s_disposed)
+                dispose();
+            return;
+        }
     }
 
     bool add_ref() 
     {
         auto current_value = m_refcount.load(std::memory_order::relaxed);
         // just need atomicity, not guarding anything
-        while (current_value && !m_refcount.compare_exchange_strong(current_value, current_value + 1, std::memory_order::relaxed, std::memory_order::relaxed)){};
-
-        return current_value;
-    }
-
-private:
-    std::atomic_size_t m_refcount{1};
-};
-
-class inner_refcount_disposable final : public interface_composite_disposable
-{
-public:
-    explicit inner_refcount_disposable(const std::shared_ptr<refocunt_disposable_state_t>& state)
-        : m_state{state}
-    {
-    }
-
-    inner_refcount_disposable(const inner_refcount_disposable&)     = delete;
-    inner_refcount_disposable(inner_refcount_disposable&&) noexcept = delete;
-
-    void dispose() noexcept override
-    {
-        if (m_disposed.exchange(true, std::memory_order::relaxed) == false)
-            m_state->try_dispose();
-    }
+        while (current_value != s_disposed && !m_refcount.compare_exchange_strong(current_value, current_value + 1, std::memory_order::relaxed, std::memory_order::relaxed)){};
 
-    using interface_composite_disposable::add;
-
-    void add(disposable_wrapper disposable) override
-    {
-        m_state->add(std::move(disposable));
+        return current_value != s_disposed;
     }
 
-    bool is_disposed() const noexcept override
-    {
-        return m_disposed.load(std::memory_order::relaxed) || m_state->is_disposed_by_any();
-    } 
-
 private:
-    std::shared_ptr<refocunt_disposable_state_t> m_state;
-    std::atomic_bool                     m_disposed{};
+    std::atomic<size_t>     m_refcount{0};
+    constexpr static size_t s_disposed = std::numeric_limits<size_t>::max();
 };
 }
 
 namespace rpp
 {
-/**
- * @brief Disposable with counter inside. Each `add_ref` increments counter, while each `dispose()` call decrements. In case of reaching zero disposes underlying disposables
- * @warning Don't use it as disposable of observer due to `is_disposed()` would be false till counter reaches zero, so, observer can be also not `is_disposed()` during this time.
- *
- * @ingroup disposables
- */
-class refcount_disposable : public interface_composite_disposable,
-                            public std::enable_shared_from_this<refcount_disposable> {
+class refcount_disposable : public std::enable_shared_from_this<refcount_disposable> {
 
 public:
     refcount_disposable() = default;
 
-    refcount_disposable(const refcount_disposable&)     = delete;
-    refcount_disposable(refcount_disposable&&) noexcept = delete;
-
     bool is_disposed_underlying() const noexcept
     {
-        return m_state.is_disposed_by_any();
+        return m_state.is_disposed();
     }
 
     composite_disposable_wrapper add_ref()
     {
         if (m_state.add_ref())
-            return composite_disposable_wrapper{std::make_shared<details::inner_refcount_disposable>(std::shared_ptr<details::refocunt_disposable_state_t>{this->shared_from_this(), &this->m_state})};
+        {
+            auto inner   = std::make_shared<rpp::composite_disposable_impl<rpp::details::disposables::dynamic_disposables_container<1>>>();
+            auto as_weak = rpp::disposable_wrapper::from_weak(inner);
+            m_state.add(as_weak);
+            inner->add([s = shared_from_this(), as_weak]() noexcept {
+                s->m_state.remove(as_weak);
+                s->m_state.release();
+            });
+            return composite_disposable_wrapper{inner};
+        }
 
         return composite_disposable_wrapper{};
     }
@@ -131,26 +97,7 @@ class refcount_disposable : public interface_composite_disposable,
         return composite_disposable_wrapper::from_shared(std::shared_ptr<rpp::composite_disposable>{this->shared_from_this(), &this->m_state});
     }
 
-    bool is_disposed() const noexcept final
-    {
-        return m_disposed.load(std::memory_order::relaxed) || m_state.is_disposed_by_any();
-    } 
-
-    void dispose() noexcept final
-    {
-        if (m_disposed.exchange(true, std::memory_order::relaxed) == false)
-            m_state.try_dispose();
-    }
-
-    using interface_composite_disposable::add;
-
-    void add(disposable_wrapper disposable) final
-    {
-        m_state.add(std::move(disposable));
-    }
-
 private:
     details::refocunt_disposable_state_t m_state{};
-    std::atomic_bool                     m_disposed{};
 };
 } // namespace rpp

src/rpp/rpp/observables/connectable_observable.hpp

@@ -56,7 +56,7 @@ struct ref_count_on_subscribe_t<rpp::connectable_observable<OriginalObservable,
             return {m_state->disposable->add_ref(), std::nullopt};
 
         m_state->disposable                        = std::make_shared<rpp::refcount_disposable>();
-        return {m_state->disposable, m_state->disposable->get_underlying()};
+        return {m_state->disposable->add_ref(), m_state->disposable->get_underlying()};
     }
 };
 }

src/rpp/rpp/operators/details/forwarding_subject.hpp

@@ -44,7 +44,7 @@ class forwarding_strategy
     explicit forwarding_strategy(std::shared_ptr<rpp::refcount_disposable> refcount)
         : m_refcount{std::move(refcount)}
     {
-        m_refcount->add(rpp::disposable_wrapper::from_weak(m_state));
+        m_refcount->get_underlying().add(rpp::disposable_wrapper::from_weak(m_state));
     }
 
     auto get_observer() const

src/rpp/rpp/operators/group_by.hpp

@@ -77,12 +77,12 @@ struct group_by_observer_strategy
 
     RPP_CALL_DURING_CONSTRUCTION(
     {
-        observer.set_upstream(rpp::disposable_wrapper{disposable});
+        observer.set_upstream(disposable->add_ref());
     });
 
     void set_upstream(const rpp::disposable_wrapper& d) const
     {
-        disposable->add(d);
+        disposable->get_underlying().add(d);
     }
 
     bool is_disposed() const
@@ -134,7 +134,7 @@ struct group_by_observer_strategy
 
         if (inserted)
         {
-            disposable->add(rpp::disposable_wrapper::from_weak(itr->second.get_disposable().get_original()));
+            disposable->get_underlying().add(rpp::disposable_wrapper::from_weak(itr->second.get_disposable().get_original()));
             obs.on_next(rpp::grouped_observable_group_by<TKey, Type>{
                 std::move(key),
                 group_by_observable_strategy<Type>{itr->second, disposable}
@@ -158,9 +158,10 @@ struct group_by_observable_strategy
     {
         if (const auto locked = disposable.lock())
         {
-            obs.set_upstream(locked->add_ref());
+            auto d = locked->add_ref();
+            obs.set_upstream(d);
             subj.get_observable()
-                .subscribe(rpp::observer<T, group_by_inner_observer_strategy<observer<T, Strategy>>>{std::move(obs), rpp::composite_disposable_wrapper::from_weak(disposable)});
+                .subscribe(rpp::observer<T, group_by_inner_observer_strategy<observer<T, Strategy>>>{std::move(obs), std::move(d)});
         }
     }
 };