code: def test_e2e_autoscaling_up(
      ray_start_10_cpus_shared, restore_data_context
  ):
      DataContext.get_current().new_execution_backend = True
      DataContext.get_current().use_streaming_executor = True
  
      @ray.remote(max_concurrency=10)
      class Barrier:
          def __init__(self, n, delay=0):
              self.n = n
              self.delay = delay
              self.max_waiters = 0
              self.cur_waiters = 0
  
          def wait(self):
              self.cur_waiters += 1
              if self.cur_waiters > self.max_waiters:
                  self.max_waiters = self.cur_waiters
              self.n -= 1
              print("wait", self.n)
              while self.n > 0:
                  time.sleep(0.1)
              time.sleep(self.delay)
              print("wait done")
              self.cur_waiters -= 1
  
          def get_max_waiters(self):
              return self.max_waiters
  
      b1 = Barrier.remote(6)
  
      class BarrierWaiter:
          def __init__(self, barrier):
              self._barrier = barrier
  
          def __call__(self, x):
              ray.get(self._barrier.wait.remote(), timeout=10)
              return x
  
      # Tests that we autoscale up to necessary size.
      # 6 tasks + 1 tasks in flight per actor => need at least 6 actors to run.
      ray.data.range(6, parallelism=6).map_batches(
          BarrierWaiter,
          fn_constructor_args=(b1,),
          compute=ray.data.ActorPoolStrategy(
              min_size=1, max_size=6, max_tasks_in_flight_per_actor=1
          ),
          batch_size=None,
      ).take_all()
      assert ray.get(b1.get_max_waiters.remote()) == 6
  
      b2 = Barrier.remote(3, delay=2)
  
      # Tests that we don't over-scale up.
      # 6 tasks + 2 tasks in flight per actor => only scale up to 3 actors
      ray.data.range(6, parallelism=6).map_batches(
          BarrierWaiter,
          fn_constructor_args=(b2,),
          compute=ray.data.ActorPoolStrategy(
              min_size=1, max_size=3, max_tasks_in_flight_per_actor=2
          ),
          batch_size=None,
      ).take_all()
      assert ray.get(b2.get_max_waiters.remote()) == 3
  
      # Tests that the max pool size is respected.
      b3 = Barrier.remote(6)
  
      # This will hang, since the actor pool is too small.
      with pytest.raises(ray.exceptions.RayTaskError):
          ray.data.range(6, parallelism=6).map(
              BarrierWaiter,
              fn_constructor_args=(b3,),
              compute=ray.data.ActorPoolStrategy(
                  min_size=1, max_size=2
              ),
          ).take_all()
  
imports: import itertools
  import random
  import threading
  import time
  from typing import Any, List
  import pandas
  import pytest
  import ray
  from ray import cloudpickle
  from ray._private.test_utils import wait_for_condition
  from ray.data._internal.execution.interfaces import ExecutionOptions, ExecutionResources, RefBundle
  from ray.data._internal.execution.operators.base_physical_operator import AllToAllOperator
  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.operators.map_transformer import create_map_transformer_from_block_fn
  from ray.data._internal.execution.operators.output_splitter import OutputSplitter
  from ray.data._internal.execution.streaming_executor import StreamingExecutor
  from ray.data._internal.execution.util import make_ref_bundles
  from ray.data.context import DataContext
  from ray.data.tests.conftest import *
  from ray.data.tests.util import extract_values
label_desc: Use image node_id 14 for calling a global function with example usage: test_e2e_autoscaling_up(ray_start_10_cpus_shared, restore_data_context) and returns: self, x