diff --git a/src/InputStream.h b/src/InputStream.h index 7225f32bd..339f9b9df 100644 --- a/src/InputStream.h +++ b/src/InputStream.h @@ -50,10 +50,16 @@ class InputStream : public AbstractRunnableModule } } + // a no-op for a policy that is not attached, so detaching twice, as a failed load's rollback may after + // stop(), never signals a removal for a policy the input no longer counts void remove_policy(const Policy *policy) { std::unique_lock lock(_input_mutex); - _policies.erase(std::remove(_policies.begin(), _policies.end(), policy), _policies.end()); + auto removed = std::remove(_policies.begin(), _policies.end(), policy); + if (removed == _policies.end()) { + return; + } + _policies.erase(removed, _policies.end()); for (auto const &proxy : _event_proxies) { proxy->policy_cb(policy, Action::RemovePolicy); } diff --git a/src/Policies.cpp b/src/Policies.cpp index 021ed23a3..360584e6a 100644 --- a/src/Policies.cpp +++ b/src/Policies.cpp @@ -83,6 +83,63 @@ std::vector PolicyManager::load(const YAML::Node &policy_yaml, bool si std::vector added_inputs; std::vector added_handlers; + // by name, like the resources policies below: once the policy is detached from an input, nothing keeps + // that input alive, and a concurrent removal of its last policy can free it while this load runs + std::vector attached_inputs; + std::vector> resources_policies; + // by name: once registered, a resources policy can be deleted through the API while this load runs + std::vector added_resources_policies; + // Undo everything this load did, one step at a time, so that a step that fails neither skips the rest + // nor replaces the exception that started the rollback. stop() undoes a completed start; after a + // partial one, Policy::start has already stopped the handlers, and the input manager stops the inputs + // it removes. The policy is detached from the inputs it was attached to: a reused input outlives this + // load and must not keep counting it, and an input it reached but never joined is left alone. + auto rollback = [&]() { + auto step = [&policy_name](auto &&undo) { + try { + undo(); + } catch (const std::exception &e) { + spdlog::get("visor")->warn("policy [{}]: rollback: {}", policy_name, e.what()); + } + }; + step([&] { policy_ptr->stop(); }); + for (auto &name : attached_inputs) { + step([&] { + // under the policy map lock, as remove_policy does: no other removal can free the input + // meanwhile. one that is already gone was freed after stop() detached the policy + std::unique_lock policies_lock(_map_mutex); + InputStream *input = nullptr; + { + auto [inputs, inputs_lock] = _registry->input_manager()->module_get_all_locked(); + auto found = inputs.find(name); + if (found == inputs.end()) { + return; + } + input = found->second.get(); + } + input->remove_policy(policy_ptr); + // a reused input outlives this load. If its last other policy was removed while this one + // still counted on it, that removal left it running, so finish the teardown here + if (std::find(added_inputs.begin(), added_inputs.end(), name) == added_inputs.end()) { + _remove_input_if_unused(name, input, policy_name); + } + }); + } + for (auto &name : added_resources_policies) { + step([&] { + auto [resources_policy, lock] = module_get_locked(name); + resources_policy->stop(); + }); + // unregistered even if it did not stop, since its input is removed below + step([&] { module_remove(name); }); + } + for (auto &m : added_handlers) { + step([&] { _registry->handler_manager()->module_remove(m); }); + } + for (auto &p : added_inputs) { + step([&] { _registry->input_manager()->module_remove(p); }); + } + }; try { for (const auto &tap_name : taps_name) { auto [tap, tap_lock] = _registry->tap_manager()->module_get_locked(tap_name); @@ -137,21 +194,15 @@ std::vector PolicyManager::load(const YAML::Node &policy_yaml, bool si // success input_ptr->add_policy(policy_ptr); + attached_inputs.push_back(input_stream_name); } - } catch (std::runtime_error &e) { + } catch (...) { // failed to create policy - for (auto &m : added_handlers) { - _registry->handler_manager()->module_remove(m); - } - for (auto &p : added_inputs) { - _registry->input_manager()->module_remove(p); - } + rollback(); throw; } - std::vector> resources_policies; - std::vector added_resources_policies; try { // every handler on a new input must be running before the input starts: an input that reads all // its data as it starts, such as a pcap_file or dnstap_file, is consumed by whichever policy starts @@ -167,26 +218,27 @@ std::vector PolicyManager::load(const YAML::Node &policy_yaml, bool si } policy_ptr->start(); for (auto &resources_policy : resources_policies) { - auto resources_policy_ptr = resources_policy.get(); - resources_policy_ptr->start(); - module_add(std::move(resources_policy)); - added_resources_policies.push_back(resources_policy_ptr->name()); - resources_policy_ptr->input_stream().back()->add_policy(resources_policy_ptr); + // attached before it is registered, and not touched after: from then on the API can delete it + auto resources_name = resources_policy->name(); + resources_policy->start(); + resources_policy->input_stream().back()->add_policy(resources_policy.get()); + try { + module_add(std::move(resources_policy)); + } catch (...) { + // not registered, so still ours: stopping it also detaches it from its input + resources_policy->stop(); + throw; + } + added_resources_policies.push_back(resources_name); } module_add(std::move(policy)); - } catch (std::runtime_error &e) { - // undoes a completed start; Policy::start undoes a partial one itself - policy_ptr->stop(); - for (auto &p : added_resources_policies) { - module_remove(p); - } - for (auto &m : added_handlers) { - _registry->handler_manager()->module_remove(m); - } - for (auto &p : added_inputs) { - _registry->input_manager()->module_remove(p); - } + } catch (const std::runtime_error &e) { + rollback(); throw PolicyException(fmt::format("policy [{}] failed to start: {}", policy_name, e.what())); + } catch (...) { + // anything else, such as a std::invalid_argument, is rethrown as it was so callers still tell it apart + rollback(); + throw; } result.push_back(policy_ptr); @@ -273,23 +325,28 @@ void PolicyManager::remove_policy(const std::string &name) } for (const auto &input : input_stream) { - if (input.second->policies_count() == 1) { - // if there is only one policy left on the input stream, and that policy is the input resources policy, then remove it - auto input_resources_name = input.first + "-resources"; - if (name != input_resources_name && _map.count(input_resources_name) != 0) { - auto resources_policy = _map[input_resources_name].get(); - resources_policy->stop(); - _registry->handler_manager()->module_remove(input_resources_name); - _map.erase(input_resources_name); - } - } + _remove_input_if_unused(input.first, input.second, name); + } - if (!input.second->policies_count()) { - _registry->input_manager()->module_remove(input.first); + _map.erase(name); +} + +void PolicyManager::_remove_input_if_unused(const std::string &input_name, InputStream *input, const std::string &leaving_policy) +{ + if (input->policies_count() == 1) { + // if there is only one policy left on the input stream, and that policy is the input resources policy, then remove it + auto input_resources_name = input_name + "-resources"; + if (leaving_policy != input_resources_name && _map.count(input_resources_name) != 0) { + auto resources_policy = _map[input_resources_name].get(); + resources_policy->stop(); + _registry->handler_manager()->module_remove(input_resources_name); + _map.erase(input_resources_name); } } - _map.erase(name); + if (!input->policies_count()) { + _registry->input_manager()->module_remove(input_name); + } } void Policy::info_json(json &j) const { diff --git a/src/Policies.h b/src/Policies.h index 507e9416f..0c35a250e 100644 --- a/src/Policies.h +++ b/src/Policies.h @@ -138,6 +138,9 @@ class PolicyManager : public AbstractManager CoreRegistry *_registry; std::string _get_policy_name(YAML::const_iterator it); + // once leaving_policy is off input: removes the input's resources policy if it is the only one left, then + // the input if none is. the caller holds _map_mutex, which keeps any other removal from freeing the input + void _remove_input_if_unused(const std::string &input_name, InputStream *input, const std::string &leaving_policy); public: PolicyManager(CoreRegistry *registry) diff --git a/src/inputs/mock/MockInputStream.cpp b/src/inputs/mock/MockInputStream.cpp index bc88e7b75..1a5196b40 100644 --- a/src/inputs/mock/MockInputStream.cpp +++ b/src/inputs/mock/MockInputStream.cpp @@ -31,6 +31,9 @@ void MockInputStream::start() if (config_exists("except_on_start")) { throw std::runtime_error("mock error on start"); } + if (config_exists("except_on_start_logic")) { + throw std::logic_error("mock logic error on start"); + } static timer timer_thread{500ms}; std::srand(std::time(nullptr)); diff --git a/src/tests/test_policies.cpp b/src/tests/test_policies.cpp index 8a66ed6bc..3e62e57c1 100644 --- a/src/tests/test_policies.cpp +++ b/src/tests/test_policies.cpp @@ -12,6 +12,11 @@ #include #include +#include +#include +#include +#include +#include #include #include #include @@ -842,6 +847,74 @@ version: "1.0" type: net )"; +auto policies_config_shared_input = R"( +version: "1.0" + +visor: + taps: + live_a: + input_type: pcap + config: + pcap_source: mock + tags: + group: g1 + broken_b: + input_type: pcap + config: + pcap_file: tests/fixtures/does_not_exist.pcap + tags: + group: g1 + policies: + first_view: + kind: collection + input: + tap: live_a + input_type: pcap + handlers: + modules: + first_net: + type: net +)"; + +auto policies_config_shared_input_failing = R"( +version: "1.0" + +visor: + policies: + second_view: + kind: collection + input: + tap_selector: + any: + - group: g1 + input_type: pcap + handlers: + modules: + second_net: + type: net +)"; + +auto policies_config_logic_error_on_start = R"( +visor: + taps: + anycast: + input_type: mock + config: + iface: eth0 + policies: + default_view: + kind: collection + input: + tap: anycast + input_type: mock + config: + except_on_start_logic: true + handlers: + modules: + default_net: + type: net +)"; + TEST_CASE("Policies", "[policies]") { @@ -1100,6 +1173,112 @@ TEST_CASE("Policies", "[policies]") CHECK_FALSE(registry.handler_manager()->module_exists("broken_view-broken-broken_net")); } + SECTION("Bad Config: a failed load leaves no reference on a shared input") + { + CoreRegistry registry; + visor::load_builtin_plugins(registry); + registry.start(nullptr); + YAML::Node config_file = YAML::Load(policies_config_shared_input); + REQUIRE_NOTHROW(registry.tap_manager()->load(config_file["visor"]["taps"])); + REQUIRE_NOTHROW(registry.policy_manager()->load(config_file["visor"]["policies"])); + + std::string input_name; + size_t policies_before; + { + auto [policy, lock] = registry.policy_manager()->module_get_locked("first_view"); + input_name = policy->input_stream().back()->name(); + policies_before = policy->input_stream().back()->policies_count(); + } + + // second_view reuses first_view's input and adds one that fails to start, so it fails before it has + // started; it must still be detached from the input it shares + YAML::Node failing = YAML::Load(policies_config_shared_input_failing); + REQUIRE_THROWS_WITH(registry.policy_manager()->load(failing["visor"]["policies"]), "policy [second_view] failed to start: Cannot open pcap/pcapng file"); + { + auto [input, lock] = registry.input_manager()->module_get_locked(input_name); + CHECK(input->policies_count() == policies_before); + } + CHECK_FALSE(registry.handler_manager()->module_exists("second_view-live_a-second_net")); + CHECK_FALSE(registry.handler_manager()->module_exists("second_view-broken_b-second_net")); + { + auto [inputs, lock] = registry.input_manager()->module_get_all_locked(); + CHECK(inputs.size() == 1); + } + + // so removing the last policy on that input still tears it down + REQUIRE_NOTHROW(registry.policy_manager()->remove_policy("first_view")); + CHECK_FALSE(registry.input_manager()->module_exists(input_name)); + } + + SECTION("Bad Config: an exception that is not a runtime_error is rolled back too") + { + CoreRegistry registry; + visor::load_builtin_plugins(registry); + registry.start(nullptr); + + REQUIRE_NOTHROW(registry.tap_manager()->load_from_str(policies_config_logic_error_on_start)); + // rethrown as it was, so the API still answers it as before; and rolled back, so a retry fails the + // same way rather than as a name clash + for (int attempt = 0; attempt < 2; ++attempt) { + REQUIRE_THROWS_AS(registry.policy_manager()->load_from_str(policies_config_logic_error_on_start), std::logic_error); + } + CHECK_FALSE(registry.policy_manager()->module_exists("default_view")); + CHECK_FALSE(registry.handler_manager()->module_exists("default_view-anycast-default_net")); + auto [inputs, lock] = registry.input_manager()->module_get_all_locked(); + CHECK(inputs.empty()); + } + + SECTION("Bad Config: a resources policy that cannot be registered is rolled back") + { + CoreRegistry registry; + visor::load_builtin_plugins(registry); + registry.start(nullptr); + REQUIRE_NOTHROW(registry.tap_manager()->load_from_str(policies_config_logic_error_on_start)); + auto policy_on = [](const std::string &name, const std::string &sample) { + return fmt::format(R"( +visor: + policies: + {0}: + kind: collection + input: + tap: anycast + input_type: mock + config: + sample: {1} + handlers: + modules: + {0}_net: + type: net +)", + name, sample); + }; + + // learn the input clash_view will get, then free it + REQUIRE_NOTHROW(registry.policy_manager()->load_from_str(policy_on("clash_view", "one"))); + std::string input_name; + { + auto [policy, lock] = registry.policy_manager()->module_get_locked("clash_view"); + input_name = policy->input_stream().back()->name(); + } + REQUIRE_NOTHROW(registry.policy_manager()->remove_policy("clash_view")); + + // a policy on another input takes the name clash_view's resources policy needs, so its registration + // fails after the policy and its input have started + auto taken = input_name + "-resources"; + REQUIRE_NOTHROW(registry.policy_manager()->load_from_str(policy_on(taken, "two"))); + REQUIRE_THROWS_WITH(registry.policy_manager()->load_from_str(policy_on("clash_view", "one")), + fmt::format("policy [clash_view] failed to start: module name '{}' already exists", taken)); + + // the rollback removes what the load made, and leaves the policy that owns the name alone + CHECK_FALSE(registry.policy_manager()->module_exists("clash_view")); + CHECK_FALSE(registry.input_manager()->module_exists(input_name)); + CHECK_FALSE(registry.handler_manager()->module_exists("clash_view-anycast-clash_view_net")); + CHECK_FALSE(registry.handler_manager()->module_exists(taken)); + REQUIRE(registry.policy_manager()->module_exists(taken)); + auto [owner, lock] = registry.policy_manager()->module_get_locked(taken); + CHECK(owner->running()); + } + SECTION("Bad Config: mis-matched input_type on tap") { CoreRegistry registry; @@ -1683,3 +1862,82 @@ TEST_CASE("Multi-policy Prometheus shared serializer", "[policies][prometheus][m CHECK(count_occurrences(output, "policy=\"policy_beta\"") >= 1); } } + +TEST_CASE("Loading and removing policies concurrently does not deadlock", "[policies][concurrency]") +{ + CoreRegistry registry; + visor::load_builtin_plugins(registry); + registry.start(nullptr); + REQUIRE_NOTHROW(registry.tap_manager()->load_from_str(R"( +visor: + taps: + anycast: + input_type: mock + config: + iface: eth0 +)")); + + // every iteration logs several lines; keep this test's output to warnings + struct QuietLogger { + std::shared_ptr logger = spdlog::get("visor"); + spdlog::level::level_enum level = logger->level(); + QuietLogger() + { + logger->set_level(spdlog::level::warn); + } + ~QuietLogger() + { + logger->set_level(level); + } + } quiet; + + // every policy gets an input of its own, so each load creates an input and its resources policy and each + // removal tears both down: load and remove_policy then take the input and policy maps on every iteration, + // and a load runs while the other thread removes. A slow build stops early; the deadlock this guards + // against showed up within 2000 iterations. + auto stop_at = std::chrono::steady_clock::now() + std::chrono::seconds(15); + auto churn = [®istry, stop_at](const std::string &prefix) { + for (int i = 0; i < 5000 && std::chrono::steady_clock::now() < stop_at; ++i) { + auto name = fmt::format("{}_{}", prefix, i); + registry.policy_manager()->load_from_str(fmt::format(R"( +visor: + policies: + {0}: + kind: collection + input: + tap: anycast + input_type: mock + config: + sample: {0} + handlers: + modules: + {0}_net: + type: net +)", + name)); + registry.policy_manager()->remove_policy(name); + } + }; + + auto a = std::async(std::launch::async, churn, std::string("a")); + auto b = std::async(std::launch::async, churn, std::string("b")); + // well past the time box, so it only fires on a hang + auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(180); + if (a.wait_until(deadline) == std::future_status::timeout || b.wait_until(deadline) == std::future_status::timeout) { + // the stuck threads can never be joined, so fail loudly instead of hanging the suite + std::fprintf(stderr, "load and remove_policy did not finish within 180 s, most likely a deadlock\n"); + std::abort(); + } + REQUIRE_NOTHROW(a.get()); + REQUIRE_NOTHROW(b.get()); + { + auto [policies, lock] = registry.policy_manager()->module_get_all_locked(); + CHECK(policies.empty()); + } + { + auto [handlers, lock] = registry.handler_manager()->module_get_all_locked(); + CHECK(handlers.empty()); + } + auto [inputs, lock] = registry.input_manager()->module_get_all_locked(); + CHECK(inputs.empty()); +}