Revert partially constructed segments on-error in segment init function

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

@@ -285,8 +285,11 @@ class DeferredWritableMultiEdge : public MultiEdgeHolder<std::size_t, T>,
                                   public DeferredWritableMultiEdgeBase
 {
   public:
-    DeferredWritableMultiEdge(determine_indices_fn_t indices_fn = nullptr, bool deep_copy = false) :
-      m_indices_fn(std::move(indices_fn))
+    DeferredWritableMultiEdge(determine_indices_fn_t indices_fn = nullptr,
+                              bool deep_copy                    = false,
+                              std::string name                  = std::string()) :
+      m_indices_fn(std::move(indices_fn)),
+      MultiEdgeHolder<std::size_t, T>(std::move(name))
     {
         // // Generate warning if deep_copy = True but type does not support it
         // if constexpr (!std::is_copy_constructible_v<T>)

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

@@ -37,7 +37,9 @@
 #include <memory>
 #include <mutex>
 #include <optional>
+#include <sstream>
 #include <stdexcept>
+#include <string>
 #include <type_traits>
 #include <typeindex>
 #include <utility>
@@ -50,20 +52,44 @@ template <typename T>
 class EdgeHolder
 {
   public:
-    EdgeHolder() = default;
+    EdgeHolder(std::string name = std::string()) : m_name(std::move(name)){};
+
     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";
+            LOG(FATAL) << "EdgeHolder(" << this << ")[" << m_name << "] "
+                       << "A node was destructed which still had dependent connections. Nodes must be kept alive while "
+                          "dependent connections are still active\n"
+                       << this->connection_info();
         }
     }
 
+    const std::string& name() const
+    {
+        return m_name;
+    };
+
   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 @@ class EdgeHolder
         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;
@@ -188,6 +221,8 @@ class EdgeHolder
     // Holds a pointer to any set edge (different from init edge). Maintains lifetime
     std::shared_ptr<Edge<T>> m_connected_edge;
 
+    std::string m_name;
+
     // Allow edge builder to call set_edge
     friend EdgeBuilder;
 
@@ -200,10 +235,25 @@ template <typename KeyT, typename T>
 class MultiEdgeHolder
 {
   public:
-    MultiEdgeHolder()          = default;
+    MultiEdgeHolder(std::string name = std::string()) : m_name(std::move(name)){};
     virtual ~MultiEdgeHolder() = default;
 
+    const std::string& name() const
+    {
+        return m_name;
+    };
+
   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);
@@ -276,7 +326,9 @@ class MultiEdgeHolder
         {
             if (create_if_missing)
             {
-                m_edges[key] = EdgeHolder<T>();
+                std::ostringstream edge_name;
+                edge_name << m_name << "_" << key;
+                m_edges[key] = EdgeHolder<T>(edge_name.str());
                 return m_edges[key];
             }
 
@@ -321,6 +373,8 @@ class MultiEdgeHolder
     // Keeps pairs of get_edge/set_edge for each key
     std::map<KeyT, EdgeHolder<T>> m_edges;
 
+    std::string m_name;
+
     // Allow edge builder to call set_edge
     friend EdgeBuilder;
 };

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

@@ -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

@@ -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,13 @@ class TypedEgress : public EgressDelegate
 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

@@ -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 @@ 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 +54,13 @@ class TypedIngress : public IngressDelegate
 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

@@ -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/generic_sink.hpp

@@ -63,7 +63,7 @@ template <typename T>
 class GenericSinkComponent : public RxSinkComponent<T>
 {
   public:
-    GenericSinkComponent()
+    GenericSinkComponent(std::string name = std::string()) : RxSinkComponent<T>(std::move(name))
     {
         RxSinkComponent<T>::set_observer(rxcpp::make_observer_dynamic<T>(
             [this](T data) {

cpp/mrc/include/mrc/node/generic_source.hpp

@@ -71,7 +71,7 @@ template <typename T>
 class GenericSourceComponent : public ForwardingEgressProvider<T>
 {
   public:
-    GenericSourceComponent()           = default;
+    GenericSourceComponent(std::string name = std::string()) : m_name(std::move(name)) {}
     ~GenericSourceComponent() override = default;
 
   private:
@@ -81,6 +81,8 @@ class GenericSourceComponent : public ForwardingEgressProvider<T>
     }
 
     virtual mrc::channel::Status get_data(T& data) = 0;
+
+    std::string m_name;
 };
 
 template <typename T>
@@ -90,6 +92,10 @@ class LambdaSourceComponent : public GenericSourceComponent<T>
     using get_data_fn_t = std::function<mrc::channel::Status(T&)>;
 
     LambdaSourceComponent(get_data_fn_t get_data_fn) : m_get_data_fn(std::move(get_data_fn)) {}
+    LambdaSourceComponent(std::string name, get_data_fn_t get_data_fn) :
+      GenericSourceComponent<T>(std::move(name)),
+      m_get_data_fn(std::move(get_data_fn))
+    {}
     ~LambdaSourceComponent() override = default;
 
   private:

cpp/mrc/include/mrc/node/operators/broadcast.hpp

@@ -34,6 +34,8 @@ namespace mrc::node {
 class BroadcastTypeless : public edge::IWritableProviderBase, public edge::IWritableAcceptorBase
 {
   public:
+    BroadcastTypeless(std::string name = std::string()) : m_name(std::move(name)) {}
+
     std::shared_ptr<edge::WritableEdgeHandle> get_writable_edge_handle() const override
     {
         auto* self = const_cast<BroadcastTypeless*>(this);
@@ -141,6 +143,7 @@ class BroadcastTypeless : public edge::IWritableProviderBase, public edge::IWrit
     }
 
   private:
+    std::string m_name;
     std::mutex m_mutex;
     std::vector<std::weak_ptr<edge::DeferredWritableMultiEdgeBase>> m_upstream_handles;
     std::vector<std::shared_ptr<edge::WritableEdgeHandle>> m_downstream_handles;
@@ -199,18 +202,18 @@ class Broadcast : public WritableProvider<T>, public edge::IWritableAcceptor<T>
 
     Broadcast(bool deep_copy = false)
     {
-        auto edge = std::make_shared<BroadcastEdge>(*this, deep_copy);
-
-        // Save to avoid casting
-        m_edge = edge;
+        init_edge(deep_copy);
+    }
 
-        WritableProvider<T>::init_owned_edge(edge);
+    Broadcast(std::string name, bool deep_copy = false) : m_name(std::move(name))
+    {
+        init_edge(deep_copy);
     }
 
     ~Broadcast()
     {
         // Debug print
-        VLOG(10) << "Destroying TestBroadcast";
+        VLOG(10) << "Destroying Broadcast " << m_name;
     }
 
     void set_writable_edge_handle(std::shared_ptr<edge::WritableEdgeHandle> ingress) override
@@ -227,11 +230,22 @@ class Broadcast : public WritableProvider<T>, public edge::IWritableAcceptor<T>
 
     void on_complete()
     {
-        VLOG(10) << "TestBroadcast completed";
+        VLOG(10) << "Broadcast completed " << m_name;
     }
 
   private:
+    void init_edge(bool deep_copy)
+    {
+        auto edge = std::make_shared<BroadcastEdge>(*this, deep_copy);
+
+        // Save to avoid casting
+        m_edge = edge;
+
+        WritableProvider<T>::init_owned_edge(edge);
+    }
+
     std::weak_ptr<BroadcastEdge> m_edge;
+    std::string m_name;
 };
 
 }  // namespace mrc::node

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

@@ -23,17 +23,28 @@
 #include "mrc/node/sink_properties.hpp"
 #include "mrc/node/source_properties.hpp"
 
+#include <string>
+
 namespace mrc::node {
 
+using namespace std::string_literals;
+
 template <typename T>
 class Queue : public WritableProvider<T>, public ReadableProvider<T>
 {
   public:
-    Queue()
+    Queue(const std::string& name = std::string()) :
+      SinkProperties<T>(name + "-sink"s),
+      SourceProperties<T>(name + "-source"s)
     {
         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)
     {