dongjoon-hyun commented on PR #58312:
URL: https://github.com/apache/spark/pull/58312#issuecomment-5457930929
Thanks for tracking this down. The direction looks right to me: putting the
whole of `release()` under the `active` lock and re-checking each eviction
candidate under the same lock is consistent with what `Lease.commit` already
does, and cleaning up the stale listing entry in `makeRoom()` is a nice
addition. I ran the new test locally and confirmed it reproduces the reported
failure on the pre-fix code:
```
- SPARK-58985: release with delete is atomic with openStore *** FAILED ***
(2 seconds, 567 ms)
java.lang.IllegalStateException: Disk usage tracker went negative (now =
-2, delta = -2)
```
Three comments.
### 1. The new test no longer exercises the interleaving it describes
With the fix in place, the latch handshake cannot work: `release()` holds
the `active` lock while it is blocked inside the mocked `sizeOf()`, so the main
thread's `openStore()` blocks on that same lock and never reaches
`openStoreDone.countDown()` in the `finally`. The release thread only proceeds
when `openStoreDone.await(2, TimeUnit.SECONDS)` **times out**.
The runtimes show this — 2.588 s with the fix, 2.567 s without; the 2
seconds are pure timeout wait.
As a result `release()` always completes first and deletes the store, so
`openStore()` always returns `None`, and this block plus its comment are dead
code after the fix:
```scala
// If openStore() handed out a path, the subsequent release in
FsHistoryProvider must not
// deduct the size again.
opened.foreach { _ =>
manager.release("app1", None, delete = true)
}
```
It is still a valid regression test, but it would read much better if it
asserted the post-fix behavior explicitly (`release()` wins the lock, therefore
`openStore()` returns `None`). Running `openStore()` on its own thread and
counting the latch down from there would also let both suites finish without
burning ~2.5 s each on a timeout.
### 2. `Lease.commit()` vs `release(delete = true)` leaves the same crash
reachable
`commit()` does `tmpPath.renameTo(dst)` and `updateUsage(newSize, committed
= true)` outside the lock, and only registers the app in `active` in a later
`synchronized` block. In that window the store is on disk but not in `active` —
exactly the shape this PR fixes for `openStore()`:
```
commit: rename tmp -> dst; committed += newSize
release: (lock) not active -> deducts sizeOf(dst), deleteStore(dst)
commit: active(app) = newSize // registers a store that is
already gone
release: (lock) deducts oldSize again -> committed goes negative
```
I verified this with a scratch test that blocks inside `commit()` between
the rename and the `active` insertion (hooking `Clock.getTimeMillis()`, which
`updateApplicationStoreInfo` calls in that window). It fails identically on
this branch and on `master`:
```
- commit vs release(delete = true) *** FAILED ***
java.lang.IllegalStateException: Disk usage tracker went negative (now =
-2, delta = -2)
at ...HistoryServerDiskManager.release(HistoryServerDiskManager.scala:195)
```
<details>
<summary>scratch test</summary>
```scala
test("commit vs release(delete = true)") {
val conf2 = new SparkConf().set(MAX_LOCAL_DISK_USAGE, MAX_USAGE)
val armed = new java.util.concurrent.atomic.AtomicBoolean(false)
val inCommit = new CountDownLatch(1)
val releaseDone = new CountDownLatch(1)
val clock = new ManualClock() {
override def getTimeMillis(): Long = {
if (armed.get()) {
inCommit.countDown()
releaseDone.await(10, TimeUnit.SECONDS)
}
super.getTimeMillis()
}
}
val manager = spy[HistoryServerDiskManager](
new HistoryServerDiskManager(conf2, testDir, store, clock))
doAnswer(AdditionalAnswers.returnsFirstArg[Long]()).when(manager)
.approximateSize(anyLong(), anyBoolean())
val leaseA = manager.lease(2)
doReturn(2L).when(manager).sizeOf(meq(leaseA.tmpPath))
val dst = leaseA.commit("app1", None)
doReturn(2L).when(manager).sizeOf(meq(dst))
manager.release("app1", None)
assert(manager.committed() === 2)
val leaseB = manager.lease(2)
doReturn(2L).when(manager).sizeOf(meq(leaseB.tmpPath))
armed.set(true)
val commitThread = new Thread(() => leaseB.commit("app1", None))
commitThread.start()
inCommit.await(10, TimeUnit.SECONDS)
// commit() has renamed the store into place and added its size, but has
not yet
// registered the app in the active map.
manager.release("app1", None, delete = true)
releaseDone.countDown()
commitThread.join(TimeUnit.SECONDS.toMillis(10))
assert(manager.committed() === 0)
// The provider that opened the store eventually releases it.
manager.release("app1", None, delete = true)
assert(manager.committed() === 0)
}
```
</details>
This is pre-existing rather than something this PR introduces, and the call
sites are real: `createDiskStore()` serving a UI request can run against
`cleanLogs()` calling `release(delete = true)` for the same app. `makeRoom()`
vs `commit()` has the same gap — a store being committed is not in `active`, so
it can still pass the new re-check and be deleted, and the `info.size` deducted
there is a stale snapshot value.
Given the PR title says "double-counting store size on concurrent release
and makeRoom", readers will likely assume the whole class of bug is closed.
Could you either note the remaining window in the description, or close it here
as well by reserving the `active` entry before the rename?
### 3. Minor
- In `makeRoom()`, `freed` is a `ListBuffer[Long]` but only `freed.size` and
`freed.sum` are used; two counters (`var freedBytes = 0L`, `var freedCount =
0`) would be simpler.
- `openStore()` still calls `updateApplicationStoreInfo` outside the lock,
so a `release()` that deletes the store in between resurrects a listing entry
for a path that no longer exists. The accounting stays correct and the new
stale-entry cleanup in `makeRoom()` eventually removes it, so this only skews
the `needed` computation during an eviction scan — not a blocker, just noting
it.
- The wider lock now covers `deleteStore()` (a recursive delete of a
potentially large RocksDB directory) in both `release()` and `makeRoom()`, so
UI-driven `openStore()` calls can block behind it. The trade-off is already
called out in the description; just flagging that `makeRoom()` inherits it too.
--
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]
