code: def test_worker_paused_threaded_actor(
      shutdown_only, actor_concurrency
  ):
      ray.init(_system_config=_SYSTEM_CONFIG)
  
      @ray.remote
      class ThreadedActor:
          def main_task(self):
              def thd_task():
                  import time
  
                  # Pause 5 seconds inside debugger
                  with ray._private.worker.global_worker.worker_paused_by_debugger():  # noqa: E501
                      time.sleep(5)
  
              # create and start 10 threads
              thds = []
              for _ in range(10):
                  thd = threading.Thread(target=thd_task)
                  thd.start()
                  thds.append(thd)
  
              # wait for all threads to finish
              for thd in thds:
                  thd.join()
  
      def verify(num_paused):
          workers = list_workers(
              filters=[("num_paused_threads", "=", num_paused)],
              detail=True,
          )
          if len(workers) == 0:
              return False
          worker = workers[0]
          return worker.num_paused_threads == num_paused
  
      threaded_actor = ThreadedActor.options(
          max_concurrency=actor_concurrency
      ).remote()
      # this one call will create 10 threads and all of them will be paused
      threaded_actor.main_task.options(
          name="ThreadedActor.main_task"
      ).remote()
      wait_for_condition(verify, num_paused=10)
  
      # After waiting for 10 threads to finish, there should be no paused threads
      wait_for_condition(verify, num_paused=0)
  
file_path: ray/python/ray/tests/test_worker_state.py
imports: import os
  import pytest
  import sys
  import threading
  import ray
  from ray._private.test_utils import wait_for_condition
  from ray.util.state import list_workers
label_desc: Use image node_id 3 for calling a global function with example usage: test_worker_paused_threaded_actor(shutdown_only, actor_concurrency) and returns: unknown, False