Revert partially constructed segments on-error in segment init function

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

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2022-2023, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-FileCopyrightText: Copyright (c) 2022-2024, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -23,7 +23,6 @@
 #include "mrc/channel/egress.hpp"
 #include "mrc/channel/ingress.hpp"
 #include "mrc/edge/forward.hpp"
-#include "mrc/exceptions/runtime_error.hpp"
 #include "mrc/type_traits.hpp"
 #include "mrc/utils/string_utils.hpp"
 
@@ -37,6 +36,7 @@
 #include <memory>
 #include <mutex>
 #include <optional>
+#include <sstream>  // for std::stringstream
 #include <stdexcept>
 #include <type_traits>
 #include <typeindex>
@@ -51,19 +51,45 @@
 {
   public:
     EdgeHolder() = default;
+
     virtual ~EdgeHolder()
     {
         // 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 "
-                          "dependent connections are still active";
+            std::stringstream msg;
+            msg << "EdgeHolder(" << this << ") "
+                << "A node was destructed which still had dependent connections. Nodes must be kept alive while "
+                   "dependent connections are still active\n"
+                << this->connection_info();
+
+#if defined(NDEBUG)
+            LOG(ERROR) << msg.str();
+#else
+            LOG(FATAL) << msg.str();
+#endif
         }
     }
 
   protected:
+    std::string connection_info() const
+    {
+        std::stringstream ss;
+        ss << "m_owned_edge=" << m_owned_edge.lock() << "\tm_owned_edge_lifetime=" << m_owned_edge_lifetime
+           << "\tm_connected_edge=" << m_connected_edge;
+
+        bool is_connected = false;
+        if (m_connected_edge)
+        {
+            is_connected = m_connected_edge->is_connected();
+        }
+
+        ss << "\tis_connected=" << is_connected << "\tcheck_active_connection=" << this->check_active_connection(false);
+        return ss.str();
+    }
+
     bool check_active_connection(bool do_throw = true) const
     {
         // Alive connection exists when the lock is true, lifetime is false or a connction object has been set
@@ -155,6 +181,13 @@
         m_connected_edge.reset();
     }
 
