[aDAG] support buffered input #47272

rkooo567 · 2024-08-22T08:58:36Z

Why are these changes needed?

Based on https://docs.google.com/document/d/1Ka_HFwPBNIY1u3kuroHOSZMEQ8AgwpYciZ4n08HJ0Xc/edit

When there are many in-flight requests (pipelining inputs to the DAG), 2 problems occur.

Input submitter timeout. InputSubmitter.write() waits until the buffer is read from downstream tasks. Since timeout count is started as soon as InputSubmitter.write() is called, when there are many in-flight requests, the later requests are likely to timeout.
Pipeline bubble. Output fetcher doesn’t read the channel until CompiledDagRef.get is called. It means the upstream task (actor 2) has to be blocked until .get is called from a driver although it can execute tasks.

This PR solves the problem by providing multiple buffer per shm channel. Note that the buffering is not supported for nccl yet (we can do it when we overlap compute/comm).

Main changes

Introduce BufferedSharedMemoryChannel which allows to create multiple buffers (10 by default). Read/write is done in round robin manner.
When you have more in-flight request than the buffer size, Dag can still have timeout error. To make debugging easy and behavior straightforward, we introduce max_buffered_inputs_ argument. If there are more than max_buffered_inputs_ requests submitted to the dag without ray.get, it immediately raises an exception.

Q&A

What's going to happen if we don't have max_buffered_inputs_?
- in this case, users can have more than max_buffered_inputs_ in-flight requests without timeout. But when timeout actually occurs, it is very difficult to tell users if this is due to pipeline being full or other errors (such as deadlock). By limiting max_buffered_inputs_ from the dag-level, we can more clearly tell users what's the case. Users can simply increase max_buffered_inputs_ if they want to submit more tasks.
why do you raise an exception from a dag when max_buffered_inputs_ is full instead of per channel?
- I tried this approach at first, but it was difficult to reliably raise a max capacity exception because of many race conditions. It complicated the PR, so I decided to make it simple.
why did you choose many channel approach vs additional thread approach?
- additional thread with python's queue introduces very high overhead (almost 30us to just coordinate synchronization), and the threading sychroniziation has to be implemented in cpp. I thought it is much simpler to use multiple channels instead.

Related issue number

Closes #47097
Closes #43826

Checks

I've signed off every commit(by using the -s flag, i.e., git commit -s) in this PR.
I've run scripts/format.sh to lint the changes in this PR.
I've included any doc changes needed for https://docs.ray.io/en/master/.
- I've added any new APIs to the API Reference. For example, if I added a
  method in Tune, I've added it in doc/source/tune/api/ under the
  corresponding .rst file.
