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

This reverts commit f900567.

Author: edoakes

python/ray/_private/worker.py

@@ -2440,6 +2440,7 @@ 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,
@@ -2458,6 +2459,7 @@ 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.
 
@@ -2486,6 +2488,7 @@ 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?"
@@ -2664,6 +2667,7 @@ 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,6 +271,7 @@
         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

@@ -3002,7 +3002,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):
+                  worker_launch_time_ms, worker_launched_time_ms, debug_source, enable_resource_isolation):
         self.is_local_mode = local_mode
 
         cdef CCoreWorkerOptions options = CCoreWorkerOptions()
@@ -3058,6 +3058,7 @@ 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

@@ -439,6 +439,7 @@ 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,40 +1,13 @@
 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",
     ],
-    visibility = ["//visibility:public"],
+    target_compatible_with = [
+        "@platforms//os:linux",
+    ],
     deps = [
         "//src/ray/common:status",
         "//src/ray/common:status_or",
@@ -46,42 +19,51 @@ ray_cc_library(
     hdrs = [
         "cgroup_manager_interface.h",
     ],
-    visibility = ["//visibility:public"],
+    target_compatible_with = [
+        "@platforms//os:linux",
+    ],
     deps = [
         "//src/ray/common:status",
         "//src/ray/common:status_or",
     ],
 )
 
 ray_cc_library(
-    name = "sysfs_cgroup_driver",
-    srcs = ["sysfs_cgroup_driver.cc"],
+    name = "cgroup_manager",
+    srcs = ["cgroup_manager.cc"],
     hdrs = [
-        "sysfs_cgroup_driver.h",
+        "cgroup_manager.h",
+        "scoped_cgroup_operation.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 = "scoped_cgroup_operation",
+    name = "sysfs_cgroup_driver",
+    srcs = ["sysfs_cgroup_driver.cc"],
     hdrs = [
-        "scoped_cgroup_operation.h",
+        "sysfs_cgroup_driver.h",
     ],
     target_compatible_with = [
         "@platforms//os:linux",
     ],
-    visibility = [":__subpackages__"],
+    deps = [
+        ":cgroup_driver_interface",
+        "//src/ray/common:status",
+        "//src/ray/common:status_or",
+        "//src/ray/util:logging",
+        "@com_google_absl//absl/strings",
+    ],
 )
 
 ray_cc_library(
@@ -92,7 +74,6 @@ ray_cc_library(
     target_compatible_with = [
         "@platforms//os:linux",
     ],
-    visibility = [":__subpackages__"],
     deps = [
         ":cgroup_driver_interface",
         "//src/ray/common:status",
@@ -106,7 +87,6 @@ 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,65 +113,48 @@ 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([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());
-    }
+  cleanup_operations_.emplace_back([cgroup = cgroup_path]() {
+    RAY_LOG(INFO) << absl::StrFormat("Deleting all cgroup %s.", cgroup);
   });
 }
 
+// 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([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());
-    }
+  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);
   });
 }
 
+// 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(
-      [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());
-        }
+      [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);
       });
 }
 
-void CgroupManager::RegisterDisableController(const std::string &cgroup_path,
+// TODO(#54703): This is a placeholder for cleanup. This will call
+// CgroupDriver::DisableController.
+void CgroupManager::RegisterDisableController(const std::string &cgroup,
                                               const std::string &controller) {
-  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());
-        }
-      });
+  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);
+  });
 }
 
 Status CgroupManager::Initialize(int64_t system_reserved_cpu_weight,
@@ -185,11 +168,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 at '%s'. Ray's cgroup "
-      "hierarchy will under the node cgroup at '%s'. The %s controllers will be "
+      "Initializing CgroupManager at base cgroup path at %s. Ray's cgroup "
+      "hierarchy will under the node cgroup %s. The %s controllers will be "
       "enabled. "
-      "The system cgroup at '%s' will have constraints [%s=%lld, %s=%lld]. "
-      "The application cgroup '%s' will have constraints [%s=%lld].",
+      "System cgroup %s will have constraints [%s=%lld, %s=%lld]. "
+      "Application cgroup %s will have constraints [%s=%lld].",
       base_cgroup_path_,
       node_cgroup_path_,
       supported_controllers,

src/ray/common/cgroup2/cgroup_manager.h

@@ -111,10 +111,9 @@ class CgroupManager : public CgroupManagerInterface {
   Status Initialize(const int64_t system_reserved_cpu_weight,
                     const int64_t system_reserved_memory_bytes);
 
-  // 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);
+  // TODO(#54703): This is a placeholder for cleanup. This will be implemented in the a
+  // future PR.
+  void RegisterDeleteCgroup(const std::string &cgroup_path);
   void RegisterMoveAllProcesses(const std::string &from, const std::string &to);
   template <typename T>
   void RegisterRemoveConstraint(const std::string &cgroup,