Add a pre-flight check when pe_count > cpu_set.weight() (#547)

* Replace the `DCHECK_LE` in the `ReusableFiberEngineFactory` class with an if/throw.
* Update `~EdgeHolder` to `LOG(DFATAL)`. The thinking here is that this error is only of use to MRC developers and not of use to downstream users (Borrowed from PR #434).

Authors:
  - David Gardner (https://github.com/dagardner-nv)

Approvers:
  - Will Killian (https://github.com/willkill07)

URL: #547

Author: dagardner-nv

.github/workflows/ci_pipe.yml

@@ -201,7 +201,7 @@ jobs:
 
   codecov:
     name: Code Coverage
-    runs-on: linux-amd64-gpu-v100-latest-1
+    runs-on: linux-amd64-gpu-l4-latest-1
     timeout-minutes: 60
     container:
       credentials:

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

@@ -37,6 +37,7 @@
 #include <memory>
 #include <mutex>
 #include <optional>
+#include <sstream>  // for std::stringstream
 #include <stdexcept>
 #include <type_traits>
 #include <typeindex>
@@ -58,12 +59,37 @@ class EdgeHolder
 
         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

…/mrc/src/internal/system/device_info.hpp …p/mrc/include/mrc/system/device_info.hppcpp/mrc/src/internal/system/device_info.hpp renamed to cpp/mrc/include/mrc/system/device_info.hpp cpp/mrc/src/internal/system/device_info.hpp renamed to cpp/mrc/include/mrc/system/device_info.hpp

@@ -45,6 +45,7 @@ struct DeviceInfo
     static auto PowerLimit(unsigned int device_id) -> double;
     static auto PowerUsage(unsigned int device_id) -> double;
     static auto UUID(unsigned int device_id) -> std::string;
+    static void Reset();
     // NOLINTEND(readability-identifier-naming)
 };
 

cpp/mrc/src/internal/runnable/engine_factory.cpp

@@ -83,7 +83,10 @@ class ReusableFiberEngineFactory final : public FiberEngineFactory
 
     std::vector<std::reference_wrapper<core::FiberTaskQueue>> get_next_n_queues(std::size_t count) final
     {
-        DCHECK_LE(count, m_pool.thread_count());
+        if (count > m_pool.thread_count())
+        {
+            throw exceptions::MrcRuntimeError("more dedicated threads/cores than available");
+        }
         std::vector<std::reference_wrapper<core::FiberTaskQueue>> queues;
 
         for (int i = 0; i < count; ++i)
@@ -193,6 +196,11 @@ class ReusableThreadEngineFactory final : public ThreadEngineFactory
   protected:
     CpuSet get_next_n_cpus(std::size_t count) final
     {
+        if (count > this->cpu_set().weight())
+        {
+            throw exceptions::MrcRuntimeError("more dedicated threads/cores than available");
+        }
+
         CpuSet cpu_set;
         for (int i = 0; i < count; ++i)
         {

cpp/mrc/src/internal/system/device_info.cpp

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-#include "internal/system/device_info.hpp"
+#include "mrc/system/device_info.hpp"
 
 #include "mrc/cuda/common.hpp"
 
@@ -33,6 +33,7 @@
 #include <cstddef>
 #include <cstdio>
 #include <memory>
+#include <new>  // for new
 #include <ostream>
 #include <set>
 #include <stdexcept>
@@ -255,6 +256,13 @@ struct NvmlState
         return NvmlState::instance().get_handle();
     }
 
+    static void reset()
+    {
+        auto& state = NvmlState::instance();
+        state.~NvmlState();
+        new (&state) NvmlState();
+    }
+
   private:
     // this object can also hold the list of device handles that we have access to.
     // - nvmlDeviceGetCount_v2 - will tell us the total number of devices we have access to, i.e. the range of [0, N)
@@ -396,4 +404,9 @@ std::string DeviceInfo::UUID(unsigned int device_id)
     return buffer.data();
 }
 
+void DeviceInfo::Reset()
+{
+    NvmlState::reset();
+}
+
 }  // namespace mrc::system

cpp/mrc/src/internal/system/topology.cpp

@@ -17,13 +17,13 @@
 
 #include "internal/system/topology.hpp"
 
-#include "internal/system/device_info.hpp"
 #include "internal/utils/ranges.hpp"
 
 #include "mrc/core/bitmap.hpp"
 #include "mrc/cuda/common.hpp"
 #include "mrc/exceptions/runtime_error.hpp"
 #include "mrc/options/topology.hpp"
+#include "mrc/system/device_info.hpp"
 
 #include <cuda_runtime.h>
 #include <glog/logging.h>

cpp/mrc/src/tests/test_topology.cpp

@@ -15,12 +15,12 @@
  * limitations under the License.
  */
 
-#include "internal/system/device_info.hpp"
 #include "internal/system/gpu_info.hpp"
 #include "internal/system/topology.hpp"
 
 #include "mrc/core/bitmap.hpp"
 #include "mrc/options/topology.hpp"
+#include "mrc/system/device_info.hpp"
 
 #include <glog/logging.h>
 #include <gtest/gtest.h>

cpp/mrc/tests/test_edges.cpp

@@ -53,15 +53,15 @@ namespace mrc {
 TEST_F(TestEdgesDeathTest, NodeDestroyedBeforeEdge)
 {
     // Reset the sink before the source which will cause an exception
-    EXPECT_DEATH(
+    EXPECT_DEBUG_DEATH(
         {
             auto source = std::make_shared<node::TestSource<int>>();
             auto sink   = std::make_shared<node::TestSink<int>>();
 
             mrc::make_edge(*source, *sink);
             sink.reset();
         },
-        "");
+        "A node was destructed which still had dependent connections");
 }
 
 TEST_F(TestEdges, SourceToSink)

cpp/mrc/tests/test_mrc.hpp

@@ -40,6 +40,14 @@ namespace mrc {
     #define SKIP_IF_CODE_COV()
 #endif
 
+// Similar to EXPECT_DEBUG_DEATH, this macro is used when a given statement is expected to abort in debug mode,
+// but throw an exception in release
+#ifdef NDEBUG
+    #define EXPECT_DEATH_OR_THROW(statement, regex, exception) EXPECT_THROW(statement, exception);
+#else
+    #define EXPECT_DEATH_OR_THROW(statement, regex, exception) EXPECT_DEATH(statement, regex);
+#endif
+
 // class that records when it's moved/copied
 struct CopyMoveCounter
 {

cpp/mrc/tests/test_node.cpp

@@ -21,15 +21,18 @@
 #include "mrc/node/rx_node.hpp"
 #include "mrc/node/rx_sink.hpp"
 #include "mrc/node/rx_source.hpp"
+#include "mrc/options/engine_groups.hpp"
 #include "mrc/options/options.hpp"
 #include "mrc/options/placement.hpp"
 #include "mrc/options/topology.hpp"
 #include "mrc/pipeline/executor.hpp"
 #include "mrc/pipeline/pipeline.hpp"
 #include "mrc/runnable/context.hpp"
 #include "mrc/runnable/launch_options.hpp"
+#include "mrc/runnable/types.hpp"  // for EngineType
 #include "mrc/segment/builder.hpp"
 #include "mrc/segment/object.hpp"
+#include "mrc/system/device_info.hpp"
 #include "mrc/utils/string_utils.hpp"
 
 #include <boost/fiber/operations.hpp>
@@ -789,4 +792,112 @@ TEST_P(ParallelTests, NodeMultiThread)
     EXPECT_EQ(complete_count, 1);
 }
 
+struct PeExceedsResourcesParams
+{
+    std::string cpu_set;
+    std::size_t pe_count;
+    std::string bad_node;
+    mrc::runnable::EngineType engine_type;
+};
+
+struct PeExceedsTests : public testing::TestWithParam<PeExceedsResourcesParams>
+{};
+
+// Run parallel tests for 1, 2 and 4 threads
+INSTANTIATE_TEST_SUITE_P(TestNode,
+                         PeExceedsTests,
+                         testing::Values(PeExceedsResourcesParams{"0-9",  // cpu set of 10 cores
+                                                                  11,     // pe count one greater
+                                                                  "sink",
+                                                                  mrc::runnable::EngineType::Fiber},
+                                         PeExceedsResourcesParams{"0-5",  // cpu set of 6 cores
+                                                                  11,
+                                                                  "source",
+                                                                  mrc::runnable::EngineType::Fiber},
+                                         PeExceedsResourcesParams{"0-3",  // cpu set of 4 cores
+                                                                  5,      // pe lower than cores
+                                                                  "node",
+                                                                  mrc::runnable::EngineType::Fiber},
+                                         PeExceedsResourcesParams{"0-9",  // cpu set of 10 cores
+                                                                  11,     // pe count one greater
+                                                                  "sink",
+                                                                  mrc::runnable::EngineType::Thread},
+                                         PeExceedsResourcesParams{"0-5",  // cpu set of 6 cores
+                                                                  11,
+                                                                  "source",
+                                                                  mrc::runnable::EngineType::Thread},
+                                         PeExceedsResourcesParams{"0-3",  // cpu set of 4 cores
+                                                                  5,      // pe lower than cores
+                                                                  "node",
+                                                                  mrc::runnable::EngineType::Thread}));
+
+TEST_P(PeExceedsTests, PeExceedsResources)
+{
+    const auto test_params = GetParam();
+
+    EXPECT_DEATH_OR_THROW(
+        {
+            // EXPECT_DEATH Executes this test in a fork, we need to reset the NvmlState singleton which would
+            // otherwise reflect the state of the parent process
+            mrc::system::DeviceInfo::Reset();
+            auto p = mrc::make_pipeline();
+
+            const std::string cpu_set  = test_params.cpu_set;
+            const std::size_t pe_count = test_params.pe_count;
+
+            auto my_segment = p->make_segment("my_segment", [&](segment::IBuilder& seg) {
+                auto source = seg.make_source<int>("src1", [&](rxcpp::subscriber<int>& s) {
+                    EXPECT_TRUE(false) << "Preflight checks should fail, this should not be "
+                                          "called";
+                });
+
+                auto node = seg.make_node<int>("node", rxcpp::operators::map([&](const int& x) {
+                                                   EXPECT_TRUE(false) << "Preflight checks should fail, this should "
+                                                                         "not be "
+                                                                         "called";
+                                                   return x;
+                                               }));
+
+                auto sink = seg.make_sink<int>(
+                    "sink",
+                    [&](const int& x) {
+                        EXPECT_TRUE(false) << "Preflight checks should fail, this should not be "
+                                              "called";
+                    },
+                    [&]() {
+                        EXPECT_TRUE(false) << "Preflight checks should fail, this should not be "
+                                              "called";
+                    });
+
+                if (test_params.bad_node == "source")
+                {
+                    source->launch_options().pe_count = pe_count;
+                }
+                else if (test_params.bad_node == "node")
+                {
+                    node->launch_options().pe_count = pe_count;
+                }
+                else
+                {
+                    sink->launch_options().pe_count = pe_count;
+                }
+
+                seg.make_edge(source, node);
+                seg.make_edge(node, sink);
+            });
+
+            auto options = std::make_unique<Options>();
+            options->topology().user_cpuset(cpu_set);
+            options->engine_factories().set_default_engine_type(test_params.engine_type);
+
+            Executor exec(std::move(options));
+            exec.register_pipeline(std::move(p));
+
+            exec.start();
+            exec.join();
+        },
+        "A node was destructed which still had dependent connections",
+        std::runtime_error);
+}
+
 }  // namespace mrc