Skip to content

[SPARK-4949]shutdownCallback in SparkDeploySchedulerBackend should be enclosed by synchronized block. #3781

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
wants to merge 7 commits into from
Closed

[SPARK-4949]shutdownCallback in SparkDeploySchedulerBackend should be enclosed by synchronized block. #3781

Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
5942765
Enclosed shutdownCallback in SparkDeploySchedulerBackend by synchroni…
sarutak Dec 24, 2014
1b60fd1
Improved the locking logics
sarutak Dec 24, 2014
f556819
Merge branch 'master' of git://git.apache.org/spark into SPARK-4949
sarutak Feb 18, 2015
552df7c
Changed the declaration of the variable "shutdownCallback" as a volat…
sarutak Feb 18, 2015
42ca528
Changed the declaration of the variable "shutdownCallback" as a volat…
sarutak Feb 18, 2015
c7265dc
Merge branch 'master' of git://git.apache.org/spark into SPARK-4949
sarutak Feb 18, 2015
c146c93
Removed "setShutdownCallback" method
sarutak Feb 18, 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
35 changes: 16 additions & 19 deletions core/src/main/scala/org/apache/spark/scheduler/cluster/SparkDeploySchedulerBackend.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@

package org.apache.spark.scheduler.cluster

import java.util.concurrent.Semaphore

import org.apache.spark.{Logging, SparkConf, SparkContext, SparkEnv}
import org.apache.spark.deploy.{ApplicationDescription, Command}
import org.apache.spark.deploy.client.{AppClient, AppClientListener}
Expand All @@ -31,16 +33,16 @@ private[spark] class SparkDeploySchedulerBackend(
with AppClientListener
with Logging {

var client: AppClient = null
var stopping = false
var shutdownCallback : (SparkDeploySchedulerBackend) => Unit = _
@volatile var appId: String = _
private var client: AppClient = null
private var stopping = false

@volatile var shutdownCallback: SparkDeploySchedulerBackend => Unit = _
@volatile private var appId: String = _

val registrationLock = new Object()
var registrationDone = false
private val registrationBarrier = new Semaphore(0)

val maxCores = conf.getOption("spark.cores.max").map(_.toInt)
val totalExpectedCores = maxCores.getOrElse(0)
private val maxCores = conf.getOption("spark.cores.max").map(_.toInt)
private val totalExpectedCores = maxCores.getOrElse(0)

override def start() {
super.start()
Expand Down Expand Up @@ -95,8 +97,10 @@ private[spark] class SparkDeploySchedulerBackend(
stopping = true
super.stop()
client.stop()
if (shutdownCallback != null) {
shutdownCallback(this)

val callback = shutdownCallback
if (callback != null) {
callback(this)
}
}

Expand Down Expand Up @@ -149,18 +153,11 @@ private[spark] class SparkDeploySchedulerBackend(
}

private def waitForRegistration() = {
registrationLock.synchronized {
while (!registrationDone) {
registrationLock.wait()
}
}
registrationBarrier.acquire()
}

private def notifyContext() = {
registrationLock.synchronized {
registrationDone = true
registrationLock.notifyAll()
}
registrationBarrier.release()
}

}