I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
Testing Strategy
- Unit tests
- Release tests
- This PR is not tested :(

ruisearch42 · 2024-08-22T15:59:48Z

python/ray/experimental/channel/shared_memory_channel.py

+        if not self._background_thread_started:
+            self.start()
+            self._background_thread_started = True


Do we really need additional threads per channel? Can we just use the caller's thread to do read and write?

if there's a clever way to make it non-blocking...

ruisearch42 · 2024-08-22T16:00:22Z

python/ray/experimental/channel/shared_memory_channel.py

+
+
+# @DeveloperAPI
+class BufferedRemoteChannel(ChannelInterface):


Maybe this is just a draft PR, but wondering why you call it a remote channel

intra process channel doesn't have to use this mechanism as it is immediately read/written

ruisearch42 · 2024-08-22T16:03:24Z

python/ray/experimental/channel/shared_memory_channel.py

+            self.start()
+            self._background_thread_started = True
+
+        self.queue.put((value, timeout))


Not sure if buffering at Python would work. I was thinking the buffering probably happens at CPP level.

at least self.queue.put seems too slow. exploring other approaches now

rkooo567 · 2024-08-24T17:31:22Z

@ruisearch42 @kevin85421 the PR is ready for a review

ruisearch42 · 2024-08-26T15:20:45Z

python/ray/dag/compiled_dag_node.py

 ) -> ChannelInterface:
    """Generic actor method to allocate an output channel.

    Args:
        reader_and_node_list: A list of tuples, where each tuple contains a reader
            actor handle and the node ID where the actor is located.
        typ: The output type hint for the channel.
+        num_shm_buffers: The number of shared memory buffer per channel.
+            It is currently ignored for nccl channel.


nit: remove "currently".
If the variable is called shm buffers, then it should always not affect nccl.

python/ray/dag/compiled_dag_node.py

python/ray/experimental/channel/shared_memory_channel.py

ruisearch42 · 2024-08-26T15:50:21Z

python/ray/experimental/channel/shared_memory_channel.py

+    exhausted. The caller of the API should guarantee to consume buffers
+    before they are exhausted, otherwise write/read raises
+    `RayChannelTimeoutError`.


Maybe saying something like the following is more accurate:
If the buffer is full for writes or empty for reads and the operation blocks longer than configured timeouts, RayChannelTimeoutError will be raised.

updated accordingly.

python/ray/experimental/channel/shared_memory_channel.py

python/ray/dag/compiled_dag_node.py

ruisearch42 · 2024-08-26T16:07:47Z

python/ray/dag/compiled_dag_node.py

+            max_buffered_inputs: The maximum number of in-flight requests that
+                are allowed to be buffered. Before submitting more requests,


hmm this is not the max number of inflight requests but rather num of buffers per channel. For example, if there are two channels, then the max in-flight requests is at least 2 * channel_num_buffers

actually I'd like to keep the semantic of max_in_flight_requests. What about

we change it to num_in_flight_requests

We use num_in_flight_requests * 2 as the num buffer size only for shm channel?

It would be even better if we could avoid adding the new config and just use self._max_buffered_results. They are trying to accomplish the same thing, I believe, just whether we buffer on the input or output.

Also, it's okay if the max buffered results is conservative, i.e. the DAG has more capacity than what is requested through max buffered results. This can always be improved later.

For now, I would say the best is to determine the number of buffers per channel to use based on the existing max_buffered_results config (which we can rename to something like max_inflight_executions) and additionally add a "global memory limit" for all buffers' sizes, in case the user puts something very high like 1M max in-flight executions.

Also, it's okay if the max buffered results is conservative, i.e. the DAG has more capacity than what is requested through max buffered results. This can always be improved later.

yeah +1 on this! And the plan we discussed exactly aligns with what you suggested (rename it to max_inflight_executions and set the buffer size the same, which is the most conservative config)

For global limit, let me follow up in other PR actually. I tried, and I think it involves in some logics, so I think it is cleaner to separate it (I think no one is going to increase this so high yet, so it should be okay)

Sounds great! How about just filing issues for now?

global memory limit for buffers

dynamically allocating additional buffers

ruisearch42 · 2024-08-26T16:10:58Z

python/ray/dag/compiled_dag_node.py

@@ -1542,6 +1562,18 @@ def run(self):
        monitor.start()
        return monitor

+    def raise_if_too_many_inflight_requests(self):
+        num_in_flight_requests = self._execution_index - self._max_execution_index
+        if num_in_flight_requests > self._max_buffered_inputs:


The comparison would be more involved:
max_num_in_flight_requests is not max_buffered_inputs, but also related to the number of channels.

See the other comment as well.

it is cleaner with a new name max_num_in_flight_requests

python/ray/dag/context.py

ruisearch42 · 2024-08-26T16:13:11Z

python/ray/experimental/channel/common.py

@@ -77,6 +77,7 @@ def create_channel(
        self,
        writer: Optional["ray.actor.ActorHandle"],
        reader_and_node_list: List[Tuple["ray.actor.ActorHandle", str]],
+        num_shm_buffers: Optional[int] = None,


At channel constructor, we can just all it num_buffers to be more general and not referring to a specific implementation of shared memory. We can still assert this is None for nccl.

Instead of this, I moved it to a constructor arg of SharedMemoryType, so it is only applied to shm channel. I currently don't really see the need of buffer in nccl. We can change it in the future if needed

python/ray/dag/compiled_dag_node.py

kevin85421 · 2024-08-27T01:20:44Z

python/ray/dag/compiled_dag_node.py

+    def raise_if_too_many_inflight_requests(self):
+        num_in_flight_requests = self._execution_index - self._max_execution_index
+        if num_in_flight_requests > self._max_buffered_inputs:
+            raise ray.exceptions.RayChannelBufferAtMaxCapacity(


What happens if the condition self.num_in_flight_requests > self._max_buffered_inputs is met? Will it cause a deadlock, or is it possible for the execution to still finish but slower?

If it is still possible to finish, maybe we should consider using a warning message instead of raising an exception. Consider the case where a job has already been running for a long time but finally fails because it executes too frequently in the last few steps.

Then it is the same semantic as before. It can timeout depending on how many channels are available. (this is the problem we want to fix. Basically problem 1 in this doc https://docs.google.com/document/d/1Ka_HFwPBNIY1u3kuroHOSZMEQ8AgwpYciZ4n08HJ0Xc/edit#heading=h.8j5v3i9ykgsy).

This raises when .execute() is called, so actually we have no problem you mentioned.

kevin85421 · 2024-08-27T01:32:54Z

python/ray/dag/context.py

+# output. The CPU memory overhead per shared memory channel is
+# DEFAULT_BUFFER_SIZE_BYTES * DEFAULT_MAX_BUFFERED_INPUTS even when channel is unused.
+# There's no additional memory impact on Nccl channels.
+DEFAULT_MAX_BUFFERED_INPUTS = int(os.environ.get("RAY_DAG_max_buffered_results", 10))


Suggested change

DEFAULT_MAX_BUFFERED_INPUTS = int(os.environ.get("RAY_DAG_max_buffered_results", 10))

DEFAULT_MAX_BUFFERED_INPUTS = int(os.environ.get("RAY_DAG_max_buffered_inputs", 10))

oops good catch

python/ray/experimental/channel/shared_memory_channel.py

kevin85421 · 2024-08-27T01:47:46Z

python/ray/experimental/channel/shared_memory_channel.py

+        self._channels = [
+            Channel(writer, reader_and_node_list, typ) for _ in range(num_shm_buffers)
+        ]
+        self._next_write_index = 0


Can you add some comments to explain how _next_write_index works? Maybe we could use self.execution_index instead and increment it whenever write or read is called on the writer and reader. When retrieving the buffer, we can use execution_index % num_shm_buffers. Using execution_index will also make debugging easier.

Are you saying we should pass self.execution_index from a caller?

rkooo567 · 2024-08-24T17:02:23Z

python/ray/dag/tests/experimental/test_accelerated_dag.py

@@ -470,39 +471,6 @@ def test_chain_dag(ray_start_regular, num_actors):
    compiled_dag.teardown()


-def test_execution_timeout(ray_start_regular):


the test is not relevant anymore because it relies on that buffering is not possible

Can we keep it while disabling buffering?

it is not possible to disable buffering. it will raise exception in this case

the test should be done by test_channel.py

python/ray/dag/compiled_dag_node.py

rkooo567 · 2024-08-27T16:22:37Z

python/ray/dag/compiled_dag_node.py

+    def raise_if_too_many_inflight_requests(self):
+        num_in_flight_requests = self._execution_index - self._max_execution_index
+        if num_in_flight_requests > self._max_buffered_inputs:
+            raise ray.exceptions.RayChannelBufferAtMaxCapacity(


Then it is the same semantic as before. It can timeout depending on how many channels are available. (this is the problem we want to fix. Basically problem 1 in this doc https://docs.google.com/document/d/1Ka_HFwPBNIY1u3kuroHOSZMEQ8AgwpYciZ4n08HJ0Xc/edit#heading=h.8j5v3i9ykgsy).

This raises when .execute() is called, so actually we have no problem you mentioned.

rkooo567 · 2024-08-27T16:23:04Z

python/ray/dag/context.py

+# output. The CPU memory overhead per shared memory channel is
+# DEFAULT_BUFFER_SIZE_BYTES * DEFAULT_MAX_BUFFERED_INPUTS even when channel is unused.
+# There's no additional memory impact on Nccl channels.
+DEFAULT_MAX_BUFFERED_INPUTS = int(os.environ.get("RAY_DAG_max_buffered_results", 10))


oops good catch

rkooo567 · 2024-08-27T16:23:54Z

python/ray/dag/compiled_dag_node.py

+            max_buffered_inputs: The maximum number of in-flight requests that
+                are allowed to be buffered. Before submitting more requests,


actually I'd like to keep the semantic of max_in_flight_requests. What about

we change it to num_in_flight_requests

We use num_in_flight_requests * 2 as the num buffer size only for shm channel?

rkooo567 · 2024-08-27T16:26:33Z

python/ray/experimental/channel/shared_memory_channel.py

+    exhausted. The caller of the API should guarantee to consume buffers
+    before they are exhausted, otherwise write/read raises
+    `RayChannelTimeoutError`.


updated accordingly.

rkooo567 · 2024-08-27T16:30:22Z

python/ray/experimental/channel/shared_memory_channel.py

+        self._channels = [
+            Channel(writer, reader_and_node_list, typ) for _ in range(num_shm_buffers)
+        ]
+        self._next_write_index = 0


Are you saying we should pass self.execution_index from a caller?

python/ray/experimental/channel/shared_memory_channel.py

rkooo567 · 2024-08-27T19:15:27Z

Discussed offline.

max_buffered_inputs is more like capacity of the dag. I will change the name to max_in_flight_requests. We are using the most conservative capacity heuristic (which is the same number of # of shm buffers) which has no timeout in any case.
num_shm_buffer is specific to shm. I will move logics to SharedMemoryType.

stephanie-wang

Looking good, I agree with the approach of having multiple channels, and it'd be great if we can later allocate them dynamically too. Will take another look once the current comments are addressed!

stephanie-wang · 2024-08-27T23:02:53Z

python/ray/dag/compiled_dag_node.py

+            max_buffered_inputs: The maximum number of in-flight requests that
+                are allowed to be buffered. Before submitting more requests,


It would be even better if we could avoid adding the new config and just use self._max_buffered_results. They are trying to accomplish the same thing, I believe, just whether we buffer on the input or output.

python/ray/dag/compiled_dag_node.py

stephanie-wang · 2024-08-27T23:07:03Z

python/ray/dag/compiled_dag_node.py

+            max_buffered_inputs: The maximum number of in-flight requests that
+                are allowed to be buffered. Before submitting more requests,


Also, it's okay if the max buffered results is conservative, i.e. the DAG has more capacity than what is requested through max buffered results. This can always be improved later.

For now, I would say the best is to determine the number of buffers per channel to use based on the existing max_buffered_results config (which we can rename to something like max_inflight_executions) and additionally add a "global memory limit" for all buffers' sizes, in case the user puts something very high like 1M max in-flight executions.

stephanie-wang · 2024-08-27T23:08:10Z

python/ray/dag/compiled_dag_node.py

@@ -1036,6 +1055,7 @@ def _get_or_compile(
                    self,
                    reader_and_node_list,
                    typ=type_hint,
+                    num_shm_buffers=self._max_buffered_inputs,


Does the num buffers only apply for the DAG input or is it to all channels in the DAG?

it is applied to all channels. This way, there's a additional benefit that the last stage is not blocked when ray.get is not called.

stephanie-wang · 2024-08-27T23:09:03Z

python/ray/experimental/channel/shared_memory_channel.py

@@ -109,6 +110,8 @@ def create_channel(
            writer: The actor that may write to the channel. None signifies the driver.
            reader_and_node_list: A list of tuples, where each tuple contains a reader
                actor handle and the node ID where the actor is located.
+            num_shm_buffers: The number of shared memory buffer per channel.
+                It is currently ignored for nccl channel.


Instead of adding this note, just raise NotImplementedError for NCCL channels.

stephanie-wang · 2024-08-27T23:09:45Z

python/ray/experimental/channel/shared_memory_channel.py

@@ -480,6 +484,71 @@ def close(self) -> None:
        self._worker.core_worker.experimental_channel_set_error(self._reader_ref)


+@DeveloperAPI
+class BufferedSharedMemoryChannel(ChannelInterface):


Why a new class instead of extending Channel?

I think we don't really need to inherit any implementation, so I made a new class

Oh I meant just to add directly to Channel class, instead of making a new class or subclass.

oh I see. I found it a little cleaner to implement it this way. But no strong opinion if you prefer that way. Let me know if you want me to change this!

Hmm I see, yeah that makes sense. Initially, I was thinking that having too many different classes gets a bit complicated once we start wanting to wrap or subclass multiple of them. But for now this seems fine, probably we should just refactor ChannelInterface at some point...

rkooo567

@stephanie-wang @ruisearch42 @kevin85421 every comment addressed. Let's try merging it today if possible!

now we are using max_in_flight_requests
As we discussed, we set buffer size to max_in_flight_requests, which is the most conservative policy but there's no false positive timeout
shm_buffer arg moved to a constructor of SharedMemoryType

rkooo567 · 2024-08-28T05:11:13Z

python/ray/dag/compiled_dag_node.py

+            max_buffered_inputs: The maximum number of in-flight requests that
+                are allowed to be buffered. Before submitting more requests,


Also, it's okay if the max buffered results is conservative, i.e. the DAG has more capacity than what is requested through max buffered results. This can always be improved later.

yeah +1 on this! And the plan we discussed exactly aligns with what you suggested (rename it to max_inflight_executions and set the buffer size the same, which is the most conservative config)

rkooo567 · 2024-08-28T05:13:24Z

python/ray/dag/compiled_dag_node.py

+            max_buffered_inputs: The maximum number of in-flight requests that
+                are allowed to be buffered. Before submitting more requests,


For global limit, let me follow up in other PR actually. I tried, and I think it involves in some logics, so I think it is cleaner to separate it (I think no one is going to increase this so high yet, so it should be okay)

python/ray/dag/compiled_dag_node.py

rkooo567 · 2024-08-28T06:08:42Z

python/ray/dag/compiled_dag_node.py

@@ -1542,6 +1562,18 @@ def run(self):
        monitor.start()
        return monitor

+    def raise_if_too_many_inflight_requests(self):
+        num_in_flight_requests = self._execution_index - self._max_execution_index
+        if num_in_flight_requests > self._max_buffered_inputs:


it is cleaner with a new name max_num_in_flight_requests

rkooo567 · 2024-08-28T06:09:26Z

python/ray/experimental/channel/common.py

@@ -77,6 +77,7 @@ def create_channel(
        self,
        writer: Optional["ray.actor.ActorHandle"],
        reader_and_node_list: List[Tuple["ray.actor.ActorHandle", str]],
+        num_shm_buffers: Optional[int] = None,


Instead of this, I moved it to a constructor arg of SharedMemoryType, so it is only applied to shm channel. I currently don't really see the need of buffer in nccl. We can change it in the future if needed

rkooo567 · 2024-08-28T06:11:29Z

python/ray/experimental/channel/shared_memory_channel.py

@@ -480,6 +484,71 @@ def close(self) -> None:
        self._worker.core_worker.experimental_channel_set_error(self._reader_ref)


+@DeveloperAPI
+class BufferedSharedMemoryChannel(ChannelInterface):


oh I see. I found it a little cleaner to implement it this way. But no strong opinion if you prefer that way. Let me know if you want me to change this!

python/ray/exceptions.py

python/ray/experimental/channel/shared_memory_channel.py

ruisearch42 · 2024-08-28T18:37:05Z

python/ray/experimental/channel/shared_memory_channel.py

+        # A single channel is not supposed to read and write at the same time.
+        assert self._next_read_index == 0


QQ: why is the value 0 special here? Should we just check if the write index equals the read index?

one channel doesn't call both read/write. So if your channel writes, there's no read

How about instead moving this logic to ensure_registered_as_writer/reader? You can initialize the indices to -1, and check the other index in those methods.

I tried this, but I feel like it is not the best way to do it because we don't require to call ensure_* before using read/write technically. So I feel like asserting in read/write is actually better. lmk if you disagree with this comment, I will follow up in other PR

ruisearch42 · 2024-08-28T18:37:21Z

python/ray/experimental/channel/shared_memory_channel.py

+        self._next_write_index %= self._num_shm_buffers
+
+    def read(self, timeout: Optional[float] = None) -> Any:
+        """Read a value to a channel.


