Merge branch 'branch-23.11' of https://github.com/nv-morpheus/mrc int…

…o pipe-more-args

.github/workflows/pull_request.yml

@@ -47,7 +47,7 @@ jobs:
     name: Prepare
     runs-on: ubuntu-latest
     container:
-      image: rapidsai/ci:latest
+      image: rapidsai/ci-conda:latest
     steps:
       - name: Get PR Info
         id: get-pr-info

Dockerfile

@@ -16,7 +16,7 @@
 # limitations under the License.
 
 
-ARG FROM_IMAGE="rapidsai/ci"
+ARG FROM_IMAGE="rapidsai/ci-conda"
 ARG CUDA_VER=11.8.0
 ARG LINUX_DISTRO=ubuntu
 ARG LINUX_VER=20.04

ci/conda/environments/clang_env.yml

@@ -26,4 +26,4 @@ dependencies:
     - libclang=16
     - libclang-cpp=16
     - llvmdev=16
-    - include-what-you-use
+    - include-what-you-use=0.20

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/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/checks.sh

@@ -24,7 +24,8 @@ update_conda_env
 
 rapids-logger "Configuring CMake"
 git submodule update --init --recursive
-cmake -B build -G Ninja ${CMAKE_BUILD_ALL_FEATURES} .
+CMAKE_CLANG_OPTIONS="-DCMAKE_C_COMPILER:FILEPATH=$(which clang) -DCMAKE_CXX_COMPILER:FILEPATH=$(which clang++) -DCMAKE_CUDA_COMPILER:FILEPATH=$(which nvcc)"
+cmake -B build -G Ninja ${CMAKE_CLANG_OPTIONS} ${CMAKE_BUILD_ALL_FEATURES} .
 
 rapids-logger "Building targets that generate source code"
 cmake --build build --target mrc_style_checks --parallel ${PARALLEL_LEVEL}

cpp/mrc/CMakeLists.txt

@@ -38,6 +38,7 @@ add_library(libmrc
   src/internal/data_plane/server.cpp
   src/internal/executor/executor_definition.cpp
   src/internal/grpc/progress_engine.cpp
+  src/internal/grpc/promise_handler.cpp
   src/internal/grpc/server.cpp
   src/internal/memory/device_resources.cpp
   src/internal/memory/host_resources.cpp

cpp/mrc/include/mrc/core/userspace_threads.hpp

@@ -19,44 +19,51 @@
 
 #include <boost/fiber/all.hpp>
 
-namespace mrc {
+namespace mrc::userspace_threads {
 
-struct userspace_threads  // NOLINT
-{
-    using mutex = boost::fibers::mutex;  // NOLINT
+// Suppress naming conventions in this file to allow matching std and boost libraries
+// NOLINTBEGIN(readability-identifier-naming)
+
+using mutex = boost::fibers::mutex;
+
+using recursive_mutex = boost::fibers::recursive_mutex;
 
-    using cv = boost::fibers::condition_variable;  // NOLINT
+using cv = boost::fibers::condition_variable;
 
-    using launch = boost::fibers::launch;  // NOLINT
+using cv_any = boost::fibers::condition_variable_any;
 
-    template <typename T>
-    using promise = boost::fibers::promise<T>;  // NOLINT
+using launch = boost::fibers::launch;
 
-    template <typename T>
-    using future = boost::fibers::future<T>;  // NOLINT
+template <typename T>
+using promise = boost::fibers::promise<T>;
 
-    template <typename T>
-    using shared_future = boost::fibers::shared_future<T>;  // NOLINT
+template <typename T>
+using future = boost::fibers::future<T>;
 
-    template <class R, class... Args>                                // NOLINT
-    using packaged_task = boost::fibers::packaged_task<R(Args...)>;  // NOLINT
+template <typename T>
+using shared_future = boost::fibers::shared_future<T>;
 
-    template <class Function, class... Args>  // NOLINT
-    static auto async(Function&& f, Args&&... args)
-    {
-        return boost::fibers::async(f, std::forward<Args>(args)...);
-    }
+template <typename SignatureT>
+using packaged_task = boost::fibers::packaged_task<SignatureT>;
+
+template <class Function, class... Args>
+static auto async(Function&& f, Args&&... args)
+{
+    return boost::fibers::async(f, std::forward<Args>(args)...);
+}
+
+template <typename Rep, typename Period>
+static void sleep_for(std::chrono::duration<Rep, Period> const& timeout_duration)
+{
+    boost::this_fiber::sleep_for(timeout_duration);
+}
+
+template <typename Clock, typename Duration>
+static void sleep_until(std::chrono::time_point<Clock, Duration> const& sleep_time_point)
+{
+    boost::this_fiber::sleep_until(sleep_time_point);
+}
 
-    template <typename Rep, typename Period>  // NOLINT
-    static void sleep_for(std::chrono::duration<Rep, Period> const& timeout_duration)
-    {
-        boost::this_fiber::sleep_for(timeout_duration);
-    }
+// NOLINTEND(readability-identifier-naming)
 
-    template <typename Clock, typename Duration>  // NOLINT
-    static void sleep_until(std::chrono::time_point<Clock, Duration> const& sleep_time_point)
-    {
-        boost::this_fiber::sleep_until(sleep_time_point);
-    }
-};
-}  // namespace mrc
+}  // namespace mrc::userspace_threads

cpp/mrc/include/mrc/core/utils.hpp

@@ -60,9 +60,12 @@ std::set<KeyT> extract_keys(const std::map<KeyT, ValT>& stdmap)
 class Unwinder
 {
   public:
-    explicit Unwinder(std::function<void()> unwind_fn) : m_unwind_fn(std::move(unwind_fn)) {}
+    explicit Unwinder(std::function<void()> unwind_fn) :
+      m_unwind_fn(std::move(unwind_fn)),
+      m_ctor_exception_count(std::uncaught_exceptions())
+    {}
 
-    ~Unwinder()
+    ~Unwinder() noexcept(false)
     {
         if (!!m_unwind_fn)
         {
@@ -71,8 +74,14 @@ class Unwinder
                 m_unwind_fn();
             } catch (...)
             {
-                LOG(ERROR) << "Fatal error during unwinder function";
-                std::terminate();
+                if (std::uncaught_exceptions() > m_ctor_exception_count)
+                {
+                    LOG(ERROR) << "Error occurred during unwinder function, but another exception is active.";
+                    std::terminate();
+                }
+
+                LOG(ERROR) << "Error occurred during unwinder function. Rethrowing";
+                throw;
             }
         }
     }
@@ -92,6 +101,9 @@ class Unwinder
     }
 
   private:
