diff --git a/src/ray/common/BUILD.bazel b/src/ray/common/BUILD.bazel index ae2113296ebc..700318572aea 100644 --- a/src/ray/common/BUILD.bazel +++ b/src/ray/common/BUILD.bazel @@ -228,7 +228,7 @@ ray_cc_library( name = "memory_monitor_factory", srcs = select({ "//bazel:is_linux": [ - "threshold_memory_monitor_factory.cc", + "multi_memory_monitor_factory.cc", ], "//conditions:default": [ "noop_memory_monitor_factory.cc", @@ -243,16 +243,18 @@ ray_cc_library( ], deps = [ ":memory_monitor_interface", - ":noop_memory_monitor", "//src/ray/common/cgroup2:cgroup_manager_interface", ] + select({ "//bazel:is_linux": [ ":memory_monitor_utils", + ":noop_memory_monitor", ":ray_config", ":threshold_memory_monitor", "//src/ray/util:logging", ], - "//conditions:default": [], + "//conditions:default": [ + ":noop_memory_monitor", + ], }), ) diff --git a/src/ray/common/cgroup2/linux_cgroup_manager_factory.cc b/src/ray/common/cgroup2/linux_cgroup_manager_factory.cc index 060b3ef8a357..9c71915e07a5 100644 --- a/src/ray/common/cgroup2/linux_cgroup_manager_factory.cc +++ b/src/ray/common/cgroup2/linux_cgroup_manager_factory.cc @@ -14,6 +14,7 @@ #include #include +#include #include #include #include @@ -66,11 +67,14 @@ std::unique_ptr CgroupManagerFactory::Create( float user_memory_proportion_high = RayConfig::instance().user_memory_proportion_high(); float user_memory_proportion_max = RayConfig::instance().user_memory_proportion_max(); - int64_t user_memory_high_bytes = - static_cast(total_memory_bytes * user_memory_proportion_high); - int64_t user_memory_max_bytes = std::min( + // The system reserved memory here already includes object store memory when + // we resolved the resource isolation config. + int64_t user_memory_high_bytes = std::min( total_memory_bytes - system_reserved_memory_bytes + object_store_memory_bytes, - static_cast(total_memory_bytes * user_memory_proportion_max)); + static_cast(total_memory_bytes * user_memory_proportion_high)); + int64_t user_memory_max_bytes = + static_cast(total_memory_bytes * user_memory_proportion_max); + StatusOr> cgroup_manager_s = CgroupManager::Create(cgroup_path, node_id, diff --git a/src/ray/common/event_memory_monitor.cc b/src/ray/common/event_memory_monitor.cc index c108f67f5e3e..45e19bddc695 100644 --- a/src/ray/common/event_memory_monitor.cc +++ b/src/ray/common/event_memory_monitor.cc @@ -192,8 +192,7 @@ void EventMemoryMonitor::MonitoringThreadMain() { if (high_modified && IsEnabled()) { Disable(); - kill_workers_callback_( - MemoryMonitorUtils::TakeSystemMemorySnapshot(cgroup_path_)); + kill_workers_callback_(); } } else { RAY_LOG(ERROR) << absl::StrFormat( diff --git a/src/ray/common/memory_monitor_factory.h b/src/ray/common/memory_monitor_factory.h index b401622c4e23..1ed45adcae03 100644 --- a/src/ray/common/memory_monitor_factory.h +++ b/src/ray/common/memory_monitor_factory.h @@ -15,38 +15,32 @@ #pragma once #include +#include #include "ray/common/cgroup2/cgroup_manager_interface.h" #include "ray/common/memory_monitor_interface.h" namespace ray { -/// Factory class for creating MemoryMonitor instances. -/// -/// This feature is only enabled on Linux. On Linux, it creates a ThresholdMemoryMonitor -/// that monitors memory usage using /proc filesystem and cgroups. -/// -/// On non-Linux platforms, this will return a no-op implementation. class MemoryMonitorFactory { public: /** - * Create a memory monitor instance. + * On Linux, creates monitors based on configuration: + * - Resource isolation disabled: ThresholdMemoryMonitor only. + * - Resource isolation enabled, ThresholdMemoryMonitor + EventMemoryMonitor. * - * On Linux, creates a ThresholdMemoryMonitor that monitors memory usage - * and triggers the callback when usage is refreshed. + * On non-Linux platforms, returns a vector with a single NoopMemoryMonitor. * - * On non-Linux platforms, creates a NoopMemoryMonitor that does nothing. - * - * @param kill_workers_callback function to execute when the memory usage is refreshed. + * @param kill_workers_callback function to invoke when memory pressure is detected. * @param resource_isolation_enabled When resource isolation is enabled, the - * memory monitor will work with the configured cgroup constraints to better + * memory monitors will work with the configured cgroup constraints to better * enforce the memory usage limit. - * @param cgroup_manager When resource isolation is enabled, the monitor will determine + * @param cgroup_manager When resource isolation is enabled, the monitors will determine * the proper memory monitoring threshold based on the set cgroup constraints provided * by the cgroup manager. - * @return a unique pointer to the memory monitor instance. + * @return a vector of memory monitor instances. */ - static std::unique_ptr Create( + static std::vector> Create( KillWorkersCallback kill_workers_callback, bool resource_isolation_enabled, const CgroupManagerInterface &cgroup_manager); diff --git a/src/ray/common/memory_monitor_interface.h b/src/ray/common/memory_monitor_interface.h index df3c0a92531a..2aad439344a6 100644 --- a/src/ray/common/memory_monitor_interface.h +++ b/src/ray/common/memory_monitor_interface.h @@ -48,11 +48,9 @@ struct SystemMemorySnapshot { using ProcessesMemorySnapshot = absl::flat_hash_map; /** - * @brief Callback that runs at each monitoring interval. - * - * \param system_memory snapshot of system memory information. + * @brief Callback to trigger worker oom killing when under memory pressure. */ -using KillWorkersCallback = std::function; +using KillWorkersCallback = std::function; /** * @brief implementations of this interface monitors the memory usage of the node diff --git a/src/ray/common/memory_monitor_utils.cc b/src/ray/common/memory_monitor_utils.cc index e7ed0d218803..ae4a4c17118a 100644 --- a/src/ray/common/memory_monitor_utils.cc +++ b/src/ray/common/memory_monitor_utils.cc @@ -318,30 +318,26 @@ int64_t MemoryMonitorUtils::GetMemoryThreshold( } if (resource_isolation_enabled) { - StatusOr user_memory_max_bytes_or = - cgroup_manager.GetUserCgroupConstraintValue("memory.max"); - RAY_CHECK(user_memory_max_bytes_or.ok()) << absl::StrFormat( - "Failed to get user cgroup memory limit when setting up memory monitor: %s", - user_memory_max_bytes_or.ToString()); - std::string user_memory_max_bytes_str = user_memory_max_bytes_or.value(); - - if (!user_memory_max_bytes_str.empty() && - std::all_of(user_memory_max_bytes_str.begin(), - user_memory_max_bytes_str.end(), + StatusOr user_slice_upper_bound_bytes_or = + cgroup_manager.GetUserCgroupConstraintValue("memory.high"); + RAY_CHECK(user_slice_upper_bound_bytes_or.ok()) << absl::StrFormat( + "Failed to get user cgroup memory limit from user cgroup %s " + "when setting up memory monitor: %s. " + "Does the cgroup path exist and/or matches the resource isolation hierarchy?", + cgroup_manager.GetUserCgroupPath(), + user_slice_upper_bound_bytes_or.ToString()); + std::string user_slice_upper_bound_bytes_str = + user_slice_upper_bound_bytes_or.value(); + RAY_CHECK(!user_slice_upper_bound_bytes_str.empty()) << absl::StrFormat( + "Failed to get upper bound memory constraints from user cgroup %s. " + "Does the cgroup path exist and/or matches the resource isolation hierarchy?", + cgroup_manager.GetUserCgroupPath()); + + if (!user_slice_upper_bound_bytes_str.empty() && + std::all_of(user_slice_upper_bound_bytes_str.begin(), + user_slice_upper_bound_bytes_str.end(), ::isdigit)) { - int64_t user_memory_max_bytes = std::stoll(user_memory_max_bytes_str); - int64_t reaction_buffer_bytes = - std::min(static_cast(total_memory_bytes * - kDefaultThresholdMonitorReactionBufferProportion), - RayConfig::instance().max_threshold_monitor_reaction_buffer_bytes()); - resolved_memory_threshold_bytes = user_memory_max_bytes - reaction_buffer_bytes; - RAY_CHECK_GE(resolved_memory_threshold_bytes, 0) << absl::StrFormat( - "Available user task memory is less than the kill memory buffer bytes: " - "%d < %d. This means the available memory for user proceses is likely " - "less than 5%% of total memory. Please consider decreasing the proportion " - "of reserved system memory if it was custom set.", - user_memory_max_bytes, - reaction_buffer_bytes); + resolved_memory_threshold_bytes = std::stoll(user_slice_upper_bound_bytes_str); } } diff --git a/src/ray/common/threshold_memory_monitor_factory.cc b/src/ray/common/multi_memory_monitor_factory.cc similarity index 68% rename from src/ray/common/threshold_memory_monitor_factory.cc rename to src/ray/common/multi_memory_monitor_factory.cc index 7c3fca8a514c..8018b78238ee 100644 --- a/src/ray/common/threshold_memory_monitor_factory.cc +++ b/src/ray/common/multi_memory_monitor_factory.cc @@ -1,4 +1,4 @@ -// Copyright 2025 The Ray Authors. +// Copyright 2026 The Ray Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -12,8 +12,8 @@ // See the License for the specific language governing permissions and // limitations under the License. -#include #include +#include #include "ray/common/memory_monitor_factory.h" #include "ray/common/memory_monitor_interface.h" @@ -25,32 +25,35 @@ namespace ray { -std::unique_ptr MemoryMonitorFactory::Create( +std::vector> MemoryMonitorFactory::Create( KillWorkersCallback kill_workers_callback, bool resource_isolation_enabled, const CgroupManagerInterface &cgroup_manager) { - int64_t memory_usage_threshold_bytes; + std::vector> monitors; uint64_t monitor_interval_ms = RayConfig::instance().memory_monitor_refresh_ms(); - if (monitor_interval_ms <= 0) { - RAY_LOG(INFO) << "MemoryMonitor disabled. Specify " - << "`RAY_memory_monitor_refresh_ms` > 0 to enable the monitor."; - return std::make_unique(); - } - int64_t total_memory_bytes = MemoryMonitorUtils::TakeSystemMemorySnapshot( MemoryMonitorInterface::kDefaultCgroupPath) .total_bytes; - memory_usage_threshold_bytes = MemoryMonitorUtils::GetMemoryThreshold( + int64_t memory_usage_threshold_bytes = MemoryMonitorUtils::GetMemoryThreshold( total_memory_bytes, RayConfig::instance().memory_usage_threshold(), RayConfig::instance().min_memory_free_bytes(), resource_isolation_enabled, cgroup_manager); - return std::make_unique(std::move(kill_workers_callback), - memory_usage_threshold_bytes, - monitor_interval_ms); + if (monitor_interval_ms > 0) { + monitors.push_back( + std::make_unique(std::move(kill_workers_callback), + memory_usage_threshold_bytes, + monitor_interval_ms)); + } else { + RAY_LOG(INFO) << "ThresholdMemoryMonitor disabled. Specify " + << "`RAY_memory_monitor_refresh_ms` > 0 to enable the monitor."; + monitors.push_back(std::make_unique()); + } + + return monitors; } } // namespace ray diff --git a/src/ray/common/noop_memory_monitor_factory.cc b/src/ray/common/noop_memory_monitor_factory.cc index a1f6af54e10a..7d2882ded95b 100644 --- a/src/ray/common/noop_memory_monitor_factory.cc +++ b/src/ray/common/noop_memory_monitor_factory.cc @@ -13,6 +13,7 @@ // limitations under the License. #include +#include #include "ray/common/memory_monitor_factory.h" #include "ray/common/memory_monitor_interface.h" @@ -20,11 +21,13 @@ namespace ray { -std::unique_ptr MemoryMonitorFactory::Create( +std::vector> MemoryMonitorFactory::Create( KillWorkersCallback kill_workers_callback, bool resource_isolation_enabled, const CgroupManagerInterface &cgroup_manager) { - return std::make_unique(); + std::vector> monitors; + monitors.push_back(std::make_unique()); + return monitors; } } // namespace ray diff --git a/src/ray/common/pressure_memory_monitor.cc b/src/ray/common/pressure_memory_monitor.cc index 9b9c367d1c8d..48fc5f879feb 100644 --- a/src/ray/common/pressure_memory_monitor.cc +++ b/src/ray/common/pressure_memory_monitor.cc @@ -161,8 +161,7 @@ void PressureMemoryMonitor::MonitoringThreadMain() { if (fds[0].revents & POLLPRI) { if (IsEnabled()) { Disable(); - kill_workers_callback_( - MemoryMonitorUtils::TakeSystemMemorySnapshot(cgroup_path_)); + kill_workers_callback_(); } } else if (fds[0].revents & POLLERR) { RAY_LOG(ERROR) << "Got POLLERR while monitoring memory pressure. " diff --git a/src/ray/common/pressure_memory_monitor.h b/src/ray/common/pressure_memory_monitor.h index 279a08fde671..d997a554c1d8 100644 --- a/src/ray/common/pressure_memory_monitor.h +++ b/src/ray/common/pressure_memory_monitor.h @@ -128,6 +128,20 @@ class PressureMemoryMonitor : public MemoryMonitorInterface { */ bool IsEnabled() const override; + /// The default monitoring mode for cgroup pressure monitor trigger. + /// Possible values are: + /// - some: At least one task is stalled for a specified duration + /// - full: All tasks are stalled for a specified duration + static constexpr char kDefaultMemoryPsiMonitoringMode[] = "some"; + + /// The default stall duration in seconds for cgroup pressure monitor trigger. + /// Possible values are multiples of 2 seconds. + static constexpr uint32_t kDefaultMemoryPsiStallDurationS = 2; + + /// The default proportion of specified duration that the task needs to + /// be stalled to trigger the pressure monitor. Defaults to 0.01%. + static constexpr float kDefaultMemoryPsiStallProportion = 0.0001; + private: /** * @brief Monitoring loop that polls on the memory pressure file, diff --git a/src/ray/common/ray_config_def.h b/src/ray/common/ray_config_def.h index 8c90d03529cb..fc8487e8f43b 100644 --- a/src/ray/common/ray_config_def.h +++ b/src/ray/common/ray_config_def.h @@ -71,15 +71,15 @@ RAY_CONFIG(uint64_t, raylet_check_gc_period_milliseconds, 100) /// memory_usage_threshold and free space is below the min_memory_free_bytes then /// it will start killing processes to free up the space. /// Note: when resource isolation is enabled, the memory usage threshold is set to -/// total memory - system reserved memory (can be specified in ray start) - -/// max(5% of total memory, max_threshold_monitor_reaction_buffer_bytes). +/// total memory - system reserved memory (can be specified in ray start). /// Notice that the formula does not account for object store memory in system reserved -/// memory. To configure the usage threshold, please adjust the system reserved memory in -/// ray start command instead. Ranging from [0, 1] +/// memory. To configure the usage threshold when resource isolation is enabled, +/// please adjust the system reserved memory in ray start command instead. +/// Ranging from [0, 1] RAY_CONFIG(float, memory_usage_threshold, 0.95) /// The interval between runs of the memory usage monitor. -/// Monitor is disabled when this value is 0. +/// ThresholdMemoryMonitor is disabled when this value is 0. RAY_CONFIG(uint64_t, memory_monitor_refresh_ms, 250) /// The minimum amount of free space. If the memory is above the @@ -97,15 +97,6 @@ RAY_CONFIG(int64_t, min_memory_free_bytes, (int64_t)-1) /// max_kill_memory_buffer_bytes. RAY_CONFIG(int64_t, max_kill_memory_buffer_bytes, 3ULL * 1024 * 1024 * 1024) // 3GiB cap -/// The threshold monitor is poll based and may miss memory bursts occurring between -/// polls. This is the maximum buffer size that can be subtracted from memory.max to give -/// the threshold monitor time to react before memory max is reached under resource -/// isolation. The system will by default provide 5% of total memory as reaction buffer, -/// capping at max_threshold_monitor_reaction_buffer_bytes. -RAY_CONFIG(int64_t, - max_threshold_monitor_reaction_buffer_bytes, - 2LL * 1024 * 1024 * 1024) // 2GiB - /// When true, use the legacy group-by-owner worker killing policy instead of the /// default time-based policy. RAY_CONFIG(bool, worker_killing_policy_by_group, false) @@ -123,14 +114,17 @@ RAY_CONFIG(int64_t, system_memory_bytes_min, 0) /// Enforced by the cgroup memory.high constraint which throttles the /// user processes' when the threshold is reached. /// Default is 1.0, meaning the user processes are allowed to use 100% of the total -/// memory. Only configure this value if you are confident that +/// memory. If resource isolation is enabled, the user memory.high constraint +/// will be set to the min of total memory - system reserved memory +/// and user_memory_proportion_high * total memory. +/// Only configure this value if you are confident that /// the configuration is desirable. Bad constraint configurations may /// lead to significant system performance degradation. RAY_CONFIG(float, user_memory_proportion_high, 1.0) /// The proportion of total memory the user processes are allowed to use. /// Enforced by the cgroup memory.max constraint which triggers the -//. kernel OOM killer when the threshold is reached. +/// kernel OOM killer when the threshold is reached. /// Default is 1.0, meaning the user processes are allowed to use 100% of the total /// memory. Only configure this value if you are confident that /// the configuration is desirable. Bad constraint configurations may diff --git a/src/ray/common/tests/BUILD.bazel b/src/ray/common/tests/BUILD.bazel index 7ab0c62e8f7c..0578d062f789 100644 --- a/src/ray/common/tests/BUILD.bazel +++ b/src/ray/common/tests/BUILD.bazel @@ -188,6 +188,28 @@ ray_cc_test( ], ) +ray_cc_test( + name = "memory_monitor_factory_test", + size = "small", + srcs = [ + "memory_monitor_factory_test.cc", + ], + tags = [ + "no_windows", + "team:core", + ], + target_compatible_with = [ + "@platforms//os:linux", + ], + deps = [ + "//src/ray/common:memory_monitor_factory", + "//src/ray/common:memory_monitor_interface", + "//src/ray/common:threshold_memory_monitor", + "//src/ray/common/cgroup2:cgroup_manager_interface", + "//src/ray/common/cgroup2:cgroup_test_utils", + ], +) + ray_cc_test( name = "grpc_util_test", size = "small", diff --git a/src/ray/common/tests/event_memory_monitor_test.cc b/src/ray/common/tests/event_memory_monitor_test.cc index 1ed244374c09..0d2eb99c26d4 100644 --- a/src/ray/common/tests/event_memory_monitor_test.cc +++ b/src/ray/common/tests/event_memory_monitor_test.cc @@ -66,8 +66,7 @@ class EventMemoryMonitorTest : public ::testing::Test { TEST_F(EventMemoryMonitorTest, TestNonexistentCgroupPathFailsGracefully) { std::string nonexistent_path = "/nonexistent/cgroup/path"; StatusSetOr, StatusT::IOError> result = - EventMemoryMonitor::Create(std::move(nonexistent_path), - [](SystemMemorySnapshot) {}); + EventMemoryMonitor::Create(std::move(nonexistent_path), []() {}); ASSERT_TRUE(result.has_error()) << "Failed to catch invalid cgroup path when creating EventMemoryMonitor"; @@ -79,7 +78,7 @@ TEST_F(EventMemoryMonitorTest, TestMissingMemoryEventsFileFailsGracefully) { RAY_CHECK(empty_dir_or.ok()) << empty_dir_or.status().ToString(); std::unique_ptr empty_dir = std::move(empty_dir_or.value()); StatusSetOr, StatusT::IOError> result = - EventMemoryMonitor::Create(empty_dir->GetPath(), [](SystemMemorySnapshot) {}); + EventMemoryMonitor::Create(empty_dir->GetPath(), []() {}); ASSERT_TRUE(result.has_error()) << "Failed to catch invalid cgroup configuration when creating EventMemoryMonitor"; @@ -88,8 +87,7 @@ TEST_F(EventMemoryMonitorTest, TestMissingMemoryEventsFileFailsGracefully) { TEST_F(EventMemoryMonitorTest, TestSuccessfulCreationWithValidPath) { StatusSetOr, StatusT::IOError> result = - EventMemoryMonitor::Create(mock_cgroup_dir_->GetPath(), - [](SystemMemorySnapshot) {}); + EventMemoryMonitor::Create(mock_cgroup_dir_->GetPath(), []() {}); ASSERT_TRUE(result.has_value()) << "Failed to create EventMemoryMonitor: " << result.message(); std::unique_ptr monitor = std::move(result.value()); @@ -100,9 +98,7 @@ TEST_F(EventMemoryMonitorTest, TestCallbackCalledWhenHighEventChanges) { WriteMemoryEventsFile(events_file_->GetPath(), 0, 0); auto callback_latch = std::make_shared(1); - KillWorkersCallback callback = [callback_latch](SystemMemorySnapshot) { - callback_latch->count_down(); - }; + KillWorkersCallback callback = [callback_latch]() { callback_latch->count_down(); }; StatusSetOr, StatusT::IOError> result = EventMemoryMonitor::Create(std::move(mock_cgroup_dir_->GetPath()), @@ -120,9 +116,7 @@ TEST_F(EventMemoryMonitorTest, TestNoCallbackWhenValuesUnchanged) { WriteMemoryEventsFile(events_file_->GetPath(), 0, 0); auto callback_latch = std::make_shared(1); - KillWorkersCallback callback = [callback_latch](SystemMemorySnapshot) { - callback_latch->count_down(); - }; + KillWorkersCallback callback = [callback_latch]() { callback_latch->count_down(); }; StatusSetOr, StatusT::IOError> result = EventMemoryMonitor::Create(mock_cgroup_dir_->GetPath(), callback); @@ -139,9 +133,7 @@ TEST_F(EventMemoryMonitorTest, TestNoCallbackWhenIrrelevantEventChanges) { WriteMemoryEventsFile(events_file_->GetPath(), 0, 0); auto callback_latch = std::make_shared(1); - KillWorkersCallback callback = [callback_latch](SystemMemorySnapshot) { - callback_latch->count_down(); - }; + KillWorkersCallback callback = [callback_latch]() { callback_latch->count_down(); }; StatusSetOr, StatusT::IOError> result = EventMemoryMonitor::Create(mock_cgroup_dir_->GetPath(), callback); @@ -161,17 +153,16 @@ TEST_F(EventMemoryMonitorTest, TestMultipleCallbacksOnMultipleChanges) { auto latch2 = std::make_shared(1); auto latch3 = std::make_shared(1); std::atomic callback_count{0}; - KillWorkersCallback callback = - [&callback_count, latch1, latch2, latch3](SystemMemorySnapshot) { - int count = ++callback_count; - if (count == 1) { - latch1->count_down(); - } else if (count == 2) { - latch2->count_down(); - } else if (count == 3) { - latch3->count_down(); - } - }; + KillWorkersCallback callback = [&callback_count, latch1, latch2, latch3]() { + int count = ++callback_count; + if (count == 1) { + latch1->count_down(); + } else if (count == 2) { + latch2->count_down(); + } else if (count == 3) { + latch3->count_down(); + } + }; StatusSetOr, StatusT::IOError> result = EventMemoryMonitor::Create(mock_cgroup_dir_->GetPath(), callback); diff --git a/src/ray/common/tests/memory_monitor_factory_test.cc b/src/ray/common/tests/memory_monitor_factory_test.cc new file mode 100644 index 000000000000..8e04862b36dd --- /dev/null +++ b/src/ray/common/tests/memory_monitor_factory_test.cc @@ -0,0 +1,104 @@ +// Copyright 2026 The Ray Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include "ray/common/memory_monitor_factory.h" + +#include +#include +#include + +#include "gtest/gtest.h" +#include "ray/common/cgroup2/cgroup_manager_interface.h" +#include "ray/common/cgroup2/cgroup_test_utils.h" +#include "ray/common/memory_monitor_interface.h" +#include "ray/common/threshold_memory_monitor.h" + +namespace ray { + +class FakeCgroupManager : public CgroupManagerInterface { + public: + explicit FakeCgroupManager(int64_t user_memory_max_bytes, + int64_t user_memory_high_bytes) + : user_memory_max_bytes_(user_memory_max_bytes), + user_memory_high_bytes_(user_memory_high_bytes) { + StatusOr> temp_dir_or = TempDirectory::Create(); + RAY_CHECK(temp_dir_or.ok()) << temp_dir_or.status().ToString(); + temp_dir_ = std::move(temp_dir_or.value()); + } + + Status AddProcessToWorkersCgroup(const std::string &) override { return Status::OK(); } + Status AddProcessToSystemCgroup(const std::string &) override { return Status::OK(); } + + std::string GetUserCgroupPath() const override { return temp_dir_->GetPath(); } + + StatusOr GetSystemCgroupConstraintValue( + const std::string &) const override { + return Status::IOError("not implemented"); + } + + StatusOr GetUserCgroupConstraintValue( + const std::string &constraint_name) const override { + if (constraint_name == "memory.max") { + return std::to_string(user_memory_max_bytes_); + } + if (constraint_name == "memory.high") { + return std::to_string(user_memory_high_bytes_); + } + return Status::IOError("constraint not found: " + constraint_name); + } + + const std::string &GetPath() const { return temp_dir_->GetPath(); } + + private: + std::unique_ptr temp_dir_; + int64_t user_memory_max_bytes_; + int64_t user_memory_high_bytes_; +}; + +class MemoryMonitorFactoryTest : public ::testing::Test { + protected: + static constexpr int64_t kUserMemoryMaxBytes = 10LL * 1024 * 1024 * 1024; // 10 GB + static constexpr int64_t kUserMemoryHighBytes = 8LL * 1024 * 1024 * 1024; // 8 GB +}; + +TEST_F(MemoryMonitorFactoryTest, + TestCreateWithResourceIsolationDisabledReturnsOnlyThresholdMonitor) { + FakeCgroupManager cgroup_manager(kUserMemoryMaxBytes, kUserMemoryHighBytes); + + std::vector> monitors = + MemoryMonitorFactory::Create([]() {}, + /*resource_isolation_enabled=*/false, + cgroup_manager); + + ASSERT_EQ(monitors.size(), 1u) << "Expected exactly one monitor"; + EXPECT_NE(dynamic_cast(monitors[0].get()), nullptr) + << "Expected the sole monitor to be a ThresholdMemoryMonitor"; +} + +TEST_F(MemoryMonitorFactoryTest, + TestCreateWithResourceIsolationEnabledReturnsThresholdMonitors) { + FakeCgroupManager cgroup_manager(kUserMemoryMaxBytes, kUserMemoryHighBytes); + TempFile pressure_file(cgroup_manager.GetPath() + "/memory.pressure"); + + std::vector> monitors = + MemoryMonitorFactory::Create([]() {}, + /*resource_isolation_enabled=*/true, + cgroup_manager); + + ASSERT_EQ(monitors.size(), 1u) << "Expected exactly one monitor"; + EXPECT_NE(dynamic_cast(monitors[0].get()), nullptr) + << "Expected the sole monitor to be a ThresholdMemoryMonitor"; +} + +} // namespace ray diff --git a/src/ray/common/tests/memory_monitor_utils_test.cc b/src/ray/common/tests/memory_monitor_utils_test.cc index 84b2391f653a..7c2470a88b9a 100644 --- a/src/ray/common/tests/memory_monitor_utils_test.cc +++ b/src/ray/common/tests/memory_monitor_utils_test.cc @@ -284,29 +284,26 @@ TEST_F( std::unique_ptr driver = FakeCgroupDriver::Create(cgroups); int64_t user_memory_max_bytes = 10LL * 1024 * 1024 * 1024; // 10 GB + int64_t user_memory_high_bytes = 8LL * 1024 * 1024 * 1024; // 8 GB StatusOr> result = CgroupManager::Create(cgroup_dir, "node_id_123", /*system_reserved_cpu_weight=*/100, /*system_memory_bytes_min=*/1LL * 1024 * 1024 * 1024, /*system_memory_bytes_low=*/1LL * 1024 * 1024 * 1024, - /*user_memory_high_bytes=*/user_memory_max_bytes, + user_memory_high_bytes, user_memory_max_bytes, std::move(driver)); std::unique_ptr cgroup_manager = std::move(result.value()); - // Reaction buffer defaults to 5% of total memory. If - // kDefaultThresholdMonitorReactionBufferProportion is changed, this should be changed - // accordingly. - int64_t expected_threshold = - user_memory_max_bytes - static_cast(16LL * 1024 * 1024 * 1024 * 0.05); + int64_t expected_default_mode_threshold = user_memory_high_bytes; ASSERT_EQ(MemoryMonitorUtils::GetMemoryThreshold( /*total_memory_bytes=*/16LL * 1024 * 1024 * 1024, /*usage_threshold=*/0.5, /*min_memory_free_bytes=*/MemoryMonitorInterface::kNull, /*resource_isolation_enabled=*/true, *cgroup_manager), - expected_threshold); + expected_default_mode_threshold); } TEST_F(MemoryMonitorUtilsTest, TestGetPidsFromDirOnlyReturnsNumericFilenames) { diff --git a/src/ray/common/tests/pressure_memory_monitor_test.cc b/src/ray/common/tests/pressure_memory_monitor_test.cc index 9badacfc7995..9a303bf64343 100644 --- a/src/ray/common/tests/pressure_memory_monitor_test.cc +++ b/src/ray/common/tests/pressure_memory_monitor_test.cc @@ -99,8 +99,7 @@ TEST_F(PressureMemoryMonitorTest, TestInvalidStallDurationReturnsFalse) { TEST_F(PressureMemoryMonitorTest, TestNonexistentCgroupPathFailsGracefully) { MemoryPsi psi = {.mode = "some", .stall_proportion = 0.5f, .stall_duration_s = 2}; std::string nonexistent_path = "/nonexistent/cgroup/path"; - auto result = PressureMemoryMonitor::Create( - psi, std::move(nonexistent_path), [](const SystemMemorySnapshot &) {}); + auto result = PressureMemoryMonitor::Create(psi, std::move(nonexistent_path), []() {}); ASSERT_TRUE(result.has_error()) << "Failed to catch invalid cgroup path when creating PressureMemoryMonitor"; @@ -110,8 +109,7 @@ TEST_F(PressureMemoryMonitorTest, TestNonexistentCgroupPathFailsGracefully) { TEST_F(PressureMemoryMonitorTest, TestMonitorCreationWritesTriggerStringToFile) { MemoryPsi psi = {.mode = "some", .stall_proportion = 0.5f, .stall_duration_s = 2}; - auto result = PressureMemoryMonitor::Create( - psi, mock_cgroup_dir_->GetPath(), [](const SystemMemorySnapshot &) {}); + auto result = PressureMemoryMonitor::Create(psi, mock_cgroup_dir_->GetPath(), []() {}); ASSERT_TRUE(result.has_value()) << "Failed to create PressureMemoryMonitor: " << result.message(); @@ -164,9 +162,7 @@ TEST_F(PressureMemoryMonitorTest, close(listener); std::shared_ptr has_called_once = std::make_shared(1); - auto kill_workers_callback = [has_called_once](const SystemMemorySnapshot &) { - has_called_once->count_down(); - }; + auto kill_workers_callback = [has_called_once]() { has_called_once->count_down(); }; std::unique_ptr monitor = std::make_unique( mock_cgroup_dir_->GetPath(), listener_fd, kill_workers_callback); diff --git a/src/ray/common/tests/threshold_memory_monitor_test.cc b/src/ray/common/tests/threshold_memory_monitor_test.cc index 9121500337bd..462239db55b2 100644 --- a/src/ray/common/tests/threshold_memory_monitor_test.cc +++ b/src/ray/common/tests/threshold_memory_monitor_test.cc @@ -54,15 +54,7 @@ TEST_F(ThresholdMemoryMonitorTest, TestMonitorTriggerCanDetectMemoryUsage) { MakeThresholdMemoryMonitor( 0 /*memory_usage_threshold_bytes*/, 1 /*refresh_interval_ms*/, - [has_checked_once](SystemMemorySnapshot system_memory) { - ASSERT_GT(system_memory.total_bytes, 0) - << "Reported total bytes from cgroup is <= 0. Is the system memory snapshot " - "taken correctly?"; - ASSERT_GE(system_memory.used_bytes, 0) - << "Reported used bytes from cgroup is < 0. Is the system memory snapshot " - "taken correctly?"; - has_checked_once->count_down(); - }, + [has_checked_once]() { has_checked_once->count_down(); }, "" /*root_cgroup_path*/); has_checked_once->wait(); } @@ -86,12 +78,7 @@ TEST_F(ThresholdMemoryMonitorTest, MakeThresholdMemoryMonitor( memory_usage_threshold_bytes, // (70%) 1 /*refresh_interval_ms*/, - [has_checked_once, cgroup_total_bytes](SystemMemorySnapshot system_memory) { - ASSERT_EQ(system_memory.total_bytes, cgroup_total_bytes) - << "Unexpected total bytes read from cgroup. Are we correctly reading memory " - "from the cgroup?"; - has_checked_once->count_down(); - }, + [has_checked_once]() { has_checked_once->count_down(); }, cgroup_dir /*root_cgroup_path*/); has_checked_once->wait(); @@ -117,9 +104,7 @@ TEST_F(ThresholdMemoryMonitorTest, MakeThresholdMemoryMonitor( memory_usage_threshold_bytes, // (70%) 1 /*refresh_interval_ms*/, - [callback_triggered](SystemMemorySnapshot system_memory) { - callback_triggered->store(true); - }, + [callback_triggered]() { callback_triggered->store(true); }, cgroup_dir /*root_cgroup_path*/); std::this_thread::sleep_for(std::chrono::seconds(5)); diff --git a/src/ray/common/threshold_memory_monitor.cc b/src/ray/common/threshold_memory_monitor.cc index 7f6e9a078197..a12fd8f5181b 100644 --- a/src/ray/common/threshold_memory_monitor.cc +++ b/src/ray/common/threshold_memory_monitor.cc @@ -60,7 +60,7 @@ ThresholdMemoryMonitor::ThresholdMemoryMonitor(KillWorkersCallback kill_workers_ if (is_usage_above_threshold && IsEnabled()) { Disable(); - kill_workers_callback_(std::move(cur_memory_snapshot)); + kill_workers_callback_(); } }, monitor_interval_ms, diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc index ea5df8ed1210..fc9a9097aa8e 100644 --- a/src/ray/raylet/node_manager.cc +++ b/src/ray/raylet/node_manager.cc @@ -244,9 +244,9 @@ NodeManager::NodeManager( ray_syncer_(io_service_, self_node_id_.Binary(), 1, 0), worker_killing_policy_(WorkerKillingPolicyFactory::Create( config.enable_resource_isolation, *cgroup_manager)), - memory_monitor_(MemoryMonitorFactory::Create(CreateKillWorkersCallback(), - config.enable_resource_isolation, - *cgroup_manager)), + memory_monitors_(MemoryMonitorFactory::Create(CreateKillWorkersCallback(), + config.enable_resource_isolation, + *cgroup_manager)), add_process_to_system_cgroup_hook_(std::move(add_process_to_system_cgroup_hook)), cgroup_manager_(std::move(cgroup_manager)), shutting_down_(shutting_down), @@ -3046,13 +3046,34 @@ std::optional NodeManager::CreateSyncMessage( return std::make_optional(std::move(msg)); } +bool NodeManager::MarkKillWorkerInProgress() { + absl::MutexLock lock(&worker_killing_in_progress_mutex_); + if (worker_killing_in_progress_) { + return false; + } + worker_killing_in_progress_ = true; + for (auto &monitor : memory_monitors_) { + monitor->Disable(); + } + return true; +} + +void NodeManager::ReleaseKillWorkerInProgress() { + absl::MutexLock lock(&worker_killing_in_progress_mutex_); + worker_killing_in_progress_ = false; + for (auto &monitor : memory_monitors_) { + monitor->Enable(); + } +} + // Picks the workers and kills the process if the memory usage is above the threshold. KillWorkersCallback NodeManager::CreateKillWorkersCallback() { - return [this](SystemMemorySnapshot system_memory_snapshot) { + return [this]() { + if (!MarkKillWorkerInProgress()) { + return; + } io_service_.post( - [this, system_memory = std::move(system_memory_snapshot)]() { - ProcessesMemorySnapshot process_memory_snapshot = - MemoryMonitorUtils::TakePerProcessMemorySnapshot(); + [this]() { std::vector> workers = worker_pool_.GetAllRegisteredWorkers(/* filter_dead_workers */ true, /* filter_io_workers */ true); @@ -3062,20 +3083,25 @@ KillWorkersCallback NodeManager::CreateKillWorkersCallback() { "killing." << "This could be due to worker memory leak and" << "idle worker are occupying most of the memory."; - memory_monitor_->Enable(); + ReleaseKillWorkerInProgress(); return; } + ProcessesMemorySnapshot process_memory_snapshot = + MemoryMonitorUtils::TakePerProcessMemorySnapshot(); + SystemMemorySnapshot system_memory_snapshot = + MemoryMonitorUtils::TakeSystemMemorySnapshot( + MemoryMonitorInterface::kDefaultCgroupPath); std::vector, bool>> workers_to_kill_and_should_retry = worker_killing_policy_->SelectWorkersToKill( - workers, process_memory_snapshot, system_memory); + workers, process_memory_snapshot, system_memory_snapshot); if (workers_to_kill_and_should_retry.empty()) { - memory_monitor_->Enable(); + ReleaseKillWorkerInProgress(); return; } // Compute the memory usage threshold - int64_t total_memory_bytes = system_memory.total_bytes; + int64_t total_memory_bytes = system_memory_snapshot.total_bytes; int64_t computed_threshold_bytes = MemoryMonitorUtils::GetMemoryThreshold( total_memory_bytes, RayConfig::instance().memory_usage_threshold(), @@ -3089,7 +3115,7 @@ KillWorkersCallback NodeManager::CreateKillWorkersCallback() { std::string oom_kill_details = CreateOomKillMessageDetails( workers_to_kill_and_should_retry, self_node_id_, - system_memory, + system_memory_snapshot, store_client_->GetMemoryUsage().value_or("Not available"), process_memory_snapshot, computed_threshold_fraction); @@ -3150,7 +3176,7 @@ KillWorkersCallback NodeManager::CreateKillWorkersCallback() { {"Name", ray_lease.GetLeaseSpecification().GetTaskName()}}); } } - memory_monitor_->Enable(); + ReleaseKillWorkerInProgress(); }, "NodeManager.KillWorkersCallback"); }; diff --git a/src/ray/raylet/node_manager.h b/src/ray/raylet/node_manager.h index 9739abb42b26..e7452dd2d4d9 100644 --- a/src/ray/raylet/node_manager.h +++ b/src/ray/raylet/node_manager.h @@ -767,6 +767,20 @@ class NodeManager : public rpc::NodeManagerServiceHandler, */ KillWorkersCallback CreateKillWorkersCallback(); + /** + * @brief Marks kill worker in progress and disables all monitors to + * prevent more than one in-flight kill worker operation. + * @return True if we successfully marked the kill worker in progress. + * False if the kill worker operation is already in progress. + */ + bool MarkKillWorkerInProgress(); + + /** + * @brief Disables kill worker in progress flag and enables all monitors + * to begin accepting new kill worker requests. + */ + void ReleaseKillWorkerInProgress(); + /** * @param workers_to_kill The workers to print the kill details for. * @param node_id The ID of the node. @@ -975,8 +989,15 @@ class NodeManager : public rpc::NodeManagerServiceHandler, /// The Policy for selecting the worker to kill when the node runs out of memory. std::unique_ptr worker_killing_policy_; - /// Monitors and reports node memory usage and whether it is above threshold. - std::unique_ptr memory_monitor_; + /// Guards worker_killing_in_progress_ and serializes kill callbacks from monitors. + absl::Mutex worker_killing_in_progress_mutex_; + + /// True while a kill-workers operation is inflight; prevents concurrent kills. + bool worker_killing_in_progress_ ABSL_GUARDED_BY(worker_killing_in_progress_mutex_) = + false; + + /// Monitors node memory usage and triggers kill callbacks when pressure is detected. + std::vector> memory_monitors_; /// Used to move the dashboard and runtime_env agents into the system cgroup. AddProcessToCgroupHook add_process_to_system_cgroup_hook_;