Skip to content

[SPARK-12049] [CORE] User JVM shutdown hook can cause deadlock at shutdown #10042

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

Closed
srowen wants to merge 2 commits into from
Closed

[SPARK-12049] [CORE] User JVM shutdown hook can cause deadlock at shutdown #10042

Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
934f237
Avoid potential deadlock with a user app's shutdown hook thread by mo…
srowen Nov 30, 2015
3da8644
Inline checkState(); avoid race condition with shuttingDown
srowen Nov 30, 2015
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
17 changes: 9 additions & 8 deletions core/src/main/scala/org/apache/spark/util/ShutdownHookManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -206,7 +206,7 @@ private[spark] object ShutdownHookManager extends Logging {
private [util] class SparkShutdownHookManager {

private val hooks = new PriorityQueue[SparkShutdownHook]()
private var shuttingDown = false
@volatile private var shuttingDown = false

/**
* Install a hook to run at shutdown and run all registered hooks in order. Hadoop 1.x does not
Expand All @@ -232,22 +232,23 @@ private [util] class SparkShutdownHookManager {
}
}

def runAll(): Unit = synchronized {
def runAll(): Unit = {
shuttingDown = true
while (!hooks.isEmpty()) {
Try(Utils.logUncaughtExceptions(hooks.poll().run()))
var nextHook: SparkShutdownHook = null
while ({nextHook = hooks synchronized { hooks.poll() }; nextHook != null}) {

@vanzin vanzin Nov 30, 2015

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

nit: we generally use hooks.synchronized.

Try(Utils.logUncaughtExceptions(nextHook.run()))
}
}

def add(priority: Int, hook: () => Unit): AnyRef = synchronized {
def add(priority: Int, hook: () => Unit): AnyRef = {
checkState()
val hookRef = new SparkShutdownHook(priority, hook)
hooks.add(hookRef)
hooks synchronized { hooks.add(hookRef) }

@vanzin vanzin Nov 30, 2015

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

So, technically, this is not really thread-safe anymore. You can have this order of things:

T1: in add(), passes checkState() check
T2: in runAll(), sets shuttingDown to true
T2: in runAll(), calls hooks.poll(), list is empty, stops calling hooks
T1: in add(), adds hook to the list, hook will never be called.

Granted, that's very unlikely. I think if you move the checkState() call inside the lock it should fix things, though. You may end up adding a new hook when shuttingDown is true in that case, but you make sure that runAll will still pick it up.

@srowen srowen Nov 30, 2015

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Yeah, you got me. That really has to be tightened up. Will address these shortly.

hookRef
}

def remove(ref: AnyRef): Boolean = synchronized {
hooks.remove(ref)
def remove(ref: AnyRef): Boolean = {
hooks synchronized { hooks.remove(ref) }
}

private def checkState(): Unit = {
Expand Down