code: def test_actor_pool_resource_reporting_with_bundling(
      ray_start_10_cpus_shared,
  ):
      input_op = InputDataBuffer(
          make_ref_bundles([[SMALL_STR] for i in range(100)])
      )
      op = MapOperator.create(
          _mul2_map_data_prcessor,
          input_op=input_op,
          name="TestMapper",
          compute_strategy=ActorPoolStrategy(min_size=2, max_size=10),
          min_rows_per_bundle=2,
      )
      op.start(ExecutionOptions())
      assert op.base_resource_usage() == ExecutionResources(
          cpu=2, gpu=0
      )
      # All actors are idle (pending creation), therefore shouldn't need to scale up when
      # submitting a new task, so incremental resource usage should be 0.
      assert op.incremental_resource_usage() == ExecutionResources(
          cpu=0, gpu=0
      )
      # Actors are pending creation, but they still count against CPU utilization.
      assert op.current_resource_usage() == ExecutionResources(
          cpu=2, gpu=0, object_store_memory=0
      )
  
      # Add inputs.
      for i in range(4):
          # Pool is still idle while waiting for actors to start, so additional tasks
          # shouldn't trigger scale-up, so incremental resource usage should still be 0.
          assert op.incremental_resource_usage() == ExecutionResources(
              cpu=0, gpu=0
          )
          op.add_input(input_op.get_next(), 0)
          usage = op.current_resource_usage()
          assert usage.cpu == 2, usage
          assert usage.gpu == 0, usage
          # Queued bundles still count against object store usage.
          assert usage.object_store_memory == pytest.approx(
              (i + 1) * 800, rel=0.5
          ), usage
      # Pool is still idle while waiting for actors to start.
      usage = op.current_resource_usage()
      assert usage.cpu == 2, usage
      assert usage.gpu == 0, usage
      # Queued bundles still count against object store usage.
      assert usage.object_store_memory == pytest.approx(
          3200, rel=0.5
      ), usage
  
      # Wait for actors to start.
      assert op.num_active_tasks() == 2
      run_op_tasks_sync(op, only_existing=True)
  
      # Now that both actors have started, a new task would trigger scale-up, so
      # incremental resource usage should be 1 CPU.
      inc_usage = op.incremental_resource_usage()
      assert inc_usage.cpu == 1, inc_usage
      assert inc_usage.gpu == 0, inc_usage
  
      # Actors have now started and the pool is actively running tasks.
      usage = op.current_resource_usage()
      assert usage.cpu == 2, usage
      assert usage.gpu == 0, usage
      assert usage.object_store_memory == pytest.approx(
          3200, rel=0.5
      ), usage
  
      # Indicate that no more inputs will arrive.
      op.all_inputs_done()
  
      # Wait until tasks are done.
      run_op_tasks_sync(op)
  
      # Work is done and the pool has been scaled down.
      usage = op.current_resource_usage()
      assert usage.cpu == 0, usage
      assert usage.gpu == 0, usage
      assert usage.object_store_memory == pytest.approx(
          5500, rel=0.5
      ), usage
  
      # Consume task outputs.
      while op.has_next():
          op.get_next()
  
      # Work is done, pool has been scaled down, and outputs have been consumed.
      usage = op.current_resource_usage()
      assert usage.cpu == 0, usage
      assert usage.gpu == 0, usage
      assert usage.object_store_memory == 0, usage
  
file_path: ray/python/ray/data/tests/test_executor_resource_management.py
imports: import pytest
  import ray
  from ray.data._internal.compute import ActorPoolStrategy, TaskPoolStrategy
  from ray.data._internal.execution.interfaces import ExecutionOptions, ExecutionResources
  from ray.data._internal.execution.operators.input_data_buffer import InputDataBuffer
  from ray.data._internal.execution.operators.map_operator import MapOperator
  from ray.data._internal.execution.util import make_ref_bundles
  from ray.data.tests.conftest import *
  from ray.data.tests.test_operators import _mul2_map_data_prcessor
  from ray.data.tests.util import run_op_tasks_sync
label_desc: Use image node_id 7 for calling a global function with example usage: test_actor_pool_resource_reporting_with_bundling(ray_start_10_cpus_shared) without return types