nit: from a channel

python/ray/experimental/channel/shared_memory_channel.py

\Based on https://docs.google.com/document/d/1Ka_HFwPBNIY1u3kuroHOSZMEQ8AgwpYciZ4n08HJ0Xc/edit When there are many in-flight requests (pipelining inputs to the DAG), 2 problems occur. Input submitter timeout. InputSubmitter.write() waits until the buffer is read from downstream tasks. Since timeout count is started as soon as InputSubmitter.write() is called, when there are many in-flight requests, the later requests are likely to timeout. Pipeline bubble. Output fetcher doesn’t read the channel until CompiledDagRef.get is called. It means the upstream task (actor 2) has to be blocked until .get is called from a driver although it can execute tasks. This PR solves the problem by providing multiple buffer per shm channel. Note that the buffering is not supported for nccl yet (we can do it when we overlap compute/comm). Main changes Introduce BufferedSharedMemoryChannel which allows to create multiple buffers (10 by default). Read/write is done in round robin manner. When you have more in-flight request than the buffer size, Dag can still have timeout error. To make debugging easy and behavior straightforward, we introduce max_buffered_inputs_ argument. If there are more than max_buffered_inputs_ requests submitted to the dag without ray.get, it immediately raises an exception. Signed-off-by: ujjawal-khare <[email protected]>

