code: def main(results=None):
      results = results or []
  
      check_optimized_build()
  
      print(
          "Tip: set TESTS_TO_RUN='pattern' to run a subset of benchmarks"
      )
  
      ray.init()
  
      value = ray.put(0)
  
      def get_small():
          ray.get(value)
  
      def put_small():
          ray.put(0)
  
      @ray.remote
      def do_put_small():
          for _ in range(100):
              ray.put(0)
  
      def put_multi_small():
          ray.get([do_put_small.remote() for _ in range(10)])
  
      arr = np.zeros(100 * 1024 * 1024, dtype=np.int64)
  
      results += timeit(
          "single client get calls (Plasma Store)", get_small
      )
  
      results += timeit(
          "single client put calls (Plasma Store)", put_small
      )
  
      results += timeit(
          "multi client put calls (Plasma Store)", put_multi_small, 1000
      )
  
      def put_large():
          ray.put(arr)
  
      results += timeit(
          "single client put gigabytes", put_large, 8 * 0.1
      )
  
      def small_value_batch():
          submitted = [small_value.remote() for _ in range(1000)]
          ray.get(submitted)
          return 0
  
      results += timeit(
          "single client tasks and get batch", small_value_batch
      )
  
      @ray.remote
      def do_put():
          for _ in range(10):
              ray.put(np.zeros(10 * 1024 * 1024, dtype=np.int64))
  
      def put_multi():
          ray.get([do_put.remote() for _ in range(10)])
  
      results += timeit(
          "multi client put gigabytes", put_multi, 10 * 8 * 0.1
      )
  
      obj_containing_ref = create_object_containing_ref.remote()
  
      def get_containing_object_ref():
          ray.get(obj_containing_ref)
  
      results += timeit(
          "single client get object containing 10k refs",
          get_containing_object_ref,
      )
  
      def wait_multiple_refs():
          num_objs = 1000
          not_ready = [small_value.remote() for _ in range(num_objs)]
          # We only need to trigger the fetch_local once for each object,
          # raylet will persist these fetch requests even after ray.wait returns.
          # See https://github.com/ray-project/ray/issues/30375.
          fetch_local = True
          for _ in range(num_objs):
              _ready, not_ready = ray.wait(
                  not_ready, fetch_local=fetch_local
              )
              if fetch_local:
                  fetch_local = False
  
      results += timeit(
          "single client wait 1k refs", wait_multiple_refs
      )
  
      def small_task():
          ray.get(small_value.remote())
  
      results += timeit("single client tasks sync", small_task)
  
      def small_task_async():
          ray.get([small_value.remote() for _ in range(1000)])
  
      results += timeit(
          "single client tasks async", small_task_async, 1000
      )
  
      n = 10000
      m = 4
      actors = [Actor.remote() for _ in range(m)]
  
      def multi_task():
          submitted = [a.small_value_batch.remote(n) for a in actors]
          ray.get(submitted)
  
      results += timeit("multi client tasks async", multi_task, n * m)
  
      a = Actor.remote()
  
      def actor_sync():
          ray.get(a.small_value.remote())
  
      results += timeit("1:1 actor calls sync", actor_sync)
  
      a = Actor.remote()
  
      def actor_async():
          ray.get([a.small_value.remote() for _ in range(1000)])
  
      results += timeit("1:1 actor calls async", actor_async, 1000)
  
      a = Actor.options(max_concurrency=16).remote()
  
      def actor_concurrent():
          ray.get([a.small_value.remote() for _ in range(1000)])
  
      results += timeit(
          "1:1 actor calls concurrent", actor_concurrent, 1000
      )
  
      n = 5000
      n_cpu = multiprocessing.cpu_count() // 2
      actors = [Actor._remote() for _ in range(n_cpu)]
      client = Client.remote(actors)
  
      def actor_async_direct():
          ray.get(client.small_value_batch.remote(n))
  
      results += timeit(
          "1:n actor calls async", actor_async_direct, n * len(actors)
      )
  
      n_cpu = multiprocessing.cpu_count() // 2
      a = [Actor.remote() for _ in range(n_cpu)]
  
      @ray.remote
      def work(actors):
          ray.get(
              [actors[i % n_cpu].small_value.remote() for i in range(n)]
          )
  
      def actor_multi2():
          ray.get([work.remote(a) for _ in range(m)])
  
      results += timeit("n:n actor calls async", actor_multi2, m * n)
  
      n = 1000
      actors = [Actor._remote() for _ in range(n_cpu)]
      clients = [Client.remote(a) for a in actors]
  
      def actor_multi2_direct_arg():
          ray.get([c.small_value_batch_arg.remote(n) for c in clients])
  
      results += timeit(
          "n:n actor calls with arg async",
          actor_multi2_direct_arg,
          n * len(clients),
      )
  
      a = AsyncActor.remote()
  
      def actor_sync():
          ray.get(a.small_value.remote())
  
      results += timeit("1:1 async-actor calls sync", actor_sync)
  
      a = AsyncActor.remote()
  
      def async_actor():
          ray.get([a.small_value.remote() for _ in range(1000)])
  
      results += timeit(
          "1:1 async-actor calls async", async_actor, 1000
      )
  
      a = AsyncActor.remote()
  
      def async_actor():
          ray.get(
              [a.small_value_with_arg.remote(i) for i in range(1000)]
          )
  
      results += timeit(
          "1:1 async-actor calls with args async", async_actor, 1000
      )
  
      n = 5000
      n_cpu = multiprocessing.cpu_count() // 2
      actors = [AsyncActor.remote() for _ in range(n_cpu)]
      client = Client.remote(actors)
  
      def async_actor_async():
          ray.get(client.small_value_batch.remote(n))
  
      results += timeit(
          "1:n async-actor calls async",
          async_actor_async,
          n * len(actors),
      )
  
      n = 5000
      m = 4
      n_cpu = multiprocessing.cpu_count() // 2
      a = [AsyncActor.remote() for _ in range(n_cpu)]
  
      @ray.remote
      def async_actor_work(actors):
          ray.get(
              [actors[i % n_cpu].small_value.remote() for i in range(n)]
          )
  
      def async_actor_multi():
          ray.get([async_actor_work.remote(a) for _ in range(m)])
  
      results += timeit(
          "n:n async-actor calls async", async_actor_multi, m * n
      )
      ray.shutdown()
  
      #################################################
      # Perf tests for channels, used in compiled DAGs.
      #################################################
      ray.init()
  
      def put_channel_small(chans, do_get=False, do_release=False):
          for chan in chans:
              chan.write(b"0")
              if do_get:
                  chan.begin_read()
              if do_release:
                  chan.end_read()
  
      @ray.remote
      class ChannelReader:
          def ready(self):
              return
  
          def read(self, chans):
              while True:
                  for chan in chans:
                      chan.begin_read()
                      chan.end_read()
  
      chans = [ray_channel.Channel(1000)]
      results += timeit(
          "[unstable] local put, single channel calls",
          lambda: put_channel_small(chans, do_release=True),
      )
      results += timeit(
          "[unstable] local put:local get, single channel calls",
          lambda: put_channel_small(
              chans, do_get=True, do_release=True
          ),
      )
  
      chans = [ray_channel.Channel(1000)]
      reader = ChannelReader.remote()
      ray.get(reader.ready.remote())
      reader.read.remote(chans)
      results += timeit(
          "[unstable] local put:1 remote get, single channel calls",
          lambda: put_channel_small(chans),
      )
      ray.kill(reader)
  
      n_cpu = multiprocessing.cpu_count() // 2
      print(f"Testing multiple readers/channels, n={n_cpu}")
  
      chans = [ray_channel.Channel(1000, num_readers=n_cpu)]
      readers = [ChannelReader.remote() for _ in range(n_cpu)]
      ray.get([reader.ready.remote() for reader in readers])
      for reader in readers:
          reader.read.remote(chans)
      results += timeit(
          "[unstable] local put:n remote get, single channel calls",
          lambda: put_channel_small(chans),
      )
      for reader in readers:
          ray.kill(reader)
  
      chans = [ray_channel.Channel(1000) for _ in range(n_cpu)]
      reader = ChannelReader.remote()
      ray.get(reader.ready.remote())
      reader.read.remote(chans)
      results += timeit(
          "[unstable] local put:1 remote get, n channels calls",
          lambda: put_channel_small(chans),
      )
      ray.kill(reader)
  
      chans = [ray_channel.Channel(1000) for _ in range(n_cpu)]
      readers = [ChannelReader.remote() for _ in range(n_cpu)]
      ray.get([reader.ready.remote() for reader in readers])
      for chan, reader in zip(chans, readers):
          reader.read.remote([chan])
      results += timeit(
          "[unstable] local put:n remote get, n channels calls",
          lambda: put_channel_small(chans),
      )
      for reader in readers:
          ray.kill(reader)
  
      # Tests for compiled DAGs.
  
      def _exec(dag):
          output_channel = dag.execute(b"x")
          output_channel.begin_read()
          output_channel.end_read()
  
      def _exec_multi_output(dag):
          output_channels = dag.execute(b"x")
          for output_channel in output_channels:
              output_channel.begin_read()
          for output_channel in output_channels:
              output_channel.end_read()
  
      a = DAGActor.remote()
      with InputNode() as inp:
          dag = a.echo.bind(inp)
  
      results += timeit(
          "[unstable] single-actor DAG calls",
          lambda: ray.get(dag.execute(b"x")),
      )
      dag = dag.experimental_compile()
      results += timeit(
          "[unstable] compiled single-actor DAG calls",
          lambda: _exec(dag),
      )
  
      del a
      n_cpu = multiprocessing.cpu_count() // 2
      actors = [DAGActor.remote() for _ in range(n_cpu)]
      with InputNode() as inp:
          dag = MultiOutputNode([a.echo.bind(inp) for a in actors])
      results += timeit(
          "[unstable] scatter-gather DAG calls, n={n_cpu} actors",
          lambda: ray.get(dag.execute(b"x")),
      )
      dag = dag.experimental_compile()
      results += timeit(
          f"[unstable] compiled scatter-gather DAG calls, n={n_cpu} actors",
          lambda: _exec_multi_output(dag),
      )
  
      actors = [DAGActor.remote() for _ in range(n_cpu)]
      with InputNode() as inp:
          dag = inp
          for a in actors:
              dag = a.echo.bind(dag)
      results += timeit(
          f"[unstable] chain DAG calls, n={n_cpu} actors",
          lambda: ray.get(dag.execute(b"x")),
      )
      dag = dag.experimental_compile()
      results += timeit(
          f"[unstable] compiled chain DAG calls, n={n_cpu} actors",
          lambda: _exec(dag),
      )
  
      ray.shutdown()
  
      ############################
      # End of channel perf tests.
      ############################
  
      NUM_PGS = 100
      NUM_BUNDLES = 1
      ray.init(resources={"custom": 100})
  
      def placement_group_create_removal(num_pgs):
          pgs = [
              ray.util.placement_group(
                  bundles=[
                      {"custom": 0.001} for _ in range(NUM_BUNDLES)
                  ]
              )
              for _ in range(num_pgs)
          ]
          [pg.wait(timeout_seconds=30) for pg in pgs]
          # Include placement group removal here to clean up.
          # If we don't clean up placement groups, the whole performance
          # gets slower as it runs more.
          # Since timeit function runs multiple times without
          # the cleaning logic, we should have this method here.
          for pg in pgs:
              ray.util.remove_placement_group(pg)
  
      results += timeit(
          "placement group create/removal",
          lambda: placement_group_create_removal(NUM_PGS),
          NUM_PGS,
      )
      ray.shutdown()
  
      client_microbenchmark_main(results)
  
      return results
  
imports: import asyncio
  import logging
  from ray._private.ray_microbenchmark_helpers import timeit
  from ray._private.ray_client_microbenchmark import main
  import numpy
  import multiprocessing
  import ray
  import ray.experimental.channel
  from ray.dag import InputNode, MultiOutputNode
label_desc: Use image node_id 5 for calling a global function with example usage: main(results) and returns: results, int