[Enable gcs actor scheduler 1/n] Raylet and GCS schedulers share cluster_task_manager

BUILD.bazel

@@ -1786,10 +1786,10 @@ cc_test(
 )
 
 cc_test(
-    name = "raylet_based_actor_scheduler_test",
+    name = "gcs_actor_scheduler_test",
     size = "small",
     srcs = [
-        "src/ray/gcs/gcs_server/test/raylet_based_actor_scheduler_test.cc",
+        "src/ray/gcs/gcs_server/test/gcs_actor_scheduler_test.cc",
     ],
     copts = COPTS,
     tags = ["team:core"],
@@ -1818,23 +1818,6 @@ cc_test(
     ],
 )
 
-cc_test(
-    name = "gcs_based_actor_scheduler_test",
-    size = "small",
-    srcs = [
-        "src/ray/gcs/gcs_server/test/gcs_based_actor_scheduler_test.cc",
-    ],
-    copts = COPTS,
-    tags = ["team:core"],
-    deps = [
-        ":gcs_server_lib",
-        ":gcs_server_test_util",
-        ":gcs_test_util_lib",
-        ":ray_mock",
-        "@com_google_googletest//:gtest_main",
-    ],
-)
-
 cc_test(
     name = "gcs_actor_manager_test",
     size = "small",

python/ray/tests/test_advanced_5.py

@@ -165,7 +165,7 @@ def test_actor_distribution_balance(ray_start_cluster_enabled, args):
     ray.init(address=cluster.address)
     cluster.wait_for_nodes()
 
-    @ray.remote(memory=100 * 1024 ** 2, num_cpus=0.01)
+    @ray.remote(memory=100 * 1024 ** 2, num_cpus=0.01, scheduling_strategy="SPREAD")
     class Foo:
         def method(self):
             return ray.worker.global_worker.node.unique_id
@@ -194,16 +194,17 @@ def test_worker_lease_reply_with_resources(ray_start_cluster_enabled):
     cluster = ray_start_cluster_enabled
     cluster.add_node(
         memory=2000 * 1024 ** 2,
+        num_cpus=1,
         _system_config={
             "gcs_resource_report_poll_period_ms": 1000000,
             "gcs_actor_scheduling_enabled": True,
         },
     )
-    node2 = cluster.add_node(memory=1000 * 1024 ** 2)
+    node2 = cluster.add_node(memory=1000 * 1024 ** 2, num_cpus=1)
     ray.init(address=cluster.address)
     cluster.wait_for_nodes()
 
-    @ray.remote(memory=1500 * 1024 ** 2)
+    @ray.remote(memory=1500 * 1024 ** 2, num_cpus=0.01)
     def fun(signal):
         signal.send.remote()
         time.sleep(30)
@@ -214,7 +215,7 @@ def fun(signal):
     # Make sure that the `fun` is running.
     ray.get(signal.wait.remote())
 
-    @ray.remote(memory=800 * 1024 ** 2)
+    @ray.remote(memory=800 * 1024 ** 2, num_cpus=0.01)
     class Foo:
         def method(self):
             return ray.worker.global_worker.node.unique_id