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 12 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
60 changes: 20 additions & 40 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 @@ -19,8 +19,7 @@ package org.apache.spark.sql.execution

import java.util.concurrent.locks.ReentrantReadWriteLock

import scala.collection.JavaConverters._
import scala.collection.mutable
import scala.collection.immutable.IndexedSeq

import org.apache.hadoop.fs.{FileSystem, Path}

Expand All @@ -47,7 +46,8 @@ case class CachedData(plan: LogicalPlan, cachedRepresentation: InMemoryRelation)
class CacheManager extends Logging {

@transient
private val cachedData = new java.util.LinkedList[CachedData]
@volatile
private var cachedData = IndexedSeq[CachedData]()
Copy link
Contributor

Choose a reason for hiding this comment

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

we'd better add some comments to explain the access pattern of cachedData. e.g. adding/removing elements should be done by creating a new seq, and locked by this.


@transient
private val cacheLock = new ReentrantReadWriteLock
Expand All @@ -72,12 +72,12 @@ class CacheManager extends Logging {

/** Clears all cached tables. */
def clearCache(): Unit = writeLock {
cachedData.asScala.foreach(_.cachedRepresentation.cacheBuilder.clearCache())
cachedData.clear()
cachedData.foreach(_.cachedRepresentation.cacheBuilder.clearCache())
cachedData = IndexedSeq[CachedData]()
}

/** Checks if the cache is empty. */
def isEmpty: Boolean = readLock {
Copy link
Contributor

Choose a reason for hiding this comment

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

can we remove the readLock method?

Copy link
Contributor

Choose a reason for hiding this comment

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

I think we can remove writeLock as well, and simply use this.synchronized

Copy link
Contributor Author

Choose a reason for hiding this comment

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

I made this change, but while doing that I realized there was an issue with the earlier change to the "partition" function. See comment below

val (plansToUncache, remainingPlans) = cachedData.partition(cd => shouldRemove(cd.plan))
// If a new plan is cached by a different thread at this point, it will be in the cachedData object,
// but not in plansToUncache or remainingPlans. So the next line will remove it.
writeLock {
cachedData = remainingPlans
}

Copy link
Contributor Author

Choose a reason for hiding this comment

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

So I reverted this back to the previous behavior.

def isEmpty: Boolean = {
cachedData.isEmpty
}

Expand Down Expand Up @@ -105,7 +105,7 @@ class CacheManager extends Logging {
if (lookupCachedData(planToCache).nonEmpty) {
logWarning("Data has already been cached.")
} else {
cachedData.add(CachedData(planToCache, inMemoryRelation))
cachedData = CachedData(planToCache, inMemoryRelation) +: cachedData
}
}
}
Expand Down Expand Up @@ -144,22 +144,12 @@ 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
}
}
}
plansToUncache.foreach { cd =>
writeLock {
cachedData.remove(cd)
}
cd.cachedRepresentation.cacheBuilder.clearCache(blocking)
val (plansToUncache, remainingPlans) = cachedData.partition(cd => shouldRemove(cd.plan))
writeLock {
cachedData = remainingPlans
}
plansToUncache.foreach { _.cachedRepresentation.cacheBuilder.clearCache(blocking) }

// Re-compile dependent cached queries after removing the cached query.
if (!cascade) {
recacheByCondition(spark, cd => {
Expand Down Expand Up @@ -194,22 +184,12 @@ 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 (needToRecache, remainingPlans) = cachedData.partition(condition)
writeLock {
// Remove the cache entry before creating a new ones.
cachedData = remainingPlans
}
needToRecache.map { cd =>
writeLock {
// Remove the cache entry before we create a new one, so that we can have a different
// physical plan.
cachedData.remove(cd)
}
cd.cachedRepresentation.cacheBuilder.clearCache()
val plan = spark.sessionState.executePlan(cd.plan).executedPlan
val newCache = InMemoryRelation(
Expand All @@ -220,20 +200,20 @@ class CacheManager extends Logging {
if (lookupCachedData(recomputedPlan.plan).nonEmpty) {
logWarning("While recaching, data was already added to cache.")
} else {
cachedData.add(recomputedPlan)
cachedData = recomputedPlan +: cachedData
}
}
}
}

/** Optionally returns cached data for the given [[Dataset]] */
def lookupCachedData(query: Dataset[_]): Option[CachedData] = readLock {
def lookupCachedData(query: Dataset[_]): Option[CachedData] = {
lookupCachedData(query.logicalPlan)
}

/** Optionally returns cached data for the given [[LogicalPlan]]. */
def lookupCachedData(plan: LogicalPlan): Option[CachedData] = readLock {
cachedData.asScala.find(cd => plan.sameResult(cd.plan))
def lookupCachedData(plan: LogicalPlan): Option[CachedData] = {
cachedData.find(cd => plan.sameResult(cd.plan))
}

/** Replaces segments of the given logical plan with cached versions where possible. */
Expand Down