ajsquared commented on code in PR #5627:
URL: https://github.com/apache/datafusion-comet/pull/5627#discussion_r3908135122
##########
spark/src/main/java/org/apache/comet/shuffle/CelebornShufflePartitionPusher.java:
##########
@@ -763,55 +814,54 @@ private void reconcileAcceptedPushes() {
ArrayList<PushReservation> completed = new ArrayList<>();
IOException detectedFailure = null;
synchronized (lifecycleLock) {
+ Iterator<PushReservation> pending = pendingPushes.iterator();
+ while (pending.hasNext()) {
+ PushReservation reservation = pending.next();
+ if (!reservation.submitted) {
+ continue;
+ }
+ registerPushStateFallback(reservation);
+ if (!reservation.pushStateFallback &&
reservation.transportPush.isComplete()) {
+ pending.remove();
+ completed.add(reservation);
+ }
+ }
for (ObservedPushState observed : observedPushStates.values()) {
Object failure = observed.exception.get();
boolean cleanedUp =
failure instanceof IOException
&& "Cleaned Up".equals(((IOException) failure).getMessage());
if (failure instanceof IOException && !cleanedUp) {
IOException observedFailure = (IOException) failure;
- // Only the raw transport callback's pinned message proves request
termination.
- // PushState can also publish interruption/backpressure exceptions
while an older
- // request is still live; those failures must never free that
request's admission.
- if (!observed.terminalFailureCredited &&
isTerminalRawPushFailure(observedFailure)) {
- observed.terminalFailureCredited = true;
- }
if (asynchronousFailure == null) {
asynchronousFailure = observedFailure;
detectedFailure = asynchronousFailure;
}
}
- if (transportCallbacks != null) {
- // Stock counters can remain positive after cancellation. The
transport path instead
- // tracks each request through the raw call, transport callbacks,
and queued retries.
+ if (hasPendingTransportOwnership(observed)) {
+ // A stock counter cannot distinguish a fallback request from a
transport-owned request.
+ // Wait for precise requests on this push state before crediting
fallback completions.
continue;
}
- long retainedRequests =
- Math.max(
- 0L, observed.inFlightRequests.sum() -
(observed.terminalFailureCredited ? 1L : 0L));
- long completions = Math.max(0L, observed.submittedPushes -
retainedRequests);
- while (observed.releasedPushes < completions) {
- PushReservation reservation = smallestSubmittedReservation(observed);
- if (reservation == null) {
+ boolean terminalFailure =
+ failure instanceof IOException
+ && !cleanedUp
+ && isTerminalRawPushFailure((IOException) failure);
+ long counterCredits = terminalFailure ? 1L : 0L;
+ long retainedRequests = Math.max(0L, observed.inFlightRequests.sum() -
counterCredits);
+ long completions = Math.max(0L, observed.fallbackSubmittedPushes -
retainedRequests);
Review Comment:
[P2][non-blocking] Recover admission after a cancelled fallback request fails
If bootstrap instrumentation fails, a subsequent map abort followed by
transport failure leaves this fallback reservation charged permanently. In
stock Celeborn 0.6.3, `cleanup()` sets `exception` to `Cleaned Up` without
resetting `totalInflightReqs`, and the raw-push failure callback returns
immediately when that exception is set, without removing the batch
([callback](https://github.com/apache/celeborn/blob/v0.6.3/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java#L1228-L1234),
[cleanup](https://github.com/apache/celeborn/blob/v0.6.3/client/src/main/java/org/apache/celeborn/client/ShuffleClientImpl.java#L1813-L1819)).
With one pending push, both `fallbackSubmittedPushes` and `retainedRequests`
remain 1, so this loop never releases its permits, even after all retained
transport ownership completes. Enough such reservations can stall later maps
sharing the client. Please retain a safe terminal-completion path for cancelled
fallback pushes; the existing `failed
transport after cleanup releases only its completed shared admission` case
could cover this with bootstrap tracking disabled.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
