External process module

modules/CMakeLists.txt

@@ -20,6 +20,7 @@ add_subdirectory(class_loader)
 add_subdirectory(controller)
 add_subdirectory(hardware)
 add_subdirectory(estimator)
+add_subdirectory(external_process)
 add_subdirectory(command)
 add_subdirectory(supervisor)
 add_subdirectory(service)

modules/external_process/CMakeLists.txt

@@ -0,0 +1,38 @@
+cmake_minimum_required(VERSION 3.15)
+
+# Allow the process tests to build without the rest of DLS or ROS installed.
+if(CMAKE_SOURCE_DIR STREQUAL CMAKE_CURRENT_SOURCE_DIR)
+    project(dls_external_process LANGUAGES CXX)
+    enable_testing()
+endif()
+
+find_package(Boost REQUIRED COMPONENTS filesystem system)
+find_package(Threads REQUIRED)
+
+add_library(dls_external_process SHARED
+	src/managed_external_process.cpp
+)
+target_compile_features(dls_external_process PUBLIC cxx_std_17)
+
+target_include_directories(dls_external_process
+	PUBLIC
+		include
+		$<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/../utils/include>
+)
+
+target_link_libraries(dls_external_process
+	PUBLIC
+		Boost::filesystem
+		Boost::system
+		Threads::Threads
+)
+
+if(COMMAND dls_install)
+    dls_install(dls_external_process)
+endif()
+
+option(DLS_EXTERNAL_PROCESS_BUILD_TESTS "Build managed process integration tests" OFF)
+if(DLS_EXTERNAL_PROCESS_BUILD_TESTS)
+    enable_testing()
+    add_subdirectory(tests)
+endif()

modules/external_process/README.md

@@ -0,0 +1,50 @@
+# Managed external processes
+
+`ManagedExternalProcess` keeps the asynchronous lifecycle API used by plugins:
+`start()`, `running()`, `requestStop()`, `stopComplete()`, and `stopAndWait()`.
+Repeated start/stop calls remain guarded. `running()` describes launcher liveness.
+
+It delegates OS process ownership to `dls::utils::OwnedProcess` in
+`dls2/util/owned_process.hpp`. That class creates a dedicated session/process group,
+tracks leader and group liveness separately, and reaps children it owns.
+`shutdownProcesses()` is shared with `run_dls2`:
+
+- The framework interrupts the whole owned group.
+- The plugin wrapper interrupts the launcher first, allowing ROS launch to stop
+  its nodes, then escalates to SIGTERM and SIGKILL for the whole group.
+- The wrapper retains configurable interrupt/terminate intervals and uses a
+  two-second final wait after SIGKILL. Failure is propagated through the future;
+  the destructor logs it. It does not report success merely because SIGKILL was sent.
+
+The owning executable must enable child subreaping if it needs to reap orphaned
+descendants itself; `run_dls2` does so. The reusable wrapper does not change this
+process-wide setting. A separately grouped descendant is not covered by its
+ancestor's group signals if its immediate owner is forcibly killed.
+
+`ShutdownSignal` handles incoming SIGINT/SIGTERM in `run_dls2` and
+`child_process_launcher`. Its callback requests application shutdown, which reaches
+the plugin's cleanup hooks. `shutdownProcesses()` sends outgoing signals to owned
+children; the plugin wrapper does not install another signal handler.
+
+## Integration test
+
+From the PEGASUS repository root, build and run this module independently:
+
+```sh
+cmake -S dls2-barebone/dls2/modules/external_process \
+  -B /tmp/dls-external-process-tests \
+  -DDLS_EXTERNAL_PROCESS_BUILD_TESTS=ON
+cmake --build /tmp/dls-external-process-tests -j2
+ctest --test-dir /tmp/dls-external-process-tests --output-on-failure
+```
+
+Requires Linux, CMake, a C++17 compiler, and Boost filesystem/system development
+libraries. ROS and the rest of DLS are not required for this standalone test build.
+
+
+The test launches dummy child/grandchild processes and checks graceful shutdown,
+forced shutdown of processes ignoring signals, cleanup after the launcher exits,
+repeated start/stop requests, reactivation, destructor cleanup, and a missing
+executable. It also checks session isolation, group liveness after leader exit,
+both initial signal policies, and compatibility with the framework shutdown API.
+It does not launch ROS or Nav2. Use `ctest -V` to see each scenario.

