cloud-fan commented on code in PR #59009:
URL: https://github.com/apache/spark/pull/59009#discussion_r4103085396
##########
core/src/main/java/org/apache/spark/unsafe/map/BytesToBytesMap.java:
##########
@@ -1109,30 +1171,50 @@ public int getNumDataPages() {
}
/**
- * Returns the underline long[] of longArray.
+ * Returns the underlying long array. If the pointer-array allocation in a
previous
+ * {@link #reset()} failed, this method retries that allocation before
returning.
+ *
+ * @throws SparkOutOfMemoryError if the pointer array still cannot be
allocated
*/
public LongArray getArray() {
+ restoreArrayAfterFailedReset();
assert(longArray != null);
return longArray;
}
/**
- * Reset this map to initialized state.
+ * Reset this map to initialized state. If {@link #free()} has been called,
the map cannot be
+ * reinitialized and this method is a no-op. If the replacement pointer
array cannot be
+ * allocated, the map remains empty and the next lookup, {@link
#getArray()}, or
+ * {@link #maxNumKeysIndex()} retries the allocation.
+ *
+ * @throws SparkOutOfMemoryError if the replacement pointer array cannot be
allocated
*/
- public void reset() {
+ public synchronized void reset() {
+ if (freed) {
+ return;
+ }
Review Comment:
Addressed in 64eaea7f1b9. reset() now throws IllegalStateException once
destructive iteration has begun, before any reset bookkeeping or allocation.
resetAfterDestructiveIterationFails pins the terminal state. Both map suites
pass: 28 tests each, 56 total.
##########
core/src/main/java/org/apache/spark/unsafe/map/BytesToBytesMap.java:
##########
@@ -1109,30 +1171,50 @@ public int getNumDataPages() {
}
/**
- * Returns the underline long[] of longArray.
+ * Returns the underlying long array. If the pointer-array allocation in a
previous
+ * {@link #reset()} failed, this method retries that allocation before
returning.
+ *
+ * @throws SparkOutOfMemoryError if the pointer array still cannot be
allocated
*/
public LongArray getArray() {
+ restoreArrayAfterFailedReset();
assert(longArray != null);
return longArray;
}
/**
- * Reset this map to initialized state.
+ * Reset this map to initialized state. If {@link #free()} has been called,
the map cannot be
+ * reinitialized and this method is a no-op. If the replacement pointer
array cannot be
+ * allocated, the map remains empty and the next lookup, {@link
#getArray()}, or
+ * {@link #maxNumKeysIndex()} retries the allocation.
+ *
+ * @throws SparkOutOfMemoryError if the replacement pointer array cannot be
allocated
*/
- public void reset() {
+ public synchronized void reset() {
+ if (freed) {
+ return;
+ }
updatePeakMemoryUsed();
+ // Put the map into its empty state up front so that if the allocate()
below fails with an OOM,
+ // the map is left consistently empty (longArray == null, no stale
currentPage) rather than
+ // half-reset. A subsequent lookup then restores the array lazily
(restoreArrayAfterFailedReset)
+ // instead of dereferencing a null longArray or writing into a freed
currentPage.
numKeys = 0;
numValues = 0;
- freeArray(longArray);
+ canGrowArray = true;
+ currentPage = null;
+ pageCursor = 0;
+ if (longArray != null) {
+ freeArray(longArray);
+ }
longArray = null;
while (dataPages.size() > 0) {
MemoryBlock dataPage = dataPages.removeLast();
freePage(dataPage);
}
+ longArrayRecoveryRequired = true;
Review Comment:
Agreed that there is no current in-tree caller combining reset() with
concurrent safeLookup(). The intent is to preserve safeLookup's thread-safe
contract for callers that catch a failed reset. In 64eaea7f1b9 I clarified that
concurrent retry is supported only after reset() has completed exceptionally;
overlap with reset() remains prohibited. I kept the flag after page cleanup
because moving it earlier could make a partially cleaned map recoverable if
freePage() unexpectedly failed, allowing stale pages to remain visible.
##########
core/src/main/java/org/apache/spark/unsafe/map/BytesToBytesMap.java:
##########
@@ -612,9 +667,13 @@ public void safeLookup(Object keyBase, long keyOffset, int
keyLength, Location l
*
* The provided hash is ignored when this map has configured key operations.
Each thread must
* supply its own {@link Location}. Probe statistics may be inaccurate under
concurrent lookup,
- * and the map must not be modified concurrently.
+ * and the map must not otherwise be modified concurrently. Concurrent calls
may safely retry a
+ * pointer-array allocation that failed during {@link #reset()}.
+ *
+ * @throws SparkOutOfMemoryError if a pointer-array allocation retried after
reset still fails
*/
public void safeLookup(Object keyBase, long keyOffset, int keyLength,
Location loc, int hash) {
+ restoreArrayAfterFailedReset();
Review Comment:
You did not miss a caller. This is intended as hardening of
BytesToBytesMap's contract for callers that recover; current in-tree hash
aggregation still propagates SparkOutOfMemoryError from reset(). The PR
description now says there is no in-tree production caller that catches it and
marks the user-facing change as No.
##########
core/src/main/java/org/apache/spark/unsafe/map/BytesToBytesMap.java:
##########
@@ -140,6 +140,18 @@ public interface KeyOperationsFactory {
* while position {@code 2 * i + 1} in the array holds key's full 32-bit
hashcode.
*/
@Nullable private LongArray longArray;
+
+ /**
+ * Whether a completed reset needs to restore longArray after its allocation
failed.
+ *
+ * Volatile so that clearing this flag safely publishes the replacement
array to concurrent
+ * safeLookup callers.
+ */
+ private volatile boolean longArrayRecoveryRequired = false;
+
+ /** Whether {@link #free()} has permanently released this map. */
+ private volatile boolean freed = false;
Review Comment:
Addressed in 64eaea7f1b9. The longArray TODO is directly below the field
again, and the two new state flags now follow the complete TODO block.
--
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]
