Draft: Revert partially constructed segments on-error in segment init function

ci/conda/environments/clang_env.yml

@@ -1,14 +1,14 @@
 # SPDX-FileCopyrightText: Copyright (c) 2022, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 #
-# Licensed under the Apache License, Version 2.0 (the "License");
+# 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,
+# 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.

cpp/mrc/include/mrc/edge/edge.hpp

@@ -109,6 +109,15 @@ class EdgeBase
         m_disconnectors.clear();
     };
 
+    void log_edges()
+    {
+        LOG(INFO) << "EdgeBase: " << this << " has " << m_linked_edges.size() << " linked edges";
+        for (auto& linked_edge : m_linked_edges)
+        {
+            LOG(INFO) << "  Linked edge: " << linked_edge.get();
+        }
+    }
+
     bool is_connected() const
     {
         return m_is_connected;
@@ -193,6 +202,12 @@ class Edge : public virtual EdgeBase
     friend class EdgeHolder;
     template <typename, typename>
     friend class MultiEdgeHolder;
+
+    void log_edges()
+    {
+        LOG(INFO) << "Edge: " << this;
+        EdgeBase::log_edges();
+    }
 };
 
 class EdgeTypeInfo

cpp/mrc/include/mrc/edge/edge_builder.hpp

@@ -85,6 +85,7 @@ struct EdgeBuilder final
     template <typename SourceT, typename SinkT = SourceT, bool AllowNarrowingV = true>
     static void make_edge_writable(IWritableAcceptor<SourceT>& source, IWritableProvider<SinkT>& sink)
     {
+        LOG(INFO) << "\t\tcreating writable edge";
         constexpr bool IsConvertable = std::is_convertible_v<SourceT, SinkT>;
         constexpr bool LessBits      = sizeof(SourceT) > sizeof(SinkT);  // Sink requires more bits than source.
         constexpr bool FloatToInt    = std::is_floating_point_v<SourceT> && std::is_integral_v<SinkT>;  // float -> int
@@ -126,6 +127,7 @@ struct EdgeBuilder final
             LOG(FATAL) << "No dynamic lookup available for statically typed objects";
         }
 
+        LOG(INFO) << "\t\twritable edge = " << edge.get() << "\t handle = " << edge->get_handle().get();
         source.set_writable_edge_handle(edge);
     }
 
@@ -173,6 +175,7 @@ struct EdgeBuilder final
             LOG(FATAL) << "No dynamic lookup available for statically typed objects";
         }
 
+        LOG(INFO) << "\t\treadable edge = " << edge.get() << "\t handle = " << edge->get_handle().get();
         sink.set_readable_edge_handle(edge);
     }
 
@@ -227,6 +230,8 @@ struct EdgeBuilder final
             auto edge_handle = edge_holder.get_connected_edge();
             edge_holder.release_edge_connection();
 
+            LOG(INFO) << "Splicing edge " << edge_handle.get() << " into " << splice_writable_acceptor;
+
             make_edge_writable(*writable_acceptor, *splice_writable_provider);
             make_edge_writable(*splice_writable_acceptor, sink);
         }
@@ -463,6 +468,7 @@ void make_edge(SourceT& source, SinkT& sink)
     using source_full_t = SourceT;
     using sink_full_t   = SinkT;
 