modules/external_process/include/dls2/external_process/managed_external_process.hpp

@@ -0,0 +1,39 @@
+#ifndef MANAGED_EXTERNAL_PROCESS_HPP
+#define MANAGED_EXTERNAL_PROCESS_HPP
+
+#include <dls2/util/owned_process.hpp>
+#include <chrono>
+#include <future>
+#include <memory>
+#include <string>
+#include <vector>
+
+namespace dls
+{
+    // Lifecycle methods are called serially by the plugin's state-machine thread.
+    class ManagedExternalProcess
+    {
+    public:
+        ManagedExternalProcess() = default;
+        ~ManagedExternalProcess();
+        ManagedExternalProcess(const ManagedExternalProcess&) = delete;
+        ManagedExternalProcess& operator=(const ManagedExternalProcess&) = delete;
+
+        void start(const std::vector<std::string>& command,
+                   std::chrono::milliseconds interrupt_timeout,
+                   std::chrono::milliseconds terminate_timeout);
+        bool running();
+        void requestStop();
+        bool stopComplete();
+        void stopAndWait();
+
+    private:
+        void stop();
+        std::shared_ptr<utils::OwnedProcess> process_;
+        std::shared_future<void> stop_result_;
+        bool shutdown_sent_{false};
+        std::chrono::milliseconds interrupt_timeout_{15000};
+        std::chrono::milliseconds terminate_timeout_{5000};
+    };
+}
+#endif

modules/external_process/src/managed_external_process.cpp

@@ -0,0 +1,102 @@
+#include "dls2/external_process/managed_external_process.hpp"
+
+#include <cerrno>
+#include <csignal>
+#include <iostream>
+#include <stdexcept>
+#include <system_error>
+#include <thread>
+#include <unistd.h>
+#include <pthread.h>
+#include <sched.h>
+
+using namespace dls;
+
+ManagedExternalProcess::~ManagedExternalProcess()
+{
+    try { stopAndWait(); }
+    catch (const std::exception& error)
+    {
+        std::cerr << "ManagedExternalProcess shutdown failed: " << error.what() << '\n';
+        // Attached Boost handles provide a final forced-termination fallback.
+    }
+}
+
+void ManagedExternalProcess::start(const std::vector<std::string>& command,
+                            std::chrono::milliseconds interrupt_timeout,
+                            std::chrono::milliseconds terminate_timeout)
+{
+    if (process_ && !shutdown_sent_)
+        return;
+
+    if (shutdown_sent_) 
+        stop_result_.get();
+
+    if (command.empty() || command.front().empty())
+        throw std::invalid_argument("Configure launch command string");
+
+    if (interrupt_timeout.count() < 0 || terminate_timeout.count() < 0)
+        throw std::invalid_argument("ManagedExternalProcess shutdown timeouts must be nonnegative");
+
+    const auto executable = command.front().find('/') == std::string::npos
+        ? boost::process::search_path(command.front())
+        : boost::filesystem::path(command.front());
+
+    if (executable.empty())
+        throw std::runtime_error("Launch executable not found: " + command.front());
+
+    auto args = command;
+    args.front() = executable.string();
+    process_ = std::make_shared<utils::OwnedProcess>(args);
+    interrupt_timeout_ = interrupt_timeout;
+    terminate_timeout_ = terminate_timeout;
+    stop_result_ = {};
+    shutdown_sent_ = false;
+}
+
+bool ManagedExternalProcess::running()
+{
+    return !shutdown_sent_ && process_ && process_->leaderRunning();
+}
+
+void ManagedExternalProcess::requestStop()
+{
+    if (shutdown_sent_) return;
+    stop_result_ = std::async(std::launch::async, [this] { stop(); }).share();
+    shutdown_sent_ = true;
+}
+
+bool ManagedExternalProcess::stopComplete()
+{
+    if (!shutdown_sent_) return false;
+    if (stop_result_.wait_for(std::chrono::milliseconds(0)) != std::future_status::ready)
+        return false;
+
+    stop_result_.get();
+    return true;
+}
+
+void ManagedExternalProcess::stopAndWait()
+{
+    requestStop();
+    stop_result_.get();
+}
+
+void ManagedExternalProcess::stop()
+{
+    if (!process_) return;
+    // A worker created during deactivation may inherit the periodic policy.
+    sched_param parameters{};
+    const int scheduler_error = ::pthread_setschedparam(::pthread_self(), SCHED_OTHER, &parameters);
+    if (scheduler_error != 0)
+        std::cerr << "External group shutdown worker could not select SCHED_OTHER: "
+                    << std::generic_category().message(scheduler_error) << '\n';
+    utils::ProcessShutdownOptions options;
+    options.interrupt_timeout = interrupt_timeout_;
+    options.terminate_timeout = terminate_timeout_;
+    options.initial_target = utils::InitialSignalTarget::leader;
+    utils::OwnedProcesses processes{{"external process", process_}};
+    if (!utils::shutdownProcesses(processes, options))
+        throw std::runtime_error("External process group did not exit after SIGKILL");
+    process_.reset();
+}

