Revert "Revert "[core] (cgroups 6/n) CgroupManager cleans up the entire cgroup hierarchy in reverse order when destroyed. (ray-project#56260)""

This reverts commit c4c774e.

Author: edoakes

python/ray/_private/worker.py

@@ -2440,7 +2440,6 @@ def is_initialized() -> bool:
     return ray._private.worker.global_worker.connected
 
 
-# TODO(hjiang): Add cgroup path along with [enable_resource_isolation].
 @with_connect_or_shutdown_lock
 def connect(
     node,
@@ -2459,7 +2458,6 @@ def connect(
     worker_launch_time_ms: int = -1,
     worker_launched_time_ms: int = -1,
     debug_source: str = "",
-    enable_resource_isolation: bool = False,
 ):
     """Connect this worker to the raylet, to Plasma, and to GCS.
 
@@ -2488,7 +2486,6 @@ def connect(
             finshes launching. If the worker is not launched by raylet (e.g.,
             driver), this must be -1 (default value).
         debug_source: Source information for `CoreWorker`, used for debugging and informational purpose, rather than functional purpose.
-        enable_resource_isolation: If true, core worker enables resource isolation by adding itself into appropriate cgroup.
     """
     # Do some basic checking to make sure we didn't call ray.init twice.
     error_message = "Perhaps you called ray.init twice by accident?"
@@ -2667,7 +2664,6 @@ def connect(
         worker_launch_time_ms,
         worker_launched_time_ms,
         debug_source,
-        enable_resource_isolation,
     )
 
     if mode == SCRIPT_MODE:

python/ray/_private/workers/default_worker.py

@@ -271,7 +271,6 @@
         ray_debugger_external=args.ray_debugger_external,
         worker_launch_time_ms=args.worker_launch_time_ms,
         worker_launched_time_ms=worker_launched_time_ms,
-        enable_resource_isolation=args.enable_resource_isolation,
     )
 
     worker = ray._private.worker.global_worker

python/ray/_raylet.pyx

@@ -3005,7 +3005,7 @@ cdef class CoreWorker:
                   local_mode, driver_name,
                   serialized_job_config, metrics_agent_port, runtime_env_hash,
                   startup_token, session_name, cluster_id, entrypoint,
-                  worker_launch_time_ms, worker_launched_time_ms, debug_source, enable_resource_isolation):
+                  worker_launch_time_ms, worker_launched_time_ms, debug_source):
         self.is_local_mode = local_mode
 
         cdef CCoreWorkerOptions options = CCoreWorkerOptions()
@@ -3061,7 +3061,6 @@ cdef class CoreWorker:
         options.worker_launch_time_ms = worker_launch_time_ms
         options.worker_launched_time_ms = worker_launched_time_ms
         options.debug_source = debug_source
-        options.enable_resource_isolation = enable_resource_isolation
         CCoreWorkerProcess.Initialize(options)
 
         self.cgname_to_eventloop_dict = None

python/ray/includes/libcoreworker.pxd

@@ -440,7 +440,6 @@ cdef extern from "ray/core_worker/core_worker.h" nogil:
         int64_t worker_launch_time_ms
         int64_t worker_launched_time_ms
         c_string debug_source
-        c_bool enable_resource_isolation
 
     cdef cppclass CCoreWorkerProcess "ray::core::CoreWorkerProcess":
         @staticmethod

src/ray/common/cgroup2/BUILD.bazel

@@ -1,13 +1,40 @@
 load("//bazel:ray.bzl", "ray_cc_library")
 
