Skip to content

[Core] (Resource Isolation 13/n) Introduce Multi Memory Monitor Factory to create new memory monitoring system #62705

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
MengjinYan merged 8 commits into from
Apr 29, 2026
Merged

[Core] (Resource Isolation 13/n) Introduce Multi Memory Monitor Factory to create new memory monitoring system #62705

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
a17f625
Wire together memory monitors through multi memory monitor factory an…
Apr 19, 2026
9c7a9ad
Lock worker killing mutex until all monitor states are updated
Apr 23, 2026
64efec0
Improve comment clarity
Apr 23, 2026
6fd9952
Wire multi-memmon just for memory.high
Apr 27, 2026
603d98a
Remove reaction buffer and unneeded dependencies
Apr 28, 2026
7fdb882
Merge branch 'master' into multi_monitor
Kunchd Apr 28, 2026
9ef8c05
Merge branch 'master' into multi_monitor
Kunchd Apr 28, 2026
db6dd99
Merge branch 'master' into multi_monitor
Kunchd Apr 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions src/ray/common/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand All @@ -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",
],
}),
)

Expand Down
12 changes: 8 additions & 4 deletions src/ray/common/cgroup2/linux_cgroup_manager_factory.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
#include <sys/types.h>
#include <unistd.h>

#include <algorithm>
#include <memory>
#include <string>
#include <utility>
Expand Down Expand Up @@ -66,11 +67,14 @@ std::unique_ptr<CgroupManagerInterface> 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<int64_t>(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<int64_t>(total_memory_bytes * user_memory_proportion_max));
static_cast<int64_t>(total_memory_bytes * user_memory_proportion_high));
int64_t user_memory_max_bytes =
static_cast<int64_t>(total_memory_bytes * user_memory_proportion_max);

StatusOr<std::unique_ptr<CgroupManagerInterface>> cgroup_manager_s =
CgroupManager::Create(cgroup_path,
node_id,
Expand Down
3 changes: 1 addition & 2 deletions src/ray/common/event_memory_monitor.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
26 changes: 10 additions & 16 deletions src/ray/common/memory_monitor_factory.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,38 +15,32 @@
#pragma once

#include <memory>
#include <vector>

#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.

@Kunchd Kunchd Apr 28, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Note: The event memory monitor will be wired in in a future PR. This PR will simply wire in the threshold monitor to maintain similar behavior with the original default_mode monitoring behavior.

*
* 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<MemoryMonitorInterface> Create(
static std::vector<std::unique_ptr<MemoryMonitorInterface>> Create(
KillWorkersCallback kill_workers_callback,
bool resource_isolation_enabled,
const CgroupManagerInterface &cgroup_manager);
Expand Down
6 changes: 2 additions & 4 deletions src/ray/common/memory_monitor_interface.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,11 +48,9 @@ struct SystemMemorySnapshot {
using ProcessesMemorySnapshot = absl::flat_hash_map<pid_t, int64_t>;

/**
* @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<void(SystemMemorySnapshot system_memory)>;
using KillWorkersCallback = std::function<void()>;

@Kunchd Kunchd Apr 22, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We are moving taking the memory snapshot logic to decide how many workers to kill to the node manager callback. This is because there could be a delay between when the memory monitor triggers the callback to when the node manager actually execute the callback. We might to make sure the decision for how many workers to kill is made at the time of the kill.


/**
* @brief implementations of this interface monitors the memory usage of the node
Expand Down
42 changes: 19 additions & 23 deletions src/ray/common/memory_monitor_utils.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -318,30 +318,26 @@ int64_t MemoryMonitorUtils::GetMemoryThreshold(
}

if (resource_isolation_enabled) {
StatusOr<std::string> 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<std::string> 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());
Comment thread
Kunchd marked this conversation as resolved.

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<int64_t>(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);
Comment thread
Kunchd marked this conversation as resolved.
resolved_memory_threshold_bytes = std::stoll(user_slice_upper_bound_bytes_str);

@Kunchd Kunchd Apr 28, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Now that we have switched to using memory.high, we no longer need a reaction buffer as the kernel will not oom kill as soon as the high boundary is met.

}
}

Expand Down
31 changes: 17 additions & 14 deletions ...ommon/threshold_memory_monitor_factory.cc → ...ay/common/multi_memory_monitor_factory.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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.
Expand All @@ -12,8 +12,8 @@
// See the License for the specific language governing permissions and
// limitations under the License.

#include <algorithm>
#include <memory>
#include <vector>

#include "ray/common/memory_monitor_factory.h"
#include "ray/common/memory_monitor_interface.h"
Expand All @@ -25,32 +25,35 @@

namespace ray {

std::unique_ptr<MemoryMonitorInterface> MemoryMonitorFactory::Create(
std::vector<std::unique_ptr<MemoryMonitorInterface>> MemoryMonitorFactory::Create(
KillWorkersCallback kill_workers_callback,
bool resource_isolation_enabled,
const CgroupManagerInterface &cgroup_manager) {
int64_t memory_usage_threshold_bytes;
std::vector<std::unique_ptr<MemoryMonitorInterface>> 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<NoopMemoryMonitor>();
}

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<ThresholdMemoryMonitor>(std::move(kill_workers_callback),
memory_usage_threshold_bytes,
monitor_interval_ms);
if (monitor_interval_ms > 0) {
monitors.push_back(
std::make_unique<ThresholdMemoryMonitor>(std::move(kill_workers_callback),
Comment thread
cursor[bot] marked this conversation as resolved.
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<NoopMemoryMonitor>());
}

return monitors;
}

} // namespace ray
7 changes: 5 additions & 2 deletions src/ray/common/noop_memory_monitor_factory.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,18 +13,21 @@
// limitations under the License.

#include <memory>
#include <vector>

#include "ray/common/memory_monitor_factory.h"
#include "ray/common/memory_monitor_interface.h"
#include "ray/common/noop_memory_monitor.h"

namespace ray {

std::unique_ptr<MemoryMonitorInterface> MemoryMonitorFactory::Create(
std::vector<std::unique_ptr<MemoryMonitorInterface>> MemoryMonitorFactory::Create(
KillWorkersCallback kill_workers_callback,
bool resource_isolation_enabled,
const CgroupManagerInterface &cgroup_manager) {
return std::make_unique<NoopMemoryMonitor>();
std::vector<std::unique_ptr<MemoryMonitorInterface>> monitors;
monitors.push_back(std::make_unique<NoopMemoryMonitor>());
return monitors;
}

} // namespace ray
3 changes: 1 addition & 2 deletions src/ray/common/pressure_memory_monitor.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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. "
Expand Down
14 changes: 14 additions & 0 deletions src/ray/common/pressure_memory_monitor.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
26 changes: 10 additions & 16 deletions src/ray/common/ray_config_def.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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)
Expand All @@ -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
Expand Down
22 changes: 22 additions & 0 deletions src/ray/common/tests/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down
Loading
Loading