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/11 15:09:05 UTC
[incubator-celeborn] branch branch-0.3 updated: [CELEBORN-783][FOLLOWUP] Private member updates and cleanup in `SortBasedPusher`
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 a711f4fc2 [CELEBORN-783][FOLLOWUP] Private member updates and cleanup in `SortBasedPusher`
a711f4fc2 is described below
commit a711f4fc23a91a6301bb33b18880f1aa5d37c551
Author: Fu Chen <cf...@gmail.com>
AuthorDate: Tue Jul 11 23:08:42 2023 +0800
[CELEBORN-783][FOLLOWUP] Private member updates and cleanup in `SortBasedPusher`
### What changes were proposed in this pull request?
As title
### Why are the changes needed?
https://github.com/apache/incubator-celeborn/pull/1699#discussion_r1259137323
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Pass GA
Closes #1704 from cfmcgrady/insert-record-followup.
Authored-by: Fu Chen <cf...@gmail.com>
Signed-off-by: zky.zhoukeyong <zk...@alibaba-inc.com>
(cherry picked from commit 90ba9f3e871e72d74eccb2c01ec076206ac5ce03)
Signed-off-by: zky.zhoukeyong <zk...@alibaba-inc.com>
---
.../spark/shuffle/celeborn/SortBasedPusher.java | 45 +++++++++-------------
.../shuffle/celeborn/SortBasedPusherSuiteJ.java | 1 -
.../shuffle/celeborn/SortBasedShuffleWriter.java | 3 --
.../shuffle/celeborn/SparkShuffleManager.java | 8 +---
.../celeborn/SortBasedShuffleWriterSuiteJ.java | 2 +-
.../shuffle/celeborn/SortBasedShuffleWriter.java | 4 --
.../shuffle/celeborn/SparkShuffleManager.java | 9 +----
7 files changed, 22 insertions(+), 50 deletions(-)
diff --git a/client-spark/common/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedPusher.java b/client-spark/common/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedPusher.java
index 330231387..5e5fc635d 100644
--- a/client-spark/common/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedPusher.java
+++ b/client-spark/common/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedPusher.java
@@ -44,6 +44,8 @@ public class SortBasedPusher extends MemoryConsumer {
private static final Logger logger = LoggerFactory.getLogger(SortBasedPusher.class);
+ private static final int UAO_SIZE = UnsafeAlignedOffset.getUaoSize();
+
/** Peak memory used by this sorter so far, in bytes. * */
private long peakMemoryUsedBytes;
@@ -56,29 +58,23 @@ public class SortBasedPusher extends MemoryConsumer {
private DataPusher dataPusher;
private final int pushBufferMaxSize;
private final long pushSortMemoryThreshold;
- final int uaoSize = UnsafeAlignedOffset.getUaoSize();
-
- String appId;
- int shuffleId;
- int mapId;
- int attemptNumber;
- long taskAttemptId;
- int numMappers;
- int numPartitions;
- CelebornConf conf;
- Consumer<Integer> afterPush;
- LongAdder[] mapStatusLengths;
+ private final int shuffleId;
+ private final int mapId;
+ private final int attemptNumber;
+ private final int numMappers;
+ private final int numPartitions;
+ private final Consumer<Integer> afterPush;
+ private final LongAdder[] mapStatusLengths;
// this lock is shared between different SortBasedPushers to synchronize pushData
- final Object sharedPushLock;
- volatile boolean asyncPushing = false;
- int[] shuffledPartitions = null;
- int[] inversedShuffledPartitions = null;
- ExecutorService executorService;
+ private final Object sharedPushLock;
+ private volatile boolean asyncPushing = false;
+ private int[] shuffledPartitions = null;
+ private int[] inversedShuffledPartitions = null;
+ private final ExecutorService executorService;
public SortBasedPusher(
TaskMemoryManager memoryManager,
ShuffleClient shuffleClient,
- String appId,
int shuffleId,
int mapId,
int attemptNumber,
@@ -98,11 +94,9 @@ public class SortBasedPusher extends MemoryConsumer {
this.shuffleClient = shuffleClient;
- this.appId = appId;
this.shuffleId = shuffleId;
this.mapId = mapId;
this.attemptNumber = attemptNumber;
- this.taskAttemptId = taskAttemptId;
this.numMappers = numMappers;
this.numPartitions = numPartitions;
@@ -112,7 +106,6 @@ public class SortBasedPusher extends MemoryConsumer {
JavaUtils.shuffleArray(shuffledPartitions, inversedShuffledPartitions);
}
- this.conf = conf;
this.afterPush = afterPush;
this.mapStatusLengths = mapStatusLengths;
@@ -193,7 +186,7 @@ public class SortBasedPusher extends MemoryConsumer {
offSet = 0;
}
- long recordReadPosition = recordOffsetInPage + uaoSize;
+ long recordReadPosition = recordOffsetInPage + UAO_SIZE;
Platform.copyMemory(
recordPage,
recordReadPosition,
@@ -223,9 +216,9 @@ public class SortBasedPusher extends MemoryConsumer {
int required;
// Need 4 or 8 bytes to store the record recordSize.
if (copySize) {
- required = recordSize + 4 + uaoSize;
+ required = recordSize + 4 + UAO_SIZE;
} else {
- required = recordSize + uaoSize;
+ required = recordSize + UAO_SIZE;
}
if (getUsed() > pushSortMemoryThreshold
@@ -245,14 +238,14 @@ public class SortBasedPusher extends MemoryConsumer {
final long recordAddress = taskMemoryManager.encodePageNumberAndOffset(currentPage, pageCursor);
if (copySize) {
UnsafeAlignedOffset.putSize(base, pageCursor, recordSize + 4);
- pageCursor += uaoSize;
+ pageCursor += UAO_SIZE;
Platform.putInt(base, pageCursor, Integer.reverseBytes(recordSize));
pageCursor += 4;
Platform.copyMemory(recordBase, recordOffset, base, pageCursor, recordSize);
pageCursor += recordSize;
} else {
UnsafeAlignedOffset.putSize(base, pageCursor, recordSize);
- pageCursor += uaoSize;
+ pageCursor += UAO_SIZE;
Platform.copyMemory(recordBase, recordOffset, base, pageCursor, recordSize);
pageCursor += recordSize;
}
diff --git a/client-spark/common/src/test/java/org/apache/spark/shuffle/celeborn/SortBasedPusherSuiteJ.java b/client-spark/common/src/test/java/org/apache/spark/shuffle/celeborn/SortBasedPusherSuiteJ.java
index 203ca46fa..3ab993937 100644
--- a/client-spark/common/src/test/java/org/apache/spark/shuffle/celeborn/SortBasedPusherSuiteJ.java
+++ b/client-spark/common/src/test/java/org/apache/spark/shuffle/celeborn/SortBasedPusherSuiteJ.java
@@ -82,7 +82,6 @@ public class SortBasedPusherSuiteJ {
new SortBasedPusher(
taskMemoryManager,
/*shuffleClient=*/ client,
- /*appId=*/ null,
/*shuffleId=*/ 0,
/*mapId=*/ 0,
/*attemptNumber=*/ 0,
diff --git a/client-spark/spark-2/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedShuffleWriter.java b/client-spark/spark-2/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedShuffleWriter.java
index 8a135ada3..1ea01ff53 100644
--- a/client-spark/spark-2/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedShuffleWriter.java
+++ b/client-spark/spark-2/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedShuffleWriter.java
@@ -95,7 +95,6 @@ public class SortBasedShuffleWriter<K, V, C> extends ShuffleWriter<K, V> {
// parameters. By the way, simplify the passed parameters.
public SortBasedShuffleWriter(
ShuffleDependency<K, V, C> dep,
- String appId,
int numMappers,
TaskContext taskContext,
CelebornConf conf,
@@ -133,7 +132,6 @@ public class SortBasedShuffleWriter<K, V, C> extends ShuffleWriter<K, V> {
new SortBasedPusher(
taskContext.taskMemoryManager(),
shuffleClient,
- appId,
shuffleId,
mapId,
taskContext.attemptNumber(),
@@ -153,7 +151,6 @@ public class SortBasedShuffleWriter<K, V, C> extends ShuffleWriter<K, V> {
new SortBasedPusher(
taskContext.taskMemoryManager(),
shuffleClient,
- appId,
shuffleId,
mapId,
taskContext.attemptNumber(),
diff --git a/client-spark/spark-2/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java b/client-spark/spark-2/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java
index 90644899e..d344deec0 100644
--- a/client-spark/spark-2/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java
+++ b/client-spark/spark-2/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java
@@ -191,13 +191,7 @@ public class SparkShuffleManager implements ShuffleManager {
ExecutorService pushThread =
celebornConf.clientPushSortPipelineEnabled() ? getPusherThread() : null;
return new SortBasedShuffleWriter<>(
- h.dependency(),
- h.appUniqueId(),
- h.numMaps(),
- context,
- celebornConf,
- client,
- pushThread);
+ h.dependency(), h.numMaps(), context, celebornConf, client, pushThread);
} else if (ShuffleMode.HASH.equals(celebornConf.shuffleWriterMode())) {
return new HashBasedShuffleWriter<>(
h, mapId, context, celebornConf, client, SendBufferPool.get(cores));
diff --git a/client-spark/spark-2/src/test/java/org/apache/spark/shuffle/celeborn/SortBasedShuffleWriterSuiteJ.java b/client-spark/spark-2/src/test/java/org/apache/spark/shuffle/celeborn/SortBasedShuffleWriterSuiteJ.java
index 85d7c8fdd..2fbd5dcf4 100644
--- a/client-spark/spark-2/src/test/java/org/apache/spark/shuffle/celeborn/SortBasedShuffleWriterSuiteJ.java
+++ b/client-spark/spark-2/src/test/java/org/apache/spark/shuffle/celeborn/SortBasedShuffleWriterSuiteJ.java
@@ -34,6 +34,6 @@ public class SortBasedShuffleWriterSuiteJ extends CelebornShuffleWriterSuiteBase
CelebornShuffleHandle handle, TaskContext context, CelebornConf conf, ShuffleClient client)
throws IOException {
return new SortBasedShuffleWriter<Integer, String, String>(
- handle.dependency(), handle.appUniqueId(), numPartitions, context, conf, client, null);
+ handle.dependency(), numPartitions, context, conf, client, null);
}
}
diff --git a/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedShuffleWriter.java b/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedShuffleWriter.java
index be0f6bba3..017b006d6 100644
--- a/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedShuffleWriter.java
+++ b/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SortBasedShuffleWriter.java
@@ -95,7 +95,6 @@ public class SortBasedShuffleWriter<K, V, C> extends ShuffleWriter<K, V> {
// parameters. By the way, simplify the passed parameters.
public SortBasedShuffleWriter(
ShuffleDependency<K, V, C> dep,
- String appId,
int numMappers,
TaskContext taskContext,
CelebornConf conf,
@@ -133,7 +132,6 @@ public class SortBasedShuffleWriter<K, V, C> extends ShuffleWriter<K, V> {
new SortBasedPusher(
taskContext.taskMemoryManager(),
shuffleClient,
- appId,
shuffleId,
mapId,
taskContext.attemptNumber(),
@@ -153,7 +151,6 @@ public class SortBasedShuffleWriter<K, V, C> extends ShuffleWriter<K, V> {
new SortBasedPusher(
taskContext.taskMemoryManager(),
shuffleClient,
- appId,
shuffleId,
mapId,
taskContext.attemptNumber(),
@@ -179,7 +176,6 @@ public class SortBasedShuffleWriter<K, V, C> extends ShuffleWriter<K, V> {
throws IOException {
this(
handle.dependency(),
- handle.appUniqueId(),
handle.numMappers(),
taskContext,
conf,
diff --git a/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java b/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java
index f121fea25..3b6c2ea65 100644
--- a/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java
+++ b/client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java
@@ -193,14 +193,7 @@ public class SparkShuffleManager implements ShuffleManager {
ExecutorService pushThread =
celebornConf.clientPushSortPipelineEnabled() ? getPusherThread() : null;
return new SortBasedShuffleWriter<>(
- h.dependency(),
- h.appUniqueId(),
- h.numMappers(),
- context,
- celebornConf,
- client,
- metrics,
- pushThread);
+ h.dependency(), h.numMappers(), context, celebornConf, client, metrics, pushThread);
} else if (ShuffleMode.HASH.equals(celebornConf.shuffleWriterMode())) {
return new HashBasedShuffleWriter<>(
h, context, celebornConf, client, metrics, SendBufferPool.get(cores));