You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@celeborn.apache.org by "SteNicholas (via GitHub)" <gi...@apache.org> on 2025/07/24 12:06:45 UTC
[PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
SteNicholas opened a new pull request, #3387:
URL: https://github.com/apache/celeborn/pull/3387
### What changes were proposed in this pull request?
Parallelize the create partition writer in `handleReserveSlots` to speed up the reserveSlots RPC process time。
### Why are the changes needed?
The creation of partition writer in `handleReserveSlots` could use parallelize way to speed up the reserveSlots RPC process time.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
CI.
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "eolivelli (via GitHub)" <gi...@apache.org>.
eolivelli commented on code in PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#discussion_r2812408466
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Controller.scala:
##########
@@ -291,15 +251,118 @@ private[deploy] class Controller(
Utils.getSlotsPerDisk(requestPrimaryLocs, requestReplicaLocs))
workerSource.incCounter(WorkerSource.SLOTS_ALLOCATED, primaryLocs.size() + replicaLocs.size())
- logInfo(s"Reserved ${primaryLocs.size()} primary location " +
- s"${primaryLocs.asScala.map(_.getUniqueId).mkString(",")} and ${replicaLocs.size()} replica location " +
- s"${replicaLocs.asScala.map(_.getUniqueId).mkString(",")} for $shuffleKey ")
+ logInfo(s"Reserved ${primaryLocs.size()} primary location" +
+ s" and ${replicaLocs.size()} replica location for $shuffleKey ")
if (log.isDebugEnabled()) {
logDebug(s"primary: $primaryLocs\nreplica: $replicaLocs.")
}
context.reply(ReserveSlotsResponse(StatusCode.SUCCESS))
}
+ private def createWriters(
+ shuffleKey: String,
+ applicationId: String,
+ shuffleId: Int,
+ requestLocs: jList[PartitionLocation],
+ splitThreshold: Long,
+ splitMode: PartitionSplitMode,
+ partitionType: PartitionType,
+ rangeReadFilter: Boolean,
+ userIdentifier: UserIdentifier,
+ partitionSplitEnabled: Boolean,
+ isSegmentGranularityVisible: Boolean,
+ isPrimary: Boolean): jList[PartitionLocation] = {
+ val partitionLocations = new jArrayList[PartitionLocation]()
+ try {
+ def createWriter(partitionLocation: PartitionLocation): PartitionLocation = {
+ createPartitionDataWriter(
+ shuffleKey,
+ applicationId,
+ shuffleId,
+ partitionLocation,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible,
+ isPrimary)
+ }
+ if (createWriterThreadPool == null) {
+ partitionLocations.addAll(requestLocs.asScala.map(createWriter).asJava)
+ } else {
+ partitionLocations.addAll(Utils.tryFuturesWithTimeout(
+ requestLocs.asScala.map(requestLoc =>
+ Utils.future(createWriter(requestLoc))(createWriterThreadPool)).toList,
+ createWriterParallelTimeout,
+ s"Create FileWriter for $shuffleKey timeout.").asJava)
+ }
+ } catch {
+ case e: Exception =>
+ logError(s"Create FileWriter for $shuffleKey failed.", e)
+ }
+ partitionLocations
+ }
+
+ private def createPartitionDataWriter(
+ shuffleKey: String,
+ applicationId: String,
+ shuffleId: Int,
+ requestLoc: PartitionLocation,
+ splitThreshold: Long,
+ splitMode: PartitionSplitMode,
+ partitionType: PartitionType,
+ rangeReadFilter: Boolean,
+ userIdentifier: UserIdentifier,
+ partitionSplitEnabled: Boolean,
+ isSegmentGranularityVisible: Boolean,
+ isPrimary: Boolean): PartitionLocation = {
+ try {
+ var location =
+ if (isPrimary) {
+ partitionLocationInfo.getPrimaryLocation(
+ shuffleKey,
+ requestLoc.getUniqueId)
+ } else {
+ partitionLocationInfo.getReplicaLocation(
+ shuffleKey,
+ requestLoc.getUniqueId)
+ }
+ if (location == null) {
+ location = requestLoc
+ val writer = storageManager.createPartitionDataWriter(
+ applicationId,
+ shuffleId,
+ location,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible)
+ new WorkingPartition(location, writer)
+ } else {
+ location
+ }
+ } catch {
+ case e: Exception =>
+ logError(s"Create FileWriter for $requestLoc $shuffleKey failed.", e)
+ throw e
+ }
+ }
+
+ private def destroyWriters(
+ partitionLocations: jList[PartitionLocation],
+ shuffleKey: String): Unit = {
+ partitionLocations.asScala.foreach { partitionLocation =>
+ val fileWriter = partitionLocation.asInstanceOf[WorkingPartition].getFileWriter
Review Comment:
makes sense
what about the other comment about destroying writers not created by this loop ?
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#discussion_r2230491592
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/WriterHelper.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.worker;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.celeborn.common.identity.UserIdentifier;
+import org.apache.celeborn.common.meta.WorkerPartitionLocationInfo;
+import org.apache.celeborn.common.protocol.PartitionLocation;
+import org.apache.celeborn.common.protocol.PartitionSplitMode;
+import org.apache.celeborn.common.protocol.PartitionType;
+import org.apache.celeborn.service.deploy.worker.storage.PartitionDataWriter;
+import org.apache.celeborn.service.deploy.worker.storage.StorageManager;
+
+public class WriterHelper {
+
+ private static final Logger LOG = LoggerFactory.getLogger(WriterHelper.class);
+
+ private final StorageManager storageManager;
+ private final WorkerPartitionLocationInfo partitionLocationInfo;
+
+ public WriterHelper(
+ StorageManager storageManager, WorkerPartitionLocationInfo partitionLocationInfo) {
+ this.storageManager = storageManager;
+ this.partitionLocationInfo = partitionLocationInfo;
+ }
+
+ public List<PartitionLocation> createWriters(
+ String shuffleKey,
+ String applicationId,
+ int shuffleId,
+ List<PartitionLocation> requestLocs,
+ long splitThreshold,
+ PartitionSplitMode splitMode,
+ PartitionType partitionType,
+ boolean rangeReadFilter,
+ UserIdentifier userIdentifier,
+ boolean partitionSplitEnabled,
+ boolean isSegmentGranularityVisible,
+ boolean isPrimary) {
+ List<PartitionLocation> locs = new ArrayList<>();
+ try {
+ ConcurrentLinkedQueue<PartitionLocation> locQueue = new ConcurrentLinkedQueue<>();
+ requestLocs
+ .parallelStream()
+ .forEach(
+ requestLoc -> {
+ PartitionLocation location =
+ isPrimary
+ ? partitionLocationInfo.getPrimaryLocation(
+ shuffleKey, requestLoc.getUniqueId())
+ : partitionLocationInfo.getReplicaLocation(
+ shuffleKey, requestLoc.getUniqueId());
+ if (location == null) {
+ location = requestLoc;
+ PartitionDataWriter writer;
+ try {
+ writer =
+ storageManager.createPartitionDataWriter(
+ applicationId,
+ shuffleId,
+ location,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
Review Comment:
Wrapping IOException in RuntimeException loses the specific exception type and may make error handling less precise. Consider creating a more specific exception or allowing the IOException to propagate appropriately.
```suggestion
throw new PartitionDataWriterException("Failed to create PartitionDataWriter", e);
```
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/WriterHelper.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.worker;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.celeborn.common.identity.UserIdentifier;
+import org.apache.celeborn.common.meta.WorkerPartitionLocationInfo;
+import org.apache.celeborn.common.protocol.PartitionLocation;
+import org.apache.celeborn.common.protocol.PartitionSplitMode;
+import org.apache.celeborn.common.protocol.PartitionType;
+import org.apache.celeborn.service.deploy.worker.storage.PartitionDataWriter;
+import org.apache.celeborn.service.deploy.worker.storage.StorageManager;
+
+public class WriterHelper {
+
+ private static final Logger LOG = LoggerFactory.getLogger(WriterHelper.class);
+
+ private final StorageManager storageManager;
+ private final WorkerPartitionLocationInfo partitionLocationInfo;
+
+ public WriterHelper(
+ StorageManager storageManager, WorkerPartitionLocationInfo partitionLocationInfo) {
+ this.storageManager = storageManager;
+ this.partitionLocationInfo = partitionLocationInfo;
+ }
+
+ public List<PartitionLocation> createWriters(
+ String shuffleKey,
+ String applicationId,
+ int shuffleId,
+ List<PartitionLocation> requestLocs,
+ long splitThreshold,
+ PartitionSplitMode splitMode,
+ PartitionType partitionType,
+ boolean rangeReadFilter,
+ UserIdentifier userIdentifier,
+ boolean partitionSplitEnabled,
+ boolean isSegmentGranularityVisible,
+ boolean isPrimary) {
+ List<PartitionLocation> locs = new ArrayList<>();
+ try {
+ ConcurrentLinkedQueue<PartitionLocation> locQueue = new ConcurrentLinkedQueue<>();
+ requestLocs
+ .parallelStream()
+ .forEach(
+ requestLoc -> {
+ PartitionLocation location =
+ isPrimary
+ ? partitionLocationInfo.getPrimaryLocation(
+ shuffleKey, requestLoc.getUniqueId())
+ : partitionLocationInfo.getReplicaLocation(
+ shuffleKey, requestLoc.getUniqueId());
+ if (location == null) {
+ location = requestLoc;
+ PartitionDataWriter writer;
+ try {
+ writer =
+ storageManager.createPartitionDataWriter(
+ applicationId,
+ shuffleId,
+ location,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ locQueue.add(new WorkingPartition(location, writer));
+ } else {
+ locQueue.add(location);
+ }
+ });
+ locs.addAll(locQueue);
+ } catch (Exception e) {
+ LOG.error("CreateWriter for {} failed.", shuffleKey, e);
+ }
+ return locs;
+ }
+
+ public void destroyWriters(List<PartitionLocation> locs, String shuffleKey) {
+ locs.parallelStream()
+ .forEach(
+ partitionLocation -> {
+ WorkingPartition workingPartition = (WorkingPartition) partitionLocation;
+ PartitionDataWriter fileWriter = workingPartition.getFileWriter();
+ fileWriter.destroy(
+ new IOException(
+ String.format(
+ "Destroy FileWriter %s caused by " + "reserving slots failed for %s.",
+ fileWriter, shuffleKey)));
Review Comment:
This cast is unsafe and will throw a ClassCastException if the PartitionLocation is not a WorkingPartition. Add type checking before casting or filter to only process WorkingPartition instances.
```suggestion
if (partitionLocation instanceof WorkingPartition) {
WorkingPartition workingPartition = (WorkingPartition) partitionLocation;
PartitionDataWriter fileWriter = workingPartition.getFileWriter();
fileWriter.destroy(
new IOException(
String.format(
"Destroy FileWriter %s caused by " + "reserving slots failed for %s.",
fileWriter, shuffleKey)));
} else {
LOG.warn("Skipping destruction of non-WorkingPartition instance: {}", partitionLocation);
}
```
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/WriterHelper.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.worker;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.celeborn.common.identity.UserIdentifier;
+import org.apache.celeborn.common.meta.WorkerPartitionLocationInfo;
+import org.apache.celeborn.common.protocol.PartitionLocation;
+import org.apache.celeborn.common.protocol.PartitionSplitMode;
+import org.apache.celeborn.common.protocol.PartitionType;
+import org.apache.celeborn.service.deploy.worker.storage.PartitionDataWriter;
+import org.apache.celeborn.service.deploy.worker.storage.StorageManager;
+
+public class WriterHelper {
+
+ private static final Logger LOG = LoggerFactory.getLogger(WriterHelper.class);
+
+ private final StorageManager storageManager;
+ private final WorkerPartitionLocationInfo partitionLocationInfo;
+
+ public WriterHelper(
+ StorageManager storageManager, WorkerPartitionLocationInfo partitionLocationInfo) {
+ this.storageManager = storageManager;
+ this.partitionLocationInfo = partitionLocationInfo;
+ }
+
+ public List<PartitionLocation> createWriters(
+ String shuffleKey,
+ String applicationId,
+ int shuffleId,
+ List<PartitionLocation> requestLocs,
+ long splitThreshold,
+ PartitionSplitMode splitMode,
+ PartitionType partitionType,
+ boolean rangeReadFilter,
+ UserIdentifier userIdentifier,
+ boolean partitionSplitEnabled,
+ boolean isSegmentGranularityVisible,
+ boolean isPrimary) {
+ List<PartitionLocation> locs = new ArrayList<>();
+ try {
+ ConcurrentLinkedQueue<PartitionLocation> locQueue = new ConcurrentLinkedQueue<>();
+ requestLocs
+ .parallelStream()
+ .forEach(
+ requestLoc -> {
+ PartitionLocation location =
+ isPrimary
+ ? partitionLocationInfo.getPrimaryLocation(
+ shuffleKey, requestLoc.getUniqueId())
+ : partitionLocationInfo.getReplicaLocation(
+ shuffleKey, requestLoc.getUniqueId());
+ if (location == null) {
+ location = requestLoc;
+ PartitionDataWriter writer;
+ try {
+ writer =
+ storageManager.createPartitionDataWriter(
+ applicationId,
+ shuffleId,
+ location,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ locQueue.add(new WorkingPartition(location, writer));
+ } else {
+ locQueue.add(location);
+ }
+ });
Review Comment:
[nitpick] Using parallelStream() with concurrent operations on a ConcurrentLinkedQueue may not provide expected performance benefits due to coordination overhead. Consider using parallel processing with a thread-safe collection strategy or evaluate if the performance gain justifies the complexity.
```suggestion
ForkJoinPool customThreadPool = new ForkJoinPool();
try {
customThreadPool.submit(() ->
requestLocs.parallelStream().forEach(
requestLoc -> {
PartitionLocation location =
isPrimary
? partitionLocationInfo.getPrimaryLocation(
shuffleKey, requestLoc.getUniqueId())
: partitionLocationInfo.getReplicaLocation(
shuffleKey, requestLoc.getUniqueId());
if (location == null) {
location = requestLoc;
PartitionDataWriter writer;
try {
writer =
storageManager.createPartitionDataWriter(
applicationId,
shuffleId,
location,
splitThreshold,
splitMode,
partitionType,
rangeReadFilter,
userIdentifier,
partitionSplitEnabled,
isSegmentGranularityVisible);
} catch (IOException e) {
throw new RuntimeException(e);
}
locQueue.add(new WorkingPartition(location, writer));
} else {
locQueue.add(location);
}
})).get();
} catch (Exception e) {
throw new RuntimeException("Error during parallel processing", e);
} finally {
customThreadPool.shutdown();
}
```
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#discussion_r2235061041
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/WriterHelper.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.worker;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.celeborn.common.identity.UserIdentifier;
+import org.apache.celeborn.common.meta.WorkerPartitionLocationInfo;
+import org.apache.celeborn.common.protocol.PartitionLocation;
+import org.apache.celeborn.common.protocol.PartitionSplitMode;
+import org.apache.celeborn.common.protocol.PartitionType;
+import org.apache.celeborn.service.deploy.worker.storage.PartitionDataWriter;
+import org.apache.celeborn.service.deploy.worker.storage.StorageManager;
+
+public class WriterHelper {
+
+ private static final Logger LOG = LoggerFactory.getLogger(WriterHelper.class);
+
+ private final StorageManager storageManager;
+ private final WorkerPartitionLocationInfo partitionLocationInfo;
+
+ public WriterHelper(
+ StorageManager storageManager, WorkerPartitionLocationInfo partitionLocationInfo) {
+ this.storageManager = storageManager;
+ this.partitionLocationInfo = partitionLocationInfo;
+ }
+
+ public List<PartitionLocation> createWriters(
+ String shuffleKey,
+ String applicationId,
+ int shuffleId,
+ List<PartitionLocation> requestLocs,
+ long splitThreshold,
+ PartitionSplitMode splitMode,
+ PartitionType partitionType,
+ boolean rangeReadFilter,
+ UserIdentifier userIdentifier,
+ boolean partitionSplitEnabled,
+ boolean isSegmentGranularityVisible,
+ boolean isPrimary) {
+ List<PartitionLocation> locs = new ArrayList<>();
+ try {
+ ConcurrentLinkedQueue<PartitionLocation> locQueue = new ConcurrentLinkedQueue<>();
+ requestLocs
+ .parallelStream()
+ .forEach(
+ requestLoc -> {
+ PartitionLocation location =
+ isPrimary
+ ? partitionLocationInfo.getPrimaryLocation(
+ shuffleKey, requestLoc.getUniqueId())
+ : partitionLocationInfo.getReplicaLocation(
+ shuffleKey, requestLoc.getUniqueId());
+ if (location == null) {
+ location = requestLoc;
+ PartitionDataWriter writer;
+ try {
+ writer =
+ storageManager.createPartitionDataWriter(
+ applicationId,
+ shuffleId,
+ location,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ locQueue.add(new WorkingPartition(location, writer));
+ } else {
+ locQueue.add(location);
+ }
+ });
+ locs.addAll(locQueue);
+ } catch (Exception e) {
+ LOG.error("CreateWriter for {} failed.", shuffleKey, e);
+ }
+ return locs;
+ }
+
+ public void destroyWriters(List<PartitionLocation> locs, String shuffleKey) {
+ locs.parallelStream()
+ .forEach(
+ partitionLocation -> {
+ WorkingPartition workingPartition = (WorkingPartition) partitionLocation;
+ PartitionDataWriter fileWriter = workingPartition.getFileWriter();
+ fileWriter.destroy(
+ new IOException(
+ String.format(
+ "Destroy FileWriter %s caused by " + "reserving slots failed for %s.",
+ fileWriter, shuffleKey)));
+ });
+ }
+}
Review Comment:
@RexXiong, WDYT?
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "eolivelli (via GitHub)" <gi...@apache.org>.
eolivelli commented on code in PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#discussion_r2811403810
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Controller.scala:
##########
@@ -291,15 +251,118 @@ private[deploy] class Controller(
Utils.getSlotsPerDisk(requestPrimaryLocs, requestReplicaLocs))
workerSource.incCounter(WorkerSource.SLOTS_ALLOCATED, primaryLocs.size() + replicaLocs.size())
- logInfo(s"Reserved ${primaryLocs.size()} primary location " +
- s"${primaryLocs.asScala.map(_.getUniqueId).mkString(",")} and ${replicaLocs.size()} replica location " +
- s"${replicaLocs.asScala.map(_.getUniqueId).mkString(",")} for $shuffleKey ")
+ logInfo(s"Reserved ${primaryLocs.size()} primary location" +
+ s" and ${replicaLocs.size()} replica location for $shuffleKey ")
if (log.isDebugEnabled()) {
logDebug(s"primary: $primaryLocs\nreplica: $replicaLocs.")
}
context.reply(ReserveSlotsResponse(StatusCode.SUCCESS))
}
+ private def createWriters(
+ shuffleKey: String,
+ applicationId: String,
+ shuffleId: Int,
+ requestLocs: jList[PartitionLocation],
+ splitThreshold: Long,
+ splitMode: PartitionSplitMode,
+ partitionType: PartitionType,
+ rangeReadFilter: Boolean,
+ userIdentifier: UserIdentifier,
+ partitionSplitEnabled: Boolean,
+ isSegmentGranularityVisible: Boolean,
+ isPrimary: Boolean): jList[PartitionLocation] = {
+ val partitionLocations = new jArrayList[PartitionLocation]()
+ try {
+ def createWriter(partitionLocation: PartitionLocation): PartitionLocation = {
+ createPartitionDataWriter(
+ shuffleKey,
+ applicationId,
+ shuffleId,
+ partitionLocation,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible,
+ isPrimary)
+ }
+ if (createWriterThreadPool == null) {
+ partitionLocations.addAll(requestLocs.asScala.map(createWriter).asJava)
+ } else {
+ partitionLocations.addAll(Utils.tryFuturesWithTimeout(
+ requestLocs.asScala.map(requestLoc =>
+ Utils.future(createWriter(requestLoc))(createWriterThreadPool)).toList,
+ createWriterParallelTimeout,
+ s"Create FileWriter for $shuffleKey timeout.").asJava)
+ }
+ } catch {
+ case e: Exception =>
+ logError(s"Create FileWriter for $shuffleKey failed.", e)
+ }
+ partitionLocations
+ }
+
+ private def createPartitionDataWriter(
+ shuffleKey: String,
+ applicationId: String,
+ shuffleId: Int,
+ requestLoc: PartitionLocation,
+ splitThreshold: Long,
+ splitMode: PartitionSplitMode,
+ partitionType: PartitionType,
+ rangeReadFilter: Boolean,
+ userIdentifier: UserIdentifier,
+ partitionSplitEnabled: Boolean,
+ isSegmentGranularityVisible: Boolean,
+ isPrimary: Boolean): PartitionLocation = {
+ try {
+ var location =
+ if (isPrimary) {
+ partitionLocationInfo.getPrimaryLocation(
+ shuffleKey,
+ requestLoc.getUniqueId)
+ } else {
+ partitionLocationInfo.getReplicaLocation(
+ shuffleKey,
+ requestLoc.getUniqueId)
+ }
+ if (location == null) {
+ location = requestLoc
+ val writer = storageManager.createPartitionDataWriter(
+ applicationId,
+ shuffleId,
+ location,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible)
+ new WorkingPartition(location, writer)
+ } else {
+ location
+ }
+ } catch {
+ case e: Exception =>
+ logError(s"Create FileWriter for $requestLoc $shuffleKey failed.", e)
+ throw e
+ }
+ }
+
+ private def destroyWriters(
+ partitionLocations: jList[PartitionLocation],
+ shuffleKey: String): Unit = {
+ partitionLocations.asScala.foreach { partitionLocation =>
+ val fileWriter = partitionLocation.asInstanceOf[WorkingPartition].getFileWriter
Review Comment:
is it possible that here the value is not an WorkingPartition ?
I am asking because in the code above it is possible that there is already a PartitionLocation, that could have been put there from other code paths
https://github.com/apache/celeborn/pull/3387/changes#diff-ddad2bbb9547ade8e07445e08a7f97dbc717b594d43e811436ea42a76f0ce0fcR347
we should destroy only the writers that have been created by this loop
in my patch I kept a list of Writers created by the loop and destroyed only them in case of failure
https://github.com/apache/celeborn/pull/3598
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#discussion_r2261800104
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Controller.scala:
##########
@@ -299,6 +258,109 @@ private[deploy] class Controller(
context.reply(ReserveSlotsResponse(StatusCode.SUCCESS))
}
+ private def createWriters(
+ shuffleKey: String,
+ applicationId: String,
+ shuffleId: Int,
+ requestLocs: jList[PartitionLocation],
+ splitThreshold: Long,
+ splitMode: PartitionSplitMode,
+ partitionType: PartitionType,
+ rangeReadFilter: Boolean,
+ userIdentifier: UserIdentifier,
+ partitionSplitEnabled: Boolean,
+ isSegmentGranularityVisible: Boolean,
+ isPrimary: Boolean): jList[PartitionLocation] = {
+ val partitionLocations = new jArrayList[PartitionLocation]()
+ try {
+ def createWriter(partitionLocation: PartitionLocation): PartitionLocation = {
+ createPartitionDataWriter(
+ shuffleKey,
+ applicationId,
+ shuffleId,
+ partitionLocation,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible,
+ isPrimary)
+ }
+ if (createWriterThreadPool == null) {
+ partitionLocations.addAll(requestLocs.asScala.map(createWriter).asJava)
+ } else {
+ partitionLocations.addAll(requestLocs.asScala.map(requestLoc =>
+ createWriterThreadPool.submit(new Callable[PartitionLocation] {
+ override def call(): PartitionLocation = createWriter(requestLoc)
+ })).map(_.get()).asJava)
+ }
+ } catch {
+ case e: Exception =>
+ logError(s"Create FileWriter for $shuffleKey failed.", e)
+ }
+ partitionLocations
+ }
+
+ private def createPartitionDataWriter(
+ shuffleKey: String,
+ applicationId: String,
+ shuffleId: Int,
+ requestLoc: PartitionLocation,
+ splitThreshold: Long,
+ splitMode: PartitionSplitMode,
+ partitionType: PartitionType,
+ rangeReadFilter: Boolean,
+ userIdentifier: UserIdentifier,
+ partitionSplitEnabled: Boolean,
+ isSegmentGranularityVisible: Boolean,
+ isPrimary: Boolean): PartitionLocation = {
+ try {
+ var location =
+ if (isPrimary) {
+ partitionLocationInfo.getPrimaryLocation(
+ shuffleKey,
+ requestLoc.getUniqueId)
+ } else {
+ partitionLocationInfo.getReplicaLocation(
+ shuffleKey,
+ requestLoc.getUniqueId)
+ }
+ if (location == null) {
+ location = requestLoc
+ val writer = storageManager.createPartitionDataWriter(
+ applicationId,
+ shuffleId,
+ location,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible)
+ new WorkingPartition(location, writer)
+ } else {
+ location
+ }
+ } catch {
+ case e: Exception =>
+ logError(s"Create FileWriter for $requestLoc $shuffleKey failed.", e)
+ throw e
Review Comment:
[nitpick] Re-throwing the exception after logging it provides no additional value since the exception will be propagated anyway. Consider removing the try-catch block or handling the exception more meaningfully.
```suggestion
var location =
if (isPrimary) {
partitionLocationInfo.getPrimaryLocation(
shuffleKey,
requestLoc.getUniqueId)
} else {
partitionLocationInfo.getReplicaLocation(
shuffleKey,
requestLoc.getUniqueId)
}
if (location == null) {
location = requestLoc
val writer = storageManager.createPartitionDataWriter(
applicationId,
shuffleId,
location,
splitThreshold,
splitMode,
partitionType,
rangeReadFilter,
userIdentifier,
partitionSplitEnabled,
isSegmentGranularityVisible)
new WorkingPartition(location, writer)
} else {
location
```
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Controller.scala:
##########
@@ -32,7 +32,7 @@ import org.roaringbitmap.RoaringBitmap
import org.apache.celeborn.common.CelebornConf
import org.apache.celeborn.common.identity.UserIdentifier
import org.apache.celeborn.common.internal.Logging
-import org.apache.celeborn.common.meta.{ReduceFileMeta, WorkerInfo, WorkerPartitionLocationInfo}
+import org.apache.celeborn.common.meta.{WorkerInfo, WorkerPartitionLocationInfo}
Review Comment:
The removal of `ReduceFileMeta` from the import statement suggests it may no longer be used in this file. However, without seeing the full file context, this could indicate dead code removal or the import might still be needed elsewhere in the file.
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Controller.scala:
##########
@@ -299,6 +258,109 @@ private[deploy] class Controller(
context.reply(ReserveSlotsResponse(StatusCode.SUCCESS))
}
+ private def createWriters(
+ shuffleKey: String,
+ applicationId: String,
+ shuffleId: Int,
+ requestLocs: jList[PartitionLocation],
+ splitThreshold: Long,
+ splitMode: PartitionSplitMode,
+ partitionType: PartitionType,
+ rangeReadFilter: Boolean,
+ userIdentifier: UserIdentifier,
+ partitionSplitEnabled: Boolean,
+ isSegmentGranularityVisible: Boolean,
+ isPrimary: Boolean): jList[PartitionLocation] = {
+ val partitionLocations = new jArrayList[PartitionLocation]()
+ try {
+ def createWriter(partitionLocation: PartitionLocation): PartitionLocation = {
+ createPartitionDataWriter(
+ shuffleKey,
+ applicationId,
+ shuffleId,
+ partitionLocation,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible,
+ isPrimary)
+ }
+ if (createWriterThreadPool == null) {
+ partitionLocations.addAll(requestLocs.asScala.map(createWriter).asJava)
+ } else {
+ partitionLocations.addAll(requestLocs.asScala.map(requestLoc =>
+ createWriterThreadPool.submit(new Callable[PartitionLocation] {
+ override def call(): PartitionLocation = createWriter(requestLoc)
+ })).map(_.get()).asJava)
Review Comment:
The parallel implementation uses `.map(_.get())` which blocks on each Future sequentially, negating the benefits of parallelization. Consider using `Future.sequence` or collecting all futures first, then waiting for completion to achieve true parallelism.
```suggestion
val futures = requestLocs.asScala.map(requestLoc =>
createWriterThreadPool.submit(new Callable[PartitionLocation] {
override def call(): PartitionLocation = createWriter(requestLoc)
}))
partitionLocations.addAll(futures.map(_.get()).asJava)
```
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#issuecomment-3177659278
Ping @RexXiong, @turboFei, @zaynt4606. I have addressed above comments.
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#issuecomment-3569540319
This PR is stale because it has been open 20 days with no activity. Remove stale label or comment or this will be closed in 10 days.
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "turboFei (via GitHub)" <gi...@apache.org>.
turboFei commented on code in PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#discussion_r2233254379
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/WriterHelper.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.worker;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.celeborn.common.identity.UserIdentifier;
+import org.apache.celeborn.common.meta.WorkerPartitionLocationInfo;
+import org.apache.celeborn.common.protocol.PartitionLocation;
+import org.apache.celeborn.common.protocol.PartitionSplitMode;
+import org.apache.celeborn.common.protocol.PartitionType;
+import org.apache.celeborn.service.deploy.worker.storage.PartitionDataWriter;
+import org.apache.celeborn.service.deploy.worker.storage.StorageManager;
+
+public class WriterHelper {
+
+ private static final Logger LOG = LoggerFactory.getLogger(WriterHelper.class);
+
+ private final StorageManager storageManager;
+ private final WorkerPartitionLocationInfo partitionLocationInfo;
+
+ public WriterHelper(
+ StorageManager storageManager, WorkerPartitionLocationInfo partitionLocationInfo) {
+ this.storageManager = storageManager;
+ this.partitionLocationInfo = partitionLocationInfo;
+ }
+
+ public List<PartitionLocation> createWriters(
+ String shuffleKey,
+ String applicationId,
+ int shuffleId,
+ List<PartitionLocation> requestLocs,
+ long splitThreshold,
+ PartitionSplitMode splitMode,
+ PartitionType partitionType,
+ boolean rangeReadFilter,
+ UserIdentifier userIdentifier,
+ boolean partitionSplitEnabled,
+ boolean isSegmentGranularityVisible,
+ boolean isPrimary) {
+ List<PartitionLocation> locs = new ArrayList<>();
+ try {
+ ConcurrentLinkedQueue<PartitionLocation> locQueue = new ConcurrentLinkedQueue<>();
+ requestLocs
+ .parallelStream()
+ .forEach(
+ requestLoc -> {
+ PartitionLocation location =
+ isPrimary
+ ? partitionLocationInfo.getPrimaryLocation(
+ shuffleKey, requestLoc.getUniqueId())
+ : partitionLocationInfo.getReplicaLocation(
+ shuffleKey, requestLoc.getUniqueId());
+ if (location == null) {
+ location = requestLoc;
+ PartitionDataWriter writer;
+ try {
+ writer =
+ storageManager.createPartitionDataWriter(
+ applicationId,
+ shuffleId,
+ location,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ locQueue.add(new WorkingPartition(location, writer));
+ } else {
+ locQueue.add(location);
+ }
+ });
+ locs.addAll(locQueue);
+ } catch (Exception e) {
+ LOG.error("CreateWriter for {} failed.", shuffleKey, e);
+ }
+ return locs;
+ }
+
+ public void destroyWriters(List<PartitionLocation> locs, String shuffleKey) {
+ locs.parallelStream()
+ .forEach(
+ partitionLocation -> {
+ WorkingPartition workingPartition = (WorkingPartition) partitionLocation;
+ PartitionDataWriter fileWriter = workingPartition.getFileWriter();
+ fileWriter.destroy(
+ new IOException(
+ String.format(
+ "Destroy FileWriter %s caused by " + "reserving slots failed for %s.",
+ fileWriter, shuffleKey)));
+ });
+ }
+}
Review Comment:
Some concern:
1. for `parallelStream`, the parallelize is determined by the system processor number
2. and seems `parallelStream` is for CPU intensive, for here, is it IO intensive?
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "FMX (via GitHub)" <gi...@apache.org>.
FMX commented on code in PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#discussion_r2275708653
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Controller.scala:
##########
@@ -299,6 +259,114 @@ private[deploy] class Controller(
context.reply(ReserveSlotsResponse(StatusCode.SUCCESS))
}
+ private def createWriters(
+ shuffleKey: String,
+ applicationId: String,
+ shuffleId: Int,
+ requestLocs: jList[PartitionLocation],
+ splitThreshold: Long,
+ splitMode: PartitionSplitMode,
+ partitionType: PartitionType,
+ rangeReadFilter: Boolean,
+ userIdentifier: UserIdentifier,
+ partitionSplitEnabled: Boolean,
+ isSegmentGranularityVisible: Boolean,
+ isPrimary: Boolean): jList[PartitionLocation] = {
+ val partitionLocations = new jArrayList[PartitionLocation]()
+ try {
+ def createWriter(partitionLocation: PartitionLocation): PartitionLocation = {
+ createPartitionDataWriter(
+ shuffleKey,
+ applicationId,
+ shuffleId,
+ partitionLocation,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible,
+ isPrimary)
+ }
+ if (createWriterThreadPool == null) {
+ partitionLocations.addAll(requestLocs.asScala.map(createWriter).asJava)
+ } else {
+ partitionLocations.addAll(requestLocs.asScala.map(requestLoc => {
+ val errorMessage = s"Create FileWriter for $requestLoc $shuffleKey timeout."
+ Utils.tryWithTimeoutAndCallback(createWriter(requestLoc))({
Review Comment:
You have waited for every create file operation, which means that the operations are still serial.
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Controller.scala:
##########
@@ -299,6 +259,114 @@ private[deploy] class Controller(
context.reply(ReserveSlotsResponse(StatusCode.SUCCESS))
}
+ private def createWriters(
+ shuffleKey: String,
+ applicationId: String,
+ shuffleId: Int,
+ requestLocs: jList[PartitionLocation],
+ splitThreshold: Long,
+ splitMode: PartitionSplitMode,
+ partitionType: PartitionType,
+ rangeReadFilter: Boolean,
+ userIdentifier: UserIdentifier,
+ partitionSplitEnabled: Boolean,
+ isSegmentGranularityVisible: Boolean,
+ isPrimary: Boolean): jList[PartitionLocation] = {
+ val partitionLocations = new jArrayList[PartitionLocation]()
+ try {
+ def createWriter(partitionLocation: PartitionLocation): PartitionLocation = {
+ createPartitionDataWriter(
+ shuffleKey,
+ applicationId,
+ shuffleId,
+ partitionLocation,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible,
+ isPrimary)
+ }
+ if (createWriterThreadPool == null) {
+ partitionLocations.addAll(requestLocs.asScala.map(createWriter).asJava)
+ } else {
+ partitionLocations.addAll(requestLocs.asScala.map(requestLoc => {
+ val errorMessage = s"Create FileWriter for $requestLoc $shuffleKey timeout."
+ Utils.tryWithTimeoutAndCallback(createWriter(requestLoc))({
Review Comment:
`Utils.tryWithTimeoutAndCallback` will create a future task and wait for its operation to be done.
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#issuecomment-3610898838
This issue was closed because it has been staled for 10 days with no activity.
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#issuecomment-3327224024
This PR is stale because it has been open 20 days with no activity. Remove stale label or comment or this will be closed in 10 days.
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#issuecomment-3405223907
This PR is stale because it has been open 20 days with no activity. Remove stale label or comment or this will be closed in 10 days.
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#issuecomment-3912931387
@RexXiong, I would like to introduce corresponding metrics in another pull request.
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "zaynt4606 (via GitHub)" <gi...@apache.org>.
zaynt4606 commented on code in PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#discussion_r2265375712
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Controller.scala:
##########
@@ -299,6 +258,109 @@ private[deploy] class Controller(
context.reply(ReserveSlotsResponse(StatusCode.SUCCESS))
}
+ private def createWriters(
+ shuffleKey: String,
+ applicationId: String,
+ shuffleId: Int,
+ requestLocs: jList[PartitionLocation],
+ splitThreshold: Long,
+ splitMode: PartitionSplitMode,
+ partitionType: PartitionType,
+ rangeReadFilter: Boolean,
+ userIdentifier: UserIdentifier,
+ partitionSplitEnabled: Boolean,
+ isSegmentGranularityVisible: Boolean,
+ isPrimary: Boolean): jList[PartitionLocation] = {
+ val partitionLocations = new jArrayList[PartitionLocation]()
+ try {
+ def createWriter(partitionLocation: PartitionLocation): PartitionLocation = {
+ createPartitionDataWriter(
+ shuffleKey,
+ applicationId,
+ shuffleId,
+ partitionLocation,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible,
+ isPrimary)
+ }
+ if (createWriterThreadPool == null) {
+ partitionLocations.addAll(requestLocs.asScala.map(createWriter).asJava)
+ } else {
+ partitionLocations.addAll(requestLocs.asScala.map(requestLoc =>
+ createWriterThreadPool.submit(new Callable[PartitionLocation] {
+ override def call(): PartitionLocation = createWriter(requestLoc)
+ })).map(_.get()).asJava)
+ }
+ } catch {
+ case e: Exception =>
+ logError(s"Create FileWriter for $shuffleKey failed.", e)
Review Comment:
+1，and a timeout cause cancellation mechanism is also needed.
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#discussion_r2230465859
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/WriterHelper.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.worker;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.celeborn.common.identity.UserIdentifier;
+import org.apache.celeborn.common.meta.WorkerPartitionLocationInfo;
+import org.apache.celeborn.common.protocol.PartitionLocation;
+import org.apache.celeborn.common.protocol.PartitionSplitMode;
+import org.apache.celeborn.common.protocol.PartitionType;
+import org.apache.celeborn.common.protocol.message.ControlMessages.ReserveSlotsResponse;
+import org.apache.celeborn.common.protocol.message.StatusCode;
+import org.apache.celeborn.common.rpc.RpcCallContext;
+import org.apache.celeborn.service.deploy.worker.storage.PartitionDataWriter;
+import org.apache.celeborn.service.deploy.worker.storage.StorageManager;
+
+public class WriterHelper {
+
+ private static final Logger LOG = LoggerFactory.getLogger(WriterHelper.class);
+
+ private final StorageManager storageManager;
+ private final WorkerPartitionLocationInfo partitionLocationInfo;
+
+ public WriterHelper(
+ StorageManager storageManager, WorkerPartitionLocationInfo partitionLocationInfo) {
+ this.storageManager = storageManager;
+ this.partitionLocationInfo = partitionLocationInfo;
+ }
+
+ public List<PartitionLocation> createWriters(
+ String shuffleKey,
+ String applicationId,
+ int shuffleId,
+ List<PartitionLocation> requestLocs,
+ long splitThreshold,
+ PartitionSplitMode splitMode,
+ PartitionType partitionType,
+ boolean rangeReadFilter,
+ UserIdentifier userIdentifier,
+ boolean partitionSplitEnabled,
+ boolean isSegmentGranularityVisible) {
+ List<PartitionLocation> locs = new ArrayList<>();
+ try {
+ ConcurrentLinkedQueue<PartitionLocation> locQueue = new ConcurrentLinkedQueue<>();
+ requestLocs
+ .parallelStream()
+ .forEach(
+ requestLoc -> {
+ PartitionLocation location =
+ partitionLocationInfo.getPrimaryLocation(shuffleKey, requestLoc.getUniqueId());
+ if (location == null) {
+ location = requestLoc;
+ PartitionDataWriter writer;
+ try {
+ writer =
+ storageManager.createPartitionDataWriter(
+ applicationId,
+ shuffleId,
+ location,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ locQueue.add(new WorkingPartition(location, writer));
+ } else {
+ locQueue.add(location);
+ }
+ });
+ locs.addAll(locQueue);
+ } catch (Exception e) {
+ LOG.error("CreateWriter for {} failed.", shuffleKey, e);
+ }
+ return locs;
+ }
+
+ public void destroyWriters(
+ List<PartitionLocation> locs, RpcCallContext context, String shuffleKey, String msg) {
+ locs.parallelStream()
+ .forEach(
+ partitionLocation -> {
+ WorkingPartition workingPartition = (WorkingPartition) partitionLocation;
+ PartitionDataWriter fileWriter = workingPartition.getFileWriter();
+ fileWriter.destroy(
+ new IOException(
+ String.format(
+ "Destroy FileWriter %s caused by " + "reserving slots failed for %s.",
+ fileWriter, shuffleKey)));
+ });
+ context.reply(new ReserveSlotsResponse(StatusCode.RESERVE_SLOTS_FAILED, msg));
Review Comment:
The destroyWriters method should not send RPC replies as it's called multiple times in the Controller. This will cause multiple replies to be sent for the same RPC call, which is incorrect.
```suggestion
```
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/WriterHelper.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.worker;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.celeborn.common.identity.UserIdentifier;
+import org.apache.celeborn.common.meta.WorkerPartitionLocationInfo;
+import org.apache.celeborn.common.protocol.PartitionLocation;
+import org.apache.celeborn.common.protocol.PartitionSplitMode;
+import org.apache.celeborn.common.protocol.PartitionType;
+import org.apache.celeborn.common.protocol.message.ControlMessages.ReserveSlotsResponse;
+import org.apache.celeborn.common.protocol.message.StatusCode;
+import org.apache.celeborn.common.rpc.RpcCallContext;
+import org.apache.celeborn.service.deploy.worker.storage.PartitionDataWriter;
+import org.apache.celeborn.service.deploy.worker.storage.StorageManager;
+
+public class WriterHelper {
+
+ private static final Logger LOG = LoggerFactory.getLogger(WriterHelper.class);
+
+ private final StorageManager storageManager;
+ private final WorkerPartitionLocationInfo partitionLocationInfo;
+
+ public WriterHelper(
+ StorageManager storageManager, WorkerPartitionLocationInfo partitionLocationInfo) {
+ this.storageManager = storageManager;
+ this.partitionLocationInfo = partitionLocationInfo;
+ }
+
+ public List<PartitionLocation> createWriters(
+ String shuffleKey,
+ String applicationId,
+ int shuffleId,
+ List<PartitionLocation> requestLocs,
+ long splitThreshold,
+ PartitionSplitMode splitMode,
+ PartitionType partitionType,
+ boolean rangeReadFilter,
+ UserIdentifier userIdentifier,
+ boolean partitionSplitEnabled,
+ boolean isSegmentGranularityVisible) {
+ List<PartitionLocation> locs = new ArrayList<>();
+ try {
+ ConcurrentLinkedQueue<PartitionLocation> locQueue = new ConcurrentLinkedQueue<>();
+ requestLocs
+ .parallelStream()
+ .forEach(
+ requestLoc -> {
+ PartitionLocation location =
+ partitionLocationInfo.getPrimaryLocation(shuffleKey, requestLoc.getUniqueId());
Review Comment:
The method is calling `getPrimaryLocation` for all request locations, but this should differentiate between primary and replica locations based on the context. The original code used `getPrimaryLocation` for primary locations and `getReplicaLocation` for replica locations.
```suggestion
PartitionLocation location;
if (requestLoc.isPrimary()) {
location = partitionLocationInfo.getPrimaryLocation(shuffleKey, requestLoc.getUniqueId());
} else {
location = partitionLocationInfo.getReplicaLocation(shuffleKey, requestLoc.getUniqueId());
}
```
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "turboFei (via GitHub)" <gi...@apache.org>.
turboFei commented on code in PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#discussion_r2233258793
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/WriterHelper.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.worker;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.celeborn.common.identity.UserIdentifier;
+import org.apache.celeborn.common.meta.WorkerPartitionLocationInfo;
+import org.apache.celeborn.common.protocol.PartitionLocation;
+import org.apache.celeborn.common.protocol.PartitionSplitMode;
+import org.apache.celeborn.common.protocol.PartitionType;
+import org.apache.celeborn.service.deploy.worker.storage.PartitionDataWriter;
+import org.apache.celeborn.service.deploy.worker.storage.StorageManager;
+
+public class WriterHelper {
+
+ private static final Logger LOG = LoggerFactory.getLogger(WriterHelper.class);
+
+ private final StorageManager storageManager;
+ private final WorkerPartitionLocationInfo partitionLocationInfo;
+
+ public WriterHelper(
+ StorageManager storageManager, WorkerPartitionLocationInfo partitionLocationInfo) {
+ this.storageManager = storageManager;
+ this.partitionLocationInfo = partitionLocationInfo;
+ }
+
+ public List<PartitionLocation> createWriters(
+ String shuffleKey,
+ String applicationId,
+ int shuffleId,
+ List<PartitionLocation> requestLocs,
+ long splitThreshold,
+ PartitionSplitMode splitMode,
+ PartitionType partitionType,
+ boolean rangeReadFilter,
+ UserIdentifier userIdentifier,
+ boolean partitionSplitEnabled,
+ boolean isSegmentGranularityVisible,
+ boolean isPrimary) {
+ List<PartitionLocation> locs = new ArrayList<>();
+ try {
+ ConcurrentLinkedQueue<PartitionLocation> locQueue = new ConcurrentLinkedQueue<>();
+ requestLocs
+ .parallelStream()
+ .forEach(
+ requestLoc -> {
+ PartitionLocation location =
+ isPrimary
+ ? partitionLocationInfo.getPrimaryLocation(
+ shuffleKey, requestLoc.getUniqueId())
+ : partitionLocationInfo.getReplicaLocation(
+ shuffleKey, requestLoc.getUniqueId());
+ if (location == null) {
+ location = requestLoc;
+ PartitionDataWriter writer;
+ try {
+ writer =
+ storageManager.createPartitionDataWriter(
+ applicationId,
+ shuffleId,
+ location,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ locQueue.add(new WorkingPartition(location, writer));
+ } else {
+ locQueue.add(location);
+ }
+ });
+ locs.addAll(locQueue);
+ } catch (Exception e) {
+ LOG.error("CreateWriter for {} failed.", shuffleKey, e);
+ }
+ return locs;
+ }
+
+ public void destroyWriters(List<PartitionLocation> locs, String shuffleKey) {
+ locs.parallelStream()
+ .forEach(
+ partitionLocation -> {
+ WorkingPartition workingPartition = (WorkingPartition) partitionLocation;
+ PartitionDataWriter fileWriter = workingPartition.getFileWriter();
+ fileWriter.destroy(
+ new IOException(
+ String.format(
+ "Destroy FileWriter %s caused by " + "reserving slots failed for %s.",
+ fileWriter, shuffleKey)));
+ });
+ }
+}
Review Comment:
and only submit task to the executor pool if the number of locs is large than a threshold
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#discussion_r2812361056
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Controller.scala:
##########
@@ -291,15 +251,118 @@ private[deploy] class Controller(
Utils.getSlotsPerDisk(requestPrimaryLocs, requestReplicaLocs))
workerSource.incCounter(WorkerSource.SLOTS_ALLOCATED, primaryLocs.size() + replicaLocs.size())
- logInfo(s"Reserved ${primaryLocs.size()} primary location " +
- s"${primaryLocs.asScala.map(_.getUniqueId).mkString(",")} and ${replicaLocs.size()} replica location " +
- s"${replicaLocs.asScala.map(_.getUniqueId).mkString(",")} for $shuffleKey ")
+ logInfo(s"Reserved ${primaryLocs.size()} primary location" +
+ s" and ${replicaLocs.size()} replica location for $shuffleKey ")
if (log.isDebugEnabled()) {
logDebug(s"primary: $primaryLocs\nreplica: $replicaLocs.")
}
context.reply(ReserveSlotsResponse(StatusCode.SUCCESS))
}
+ private def createWriters(
+ shuffleKey: String,
+ applicationId: String,
+ shuffleId: Int,
+ requestLocs: jList[PartitionLocation],
+ splitThreshold: Long,
+ splitMode: PartitionSplitMode,
+ partitionType: PartitionType,
+ rangeReadFilter: Boolean,
+ userIdentifier: UserIdentifier,
+ partitionSplitEnabled: Boolean,
+ isSegmentGranularityVisible: Boolean,
+ isPrimary: Boolean): jList[PartitionLocation] = {
+ val partitionLocations = new jArrayList[PartitionLocation]()
+ try {
+ def createWriter(partitionLocation: PartitionLocation): PartitionLocation = {
+ createPartitionDataWriter(
+ shuffleKey,
+ applicationId,
+ shuffleId,
+ partitionLocation,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible,
+ isPrimary)
+ }
+ if (createWriterThreadPool == null) {
+ partitionLocations.addAll(requestLocs.asScala.map(createWriter).asJava)
+ } else {
+ partitionLocations.addAll(Utils.tryFuturesWithTimeout(
+ requestLocs.asScala.map(requestLoc =>
+ Utils.future(createWriter(requestLoc))(createWriterThreadPool)).toList,
+ createWriterParallelTimeout,
+ s"Create FileWriter for $shuffleKey timeout.").asJava)
+ }
+ } catch {
+ case e: Exception =>
+ logError(s"Create FileWriter for $shuffleKey failed.", e)
+ }
+ partitionLocations
+ }
+
+ private def createPartitionDataWriter(
+ shuffleKey: String,
+ applicationId: String,
+ shuffleId: Int,
+ requestLoc: PartitionLocation,
+ splitThreshold: Long,
+ splitMode: PartitionSplitMode,
+ partitionType: PartitionType,
+ rangeReadFilter: Boolean,
+ userIdentifier: UserIdentifier,
+ partitionSplitEnabled: Boolean,
+ isSegmentGranularityVisible: Boolean,
+ isPrimary: Boolean): PartitionLocation = {
+ try {
+ var location =
+ if (isPrimary) {
+ partitionLocationInfo.getPrimaryLocation(
+ shuffleKey,
+ requestLoc.getUniqueId)
+ } else {
+ partitionLocationInfo.getReplicaLocation(
+ shuffleKey,
+ requestLoc.getUniqueId)
+ }
+ if (location == null) {
+ location = requestLoc
+ val writer = storageManager.createPartitionDataWriter(
+ applicationId,
+ shuffleId,
+ location,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible)
+ new WorkingPartition(location, writer)
+ } else {
+ location
+ }
+ } catch {
+ case e: Exception =>
+ logError(s"Create FileWriter for $requestLoc $shuffleKey failed.", e)
+ throw e
+ }
+ }
+
+ private def destroyWriters(
+ partitionLocations: jList[PartitionLocation],
+ shuffleKey: String): Unit = {
+ partitionLocations.asScala.foreach { partitionLocation =>
+ val fileWriter = partitionLocation.asInstanceOf[WorkingPartition].getFileWriter
Review Comment:
@eolivelli, it's impossible that the value is not a `WorkingPartition`. The `destroyWriters` method only extracts the common codes from the following codes:
```
primaryLocs.asScala.foreach { partitionLocation =>
val fileWriter = partitionLocation.asInstanceOf[WorkingPartition].getFileWriter
fileWriter.destroy(new IOException(s"Destroy FileWriter $fileWriter caused by " +
s"reserving slots failed for $shuffleKey."))
}
replicaLocs.asScala.foreach { partitionLocation =>
val fileWriter = partitionLocation.asInstanceOf[WorkingPartition].getFileWriter
fileWriter.destroy(new IOException(s"Destroy FileWriter $fileWriter caused by " +
s"reserving slots failed for $shuffleKey."))
}
```
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas closed pull request #3387: [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time
URL: https://github.com/apache/celeborn/pull/3387
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#discussion_r2230291202
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/WriterHelper.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.worker;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.celeborn.common.identity.UserIdentifier;
+import org.apache.celeborn.common.meta.WorkerPartitionLocationInfo;
+import org.apache.celeborn.common.protocol.PartitionLocation;
+import org.apache.celeborn.common.protocol.PartitionSplitMode;
+import org.apache.celeborn.common.protocol.PartitionType;
+import org.apache.celeborn.common.protocol.message.ControlMessages.ReserveSlotsResponse;
+import org.apache.celeborn.common.protocol.message.StatusCode;
+import org.apache.celeborn.common.rpc.RpcCallContext;
+import org.apache.celeborn.service.deploy.worker.storage.PartitionDataWriter;
+import org.apache.celeborn.service.deploy.worker.storage.StorageManager;
+
+public class WriterHelper {
+
+ private static final Logger LOG = LoggerFactory.getLogger(WriterHelper.class);
+
+ private final StorageManager storageManager;
+ private final WorkerPartitionLocationInfo partitionLocationInfo;
+
+ public WriterHelper(
+ StorageManager storageManager, WorkerPartitionLocationInfo partitionLocationInfo) {
+ this.storageManager = storageManager;
+ this.partitionLocationInfo = partitionLocationInfo;
+ }
+
+ public List<PartitionLocation> createWriters(
+ String shuffleKey,
+ String applicationId,
+ int shuffleId,
+ List<PartitionLocation> requestLocs,
+ long splitThreshold,
+ PartitionSplitMode splitMode,
+ PartitionType partitionType,
+ boolean rangeReadFilter,
+ UserIdentifier userIdentifier,
+ boolean partitionSplitEnabled,
+ boolean isSegmentGranularityVisible) {
+ List<PartitionLocation> locs = new ArrayList<>();
+ try {
+ ConcurrentLinkedQueue<PartitionLocation> locQueue = new ConcurrentLinkedQueue<>();
+ requestLocs
+ .parallelStream()
Review Comment:
The reason why use java `.parallelStream()` instead of scala `.par` is that there is no good way to solve the compatibility of scala `.par` between scala 2.12 and 2.13.
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "turboFei (via GitHub)" <gi...@apache.org>.
turboFei commented on code in PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#discussion_r2233254379
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/WriterHelper.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.worker;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.celeborn.common.identity.UserIdentifier;
+import org.apache.celeborn.common.meta.WorkerPartitionLocationInfo;
+import org.apache.celeborn.common.protocol.PartitionLocation;
+import org.apache.celeborn.common.protocol.PartitionSplitMode;
+import org.apache.celeborn.common.protocol.PartitionType;
+import org.apache.celeborn.service.deploy.worker.storage.PartitionDataWriter;
+import org.apache.celeborn.service.deploy.worker.storage.StorageManager;
+
+public class WriterHelper {
+
+ private static final Logger LOG = LoggerFactory.getLogger(WriterHelper.class);
+
+ private final StorageManager storageManager;
+ private final WorkerPartitionLocationInfo partitionLocationInfo;
+
+ public WriterHelper(
+ StorageManager storageManager, WorkerPartitionLocationInfo partitionLocationInfo) {
+ this.storageManager = storageManager;
+ this.partitionLocationInfo = partitionLocationInfo;
+ }
+
+ public List<PartitionLocation> createWriters(
+ String shuffleKey,
+ String applicationId,
+ int shuffleId,
+ List<PartitionLocation> requestLocs,
+ long splitThreshold,
+ PartitionSplitMode splitMode,
+ PartitionType partitionType,
+ boolean rangeReadFilter,
+ UserIdentifier userIdentifier,
+ boolean partitionSplitEnabled,
+ boolean isSegmentGranularityVisible,
+ boolean isPrimary) {
+ List<PartitionLocation> locs = new ArrayList<>();
+ try {
+ ConcurrentLinkedQueue<PartitionLocation> locQueue = new ConcurrentLinkedQueue<>();
+ requestLocs
+ .parallelStream()
+ .forEach(
+ requestLoc -> {
+ PartitionLocation location =
+ isPrimary
+ ? partitionLocationInfo.getPrimaryLocation(
+ shuffleKey, requestLoc.getUniqueId())
+ : partitionLocationInfo.getReplicaLocation(
+ shuffleKey, requestLoc.getUniqueId());
+ if (location == null) {
+ location = requestLoc;
+ PartitionDataWriter writer;
+ try {
+ writer =
+ storageManager.createPartitionDataWriter(
+ applicationId,
+ shuffleId,
+ location,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ locQueue.add(new WorkingPartition(location, writer));
+ } else {
+ locQueue.add(location);
+ }
+ });
+ locs.addAll(locQueue);
+ } catch (Exception e) {
+ LOG.error("CreateWriter for {} failed.", shuffleKey, e);
+ }
+ return locs;
+ }
+
+ public void destroyWriters(List<PartitionLocation> locs, String shuffleKey) {
+ locs.parallelStream()
+ .forEach(
+ partitionLocation -> {
+ WorkingPartition workingPartition = (WorkingPartition) partitionLocation;
+ PartitionDataWriter fileWriter = workingPartition.getFileWriter();
+ fileWriter.destroy(
+ new IOException(
+ String.format(
+ "Destroy FileWriter %s caused by " + "reserving slots failed for %s.",
+ fileWriter, shuffleKey)));
+ });
+ }
+}
Review Comment:
Some concern:
for `parallelStream`, the parallelize is determined by the system processor number and seems `parallelStream` is for CPU intensive, for here, is it IO intensive?
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "turboFei (via GitHub)" <gi...@apache.org>.
turboFei commented on code in PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#discussion_r2233258793
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/WriterHelper.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.worker;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.celeborn.common.identity.UserIdentifier;
+import org.apache.celeborn.common.meta.WorkerPartitionLocationInfo;
+import org.apache.celeborn.common.protocol.PartitionLocation;
+import org.apache.celeborn.common.protocol.PartitionSplitMode;
+import org.apache.celeborn.common.protocol.PartitionType;
+import org.apache.celeborn.service.deploy.worker.storage.PartitionDataWriter;
+import org.apache.celeborn.service.deploy.worker.storage.StorageManager;
+
+public class WriterHelper {
+
+ private static final Logger LOG = LoggerFactory.getLogger(WriterHelper.class);
+
+ private final StorageManager storageManager;
+ private final WorkerPartitionLocationInfo partitionLocationInfo;
+
+ public WriterHelper(
+ StorageManager storageManager, WorkerPartitionLocationInfo partitionLocationInfo) {
+ this.storageManager = storageManager;
+ this.partitionLocationInfo = partitionLocationInfo;
+ }
+
+ public List<PartitionLocation> createWriters(
+ String shuffleKey,
+ String applicationId,
+ int shuffleId,
+ List<PartitionLocation> requestLocs,
+ long splitThreshold,
+ PartitionSplitMode splitMode,
+ PartitionType partitionType,
+ boolean rangeReadFilter,
+ UserIdentifier userIdentifier,
+ boolean partitionSplitEnabled,
+ boolean isSegmentGranularityVisible,
+ boolean isPrimary) {
+ List<PartitionLocation> locs = new ArrayList<>();
+ try {
+ ConcurrentLinkedQueue<PartitionLocation> locQueue = new ConcurrentLinkedQueue<>();
+ requestLocs
+ .parallelStream()
+ .forEach(
+ requestLoc -> {
+ PartitionLocation location =
+ isPrimary
+ ? partitionLocationInfo.getPrimaryLocation(
+ shuffleKey, requestLoc.getUniqueId())
+ : partitionLocationInfo.getReplicaLocation(
+ shuffleKey, requestLoc.getUniqueId());
+ if (location == null) {
+ location = requestLoc;
+ PartitionDataWriter writer;
+ try {
+ writer =
+ storageManager.createPartitionDataWriter(
+ applicationId,
+ shuffleId,
+ location,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ locQueue.add(new WorkingPartition(location, writer));
+ } else {
+ locQueue.add(location);
+ }
+ });
+ locs.addAll(locQueue);
+ } catch (Exception e) {
+ LOG.error("CreateWriter for {} failed.", shuffleKey, e);
+ }
+ return locs;
+ }
+
+ public void destroyWriters(List<PartitionLocation> locs, String shuffleKey) {
+ locs.parallelStream()
+ .forEach(
+ partitionLocation -> {
+ WorkingPartition workingPartition = (WorkingPartition) partitionLocation;
+ PartitionDataWriter fileWriter = workingPartition.getFileWriter();
+ fileWriter.destroy(
+ new IOException(
+ String.format(
+ "Destroy FileWriter %s caused by " + "reserving slots failed for %s.",
+ fileWriter, shuffleKey)));
+ });
+ }
+}
Review Comment:
and only submit task to the executor pool if the number of locs large than a threshold
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "turboFei (via GitHub)" <gi...@apache.org>.
turboFei commented on code in PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#discussion_r2233257638
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/WriterHelper.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.worker;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.celeborn.common.identity.UserIdentifier;
+import org.apache.celeborn.common.meta.WorkerPartitionLocationInfo;
+import org.apache.celeborn.common.protocol.PartitionLocation;
+import org.apache.celeborn.common.protocol.PartitionSplitMode;
+import org.apache.celeborn.common.protocol.PartitionType;
+import org.apache.celeborn.service.deploy.worker.storage.PartitionDataWriter;
+import org.apache.celeborn.service.deploy.worker.storage.StorageManager;
+
+public class WriterHelper {
+
+ private static final Logger LOG = LoggerFactory.getLogger(WriterHelper.class);
+
+ private final StorageManager storageManager;
+ private final WorkerPartitionLocationInfo partitionLocationInfo;
+
+ public WriterHelper(
+ StorageManager storageManager, WorkerPartitionLocationInfo partitionLocationInfo) {
+ this.storageManager = storageManager;
+ this.partitionLocationInfo = partitionLocationInfo;
+ }
+
+ public List<PartitionLocation> createWriters(
+ String shuffleKey,
+ String applicationId,
+ int shuffleId,
+ List<PartitionLocation> requestLocs,
+ long splitThreshold,
+ PartitionSplitMode splitMode,
+ PartitionType partitionType,
+ boolean rangeReadFilter,
+ UserIdentifier userIdentifier,
+ boolean partitionSplitEnabled,
+ boolean isSegmentGranularityVisible,
+ boolean isPrimary) {
+ List<PartitionLocation> locs = new ArrayList<>();
+ try {
+ ConcurrentLinkedQueue<PartitionLocation> locQueue = new ConcurrentLinkedQueue<>();
+ requestLocs
+ .parallelStream()
+ .forEach(
+ requestLoc -> {
+ PartitionLocation location =
+ isPrimary
+ ? partitionLocationInfo.getPrimaryLocation(
+ shuffleKey, requestLoc.getUniqueId())
+ : partitionLocationInfo.getReplicaLocation(
+ shuffleKey, requestLoc.getUniqueId());
+ if (location == null) {
+ location = requestLoc;
+ PartitionDataWriter writer;
+ try {
+ writer =
+ storageManager.createPartitionDataWriter(
+ applicationId,
+ shuffleId,
+ location,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ locQueue.add(new WorkingPartition(location, writer));
+ } else {
+ locQueue.add(location);
+ }
+ });
+ locs.addAll(locQueue);
+ } catch (Exception e) {
+ LOG.error("CreateWriter for {} failed.", shuffleKey, e);
+ }
+ return locs;
+ }
+
+ public void destroyWriters(List<PartitionLocation> locs, String shuffleKey) {
+ locs.parallelStream()
+ .forEach(
+ partitionLocation -> {
+ WorkingPartition workingPartition = (WorkingPartition) partitionLocation;
+ PartitionDataWriter fileWriter = workingPartition.getFileWriter();
+ fileWriter.destroy(
+ new IOException(
+ String.format(
+ "Destroy FileWriter %s caused by " + "reserving slots failed for %s.",
+ fileWriter, shuffleKey)));
+ });
+ }
+}
Review Comment:
maybe use `newDaemonFixedThreadPool`?
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#issuecomment-3116755318
Ping @RexXiong, @FMX.
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "turboFei (via GitHub)" <gi...@apache.org>.
turboFei commented on code in PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#discussion_r2233254379
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/WriterHelper.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.worker;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.celeborn.common.identity.UserIdentifier;
+import org.apache.celeborn.common.meta.WorkerPartitionLocationInfo;
+import org.apache.celeborn.common.protocol.PartitionLocation;
+import org.apache.celeborn.common.protocol.PartitionSplitMode;
+import org.apache.celeborn.common.protocol.PartitionType;
+import org.apache.celeborn.service.deploy.worker.storage.PartitionDataWriter;
+import org.apache.celeborn.service.deploy.worker.storage.StorageManager;
+
+public class WriterHelper {
+
+ private static final Logger LOG = LoggerFactory.getLogger(WriterHelper.class);
+
+ private final StorageManager storageManager;
+ private final WorkerPartitionLocationInfo partitionLocationInfo;
+
+ public WriterHelper(
+ StorageManager storageManager, WorkerPartitionLocationInfo partitionLocationInfo) {
+ this.storageManager = storageManager;
+ this.partitionLocationInfo = partitionLocationInfo;
+ }
+
+ public List<PartitionLocation> createWriters(
+ String shuffleKey,
+ String applicationId,
+ int shuffleId,
+ List<PartitionLocation> requestLocs,
+ long splitThreshold,
+ PartitionSplitMode splitMode,
+ PartitionType partitionType,
+ boolean rangeReadFilter,
+ UserIdentifier userIdentifier,
+ boolean partitionSplitEnabled,
+ boolean isSegmentGranularityVisible,
+ boolean isPrimary) {
+ List<PartitionLocation> locs = new ArrayList<>();
+ try {
+ ConcurrentLinkedQueue<PartitionLocation> locQueue = new ConcurrentLinkedQueue<>();
+ requestLocs
+ .parallelStream()
+ .forEach(
+ requestLoc -> {
+ PartitionLocation location =
+ isPrimary
+ ? partitionLocationInfo.getPrimaryLocation(
+ shuffleKey, requestLoc.getUniqueId())
+ : partitionLocationInfo.getReplicaLocation(
+ shuffleKey, requestLoc.getUniqueId());
+ if (location == null) {
+ location = requestLoc;
+ PartitionDataWriter writer;
+ try {
+ writer =
+ storageManager.createPartitionDataWriter(
+ applicationId,
+ shuffleId,
+ location,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ locQueue.add(new WorkingPartition(location, writer));
+ } else {
+ locQueue.add(location);
+ }
+ });
+ locs.addAll(locQueue);
+ } catch (Exception e) {
+ LOG.error("CreateWriter for {} failed.", shuffleKey, e);
+ }
+ return locs;
+ }
+
+ public void destroyWriters(List<PartitionLocation> locs, String shuffleKey) {
+ locs.parallelStream()
+ .forEach(
+ partitionLocation -> {
+ WorkingPartition workingPartition = (WorkingPartition) partitionLocation;
+ PartitionDataWriter fileWriter = workingPartition.getFileWriter();
+ fileWriter.destroy(
+ new IOException(
+ String.format(
+ "Destroy FileWriter %s caused by " + "reserving slots failed for %s.",
+ fileWriter, shuffleKey)));
+ });
+ }
+}
Review Comment:
not sure how much benefit with parallelStream.
Some concern:
1. for `parallelStream`, the parallelize is determined by the system processor number
2. and `parallelStream` is for CPU intensive, for here, is it IO intensive?
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/WriterHelper.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.worker;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.celeborn.common.identity.UserIdentifier;
+import org.apache.celeborn.common.meta.WorkerPartitionLocationInfo;
+import org.apache.celeborn.common.protocol.PartitionLocation;
+import org.apache.celeborn.common.protocol.PartitionSplitMode;
+import org.apache.celeborn.common.protocol.PartitionType;
+import org.apache.celeborn.service.deploy.worker.storage.PartitionDataWriter;
+import org.apache.celeborn.service.deploy.worker.storage.StorageManager;
+
+public class WriterHelper {
+
+ private static final Logger LOG = LoggerFactory.getLogger(WriterHelper.class);
+
+ private final StorageManager storageManager;
+ private final WorkerPartitionLocationInfo partitionLocationInfo;
+
+ public WriterHelper(
+ StorageManager storageManager, WorkerPartitionLocationInfo partitionLocationInfo) {
+ this.storageManager = storageManager;
+ this.partitionLocationInfo = partitionLocationInfo;
+ }
+
+ public List<PartitionLocation> createWriters(
+ String shuffleKey,
+ String applicationId,
+ int shuffleId,
+ List<PartitionLocation> requestLocs,
+ long splitThreshold,
+ PartitionSplitMode splitMode,
+ PartitionType partitionType,
+ boolean rangeReadFilter,
+ UserIdentifier userIdentifier,
+ boolean partitionSplitEnabled,
+ boolean isSegmentGranularityVisible,
+ boolean isPrimary) {
+ List<PartitionLocation> locs = new ArrayList<>();
+ try {
+ ConcurrentLinkedQueue<PartitionLocation> locQueue = new ConcurrentLinkedQueue<>();
+ requestLocs
+ .parallelStream()
+ .forEach(
+ requestLoc -> {
+ PartitionLocation location =
+ isPrimary
+ ? partitionLocationInfo.getPrimaryLocation(
+ shuffleKey, requestLoc.getUniqueId())
+ : partitionLocationInfo.getReplicaLocation(
+ shuffleKey, requestLoc.getUniqueId());
+ if (location == null) {
+ location = requestLoc;
+ PartitionDataWriter writer;
+ try {
+ writer =
+ storageManager.createPartitionDataWriter(
+ applicationId,
+ shuffleId,
+ location,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ locQueue.add(new WorkingPartition(location, writer));
+ } else {
+ locQueue.add(location);
+ }
+ });
+ locs.addAll(locQueue);
+ } catch (Exception e) {
+ LOG.error("CreateWriter for {} failed.", shuffleKey, e);
+ }
+ return locs;
+ }
+
+ public void destroyWriters(List<PartitionLocation> locs, String shuffleKey) {
+ locs.parallelStream()
+ .forEach(
+ partitionLocation -> {
+ WorkingPartition workingPartition = (WorkingPartition) partitionLocation;
+ PartitionDataWriter fileWriter = workingPartition.getFileWriter();
+ fileWriter.destroy(
+ new IOException(
+ String.format(
+ "Destroy FileWriter %s caused by " + "reserving slots failed for %s.",
+ fileWriter, shuffleKey)));
+ });
+ }
+}
Review Comment:
Some concern:
1. for `parallelStream`, the parallelize is determined by the system processor number
2. and `parallelStream` is for CPU intensive, for here, is it IO intensive?
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "FMX (via GitHub)" <gi...@apache.org>.
FMX commented on PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#issuecomment-3226558298
Plz share some stats about this PR. I was wondering if this PR is helpful for the Spark job's e2e time.
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "eolivelli (via GitHub)" <gi...@apache.org>.
eolivelli commented on code in PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#discussion_r2812469609
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Controller.scala:
##########
@@ -291,15 +251,118 @@ private[deploy] class Controller(
Utils.getSlotsPerDisk(requestPrimaryLocs, requestReplicaLocs))
workerSource.incCounter(WorkerSource.SLOTS_ALLOCATED, primaryLocs.size() + replicaLocs.size())
- logInfo(s"Reserved ${primaryLocs.size()} primary location " +
- s"${primaryLocs.asScala.map(_.getUniqueId).mkString(",")} and ${replicaLocs.size()} replica location " +
- s"${replicaLocs.asScala.map(_.getUniqueId).mkString(",")} for $shuffleKey ")
+ logInfo(s"Reserved ${primaryLocs.size()} primary location" +
+ s" and ${replicaLocs.size()} replica location for $shuffleKey ")
if (log.isDebugEnabled()) {
logDebug(s"primary: $primaryLocs\nreplica: $replicaLocs.")
}
context.reply(ReserveSlotsResponse(StatusCode.SUCCESS))
}
+ private def createWriters(
+ shuffleKey: String,
+ applicationId: String,
+ shuffleId: Int,
+ requestLocs: jList[PartitionLocation],
+ splitThreshold: Long,
+ splitMode: PartitionSplitMode,
+ partitionType: PartitionType,
+ rangeReadFilter: Boolean,
+ userIdentifier: UserIdentifier,
+ partitionSplitEnabled: Boolean,
+ isSegmentGranularityVisible: Boolean,
+ isPrimary: Boolean): jList[PartitionLocation] = {
+ val partitionLocations = new jArrayList[PartitionLocation]()
+ try {
+ def createWriter(partitionLocation: PartitionLocation): PartitionLocation = {
+ createPartitionDataWriter(
+ shuffleKey,
+ applicationId,
+ shuffleId,
+ partitionLocation,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible,
+ isPrimary)
+ }
+ if (createWriterThreadPool == null) {
+ partitionLocations.addAll(requestLocs.asScala.map(createWriter).asJava)
+ } else {
+ partitionLocations.addAll(Utils.tryFuturesWithTimeout(
+ requestLocs.asScala.map(requestLoc =>
+ Utils.future(createWriter(requestLoc))(createWriterThreadPool)).toList,
+ createWriterParallelTimeout,
+ s"Create FileWriter for $shuffleKey timeout.").asJava)
+ }
+ } catch {
+ case e: Exception =>
+ logError(s"Create FileWriter for $shuffleKey failed.", e)
+ }
+ partitionLocations
+ }
+
+ private def createPartitionDataWriter(
+ shuffleKey: String,
+ applicationId: String,
+ shuffleId: Int,
+ requestLoc: PartitionLocation,
+ splitThreshold: Long,
+ splitMode: PartitionSplitMode,
+ partitionType: PartitionType,
+ rangeReadFilter: Boolean,
+ userIdentifier: UserIdentifier,
+ partitionSplitEnabled: Boolean,
+ isSegmentGranularityVisible: Boolean,
+ isPrimary: Boolean): PartitionLocation = {
+ try {
+ var location =
+ if (isPrimary) {
+ partitionLocationInfo.getPrimaryLocation(
+ shuffleKey,
+ requestLoc.getUniqueId)
+ } else {
+ partitionLocationInfo.getReplicaLocation(
+ shuffleKey,
+ requestLoc.getUniqueId)
+ }
+ if (location == null) {
+ location = requestLoc
+ val writer = storageManager.createPartitionDataWriter(
+ applicationId,
+ shuffleId,
+ location,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible)
+ new WorkingPartition(location, writer)
+ } else {
+ location
+ }
+ } catch {
+ case e: Exception =>
+ logError(s"Create FileWriter for $requestLoc $shuffleKey failed.", e)
+ throw e
+ }
+ }
+
+ private def destroyWriters(
+ partitionLocations: jList[PartitionLocation],
+ shuffleKey: String): Unit = {
+ partitionLocations.asScala.foreach { partitionLocation =>
+ val fileWriter = partitionLocation.asInstanceOf[WorkingPartition].getFileWriter
Review Comment:
The behavior is the same as before, this patch does not break it.
okay
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "eolivelli (via GitHub)" <gi...@apache.org>.
eolivelli commented on PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#issuecomment-3907073180
I confirm that this PR solves the same problem as in https://github.com/apache/celeborn/pull/3598
The PR is closed so I cannot "approve", but the patch LGTM
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "RexXiong (via GitHub)" <gi...@apache.org>.
RexXiong commented on code in PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#discussion_r2254291168
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/WriterHelper.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.worker;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.celeborn.common.identity.UserIdentifier;
+import org.apache.celeborn.common.meta.WorkerPartitionLocationInfo;
+import org.apache.celeborn.common.protocol.PartitionLocation;
+import org.apache.celeborn.common.protocol.PartitionSplitMode;
+import org.apache.celeborn.common.protocol.PartitionType;
+import org.apache.celeborn.service.deploy.worker.storage.PartitionDataWriter;
+import org.apache.celeborn.service.deploy.worker.storage.StorageManager;
+
+public class WriterHelper {
+
+ private static final Logger LOG = LoggerFactory.getLogger(WriterHelper.class);
+
+ private final StorageManager storageManager;
+ private final WorkerPartitionLocationInfo partitionLocationInfo;
+
+ public WriterHelper(
+ StorageManager storageManager, WorkerPartitionLocationInfo partitionLocationInfo) {
+ this.storageManager = storageManager;
+ this.partitionLocationInfo = partitionLocationInfo;
+ }
+
+ public List<PartitionLocation> createWriters(
+ String shuffleKey,
+ String applicationId,
+ int shuffleId,
+ List<PartitionLocation> requestLocs,
+ long splitThreshold,
+ PartitionSplitMode splitMode,
+ PartitionType partitionType,
+ boolean rangeReadFilter,
+ UserIdentifier userIdentifier,
+ boolean partitionSplitEnabled,
+ boolean isSegmentGranularityVisible,
+ boolean isPrimary) {
+ List<PartitionLocation> locs = new ArrayList<>();
+ try {
+ ConcurrentLinkedQueue<PartitionLocation> locQueue = new ConcurrentLinkedQueue<>();
+ requestLocs
+ .parallelStream()
+ .forEach(
+ requestLoc -> {
+ PartitionLocation location =
+ isPrimary
+ ? partitionLocationInfo.getPrimaryLocation(
+ shuffleKey, requestLoc.getUniqueId())
+ : partitionLocationInfo.getReplicaLocation(
+ shuffleKey, requestLoc.getUniqueId());
+ if (location == null) {
+ location = requestLoc;
+ PartitionDataWriter writer;
+ try {
+ writer =
+ storageManager.createPartitionDataWriter(
+ applicationId,
+ shuffleId,
+ location,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ locQueue.add(new WorkingPartition(location, writer));
+ } else {
+ locQueue.add(location);
+ }
+ });
+ locs.addAll(locQueue);
+ } catch (Exception e) {
+ LOG.error("CreateWriter for {} failed.", shuffleKey, e);
+ }
+ return locs;
+ }
+
+ public void destroyWriters(List<PartitionLocation> locs, String shuffleKey) {
+ locs.parallelStream()
+ .forEach(
+ partitionLocation -> {
+ WorkingPartition workingPartition = (WorkingPartition) partitionLocation;
+ PartitionDataWriter fileWriter = workingPartition.getFileWriter();
+ fileWriter.destroy(
+ new IOException(
+ String.format(
+ "Destroy FileWriter %s caused by " + "reserving slots failed for %s.",
+ fileWriter, shuffleKey)));
+ });
+ }
+}
Review Comment:
Using parallelStream to destroyWriters may cause the worker to lose control over IO concurrency, Agree with @turboFei may be we can use newDaemonFixedThreadPool instead.
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "RexXiong (via GitHub)" <gi...@apache.org>.
RexXiong commented on code in PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#discussion_r2254291168
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/WriterHelper.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.worker;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.celeborn.common.identity.UserIdentifier;
+import org.apache.celeborn.common.meta.WorkerPartitionLocationInfo;
+import org.apache.celeborn.common.protocol.PartitionLocation;
+import org.apache.celeborn.common.protocol.PartitionSplitMode;
+import org.apache.celeborn.common.protocol.PartitionType;
+import org.apache.celeborn.service.deploy.worker.storage.PartitionDataWriter;
+import org.apache.celeborn.service.deploy.worker.storage.StorageManager;
+
+public class WriterHelper {
+
+ private static final Logger LOG = LoggerFactory.getLogger(WriterHelper.class);
+
+ private final StorageManager storageManager;
+ private final WorkerPartitionLocationInfo partitionLocationInfo;
+
+ public WriterHelper(
+ StorageManager storageManager, WorkerPartitionLocationInfo partitionLocationInfo) {
+ this.storageManager = storageManager;
+ this.partitionLocationInfo = partitionLocationInfo;
+ }
+
+ public List<PartitionLocation> createWriters(
+ String shuffleKey,
+ String applicationId,
+ int shuffleId,
+ List<PartitionLocation> requestLocs,
+ long splitThreshold,
+ PartitionSplitMode splitMode,
+ PartitionType partitionType,
+ boolean rangeReadFilter,
+ UserIdentifier userIdentifier,
+ boolean partitionSplitEnabled,
+ boolean isSegmentGranularityVisible,
+ boolean isPrimary) {
+ List<PartitionLocation> locs = new ArrayList<>();
+ try {
+ ConcurrentLinkedQueue<PartitionLocation> locQueue = new ConcurrentLinkedQueue<>();
+ requestLocs
+ .parallelStream()
+ .forEach(
+ requestLoc -> {
+ PartitionLocation location =
+ isPrimary
+ ? partitionLocationInfo.getPrimaryLocation(
+ shuffleKey, requestLoc.getUniqueId())
+ : partitionLocationInfo.getReplicaLocation(
+ shuffleKey, requestLoc.getUniqueId());
+ if (location == null) {
+ location = requestLoc;
+ PartitionDataWriter writer;
+ try {
+ writer =
+ storageManager.createPartitionDataWriter(
+ applicationId,
+ shuffleId,
+ location,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ locQueue.add(new WorkingPartition(location, writer));
+ } else {
+ locQueue.add(location);
+ }
+ });
+ locs.addAll(locQueue);
+ } catch (Exception e) {
+ LOG.error("CreateWriter for {} failed.", shuffleKey, e);
+ }
+ return locs;
+ }
+
+ public void destroyWriters(List<PartitionLocation> locs, String shuffleKey) {
+ locs.parallelStream()
+ .forEach(
+ partitionLocation -> {
+ WorkingPartition workingPartition = (WorkingPartition) partitionLocation;
+ PartitionDataWriter fileWriter = workingPartition.getFileWriter();
+ fileWriter.destroy(
+ new IOException(
+ String.format(
+ "Destroy FileWriter %s caused by " + "reserving slots failed for %s.",
+ fileWriter, shuffleKey)));
+ });
+ }
+}
Review Comment:
Using parallelStream to createWriters may cause the worker to lose control over IO concurrency, Agree with @turboFei may be we can use newDaemonFixedThreadPool instead.
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "turboFei (via GitHub)" <gi...@apache.org>.
turboFei commented on code in PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#discussion_r2261796928
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Controller.scala:
##########
@@ -299,6 +258,109 @@ private[deploy] class Controller(
context.reply(ReserveSlotsResponse(StatusCode.SUCCESS))
}
+ private def createWriters(
+ shuffleKey: String,
+ applicationId: String,
+ shuffleId: Int,
+ requestLocs: jList[PartitionLocation],
+ splitThreshold: Long,
+ splitMode: PartitionSplitMode,
+ partitionType: PartitionType,
+ rangeReadFilter: Boolean,
+ userIdentifier: UserIdentifier,
+ partitionSplitEnabled: Boolean,
+ isSegmentGranularityVisible: Boolean,
+ isPrimary: Boolean): jList[PartitionLocation] = {
+ val partitionLocations = new jArrayList[PartitionLocation]()
+ try {
+ def createWriter(partitionLocation: PartitionLocation): PartitionLocation = {
+ createPartitionDataWriter(
+ shuffleKey,
+ applicationId,
+ shuffleId,
+ partitionLocation,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible,
+ isPrimary)
+ }
+ if (createWriterThreadPool == null) {
+ partitionLocations.addAll(requestLocs.asScala.map(createWriter).asJava)
+ } else {
+ partitionLocations.addAll(requestLocs.asScala.map(requestLoc =>
+ createWriterThreadPool.submit(new Callable[PartitionLocation] {
+ override def call(): PartitionLocation = createWriter(requestLoc)
+ })).map(_.get()).asJava)
+ }
+ } catch {
+ case e: Exception =>
+ logError(s"Create FileWriter for $shuffleKey failed.", e)
Review Comment:
should we track these futures and cancel the remaining futures on exception?
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#issuecomment-3913020089
Merged to main(v0.7.0) and branch-0.6(v0.6.3).
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #3387:
URL: https://github.com/apache/celeborn/pull/3387#discussion_r2812429729
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Controller.scala:
##########
@@ -291,15 +251,118 @@ private[deploy] class Controller(
Utils.getSlotsPerDisk(requestPrimaryLocs, requestReplicaLocs))
workerSource.incCounter(WorkerSource.SLOTS_ALLOCATED, primaryLocs.size() + replicaLocs.size())
- logInfo(s"Reserved ${primaryLocs.size()} primary location " +
- s"${primaryLocs.asScala.map(_.getUniqueId).mkString(",")} and ${replicaLocs.size()} replica location " +
- s"${replicaLocs.asScala.map(_.getUniqueId).mkString(",")} for $shuffleKey ")
+ logInfo(s"Reserved ${primaryLocs.size()} primary location" +
+ s" and ${replicaLocs.size()} replica location for $shuffleKey ")
if (log.isDebugEnabled()) {
logDebug(s"primary: $primaryLocs\nreplica: $replicaLocs.")
}
context.reply(ReserveSlotsResponse(StatusCode.SUCCESS))
}
+ private def createWriters(
+ shuffleKey: String,
+ applicationId: String,
+ shuffleId: Int,
+ requestLocs: jList[PartitionLocation],
+ splitThreshold: Long,
+ splitMode: PartitionSplitMode,
+ partitionType: PartitionType,
+ rangeReadFilter: Boolean,
+ userIdentifier: UserIdentifier,
+ partitionSplitEnabled: Boolean,
+ isSegmentGranularityVisible: Boolean,
+ isPrimary: Boolean): jList[PartitionLocation] = {
+ val partitionLocations = new jArrayList[PartitionLocation]()
+ try {
+ def createWriter(partitionLocation: PartitionLocation): PartitionLocation = {
+ createPartitionDataWriter(
+ shuffleKey,
+ applicationId,
+ shuffleId,
+ partitionLocation,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible,
+ isPrimary)
+ }
+ if (createWriterThreadPool == null) {
+ partitionLocations.addAll(requestLocs.asScala.map(createWriter).asJava)
+ } else {
+ partitionLocations.addAll(Utils.tryFuturesWithTimeout(
+ requestLocs.asScala.map(requestLoc =>
+ Utils.future(createWriter(requestLoc))(createWriterThreadPool)).toList,
+ createWriterParallelTimeout,
+ s"Create FileWriter for $shuffleKey timeout.").asJava)
+ }
+ } catch {
+ case e: Exception =>
+ logError(s"Create FileWriter for $shuffleKey failed.", e)
+ }
+ partitionLocations
+ }
+
+ private def createPartitionDataWriter(
+ shuffleKey: String,
+ applicationId: String,
+ shuffleId: Int,
+ requestLoc: PartitionLocation,
+ splitThreshold: Long,
+ splitMode: PartitionSplitMode,
+ partitionType: PartitionType,
+ rangeReadFilter: Boolean,
+ userIdentifier: UserIdentifier,
+ partitionSplitEnabled: Boolean,
+ isSegmentGranularityVisible: Boolean,
+ isPrimary: Boolean): PartitionLocation = {
+ try {
+ var location =
+ if (isPrimary) {
+ partitionLocationInfo.getPrimaryLocation(
+ shuffleKey,
+ requestLoc.getUniqueId)
+ } else {
+ partitionLocationInfo.getReplicaLocation(
+ shuffleKey,
+ requestLoc.getUniqueId)
+ }
+ if (location == null) {
+ location = requestLoc
+ val writer = storageManager.createPartitionDataWriter(
+ applicationId,
+ shuffleId,
+ location,
+ splitThreshold,
+ splitMode,
+ partitionType,
+ rangeReadFilter,
+ userIdentifier,
+ partitionSplitEnabled,
+ isSegmentGranularityVisible)
+ new WorkingPartition(location, writer)
+ } else {
+ location
+ }
+ } catch {
+ case e: Exception =>
+ logError(s"Create FileWriter for $requestLoc $shuffleKey failed.", e)
+ throw e
+ }
+ }
+
+ private def destroyWriters(
+ partitionLocations: jList[PartitionLocation],
+ shuffleKey: String): Unit = {
+ partitionLocations.asScala.foreach { partitionLocation =>
+ val fileWriter = partitionLocation.asInstanceOf[WorkingPartition].getFileWriter
Review Comment:
@eolivelli, all locations including created by this loop and get from `partitionLocationInfo` should destroy the writers via the following code:
```
var location = partitionLocationInfo.getPrimaryLocation(shuffleKey, requestPrimaryLocs.get(ind).getUniqueId)
if (location == null) {
...
} else {
primaryLocs.add(location)
}
```
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time [celeborn]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #3387: [CELEBORN-2063] Parallelize the create partition writer in handleReserveSlots to speed up the reserveSlots RPC process time
URL: https://github.com/apache/celeborn/pull/3387
--
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@celeborn.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org