code: def test_threaded_actor_basic(shutdown_only):
      """Test the basic threaded actor."""
      ray.init(num_cpus=1)
  
      @ray.remote(num_cpus=1)
      class ThreadedActor:
          def __init__(self):
              self.received = []
              self.lock = threading.Lock()
  
          def add(self, seqno):
              with self.lock:
                  self.received.append(seqno)
  
          def get_all(self):
              with self.lock:
                  return self.received
  
      a = ThreadedActor.options(max_concurrency=10).remote()
      max_seq = 50
      ray.get([a.add.remote(seqno) for seqno in range(max_seq)])
      seqnos = ray.get(a.get_all.remote())
      # Currently, the caller submission order is not guaranteed
      # when the threaded actor is used.
      assert sorted(seqnos) == list(range(max_seq))
      ray.kill(a)
      ensure_cpu_returned(1)
  
file_path: ray/python/ray/tests/test_threaded_actor.py
imports: import sys
  import threading
  import time
  import numpy
  import pytest
  import ray
  import ray._private.test_utils
  from ray._private.state import available_resources
label_desc: Use image node_id 2 for calling a global function with example usage: test_threaded_actor_basic(shutdown_only) and returns: self