Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[SPARK-26917][SQL] Further reduce locks in CacheManager #24028

Closed
wants to merge 17 commits into from
Closed

[SPARK-26917][SQL] Further reduce locks in CacheManager #24028

Changes from 9 commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
22bd4bd
Merge pull request #1 from apache/master
DaveDeCaprio Nov 27, 2018
9219c01
Merge branch 'master' of https://github.com/apache/spark
DaveDeCaprio Jan 5, 2019
771d12d
Merge branch 'master' of https://github.com/apache/spark
DaveDeCaprio Jan 14, 2019
0277891
Merge branch 'master' of https://github.com/apache/spark
DaveDeCaprio Feb 18, 2019
0971a06
Merge branch 'master' of https://github.com/apache/spark
DaveDeCaprio Feb 21, 2019
697796e
Merge branch 'master' of https://github.com/apache/spark
DaveDeCaprio Mar 1, 2019
375ee6a
Merge branch 'master' of https://github.com/apache/spark
DaveDeCaprio Mar 8, 2019
fc7487b
Reduced readLocks since they cna cause contention also.
DaveDeCaprio Mar 8, 2019
27ab9f0
Removed extra code
DaveDeCaprio Mar 9, 2019
1d6f84a
Changed cachedData to a var of an immutable IndexedSeq.
DaveDeCaprio Mar 11, 2019
a5977f0
Minor updates from code review. Added volatile
DaveDeCaprio Mar 12, 2019
1de029c
Tabs to spaces
DaveDeCaprio Mar 12, 2019
755d484
Put annotations on one line.
DaveDeCaprio Mar 13, 2019
032c606
Removed read and write lock functions.
DaveDeCaprio Mar 13, 2019
d402533
Added comments.
DaveDeCaprio Mar 13, 2019
1d693af
Fixed comment formatting
DaveDeCaprio Mar 13, 2019
1bb2511
Changed to reference equals for plan removal.
DaveDeCaprio Mar 14, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 6 additions & 18 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -144,16 +144,10 @@ class CacheManager extends Logging {
} else {
_.sameResult(plan)
}
val plansToUncache = mutable.Buffer[CachedData]()
readLock {
val it = cachedData.iterator()
while (it.hasNext) {
val cd = it.next()
if (shouldRemove(cd.plan)) {
plansToUncache += cd
}
}
val cachedDataCopy = readLock {
cachedData.asScala.clone()
}
val plansToUncache = cachedDataCopy.filter(cd => shouldRemove(cd.plan))
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It seems we always use whole loops when performance should be a concern:
https://github.com/databricks/scala-style-guide#traversal-and-zipwithindex
No actual performance impact between the two pattens on your heavy workload?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I suspect that it's the logic in shouldRemove that takes the time here, and can be done without the lock.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, the problem is that the "shouldRemove" function is passed into this call. If that call is expensive, it causes the lock to be held for arbitrarily long amounts of time.

"shouldRemove" when called from "recacheByPath" causes a full traversal of the entire logical plan tree for every cached plan. In the process of doing this it will regenerate path strings for every file referenced by every single plan. In our situations at least this is easily many orders of magnitude more memory overhead than a shallow copy of the list.

plansToUncache.foreach { cd =>
writeLock {
cachedData.remove(cd)
Expand Down Expand Up @@ -194,16 +188,10 @@ class CacheManager extends Logging {
private def recacheByCondition(
spark: SparkSession,
condition: CachedData => Boolean): Unit = {
val needToRecache = scala.collection.mutable.ArrayBuffer.empty[CachedData]
readLock {
val it = cachedData.iterator()
while (it.hasNext) {
val cd = it.next()
if (condition(cd)) {
needToRecache += cd
}
}
val cachedDataCopy = readLock {
cachedData.asScala.clone()
}
val needToRecache = cachedDataCopy.filter(condition)
needToRecache.map { cd =>
writeLock {
// Remove the cache entry before we create a new one, so that we can have a different
Expand Down