code: def start_actors(total_num_actors, num_nodes):
      """Create actors and run the computation loop."""
      total_num_actors = int(total_num_actors)
      actors_per_node = int(total_num_actors / num_nodes)
      start = time.time()
      nodes = []
      # Place an actor per node in round-robin.
      # It is added here to simulate the real user workload.
      while len(nodes) < num_nodes:
          nodes = [
              next((r for r in n["Resources"] if "node" in r), None)
              for n in ray.nodes()
              if n["Alive"]
          ]
          nodes = [n for n in nodes if n is not None]
      pi_actors = [
          PiCalculator.options(
              resources={n: 0.01}, max_concurrency=10
          ).remote({"meta": 1})
          for n in nodes
          for _ in range(actors_per_node)
      ]
      ray.get([actor.ready.remote() for actor in pi_actors])
      print(
          f"Took {time.time() - start} to create {total_num_actors} actors"
      )
      # Start the computation loop.
      for actor in pi_actors:
          actor.run_compute.remote()
      return pi_actors
  
imports: import argparse
  import json
  import logging
  import numpy
  import os
  import threading
  import time
  from ray._private.test_utils import monitor_memory_usage
  from ray.data._internal.progress_bar import ProgressBar
  from collections import namedtuple
  from queue import Queue
  import ray
label_desc: Use image node_id 1 for calling a global function with example usage: start_actors(total_num_actors, num_nodes) and returns: pi_actors