modules/external_process/tests/CMakeLists.txt

@@ -0,0 +1,7 @@
+add_executable(test_external_process_managed_process managed_external_process_test.cpp)
+target_link_libraries(test_external_process_managed_process PRIVATE dls_external_process)
+add_test(NAME external_process_managed_process COMMAND test_external_process_managed_process)
+set_tests_properties(external_process_managed_process PROPERTIES TIMEOUT 15)
+if(TARGET dls2-tests)
+    add_dependencies(dls2-tests test_external_process_managed_process)
+endif()

modules/external_process/tests/managed_external_process_test.cpp

@@ -0,0 +1,149 @@
+#include <dls2/external_process/managed_external_process.hpp>
+
+#include <csignal>
+#include <filesystem>
+#include <fstream>
+#include <iostream>
+#include <stdexcept>
+#include <thread>
+#include <sys/prctl.h>
+#include <sys/wait.h>
+#include <unistd.h>
+
+using namespace std::chrono_literals;
+namespace fs = std::filesystem;
+using dls::ManagedExternalProcess;
+
+namespace
+{
+    volatile std::sig_atomic_t interrupted = 0;
+    void interrupt(int) { interrupted = 1; }
+    void require(bool condition, const char* message)
+    {
+        if (!condition) throw std::runtime_error(message);
+    }
+    template<class Predicate> void await(Predicate predicate)
+    {
+        const auto deadline = std::chrono::steady_clock::now() + 3s;
+        while (!predicate())
+        {
+            require(std::chrono::steady_clock::now() < deadline, "Timed out");
+            std::this_thread::sleep_for(5ms);
+        }
+    }
+
+    int fixture(const std::string& mode, const std::string& ready)
+    {
+        std::signal(SIGINT, (mode == "graceful" || mode == "policy") ? interrupt : SIG_IGN);
+        std::signal(SIGTERM, SIG_IGN);
+        const auto node = ::fork();
+        if (node == 0)
+        {
+            if (mode == "policy") {
+                while (!interrupted) std::this_thread::sleep_for(5ms);
+                std::ofstream(ready + ".node-interrupted") << "SIGINT\n";
+                return 0;
+            }
+            std::signal(SIGINT, SIG_IGN);
+            for (;;) ::pause();
+        }
+        require(node > 0, "fork failed");
+        { std::ofstream output(ready); output << ::getpid() << ' ' << node << '\n'; }
+        if (mode == "orphan") return 0;
+        while (!interrupted) std::this_thread::sleep_for(5ms);
+        if (mode == "policy") std::this_thread::sleep_for(100ms);
+        ::kill(node, SIGKILL);
+        ::waitpid(node, nullptr, 0);
+        return 0;
+    }
+
+    void checkGone(pid_t launcher, pid_t node)
+    {
+        // Adopt grandchildren so the test can reap them like a container init.
+        await([&] {
+            while (::waitpid(-1, nullptr, WNOHANG) > 0) {}
+            return ::kill(launcher, 0) == -1 && ::kill(node, 0) == -1;
+        });
+    }
+}
+
+int main(int argc, char** argv)
+{
+    if (argc == 4) return fixture(argv[2], argv[3]);
+    require(::prctl(PR_SET_CHILD_SUBREAPER, 1) == 0, "subreaper failed");
+    const auto executable = fs::canonical("/proc/self/exe").string();
+    const auto directory = fs::temp_directory_path() / ("dls-process-test-" + std::to_string(::getpid()));
+    fs::create_directory(directory);
+    ManagedExternalProcess process;
+    int cycle = 0;
+    for (const auto& mode : {"graceful", "stubborn", "orphan", "graceful"})
+    {
+        std::cout << "Testing " << mode << " shutdown, cycle " << cycle << std::endl;
+        const auto ready = (directory / std::to_string(cycle++)).string();
+        const std::vector<std::string> command{executable, "fixture", mode, ready};
+        process.start(command, 100ms, 100ms);
+        // Repeated activation must not spawn another process.
+        process.start({"does-not-exist"}, 100ms, 100ms);
+        pid_t launcher = 0, node = 0;
+        await([&] { std::ifstream input(ready); return bool(input >> launcher >> node); });
+        require(::getpgid(launcher) != ::getpgrp(), "Shared process group");
+        require(::getsid(launcher) == launcher, "Child did not create a dedicated session");
+        if (std::string(mode) == "orphan") await([&] { return !process.running(); });
+        const auto before = std::chrono::steady_clock::now();
+        process.requestStop();
+        require(std::chrono::steady_clock::now() - before < 100ms, "Stop request blocked");
+        await([&] { process.requestStop(); return process.stopComplete(); });
+        require(!process.running(), "Launcher still running");
+        process.stopAndWait();
+        checkGone(launcher, node);
+    }
+    std::cout << "Testing destructor cleanup" << std::endl;
+    pid_t launcher = 0, node = 0;
+    {
+        ManagedExternalProcess scoped;
+        const auto ready = (directory / "destructor").string();
+        scoped.start({executable, "fixture", "stubborn", ready}, 50ms, 50ms);
+        await([&] { std::ifstream input(ready); return bool(input >> launcher >> node); });
+    }
+    checkGone(launcher, node);
+    std::cout << "Testing missing executable" << std::endl;
+    bool rejected = false;
+    try { process.start({"dls-missing-executable-for-process-test"}, 50ms, 50ms); }
+    catch (const std::exception&) { rejected = true; }
+    require(rejected, "Missing executable was accepted");
+    process.stopAndWait();
+
+    for (const auto target : {dls::utils::InitialSignalTarget::leader,
+                              dls::utils::InitialSignalTarget::group}) {
+        const bool whole_group = target == dls::utils::InitialSignalTarget::group;
+        std::cout << "Testing initial signal target: " << (whole_group ? "group" : "leader") << std::endl;
+        const auto ready = (directory / (whole_group ? "group-policy" : "leader-policy")).string();
+        auto owned = std::make_shared<dls::utils::OwnedProcess>(
+            std::vector<std::string>{executable, "fixture", "policy", ready});
+        await([&] { std::ifstream input(ready); return bool(input >> launcher >> node); });
+        dls::utils::OwnedProcesses children{{"policy fixture", owned}};
+        dls::utils::ProcessShutdownOptions options;
+        options.initial_target = target;
+        options.interrupt_timeout = 500ms;
+        options.terminate_timeout = 100ms;
+        options.kill_timeout = 500ms;
+        require(dls::utils::shutdownProcesses(children, options), "Policy shutdown failed");
+        require(fs::exists(ready + ".node-interrupted") == whole_group,
+                "SIGINT reached the wrong process set");
+        checkGone(launcher, node);
+    }
+
+    std::cout << "Testing group liveness after leader exit and legacy shutdown API" << std::endl;
+    {
+        const auto ready = (directory / "owned-orphan").string();
+        auto owned = std::make_shared<dls::utils::OwnedProcess>(
+            std::vector<std::string>{executable, "fixture", "orphan", ready});
+        await([&] { std::ifstream input(ready); return bool(input >> launcher >> node); });
+        await([&] { return !owned->leaderRunning(); });
+        require(owned->running(), "Surviving descendant was not tracked");
+        dls::utils::OwnedProcesses children{{"orphan fixture", owned}};
+        require(dls::utils::shutdownProcesses(children, 10ms), "Legacy shutdown failed");
+        checkGone(launcher, node);
+    }
+    fs::remove_all(directory);
+}