ODIn/spark-instrumented-optimizer
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

[SPARK-31757][CORE] Improve HistoryServerDiskManager.updateAccessTime()

Browse source 
### What changes were proposed in this pull request?

The function `HistoryServerDiskManager`.`updateAccessTime()` would recompute the application store directory size every time it's triggered, this effort could be avoided because we already computed the new size outside the function call.

### Does this PR introduce _any_ user-facing change?

No

### How was this patch tested?

Existing test cases.

Closes #28579 from jiangxb1987/updateInfo.

Authored-by: Xingbo Jiang <xingbo.jiang@databricks.com>
Signed-off-by: Xingbo Jiang <xingbo.jiang@databricks.com>
This commit is contained in:
Xingbo Jiang 2020-05-21 23:36:55 -07:00
parent 83d0967dcc
commit 245aee9fc8
1 changed files with 9 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
core/src/main/scala/org/apache/spark/deploy/history/HistoryServerDiskManager.scala
View file

@ -122,10 +122,12 @@ 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 {
val path = appStorePath(appId, attemptId)
if (path.isDirectory()) {
active(appId -> attemptId) = sizeOf(path)
newSize = sizeOf(path)
active(appId -> attemptId) = newSize
Some(path)
} else {
None
@ -133,7 +135,7 @@ private class HistoryServerDiskManager(
}
storePath.foreach { path =>
updateAccessTime(appId, attemptId)
updateApplicationStoreInfo(appId, attemptId, newSize)
}
storePath
@ -238,10 +240,11 @@ private class HistoryServerDiskManager(
new File(appStoreDir, fileName)
}
private def updateAccessTime(appId: String, attemptId: Option[String]): Unit = {
private def updateApplicationStoreInfo(
appId: String, attemptId: Option[String], newSize: Long): Unit = {
val path = appStorePath(appId, attemptId)
val info = ApplicationStoreInfo(path.getAbsolutePath(), clock.getTimeMillis(), appId, attemptId,
sizeOf(path))
val info = ApplicationStoreInfo(path.getAbsolutePath(), clock.getTimeMillis(), appId,
attemptId, newSize)
listing.write(info)
}
@ -297,7 +300,7 @@ private class HistoryServerDiskManager(
s"exceeded ($current > $max)")
}
updateAccessTime(appId, attemptId)
updateApplicationStoreInfo(appId, attemptId, newSize)
active.synchronized {
active(appId -> attemptId) = newSize