From a17f625d9b08b39c604c410c6653f575f6c82d41 Mon Sep 17 00:00:00 2001 From: davik Date: Sun, 19 Apr 2026 18:47:18 +0000 Subject: [PATCH 1/5] Wire together memory monitors through multi memory monitor factory and update cgroup bounds Signed-off-by: davik --- python/ray/_private/ray_constants.py | 2 +- src/ray/common/BUILD.bazel | 11 +- .../cgroup2/linux_cgroup_manager_factory.cc | 14 +- src/ray/common/event_memory_monitor.cc | 3 +- src/ray/common/memory_monitor_factory.h | 32 +++-- src/ray/common/memory_monitor_interface.h | 6 +- src/ray/common/memory_monitor_utils.cc | 57 +++++--- src/ray/common/memory_monitor_utils.h | 3 + .../common/multi_memory_monitor_factory.cc | 101 +++++++++++++ src/ray/common/noop_memory_monitor_factory.cc | 8 +- src/ray/common/pressure_memory_monitor.cc | 3 +- src/ray/common/pressure_memory_monitor.h | 14 ++ src/ray/common/ray_config_def.h | 14 +- src/ray/common/tests/BUILD.bazel | 25 ++++ .../common/tests/event_memory_monitor_test.cc | 41 +++--- .../tests/memory_monitor_factory_test.cc | 135 ++++++++++++++++++ .../common/tests/memory_monitor_utils_test.cc | 66 +++++---- .../tests/pressure_memory_monitor_test.cc | 10 +- .../tests/threshold_memory_monitor_test.cc | 25 +--- src/ray/common/threshold_memory_monitor.cc | 2 +- .../threshold_memory_monitor_factory.cc | 56 -------- src/ray/raylet/node_manager.cc | 59 ++++++-- src/ray/raylet/node_manager.h | 24 +++- .../raylet/worker_killing_policy_factory.cc | 1 + 24 files changed, 508 insertions(+), 204 deletions(-) create mode 100644 src/ray/common/multi_memory_monitor_factory.cc create mode 100644 src/ray/common/tests/memory_monitor_factory_test.cc delete mode 100644 src/ray/common/threshold_memory_monitor_factory.cc diff --git a/python/ray/_private/ray_constants.py b/python/ray/_private/ray_constants.py index 864e4429bc83..18af055239e6 100644 --- a/python/ray/_private/ray_constants.py +++ b/python/ray/_private/ray_constants.py @@ -63,7 +63,7 @@ def env_set_by_user(key): # The default minimum number of bytes to reserve for ray system processes. # This value is used if the available_memory * DEFAULT_SYSTEM_RESERVED_MEMORY_PROPORTION < this value. DEFAULT_MIN_SYSTEM_RESERVED_MEMORY_BYTES = env_integer( - "RAY_DEFAULT_MIN_SYSTEM_RESERVED_MEMORY_BYTES", (1) * (1024**3) + "RAY_DEFAULT_MIN_SYSTEM_RESERVED_MEMORY_BYTES", 500 * (1024**2) ) # The default maximum number of bytes to reserve for ray system processes. # This value is used if the available_memory * DEFAULT_SYSTEM_RESERVED_MEMORY_PROPORTION > this value. diff --git a/src/ray/common/BUILD.bazel b/src/ray/common/BUILD.bazel index ae2113296ebc..d2d4727dae1c 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,21 @@ ray_cc_library( ], deps = [ ":memory_monitor_interface", - ":noop_memory_monitor", "//src/ray/common/cgroup2:cgroup_manager_interface", ] + select({ "//bazel:is_linux": [ + ":event_memory_monitor", ":memory_monitor_utils", + ":noop_memory_monitor", + ":pressure_memory_monitor", ":ray_config", + ":status_or", ":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..70b583658a65 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 @@ -68,9 +69,16 @@ std::unique_ptr CgroupManagerFactory::Create( int64_t user_memory_high_bytes = static_cast(total_memory_bytes * user_memory_proportion_high); - int64_t user_memory_max_bytes = std::min( - total_memory_bytes - system_reserved_memory_bytes + object_store_memory_bytes, - static_cast(total_memory_bytes * user_memory_proportion_max)); + int64_t user_memory_max_bytes = + static_cast(total_memory_bytes * user_memory_proportion_max); + if (RayConfig::instance().enable_memory_throttling_mode()) { + user_memory_high_bytes = std::min(total_memory_bytes - system_reserved_memory_bytes, + user_memory_high_bytes); + } else { + user_memory_max_bytes = std::min( + total_memory_bytes - system_reserved_memory_bytes + object_store_memory_bytes, + user_memory_max_bytes); + } 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..8ee352a22526 100644 --- a/src/ray/common/memory_monitor_factory.h +++ b/src/ray/common/memory_monitor_factory.h @@ -15,40 +15,42 @@ #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. + * Create memory monitor instances based on configuration. * - * On Linux, creates a ThresholdMemoryMonitor that monitors memory usage - * and triggers the callback when usage is refreshed. + * On Linux, creates monitors based on configuration: + * - Resource isolation disabled: ThresholdMemoryMonitor only. + * - Resource isolation enabled, throttling disabled: ThresholdMemoryMonitor + + * PressureMemoryMonitor. + * - Resource isolation enabled, throttling enabled: EventMemoryMonitor only. * - * On non-Linux platforms, creates a NoopMemoryMonitor that does nothing. + * On non-Linux platforms, returns a vector with a single NoopMemoryMonitor. * - * @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 memory_throttling_mode_enabled when enabled, the memory monitor will work + * with cgroup constraints to balance between memory throttling and worker killing to + * enforce stronger resource isolation between user and system slice. + * @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, + bool memory_throttling_mode_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..9d9abf9e63f7 100644 --- a/src/ray/common/memory_monitor_utils.cc +++ b/src/ray/common/memory_monitor_utils.cc @@ -298,6 +298,7 @@ int64_t MemoryMonitorUtils::GetMemoryThreshold( float usage_threshold, int64_t min_memory_free_bytes, bool resource_isolation_enabled, + bool memory_throttling_mode_enabled, const CgroupManagerInterface &cgroup_manager) { RAY_CHECK_GE(total_memory_bytes, MemoryMonitorInterface::kNull); RAY_CHECK_GE(min_memory_free_bytes, MemoryMonitorInterface::kNull); @@ -318,30 +319,44 @@ int64_t MemoryMonitorUtils::GetMemoryThreshold( } if (resource_isolation_enabled) { - StatusOr user_memory_max_bytes_or = + StatusOr user_slice_upper_bound_bytes_or = cgroup_manager.GetUserCgroupConstraintValue("memory.max"); - RAY_CHECK(user_memory_max_bytes_or.ok()) << absl::StrFormat( + if (memory_throttling_mode_enabled) { + 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 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(), + user_slice_upper_bound_bytes_or.ToString()); + std::string user_slice_upper_bound_bytes_str = + std::string(absl::StripAsciiWhitespace(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. " + "Please check that the cgroup path for resource isolation is correct.", + 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); + int64_t user_slice_upper_bound_bytes = std::stoll(user_slice_upper_bound_bytes_str); + if (memory_throttling_mode_enabled) { + resolved_memory_threshold_bytes = user_slice_upper_bound_bytes; + } else { + 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_slice_upper_bound_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_slice_upper_bound_bytes, + reaction_buffer_bytes); + } } } diff --git a/src/ray/common/memory_monitor_utils.h b/src/ray/common/memory_monitor_utils.h index cc6e49596123..50a652cf6385 100644 --- a/src/ray/common/memory_monitor_utils.h +++ b/src/ray/common/memory_monitor_utils.h @@ -73,6 +73,8 @@ class MemoryMonitorUtils { * @param resource_isolation_enabled Whether resource isolation is enabled. Used * to determine if the threshold should be calculated based on the cgroup * constraints. + * @param memory_throttling_mode_enabled When true and resource isolation is enabled, + * compute the memory threshold for memory throttling mode. * @param cgroup_manager The cgroup manager to fetch the upper bound memory constraints * from. * @return The memory threshold. @@ -81,6 +83,7 @@ class MemoryMonitorUtils { float usage_threshold, int64_t min_memory_free_bytes, bool resource_isolation_enabled, + bool memory_throttling_mode_enabled, const CgroupManagerInterface &cgroup_manager); /** diff --git a/src/ray/common/multi_memory_monitor_factory.cc b/src/ray/common/multi_memory_monitor_factory.cc new file mode 100644 index 000000000000..1a4ba8ac069b --- /dev/null +++ b/src/ray/common/multi_memory_monitor_factory.cc @@ -0,0 +1,101 @@ +// 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 +#include + +#include "ray/common/event_memory_monitor.h" +#include "ray/common/memory_monitor_factory.h" +#include "ray/common/memory_monitor_interface.h" +#include "ray/common/memory_monitor_utils.h" +#include "ray/common/noop_memory_monitor.h" +#include "ray/common/pressure_memory_monitor.h" +#include "ray/common/ray_config.h" +#include "ray/common/status_or.h" +#include "ray/common/threshold_memory_monitor.h" +#include "ray/util/logging.h" + +namespace ray { + +std::vector> MemoryMonitorFactory::Create( + KillWorkersCallback kill_workers_callback, + bool resource_isolation_enabled, + bool memory_throttling_mode_enabled, + const CgroupManagerInterface &cgroup_manager) { + std::vector> monitors; + + uint64_t monitor_interval_ms = RayConfig::instance().memory_monitor_refresh_ms(); + int64_t total_memory_bytes = MemoryMonitorUtils::TakeSystemMemorySnapshot( + MemoryMonitorInterface::kDefaultCgroupPath) + .total_bytes; + 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, + memory_throttling_mode_enabled, + cgroup_manager); + + if (resource_isolation_enabled) { + std::string user_cgroup_path = cgroup_manager.GetUserCgroupPath(); + + if (memory_throttling_mode_enabled) { + StatusSetOr, StatusT::IOError> + event_monitor_or = EventMemoryMonitor::Create(user_cgroup_path, + std::move(kill_workers_callback)); + RAY_CHECK(event_monitor_or.has_value()) + << "Failed to create EventMemoryMonitor: " << event_monitor_or.message(); + monitors.push_back(std::move(event_monitor_or.value())); + } else { + if (monitor_interval_ms > 0) { + monitors.push_back(std::make_unique( + kill_workers_callback, memory_usage_threshold_bytes, monitor_interval_ms)); + } else { + RAY_LOG(INFO) + << "ThresholdMemoryMonitor disabled. Relying on PressureMemoryMonitor only. " + << "Specify `RAY_memory_monitor_refresh_ms` > 0 to enable the " + "ThresholdMemoryMonitor."; + } + + MemoryPsi pressure_threshold{ + PressureMemoryMonitor::kDefaultMemoryPsiMonitoringMode, + PressureMemoryMonitor::kDefaultMemoryPsiStallProportion, + PressureMemoryMonitor::kDefaultMemoryPsiStallDurationS}; + + StatusSetOr, + StatusT::InvalidArgument, + StatusT::IOError> + pressure_monitor_or = PressureMemoryMonitor::Create( + pressure_threshold, user_cgroup_path, std::move(kill_workers_callback)); + RAY_CHECK(pressure_monitor_or.has_value()) + << "Failed to create PressureMemoryMonitor: " << pressure_monitor_or.message(); + monitors.push_back(std::move(pressure_monitor_or.value())); + } + } else { + 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..83b003263348 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,14 @@ namespace ray { -std::unique_ptr MemoryMonitorFactory::Create( +std::vector> MemoryMonitorFactory::Create( KillWorkersCallback kill_workers_callback, bool resource_isolation_enabled, + bool memory_throttling_mode_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..fbbfaeff1414 100644 --- a/src/ray/common/ray_config_def.h +++ b/src/ray/common/ray_config_def.h @@ -79,7 +79,7 @@ RAY_CONFIG(uint64_t, raylet_check_gc_period_milliseconds, 100) 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 @@ -110,6 +110,18 @@ RAY_CONFIG(int64_t, /// default time-based policy. RAY_CONFIG(bool, worker_killing_policy_by_group, false) +/// Whether to enable the memory throttling monitoring system. +/// When true, the memory monitor will work with cgroup constraints to +/// balance between memory throttling and worker killing to enforce +/// stronger resource isolation between user and system processes. +/// This mode restrict the available heap memory for user processes to +/// total system memory - system reserved memory - object store memory. +/// +/// This mode is only supported when resource isolation is enabled, and +/// should only be enabled if the workload is experiencing node deaths +/// due to memory pressure. +RAY_CONFIG(bool, enable_memory_throttling_mode, false) + /// The reserved memory bytes for system processes /// enforced via cgroup memory.min constraint which guarantees /// that the system processes' memory will not be reclaimed under any conditions. diff --git a/src/ray/common/tests/BUILD.bazel b/src/ray/common/tests/BUILD.bazel index 7ab0c62e8f7c..7375c4914884 100644 --- a/src/ray/common/tests/BUILD.bazel +++ b/src/ray/common/tests/BUILD.bazel @@ -188,6 +188,31 @@ 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:event_memory_monitor", + "//src/ray/common:memory_monitor_factory", + "//src/ray/common:memory_monitor_interface", + "//src/ray/common:pressure_memory_monitor", + "//src/ray/common:threshold_memory_monitor", + "//src/ray/common/cgroup2:cgroup_manager_interface", + "//src/ray/common/cgroup2:cgroup_test_utils", + "//src/ray/common/cgroup2:noop_cgroup_manager", + ], +) + 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..48bcb02f2e0d --- /dev/null +++ b/src/ray/common/tests/memory_monitor_factory_test.cc @@ -0,0 +1,135 @@ +// 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 + +#include "gtest/gtest.h" +#include "ray/common/cgroup2/cgroup_manager_interface.h" +#include "ray/common/cgroup2/cgroup_test_utils.h" +#include "ray/common/cgroup2/noop_cgroup_manager.h" +#include "ray/common/event_memory_monitor.h" +#include "ray/common/memory_monitor_interface.h" +#include "ray/common/pressure_memory_monitor.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: + // Large enough that threshold (memory.max - reaction_buffer) is > 0. + static constexpr int64_t kUserMemoryMaxBytes = 10LL * 1024 * 1024 * 1024; // 10 GB + static constexpr int64_t kUserMemoryHighBytes = 8LL * 1024 * 1024 * 1024; // 5 GB +}; + +TEST_F(MemoryMonitorFactoryTest, + TestCreateWithResourceIsolationDisabledReturnsOnlyThresholdMonitor) { + FakeCgroupManager cgroup_manager(kUserMemoryMaxBytes, kUserMemoryHighBytes); + + std::vector> monitors = + MemoryMonitorFactory::Create([]() {}, + /*resource_isolation_enabled=*/false, + /*memory_throttling_mode_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, + TestCreateWithResourceIsolationEnabledThrottlingDisabledReturnsPressureAndThresholdMonitors) { + FakeCgroupManager cgroup_manager(kUserMemoryMaxBytes, kUserMemoryHighBytes); + TempFile pressure_file(cgroup_manager.GetPath() + "/memory.pressure"); + + std::vector> monitors = + MemoryMonitorFactory::Create([]() {}, + /*resource_isolation_enabled=*/true, + /*memory_throttling_mode_enabled=*/false, + cgroup_manager); + + ASSERT_EQ(monitors.size(), 2u) + << "Expected ThresholdMemoryMonitor + PressureMemoryMonitor"; + bool has_threshold = false; + bool has_pressure = false; + for (const std::unique_ptr &m : monitors) { + if (dynamic_cast(m.get()) != nullptr) has_threshold = true; + if (dynamic_cast(m.get()) != nullptr) has_pressure = true; + } + EXPECT_TRUE(has_threshold) << "Expected a ThresholdMemoryMonitor"; + EXPECT_TRUE(has_pressure) << "Expected a PressureMemoryMonitor"; +} + +TEST_F(MemoryMonitorFactoryTest, + TestCreateWithResourceIsolationEnabledThrottlingEnabledReturnsEventMonitor) { + FakeCgroupManager cgroup_manager(kUserMemoryMaxBytes, kUserMemoryHighBytes); + TempFile pressure_file(cgroup_manager.GetPath() + "/memory.events"); + + std::vector> monitors = + MemoryMonitorFactory::Create([]() {}, + /*resource_isolation_enabled=*/true, + /*memory_throttling_mode_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 EventMemoryMonitor"; +} + +} // 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..5ce114ebbe40 100644 --- a/src/ray/common/tests/memory_monitor_utils_test.cc +++ b/src/ray/common/tests/memory_monitor_utils_test.cc @@ -234,36 +234,39 @@ TEST_F(MemoryMonitorUtilsTest, TestCgroupNonexistentUsageFileReturnskNull) { TEST_F(MemoryMonitorUtilsTest, TestGetMemoryThresholdTakeGreaterOfTheTwoValues) { NoopCgroupManager noop_cgroup_manager; - ASSERT_EQ( - MemoryMonitorUtils::GetMemoryThreshold(100, 0.5, 0, false, noop_cgroup_manager), - 100); - ASSERT_EQ( - MemoryMonitorUtils::GetMemoryThreshold(100, 0.5, 60, false, noop_cgroup_manager), - 50); - - ASSERT_EQ( - MemoryMonitorUtils::GetMemoryThreshold(100, 1, 10, false, noop_cgroup_manager), - 100); - ASSERT_EQ( - MemoryMonitorUtils::GetMemoryThreshold(100, 1, 100, false, noop_cgroup_manager), - 100); + ASSERT_EQ(MemoryMonitorUtils::GetMemoryThreshold( + 100, 0.5, 0, false, false, noop_cgroup_manager), + 100); + ASSERT_EQ(MemoryMonitorUtils::GetMemoryThreshold( + 100, 0.5, 60, false, false, noop_cgroup_manager), + 50); - ASSERT_EQ( - MemoryMonitorUtils::GetMemoryThreshold(100, 0.1, 100, false, noop_cgroup_manager), - 10); - ASSERT_EQ( - MemoryMonitorUtils::GetMemoryThreshold(100, 0, 10, false, noop_cgroup_manager), 90); - ASSERT_EQ( - MemoryMonitorUtils::GetMemoryThreshold(100, 0, 100, false, noop_cgroup_manager), 0); + ASSERT_EQ(MemoryMonitorUtils::GetMemoryThreshold( + 100, 1, 10, false, false, noop_cgroup_manager), + 100); + ASSERT_EQ(MemoryMonitorUtils::GetMemoryThreshold( + 100, 1, 100, false, false, noop_cgroup_manager), + 100); ASSERT_EQ(MemoryMonitorUtils::GetMemoryThreshold( - 100, 0, MemoryMonitorInterface::kNull, false, noop_cgroup_manager), + 100, 0.1, 100, false, false, noop_cgroup_manager), + 10); + ASSERT_EQ(MemoryMonitorUtils::GetMemoryThreshold( + 100, 0, 10, false, false, noop_cgroup_manager), + 90); + ASSERT_EQ(MemoryMonitorUtils::GetMemoryThreshold( + 100, 0, 100, false, false, noop_cgroup_manager), 0); + ASSERT_EQ(MemoryMonitorUtils::GetMemoryThreshold( - 100, 0.5, MemoryMonitorInterface::kNull, false, noop_cgroup_manager), - 50); + 100, 0, MemoryMonitorInterface::kNull, false, false, noop_cgroup_manager), + 0); + ASSERT_EQ( + MemoryMonitorUtils::GetMemoryThreshold( + 100, 0.5, MemoryMonitorInterface::kNull, false, false, noop_cgroup_manager), + 50); ASSERT_EQ(MemoryMonitorUtils::GetMemoryThreshold( - 100, 1, MemoryMonitorInterface::kNull, false, noop_cgroup_manager), + 100, 1, MemoryMonitorInterface::kNull, false, false, noop_cgroup_manager), 100); } @@ -298,15 +301,26 @@ TEST_F( // Reaction buffer defaults to 5% of total memory. If // kDefaultThresholdMonitorReactionBufferProportion is changed, this should be changed // accordingly. - int64_t expected_threshold = + int64_t expected_default_mode_threshold = user_memory_max_bytes - static_cast(16LL * 1024 * 1024 * 1024 * 0.05); 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, + /*memory_throttling_mode_enabled=*/false, + *cgroup_manager), + expected_default_mode_threshold); + + int64_t expected_throttling_mode_threshold = user_memory_max_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, + /*memory_throttling_mode_enabled=*/true, *cgroup_manager), - expected_threshold); + expected_throttling_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..74bf6e7f2716 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(); } @@ -82,16 +74,11 @@ TEST_F(ThresholdMemoryMonitorTest, NoopCgroupManager noop_cgroup_manager; int64_t memory_usage_threshold_bytes = MemoryMonitorUtils::GetMemoryThreshold( - cgroup_total_bytes, 0.7f, -1, false, noop_cgroup_manager); + cgroup_total_bytes, 0.7f, -1, false, false, noop_cgroup_manager); 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(); @@ -113,13 +100,11 @@ TEST_F(ThresholdMemoryMonitorTest, NoopCgroupManager noop_cgroup_manager; int64_t memory_usage_threshold_bytes = MemoryMonitorUtils::GetMemoryThreshold( - cgroup_total_bytes, 0.7f, -1, false, noop_cgroup_manager); + cgroup_total_bytes, 0.7f, -1, false, false, noop_cgroup_manager); 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/common/threshold_memory_monitor_factory.cc b/src/ray/common/threshold_memory_monitor_factory.cc deleted file mode 100644 index 7c3fca8a514c..000000000000 --- a/src/ray/common/threshold_memory_monitor_factory.cc +++ /dev/null @@ -1,56 +0,0 @@ -// Copyright 2025 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 -#include - -#include "ray/common/memory_monitor_factory.h" -#include "ray/common/memory_monitor_interface.h" -#include "ray/common/memory_monitor_utils.h" -#include "ray/common/noop_memory_monitor.h" -#include "ray/common/ray_config.h" -#include "ray/common/threshold_memory_monitor.h" -#include "ray/util/logging.h" - -namespace ray { - -std::unique_ptr MemoryMonitorFactory::Create( - KillWorkersCallback kill_workers_callback, - bool resource_isolation_enabled, - const CgroupManagerInterface &cgroup_manager) { - int64_t memory_usage_threshold_bytes; - - 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( - 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); -} - -} // namespace ray diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc index 4bf70b196751..2e1327056595 100644 --- a/src/ray/raylet/node_manager.cc +++ b/src/ray/raylet/node_manager.cc @@ -244,9 +244,11 @@ 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, + RayConfig::instance().enable_memory_throttling_mode(), + *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), @@ -3050,13 +3052,38 @@ 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); @@ -3066,25 +3093,31 @@ 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(), RayConfig::instance().min_memory_free_bytes(), initial_config_.enable_resource_isolation, + RayConfig::instance().enable_memory_throttling_mode(), *cgroup_manager_); float computed_threshold_fraction = static_cast(computed_threshold_bytes) / @@ -3093,7 +3126,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); @@ -3154,7 +3187,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 25fbca29d87d..c78555b5e428 100644 --- a/src/ray/raylet/node_manager.h +++ b/src/ray/raylet/node_manager.h @@ -767,6 +767,19 @@ 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 the kill worker operation is not previously 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 +988,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_; diff --git a/src/ray/raylet/worker_killing_policy_factory.cc b/src/ray/raylet/worker_killing_policy_factory.cc index ed3e3f1196a1..e0aea50af80f 100644 --- a/src/ray/raylet/worker_killing_policy_factory.cc +++ b/src/ray/raylet/worker_killing_policy_factory.cc @@ -41,6 +41,7 @@ std::unique_ptr WorkerKillingPolicyFactory::Create RayConfig::instance().memory_usage_threshold(), RayConfig::instance().min_memory_free_bytes(), resource_isolation_enabled, + RayConfig::instance().enable_memory_throttling_mode(), cgroup_manager); int64_t kill_memory_buffer_bytes = From 9c7a9ada7d976ae3420d657a151e73832d9ea25a Mon Sep 17 00:00:00 2001 From: davik Date: Thu, 23 Apr 2026 06:02:01 +0000 Subject: [PATCH 2/5] Lock worker killing mutex until all monitor states are updated Signed-off-by: davik --- src/ray/raylet/node_manager.cc | 16 ++++++---------- 1 file changed, 6 insertions(+), 10 deletions(-) diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc index 2e1327056595..4ee15cd764aa 100644 --- a/src/ray/raylet/node_manager.cc +++ b/src/ray/raylet/node_manager.cc @@ -3053,13 +3053,11 @@ std::optional NodeManager::CreateSyncMessage( } bool NodeManager::MarkKillWorkerInProgress() { - { - absl::MutexLock lock(&worker_killing_in_progress_mutex_); - if (worker_killing_in_progress_) { - return false; - } - worker_killing_in_progress_ = true; + 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(); } @@ -3067,10 +3065,8 @@ bool NodeManager::MarkKillWorkerInProgress() { } void NodeManager::ReleaseKillWorkerInProgress() { - { - absl::MutexLock lock(&worker_killing_in_progress_mutex_); - worker_killing_in_progress_ = false; - } + absl::MutexLock lock(&worker_killing_in_progress_mutex_); + worker_killing_in_progress_ = false; for (auto &monitor : memory_monitors_) { monitor->Enable(); } From 64efec080321c072093fbf7c329c31393bd8ae65 Mon Sep 17 00:00:00 2001 From: davik Date: Thu, 23 Apr 2026 23:48:21 +0000 Subject: [PATCH 3/5] Improve comment clarity Signed-off-by: davik --- src/ray/common/memory_monitor_utils.cc | 7 +++++-- src/ray/common/tests/memory_monitor_utils_test.cc | 5 +++-- src/ray/raylet/node_manager.h | 3 ++- 3 files changed, 10 insertions(+), 5 deletions(-) diff --git a/src/ray/common/memory_monitor_utils.cc b/src/ray/common/memory_monitor_utils.cc index 9d9abf9e63f7..f8afe9feed5c 100644 --- a/src/ray/common/memory_monitor_utils.cc +++ b/src/ray/common/memory_monitor_utils.cc @@ -326,13 +326,16 @@ int64_t MemoryMonitorUtils::GetMemoryThreshold( cgroup_manager.GetUserCgroupConstraintValue("memory.high"); } RAY_CHECK(user_slice_upper_bound_bytes_or.ok()) << absl::StrFormat( - "Failed to get user cgroup memory limit when setting up memory monitor: %s", + "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 = std::string(absl::StripAsciiWhitespace(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. " - "Please check that the cgroup path for resource isolation is correct.", + "Does the cgroup path exist and/or matches the resource isolation hierarchy?", cgroup_manager.GetUserCgroupPath()); if (!user_slice_upper_bound_bytes_str.empty() && diff --git a/src/ray/common/tests/memory_monitor_utils_test.cc b/src/ray/common/tests/memory_monitor_utils_test.cc index 5ce114ebbe40..2746cc446d56 100644 --- a/src/ray/common/tests/memory_monitor_utils_test.cc +++ b/src/ray/common/tests/memory_monitor_utils_test.cc @@ -287,13 +287,14 @@ 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()); @@ -312,7 +313,7 @@ TEST_F( *cgroup_manager), expected_default_mode_threshold); - int64_t expected_throttling_mode_threshold = user_memory_max_bytes; + int64_t expected_throttling_mode_threshold = user_memory_high_bytes; ASSERT_EQ(MemoryMonitorUtils::GetMemoryThreshold( /*total_memory_bytes=*/16LL * 1024 * 1024 * 1024, /*usage_threshold=*/0.5, diff --git a/src/ray/raylet/node_manager.h b/src/ray/raylet/node_manager.h index c78555b5e428..9c408a08d2c3 100644 --- a/src/ray/raylet/node_manager.h +++ b/src/ray/raylet/node_manager.h @@ -770,7 +770,8 @@ class NodeManager : public rpc::NodeManagerServiceHandler, /** * @brief Marks kill worker in progress and disables all monitors to * prevent more than one in-flight kill worker operation. - * @return True if the kill worker operation is not previously in progress. + * @return True if we successfully marked the kill worker in progress. + * False if the kill worker operation is already in progress. */ bool MarkKillWorkerInProgress(); From 6fd99528f40d65458171b092720393c41d725389 Mon Sep 17 00:00:00 2001 From: davik Date: Mon, 27 Apr 2026 23:43:06 +0000 Subject: [PATCH 4/5] Wire multi-memmon just for memory.high Signed-off-by: davik --- python/ray/_private/ray_constants.py | 2 +- .../cgroup2/linux_cgroup_manager_factory.cc | 16 ++--- src/ray/common/memory_monitor_factory.h | 10 +-- src/ray/common/memory_monitor_utils.cc | 28 +------- src/ray/common/memory_monitor_utils.h | 3 - .../common/multi_memory_monitor_factory.cc | 55 +++------------- src/ray/common/noop_memory_monitor_factory.cc | 1 - src/ray/common/ray_config_def.h | 19 ++---- .../tests/memory_monitor_factory_test.cc | 34 ++-------- .../common/tests/memory_monitor_utils_test.cc | 65 +++++++------------ .../tests/threshold_memory_monitor_test.cc | 4 +- src/ray/raylet/node_manager.cc | 9 +-- .../raylet/worker_killing_policy_factory.cc | 1 - 13 files changed, 58 insertions(+), 189 deletions(-) diff --git a/python/ray/_private/ray_constants.py b/python/ray/_private/ray_constants.py index 18af055239e6..864e4429bc83 100644 --- a/python/ray/_private/ray_constants.py +++ b/python/ray/_private/ray_constants.py @@ -63,7 +63,7 @@ def env_set_by_user(key): # The default minimum number of bytes to reserve for ray system processes. # This value is used if the available_memory * DEFAULT_SYSTEM_RESERVED_MEMORY_PROPORTION < this value. DEFAULT_MIN_SYSTEM_RESERVED_MEMORY_BYTES = env_integer( - "RAY_DEFAULT_MIN_SYSTEM_RESERVED_MEMORY_BYTES", 500 * (1024**2) + "RAY_DEFAULT_MIN_SYSTEM_RESERVED_MEMORY_BYTES", (1) * (1024**3) ) # The default maximum number of bytes to reserve for ray system processes. # This value is used if the available_memory * DEFAULT_SYSTEM_RESERVED_MEMORY_PROPORTION > this value. diff --git a/src/ray/common/cgroup2/linux_cgroup_manager_factory.cc b/src/ray/common/cgroup2/linux_cgroup_manager_factory.cc index 70b583658a65..9c71915e07a5 100644 --- a/src/ray/common/cgroup2/linux_cgroup_manager_factory.cc +++ b/src/ray/common/cgroup2/linux_cgroup_manager_factory.cc @@ -67,18 +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); + // 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_high)); int64_t user_memory_max_bytes = static_cast(total_memory_bytes * user_memory_proportion_max); - if (RayConfig::instance().enable_memory_throttling_mode()) { - user_memory_high_bytes = std::min(total_memory_bytes - system_reserved_memory_bytes, - user_memory_high_bytes); - } else { - user_memory_max_bytes = std::min( - total_memory_bytes - system_reserved_memory_bytes + object_store_memory_bytes, - user_memory_max_bytes); - } + StatusOr> cgroup_manager_s = CgroupManager::Create(cgroup_path, node_id, diff --git a/src/ray/common/memory_monitor_factory.h b/src/ray/common/memory_monitor_factory.h index 8ee352a22526..1ed45adcae03 100644 --- a/src/ray/common/memory_monitor_factory.h +++ b/src/ray/common/memory_monitor_factory.h @@ -25,13 +25,9 @@ namespace ray { class MemoryMonitorFactory { public: /** - * Create memory monitor instances based on configuration. - * * On Linux, creates monitors based on configuration: * - Resource isolation disabled: ThresholdMemoryMonitor only. - * - Resource isolation enabled, throttling disabled: ThresholdMemoryMonitor + - * PressureMemoryMonitor. - * - Resource isolation enabled, throttling enabled: EventMemoryMonitor only. + * - Resource isolation enabled, ThresholdMemoryMonitor + EventMemoryMonitor. * * On non-Linux platforms, returns a vector with a single NoopMemoryMonitor. * @@ -39,9 +35,6 @@ class MemoryMonitorFactory { * @param resource_isolation_enabled When resource isolation is enabled, the * memory monitors will work with the configured cgroup constraints to better * enforce the memory usage limit. - * @param memory_throttling_mode_enabled when enabled, the memory monitor will work - * with cgroup constraints to balance between memory throttling and worker killing to - * enforce stronger resource isolation between user and system slice. * @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. @@ -50,7 +43,6 @@ class MemoryMonitorFactory { static std::vector> Create( KillWorkersCallback kill_workers_callback, bool resource_isolation_enabled, - bool memory_throttling_mode_enabled, const CgroupManagerInterface &cgroup_manager); }; diff --git a/src/ray/common/memory_monitor_utils.cc b/src/ray/common/memory_monitor_utils.cc index f8afe9feed5c..ae4a4c17118a 100644 --- a/src/ray/common/memory_monitor_utils.cc +++ b/src/ray/common/memory_monitor_utils.cc @@ -298,7 +298,6 @@ int64_t MemoryMonitorUtils::GetMemoryThreshold( float usage_threshold, int64_t min_memory_free_bytes, bool resource_isolation_enabled, - bool memory_throttling_mode_enabled, const CgroupManagerInterface &cgroup_manager) { RAY_CHECK_GE(total_memory_bytes, MemoryMonitorInterface::kNull); RAY_CHECK_GE(min_memory_free_bytes, MemoryMonitorInterface::kNull); @@ -320,11 +319,7 @@ int64_t MemoryMonitorUtils::GetMemoryThreshold( if (resource_isolation_enabled) { StatusOr user_slice_upper_bound_bytes_or = - cgroup_manager.GetUserCgroupConstraintValue("memory.max"); - if (memory_throttling_mode_enabled) { - user_slice_upper_bound_bytes_or = - cgroup_manager.GetUserCgroupConstraintValue("memory.high"); - } + 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. " @@ -332,7 +327,7 @@ int64_t MemoryMonitorUtils::GetMemoryThreshold( cgroup_manager.GetUserCgroupPath(), user_slice_upper_bound_bytes_or.ToString()); std::string user_slice_upper_bound_bytes_str = - std::string(absl::StripAsciiWhitespace(user_slice_upper_bound_bytes_or.value())); + 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?", @@ -342,24 +337,7 @@ int64_t MemoryMonitorUtils::GetMemoryThreshold( std::all_of(user_slice_upper_bound_bytes_str.begin(), user_slice_upper_bound_bytes_str.end(), ::isdigit)) { - int64_t user_slice_upper_bound_bytes = std::stoll(user_slice_upper_bound_bytes_str); - if (memory_throttling_mode_enabled) { - resolved_memory_threshold_bytes = user_slice_upper_bound_bytes; - } else { - 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_slice_upper_bound_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_slice_upper_bound_bytes, - reaction_buffer_bytes); - } + resolved_memory_threshold_bytes = std::stoll(user_slice_upper_bound_bytes_str); } } diff --git a/src/ray/common/memory_monitor_utils.h b/src/ray/common/memory_monitor_utils.h index 50a652cf6385..cc6e49596123 100644 --- a/src/ray/common/memory_monitor_utils.h +++ b/src/ray/common/memory_monitor_utils.h @@ -73,8 +73,6 @@ class MemoryMonitorUtils { * @param resource_isolation_enabled Whether resource isolation is enabled. Used * to determine if the threshold should be calculated based on the cgroup * constraints. - * @param memory_throttling_mode_enabled When true and resource isolation is enabled, - * compute the memory threshold for memory throttling mode. * @param cgroup_manager The cgroup manager to fetch the upper bound memory constraints * from. * @return The memory threshold. @@ -83,7 +81,6 @@ class MemoryMonitorUtils { float usage_threshold, int64_t min_memory_free_bytes, bool resource_isolation_enabled, - bool memory_throttling_mode_enabled, const CgroupManagerInterface &cgroup_manager); /** diff --git a/src/ray/common/multi_memory_monitor_factory.cc b/src/ray/common/multi_memory_monitor_factory.cc index 1a4ba8ac069b..8ba945b3cab7 100644 --- a/src/ray/common/multi_memory_monitor_factory.cc +++ b/src/ray/common/multi_memory_monitor_factory.cc @@ -31,7 +31,6 @@ namespace ray { std::vector> MemoryMonitorFactory::Create( KillWorkersCallback kill_workers_callback, bool resource_isolation_enabled, - bool memory_throttling_mode_enabled, const CgroupManagerInterface &cgroup_manager) { std::vector> monitors; @@ -44,55 +43,17 @@ std::vector> MemoryMonitorFactory::Creat RayConfig::instance().memory_usage_threshold(), RayConfig::instance().min_memory_free_bytes(), resource_isolation_enabled, - memory_throttling_mode_enabled, cgroup_manager); - if (resource_isolation_enabled) { - std::string user_cgroup_path = cgroup_manager.GetUserCgroupPath(); - - if (memory_throttling_mode_enabled) { - StatusSetOr, StatusT::IOError> - event_monitor_or = EventMemoryMonitor::Create(user_cgroup_path, - std::move(kill_workers_callback)); - RAY_CHECK(event_monitor_or.has_value()) - << "Failed to create EventMemoryMonitor: " << event_monitor_or.message(); - monitors.push_back(std::move(event_monitor_or.value())); - } else { - if (monitor_interval_ms > 0) { - monitors.push_back(std::make_unique( - kill_workers_callback, memory_usage_threshold_bytes, monitor_interval_ms)); - } else { - RAY_LOG(INFO) - << "ThresholdMemoryMonitor disabled. Relying on PressureMemoryMonitor only. " - << "Specify `RAY_memory_monitor_refresh_ms` > 0 to enable the " - "ThresholdMemoryMonitor."; - } - - MemoryPsi pressure_threshold{ - PressureMemoryMonitor::kDefaultMemoryPsiMonitoringMode, - PressureMemoryMonitor::kDefaultMemoryPsiStallProportion, - PressureMemoryMonitor::kDefaultMemoryPsiStallDurationS}; - - StatusSetOr, - StatusT::InvalidArgument, - StatusT::IOError> - pressure_monitor_or = PressureMemoryMonitor::Create( - pressure_threshold, user_cgroup_path, std::move(kill_workers_callback)); - RAY_CHECK(pressure_monitor_or.has_value()) - << "Failed to create PressureMemoryMonitor: " << pressure_monitor_or.message(); - monitors.push_back(std::move(pressure_monitor_or.value())); - } + if (monitor_interval_ms > 0) { + monitors.push_back( + std::make_unique(std::move(kill_workers_callback), + memory_usage_threshold_bytes, + monitor_interval_ms)); } else { - 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()); - } + RAY_LOG(INFO) << "ThresholdMemoryMonitor disabled. Specify " + << "`RAY_memory_monitor_refresh_ms` > 0 to enable the monitor."; + monitors.push_back(std::make_unique()); } return monitors; diff --git a/src/ray/common/noop_memory_monitor_factory.cc b/src/ray/common/noop_memory_monitor_factory.cc index 83b003263348..7d2882ded95b 100644 --- a/src/ray/common/noop_memory_monitor_factory.cc +++ b/src/ray/common/noop_memory_monitor_factory.cc @@ -24,7 +24,6 @@ namespace ray { std::vector> MemoryMonitorFactory::Create( KillWorkersCallback kill_workers_callback, bool resource_isolation_enabled, - bool memory_throttling_mode_enabled, const CgroupManagerInterface &cgroup_manager) { std::vector> monitors; monitors.push_back(std::make_unique()); diff --git a/src/ray/common/ray_config_def.h b/src/ray/common/ray_config_def.h index fbbfaeff1414..dbeea5ab98bc 100644 --- a/src/ray/common/ray_config_def.h +++ b/src/ray/common/ray_config_def.h @@ -110,18 +110,6 @@ RAY_CONFIG(int64_t, /// default time-based policy. RAY_CONFIG(bool, worker_killing_policy_by_group, false) -/// Whether to enable the memory throttling monitoring system. -/// When true, the memory monitor will work with cgroup constraints to -/// balance between memory throttling and worker killing to enforce -/// stronger resource isolation between user and system processes. -/// This mode restrict the available heap memory for user processes to -/// total system memory - system reserved memory - object store memory. -/// -/// This mode is only supported when resource isolation is enabled, and -/// should only be enabled if the workload is experiencing node deaths -/// due to memory pressure. -RAY_CONFIG(bool, enable_memory_throttling_mode, false) - /// The reserved memory bytes for system processes /// enforced via cgroup memory.min constraint which guarantees /// that the system processes' memory will not be reclaimed under any conditions. @@ -135,14 +123,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/memory_monitor_factory_test.cc b/src/ray/common/tests/memory_monitor_factory_test.cc index 48bcb02f2e0d..6c00b741c0ed 100644 --- a/src/ray/common/tests/memory_monitor_factory_test.cc +++ b/src/ray/common/tests/memory_monitor_factory_test.cc @@ -84,7 +84,6 @@ TEST_F(MemoryMonitorFactoryTest, std::vector> monitors = MemoryMonitorFactory::Create([]() {}, /*resource_isolation_enabled=*/false, - /*memory_throttling_mode_enabled=*/false, cgroup_manager); ASSERT_EQ(monitors.size(), 1u) << "Expected exactly one monitor"; @@ -92,44 +91,19 @@ TEST_F(MemoryMonitorFactoryTest, << "Expected the sole monitor to be a ThresholdMemoryMonitor"; } -TEST_F( - MemoryMonitorFactoryTest, - TestCreateWithResourceIsolationEnabledThrottlingDisabledReturnsPressureAndThresholdMonitors) { - FakeCgroupManager cgroup_manager(kUserMemoryMaxBytes, kUserMemoryHighBytes); - TempFile pressure_file(cgroup_manager.GetPath() + "/memory.pressure"); - - std::vector> monitors = - MemoryMonitorFactory::Create([]() {}, - /*resource_isolation_enabled=*/true, - /*memory_throttling_mode_enabled=*/false, - cgroup_manager); - - ASSERT_EQ(monitors.size(), 2u) - << "Expected ThresholdMemoryMonitor + PressureMemoryMonitor"; - bool has_threshold = false; - bool has_pressure = false; - for (const std::unique_ptr &m : monitors) { - if (dynamic_cast(m.get()) != nullptr) has_threshold = true; - if (dynamic_cast(m.get()) != nullptr) has_pressure = true; - } - EXPECT_TRUE(has_threshold) << "Expected a ThresholdMemoryMonitor"; - EXPECT_TRUE(has_pressure) << "Expected a PressureMemoryMonitor"; -} - TEST_F(MemoryMonitorFactoryTest, - TestCreateWithResourceIsolationEnabledThrottlingEnabledReturnsEventMonitor) { + TestCreateWithResourceIsolationEnabledReturnsThresholdMonitors) { FakeCgroupManager cgroup_manager(kUserMemoryMaxBytes, kUserMemoryHighBytes); - TempFile pressure_file(cgroup_manager.GetPath() + "/memory.events"); + TempFile pressure_file(cgroup_manager.GetPath() + "/memory.pressure"); std::vector> monitors = MemoryMonitorFactory::Create([]() {}, /*resource_isolation_enabled=*/true, - /*memory_throttling_mode_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 EventMemoryMonitor"; + 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 2746cc446d56..45e310773863 100644 --- a/src/ray/common/tests/memory_monitor_utils_test.cc +++ b/src/ray/common/tests/memory_monitor_utils_test.cc @@ -234,39 +234,36 @@ TEST_F(MemoryMonitorUtilsTest, TestCgroupNonexistentUsageFileReturnskNull) { TEST_F(MemoryMonitorUtilsTest, TestGetMemoryThresholdTakeGreaterOfTheTwoValues) { NoopCgroupManager noop_cgroup_manager; - ASSERT_EQ(MemoryMonitorUtils::GetMemoryThreshold( - 100, 0.5, 0, false, false, noop_cgroup_manager), - 100); - ASSERT_EQ(MemoryMonitorUtils::GetMemoryThreshold( - 100, 0.5, 60, false, false, noop_cgroup_manager), - 50); + ASSERT_EQ( + MemoryMonitorUtils::GetMemoryThreshold(100, 0.5, 0, false, noop_cgroup_manager), + 100); + ASSERT_EQ( + MemoryMonitorUtils::GetMemoryThreshold(100, 0.5, 60, false, noop_cgroup_manager), + 50); - ASSERT_EQ(MemoryMonitorUtils::GetMemoryThreshold( - 100, 1, 10, false, false, noop_cgroup_manager), - 100); - ASSERT_EQ(MemoryMonitorUtils::GetMemoryThreshold( - 100, 1, 100, false, false, noop_cgroup_manager), - 100); + ASSERT_EQ( + MemoryMonitorUtils::GetMemoryThreshold(100, 1, 10, false, noop_cgroup_manager), + 100); + ASSERT_EQ( + MemoryMonitorUtils::GetMemoryThreshold(100, 1, 100, false, noop_cgroup_manager), + 100); - ASSERT_EQ(MemoryMonitorUtils::GetMemoryThreshold( - 100, 0.1, 100, false, false, noop_cgroup_manager), - 10); - ASSERT_EQ(MemoryMonitorUtils::GetMemoryThreshold( - 100, 0, 10, false, false, noop_cgroup_manager), - 90); - ASSERT_EQ(MemoryMonitorUtils::GetMemoryThreshold( - 100, 0, 100, false, false, noop_cgroup_manager), - 0); + ASSERT_EQ( + MemoryMonitorUtils::GetMemoryThreshold(100, 0.1, 100, false, noop_cgroup_manager), + 10); + ASSERT_EQ( + MemoryMonitorUtils::GetMemoryThreshold(100, 0, 10, false, noop_cgroup_manager), 90); + ASSERT_EQ( + MemoryMonitorUtils::GetMemoryThreshold(100, 0, 100, false, noop_cgroup_manager), 0); ASSERT_EQ(MemoryMonitorUtils::GetMemoryThreshold( - 100, 0, MemoryMonitorInterface::kNull, false, false, noop_cgroup_manager), + 100, 0, MemoryMonitorInterface::kNull, false, noop_cgroup_manager), 0); - ASSERT_EQ( - MemoryMonitorUtils::GetMemoryThreshold( - 100, 0.5, MemoryMonitorInterface::kNull, false, false, noop_cgroup_manager), - 50); ASSERT_EQ(MemoryMonitorUtils::GetMemoryThreshold( - 100, 1, MemoryMonitorInterface::kNull, false, false, noop_cgroup_manager), + 100, 0.5, MemoryMonitorInterface::kNull, false, noop_cgroup_manager), + 50); + ASSERT_EQ(MemoryMonitorUtils::GetMemoryThreshold( + 100, 1, MemoryMonitorInterface::kNull, false, noop_cgroup_manager), 100); } @@ -302,26 +299,14 @@ TEST_F( // Reaction buffer defaults to 5% of total memory. If // kDefaultThresholdMonitorReactionBufferProportion is changed, this should be changed // accordingly. - int64_t expected_default_mode_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, - /*memory_throttling_mode_enabled=*/false, *cgroup_manager), expected_default_mode_threshold); - - int64_t expected_throttling_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, - /*memory_throttling_mode_enabled=*/true, - *cgroup_manager), - expected_throttling_mode_threshold); } TEST_F(MemoryMonitorUtilsTest, TestGetPidsFromDirOnlyReturnsNumericFilenames) { diff --git a/src/ray/common/tests/threshold_memory_monitor_test.cc b/src/ray/common/tests/threshold_memory_monitor_test.cc index 74bf6e7f2716..462239db55b2 100644 --- a/src/ray/common/tests/threshold_memory_monitor_test.cc +++ b/src/ray/common/tests/threshold_memory_monitor_test.cc @@ -74,7 +74,7 @@ TEST_F(ThresholdMemoryMonitorTest, NoopCgroupManager noop_cgroup_manager; int64_t memory_usage_threshold_bytes = MemoryMonitorUtils::GetMemoryThreshold( - cgroup_total_bytes, 0.7f, -1, false, false, noop_cgroup_manager); + cgroup_total_bytes, 0.7f, -1, false, noop_cgroup_manager); MakeThresholdMemoryMonitor( memory_usage_threshold_bytes, // (70%) 1 /*refresh_interval_ms*/, @@ -100,7 +100,7 @@ TEST_F(ThresholdMemoryMonitorTest, NoopCgroupManager noop_cgroup_manager; int64_t memory_usage_threshold_bytes = MemoryMonitorUtils::GetMemoryThreshold( - cgroup_total_bytes, 0.7f, -1, false, false, noop_cgroup_manager); + cgroup_total_bytes, 0.7f, -1, false, noop_cgroup_manager); MakeThresholdMemoryMonitor( memory_usage_threshold_bytes, // (70%) 1 /*refresh_interval_ms*/, diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc index 4ee15cd764aa..53b80241c1bf 100644 --- a/src/ray/raylet/node_manager.cc +++ b/src/ray/raylet/node_manager.cc @@ -244,11 +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_monitors_(MemoryMonitorFactory::Create( - CreateKillWorkersCallback(), - config.enable_resource_isolation, - RayConfig::instance().enable_memory_throttling_mode(), - *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), @@ -3113,7 +3111,6 @@ KillWorkersCallback NodeManager::CreateKillWorkersCallback() { RayConfig::instance().memory_usage_threshold(), RayConfig::instance().min_memory_free_bytes(), initial_config_.enable_resource_isolation, - RayConfig::instance().enable_memory_throttling_mode(), *cgroup_manager_); float computed_threshold_fraction = static_cast(computed_threshold_bytes) / diff --git a/src/ray/raylet/worker_killing_policy_factory.cc b/src/ray/raylet/worker_killing_policy_factory.cc index e0aea50af80f..ed3e3f1196a1 100644 --- a/src/ray/raylet/worker_killing_policy_factory.cc +++ b/src/ray/raylet/worker_killing_policy_factory.cc @@ -41,7 +41,6 @@ std::unique_ptr WorkerKillingPolicyFactory::Create RayConfig::instance().memory_usage_threshold(), RayConfig::instance().min_memory_free_bytes(), resource_isolation_enabled, - RayConfig::instance().enable_memory_throttling_mode(), cgroup_manager); int64_t kill_memory_buffer_bytes = From 603d98abc795ce7a9a344833bcf442d2ba1dc49a Mon Sep 17 00:00:00 2001 From: davik Date: Tue, 28 Apr 2026 00:15:31 +0000 Subject: [PATCH 5/5] Remove reaction buffer and unneeded dependencies Signed-off-by: davik --- src/ray/common/BUILD.bazel | 3 --- src/ray/common/multi_memory_monitor_factory.cc | 3 --- src/ray/common/ray_config_def.h | 17 ++++------------- src/ray/common/tests/BUILD.bazel | 3 --- .../common/tests/memory_monitor_factory_test.cc | 7 +------ .../common/tests/memory_monitor_utils_test.cc | 3 --- 6 files changed, 5 insertions(+), 31 deletions(-) diff --git a/src/ray/common/BUILD.bazel b/src/ray/common/BUILD.bazel index d2d4727dae1c..700318572aea 100644 --- a/src/ray/common/BUILD.bazel +++ b/src/ray/common/BUILD.bazel @@ -246,12 +246,9 @@ ray_cc_library( "//src/ray/common/cgroup2:cgroup_manager_interface", ] + select({ "//bazel:is_linux": [ - ":event_memory_monitor", ":memory_monitor_utils", ":noop_memory_monitor", - ":pressure_memory_monitor", ":ray_config", - ":status_or", ":threshold_memory_monitor", "//src/ray/util:logging", ], diff --git a/src/ray/common/multi_memory_monitor_factory.cc b/src/ray/common/multi_memory_monitor_factory.cc index 8ba945b3cab7..8018b78238ee 100644 --- a/src/ray/common/multi_memory_monitor_factory.cc +++ b/src/ray/common/multi_memory_monitor_factory.cc @@ -15,14 +15,11 @@ #include #include -#include "ray/common/event_memory_monitor.h" #include "ray/common/memory_monitor_factory.h" #include "ray/common/memory_monitor_interface.h" #include "ray/common/memory_monitor_utils.h" #include "ray/common/noop_memory_monitor.h" -#include "ray/common/pressure_memory_monitor.h" #include "ray/common/ray_config.h" -#include "ray/common/status_or.h" #include "ray/common/threshold_memory_monitor.h" #include "ray/util/logging.h" diff --git a/src/ray/common/ray_config_def.h b/src/ray/common/ray_config_def.h index dbeea5ab98bc..fc8487e8f43b 100644 --- a/src/ray/common/ray_config_def.h +++ b/src/ray/common/ray_config_def.h @@ -71,11 +71,11 @@ 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. @@ -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) diff --git a/src/ray/common/tests/BUILD.bazel b/src/ray/common/tests/BUILD.bazel index 7375c4914884..0578d062f789 100644 --- a/src/ray/common/tests/BUILD.bazel +++ b/src/ray/common/tests/BUILD.bazel @@ -202,14 +202,11 @@ ray_cc_test( "@platforms//os:linux", ], deps = [ - "//src/ray/common:event_memory_monitor", "//src/ray/common:memory_monitor_factory", "//src/ray/common:memory_monitor_interface", - "//src/ray/common:pressure_memory_monitor", "//src/ray/common:threshold_memory_monitor", "//src/ray/common/cgroup2:cgroup_manager_interface", "//src/ray/common/cgroup2:cgroup_test_utils", - "//src/ray/common/cgroup2:noop_cgroup_manager", ], ) diff --git a/src/ray/common/tests/memory_monitor_factory_test.cc b/src/ray/common/tests/memory_monitor_factory_test.cc index 6c00b741c0ed..8e04862b36dd 100644 --- a/src/ray/common/tests/memory_monitor_factory_test.cc +++ b/src/ray/common/tests/memory_monitor_factory_test.cc @@ -16,16 +16,12 @@ #include #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/cgroup2/noop_cgroup_manager.h" -#include "ray/common/event_memory_monitor.h" #include "ray/common/memory_monitor_interface.h" -#include "ray/common/pressure_memory_monitor.h" #include "ray/common/threshold_memory_monitor.h" namespace ray { @@ -72,9 +68,8 @@ class FakeCgroupManager : public CgroupManagerInterface { class MemoryMonitorFactoryTest : public ::testing::Test { protected: - // Large enough that threshold (memory.max - reaction_buffer) is > 0. static constexpr int64_t kUserMemoryMaxBytes = 10LL * 1024 * 1024 * 1024; // 10 GB - static constexpr int64_t kUserMemoryHighBytes = 8LL * 1024 * 1024 * 1024; // 5 GB + static constexpr int64_t kUserMemoryHighBytes = 8LL * 1024 * 1024 * 1024; // 8 GB }; TEST_F(MemoryMonitorFactoryTest, diff --git a/src/ray/common/tests/memory_monitor_utils_test.cc b/src/ray/common/tests/memory_monitor_utils_test.cc index 45e310773863..7c2470a88b9a 100644 --- a/src/ray/common/tests/memory_monitor_utils_test.cc +++ b/src/ray/common/tests/memory_monitor_utils_test.cc @@ -296,9 +296,6 @@ TEST_F( 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_default_mode_threshold = user_memory_high_bytes; ASSERT_EQ(MemoryMonitorUtils::GetMemoryThreshold( /*total_memory_bytes=*/16LL * 1024 * 1024 * 1024,