diff --git a/python/ray/dag/compiled_dag_node.py b/python/ray/dag/compiled_dag_node.py index 27a68595d858..b8a3cb7bf209 100644 --- a/python/ray/dag/compiled_dag_node.py +++ b/python/ray/dag/compiled_dag_node.py @@ -661,7 +661,7 @@ def __init__( # We conservatively set num_shm_buffers to _max_inflight_executions. # It means that the DAG can be underutilized, but it guarantees there's # no false positive timeouts. - num_shm_buffers=self._max_inflight_executions, + num_shm_buffers=1, ) if not isinstance(self._buffer_size_bytes, int) or self._buffer_size_bytes <= 0: raise ValueError( diff --git a/python/ray/dag/tests/experimental/test_accelerated_dag.py b/python/ray/dag/tests/experimental/test_accelerated_dag.py index f30f39a65e42..50b08550cb25 100644 --- a/python/ray/dag/tests/experimental/test_accelerated_dag.py +++ b/python/ray/dag/tests/experimental/test_accelerated_dag.py @@ -1970,6 +1970,7 @@ def test_driver_and_actor_as_readers(ray_start_cluster): dag.experimental_compile() +@pytest.mark.skip("Currently buffer size is set to 1 because of regression.") @pytest.mark.parametrize("temporary_change_timeout", [1], indirect=True) def test_buffered_inputs(shutdown_only, temporary_change_timeout): ray.init()