[core][ci] Update state api scale test
diff --git a/python/ray/tests/test_state_api_2.py b/python/ray/tests/test_state_api_2.py
index cbb654565aacb..c5f3ce4276c56 100644
--- a/python/ray/tests/test_state_api_2.py
+++ b/python/ray/tests/test_state_api_2.py
@@ -1,8 +1,11 @@
 import asyncio
 import json
 import os
+import sys
+from pathlib import Path
 
 from collections import defaultdict
+from ray._private.test_utils import check_call_subprocess
 
 import ray
 import requests
@@ -277,7 +280,7 @@ def test_experimental_import_deprecation():
         from ray.experimental.state.util import convert_string_to_type  # noqa: F401
 
 
-def test_actor_task_with_repr_name():
+def test_actor_task_with_repr_name(ray_start_with_dashboard):
     @ray.remote
     class ReprActor:
         def __init__(self, x) -> None:
@@ -338,6 +341,20 @@ def verify():
     wait_for_condition(verify)
 
 
+@pytest.mark.skipif(
+    sys.platform == "win32", reason="Release test not expected to work on non-linux."
+)
+def test_state_api_scale_smoke(shutdown_only):
+    ray.init()
+    release_test_file_path = (
+        "../../release/nightly_tests/stress_tests/test_state_api_scale.py"
+    )
+    full_path = Path(ray.__file__).parents[0] / release_test_file_path
+    assert full_path.exists()
+
+    check_call_subprocess(["python", str(full_path), "--smoke-test"])
+
+
 if __name__ == "__main__":
     import sys
 
diff --git a/release/nightly_tests/stress_tests/test_state_api_scale.py b/release/nightly_tests/stress_tests/test_state_api_scale.py
index b50c008c86115..466c62a8835b8 100644
--- a/release/nightly_tests/stress_tests/test_state_api_scale.py
+++ b/release/nightly_tests/stress_tests/test_state_api_scale.py
@@ -1,6 +1,7 @@
 import click
 import json
 import ray
