cloud-fan commented on code in PR #59009:
URL: https://github.com/apache/spark/pull/59009#discussion_r4106947029
##########
core/src/main/java/org/apache/spark/unsafe/map/BytesToBytesMap.java:
##########
@@ -1109,30 +1173,55 @@ 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
Review Comment:
Addressed in 60eea7415ee. Failed reset no longer self-heals through lookup.
The map remains empty and unusable until the caller explicitly retries reset(),
so catching the reset OOM and continuing with lookup fails fast instead of
silently discarding sorter-held data.
##########
core/src/main/java/org/apache/spark/unsafe/map/BytesToBytesMap.java:
##########
@@ -612,9 +668,14 @@ 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. After {@link
#reset()} has completed
+ * exceptionally because its pointer-array allocation failed, concurrent
calls may safely retry
+ * that allocation.
+ *
+ * @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:
Addressed in 60eea7415ee. lookup() and safeLookup() no longer allocate
during recovery or throw SparkOutOfMemoryError from that recovery path. They
perform a state check and fail with IllegalStateException after a failed reset;
only reset() allocates.
##########
core/src/main/java/org/apache/spark/unsafe/map/BytesToBytesMap.java:
##########
@@ -587,12 +604,50 @@ public Location lookup(Object keyBase, long keyOffset,
int keyLength, int hash)
return loc;
}
+ /**
+ * Restores the hash array after a {@link #reset()} freed it but its eager
reallocation failed
+ * with an OOM, leaving {@code longArray == null}. A reset() empties the
map, so re-allocating at
+ * {@code initialCapacity} here is correct. If memory is still unavailable
this throws a
+ * SparkOutOfMemoryError instead of exposing the null array.
+ */
+ private void restoreArrayAfterFailedReset() {
+ if (freed) {
+ throw new IllegalStateException("BytesToBytesMap has already been
freed");
+ }
+ if (longArrayRecoveryRequired) {
+ // Only the failed-reset state is recoverable here: the map is empty, so
a fresh
+ // initial-capacity array is correct. `destructiveIterator != null`
means destructive
+ // iteration has begun, after which map operations are illegal.
+ //
+ // allocate() acquires the TaskMemoryManager monitor while this monitor
is held. The memory
+ // manager may call spill() while holding its monitor, so spill() and
anything reachable
+ // from it must not acquire this monitor.
+ synchronized (this) {
+ if (freed) {
+ throw new IllegalStateException("BytesToBytesMap has already been
freed");
+ }
+ if (destructiveIterator != null) {
Review Comment:
Addressed in 60eea7415ee. A single lifecycle state is checked by all
array-dependent operations regardless of history. Lookups, accessors, and
iterators after destructive iteration now consistently throw
IllegalStateException, and operationsAfterDestructiveIterationFail covers the
normal destructive path.
##########
core/src/main/java/org/apache/spark/unsafe/map/BytesToBytesMap.java:
##########
@@ -1109,30 +1173,55 @@ 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
+ * @throws IllegalStateException if destructive iteration has begun
*/
- public void reset() {
+ public synchronized void reset() {
+ if (freed) {
Review Comment:
Addressed in 60eea7415ee. reset() after free() now throws
IllegalStateException, as does reset() after destructive iteration.
operationsAfterFreeFail covers reset and the array-dependent accessors.
##########
core/src/main/java/org/apache/spark/unsafe/map/BytesToBytesMap.java:
##########
@@ -1109,30 +1173,55 @@ 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
+ * @throws IllegalStateException if destructive iteration has begun
*/
- public void reset() {
+ public synchronized void reset() {
Review Comment:
I used the explicit-reset alternative in 60eea7415ee. Rather than retaining
and zeroing a potentially large old pointer array, which would work against
reset releasing memory, reset frees the old storage and enters RESET_FAILED if
replacement allocation fails. Only a later explicit reset can recover. This
also removes all lazy-recovery machinery.
--
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]
