Hendrik/xgboost
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[jvm-packages] better fix for shutdown applications (#4108)

Browse Source 
* intentionally failed task

* throw exception

* more

* stop sparkcontext directly

* stop from another thread

* new scope

* use a new thread

* daemon threads

* don't join the killer thread

* remove injected errors

* add comments
This commit is contained in:
Nan Zhu 2019-02-07 09:02:17 -08:00 committed by GitHub
parent 017c97b8ce
commit 05243642bb
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 17 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
jvm-packages/xgboost4j-spark/src/main/scala/org/apache/spark/SparkParallelismTracker.scala
View File

@ -20,7 +20,7 @@ import java.net.URL
import org.apache.commons.logging.LogFactory import org.apache.commons.logging.LogFactory
import org.apache.spark.scheduler.{SparkListener, SparkListenerExecutorRemoved, SparkListenerTaskEnd} import org.apache.spark.scheduler._
import org.codehaus.jackson.map.ObjectMapper import org.codehaus.jackson.map.ObjectMapper
import scala.collection.JavaConverters._ import scala.collection.JavaConverters._
import scala.concurrent.ExecutionContext.Implicits.global import scala.concurrent.ExecutionContext.Implicits.global
@ -114,21 +114,26 @@ class SparkParallelismTracker(
} }
} }
private class ErrorInXGBoostTraining(msg: String) extends ControlThrowable {
override def toString: String = s"ErrorInXGBoostTraining: $msg"
}
private[spark] class TaskFailedListener extends SparkListener { private[spark] class TaskFailedListener extends SparkListener {
private[this] val logger = LogFactory.getLog("XGBoostTaskFailedListener")
override def onTaskEnd(taskEnd: SparkListenerTaskEnd): Unit = { override def onTaskEnd(taskEnd: SparkListenerTaskEnd): Unit = {
taskEnd.reason match { taskEnd.reason match {
case taskEnd: SparkListenerTaskEnd => case taskEndReason: TaskFailedReason =>
if (taskEnd.reason.isInstanceOf[TaskFailedReason]) { logger.error(s"Training Task Failed during XGBoost Training: " +
throw new ErrorInXGBoostTraining(s"TaskFailed during XGBoost Training: " + s"$taskEndReason, stopping SparkContext")
s"${taskEnd.reason}") // Spark does not allow ListenerThread to shutdown SparkContext so that we have to do it
// in a separate thread
val sparkContextKiller = new Thread() {
override def run(): Unit = {
LiveListenerBus.withinListenerThread.withValue(false) {
SparkContext.getOrCreate().stop()
}
}
} }
case executorRemoved: SparkListenerExecutorRemoved => sparkContextKiller.setDaemon(true)
throw new ErrorInXGBoostTraining(s"Executor lost during XGBoost Training: " + sparkContextKiller.start()
s"${executorRemoved.reason}")
case _ => case _ =>
} }
} }