+    // Stores the number of active exceptions during creation. If the number of active exceptions during destruction is
+    // greater, we do not throw and log error and terminate
+    int m_ctor_exception_count;
     std::function<void()> m_unwind_fn;
 };
 

cpp/mrc/include/mrc/types.hpp

@@ -24,33 +24,40 @@
 
 namespace mrc {
 
+// Suppress naming conventions in this file to allow matching std and boost libraries
+// NOLINTBEGIN(readability-identifier-naming)
+
 // Typedefs
 template <typename T>
-using Promise = userspace_threads::promise<T>;  // NOLINT(readability-identifier-naming)
+using Promise = userspace_threads::promise<T>;
 
 template <typename T>
-using Future = userspace_threads::future<T>;  // NOLINT(readability-identifier-naming)
+using Future = userspace_threads::future<T>;
 
 template <typename T>
-using SharedFuture = userspace_threads::shared_future<T>;  // NOLINT(readability-identifier-naming)
+using SharedFuture = userspace_threads::shared_future<T>;
+
+using Mutex = userspace_threads::mutex;
 
-using Mutex = userspace_threads::mutex;  // NOLINT(readability-identifier-naming)
+using RecursiveMutex = userspace_threads::recursive_mutex;
 
-using CondV = userspace_threads::cv;  // NOLINT(readability-identifier-naming)
+using CondV = userspace_threads::cv;
 
-using MachineID  = std::uint64_t;  // NOLINT(readability-identifier-naming)
-using InstanceID = std::uint64_t;  // NOLINT(readability-identifier-naming)
-using TagID      = std::uint64_t;  // NOLINT(readability-identifier-naming)
+using MachineID  = std::uint64_t;
+using InstanceID = std::uint64_t;
+using TagID      = std::uint64_t;
 
 template <typename T>
-using Handle = std::shared_ptr<T>;  // NOLINT(readability-identifier-naming)
+using Handle = std::shared_ptr<T>;
+
+using SegmentID      = std::uint16_t;
+using SegmentRank    = std::uint16_t;
+using SegmentAddress = std::uint32_t;  // id + rank
 
-using SegmentID      = std::uint16_t;  // NOLINT(readability-identifier-naming)
-using SegmentRank    = std::uint16_t;  // NOLINT(readability-identifier-naming)
-using SegmentAddress = std::uint32_t;  // NOLINT(readability-identifier-naming) // id + rank
+using PortName    = std::string;
+using PortID      = std::uint16_t;
+using PortAddress = std::uint64_t;  // id + rank + port
 
-using PortName    = std::string;    // NOLINT(readability-identifier-naming)
-using PortID      = std::uint16_t;  // NOLINT(readability-identifier-naming)
-using PortAddress = std::uint64_t;  // NOLINT(readability-identifier-naming)  // id + rank + port
+// NOLINTEND(readability-identifier-naming)
 
 }  // namespace mrc