+from ray._private.ray_constants import LOG_PREFIX_ACTOR_NAME, LOG_PREFIX_JOB_ID
 from ray._private.state_api_test_utils import (
     STATE_LIST_LIMIT,
     StateAPIMetric,
@@ -11,10 +12,16 @@
 
 import ray._private.test_utils as test_utils
 import tqdm
-import asyncio
 import time
 import os
 
+from ray.util.placement_group import (
+    placement_group,
+    remove_placement_group,
+)
+from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
+
+
 from ray.util.state import (
     get_log,
     list_actors,
@@ -26,22 +33,6 @@
 MiB = 1024 * 1024
 
 
-# We set num_cpus to zero because this actor will mostly just block on I/O.
-@ray.remote(num_cpus=0)
-class SignalActor:
-    def __init__(self):
-        self.ready_event = asyncio.Event()
-
-    def send(self, clear=False):
-        self.ready_event.set()
-        if clear:
-            self.ready_event.clear()
-
-    async def wait(self, should_wait=True):
-        if should_wait:
-            await self.ready_event.wait()
-
-
 def invoke_state_api_n(*args, **kwargs):
     def verify():
         NUM_API_CALL_SAMPLES = 10
@@ -53,6 +44,7 @@ def verify():
 
 
 def test_many_tasks(num_tasks: int):
+    TASK_NAME_TEMPLATE = "pi4_sample_{num_tasks}"
     if num_tasks == 0:
         print("Skipping test with no tasks")
         return
@@ -61,10 +53,13 @@ def test_many_tasks(num_tasks: int):
     invoke_state_api_n(
         lambda res: len(res) == 0,
         list_tasks,
-        filters=[("name", "=", "pi4_sample"), ("state", "=", "RUNNING")],
+        filters=[("name", "=", TASK_NAME_TEMPLATE.format(num_tasks=num_tasks))],
         key_suffix="0",
         limit=STATE_LIST_LIMIT,
-        err_msg="Expect 0 running tasks.",
+        err_msg=(
+            "Expect 0 running tasks for "
+            f"{TASK_NAME_TEMPLATE.format(num_tasks=num_tasks)}"
+        ),
     )
 
     # Task definition adopted from:
@@ -73,37 +68,32 @@ def test_many_tasks(num_tasks: int):
 
     SAMPLES = 100
 
-    # `num_cpus` required obtained from 1 / (num_tasks /num_total_cpus)
-    #  where num_total_cpus obtained from cluster_compute in `release_tests.yaml`
-    # 1 / (10k/45 * 7) ~= 0.03, taking a smaller value to make sure all tasks could be
-    # running at the same time.
-    @ray.remote(num_cpus=0.02)
-    def pi4_sample(signal):
+    @ray.remote
+    def pi4_sample():
         in_count = 0
         for _ in range(SAMPLES):
             x, y = random(), random()
             if x * x + y * y <= 1:
                 in_count += 1
-        # Block on signal
-        ray.get(signal.wait.remote())
         return in_count
 
     results = []
-    signal = SignalActor.remote()
     for _ in tqdm.trange(num_tasks, desc="Launching tasks"):
-        results.append(pi4_sample.remote(signal))
+        results.append(
+            pi4_sample.options(
+                name=TASK_NAME_TEMPLATE.format(num_tasks=num_tasks)
+            ).remote()
+        )
 
     invoke_state_api_n(
         lambda res: len(res) == num_tasks,
         list_tasks,
-        filters=[("name", "=", "pi4_sample"), ("state", "!=", "FINISHED")],
+        filters=[("name", "=", TASK_NAME_TEMPLATE.format(num_tasks=num_tasks))],
         key_suffix=f"{num_tasks}",
         limit=STATE_LIST_LIMIT,
         err_msg=f"Expect {num_tasks} non finished tasks.",
     )
 
-    print("Waiting for tasks to finish...")
-    ray.get(signal.send.remote())
     ray.get(results)
 
     # Clean up
@@ -111,14 +101,15 @@ def pi4_sample(signal):
     invoke_state_api_n(
         lambda res: len(res) == 0,
         list_tasks,
-        filters=[("name", "=", "pi4_sample"), ("state", "=", "RUNNING")],
+        filters=[
+            ("name", "=", TASK_NAME_TEMPLATE.format(num_tasks=num_tasks)),
+            ("state", "=", "RUNNING"),
+        ],
         key_suffix="0",
         limit=STATE_LIST_LIMIT,
         err_msg="Expect 0 running tasks",
     )
 
-    del signal
-
 
 def test_many_actors(num_actors: int):
     if num_actors == 0:
@@ -177,7 +168,11 @@ def test_many_objects(num_objects, num_actors):
         print("Skipping test with no objects")
         return
 
-    @ray.remote(num_cpus=0.1)
+    pg = placement_group([{"CPU": 1}] * num_actors, strategy="SPREAD")
+    ray.get(pg.ready())
+
+    # We will try to put actors on multiple nodes.
+    @ray.remote
     class ObjectActor:
         def __init__(self):
             self.objs = []
@@ -191,11 +186,13 @@ def create_objs(self, num_objects):
 
             return self.objs
 
-        def exit(self):
-            ray.actor.exit_actor()
-
     actors = [
-        ObjectActor.remote() for _ in tqdm.trange(num_actors, desc="Creating actors...")
+        ObjectActor.options(
+            scheduling_strategy=PlacementGroupSchedulingStrategy(
+                placement_group=pg,
+            )
+        ).remote()
+        for _ in tqdm.trange(num_actors, desc="Creating actors...")
     ]
 
     # Splitting objects to multiple actors for creation,
@@ -231,9 +228,8 @@ def _split(a, n):
         limit=STATE_LIST_LIMIT,
     )
 
-    exiting_actors = [actor.exit.remote() for actor in actors]
-    for _ in tqdm.trange(len(actors), desc="Destroying actors..."):
-        _exitted, exiting_actors = ray.wait(exiting_actors)
+    del actors
+    remove_placement_group(pg)
 
 
 def test_large_log_file(log_file_size_byte: int):
@@ -250,6 +246,9 @@ def test_large_log_file(log_file_size_byte: int):
     class LogActor:
         def write_log(self, log_file_size_byte: int):
             ctx = hashlib.md5()
+            job_id = ray.get_runtime_context().get_job_id()
+            prefix = f"{LOG_PREFIX_JOB_ID}{job_id}\n{LOG_PREFIX_ACTOR_NAME}LogActor\n"
+            ctx.update(prefix.encode())
             while log_file_size_byte > 0:
                 n = min(log_file_size_byte, 4 * MiB)
                 chunk = "".join(random.choices(string.ascii_letters, k=n))
@@ -272,7 +271,7 @@ def write_log(self, log_file_size_byte: int):
 
     time_taken = 0
     t_start = time.perf_counter()
-    for s in get_log(task_id=task.task_id().hex(), tail=1000000000):
+    for s in get_log(actor_id=actor._actor_id.hex(), tail=1000000000):
         t_end = time.perf_counter()
         time_taken += t_end - t_start
         # Not including this time
@@ -359,10 +358,13 @@ def test(
     ray.init(address="auto", log_to_driver=False)
 
     if smoke_test:
-        num_tasks = "100"
-        num_actors = "10"
-        num_objects = "100"
-        log_file_size_byte = f"{16*MiB}"
+        num_tasks = "1,100"
+        num_actors = "1,10"
+        num_objects = "1,100"
+        num_actors_for_objects = 1
+        log_file_size_byte = f"64,{16*MiB}"
+        global STATE_LIST_LIMIT
+        STATE_LIST_LIMIT = STATE_LIST_LIMIT // 1000
 
     # Parse the input
     num_tasks_arr, num_actors_arr, num_objects_arr, log_file_size_arr = _parse_input(
@@ -420,7 +422,33 @@ def test(
                 "perf_metric_name": "avg_state_api_latency_sec",
                 "perf_metric_value": state_perf_result["avg_state_api_latency_sec"],
                 "perf_metric_type": "LATENCY",
-            }
+            },
+            {
+                "perf_metric_name": "avg_state_api_get_log_latency_sec",
+                "perf_metric_value": state_perf_result["avg_get_log_latency_sec"],
+                "perf_metric_type": "LATENCY",
+            },
+            {
+                "perf_metric_name": "avg_state_api_list_tasks_10000_latency_sec",
+                "perf_metric_value": state_perf_result[
+                    "avg_list_tasks_10000_latency_sec"
+                ],
+                "perf_metric_type": "LATENCY",
+            },
+            {
+                "perf_metric_name": "avg_state_api_list_actors_5000_latency_sec",
+                "perf_metric_value": state_perf_result[
+                    "avg_list_actors_5000_latency_sec"
+                ],
+                "perf_metric_type": "LATENCY",
+            },
+            {
+                "perf_metric_name": "avg_state_api_list_objects_50000_latency_sec",
+                "perf_metric_value": state_perf_result[
+                    "avg_list_objects_50000_latency_sec"
+                ],
+                "perf_metric_type": "LATENCY",
+            },
         ]
 
     if "TEST_OUTPUT_JSON" in os.environ:

<!-- Thank you for your contribution! Please review https://github.com/ray-project/ray/blob/master/CONTRIBUTING.rst before opening a pull request. -->

<!-- Please add a reviewer to the assignee section when you create a PR. If you don't have the access to it, we will shortly find a reviewer and assign them to your PR. -->

## Why are these changes needed?
We are using a signal block to block the execution of tasks because tasks states were not persistent when they finished before (so we need to have X number of tasks running to simulate a large task response). 

However, this is no longer true.


<!-- Please give a short summary of the change and the problem this solves. -->

## Related issue number

Closes https://github.com/ray-project/ray/issues/35460

<!-- For example: "Closes #1234" -->

## Checks

- [ ] I've signed off every commit(by using the -s flag, i.e., `git commit -s`) in this PR.
- [ ] I've run `scripts/format.sh` to lint the changes in this PR.
- [ ] I've included any doc changes needed for https://docs.ray.io/en/master/.
    - [ ] I've added any new APIs to the API Reference. For example, if I added a 
           method in Tune, I've added it in `doc/source/tune/api/` under the 
           corresponding `.rst` file.
- [ ] I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
- Testing Strategy
   - [ ] Unit tests
   - [ ] Release tests
   - [ ] This PR is not tested :(
