From adeae393059d6cb5f7eeb9ca789e4aa761885b15 Mon Sep 17 00:00:00 2001 From: Marcelo Vanzin Date: Fri, 19 Jul 2019 15:00:25 -0700 Subject: [PATCH 1/2] [SPARK-28455][core] Avoid overflow when calculating executor timeout time. This would cause the timeout time to be negative, so executors would be timed out immediately (instead of never). I also tweaked a couple of log messages that could get pretty long when lots of executors were active. Added unit test (which failed without the fix). --- .../scheduler/dynalloc/ExecutorMonitor.scala | 19 ++++++++++++++---- .../dynalloc/ExecutorMonitorSuite.scala | 20 +++++++++++++++++++ 2 files changed, 35 insertions(+), 4 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitor.scala b/core/src/main/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitor.scala index f5beb403555e..dd3ec5a16b8b 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitor.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitor.scala @@ -188,7 +188,12 @@ private[spark] class ExecutorMonitor( exec.updateActiveShuffles(activeShuffleIds) if (exec.hasActiveShuffle) { needTimeoutUpdate = true - activatedExecs += id + // Don't collect too many ids or the debug message becomes not very useful. + if (activatedExecs.size < 10) { + activatedExecs += id + } else if (activatedExecs.size == 10) { + activatedExecs += "..." + } } } } @@ -237,8 +242,13 @@ private[spark] class ExecutorMonitor( executors.asScala.foreach { case (id, exec) => if (exec.hasActiveShuffle) { exec.updateActiveShuffles(activeShuffles) - if (!exec.hasActiveShuffle) { - deactivatedExecs += id + if (!exec.hasActiveShuffle && log.isDebugEnabled) { + // Don't collect too many ids or the debug message becomes not very useful. + if (deactivatedExecs.size < 10) { + deactivatedExecs += id + } else if (deactivatedExecs.size == 10) { + deactivatedExecs += "..." + } } } } @@ -448,7 +458,8 @@ private[spark] class ExecutorMonitor( } else { idleTimeoutMs } - idleStart + timeout + val deadline = idleStart + timeout + if (deadline >= 0) deadline else Long.MaxValue } else { Long.MaxValue } diff --git a/core/src/test/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitorSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitorSuite.scala index e11ee97469b0..6a25754fcbe5 100644 --- a/core/src/test/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitorSuite.scala +++ b/core/src/test/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitorSuite.scala @@ -367,6 +367,26 @@ class ExecutorMonitorSuite extends SparkFunSuite { assert(monitor.timedOutExecutors(idleDeadline).toSet === Set("1", "2")) } + test("SPARK-28455: avoid overflow in timeout calculation") { + conf + .set(DYN_ALLOCATION_SHUFFLE_TIMEOUT, Long.MaxValue) + .set(DYN_ALLOCATION_SHUFFLE_TRACKING, true) + .set(SHUFFLE_SERVICE_ENABLED, false) + monitor = new ExecutorMonitor(conf, client, null, clock) + + // Generate events that will make executor 1 be idle, while still holding shuffle data. + // The executor should not be eligible for removal since the timeout is basically "infinite". + val stage = stageInfo(1, shuffleId = 0) + monitor.onJobStart(SparkListenerJobStart(1, clock.getTimeMillis(), Seq(stage))) + clock.advance(1000L) + monitor.onExecutorAdded(SparkListenerExecutorAdded(clock.getTimeMillis(), "1", null)) + monitor.onTaskStart(SparkListenerTaskStart(1, 0, taskInfo("1", 1))) + monitor.onTaskEnd(SparkListenerTaskEnd(1, 0, "foo", Success, taskInfo("1", 1), null)) + monitor.onJobEnd(SparkListenerJobEnd(1, clock.getTimeMillis(), JobSucceeded)) + + assert(monitor.timedOutExecutors(idleDeadline).isEmpty) + } + private def idleDeadline: Long = clock.getTimeMillis() + idleTimeoutMs + 1 private def storageDeadline: Long = clock.getTimeMillis() + storageTimeoutMs + 1 private def shuffleDeadline: Long = clock.getTimeMillis() + shuffleTimeoutMs + 1 From 392e5858c9029dc2e4d81198e241b2fadc5335eb Mon Sep 17 00:00:00 2001 From: Marcelo Vanzin Date: Mon, 22 Jul 2019 10:25:58 -0700 Subject: [PATCH 2/2] Log tweak. --- .../scheduler/dynalloc/ExecutorMonitor.scala | 46 +++++++++++-------- 1 file changed, 27 insertions(+), 19 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitor.scala b/core/src/main/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitor.scala index dd3ec5a16b8b..d0337b6e3496 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitor.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitor.scala @@ -182,24 +182,19 @@ private[spark] class ExecutorMonitor( if (updateExecutors) { val activeShuffleIds = shuffleStages.map(_._2).toSeq var needTimeoutUpdate = false - val activatedExecs = new mutable.ArrayBuffer[String]() + val activatedExecs = new ExecutorIdCollector() executors.asScala.foreach { case (id, exec) => if (!exec.hasActiveShuffle) { exec.updateActiveShuffles(activeShuffleIds) if (exec.hasActiveShuffle) { needTimeoutUpdate = true - // Don't collect too many ids or the debug message becomes not very useful. - if (activatedExecs.size < 10) { - activatedExecs += id - } else if (activatedExecs.size == 10) { - activatedExecs += "..." - } + activatedExecs.add(id) } } } - logDebug(s"Activated executors ${activatedExecs.mkString(",")} due to shuffle data " + - s"needed by new job ${event.jobId}.") + logDebug(s"Activated executors $activatedExecs due to shuffle data needed by new job" + + s"${event.jobId}.") if (needTimeoutUpdate) { nextTimeout.set(Long.MinValue) @@ -238,23 +233,18 @@ private[spark] class ExecutorMonitor( } } - val deactivatedExecs = new mutable.ArrayBuffer[String]() + val deactivatedExecs = new ExecutorIdCollector() executors.asScala.foreach { case (id, exec) => if (exec.hasActiveShuffle) { exec.updateActiveShuffles(activeShuffles) - if (!exec.hasActiveShuffle && log.isDebugEnabled) { - // Don't collect too many ids or the debug message becomes not very useful. - if (deactivatedExecs.size < 10) { - deactivatedExecs += id - } else if (deactivatedExecs.size == 10) { - deactivatedExecs += "..." - } + if (!exec.hasActiveShuffle) { + deactivatedExecs.add(id) } } } - logDebug(s"Executors ${deactivatedExecs.mkString(",")} do not have active shuffle data " + - s"after job ${event.jobId} finished.") + logDebug(s"Executors $deactivatedExecs do not have active shuffle data after job " + + s"${event.jobId} finished.") } jobToStageIDs.remove(event.jobId).foreach { stages => @@ -502,4 +492,22 @@ private[spark] class ExecutorMonitor( private case class ShuffleCleanedEvent(id: Int) extends SparkListenerEvent { override protected[spark] def logEvent: Boolean = false } + + /** Used to collect executor IDs for debug messages (and avoid too long messages). */ + private class ExecutorIdCollector { + private val ids = if (log.isDebugEnabled) new mutable.ArrayBuffer[String]() else null + private var excess = 0 + + def add(id: String): Unit = if (log.isDebugEnabled) { + if (ids.size < 10) { + ids += id + } else { + excess += 1 + } + } + + override def toString(): String = { + ids.mkString(",") + (if (excess > 0) s" (and $excess more)" else "") + } + } }