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.

ci/conda/environments/dev_env.yml

@@ -25,7 +25,7 @@ dependencies:
   - autoconf>=2.69
   - bash-completion
   - benchmark=1.6.0
-  - boost-cpp=1.74
+  - boost-cpp=1.82
   - ccache
   - cmake=3.24
   - cuda-toolkit # Version comes from the channel above
@@ -46,7 +46,7 @@ dependencies:
   - isort
   - jinja2=3.0
   - lcov=1.15
-  - libhwloc=2.5
+  - libhwloc=2.9.2
   - libprotobuf=3.21
   - librmm=23.06
   - libtool

ci/release/update-version.sh

@@ -60,7 +60,10 @@ function sed_runner() {
 
 # .gitmodules
 git submodule set-branch -b branch-${NEXT_SHORT_TAG} morpheus_utils
-git submodule update --remote
+if [[ "$(git diff --name-only | grep .gitmodules)" != "" ]]; then
+   # Only update the submodules if setting the branch changed .gitmodules
+   git submodule update --remote
+fi
 
 # Root CMakeLists.txt
 sed_runner 's/'"VERSION ${CURRENT_FULL_VERSION}.*"'/'"VERSION ${NEXT_FULL_VERSION}"'/g' CMakeLists.txt

ci/scripts/cpp_checks.sh

@@ -80,9 +80,22 @@ if [[ -n "${MRC_MODIFIED_FILES}" ]]; then
 
    # Include What You Use
    if [[ "${SKIP_IWYU}" == "" ]]; then
-      IWYU_DIRS="cpp python"
+      # Remove .h, .hpp, and .cu files from the modified list
+      shopt -s extglob
+      IWYU_MODIFIED_FILES=( "${MRC_MODIFIED_FILES[@]/*.@(h|hpp|cu)/}" )
+
+      # Get the list of compiled files relative to this directory
+      WORKING_PREFIX="${PWD}/"
+      COMPILED_FILES=( $(jq -r .[].file ${BUILD_DIR}/compile_commands.json | sort -u ) )
+      COMPILED_FILES=( "${COMPILED_FILES[@]/#$WORKING_PREFIX/}" )
+      COMBINED_FILES=("${COMPILED_FILES[@]}")
+      COMBINED_FILES+=("${IWYU_MODIFIED_FILES[@]}")
+
+      # Find the intersection between compiled files and modified files
+      IWYU_MODIFIED_FILES=( $(printf '%s\0' "${COMBINED_FILES[@]}" | sort -z | uniq -d -z | xargs -0n1) )
+
       NUM_PROC=$(get_num_proc)
-      IWYU_OUTPUT=`${IWYU_TOOL} -p ${BUILD_DIR} -j ${NUM_PROC} ${IWYU_DIRS} 2>&1`
+      IWYU_OUTPUT=`${IWYU_TOOL} -p ${BUILD_DIR} -j ${NUM_PROC} ${IWYU_MODIFIED_FILES[@]} 2>&1`
       IWYU_RETVAL=$?
    fi
 else

ci/scripts/github/common.sh

@@ -106,16 +106,18 @@ function update_conda_env() {
 print_env_vars
 
 function fetch_base_branch() {
-    rapids-logger "Retrieving base branch from GitHub API"
-    [[ -n "$GH_TOKEN" ]] && CURL_HEADERS=('-H' "Authorization: token ${GH_TOKEN}")
-    RESP=$(
-    curl -s \
-        -H "Accept: application/vnd.github.v3+json" \
-        "${CURL_HEADERS[@]}" \
-        "${GITHUB_API_URL}/repos/${ORG_NAME}/${REPO_NAME}/pulls/${PR_NUM}"
-    )
-
-    BASE_BRANCH=$(echo "${RESP}" | jq -r '.base.ref')
+    if [[ "${BASE_BRANCH}" == "" ]]; then
+        rapids-logger "Retrieving base branch from GitHub API"
+        [[ -n "$GH_TOKEN" ]] && CURL_HEADERS=('-H' "Authorization: token ${GH_TOKEN}")
+        RESP=$(
+        curl -s \
+            -H "Accept: application/vnd.github.v3+json" \
+            "${CURL_HEADERS[@]}" \
+            "${GITHUB_API_URL}/repos/${ORG_NAME}/${REPO_NAME}/pulls/${PR_NUM}"
+        )
+
+        BASE_BRANCH=$(echo "${RESP}" | jq -r '.base.ref')
+    fi
 
     # Change target is the branch name we are merging into but due to the weird way jenkins does
     # the checkout it isn't recognized by git without the origin/ prefix

cpp/mrc/src/internal/segment/builder_definition.cpp

@@ -284,6 +284,10 @@ void BuilderDefinition::initialize()
                    << ", Segment Rank: " << m_rank << ". Exception message:\n"
                    << e.what();
 
+        m_objects.clear();
+        m_ingress_ports.clear();
+        m_egress_ports.clear();
+
         // Rethrow after logging
         std::rethrow_exception(std::current_exception());
     }

cpp/mrc/tests/test_pipeline.cpp

@@ -36,6 +36,7 @@
 #include <map>
 #include <memory>
 #include <ostream>
+#include <stdexcept>
 #include <string>
 #include <utility>
 #include <vector>
@@ -105,8 +106,6 @@ TEST_F(TestPipeline, DuplicateSegments)
 
 TEST_F(TestPipeline, TwoSegment)
 {
-    GTEST_SKIP() << "#185";
-
     std::atomic<int> next_count     = 0;
     std::atomic<int> complete_count = 0;
 
@@ -162,6 +161,51 @@ TEST_F(TestPipeline, TwoSegment)
     LOG(INFO) << "Done" << std::endl;
 }
 
+TEST_F(TestPipeline, SegmentInitErrorHandling)
+{
+    // Test to reproduce issue #360
+    auto pipeline = mrc::make_pipeline();
+
+    auto seg_1 =
+        pipeline->make_segment("seg_1", segment::EgressPorts<float>({"float_port"}), [](segment::IBuilder& seg) {
+            auto rx_source = seg.make_source<float>("rx_source", [](rxcpp::subscriber<float> s) {
+                FAIL() << "This should not be called";
+            });
+
+            auto my_float_egress = seg.get_egress<float>("float_port");
+
+            seg.make_edge(rx_source, my_float_egress);
+        });
+
+    auto seg_2 = pipeline->make_segment("seg_2",
+                                        segment::IngressPorts<float>({"float_port"}),
+                                        [&](segment::IBuilder& seg) {
+                                            auto my_float_ingress = seg.get_ingress<float>("float_port");
+
+                                            auto rx_sink = seg.make_sink<float>("rx_sink",
+                                                                                rxcpp::make_observer_dynamic<float>(
+                                                                                    [&](float x) {
+                                                                                        FAIL() << "This should not be "
+                                                                                                  "called";
+                                                                                    },
+                                                                                    [&]() {
+                                                                                        FAIL() << "This should not be "
+                                                                                                  "called";
+                                                                                    }));
+
+                                            seg.make_edge(my_float_ingress, rx_sink);
+                                            throw std::runtime_error("Error in initializer");
+                                        });
+
+    Executor exec(std::move(m_options));
+
+    exec.register_pipeline(std::move(pipeline));
+
+    exec.start();
+
+    EXPECT_THROW(exec.join(), std::runtime_error);
+}
+
 /*
 TEST_F(TestPipeline, TwoSegmentManualTag)
 {