You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@uniffle.apache.org by "zhengchenyu (via GitHub)" <gi...@apache.org> on 2024/10/18 13:01:53 UTC
[PR] [#2173] feat(remote merge): support netty for remote merge. [incubator-uniffle]
zhengchenyu opened a new pull request, #2202:
URL: https://github.com/apache/incubator-uniffle/pull/2202
### What changes were proposed in this pull request?
Support netty for remote merge. Use direct ByteBuf to replace with byte[] when netty is enable. And optimized code structure to avoid memory leaks
### Why are the changes needed?
Fix: #2173
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
unit test, integration test, real job in cluster.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#2173] feat(remote merge): support netty for remote merge. [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #2202:
URL: https://github.com/apache/incubator-uniffle/pull/2202#discussion_r1810041194
##########
common/src/main/java/org/apache/uniffle/common/serializer/writable/WritableSerializerInstance.java:
##########
@@ -47,19 +47,28 @@ public <T> T deserialize(DataInputBuffer buffer, Class vClass) throws IOExceptio
}
@Override
- public <K, V> SerializationStream serializeStream(OutputStream output, boolean raw) {
+ public <K, V> SerializationStream serializeStream(
+ SerOutputStream output, boolean raw, boolean shared) {
if (raw) {
- return new RawWritableSerializationStream(this, output);
+ if (shared) {
Review Comment:
BufferDeserializationStream and PartitionDeserializationStream may be better.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#2173] feat(remote merge): support netty for remote merge. [incubator-uniffle]
Posted by "zhengchenyu (via GitHub)" <gi...@apache.org>.
zhengchenyu commented on code in PR #2202:
URL: https://github.com/apache/incubator-uniffle/pull/2202#discussion_r1809821621
##########
common/src/main/java/org/apache/uniffle/common/serializer/writable/WritableSerializerInstance.java:
##########
@@ -47,19 +47,28 @@ public <T> T deserialize(DataInputBuffer buffer, Class vClass) throws IOExceptio
}
@Override
- public <K, V> SerializationStream serializeStream(OutputStream output, boolean raw) {
+ public <K, V> SerializationStream serializeStream(
+ SerOutputStream output, boolean raw, boolean shared) {
if (raw) {
- return new RawWritableSerializationStream(this, output);
+ if (shared) {
Review Comment:
SharedSerializationStream will not be accessed by multiple threads, can only be executed under the merge thread corresponding to one partition. Here, 'Shared' means use shared buffer to merge.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#2173] feat(remote merge): support netty for remote merge. [incubator-uniffle]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #2202:
URL: https://github.com/apache/incubator-uniffle/pull/2202#issuecomment-2422479522
## Test Results
2 904 files + 9 2 904 suites +9 5h 49m 16s :stopwatch: + 2m 11s
1 082 tests + 39 1 078 :white_check_mark: + 38 2 :zzz: ±0 1 :x: ±0 1 :fire: +1
13 567 runs +594 13 535 :white_check_mark: +593 30 :zzz: ±0 1 :x: ±0 1 :fire: +1
For more details on these failures and errors, see [this check](https://github.com/apache/incubator-uniffle/runs/31734266447).
Results for commit 7826b165. ± Comparison against base commit f9b4c0e0.
<details>
<summary>This pull request <b>removes</b> 20 and <b>adds</b> 59 tests. <i>Note that renamed tests count towards both.</i></summary>
```
org.apache.uniffle.common.records.RecordsReaderWriterTest ‑ testWriteAndReadRecordFile1{String, File}[1]
org.apache.uniffle.common.records.RecordsReaderWriterTest ‑ testWriteAndReadRecordFile1{String, File}[2]
org.apache.uniffle.common.records.RecordsReaderWriterTest ‑ testWriteAndReadRecordFile2{String, File}[1]
org.apache.uniffle.common.records.RecordsReaderWriterTest ‑ testWriteAndReadRecordFile2{String, File}[2]
org.apache.uniffle.common.records.RecordsReaderWriterTest ‑ testWriteAndReadRecordFile3{String, File}[1]
org.apache.uniffle.common.records.RecordsReaderWriterTest ‑ testWriteAndReadRecordFile3{String, File}[2]
org.apache.uniffle.common.records.RecordsReaderWriterTest ‑ testWriteAndReadRecordFile4{String, File}[1]
org.apache.uniffle.common.records.RecordsReaderWriterTest ‑ testWriteAndReadRecordFile4{String, File}[2]
org.apache.uniffle.common.serializer.PartialInputStreamTest ‑ testReadFileInputStream
org.apache.uniffle.common.serializer.PartialInputStreamTest ‑ testReadMemroyInputStream
…
```
```
org.apache.uniffle.common.merger.MergerTest ‑ testMergeSegmentToFile{String, File}[2]
org.apache.uniffle.common.merger.MergerTest ‑ testMergeSegmentToFile{String, File}[3]
org.apache.uniffle.common.merger.MergerTest ‑ testMergeSegmentToFile{String, File}[4]
org.apache.uniffle.common.netty.protocol.NettyProtocolTest ‑ testGetSortedShuffleDataRequest
org.apache.uniffle.common.netty.protocol.NettyProtocolTest ‑ testGetSortedShuffleDataResponse
org.apache.uniffle.common.records.RecordsReaderWriterTest ‑ testWriteAndReadRecordFileUseDirect{String, File}[1]
org.apache.uniffle.common.records.RecordsReaderWriterTest ‑ testWriteAndReadRecordFileUseDirect{String, File}[2]
org.apache.uniffle.common.records.RecordsReaderWriterTest ‑ testWriteAndReadRecordFile{String, File}[1]
org.apache.uniffle.common.records.RecordsReaderWriterTest ‑ testWriteAndReadRecordFile{String, File}[2]
org.apache.uniffle.common.records.RecordsReaderWriterTest ‑ testWriteAndReadRecordFile{String, File}[3]
…
```
</details>
[test-results]:data:application/gzip;base64,H4sIAGZiEmcC/02MzQ6DIBAGX8Vw7gFZd4G+TKMgCak/DcKp6btXqFKPM/vtvJnz07ixeyM0724N25KPV7Yp9NGvSzZcI+1qv8e8aLkSJz62ZExxUv3d07/yXxWu91MeVTGGsIbDhLSUKiDJA2sVELC6XxX4KS7VwteoWefZxx2YVIKGlnBAR1oKZZFIDd3YGpBW6o4LARzBss8XHxBY8xQBAAA=
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#2173] feat(remote merge): support netty for remote merge. [incubator-uniffle]
Posted by "zhengchenyu (via GitHub)" <gi...@apache.org>.
zhengchenyu commented on code in PR #2202:
URL: https://github.com/apache/incubator-uniffle/pull/2202#discussion_r1810128492
##########
common/src/main/java/org/apache/uniffle/common/serializer/writable/WritableSerializerInstance.java:
##########
@@ -47,19 +47,28 @@ public <T> T deserialize(DataInputBuffer buffer, Class vClass) throws IOExceptio
}
@Override
- public <K, V> SerializationStream serializeStream(OutputStream output, boolean raw) {
+ public <K, V> SerializationStream serializeStream(
+ SerOutputStream output, boolean raw, boolean shared) {
if (raw) {
- return new RawWritableSerializationStream(this, output);
+ if (shared) {
Review Comment:
We have three stream:
* WritableSerializationStream
* RawWritableDeserializationStream
* SharedRawWritableDeserializationStream
`WritableSerializationStream` is used to parse bytes into Java objects, mainly used on the reduce side of tez and spark.
`RawWritableSerializationStream` directly copies bytes without doing any actual deserialization. It is mainly used on the reduce side of mr, because mr requires raw interface.
`SharedRawWritableDeserializationStream` is similar to `RawWritableSerializationStream`, but uses some memory optimization methods. Mainly used for server-side merge. So deserialization is no needed.
The `Raw` prefix means that the bytes are copied directly without unnecessary serialization. I think it should not be deleted.
Now that, I think WritableSerializationStream and RawWritableDeserializationStream names are not changed, rename SharedRawWritableDeserializationStream to BufferRawDeserializationStream. How about this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#2173] feat(remote merge): support netty for remote merge. [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #2202:
URL: https://github.com/apache/incubator-uniffle/pull/2202#discussion_r1809768809
##########
common/src/main/java/org/apache/uniffle/common/serializer/writable/WritableSerializerInstance.java:
##########
@@ -47,19 +47,28 @@ public <T> T deserialize(DataInputBuffer buffer, Class vClass) throws IOExceptio
}
@Override
- public <K, V> SerializationStream serializeStream(OutputStream output, boolean raw) {
+ public <K, V> SerializationStream serializeStream(
+ SerOutputStream output, boolean raw, boolean shared) {
if (raw) {
- return new RawWritableSerializationStream(this, output);
+ if (shared) {
Review Comment:
Why do we use a shared serialization stream?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#2173] feat(remote merge): support netty for remote merge. [incubator-uniffle]
Posted by "zhengchenyu (via GitHub)" <gi...@apache.org>.
zhengchenyu closed pull request #2202: [#2173] feat(remote merge): support netty for remote merge.
URL: https://github.com/apache/incubator-uniffle/pull/2202
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#2173] feat(remote merge): support netty for remote merge. [incubator-uniffle]
Posted by "zhengchenyu (via GitHub)" <gi...@apache.org>.
zhengchenyu merged PR #2202:
URL: https://github.com/apache/incubator-uniffle/pull/2202
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#2173] feat(remote merge): support netty for remote merge. [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #2202:
URL: https://github.com/apache/incubator-uniffle/pull/2202#discussion_r1809814217
##########
common/src/main/java/org/apache/uniffle/common/serializer/writable/WritableSerializerInstance.java:
##########
@@ -47,19 +47,28 @@ public <T> T deserialize(DataInputBuffer buffer, Class vClass) throws IOExceptio
}
@Override
- public <K, V> SerializationStream serializeStream(OutputStream output, boolean raw) {
+ public <K, V> SerializationStream serializeStream(
+ SerOutputStream output, boolean raw, boolean shared) {
if (raw) {
- return new RawWritableSerializationStream(this, output);
+ if (shared) {
Review Comment:
Name makes me confused. Because `SharedSerializationStream` needs to be operated by multiple threads. It will need the lock.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#2173] feat(remote merge): support netty for remote merge. [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #2202:
URL: https://github.com/apache/incubator-uniffle/pull/2202#discussion_r1810000421
##########
common/src/main/java/org/apache/uniffle/common/serializer/writable/WritableSerializerInstance.java:
##########
@@ -47,19 +47,28 @@ public <T> T deserialize(DataInputBuffer buffer, Class vClass) throws IOExceptio
}
@Override
- public <K, V> SerializationStream serializeStream(OutputStream output, boolean raw) {
+ public <K, V> SerializationStream serializeStream(
+ SerOutputStream output, boolean raw, boolean shared) {
if (raw) {
- return new RawWritableSerializationStream(this, output);
+ if (shared) {
Review Comment:
Could we give a better name for it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#2173] feat(remote merge): support netty for remote merge. [incubator-uniffle]
Posted by "zhengchenyu (via GitHub)" <gi...@apache.org>.
zhengchenyu commented on PR #2202:
URL: https://github.com/apache/incubator-uniffle/pull/2202#issuecomment-2428087162
@jerqi Can you please review this PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#2173] feat(remote merge): support netty for remote merge. [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #2202:
URL: https://github.com/apache/incubator-uniffle/pull/2202#discussion_r1810531349
##########
common/src/main/java/org/apache/uniffle/common/serializer/writable/WritableSerializerInstance.java:
##########
@@ -47,19 +47,28 @@ public <T> T deserialize(DataInputBuffer buffer, Class vClass) throws IOExceptio
}
@Override
- public <K, V> SerializationStream serializeStream(OutputStream output, boolean raw) {
+ public <K, V> SerializationStream serializeStream(
+ SerOutputStream output, boolean raw, boolean shared) {
if (raw) {
- return new RawWritableSerializationStream(this, output);
+ if (shared) {
Review Comment:
Ok for. `Buffer` -> `Buffered`.
##########
common/src/main/java/org/apache/uniffle/common/serializer/writable/WritableSerializerInstance.java:
##########
@@ -47,19 +47,28 @@ public <T> T deserialize(DataInputBuffer buffer, Class vClass) throws IOExceptio
}
@Override
- public <K, V> SerializationStream serializeStream(OutputStream output, boolean raw) {
+ public <K, V> SerializationStream serializeStream(
+ SerOutputStream output, boolean raw, boolean shared) {
if (raw) {
- return new RawWritableSerializationStream(this, output);
+ if (shared) {
Review Comment:
Ok for me. `Buffer` -> `Buffered`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#2173] feat(remote merge): support netty for remote merge. [incubator-uniffle]
Posted by "zhengchenyu (via GitHub)" <gi...@apache.org>.
zhengchenyu commented on code in PR #2202:
URL: https://github.com/apache/incubator-uniffle/pull/2202#discussion_r1809821621
##########
common/src/main/java/org/apache/uniffle/common/serializer/writable/WritableSerializerInstance.java:
##########
@@ -47,19 +47,28 @@ public <T> T deserialize(DataInputBuffer buffer, Class vClass) throws IOExceptio
}
@Override
- public <K, V> SerializationStream serializeStream(OutputStream output, boolean raw) {
+ public <K, V> SerializationStream serializeStream(
+ SerOutputStream output, boolean raw, boolean shared) {
if (raw) {
- return new RawWritableSerializationStream(this, output);
+ if (shared) {
Review Comment:
A SharedSerializationStream corresponds to a block. SharedSerializationStream will not be accessed by multiple threads, can only be executed under the merge thread corresponding to one partition. Here, 'Shared' means use shared buffer to merge.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#2173] feat(remote merge): support netty for remote merge. [incubator-uniffle]
Posted by "zhengchenyu (via GitHub)" <gi...@apache.org>.
zhengchenyu commented on code in PR #2202:
URL: https://github.com/apache/incubator-uniffle/pull/2202#discussion_r1809790751
##########
common/src/main/java/org/apache/uniffle/common/serializer/writable/WritableSerializerInstance.java:
##########
@@ -47,19 +47,28 @@ public <T> T deserialize(DataInputBuffer buffer, Class vClass) throws IOExceptio
}
@Override
- public <K, V> SerializationStream serializeStream(OutputStream output, boolean raw) {
+ public <K, V> SerializationStream serializeStream(
+ SerOutputStream output, boolean raw, boolean shared) {
if (raw) {
- return new RawWritableSerializationStream(this, output);
+ if (shared) {
Review Comment:
In client side, the parsed record will be used by reduce, so we need a deep copy instance, every record have their own buffer. if we use shared buffer, error will occur.
But in server side, we write the record to the mergedblock immediately after parsing the record, so there is no need for each record to have a separate memory copy. For a segment/block, we can use only two shared buffer, this saves more memory.
BTW, although this PR is about Netty, a lot of work has actually been done on saving memory
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#2173] feat(remote merge): support netty for remote merge. [incubator-uniffle]
Posted by "zhengchenyu (via GitHub)" <gi...@apache.org>.
zhengchenyu commented on code in PR #2202:
URL: https://github.com/apache/incubator-uniffle/pull/2202#discussion_r1810037857
##########
common/src/main/java/org/apache/uniffle/common/serializer/writable/WritableSerializerInstance.java:
##########
@@ -47,19 +47,28 @@ public <T> T deserialize(DataInputBuffer buffer, Class vClass) throws IOExceptio
}
@Override
- public <K, V> SerializationStream serializeStream(OutputStream output, boolean raw) {
+ public <K, V> SerializationStream serializeStream(
+ SerOutputStream output, boolean raw, boolean shared) {
if (raw) {
- return new RawWritableSerializationStream(this, output);
+ if (shared) {
Review Comment:
> Could we give a better name for it?
In fact, I used to use `shallow` as name before. But I change to 'shared'.
Compare to `RawWritableDeserializationStream`, `SharedRawWritableDeserializationStream` use shared buffer to store record, but `RawWritableDeserializationStream` allocates a new buffer for each record.
If we need changed, `RawWritableDeserializationStream` rename to `DeepRawWritableDeserializationStream`, and `SharedRawWritableDeserializationStream` rename to `ShallowRawWritableDeserializationStream`. How about it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org