…project#47320) When users do read(timeout=0) or write(timeout=0), one may expect it returns immediately if the buffer is readable/writable. It is not true in some cases (I found it from ray-project#47272). The root cause is that we apply timeout when we acquire headers. When a buffer is writable (its obj buffer is not acquired), it is possible write(timeout=0) fails if there are readers trying to acquire a header because we apply the same timeout when we acquire a header. This PR fixes the isuse by not applying the timeout in this case. It is okay because acquiring header takes a very short time, and it is immediately released. Signed-off-by: ujjawal-khare <[email protected]>

\Based on https://docs.google.com/document/d/1Ka_HFwPBNIY1u3kuroHOSZMEQ8AgwpYciZ4n08HJ0Xc/edit When there are many in-flight requests (pipelining inputs to the DAG), 2 problems occur. Input submitter timeout. InputSubmitter.write() waits until the buffer is read from downstream tasks. Since timeout count is started as soon as InputSubmitter.write() is called, when there are many in-flight requests, the later requests are likely to timeout. Pipeline bubble. Output fetcher doesn’t read the channel until CompiledDagRef.get is called. It means the upstream task (actor 2) has to be blocked until .get is called from a driver although it can execute tasks. This PR solves the problem by providing multiple buffer per shm channel. Note that the buffering is not supported for nccl yet (we can do it when we overlap compute/comm). Main changes Introduce BufferedSharedMemoryChannel which allows to create multiple buffers (10 by default). Read/write is done in round robin manner. When you have more in-flight request than the buffer size, Dag can still have timeout error. To make debugging easy and behavior straightforward, we introduce max_buffered_inputs_ argument. If there are more than max_buffered_inputs_ requests submitted to the dag without ray.get, it immediately raises an exception. Signed-off-by: ujjawal-khare <[email protected]>

