class_docstr: Tests for StreamOutputBackpressurePolicy.
class_docstr_tok: ["Tests","for","StreamOutputBackpressurePolicy","."]
code: def _run_dataset(self, producer_num_cpus, consumer_num_cpus):
      # Create a dataset with 2 operators:
      # - The producer op has only 1 task, which produces 5 blocks, each of which
      #   has 100MB data.
      # - The consumer op has 5 slow tasks, each of which consumes 1 block.
      # Return the timestamps at the producer and consumer tasks for each block.
      num_blocks = self._num_blocks
      block_size = self._block_size
      ray.data.DataContext.get_current().target_max_block_size = (
          block_size
      )
  
      def producer(batch):
          for i in range(num_blocks):
              print("Producing block", i)
              yield {
                  "id": [i],
                  "data": [np.zeros(block_size, dtype=np.uint8)],
                  "producer_timestamp": [time.time()],
              }
  
      def consumer(batch):
          assert len(batch["id"]) == 1
          print("Consuming block", batch["id"][0])
          batch["consumer_timestamp"] = [time.time()]
          time.sleep(0.1)
          del batch["data"]
          return batch
  
      ds = ray.data.range(1, parallelism=1).materialize()
      ds = ds.map_batches(
          producer, batch_size=None, num_cpus=producer_num_cpus
      )
      ds = ds.map_batches(
          consumer, batch_size=None, num_cpus=consumer_num_cpus
      )
  
      res = ds.take_all()
      assert [row["id"] for row in res] == list(range(self._num_blocks))
      return (
          [row["producer_timestamp"] for row in res],
          [row["consumer_timestamp"] for row in res],
      )
  
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 5 for calling the TestStreamOutputBackpressurePolicy obj's underlying member method code with example usage: obj._run_dataset(producer_num_cpus, consumer_num_cpus) and returns: unknown,  unknown, batch