LuciferYang commented on code in PR #59009:
URL: https://github.com/apache/spark/pull/59009#discussion_r4104095287
##########
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:
Thanks — hardening is the right framing, and marking the user-facing change
as No, with the note that no in-tree caller catches the error, resolves this
from my side.
##########
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 on both counts: the narrowed contract (concurrent retry only after
`reset()` has returned exceptionally) covers the concern, and keeping the flag
after page cleanup is the safer choice given a mid-cleanup `freePage()`
failure. Withdrawing the move suggestion.
--
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]
