sunchao commented on code in PR #3784:
URL: https://github.com/apache/celeborn/pull/3784#discussion_r3846214488
##########
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)
+ : existing);
+ } catch (RejectedExecutionException e) {
+ return failedSortedFileInfo(new IOException("Partition sorter is
closed.", e));
+ }
+ if (!sorting.contains(fileId)) {
+ try {
+ FileSorter fileSorter = new FileSorter(diskFileInfo, fileId,
shuffleKey);
+ sorting.add(fileId);
+ logger.debug(
+ "Adding sorter to sort queue shuffle key {}, file name {}",
shuffleKey, fileName);
+ shuffleSortTaskDeque.put(fileSorter);
+ } catch (InterruptedException e) {
+ logger.error(
+ "Sorter scheduler thread is interrupted means worker is
shutting down.", e);
+ sorting.remove(fileId);
+ sortCompletionFuture.completeExceptionally(
+ new IOException(
+ "Sort scheduler thread is interrupted means worker is
shutting down.", e));
+ } catch (IOException e) {
+ logger.error("File sorter access DFS failed.", e);
+ sortCompletionFuture.completeExceptionally(
+ new IOException("File sorter access DFS failed.", e));
+ }
+ }
+ }
+ }
+
+ if (shutdown) {
+ sortCompletionFuture.completeExceptionally(new IOException("Partition
sorter is closed."));
+ }
+ sortedFileWaiterCount.incrementAndGet();
+ return sortCompletionFuture
Review Comment:
Fixed in 6f8522f. Pending requests are now tracked through index resolution,
and cache publication and stream registration are guarded by the request's
original shuffle lifecycle. Worker cleanup expires fetch admissions before
retiring sorter state, so delayed file lookups cannot admit work after cleanup.
Added regression coverage for cleanup while index resolution is queued or
paused after parsing, and while the initial file lookup is blocked. After
releasing and draining the work, the requests fail without recreating cache
entries or streams.
##########
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)
+ : existing);
+ } catch (RejectedExecutionException e) {
+ return failedSortedFileInfo(new IOException("Partition sorter is
closed.", e));
+ }
+ if (!sorting.contains(fileId)) {
+ try {
+ FileSorter fileSorter = new FileSorter(diskFileInfo, fileId,
shuffleKey);
Review Comment:
Fixed in 6f8522f. `FileSorter` construction now only records metadata;
local/DFS existence checks and deletion of leftover output files run in
`prepareFiles()` on the sorter executor.
`testDfsSorterPreparationDoesNotBlockReceiveCaller` uses the real fetch
handler and sorter, blocks DFS `exists`, and verifies that `receive` and
subsequent work on the request executor finish before the metadata operation is
released.
##########
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:
Fixed in 6f8522f by restoring per-request timeouts. Only the underlying sort
completion is shared; each waiter has its own deadline/cancellation signal. A
late reader gets its full wait budget, and another reader timing out does not
cancel its wait or the sort.
`testAsyncSortedFileWaitersHaveIndependentTimeouts` explicitly fires the
first waiter's timer, verifies the later waiter and retry remain pending, and
then lets both succeed from one physical sort. It also covers cancellation
isolation and repeated abandoned readers without retained sort-future callbacks.
--
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]
