diff --git a/docs/source/device/add_device.rst b/docs/source/device/add_device.rst index d5be9b5da4..bd7aeefdda 100644 --- a/docs/source/device/add_device.rst +++ b/docs/source/device/add_device.rst @@ -93,6 +93,9 @@ which uses ``SchemaPusher``). ``Pack()``, then call ``pusher_.push_buffer(ptr, size, sample_time_local_common_clock_ns, sample_time_raw_device_clock_ns)``. Use a monotonic clock (e.g. ``core::os_monotonic_now_ns()``) for the local common clock; use the device's own clock for the raw device clock if available. +- **Shutdown** — To stop a plugin, the plugin manager sends ``SIGINT`` and kills the process if + it is still running 2 seconds later. A plugin that needs longer to close its files sets + ``shutdown_timeout_seconds`` (greater than 0, at most 3600) in its ``plugin.yaml``. Reference implementation: :code-dir:`src/plugins/generic_3axis_pedal`. The plugin holds a ``core::SchemaPusher`` member, opens the Linux joystick device, maps axes to diff --git a/src/core/plugin_manager/cpp/inc/plugin_manager/plugin.hpp b/src/core/plugin_manager/cpp/inc/plugin_manager/plugin.hpp index 379e3745b8..73ec027688 100644 --- a/src/core/plugin_manager/cpp/inc/plugin_manager/plugin.hpp +++ b/src/core/plugin_manager/cpp/inc/plugin_manager/plugin.hpp @@ -7,6 +7,7 @@ # include #endif +#include #include #include #include @@ -17,6 +18,17 @@ namespace core { +// How long stop() waits after SIGINT before it sends SIGKILL, unless the +// plugin's plugin.yaml sets shutdown_timeout_seconds. +inline constexpr double DEFAULT_PLUGIN_SHUTDOWN_TIMEOUT_SECONDS = 2.0; +inline constexpr double MAX_PLUGIN_SHUTDOWN_TIMEOUT_SECONDS = 3600.0; + +/** + * @brief Throw std::invalid_argument unless seconds is greater than 0 and at most + * MAX_PLUGIN_SHUTDOWN_TIMEOUT_SECONDS. + */ +void validate_plugin_shutdown_timeout(double seconds); + enum class ProcessState { RUNNING, @@ -68,11 +80,14 @@ class Plugin * @param working_dir The directory to run the plugin in (where metadata was found) * @param plugin_root_id The root ID for the plugin * @param plugin_args Optional list of arguments to append to the command + * @param shutdown_timeout_seconds Time stop() allows between SIGINT and SIGKILL + * @throws std::invalid_argument if shutdown_timeout_seconds is out of range */ Plugin(const std::string& command, const std::string& working_dir, const std::string& plugin_root_id, - const std::vector& plugin_args = {}); + const std::vector& plugin_args = {}, + double shutdown_timeout_seconds = DEFAULT_PLUGIN_SHUTDOWN_TIMEOUT_SECONDS); /** * @brief Destructor - stops the plugin process @@ -114,8 +129,10 @@ class Plugin int m_pid = -1; #endif bool m_stop_requested = false; + std::chrono::duration m_shutdown_timeout; ProcessSnapshot m_process_snapshot; std::mutex m_process_mutex; + std::mutex m_stop_mutex; // Held for a whole stop_process(); taken before m_process_mutex. }; } // namespace core diff --git a/src/core/plugin_manager/cpp/inc/plugin_manager/plugin_manager.hpp b/src/core/plugin_manager/cpp/inc/plugin_manager/plugin_manager.hpp index d236fae618..a6a28eb08c 100644 --- a/src/core/plugin_manager/cpp/inc/plugin_manager/plugin_manager.hpp +++ b/src/core/plugin_manager/cpp/inc/plugin_manager/plugin_manager.hpp @@ -30,6 +30,7 @@ struct PluginInfo std::string working_dir; std::vector args; std::vector devices; + double shutdown_timeout_seconds = DEFAULT_PLUGIN_SHUTDOWN_TIMEOUT_SECONDS; }; class PluginManager @@ -66,7 +67,8 @@ class PluginManager * @param plugin_name The name of the plugin to start. * @param plugin_root_id The root ID for the plugin. * @param plugin_args Optional arguments appended after plugin.yaml args. - * @return A unique pointer to the Plugin instance (stops on destruction). + * @return A unique pointer to the Plugin instance (stops on destruction), with the + * shutdown timeout from plugin.yaml. */ std::unique_ptr start(const std::string& plugin_name, const std::string& plugin_root_id, diff --git a/src/core/plugin_manager/cpp/plugin.cpp b/src/core/plugin_manager/cpp/plugin.cpp index 600ea872ed..97bce0ba05 100644 --- a/src/core/plugin_manager/cpp/plugin.cpp +++ b/src/core/plugin_manager/cpp/plugin.cpp @@ -162,11 +162,26 @@ std::string child_launch_error_message(const ChildLaunchError& error, } // namespace #endif +void validate_plugin_shutdown_timeout(double seconds) +{ + // Written so that NaN fails too. + if (!(seconds > 0.0 && seconds <= MAX_PLUGIN_SHUTDOWN_TIMEOUT_SECONDS)) + { + std::ostringstream message; + message << "shutdown_timeout_seconds must be greater than 0 and at most " << MAX_PLUGIN_SHUTDOWN_TIMEOUT_SECONDS + << ", got " << seconds; + throw std::invalid_argument(message.str()); + } +} + Plugin::Plugin(const std::string& command, const std::string& working_dir, const std::string& plugin_root_id, - const std::vector& plugin_args) + const std::vector& plugin_args, + double shutdown_timeout_seconds) + : m_shutdown_timeout(shutdown_timeout_seconds) { + validate_plugin_shutdown_timeout(shutdown_timeout_seconds); start_process(command, working_dir, plugin_root_id, plugin_args); } @@ -459,6 +474,9 @@ void Plugin::start_process(const std::string& command, void Plugin::stop_process() { #ifndef _WIN32 + // A concurrent caller waits here for the stop in progress, then finds the process + // gone, instead of sending a second SIGINT into the plugin's cleanup. + std::lock_guard stop_lock(m_stop_mutex); std::unique_lock lock(m_process_mutex); if (m_pid == -1) { @@ -478,7 +496,8 @@ void Plugin::stop_process() throw PluginCrashException(m_process_snapshot.error); } - const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(2); + const auto deadline = std::chrono::steady_clock::now() + + std::chrono::duration_cast(m_shutdown_timeout); while (m_pid != -1 && std::chrono::steady_clock::now() < deadline) { refresh_process_snapshot_locked(false); @@ -496,6 +515,11 @@ void Plugin::stop_process() return; } + isaaccapture::Logger::get("isaaccapture.core.Plugin") + ->warn( + "Plugin process {} did not exit within {} s of SIGINT; sending SIGKILL. Raise " + "shutdown_timeout_seconds in its plugin.yaml if it needs longer to shut down.", + m_pid, m_shutdown_timeout.count()); if (kill(m_pid, SIGKILL) == -1) { cache_signal_error_locked(errno, "send SIGKILL to plugin process"); diff --git a/src/core/plugin_manager/cpp/plugin_manager.cpp b/src/core/plugin_manager/cpp/plugin_manager.cpp index 00283e6a27..3b186feafe 100644 --- a/src/core/plugin_manager/cpp/plugin_manager.cpp +++ b/src/core/plugin_manager/cpp/plugin_manager.cpp @@ -39,6 +39,11 @@ static PluginInfo parse_plugin_yaml(const std::string& path) info.args.push_back(arg_node.as()); } } + if (config["shutdown_timeout_seconds"]) + { + info.shutdown_timeout_seconds = config["shutdown_timeout_seconds"].as(); + validate_plugin_shutdown_timeout(info.shutdown_timeout_seconds); + } // Parse devices list if (config["devices"] && config["devices"].IsSequence()) @@ -171,7 +176,7 @@ std::unique_ptr PluginManager::start(const std::string& plugin_name, const auto& info = it->second; std::vector args = info.args; args.insert(args.end(), plugin_args.begin(), plugin_args.end()); - return std::make_unique(info.command, info.working_dir, plugin_root_id, args); + return std::make_unique(info.command, info.working_dir, plugin_root_id, args, info.shutdown_timeout_seconds); } } // namespace core diff --git a/src/core/plugin_manager/python/plugin_manager_bindings.cpp b/src/core/plugin_manager/python/plugin_manager_bindings.cpp index 3e8fcba078..10c3a0a0cb 100644 --- a/src/core/plugin_manager/python/plugin_manager_bindings.cpp +++ b/src/core/plugin_manager/python/plugin_manager_bindings.cpp @@ -9,6 +9,32 @@ namespace py = pybind11; using namespace core; +namespace +{ + +// ~Plugin() stops a running plugin and can wait up to its shutdown timeout, so a +// Python-owned Plugin is destroyed without the GIL. (pybind11 2.13 has no +// release_gil_before_calling_cpp_dtor.) +struct ReleaseGilDeleter +{ + void operator()(Plugin* plugin) const + { + if (PyGILState_Check()) + { + py::gil_scoped_release release; + delete plugin; + } + else + { + delete plugin; + } + } +}; + +using PythonPlugin = std::unique_ptr; + +} // namespace + PYBIND11_MODULE(_plugin_manager, m) { m.doc() = "Isaac Teleop Plugin Manager bindings"; @@ -52,6 +78,7 @@ PYBIND11_MODULE(_plugin_manager, m) .def_readonly("command", &PluginInfo::command) .def_readonly("version", &PluginInfo::version) .def_readonly("working_dir", &PluginInfo::working_dir) + .def_readonly("shutdown_timeout_seconds", &PluginInfo::shutdown_timeout_seconds) .def_property_readonly("args", [](const PluginInfo& info) { @@ -73,13 +100,20 @@ PYBIND11_MODULE(_plugin_manager, m) return devices; }); - py::class_>(m, "Plugin") - .def("stop", &Plugin::stop, "Explicitly stop the plugin (throws PluginCrashException if crashed)") + // stop() can wait up to the plugin's shutdown timeout, so it runs without the GIL. + py::class_(m, "Plugin") + .def("stop", &Plugin::stop, py::call_guard(), + "Explicitly stop the plugin (throws PluginCrashException if crashed)") .def("check_health", &Plugin::check_health, "Check if plugin has crashed (throws PluginCrashException if crashed)") .def("get_process_snapshot", &Plugin::get_process_snapshot, "Poll and return the cached process state without throwing process-health exceptions") .def("__enter__", [](Plugin* self) { return self; }) - .def("__exit__", [](Plugin* self, py::object, py::object, py::object) { self->stop(); }); + .def("__exit__", + [](Plugin* self, py::object, py::object, py::object) + { + py::gil_scoped_release release; + self->stop(); + }); py::class_(m, "PluginManager") .def(py::init&>(), py::arg("search_paths"), @@ -89,7 +123,11 @@ PYBIND11_MODULE(_plugin_manager, m) "Get the complete immutable descriptor for a discovered plugin") .def("query_devices", &PluginManager::query_devices, py::arg("plugin_name"), "Query available devices from a plugin") - .def("start", &PluginManager::start, py::arg("plugin_name"), py::arg("plugin_root_id"), - py::arg("plugin_args") = std::vector{}, - "Start a plugin and return a RAII handle. plugin_args are appended after plugin.yaml args."); + .def( + "start", + [](PluginManager& self, const std::string& plugin_name, const std::string& plugin_root_id, + const std::vector& plugin_args) + { return PythonPlugin(self.start(plugin_name, plugin_root_id, plugin_args).release()); }, + py::arg("plugin_name"), py::arg("plugin_root_id"), py::arg("plugin_args") = std::vector{}, + "Start a plugin and return a RAII handle. plugin_args are appended after plugin.yaml args."); } diff --git a/tests/cpp/core/plugin_manager/temporary_directory.hpp b/tests/cpp/core/plugin_manager/temporary_directory.hpp new file mode 100644 index 0000000000..5d9bf4f0f4 --- /dev/null +++ b/tests/cpp/core/plugin_manager/temporary_directory.hpp @@ -0,0 +1,46 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#pragma once + +#include +#include +#include +#include +#include + +class TemporaryDirectory +{ +public: + TemporaryDirectory() + { + const auto seed = std::chrono::steady_clock::now().time_since_epoch().count(); + for (int suffix = 0; suffix < 100; ++suffix) + { + m_path = std::filesystem::temp_directory_path() / + ("isaacteleop_plugin_manager_" + std::to_string(seed) + "_" + std::to_string(suffix)); + if (std::filesystem::create_directory(m_path)) + { + return; + } + } + throw std::runtime_error("Failed to create temporary test directory"); + } + + ~TemporaryDirectory() + { + std::error_code error; + std::filesystem::remove_all(m_path, error); + } + + TemporaryDirectory(const TemporaryDirectory&) = delete; + TemporaryDirectory& operator=(const TemporaryDirectory&) = delete; + + const std::filesystem::path& path() const + { + return m_path; + } + +private: + std::filesystem::path m_path; +}; diff --git a/tests/cpp/core/plugin_manager/test_plugin_manager.cpp b/tests/cpp/core/plugin_manager/test_plugin_manager.cpp index 1bb732daa2..daf3782c0e 100644 --- a/tests/cpp/core/plugin_manager/test_plugin_manager.cpp +++ b/tests/cpp/core/plugin_manager/test_plugin_manager.cpp @@ -1,12 +1,15 @@ // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 +#include "temporary_directory.hpp" + #include #include -#include +#include #include #include +#include #include #include #include @@ -14,38 +17,15 @@ namespace { -class TemporaryDirectory +void write_metadata(const std::filesystem::path& search_directory, const std::string& extra_lines) { -public: - TemporaryDirectory() - { - const auto seed = std::chrono::steady_clock::now().time_since_epoch().count(); - for (int suffix = 0; suffix < 100; ++suffix) - { - m_path = std::filesystem::temp_directory_path() / - ("isaacteleop_plugin_manager_" + std::to_string(seed) + "_" + std::to_string(suffix)); - if (std::filesystem::create_directory(m_path)) - { - return; - } - } - throw std::runtime_error("Failed to create temporary test directory"); - } - - ~TemporaryDirectory() - { - std::error_code error; - std::filesystem::remove_all(m_path, error); - } - - const std::filesystem::path& path() const - { - return m_path; - } - -private: - std::filesystem::path m_path; -}; + const std::filesystem::path plugin_directory = search_directory / "sample"; + std::filesystem::create_directory(plugin_directory); + std::ofstream metadata(plugin_directory / "plugin.yaml"); + metadata << "name: sample_plugin\n" + "command: /bin/true\n" + << extra_lines; +} } // namespace @@ -87,5 +67,41 @@ TEST_CASE("plugin manager returns complete descriptors", "[plugin_manager][metad REQUIRE(info.devices[0].description == "Left hand"); REQUIRE(info.devices[1].path == "/camera"); REQUIRE((manager.query_devices("sample_plugin") == std::vector{ "/hands/left", "/camera" })); + REQUIRE(info.shutdown_timeout_seconds == core::DEFAULT_PLUGIN_SHUTDOWN_TIMEOUT_SECONDS); REQUIRE_THROWS_AS(manager.get_plugin_info("missing"), std::runtime_error); } + +TEST_CASE("plugin.yaml sets the plugin's shutdown timeout", "[plugin_manager][metadata]") +{ + SECTION("a valid value is reported") + { + TemporaryDirectory search_directory; + write_metadata(search_directory.path(), "shutdown_timeout_seconds: 15.5\n"); + + const core::PluginManager manager({ search_directory.path().string() }); + REQUIRE(manager.get_plugin_info("sample_plugin").shutdown_timeout_seconds == 15.5); + } + + SECTION("an invalid value skips the plugin") + { + for (const std::string value : { "0", "-3", ".nan", ".inf", "3601", "soon" }) + { + CAPTURE(value); + TemporaryDirectory search_directory; + write_metadata(search_directory.path(), "shutdown_timeout_seconds: " + value + "\n"); + + const core::PluginManager manager({ search_directory.path().string() }); + REQUIRE(manager.get_plugin_names().empty()); + } + } +} + +TEST_CASE("plugin rejects an out-of-range shutdown timeout before launching", "[plugin_manager]") +{ + for (const double seconds : { 0.0, -1.0, std::nan(""), std::numeric_limits::infinity(), + core::MAX_PLUGIN_SHUTDOWN_TIMEOUT_SECONDS + 1.0 }) + { + CAPTURE(seconds); + REQUIRE_THROWS_AS(core::Plugin("never-launched", "", "test-root", {}, seconds), std::invalid_argument); + } +} diff --git a/tests/cpp/core/plugin_manager/test_plugin_process.cpp b/tests/cpp/core/plugin_manager/test_plugin_process.cpp index 654e5da5d2..86de3a553a 100644 --- a/tests/cpp/core/plugin_manager/test_plugin_process.cpp +++ b/tests/cpp/core/plugin_manager/test_plugin_process.cpp @@ -1,11 +1,17 @@ // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 +#include "temporary_directory.hpp" + #include #include +#include #include #include +#include +#include +#include #include #include #include @@ -145,6 +151,70 @@ TEST_CASE("explicit stop is cached and non-failing", "[plugin_manager][process]" REQUIRE(second.reason == first.reason); } +TEST_CASE("stop uses the shutdown timeout from plugin.yaml", "[plugin_manager][process]") +{ + TemporaryDirectory search_directory; + const std::filesystem::path plugin_directory = search_directory.path() / "finalizing"; + std::filesystem::create_directory(plugin_directory); + const std::filesystem::path marker = search_directory.path() / "finalized"; + + const auto start = [&](const std::string& timeout_seconds, int finalize_milliseconds) + { + std::ofstream metadata(plugin_directory / "plugin.yaml"); + metadata << "name: finalizing_plugin\n" + << "command: \"" << PLUGIN_MANAGER_TEST_PROCESS << "\"\n" + << "shutdown_timeout_seconds: " << timeout_seconds << "\n"; + metadata.close(); + core::PluginManager manager({ search_directory.path().string() }); + auto plugin = manager.start( + "finalizing_plugin", "test-root", { "finalize", marker.string(), std::to_string(finalize_milliseconds) }); + + const std::filesystem::path ready = marker.string() + ".ready"; + const auto deadline = std::chrono::steady_clock::now() + PROCESS_TIMEOUT; + while (!std::filesystem::exists(ready) && std::chrono::steady_clock::now() < deadline) + { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + REQUIRE(std::filesystem::exists(ready)); + return plugin; + }; + + SECTION("a plugin that finishes within its timeout is not killed") + { + const std::unique_ptr plugin = start("5", 300); + REQUIRE_NOTHROW(plugin->stop()); + REQUIRE(std::filesystem::exists(marker)); + REQUIRE(plugin->get_process_snapshot().state == core::ProcessState::STOPPED); + } + + SECTION("a concurrent stop waits for the shutdown already in progress") + { + const std::unique_ptr plugin = start("5", 500); + std::thread second_caller( + [&] + { + // Let the first stop's SIGINT reach the plugin before this one starts. + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + plugin->stop(); + }); + REQUIRE_NOTHROW(plugin->stop()); + second_caller.join(); + REQUIRE(std::filesystem::exists(marker)); + REQUIRE(plugin->get_process_snapshot().state == core::ProcessState::STOPPED); + } + + SECTION("a plugin that outlives its timeout is killed when it expires") + { + const std::unique_ptr plugin = start("0.2", 5000); + const auto stop_started = std::chrono::steady_clock::now(); + REQUIRE_NOTHROW(plugin->stop()); + // Well below the 2 s default, so the manifest value was used. + REQUIRE(std::chrono::steady_clock::now() - stop_started < std::chrono::milliseconds(1500)); + REQUIRE_FALSE(std::filesystem::exists(marker)); + REQUIRE(plugin->get_process_snapshot().state == core::ProcessState::STOPPED); + } +} + TEST_CASE("pre-exec failures are reported by the parent", "[plugin_manager][process]") { SECTION("working directory failure") diff --git a/tests/cpp/core/plugin_manager/test_process/test_process.cpp b/tests/cpp/core/plugin_manager/test_process/test_process.cpp index c553f139b6..66de0a1c89 100644 --- a/tests/cpp/core/plugin_manager/test_process/test_process.cpp +++ b/tests/cpp/core/plugin_manager/test_process/test_process.cpp @@ -4,6 +4,7 @@ #include #include #include +#include #include #include #include @@ -12,11 +13,12 @@ namespace { +// Counts SIGINTs; only the signal handler writes it. volatile std::sig_atomic_t stop_requested = 0; void request_stop(int) { - stop_requested = 1; + stop_requested = stop_requested + 1; } std::vector plugin_arguments(int argc, char** argv) @@ -58,6 +60,30 @@ int main(int argc, char** argv) return 2; } + // finalize : write .ready once SIGINT is handled; after SIGINT, + // keep working for , then write . A second SIGINT abandons the work, as a + // second Ctrl-C does in many plugins. + if (arguments[0] == "finalize") + { + std::signal(SIGINT, request_stop); + std::ofstream(arguments[1] + ".ready") << "ready\n"; + while (!stop_requested) + { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + const auto done = std::chrono::steady_clock::now() + std::chrono::milliseconds(std::stoi(arguments[2])); + while (std::chrono::steady_clock::now() < done) + { + if (stop_requested > 1) + { + return 4; + } + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + std::ofstream(arguments[1]) << "finalized\n"; + return 0; + } + std::this_thread::sleep_for(std::chrono::milliseconds(std::stoi(arguments[2]))); if (arguments[0] == "exit") {