[#590][part-1] ManagedBuffer instead ByteBuf to hold ShuffleData #906
Conversation
…ffleDataResponse and ShuffleResultData
Codecov Report
@@ Coverage Diff @@
## master #906 +/- ##
============================================
+ Coverage 55.22% 56.99% +1.77%
- Complexity 2197 2280 +83
============================================
Files 333 327 -6
Lines 16444 14675 -1769
Branches 1306 1354 +48
============================================
- Hits 9081 8364 -717
+ Misses 6851 5834 -1017
+ Partials 512 477 -35
... and 93 files with indirect coverage changes 📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more |
|
@leixm Could you help me review this pr? |
Sure. |
| ByteBufUtils.copyByteBuf(data, buf); | ||
| data.release(); | ||
| if (buffer instanceof FileSegmentManagedBuffer) { | ||
| buf.writeInt(buffer.size()); |
There was a problem hiding this comment.
Why do we only write the size of buffer?
There was a problem hiding this comment.
FileSegmentManagedBuffer.convertToNetty() returned DefaultFileRegion type object , in order to achieve sendfile the DefaultFileRegion type object should writed by channel.write(object).
could you give same other suggest of realization for sendfile in netty?
There was a problem hiding this comment.
How do the other systems like Spark, Celeborn implement this?
There was a problem hiding this comment.
In order to keep the uniffle interface design as much as possible, reference Celeborn&Spark add ManagedBuffer to hold different type ShuffleData(bytebuf or FileRegion).
There was a problem hiding this comment.
In order to keep the uniffle interface design as much as possible, reference Celeborn&Spark add ManagedBuffer to hold different type ShuffleData(bytebuf or FileRegion).
Ok for me.
| @@ -59,5 +59,9 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) | |||
| byteBuf.release(); | |||
| } | |||
| ctx.writeAndFlush(byteBuf); | |||
| // do transferTo send data after encode buffer send. | |||
| if (message instanceof Transferable) { | |||
| ((Transferable) message).transferTo(ctx.channel()); | |||
There was a problem hiding this comment.
@jerqi the line is write the FileSegmentManageBuffer ShuffleData.
| } | ||
| return ByteBufUtils.readBytes(data); | ||
| return ByteBufUtils.readBytes(Unpooled.wrappedBuffer(buffer.nioByteBuffer())); |
There was a problem hiding this comment.
Maybe we can return ByteBufUtils.readBytes(buffer.byteBuf());
| } | ||
|
|
||
| public ByteBuf getDataBuf() { | ||
| return data; | ||
| return Unpooled.wrappedBuffer(buffer.nioByteBuffer()); |
|
Maybe we can also make |
What's the benefit? |
What changes were proposed in this pull request?
part-1:
Why are the changes needed?
Fix: #590
Does this PR introduce any user-facing change?
(Please list the user-facing changes introduced by your change, including
No.
How was this patch tested?
Integration Testing