Chunk with multiple messages #251
Open
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 <[email protected]>
|
If you're merging branch-25.08 into your PR you need to retarget it to branch-25.08 as well. Was that intended @nirandaperera ? |
…ti-packed-data-chunk
Signed-off-by: niranda perera <[email protected]>
nirandaperera
force-pushed
the
multi-packed-data-chunk
branch
from
fd59be4 to
aba10c0
Compare
Thanks @pentschev. It was a mistake. I force pushed the changes now |
Signed-off-by: niranda perera <[email protected]>
Signed-off-by: niranda perera <[email protected]>
nirandaperera
added
breaking
Signed-off-by: niranda perera <[email protected]>
Signed-off-by: niranda perera <[email protected]>
wence-
reviewed
May 7, 2025
| /** | ||
| * @brief Chunk with multiple messages. | ||
| * | ||
| * This class will have two buffers: |
There was a problem hiding this comment.
Suggested change
| * This class will have two buffers: | |
| * This class will has two buffers: |
| * - data_: The data buffer that contains the concatenateddata of the messages in the | ||
| * chunk. | ||
| * | ||
| * The metadata_ buffer will have the following format: |
There was a problem hiding this comment.
Suggested change
| * The metadata_ buffer will have the following format: | |
| * The metadata_ buffer has the following format: |
Comment on lines
45
to
47
| * - [psum_meta]: std::vector<uint32_t>, Prefix sums (excluding 0) of the metadata | ||
| * sizes of the messages, size = n_elements | ||
| * - [psum_data]: std::vector<uint64_t>, Prefix sums (excluding 0) of the data sizes of |
There was a problem hiding this comment.
Can we call this something like metadata_offsets and data_offsets respectively?
| * @return The number of messages in the chunk. | ||
| */ | ||
| inline size_t n_messages() const { | ||
| return *reinterpret_cast<size_t*>(metadata_->data() + sizeof(ChunkID)); |
There was a problem hiding this comment.
All these reinterpret_cast type-punning approaches break strict-aliasing rules unfortunately.
What we should do is:
- (C++ 20) use
std::bit_cast(but it's messy because we're carrying aroundstd::byteandbit_casttakes values, not pointer + size). - Use
memcpy(the compiler will optimise this)
So, this would be (for example):
inline size_t n_messages() const {
size_t result;
memcpy(&result, metadata_->data() + sizeof(ChunkID), sizeof(result));
return result;
}
There was a problem hiding this comment.
I see. TIL. Sure, I will change to memcpy.
Signed-off-by: niranda perera <[email protected]>
Signed-off-by: niranda perera <[email protected]>
Signed-off-by: niranda perera <[email protected]>
Signed-off-by: niranda perera <[email protected]>
…ti-packed-data-chunk Signed-off-by: niranda perera <[email protected]>
|
/ok to test |
Signed-off-by: niranda perera <[email protected]>
Signed-off-by: niranda perera <[email protected]>
madsbk
reviewed
May 9, 2025
There was a problem hiding this comment.
Overall looks good!
@nirandaperera, let's prioritize getting full chunk support. I think it will have a significant impact!
| * is a control message, otherwise zero (data message). | ||
| */ | ||
| inline size_t expected_num_chunks(size_t i) const { | ||
| return expected_num_chunks_[i]; |
There was a problem hiding this comment.
Suggested change
| return expected_num_chunks_[i]; | |
| return expected_num_chunks_.at(i); |
| * @return True if the message is a control message, false otherwise. | ||
| */ | ||
| inline bool is_control_message(size_t i) const { | ||
| return expected_num_chunks_[i] > 0; |
There was a problem hiding this comment.
Suggested change
| return expected_num_chunks_[i] > 0; | |
| return expected_num_chunks(i) > 0; |
| */ | ||
| [[nodiscard]] std::unique_ptr<cudf::table> unpack(rmm::cuda_stream_view stream) const; | ||
| static Chunk from_serialized_buf( |
There was a problem hiding this comment.
Suggested change
| static Chunk from_serialized_buf( | |
| static Chunk from_serialized_buffer( |
cpp/src/shuffler/chunk.cpp
Outdated
| return {chunk_id, 1, {part_id}, {expected_num_chunks}, {0}, {0}}; | ||
| } | ||
|
|
||
| Chunk Chunk::from_serialized_buf(std::vector<uint8_t> const& msg, bool validate) { |
There was a problem hiding this comment.
Suggested change
| Chunk Chunk::from_serialized_buf(std::vector<uint8_t> const& msg, bool validate) { | |
| Chunk Chunk::from_serialized_buffer(std::vector<uint8_t> const& msg, bool validate) { |
Signed-off-by: niranda perera <[email protected]>
…ti-packed-data-chunk
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
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.
Chunk with multiple messages. This PR only moves the existing
Chunkclass to the new impl and it would only have 1 message in it.This class has two buffers:
metadata_: The metadata buffer that contains information about the messages in the chunks and the concatenated metadata of the messages.data_: The data buffer that contains the concatenateddata of the messages in the chunk.All the chunk information will be encoded to the metadata_ buffer as follows.
The metadata_ buffer uses the following format:
chunk_id: uint64_t, ID of the chunkn_elements: size_t, Number of messages in the chunk[partition_ids]: vector, Partition IDs of the messages, size = n_elements[expected_num_chunks]: vector<size_t>, Expected number of chunks of the messages, size = n_elements[meta_offsets]: vector<uint32_t>, Offsets (excluding 0) of the metadata sizes of the messages, size = n_elements[data_offsets]: vector<uint64_t>, Offsets (excluding 0) of the data sizes of the messages, size = n_elements[concat_metadata]: vector<uint8_t>, Concatenated metadata of the messages, size = meta_offsets[n_elements - 1]For a chunk with N messages with M bytes of concat metadata the size of metadata_ buffer is
sizeof(ChunkID) + sizeof(size_t) + N (sizeof(PartID) + sizeof(size_t) + sizeof(uint32_t) + sizeof(uint64_t)) + M = 16 + N 24 + Mbytes.For a chunk with a single control message, the size of the metadata_ buffer is
sizeof(ChunkID) + sizeof(PartID)+ 2*sizeof(size_t) + sizeof(uint32_t) + sizeof(uint64_t) = 40bytes.For a chunk with a single message with M bytes of metadata, the size of the metadata_ buffer is
sizeof(ChunkID) + sizeof(PartID) + sizeof(size_t) + sizeof(uint32_t) + sizeof(ChunkID) + sizeof(PartID) + sizeof(size_t) + sizeof(uint32_t) + sizeof(uint64_t) + M = 40 + Mbytes.