SteNicholas commented on code in PR #3784:
URL: https://github.com/apache/celeborn/pull/3784#discussion_r3845601589
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java:
##########
@@ -314,6 +327,174 @@ public FileInfo getSortedFileInfo(
}
}
+ public CompletableFuture<FileInfo> getSortedFileInfoAsync(
+ String shuffleKey, String fileName, FileInfo fileInfo, int
startMapIndex, int endMapIndex) {
+ if (shutdown) {
+ return failedSortedFileInfo(new IOException("Partition sorter is
closed."));
+ }
+ if (fileInfo instanceof MemoryFileInfo) {
+ logger.debug(
+ "Sorting memory shuffle file synchronously for shuffle key {}, file
name {}, map range [{}, {})",
+ shuffleKey,
+ fileName,
+ startMapIndex,
+ endMapIndex);
+ try {
+ return CompletableFuture.completedFuture(
+ getSortedFileInfo(shuffleKey, fileName, fileInfo, startMapIndex,
endMapIndex));
+ } catch (IOException e) {
+ return failedSortedFileInfo(e);
+ }
+ }
+
+ DiskFileInfo diskFileInfo = (DiskFileInfo) fileInfo;
+ String fileId = shuffleKey + "-" + fileName;
+ UserIdentifier userIdentifier = diskFileInfo.getUserIdentifier();
+ Set<String> sorted =
+ sortedShuffleFiles.computeIfAbsent(shuffleKey, ignored ->
ConcurrentHashMap.newKeySet());
+ Set<String> sorting =
+ sortingShuffleFiles.computeIfAbsent(shuffleKey, ignored ->
ConcurrentHashMap.newKeySet());
+ String sortedFilePath =
Utils.getSortedFilePath(diskFileInfo.getFilePath());
+ String indexFilePath = Utils.getIndexFilePath(diskFileInfo.getFilePath());
+
+ CompletableFuture<Void> sortCompletionFuture;
+ synchronized (sorting) {
+ if (shutdown) {
+ return failedSortedFileInfo(new IOException("Partition sorter is
closed."));
+ }
+ if (sorted.contains(fileId)) {
+ sortCompletionFuture = CompletableFuture.completedFuture(null);
+ } else {
+ try {
+ sortCompletionFuture =
+ sortCompletionFutures
+ .computeIfAbsent(shuffleKey, ignored ->
JavaUtils.newConcurrentHashMap())
+ .compute(
+ fileId,
+ (ignored, existing) ->
+ existing == null || existing.isDone()
+ ? createSortCompletionFuture(shuffleKey, fileId,
diskFileInfo)
Review Comment:
All waiters reuse this future and therefore share one timeout that starts
with the first waiter. A later request inherits only the remaining time,
potentially almost zero, while a request arriving just after this future times
out gets a fresh full timeout. The previous synchronous path applied the
timeout per request. Please share the raw sort-completion future but apply a
timeout to each returned waiter, or document and test the new global
sort-attempt timeout semantics.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