ip

2c7645b

rkooo567 added the go add ONLY when ready to merge, run all tests label Aug 22, 2024

.

1c67606

ruisearch42 reviewed Aug 22, 2024

View reviewed changes

kevin85421 assigned kevin85421 and ruisearch42 Aug 22, 2024

SangBin Cho added 9 commits August 22, 2024 23:54

ip

6997436

ip 2

43cf80b

fully working.

15d5a1e

.

a8555f8

ip

da826d2

remove breakpoint

256b0ec

ip

fe0ff80

done

f560a18

done

36c36c5

rkooo567 mentioned this pull request Aug 24, 2024

[Core][aDAG] Remove timeout for acquiring header for read/write #47320

Merged

8 tasks

ruisearch42 reviewed Aug 26, 2024

View reviewed changes

kevin85421 reviewed Aug 27, 2024

View reviewed changes

Merge branch 'master' into buffer-input-DAG

5d7f2ce

rkooo567 commented Aug 27, 2024

View reviewed changes

stephanie-wang reviewed Aug 27, 2024

View reviewed changes

rkooo567 added 2 commits August 27, 2024 22:50

addressed the first set of reviews.

bf690ac

Addressed all code review.

286e633

rkooo567 requested a review from a team as a code owner August 28, 2024 06:19

