Make unbounded fanout messages spillable #711
Merged
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Signed-off-by: niranda perera <niranda.perera@gmail.com>
|
Auto-sync is disabled for draft pull requests in this repository. Workflows must be run manually. Contributors can view more details about this message here. |
nirandaperera
added
improvement
…ded-fanout-state-spillable
nirandaperera
commented
Dec 5, 2025
madsbk
reviewed
Dec 5, 2025
rapids-bot bot
pushed a commit
that referenced
this pull request
Dec 5, 2025
```c++
/**
* @brief Create a deep copy of a message without removing it.
*
* This method duplicates the message identified by `mid` while leaving the
* original message intact inside the container. The returned message is a
* full deep copy of the payload. If the message is currently being spilled
* by another thread, this call waits until spilling completes.
*
* @param mid Message identifier.
* @param reservation Memory reservation used for allocating buffers during
* the deep copy. The reservation also determines the memory type of the
* returned message.
*
* @return A deep copy of the referenced `Message`.
*
* @throws std::out_of_range If the message has already been extracted.
* @throws std::runtime_error If required memory cannot be allocated using
* the provided reservation.
*/
[[nodiscard]] Message copy(MessageId mid, MemoryReservation& reservation);
```
Based on #711
Authors:
- Mads R. B. Kristensen (https://github.com/madsbk)
Approvers:
- Peter Andreas Entschev (https://github.com/pentschev)
URL: #713
…ded-fanout-state-spillable Signed-off-by: niranda perera <niranda.perera@gmail.com>
Signed-off-by: niranda perera <niranda.perera@gmail.com>
…ded-fanout-state-spillable
…ded-fanout-state-spillable Signed-off-by: niranda perera <niranda.perera@gmail.com>
…ded-fanout-state-spillable Signed-off-by: niranda perera <niranda.perera@gmail.com>
nirandaperera
force-pushed
the
Make-unbounded-fanout-state-spillable
branch
from
ab0a1e5 to
feb04a8
Compare
Signed-off-by: niranda perera <niranda.perera@gmail.com>
Signed-off-by: niranda perera <niranda.perera@gmail.com>
Signed-off-by: niranda perera <niranda.perera@gmail.com>
madsbk
requested changes
Dec 9, 2025
Co-authored-by: Mads R. B. Kristensen <madsbk@gmail.com>
Signed-off-by: niranda perera <niranda.perera@gmail.com>
This reverts commit 5606f6a. Signed-off-by: niranda perera <niranda.perera@gmail.com>
…ded-fanout-state-spillable
…randaperera/rapidsmpf into Make-unbounded-fanout-state-spillable
Signed-off-by: niranda perera <niranda.perera@gmail.com>
madsbk
requested changes
Dec 11, 2025
wence-
reviewed
Dec 11, 2025
Comment on lines
+48
to
+61
| constexpr std::span<MemoryType const> leq_memory_types(MemoryType mem_type) noexcept { | ||
| return std::views::drop_while(MEMORY_TYPES, [&](MemoryType const& mt) { | ||
| return mt != mem_type; | ||
| }); | ||
| } | ||
|
|
||
| static_assert(std::ranges::equal(leq_memory_types(MemoryType::DEVICE), MEMORY_TYPES)); | ||
| static_assert(std::ranges::equal( | ||
| leq_memory_types(MemoryType::HOST), std::ranges::single_view{MemoryType::HOST} | ||
| )); | ||
| // unknown memory type should return an empty view | ||
| static_assert(std::ranges::equal( | ||
| leq_memory_types(static_cast<MemoryType>(-1)), std::ranges::empty_view<MemoryType>{} | ||
| )); |
Contributor
There was a problem hiding this comment.
Why not immediately just have:
template<typename Pred>
constexpr std::span<MemoryType const> filter_types(MemoryType mem_type, Pred pred) noexcept {
return std::views::filter(MEMORY_TYPES, pred);
}
?
Contributor
Author
There was a problem hiding this comment.
This would return a Range isnt it? But yes, sure.
Co-authored-by: Mads R. B. Kristensen <madsbk@gmail.com>
Signed-off-by: niranda perera <niranda.perera@gmail.com>
…ded-fanout-state-spillable
madsbk
approved these changes
Dec 11, 2025
cpp/tests/streaming/test_fanout.cpp
Outdated
| } | ||
|
|
||
| nodes.push_back(node::fanout(ctx, in, out_chs, policy)); | ||
|
|
| nodes.push_back( | ||
| many_input_sink(ctx, out_chs, ConsumePolicy::CHANNEL_ORDER, outs) | ||
| ); | ||
|
|
Signed-off-by: niranda perera <niranda.perera@gmail.com>
…ded-fanout-state-spillable
Signed-off-by: niranda perera <niranda.perera@gmail.com>
Contributor
Author
|
/merge |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Closes #675