class_docstr: Tests for ConcurrencyCapBackpressurePolicy.
class_docstr_tok: ["Tests","for","ConcurrencyCapBackpressurePolicy","."]
docstr: A simple E2E test with ConcurrencyCapBackpressurePolicy enabled.
docstr_tok: ["A","simple","E2E","test","with","ConcurrencyCapBackpressurePolicy","enabled","."]
code: def test_e2e_normal(self):
      """A simple E2E test with ConcurrencyCapBackpressurePolicy enabled."""
      actor = self._create_record_time_actor()
      map_func1 = self._get_map_func(actor, 1)
      map_func2 = self._get_map_func(actor, 2)
  
      # Creat a dataset with 2 map ops. Each map op has N tasks, where N is
      # the number of cluster CPUs.
      N = self.__class__._cluster_cpus
      ds = ray.data.range(N, parallelism=N)
      # Use different `num_cpus` to make sure they don't fuse.
      ds = ds.map_batches(map_func1, batch_size=None, num_cpus=1)
      ds = ds.map_batches(map_func2, batch_size=None, num_cpus=1.1)
      res = ds.take_all()
      self.assertEqual(len(res), N)
  
      # We recorded the start and end time of each op,
      # check that these 2 ops are executed interleavingly.
      # This means that the executor didn't allocate all resources to the first
      # op in the beginning.
      start1, end1 = ray.get(
          actor.get_start_and_end_time_for_op.remote(1)
      )
      start2, end2 = ray.get(
          actor.get_start_and_end_time_for_op.remote(2)
      )
      assert start1 < start2 < end1 < end2, (start1, start2, end1, end2)
  
imports: import functools
  import time
  import unittest
  from collections import defaultdict
  from contextlib import contextmanager
  from unittest.mock import MagicMock, patch
  import numpy
  import pytest
  import ray
  from ray.data._internal.execution.backpressure_policy import ENABLED_BACKPRESSURE_POLICIES_CONFIG_KEY, ConcurrencyCapBackpressurePolicy, StreamingOutputBackpressurePolicy
  from ray.data.tests.conftest import restore_data_context
  from ray.data.tests.conftest import CoreExecutionMetrics, assert_core_execution_metrics_equals, get_initial_core_execution_metrics_snapshot
  from ray.tests.conftest import shutdown_only
label_desc: Use image node_id 8 for calling the TestConcurrencyCapBackpressurePolicy obj's underlying member method code with example usage: obj.test_e2e_normal() without return types