rkooo567 commented Aug 28, 2024

View reviewed changes

ruisearch42 reviewed Aug 28, 2024

View reviewed changes

ruisearch42 mentioned this pull request Dec 3, 2024

[core][compiled graphs] Fix and re-enable shared memory channel buffering support #49044

Closed



		# @DeveloperAPI
		class BufferedRemoteChannel(ChannelInterface):

		max_buffered_inputs: The maximum number of in-flight requests that
		are allowed to be buffered. Before submitting more requests,

	DEFAULT_MAX_BUFFERED_INPUTS = int(os.environ.get("RAY_DAG_max_buffered_results", 10))
	DEFAULT_MAX_BUFFERED_INPUTS = int(os.environ.get("RAY_DAG_max_buffered_inputs", 10))

		@@ -470,39 +471,6 @@ def test_chain_dag(ray_start_regular, num_actors):
		compiled_dag.teardown()


		def test_execution_timeout(ray_start_regular):

		# A single channel is not supposed to read and write at the same time.
		assert self._next_read_index == 0

[aDAG] support buffered input #47272

[aDAG] support buffered input #47272

Conversation

rkooo567 commented Aug 22, 2024 • edited Loading

Why are these changes needed?

Q&A

Related issue number

Checks

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rkooo567 commented Aug 24, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rkooo567 commented Aug 27, 2024

stephanie-wang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rkooo567 left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rkooo567 commented Aug 22, 2024 •

edited

Loading

rkooo567 left a comment •

edited

Loading