+    LOG(INFO) << "\tedge_builder::make_edge";
     if constexpr (is_base_of_template<edge::IWritableAcceptor, source_full_t>::value &&
                   is_base_of_template<edge::IWritableProvider, sink_full_t>::value)
     {

cpp/mrc/include/mrc/edge/edge_holder.hpp

@@ -50,15 +50,27 @@ template <typename T>
 class EdgeHolder
 {
   public:
-    EdgeHolder() = default;
+    EdgeHolder()
+    {
+        LOG(INFO) << "EdgeHolder() = " << this;
+    };
+
     virtual ~EdgeHolder()
     {
+        auto p = m_connected_edge.get();
         // Drop any edge connections before this object goes out of scope. This should execute any disconnectors
         m_connected_edge.reset();
 
         if (this->check_active_connection(false))
         {
-            LOG(FATAL) << "A node was destructed which still had dependent connections. Nodes must be kept alive while "
+            LOG(INFO) << "edge = " << p << " edge_holder=" << this;
+            if (p)
+            {
+                p->log_edges();
+            }
+            LOG(FATAL) << this
+                       << " A node was destructed which still had dependent connections. Nodes must be kept alive "
+                          "while "
                           "dependent connections are still active";
         }
     }
@@ -69,6 +81,7 @@ class EdgeHolder
         // Alive connection exists when the lock is true, lifetime is false or a connction object has been set
         if (m_owned_edge.lock() && !m_owned_edge_lifetime)
         {
+            LOG(INFO) << "check_active_connection = " << this << " m_owned_edge.lock() && !m_owned_edge_lifetime";
             // Then someone is using this edge already, cant be changed
             if (do_throw)
             {
@@ -80,6 +93,8 @@ class EdgeHolder
         // Check for set connections. Must be connected to throw error
         if (m_connected_edge && m_connected_edge->is_connected())
         {
+            LOG(INFO) << "check_active_connection = " << this
+                      << " m_connected_edge && m_connected_edge->is_connected()";
             // Then someone is using this edge already, cant be changed
             if (do_throw)
             {
@@ -152,8 +167,10 @@ class EdgeHolder
 
     void release_edge_connection()
     {
+        LOG(INFO) << "Releasing edge connection for edge_holder=" << this;
         m_owned_edge_lifetime.reset();
         m_connected_edge.reset();
+        m_owned_edge.reset();
     }
 
     const std::shared_ptr<Edge<T>>& get_connected_edge() const

cpp/mrc/include/mrc/manifold/composite_manifold.hpp

@@ -59,6 +59,20 @@ class CompositeManifold : public Manifold
         mrc::make_edge(*m_ingress, *m_egress);
     }
 
+    ~CompositeManifold() override
+    {
+        LOG(INFO) << "CompositeManifold::~CompositeManifold(): " << info();
+        shutdown();
+    };
+
+    void shutdown() final
+    {
+        LOG(INFO) << "CompositeManifold::shutdown(): " << info();
+        m_ingress->shutdown();
+        m_egress->shutdown();
+        LOG(INFO) << "CompositeManifold::shutdown(): done";
+    }
+
   protected:
     IngressT& ingress()
     {

cpp/mrc/include/mrc/manifold/egress.hpp

@@ -35,6 +35,7 @@ struct EgressDelegate
 {
     virtual ~EgressDelegate()                                                                        = default;
     virtual void add_output(const SegmentAddress& address, edge::IWritableProviderBase* output_sink) = 0;
+    virtual void shutdown(){};
 };
 
 template <typename T>
@@ -55,6 +56,12 @@ class TypedEgress : public EgressDelegate
 template <typename T>
 class RoundRobinEgress : public node::Router<SegmentAddress, T>, public TypedEgress<T>
 {
+  public:
+    void shutdown() final
+    {
+        node::Router<SegmentAddress, T>::release_edge_connections();
+    }
+
   protected:
     SegmentAddress determine_key_for_value(const T& t) override
     {

cpp/mrc/include/mrc/manifold/ingress.hpp

@@ -31,6 +31,7 @@ struct IngressDelegate
 {
     virtual ~IngressDelegate()                                                                       = default;
     virtual void add_input(const SegmentAddress& address, edge::IWritableAcceptorBase* input_source) = 0;
+    virtual void shutdown(){};
 };
 
 template <typename T>
@@ -51,6 +52,12 @@ class TypedIngress : public IngressDelegate
 template <typename T>
 class MuxedIngress : public node::Muxer<T>, public TypedIngress<T>
 {
+  public:
+    void shutdown() final
+    {
+        node::SourceProperties<T>::release_edge_connection();
+    }
+
   protected:
     void do_add_input(const SegmentAddress& address, edge::IWritableAcceptor<T>* source) final
     {

cpp/mrc/include/mrc/manifold/interface.hpp

@@ -27,9 +27,11 @@ struct Interface
     virtual ~Interface() = default;
 
     virtual const PortName& port_name() const = 0;
+    virtual const std::string& info() const   = 0;
 
-    virtual void start() = 0;
-    virtual void join()  = 0;
+    virtual void start()    = 0;
+    virtual void join()     = 0;
+    virtual void shutdown() = 0;
 
     virtual void add_input(const SegmentAddress& address, edge::IWritableAcceptorBase* input_source) = 0;
     virtual void add_output(const SegmentAddress& address, edge::IWritableProviderBase* output_sink) = 0;

cpp/mrc/include/mrc/manifold/manifold.hpp

@@ -39,12 +39,12 @@ class Manifold : public Interface
     ~Manifold() override;
 
     const PortName& port_name() const final;
+    const std::string& info() const final;
+    void shutdown() override;
 
   protected:
     runnable::IRunnableResources& resources();
 
-    const std::string& info() const;
-
   private:
     void add_input(const SegmentAddress& address, edge::IWritableAcceptorBase* input_source) final;
     void add_output(const SegmentAddress& address, edge::IWritableProviderBase* output_sink) final;

cpp/mrc/include/mrc/node/queue.hpp

@@ -33,7 +33,12 @@ class Queue : public WritableProvider<T>, public ReadableProvider<T>
     {
         this->set_channel(std::make_unique<mrc::channel::BufferedChannel<T>>());
     }
-    ~Queue() override = default;
+
+    ~Queue() override
+    {
+        SinkProperties<T>::release_edge_connection();
+        SourceProperties<T>::release_edge_connection();
+    };
 
     void set_channel(std::unique_ptr<mrc::channel::Channel<T>> channel)
     {

cpp/mrc/include/mrc/node/rx_sink_base.hpp

@@ -45,10 +45,11 @@ class RxSinkBase : public WritableProvider<T>, public ReadableAcceptor<T>, publi
   public:
     void sink_add_watcher(std::shared_ptr<WatcherInterface> watcher);
     void sink_remove_watcher(std::shared_ptr<WatcherInterface> watcher);
+    void release_edge_connection();
 
   protected:
     RxSinkBase();
-    ~RxSinkBase() override = default;
+    ~RxSinkBase() override;
 
     const rxcpp::observable<T>& observable() const;
 
@@ -70,6 +71,21 @@ RxSinkBase<T>::RxSinkBase() :
     this->set_channel(std::make_unique<mrc::channel::BufferedChannel<T>>());
 }
 
+template <typename T>
+RxSinkBase<T>::~RxSinkBase()
+{
+    LOG(INFO) << "~RxSinkBase()";
+    this->release_edge_connection();
+    LOG(INFO) << "~RxSinkBase() - released";
+}
+
+template <typename T>
+void RxSinkBase<T>::release_edge_connection()
+{
+    LOG(ERROR) << "RxSinkBase releasing edge connection";
+    SinkChannelOwner<T>::release_edge_connection();
+}
+
 template <typename T>
 const rxcpp::observable<T>& RxSinkBase<T>::observable() const
 {

cpp/mrc/include/mrc/node/rx_source_base.hpp

@@ -53,10 +53,11 @@ class RxSourceBase : public ReadableProvider<T>,
   public:
     void source_add_watcher(std::shared_ptr<WatcherInterface> watcher);
     void source_remove_watcher(std::shared_ptr<WatcherInterface> watcher);
+    void release_edge_connection();
 
   protected:
     RxSourceBase();
-    ~RxSourceBase() override = default;
+    ~RxSourceBase() override;
 
     const rxcpp::observer<T>& observer() const;
 
@@ -84,6 +85,21 @@ RxSourceBase<T>::RxSourceBase() :
     this->set_channel(std::make_unique<mrc::channel::BufferedChannel<T>>());
 }
 
+template <typename T>
+RxSourceBase<T>::~RxSourceBase()
+{
+    LOG(ERROR) << "~RxSourceBase";
+    this->release_edge_connection();
+    LOG(ERROR) << "~RxSourceBase - done";
+}
+
+template <typename T>
+void RxSourceBase<T>::release_edge_connection()
+{
+    LOG(ERROR) << "RxSourceBase releasing edge connection";
+    SourceChannelOwner<T>::release_edge_connection();
+}
+
 template <typename T>
 const rxcpp::observer<T>& RxSourceBase<T>::observer() const
 {

cpp/mrc/include/mrc/segment/builder.hpp

@@ -79,14 +79,14 @@ namespace {
 namespace hana = boost::hana;
 
 template <typename T>
-auto has_source_add_watcher = hana::is_valid(
-    [](auto&& thing) -> decltype(std::forward<decltype(thing)>(thing).source_add_watcher(
-                         std::declval<std::shared_ptr<mrc::WatcherInterface>>())) {});
+auto has_source_add_watcher =
+    hana::is_valid([](auto&& thing) -> decltype(std::forward<decltype(thing)>(thing).source_add_watcher(
+                                        std::declval<std::shared_ptr<mrc::WatcherInterface>>())) {});
 
 template <typename T>
-auto has_sink_add_watcher = hana::is_valid(
-    [](auto&& thing) -> decltype(std::forward<decltype(thing)>(thing).sink_add_watcher(
-                         std::declval<std::shared_ptr<mrc::WatcherInterface>>())) {});
+auto has_sink_add_watcher =
+    hana::is_valid([](auto&& thing) -> decltype(std::forward<decltype(thing)>(thing).sink_add_watcher(
+                                        std::declval<std::shared_ptr<mrc::WatcherInterface>>())) {});
 
 template <typename T>
 void add_stats_watcher_if_rx_source(T& thing, std::string name)
@@ -464,6 +464,8 @@ void IBuilder::make_edge(SourceObjectT source, SinkObjectT sink)
     auto& source_object = to_object_properties(source);
     auto& sink_object   = to_object_properties(sink);
 
+    LOG(INFO) << "Creating edge from " << source_object.name() << " to " << sink_object.name();
+
     // If we can determine the type from the actual object, use that, then fall back to hints or defaults.
     using deduced_source_type_t = first_non_void_type_t<source_sp_type_t,  // Deduced type (if possible)
                                                         SourceNodeTypeT,   // Explicit type hint
@@ -479,13 +481,18 @@ void IBuilder::make_edge(SourceObjectT source, SinkObjectT sink)
 
     if (source_object.is_writable_acceptor() && sink_object.is_writable_provider())
     {
+        LOG(INFO) << "\tCreating edge from WritableAcceptor to WritableProvider for " << source_object.name() << " to "
+                  << sink_object.name();
         mrc::make_edge(source_object.template writable_acceptor_typed<deduced_source_type_t>(),
                        sink_object.template writable_provider_typed<deduced_sink_type_t>());
+        LOG(INFO) << "\tCreating edge from WritableAcceptor to WritableProvider for " << source_object.name() << " to "
+                  << sink_object.name() << " - done";
         return;
     }
 
     if (source_object.is_readable_provider() && sink_object.is_readable_acceptor())
     {
+        LOG(INFO) << "\tCreating edge from ReadableProvider to ReadableAcceptor";
         mrc::make_edge(source_object.template readable_provider_typed<deduced_source_type_t>(),
                        sink_object.template readable_acceptor_typed<deduced_sink_type_t>());
         return;
@@ -609,6 +616,7 @@ void IBuilder::add_throughput_counter(std::shared_ptr<Object<ObjectT>> segment_o
     CHECK(segment_object->is_source());
     using source_type_t = typename ObjectT::source_type_t;
     auto counter        = this->make_throughput_counter(runnable->name());
+
     runnable->object().add_epilogue_tap([counter](const source_type_t& data) {
         counter(1);
     });

cpp/mrc/include/mrc/segment/component.hpp

@@ -34,6 +34,11 @@ class Component final : public Object<ResourceT>
     Component(std::unique_ptr<ResourceT> resource) : m_resource(std::move(resource)) {}
     ~Component() final = default;
 
+    void destroy() final
+    {
+        m_resource.reset();
+    }
+
   private:
     ResourceT* get_object() const final
     {