You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ch...@apache.org on 2026/09/07 07:45:21 UTC
(spark) branch master updated: [SPARK-58985][CORE] Fix HistoryServerDiskManager double-counting store size on concurrent release and makeRoom
This is an automated email from the ASF dual-hosted git repository.
pan3793 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new ff9d3e357618 [SPARK-58985][CORE] Fix HistoryServerDiskManager double-counting store size on concurrent release and makeRoom
ff9d3e357618 is described below
commit ff9d3e3576186bb6cb9788f4b1156579f90f9587
Author: Cheng Pan <pa...@gmail.com>
AuthorDate: Mon Sep 7 15:45:08 2026 +0800
[SPARK-58985][CORE] Fix HistoryServerDiskManager double-counting store size on concurrent release and makeRoom
### What changes were proposed in this pull request?
Make the disk usage accounting in `HistoryServerDiskManager` atomic with the store operation it accompanies:
- `release()` now performs the whole operation -- removing the app from the `active` map, updating usage accounting, and deleting or re-measuring the store -- under the `active` lock. When the store directory is already gone, it also drops the listing entry so the store is not deducted again later.
- `Lease.commit()` now moves the store into place, updates usage accounting, writes the listing entry, and registers the app in the `active` map under the `active` lock, so the store is never on disk without being tracked. A failed rename raises an `IOException` instead of recording a store that is not there. When the previous store directory is already gone but still listed, its listing entry and accounting are dropped before the new store takes its place.
- `openStore()` now writes the listing entry under the `active` lock, so a concurrent `release()` cannot leave a listing entry for a deleted store.
- `makeRoom()` re-checks each eviction candidate under the `active` lock before deleting it: it skips candidates that became active or that a concurrent `release()` or `commit()` already deleted, and deducts the size the listing holds at that point. The summary log reports the stores actually deleted and the space actually freed, and warns when nothing could be freed. On that log line, `NUM_BYTES_EVICTED` now carries the freed bytes and `NUM_BYTES_TO_FREE` the target, and the store co [...]
The lock now covers directory I/O (`sizeOf`, rename, deletion, listing read/write). As a trade-off, `openStore()` from UI requests can block while a concurrent `release()` (e.g., in the `cleanLogs` loop) or `makeRoom()` eviction deletes a store; this is deliberate to keep the accounting accurate.
### Why are the changes needed?
`HistoryServerDiskManager` can deduct the same store size twice, driving the committed usage negative and making the History Server throw `IllegalStateException: Disk usage tracker went negative`.
The race is longstanding: `release()` updates usage and operates on the store directory outside the `active` lock, `commit()` moves the store into place before registering it, and `makeRoom()` deletes eviction candidates without re-checking, so two paths have been able to deduct the same store since the disk manager was introduced by SPARK-20654 (2.3.0).
SPARK-56044 (4.0.3) widened the race. By adding a deduction in `release()` based on the size measured from disk for apps not in the `active` map, a double deduction no longer requires the application to be actively open:
- `release(delete = true)` vs `openStore()`: `release()` deducts the measured size and deletes the store, but a concurrent `openStore()` re-registered the app in `active`, so a subsequent `release()` deducts the size again.
- `release(delete = true)` vs `commit()`: `commit()` has moved the store into place but not yet registered it, so `release()` deducts the measured size and deletes the store; `commit()` then registers the deleted store, and a subsequent `release()` deducts the size again.
- `release(delete = true)` vs `makeRoom()`: both paths deduct the size of the same store.
This makes the race reachable in normal History Server operation, e.g. log cleanup calling `release(delete = true)` for an app never opened after a restart while a concurrent UI request opens, loads, or evicts the same store. The fix also closes two concurrent `makeRoom()` calls evicting the same store twice.
This crash was observed on a production History Server (4.1-based build), in the periodic log cleanup path:
```
java.lang.IllegalStateException: Disk usage tracker went negative (now = -118595158, delta = -151974353)
at o.a.s.deploy.history.HistoryServerDiskManager.updateUsage(HistoryServerDiskManager.scala:285)
at o.a.s.deploy.history.HistoryServerDiskManager.release(HistoryServerDiskManager.scala:186)
at o.a.s.deploy.history.FsHistoryProvider.cleanAppData(FsHistoryProvider.scala:746)
at o.a.s.deploy.history.FsHistoryProvider.deleteAttemptLogs(FsHistoryProvider.scala:1132)
at o.a.s.deploy.history.FsHistoryProvider.cleanLogs(FsHistoryProvider.scala:1063)
at o.a.s.deploy.history.FsHistoryProvider.$anonfun$startPolling$4(FsHistoryProvider.scala:305)
```
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Added five tests to `HistoryServerDiskManagerSuite`. The three race tests (`release with delete is atomic with openStore`, `makeRoom is atomic with release`, `commit is atomic with release`) park one operation mid-way, run the competing operation on another thread, and let the parked one proceed once the competing thread has finished or blocked on the lock, so none relies on a fixed wait. Each fails on the pre-fix code, the first with `openStore()` handing out a store being deleted, t [...]
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Qwen 3.8 Max
Closes #58312 from pan3793/SPARK-58985.
Authored-by: Cheng Pan <pa...@gmail.com>
Signed-off-by: Cheng Pan <ch...@apache.org>
---
.../deploy/history/HistoryServerDiskManager.scala | 147 +++++++++++-------
.../history/HistoryServerDiskManagerSuite.scala | 167 ++++++++++++++++++++-
2 files changed, 255 insertions(+), 59 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/deploy/history/HistoryServerDiskManager.scala b/core/src/main/scala/org/apache/spark/deploy/history/HistoryServerDiskManager.scala
index 35cdd7880f23..0f29d19ecc84 100644
--- a/core/src/main/scala/org/apache/spark/deploy/history/HistoryServerDiskManager.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/history/HistoryServerDiskManager.scala
@@ -17,7 +17,7 @@
package org.apache.spark.deploy.history
-import java.io.File
+import java.io.{File, IOException}
import java.util.concurrent.atomic.AtomicLong
import scala.collection.mutable.{HashMap, ListBuffer}
@@ -67,6 +67,8 @@ private class HistoryServerDiskManager(
private val maxUsage = conf.get(MAX_LOCAL_DISK_USAGE)
private val currentUsage = new AtomicLong(0L)
private val committedUsage = new AtomicLong(0L)
+ // Besides the map itself, this lock guards the store directories, their listing entries and
+ // the committed usage: whether a store is still counted is decided and acted on in one hold.
private val active = new HashMap[(String, Option[String]), Long]()
def initialize(): Unit = {
@@ -139,60 +141,64 @@ private class HistoryServerDiskManager(
* being used so that it's not evicted when running out of designated space.
*/
def openStore(appId: String, attemptId: Option[String]): Option[File] = {
- var newSize: Long = 0
- val storePath = active.synchronized {
+ active.synchronized {
val path = appStorePath(appId, attemptId)
if (path.isDirectory()) {
- newSize = sizeOf(path)
+ val newSize = sizeOf(path)
active(appId -> attemptId) = newSize
+ updateApplicationStoreInfo(appId, attemptId, newSize)
Some(path)
} else {
None
}
}
-
- storePath.foreach { path =>
- updateApplicationStoreInfo(appId, attemptId, newSize)
- }
-
- storePath
}
/**
* Tell the disk manager that the store for the given application is not being used anymore.
+ * A tracked store whose directory is already gone is dropped from the listing regardless of
+ * `delete`.
*
* @param delete Whether to delete the store from disk.
*/
def release(appId: String, attemptId: Option[String], delete: Boolean = false): Unit = {
// Because disk-based stores may modify the structure of the store files even when just reading,
- // update the accounting for this application when it's closed.
- val oldSizeOpt = active.synchronized {
- active.remove(appId -> attemptId)
- }
-
- oldSizeOpt.foreach { oldSize =>
- updateUsage(-oldSize, committed = true)
- }
+ // update the accounting for this application when it's closed. Do the whole operation under
+ // the active lock so a concurrent openStore(), commit() or makeRoom() cannot deduct the same
+ // store size twice.
+ active.synchronized {
+ val oldSizeOpt = active.remove(appId -> attemptId)
+
+ oldSizeOpt.foreach { oldSize =>
+ updateUsage(-oldSize, committed = true)
+ }
- // Apply the store operation regardless of whether the app was in the active map, since
- // the store directory may still exist on disk (e.g., when the app was never opened after
- // a restart).
- val path = appStorePath(appId, attemptId)
- if (path.isDirectory()) {
- if (delete) {
- // If the app was not actively tracked, its size was not deducted above; do it now.
- if (oldSizeOpt.isEmpty) {
- val size = sizeOf(path)
- updateUsage(-size, committed = true)
+ // Apply the store operation regardless of whether the app was in the active map, since
+ // the store directory may still exist on disk (e.g., when the app was never opened after
+ // a restart).
+ val path = appStorePath(appId, attemptId)
+ if (path.isDirectory()) {
+ if (delete) {
+ // If the app was not actively tracked, its size was not deducted above; do it now.
+ // Measure it from disk rather than trusting the listing: a store directory without a
+ // listing entry is still counted, since initialize() measures the whole directory.
+ if (oldSizeOpt.isEmpty) {
+ val size = sizeOf(path)
+ updateUsage(-size, committed = true)
+ }
+ deleteStore(path)
+ } else if (oldSizeOpt.isDefined) {
+ // Re-measure the size since the store may have changed while it was open.
+ val newSize = sizeOf(path)
+ val newInfo = listing.read(classOf[ApplicationStoreInfo], path.getAbsolutePath())
+ .copy(size = newSize)
+ listing.write(newInfo)
+ updateUsage(newSize, committed = true)
}
- deleteStore(path)
} else if (oldSizeOpt.isDefined) {
- // Re-measure the size since the store may have changed while it was open.
- val newSize = sizeOf(path)
- val newInfo = listing.read(classOf[ApplicationStoreInfo], path.getAbsolutePath())
- .copy(size = newSize)
- listing.write(newInfo)
- updateUsage(newSize, committed = true)
+ // The store directory is gone (e.g., deleted out of band) and its size was deducted
+ // above; drop its listing entry too, so makeRoom() does not deduct it again.
+ deleteStore(path)
}
}
}
@@ -226,6 +232,15 @@ private class HistoryServerDiskManager(
listing.delete(classOf[ApplicationStoreInfo], path.getAbsolutePath())
}
+ /** Returns the listing entry of the store at `path`, if any. */
+ private def readStoreInfo(path: String): Option[ApplicationStoreInfo] = {
+ try {
+ Some(listing.read(classOf[ApplicationStoreInfo], path))
+ } catch {
+ case _: NoSuchElementException => None
+ }
+ }
+
private def makeRoom(size: Long): Unit = {
if (free() < size) {
logDebug(s"Not enough free space, looking at candidates for deletion...")
@@ -246,18 +261,30 @@ private class HistoryServerDiskManager(
}
}
- if (evicted.nonEmpty) {
- val freed = evicted.map { info =>
- logInfo(log"Deleting store for" +
- log" ${MDC(APP_ID, info.appId)}/${MDC(APP_ATTEMPT_ID, info.attemptId)}.")
- deleteStore(new File(info.path))
- updateUsage(-info.size, committed = true)
- info.size
- }.sum
-
- logInfo(log"Deleted ${MDC(NUM_BYTES_EVICTED, evicted.size)} store(s)" +
- log" to free ${MDC(NUM_BYTES_TO_FREE, Utils.bytesToString(freed))}" +
- log" (target = ${MDC(NUM_BYTES, Utils.bytesToString(size))}).")
+ var freedCount = 0
+ var freedBytes = 0L
+ evicted.foreach { candidate =>
+ active.synchronized {
+ // Re-check under the lock: since the candidate was collected, it may have become
+ // active, or a concurrent release() or commit() may have deleted it along with its
+ // listing entry. Deduct the size the listing holds now, which may have changed as well.
+ if (!active.contains(candidate.appId -> candidate.attemptId)) {
+ readStoreInfo(candidate.path).foreach { info =>
+ logInfo(log"Deleting store for" +
+ log" ${MDC(APP_ID, info.appId)}/${MDC(APP_ATTEMPT_ID, info.attemptId)}.")
+ deleteStore(new File(info.path))
+ updateUsage(-info.size, committed = true)
+ freedCount += 1
+ freedBytes += info.size
+ }
+ }
+ }
+ }
+
+ if (freedCount > 0) {
+ logInfo(log"Deleted ${MDC(NUM_APPS, freedCount)} store(s)" +
+ log" to free ${MDC(NUM_BYTES_EVICTED, Utils.bytesToString(freedBytes))}" +
+ log" (target = ${MDC(NUM_BYTES_TO_FREE, Utils.bytesToString(size))}).")
} else {
logWarning(log"Unable to free any space to make room for " +
log"${MDC(NUM_BYTES, Utils.bytesToString(size))}.")
@@ -320,6 +347,13 @@ private class HistoryServerDiskManager(
val size = sizeOf(dst)
deleteStore(dst)
updateUsage(-size, committed = true)
+ } else {
+ // The store directory is gone (e.g., deleted out of band) but may still be listed and
+ // counted; drop it before the new store takes over its listing entry.
+ readStoreInfo(dst.getAbsolutePath()).foreach { info =>
+ deleteStore(dst)
+ updateUsage(-info.size, committed = true)
+ }
}
}
@@ -327,9 +361,18 @@ private class HistoryServerDiskManager(
val newSize = sizeOf(tmpPath)
makeRoom(newSize)
- tmpPath.renameTo(dst)
- updateUsage(newSize, committed = true)
+ // Move the store into place, account for it and mark the app active in one step, so a
+ // concurrent release() or makeRoom() cannot delete the store before it is tracked.
+ active.synchronized {
+ if (!tmpPath.renameTo(dst)) {
+ throw new IOException(s"Failed to move the store from $tmpPath to $dst")
+ }
+ updateUsage(newSize, committed = true)
+ updateApplicationStoreInfo(appId, attemptId, newSize)
+ active(appId -> attemptId) = newSize
+ }
+
if (committedUsage.get() > maxUsage) {
val current = Utils.bytesToString(committedUsage.get())
val max = Utils.bytesToString(maxUsage)
@@ -337,12 +380,6 @@ private class HistoryServerDiskManager(
log"${MDC(APP_ATTEMPT_ID, attemptId)} causes maximum disk usage to be " +
log"exceeded (${MDC(NUM_BYTES, current)} > ${MDC(NUM_BYTES_MAX, max)}")
}
-
- updateApplicationStoreInfo(appId, attemptId, newSize)
-
- active.synchronized {
- active(appId -> attemptId) = newSize
- }
dst
}
diff --git a/core/src/test/scala/org/apache/spark/deploy/history/HistoryServerDiskManagerSuite.scala b/core/src/test/scala/org/apache/spark/deploy/history/HistoryServerDiskManagerSuite.scala
index c82a0e4321d3..f9ab2de28c59 100644
--- a/core/src/test/scala/org/apache/spark/deploy/history/HistoryServerDiskManagerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/deploy/history/HistoryServerDiskManagerSuite.scala
@@ -18,21 +18,26 @@
package org.apache.spark.deploy.history
import java.io.File
+import java.util.concurrent.{CountDownLatch, FutureTask, TimeUnit}
+
+import scala.concurrent.duration._
import org.mockito.AdditionalAnswers
import org.mockito.ArgumentMatchers.{anyBoolean, anyLong, eq => meq}
import org.mockito.Mockito.{doAnswer, spy}
import org.scalatest.BeforeAndAfter
+import org.scalatest.concurrent.Eventually
import org.apache.spark.{SparkConf, SparkFunSuite}
import org.apache.spark.internal.config.History._
import org.apache.spark.internal.config.History.HybridStoreDiskBackend
import org.apache.spark.status.KVUtils
import org.apache.spark.tags.ExtendedLevelDBTest
-import org.apache.spark.util.{ManualClock, Utils}
+import org.apache.spark.util.{Clock, ManualClock, Utils}
import org.apache.spark.util.kvstore.KVStore
-abstract class HistoryServerDiskManagerSuite extends SparkFunSuite with BeforeAndAfter {
+abstract class HistoryServerDiskManagerSuite extends SparkFunSuite with BeforeAndAfter
+ with Eventually {
protected def backend: HybridStoreDiskBackend.Value
@@ -60,15 +65,60 @@ abstract class HistoryServerDiskManagerSuite extends SparkFunSuite with BeforeAn
}
}
- private def mockManager(): HistoryServerDiskManager = {
+ private def mockManager(clock: Clock = new ManualClock()): HistoryServerDiskManager = {
val conf = new SparkConf().set(MAX_LOCAL_DISK_USAGE, MAX_USAGE)
val manager = spy[HistoryServerDiskManager](
- new HistoryServerDiskManager(conf, testDir, store, new ManualClock()))
+ new HistoryServerDiskManager(conf, testDir, store, clock))
doAnswer(AdditionalAnswers.returnsFirstArg[Long]()).when(manager)
.approximateSize(anyLong(), anyBoolean())
manager
}
+ /**
+ * Waits until `thread` is blocked on the lock another thread holds, or has finished, which
+ * lets a regression that no longer takes the lock fail fast instead of timing out.
+ */
+ private def awaitBlockedOrDone(thread: Thread): Unit = {
+ eventually(timeout(10.seconds), interval(10.millis)) {
+ assert(Set(Thread.State.BLOCKED, Thread.State.TERMINATED).contains(thread.getState))
+ }
+ }
+
+ /** Commits a store of size 2 for app1 and releases it, so it stays on disk and evictable. */
+ private def commitIdleStore(manager: HistoryServerDiskManager): File = {
+ val lease = manager.lease(2)
+ doReturn(2L).when(manager).sizeOf(meq(lease.tmpPath))
+ val dst = lease.commit("app1", None)
+ doReturn(2L).when(manager).sizeOf(meq(dst))
+ manager.release("app1", None)
+ assert(manager.committed() === 2)
+ dst
+ }
+
+ private case class ParkedRelease(thread: Thread, task: FutureTask[Unit], resume: CountDownLatch)
+
+ /**
+ * Runs release(delete = true) for app1 on its own thread, and returns once it is parked inside
+ * sizeOf(dst), i.e. after it decided to deduct the store but before it deleted it.
+ */
+ private def parkReleaseInSizeOf(manager: HistoryServerDiskManager, dst: File): ParkedRelease = {
+ val parked = new CountDownLatch(1)
+ val resume = new CountDownLatch(1)
+ val task = new FutureTask[Unit](() => manager.release("app1", None, delete = true))
+ val thread = new Thread(task)
+ doAnswer(_ => {
+ // Only park the release thread; on the pre-fix code, openStore() also measures the store.
+ if (Thread.currentThread() eq thread) {
+ parked.countDown()
+ resume.await(10, TimeUnit.SECONDS)
+ }
+ 2L
+ }).when(manager).sizeOf(meq(dst))
+ thread.start()
+ assert(parked.await(10, TimeUnit.SECONDS))
+ ParkedRelease(thread, task, resume)
+ }
+
test("leasing space") {
val manager = mockManager()
@@ -252,6 +302,115 @@ abstract class HistoryServerDiskManagerSuite extends SparkFunSuite with BeforeAn
assert(manager2.free() === MAX_USAGE)
}
+ test("SPARK-58985: release with delete is atomic with openStore") {
+ val manager = mockManager()
+ val dst = commitIdleStore(manager)
+
+ val release = parkReleaseInSizeOf(manager, dst)
+ val openStoreTask = new FutureTask[Option[File]](() => manager.openStore("app1", None))
+ val openStoreThread = new Thread(openStoreTask)
+ try {
+ openStoreThread.start()
+ awaitBlockedOrDone(openStoreThread)
+ } finally {
+ release.resume.countDown()
+ }
+ release.task.get(10, TimeUnit.SECONDS)
+
+ // release() holds the lock until the store is deleted, so openStore() must not hand out the
+ // path, and the size is deducted exactly once.
+ assert(openStoreTask.get(10, TimeUnit.SECONDS).isEmpty)
+ assert(manager.committed() === 0)
+ assert(!dst.exists())
+ }
+
+ test("SPARK-58985: makeRoom is atomic with release") {
+ val manager = mockManager()
+ val dst = commitIdleStore(manager)
+
+ // Run a lease() that has to evict the store while release() is deleting it.
+ val release = parkReleaseInSizeOf(manager, dst)
+ val leaseTask = new FutureTask[Unit](() => manager.lease(2))
+ val leaseThread = new Thread(leaseTask)
+ try {
+ leaseThread.start()
+ awaitBlockedOrDone(leaseThread)
+ } finally {
+ release.resume.countDown()
+ }
+ release.task.get(10, TimeUnit.SECONDS)
+ leaseTask.get(10, TimeUnit.SECONDS)
+
+ // release() deleted the store first, so makeRoom() found nothing left to deduct.
+ assert(manager.committed() === 0)
+ assert(!dst.exists())
+ assert(manager.free() === 1)
+ }
+
+ test("SPARK-58985: makeRoom deducts a store deleted out of band") {
+ val manager = mockManager()
+ val dst = commitIdleStore(manager)
+ Utils.deleteRecursively(dst)
+
+ // The store is still counted and listed, so evicting it must deduct its size.
+ manager.lease(2)
+ assert(manager.committed() === 0)
+ assert(!store.view(classOf[ApplicationStoreInfo]).iterator().hasNext)
+ }
+
+ test("SPARK-58985: commit deducts a store deleted out of band") {
+ val manager = mockManager()
+ val dst = commitIdleStore(manager)
+ Utils.deleteRecursively(dst)
+
+ // A lease that fits without evicting anything, so commit() meets the leftover entry itself.
+ val lease = manager.lease(1)
+ doReturn(1L).when(manager).sizeOf(meq(lease.tmpPath))
+ assert(lease.commit("app1", None) === dst)
+ assert(manager.committed() === 1)
+ assert(store.read(classOf[ApplicationStoreInfo], dst.getAbsolutePath).size === 1)
+ }
+
+ test("SPARK-58985: commit is atomic with release") {
+ // updateApplicationStoreInfo() reads the clock after commit() has moved the store into
+ // place; park commit() there and run release() concurrently.
+ val inCommit = new CountDownLatch(1)
+ val resumeCommit = new CountDownLatch(1)
+ val manager = mockManager(new ManualClock() {
+ override def getTimeMillis(): Long = {
+ inCommit.countDown()
+ resumeCommit.await(10, TimeUnit.SECONDS)
+ super.getTimeMillis()
+ }
+ })
+ val lease = manager.lease(2)
+ doReturn(2L).when(manager).sizeOf(meq(lease.tmpPath))
+ val dst = manager.appStorePath("app1", None)
+ doReturn(2L).when(manager).sizeOf(meq(dst))
+
+ val commitTask = new FutureTask[File](() => lease.commit("app1", None))
+ val commitThread = new Thread(commitTask)
+ val releaseTask = new FutureTask[Unit](() => manager.release("app1", None, delete = true))
+ val releaseThread = new Thread(releaseTask)
+ try {
+ commitThread.start()
+ assert(inCommit.await(10, TimeUnit.SECONDS))
+ releaseThread.start()
+ awaitBlockedOrDone(releaseThread)
+ } finally {
+ resumeCommit.countDown()
+ }
+ commitTask.get(10, TimeUnit.SECONDS)
+ releaseTask.get(10, TimeUnit.SECONDS)
+
+ // release() cannot see the store until commit() has registered it, so it deducts the size
+ // exactly once; the provider's own release of the committed store then finds nothing left.
+ assert(manager.committed() === 0)
+ assert(!dst.exists())
+ manager.release("app1", None, delete = true)
+ assert(manager.committed() === 0)
+ }
+
test("SPARK-38095: appStorePath should use backend extensions") {
val conf = new SparkConf().set(HYBRID_STORE_DISK_BACKEND, backend.toString)
val manager = new HistoryServerDiskManager(conf, testDir, store, new ManualClock())
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org