+config_setting(
+    name = "is_linux",
+    constraint_values = ["@platforms//os:linux"],
+)
+
+# Public targets.
+ray_cc_library(
+    name = "cgroup_manager",
+    srcs = select({
+        ":is_linux": ["cgroup_manager.cc"],
+        "//conditions:default": ["noop_cgroup_manager.cc"],
+    }),
+    hdrs = ["cgroup_manager.h"],
+    visibility = ["//visibility:public"],
+    deps = [
+        ":cgroup_driver_interface",
+        ":cgroup_manager_interface",
+        "//src/ray/common:status",
+        "//src/ray/common:status_or",
+    ] + select({
+        ":is_linux": [
+            ":scoped_cgroup_operation",
+            "//src/ray/util:logging",
+            "@com_google_absl//absl/strings",
+        ],
+        "//conditions:default": [],
+    }),
+)
+
 ray_cc_library(
     name = "cgroup_driver_interface",
     hdrs = [
         "cgroup_driver_interface.h",
     ],
-    target_compatible_with = [
-        "@platforms//os:linux",
-    ],
+    visibility = ["//visibility:public"],
     deps = [
         "//src/ray/common:status",
         "//src/ray/common:status_or",
@@ -19,51 +46,42 @@ ray_cc_library(
     hdrs = [
         "cgroup_manager_interface.h",
     ],
-    target_compatible_with = [
-        "@platforms//os:linux",
-    ],
+    visibility = ["//visibility:public"],
     deps = [
         "//src/ray/common:status",
         "//src/ray/common:status_or",
     ],
 )
 
 ray_cc_library(
-    name = "cgroup_manager",
-    srcs = ["cgroup_manager.cc"],
+    name = "sysfs_cgroup_driver",
+    srcs = ["sysfs_cgroup_driver.cc"],
     hdrs = [
-        "cgroup_manager.h",
-        "scoped_cgroup_operation.h",
+        "sysfs_cgroup_driver.h",
     ],
     target_compatible_with = [
         "@platforms//os:linux",
     ],
+    visibility = ["//visibility:public"],
     deps = [
         ":cgroup_driver_interface",
-        ":cgroup_manager_interface",
         "//src/ray/common:status",
         "//src/ray/common:status_or",
         "//src/ray/util:logging",
         "@com_google_absl//absl/strings",
     ],
 )
 
+# Private Targets.
 ray_cc_library(
-    name = "sysfs_cgroup_driver",
-    srcs = ["sysfs_cgroup_driver.cc"],
+    name = "scoped_cgroup_operation",
     hdrs = [
-        "sysfs_cgroup_driver.h",
+        "scoped_cgroup_operation.h",
     ],
     target_compatible_with = [
         "@platforms//os:linux",
     ],
-    deps = [
-        ":cgroup_driver_interface",
-        "//src/ray/common:status",
-        "//src/ray/common:status_or",
-        "//src/ray/util:logging",
-        "@com_google_absl//absl/strings",
-    ],
+    visibility = [":__subpackages__"],
 )
 
 ray_cc_library(
@@ -74,6 +92,7 @@ ray_cc_library(
     target_compatible_with = [
         "@platforms//os:linux",
     ],
+    visibility = [":__subpackages__"],
     deps = [
         ":cgroup_driver_interface",
         "//src/ray/common:status",
@@ -87,6 +106,7 @@ ray_cc_library(
     target_compatible_with = [
         "@platforms//os:linux",
     ],
+    visibility = [":__subpackages__"],
     deps = [
         "//src/ray/common:id",
         "//src/ray/common:status",

src/ray/common/cgroup2/cgroup_manager.cc

@@ -113,48 +113,65 @@ StatusOr<std::unique_ptr<CgroupManager>> CgroupManager::Create(
   return cgroup_manager;
 }
 
-// TODO(#54703): This is a placeholder for cleanup. This will call
-// CgroupDriver::DeleteCgroup.
 void CgroupManager::RegisterDeleteCgroup(const std::string &cgroup_path) {
-  cleanup_operations_.emplace_back([cgroup = cgroup_path]() {
-    RAY_LOG(INFO) << absl::StrFormat("Deleting all cgroup %s.", cgroup);
+  cleanup_operations_.emplace_back([this, cgroup = cgroup_path]() {
+    Status s = this->cgroup_driver_->DeleteCgroup(cgroup);
+    if (!s.ok()) {
+      RAY_LOG(WARNING) << absl::StrFormat(
+          "Failed to delete cgroup %s with error %s.", cgroup, s.ToString());
+    }
   });
 }
 
-// TODO(#54703): This is a placeholder for cleanup. This will call
-// CgroupDriver::MoveAllProcesses.
 void CgroupManager::RegisterMoveAllProcesses(const std::string &from,
                                              const std::string &to) {
-  cleanup_operations_.emplace_back([from_cgroup = from, to_cgroup = to]() {
-    RAY_LOG(INFO) << absl::StrFormat(
-        "Moved All Processes from %s to %s.", from_cgroup, to_cgroup);
+  cleanup_operations_.emplace_back([this, from_cgroup = from, to_cgroup = to]() {
+    Status s = this->cgroup_driver_->MoveAllProcesses(from_cgroup, to_cgroup);
+    if (!s.ok()) {
+      RAY_LOG(WARNING) << absl::StrFormat(
+          "Failed to move all processes from %s to %s with error %s",
+          from_cgroup,
+          to_cgroup,
+          s.ToString());
+    }
   });
 }
 
-// TODO(#54703): This is a placeholder for cleanup. This will call
-// CgroupDriver::AddConstraint(cgroup, constraint, default_value).
 template <typename T>
 void CgroupManager::RegisterRemoveConstraint(const std::string &cgroup,
                                              const Constraint<T> &constraint) {
   cleanup_operations_.emplace_back(
-      [constrained_cgroup = cgroup, constraint_to_remove = constraint]() {
-        RAY_LOG(INFO) << absl::StrFormat(
-            "Setting constraint %s to default value %lld for cgroup %s",
-            constraint_to_remove.name_,
-            constraint_to_remove.default_value_,
-            constrained_cgroup);
+      [this, constrained_cgroup = cgroup, constraint_to_remove = constraint]() {
+        std::string default_value = std::to_string(constraint_to_remove.default_value_);
+        Status s = this->cgroup_driver_->AddConstraint(constrained_cgroup,
+                                                       constraint_to_remove.controller_,
+                                                       constraint_to_remove.name_,
+                                                       default_value);
+        if (!s.ok()) {
+          RAY_LOG(WARNING) << absl::StrFormat(
+              "Failed to set constraint %s=%s to default value for cgroup %s with error "
+              "%s.",
+              constraint_to_remove.name_,
+              default_value,
+              constrained_cgroup,
+              s.ToString());
+        }
       });
 }
 
-// TODO(#54703): This is a placeholder for cleanup. This will call
-// CgroupDriver::DisableController.
-void CgroupManager::RegisterDisableController(const std::string &cgroup,
+void CgroupManager::RegisterDisableController(const std::string &cgroup_path,
                                               const std::string &controller) {
-  cleanup_operations_.emplace_back([cgroup_to_clean = cgroup,
-                                    controller_to_disable = controller]() {
-    RAY_LOG(INFO) << absl::StrFormat(
-        "Disabling controller %s for cgroup %s.", controller_to_disable, cgroup_to_clean);
-  });
+  cleanup_operations_.emplace_back(
+      [this, cgroup = cgroup_path, controller_to_disable = controller]() {
+        Status s = this->cgroup_driver_->DisableController(cgroup, controller_to_disable);
+        if (!s.ok()) {
+          RAY_LOG(WARNING) << absl::StrFormat(
+              "Failed to disable controller %s for cgroup %s with error %s",
+              controller_to_disable,
+              cgroup,
+              s.ToString());
+        }
+      });
 }
 
 Status CgroupManager::Initialize(int64_t system_reserved_cpu_weight,
@@ -168,11 +185,11 @@ Status CgroupManager::Initialize(int64_t system_reserved_cpu_weight,
       cpu_weight_constraint_.Max() - system_reserved_cpu_weight;
 
   RAY_LOG(INFO) << absl::StrFormat(
-      "Initializing CgroupManager at base cgroup path at %s. Ray's cgroup "
-      "hierarchy will under the node cgroup %s. The %s controllers will be "
+      "Initializing CgroupManager at base cgroup at '%s'. Ray's cgroup "
+      "hierarchy will under the node cgroup at '%s'. The %s controllers will be "
       "enabled. "
-      "System cgroup %s will have constraints [%s=%lld, %s=%lld]. "
-      "Application cgroup %s will have constraints [%s=%lld].",
+      "The system cgroup at '%s' will have constraints [%s=%lld, %s=%lld]. "
+      "The application cgroup '%s' will have constraints [%s=%lld].",
       base_cgroup_path_,
       node_cgroup_path_,
       supported_controllers,

src/ray/common/cgroup2/cgroup_manager.h

@@ -111,9 +111,10 @@ class CgroupManager : public CgroupManagerInterface {
   Status Initialize(const int64_t system_reserved_cpu_weight,
                     const int64_t system_reserved_memory_bytes);
 
-  // TODO(#54703): This is a placeholder for cleanup. This will be implemented in the a
-  // future PR.
-  void RegisterDeleteCgroup(const std::string &cgroup_path);
+  // The Register* methods register a callback that will execute in the destructor
+  // in FILO order. All callbacks required the cgroup_driver_ to be available to
+  // remove the cgroup hierarchy.
+  void RegisterDeleteCgroup(const std::string &cgroup);
   void RegisterMoveAllProcesses(const std::string &from, const std::string &to);
   template <typename T>
   void RegisterRemoveConstraint(const std::string &cgroup,