+    void disconnect()
+    {
+        m_connected_edge.reset();
+        m_owned_edge_lifetime.reset();
+        m_owned_edge.reset();
+    }
+
     const std::shared_ptr<Edge<T>>& get_connected_edge() const
     {
         return m_connected_edge;
@@ -204,6 +237,17 @@
     virtual ~MultiEdgeHolder() = default;
 
   protected:
+    std::string connection_info() const
+    {
+        std::stringstream ss;
+        ss << "m_edges.size()=" << m_edges.size();
+        for (const auto& [key, edge_pair] : m_edges)
+        {
+            ss << "\n\tkey=" << key << "\t" << edge_pair.connection_info();
+        }
+        return ss.str();
+    }
+
     void init_owned_edge(KeyT key, std::shared_ptr<Edge<T>> edge)
     {
         auto& edge_pair = this->get_edge_pair(key, true);

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

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2021-2023, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-FileCopyrightText: Copyright (c) 2021-2024, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -59,6 +59,17 @@ class CompositeManifold : public Manifold
         mrc::make_edge(*m_ingress, *m_egress);
     }
 
+    ~CompositeManifold() override
+    {
+        shutdown();
+    };
+
+    void shutdown() final
+    {
+        m_ingress->shutdown();
+        m_egress->shutdown();
+    }
+
   protected:
     IngressT& ingress()
     {

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

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2021-2023, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-FileCopyrightText: Copyright (c) 2021-2024, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -35,6 +35,7 @@
 {
     virtual ~EgressDelegate()                                                                        = default;
     virtual void add_output(const SegmentAddress& address, edge::IWritableProviderBase* output_sink) = 0;
+    virtual void shutdown(){};
 };
 
 template <typename T>
@@ -55,6 +56,13 @@
 template <typename T>
 class RoundRobinEgress : public node::Router<SegmentAddress, T>, public TypedEgress<T>
 {
+  public:
+    void shutdown() final
+    {
+        DVLOG(10) << "Releasing edges from manifold egress";
+        node::Router<SegmentAddress, T>::release_edge_connections();
+    }
+
   protected:
     SegmentAddress determine_key_for_value(const T& t) override
     {

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

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2021-2023, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-FileCopyrightText: Copyright (c) 2021-2024, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -23,6 +23,8 @@
 #include "mrc/node/sink_properties.hpp"
 #include "mrc/node/source_properties.hpp"
 
+#include <glog/logging.h>
+
 #include <memory>
 
 namespace mrc::manifold {
@@ -31,6 +33,7 @@
 {
     virtual ~IngressDelegate()                                                                       = default;
     virtual void add_input(const SegmentAddress& address, edge::IWritableAcceptorBase* input_source) = 0;
+    virtual void shutdown(){};
 };
 
 template <typename T>
@@ -51,6 +54,13 @@
 template <typename T>
 class MuxedIngress : public node::Muxer<T>, public TypedIngress<T>
 {
+  public:
+    void shutdown() final
+    {
+        DVLOG(10) << "Releasing edges from manifold ingress";
+        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

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2021-2023, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-FileCopyrightText: Copyright (c) 2021-2024, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -28,8 +28,9 @@ struct Interface
 
     virtual const PortName& port_name() 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

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2021-2023, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-FileCopyrightText: Copyright (c) 2021-2024, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -39,6 +39,7 @@ class Manifold : public Interface
     ~Manifold() override;
 
     const PortName& port_name() const final;
+    void shutdown() override;
 
   protected:
     runnable::IRunnableResources& resources();

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

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2021-2023, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-FileCopyrightText: Copyright (c) 2021-2024, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -33,7 +33,11 @@ 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_node.hpp

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2021-2023, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-FileCopyrightText: Copyright (c) 2021-2024, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -68,13 +68,14 @@ class RxNode : public RxSinkBase<InputT>,
 
     void make_stream(stream_fn_t fn);
 
+    void on_shutdown_critical_section() final;
+
   private:
     // the following method(s) are moved to private from their original scopes to prevent access from deriving classes
     using RxSinkBase<InputT>::observable;
     using RxSourceBase<OutputT>::observer;
 
     void do_subscribe(rxcpp::composite_subscription& subscription) final;
-    void on_shutdown_critical_section() final;
 
     void on_stop(const rxcpp::subscription& subscription) override;
     void on_kill(const rxcpp::subscription& subscription) final;

cpp/mrc/include/mrc/node/source_properties.hpp

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2021-2023, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-FileCopyrightText: Copyright (c) 2021-2024, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -23,6 +23,7 @@
 #include "mrc/edge/edge_writable.hpp"
 #include "mrc/node/forward.hpp"
 #include "mrc/type_traits.hpp"
+#include "mrc/types.hpp"  // for Mutex
 #include "mrc/utils/type_utils.hpp"
 
 #include <memory>
@@ -208,37 +209,68 @@
 class ForwardingEgressProvider : public ReadableProvider<T>
 {
   protected:
+    struct State
+    {
+        Mutex m_mutex;
+        bool m_is_destroyed{false};
+    };
+
     class ForwardingEdge : public edge::IEdgeReadable<T>
     {
       public:
-        ForwardingEdge(ForwardingEgressProvider<T>& parent) : m_parent(parent) {}
+        ForwardingEdge(ForwardingEgressProvider<T>& parent, std::shared_ptr<State> state) :
+          m_parent(parent),
+          m_state(std::move(state))
+        {}
 
         ~ForwardingEdge() = default;
 
         channel::Status await_read(T& t) override
         {
-            return m_parent.get_next(t);
+            std::lock_guard<decltype(m_state->m_mutex)> lock(m_state->m_mutex);
+            if (!(m_state->m_is_destroyed))
+            {
+                return m_parent.get_next(t);
+            }
+
+            return channel::Status::closed;
         }
 
       private:
         ForwardingEgressProvider<T>& m_parent;
+        std::shared_ptr<State> m_state;
     };
 
-    ForwardingEgressProvider()
+    ForwardingEgressProvider() : m_state(std::make_shared<State>())
     {
-        auto inner_edge = std::make_shared<ForwardingEdge>(*this);
+        auto inner_edge = std::make_shared<ForwardingEdge>(*this, m_state);
 
-        inner_edge->add_disconnector([this]() {
-            // Only call the on_complete if we have been connected
-            this->on_complete();
+        inner_edge->add_disconnector([this, state = m_state]() {
+            std::lock_guard<decltype(state->m_mutex)> lock(state->m_mutex);
+            if (!(state->m_is_destroyed))
+            {
+                // Only call the on_complete if we have been connected and `this` is still alive
+                this->on_complete();
+            }
         });
 
         ReadableProvider<T>::init_owned_edge(inner_edge);
     }
 
+    ~ForwardingEgressProvider()
+    {
+        SourceProperties<T>::disconnect();
+        {
+            std::lock_guard<decltype(m_state->m_mutex)> lock(m_state->m_mutex);
+            m_state->m_is_destroyed = true;
+        }
+    }
+
     virtual channel::Status get_next(T& t) = 0;
 
     virtual void on_complete() {}
+
+    std::shared_ptr<State> m_state;
 };
 
 }  // namespace mrc::node

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

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2021-2023, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-FileCopyrightText: Copyright (c) 2021-2024, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
  * SPDX-License-Identifier: Apache-2.0
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -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
     {