You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by zh...@apache.org on 2023/07/01 10:27:59 UTC
[incubator-celeborn] branch branch-0.3 updated: [CELEBORN-755][FOLLOWUP] Avoid unnecessary memory copy when compression disabled
This is an automated email from the ASF dual-hosted git repository.
zhouky pushed a commit to branch branch-0.3
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
The following commit(s) were added to refs/heads/branch-0.3 by this push:
new 893f8c5fe [CELEBORN-755][FOLLOWUP] Avoid unnecessary memory copy when compression disabled
893f8c5fe is described below
commit 893f8c5fe8ebbf48759e8563d652e843ff397aa1
Author: zky.zhoukeyong <zk...@alibaba-inc.com>
AuthorDate: Sat Jul 1 18:27:33 2023 +0800
[CELEBORN-755][FOLLOWUP] Avoid unnecessary memory copy when compression disabled
…sion disabled
### What changes were proposed in this pull request?
Avoid memory copy for code path where compression is disabled. Followup of https://github.com/apache/incubator-celeborn/pull/1669
### Why are the changes needed?
ditto
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Pass GA
Closes #1671 from waitinfuture/755.
Authored-by: zky.zhoukeyong <zk...@alibaba-inc.com>
Signed-off-by: zky.zhoukeyong <zk...@alibaba-inc.com>
(cherry picked from commit af0f5e57846ffab3ac052b9fc0710ff1c660a144)
Signed-off-by: zky.zhoukeyong <zk...@alibaba-inc.com>
---
.../org/apache/celeborn/client/ShuffleClientImpl.java | 16 ++++++----------
.../org/apache/celeborn/client/read/RssInputStream.java | 2 +-
2 files changed, 7 insertions(+), 11 deletions(-)
diff --git a/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java b/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
index 0e943dacd..a809bddae 100644
--- a/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
+++ b/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java
@@ -836,26 +836,22 @@ public class ShuffleClientImpl extends ShuffleClient {
// increment batchId
final int nextBatchId = pushState.nextBatchId();
- int totalSize = length;
- byte[] shuffleDataBuf = new byte[length];
-
if (shuffleCompressionEnabled) {
// compress data
final Compressor compressor = compressorThreadLocal.get();
compressor.compress(data, offset, length);
- totalSize = compressor.getCompressedTotalSize();
- shuffleDataBuf = compressor.getCompressedBuffer();
- } else {
- System.arraycopy(data, offset, shuffleDataBuf, 0, length);
+ data = compressor.getCompressedBuffer();
+ offset = 0;
+ length = compressor.getCompressedTotalSize();
}
- final byte[] body = new byte[BATCH_HEADER_SIZE + totalSize];
+ final byte[] body = new byte[BATCH_HEADER_SIZE + length];
Platform.putInt(body, Platform.BYTE_ARRAY_OFFSET, mapId);
Platform.putInt(body, Platform.BYTE_ARRAY_OFFSET + 4, attemptId);
Platform.putInt(body, Platform.BYTE_ARRAY_OFFSET + 8, nextBatchId);
- Platform.putInt(body, Platform.BYTE_ARRAY_OFFSET + 12, totalSize);
- System.arraycopy(shuffleDataBuf, 0, body, BATCH_HEADER_SIZE, totalSize);
+ Platform.putInt(body, Platform.BYTE_ARRAY_OFFSET + 12, length);
+ System.arraycopy(data, offset, body, BATCH_HEADER_SIZE, length);
if (doPush) {
// check limit
diff --git a/client/src/main/java/org/apache/celeborn/client/read/RssInputStream.java b/client/src/main/java/org/apache/celeborn/client/read/RssInputStream.java
index bcc639780..0d2aaa73c 100644
--- a/client/src/main/java/org/apache/celeborn/client/read/RssInputStream.java
+++ b/client/src/main/java/org/apache/celeborn/client/read/RssInputStream.java
@@ -166,7 +166,7 @@ public abstract class RssInputStream extends InputStream {
int blockSize = conf.clientPushBufferMaxSize();
if (shuffleCompressionEnabled) {
int headerLen = Decompressor.getCompressionHeaderLength(conf);
- blockSize = conf.clientPushBufferMaxSize() + headerLen;
+ blockSize += headerLen;
compressedBuf = new byte[blockSize];
decompressor = Decompressor.getDecompressor(conf);