diff --git a/modules/CMakeLists.txt b/modules/CMakeLists.txt index 20264c96..0075c961 100644 --- a/modules/CMakeLists.txt +++ b/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) diff --git a/modules/external_process/CMakeLists.txt b/modules/external_process/CMakeLists.txt new file mode 100644 index 00000000..522a5ca3 --- /dev/null +++ b/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 + $ +) + +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() diff --git a/modules/external_process/README.md b/modules/external_process/README.md new file mode 100644 index 00000000..365e26ee --- /dev/null +++ b/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. diff --git a/modules/external_process/include/dls2/external_process/managed_external_process.hpp b/modules/external_process/include/dls2/external_process/managed_external_process.hpp new file mode 100644 index 00000000..1e00f144 --- /dev/null +++ b/modules/external_process/include/dls2/external_process/managed_external_process.hpp @@ -0,0 +1,40 @@ +#ifndef MANAGED_EXTERNAL_PROCESS_HPP +#define MANAGED_EXTERNAL_PROCESS_HPP + +#include +#include +#include +#include +#include +#include + +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& 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 process_; + std::shared_future stop_result_; + std::promise stop_request_; + bool shutdown_sent_{false}; + std::chrono::milliseconds interrupt_timeout_{15000}; + std::chrono::milliseconds terminate_timeout_{5000}; + }; +} +#endif diff --git a/modules/external_process/src/managed_external_process.cpp b/modules/external_process/src/managed_external_process.cpp new file mode 100644 index 00000000..2f92a74e --- /dev/null +++ b/modules/external_process/src/managed_external_process.cpp @@ -0,0 +1,116 @@ +#include "dls2/external_process/managed_external_process.hpp" + +#include +#include +#include +#include +#include + +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& command, + std::chrono::milliseconds interrupt_timeout, + std::chrono::milliseconds terminate_timeout) +{ + if (process_ && !shutdown_sent_) + return; + + if (shutdown_sent_ && stop_result_.valid()) + 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(args); + interrupt_timeout_ = interrupt_timeout; + terminate_timeout_ = terminate_timeout; + stop_result_ = {}; + shutdown_sent_ = false; + stop_request_ = std::promise{}; + try + { + // Activation runs outside SCHED_DEADLINE. Create the worker here: + // a deadline thread cannot create it later from requestStop(). + stop_result_ = std::async(std::launch::async, + [this, request = stop_request_.get_future()]() mutable { + request.wait(); + stop(); + }).share(); + } + catch (...) + { + stop(); + throw; + } +} + +bool ManagedExternalProcess::running() +{ + return !shutdown_sent_ && process_ && process_->leaderRunning(); +} + +void ManagedExternalProcess::requestStop() +{ + if (shutdown_sent_) + return; + + if (stop_result_.valid()) stop_request_.set_value(); + shutdown_sent_ = true; +} + +bool ManagedExternalProcess::stopComplete() +{ + if (!shutdown_sent_) return false; + if (!stop_result_.valid()) return true; + if (stop_result_.wait_for(std::chrono::milliseconds(0)) != std::future_status::ready) + return false; + + stop_result_.get(); + return true; +} + +void ManagedExternalProcess::stopAndWait() +{ + requestStop(); + if (stop_result_.valid()) stop_result_.get(); +} + +void ManagedExternalProcess::stop() +{ + if (!process_) return; + sched_param parameters{}; + const int scheduler_error = ::pthread_setschedparam(::pthread_self(), SCHED_OTHER, ¶meters); + 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(); +} diff --git a/modules/external_process/tests/CMakeLists.txt b/modules/external_process/tests/CMakeLists.txt new file mode 100644 index 00000000..5ab2d198 --- /dev/null +++ b/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() diff --git a/modules/external_process/tests/managed_external_process_test.cpp b/modules/external_process/tests/managed_external_process_test.cpp new file mode 100644 index 00000000..411ff669 --- /dev/null +++ b/modules/external_process/tests/managed_external_process_test.cpp @@ -0,0 +1,182 @@ +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +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 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 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( + std::vector{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( + std::vector{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); + } + std::cout << "Testing stop request from SCHED_DEADLINE" << std::endl; + { + ManagedExternalProcess realtime; + const auto ready = (directory / "deadline").string(); + realtime.start({executable, "fixture", "graceful", ready}, 100ms, 100ms); + await([&] { std::ifstream input(ready); return bool(input >> launcher >> node); }); + struct sched_attr attributes{}; + attributes.size = sizeof(attributes); + attributes.sched_policy = SCHED_DEADLINE; + attributes.sched_runtime = 1000000; + attributes.sched_deadline = 100000000; + attributes.sched_period = 100000000; + if (::syscall(SYS_sched_setattr, 0, &attributes, 0) == 0) { + bool requested = false; + try { realtime.requestStop(); requested = true; } + catch (const std::exception& error) { + std::cerr << "Real-time stop failed: " << error.what() << std::endl; + } + attributes.sched_policy = SCHED_OTHER; + attributes.sched_runtime = attributes.sched_deadline = attributes.sched_period = 0; + require(::syscall(SYS_sched_setattr, 0, &attributes, 0) == 0, "Restore scheduler failed"); + realtime.stopAndWait(); + checkGone(launcher, node); + require(requested, "requestStop could not run under SCHED_DEADLINE"); + } else { + std::cout << "SKIP: SCHED_DEADLINE unavailable (requires CAP_SYS_NICE)" << std::endl; + realtime.stopAndWait(); + checkGone(launcher, node); + } + } + fs::remove_all(directory); +} diff --git a/modules/utils/include/dls2/util/owned_process.hpp b/modules/utils/include/dls2/util/owned_process.hpp index fee00f0d..01ed2971 100644 --- a/modules/utils/include/dls2/util/owned_process.hpp +++ b/modules/utils/include/dls2/util/owned_process.hpp @@ -9,6 +9,8 @@ #include #include #include +#include +#include #include #include #include @@ -42,22 +44,38 @@ class OwnedProcess bool running() { if (finished_) return false; - if (!leader_exited_) { - std::error_code error; - leader_exited_ = !proc.running(error); - if (error) throw std::system_error(error, "checking owned child"); - } + leaderRunning(); if (!own_group_) return !leader_exited_; if (leader_exited_) { int status; while (::waitpid(-group_.native_handle(), &status, WNOHANG) > 0) {} } - if (::kill(-group_.native_handle(), 0) == 0 || errno != ESRCH) return true; + if (::kill(-group_.native_handle(), 0) == 0) return true; + if (errno != ESRCH) + throw std::system_error(errno, std::generic_category(), "checking owned process group"); finished_ = true; group_.detach(); return false; } + // Startup readiness needs the launcher itself, not surviving descendants. + bool leaderRunning() + { + if (finished_) return false; + if (!leader_exited_) { + std::error_code error; + leader_exited_ = !proc.running(error); + if (error) throw std::system_error(error, "checking owned child"); + } + return !leader_exited_; + } + + void signalLeader(int value) + { + if (leaderRunning() && ::kill(proc.id(), value) != 0 && errno != ESRCH) + throw std::system_error(errno, std::generic_category(), "signalling owned child"); + } + void signal(int value) { if (!finished_ && ::kill(own_group_ ? -group_.native_handle() : proc.id(), value) != 0 && errno != ESRCH) @@ -79,9 +97,22 @@ class OwnedProcess using OwnedProcesses = std::map>; -inline bool shutdownProcesses(OwnedProcesses& processes, std::chrono::milliseconds grace) +enum class InitialSignalTarget { group, leader }; + +struct ProcessShutdownOptions +{ + std::chrono::milliseconds interrupt_timeout{10000}; + std::chrono::milliseconds terminate_timeout{2000}; + std::chrono::milliseconds kill_timeout{2000}; + InitialSignalTarget initial_target{InitialSignalTarget::group}; +}; + +inline bool shutdownProcesses(OwnedProcesses& processes, const ProcessShutdownOptions& options) { using namespace std::chrono_literals; + if (options.interrupt_timeout.count() < 0 || options.terminate_timeout.count() < 0 || + options.kill_timeout.count() < 0) + throw std::invalid_argument("Process shutdown timeouts must be nonnegative"); const auto wait = [&](std::chrono::milliseconds duration) { const auto deadline = std::chrono::steady_clock::now() + duration; for (;;) { @@ -97,15 +128,18 @@ inline bool shutdownProcesses(OwnedProcesses& processes, std::chrono::millisecon if (!process->running()) continue; if (report) std::cerr << "Shutdown: " << name << " (PID/PGID " << process->id() << ") still running; sending signal " << value << std::endl; - process->signal(value); + if (value == SIGINT && options.initial_target == InitialSignalTarget::leader) + process->signalLeader(value); + else + process->signal(value); } }; signal(SIGINT, false); - if (wait(grace)) return true; + if (wait(options.interrupt_timeout)) return true; signal(SIGTERM, true); - if (wait(2s)) return true; + if (wait(options.terminate_timeout)) return true; signal(SIGKILL, true); - if (wait(2s)) return true; + if (wait(options.kill_timeout)) return true; for (auto& [name, process] : processes) { if (process->running()) { std::cerr << "Shutdown: " << name << " (PID/PGID " << process->id() @@ -115,4 +149,12 @@ inline bool shutdownProcesses(OwnedProcesses& processes, std::chrono::millisecon } return false; } + +// Preserve the framework's existing group-shutdown interface and defaults. +inline bool shutdownProcesses(OwnedProcesses& processes, std::chrono::milliseconds grace) +{ + ProcessShutdownOptions options; + options.interrupt_timeout = grace; + return shutdownProcesses(processes, options); +} }