[SPARK-5529][CORE]Add expireDeadHosts in HeartbeatReceiver #4363
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -17,33 +17,84 @@ | |
|
|
||
| package org.apache.spark | ||
|
|
||
| import scala.concurrent.duration._ | ||
| import scala.collection.mutable | ||
|
|
||
| import akka.actor.{Actor, Cancellable} | ||
|
|
||
| import org.apache.spark.executor.TaskMetrics | ||
| import org.apache.spark.storage.BlockManagerId | ||
| import org.apache.spark.scheduler.{SlaveLost, TaskScheduler} | ||
| import org.apache.spark.util.ActorLogReceive | ||
|
|
||
| /** | ||
| * A heartbeat from executors to the driver. This is a shared message used by several internal | ||
| * components to convey liveness or execution information for in-progress tasks. It will also | ||
| * expire the hosts that have not heartbeated for more than spark.driver.executorTimeoutMs. | ||
| */ | ||
| private[spark] case class Heartbeat( | ||
| executorId: String, | ||
| taskMetrics: Array[(Long, TaskMetrics)], // taskId -> TaskMetrics | ||
| blockManagerId: BlockManagerId) | ||
|
|
||
| private[spark] case object ExpireDeadHosts | ||
|
|
||
| private[spark] case class HeartbeatResponse(reregisterBlockManager: Boolean) | ||
|
|
||
| /** | ||
| * Lives in the driver to receive heartbeats from executors.. | ||
|
There was a problem hiding this comment. We should add a header comment that communicates that one of the functions of the HeartbeatReceiver is to expire executors. There was a problem hiding this comment. Hi Sryza, thanks for your review. |
||
| */ | ||
| private[spark] class HeartbeatReceiver(sc: SparkContext, scheduler: TaskScheduler) | ||
|
There was a problem hiding this comment. I would recommend limit the number of things we pass to this receiver to the following set of smaller things instead of the whole The There was a problem hiding this comment. It will not easy to understand, on the other hand the SparkContext is use in a lot of place. There was a problem hiding this comment. Yes, but it's bad design pattern to pass in the whole Anyway, not a big deal since this is private. I will merge it as is. |
||
| extends Actor with ActorLogReceive with Logging { | ||
|
|
||
| val executorLastSeen = new mutable.HashMap[String, Long] | ||
|
There was a problem hiding this comment. Please add a comment on what the keys and values are: |
||
|
|
||
| val executorTimeout = sc.conf.getLong("spark.driver.executorTimeoutMs", | ||
|
There was a problem hiding this comment.
There was a problem hiding this comment. actually, can this just use |
||
| sc.conf.getLong("spark.storage.blockManagerSlaveTimeoutMs", 120 * 1000)) | ||
|
|
||
| val checkTimeoutInterval = sc.conf.getLong("spark.driver.executorTimeoutIntervalMs", | ||
|
There was a problem hiding this comment. this isn't really a driver property. I would just use There was a problem hiding this comment.
There was a problem hiding this comment. @sryza if you read the documentation of There was a problem hiding this comment. As for There was a problem hiding this comment.
You are right. However, in all the other places that There was a problem hiding this comment. well we do already have the There was a problem hiding this comment. The There was a problem hiding this comment. Yes, I'd still like to avoid introducing new configs unless we have to since we can't remove those easily. I prefer to just make this consider only the There was a problem hiding this comment. yes, Now i agree with considering the spark.network.* and spark.storage.blockManager.* configs. |
||
| sc.conf.getLong("spark.storage.blockManagerTimeoutIntervalMs", 60000)) | ||
|
|
||
| var timeoutCheckingTask: Cancellable = null | ||
|
There was a problem hiding this comment. these could all be private |
||
|
|
||
| override def preStart(): Unit = { | ||
| import context.dispatcher | ||
| timeoutCheckingTask = context.system.scheduler.schedule(0.seconds, | ||
| checkTimeoutInterval.milliseconds, self, ExpireDeadHosts) | ||
| super.preStart() | ||
| } | ||
|
|
||
| override def receiveWithLogging = { | ||
| case Heartbeat(executorId, taskMetrics, blockManagerId) => | ||
| val response = HeartbeatResponse( | ||
| !scheduler.executorHeartbeatReceived(executorId, taskMetrics, blockManagerId)) | ||
|
There was a problem hiding this comment. not your code, but can you put this in a separate val: |
||
| executorLastSeen(executorId) = System.currentTimeMillis() | ||
| sender ! response | ||
| case ExpireDeadHosts => | ||
| expireDeadHosts() | ||
| } | ||
|
|
||
| private def expireDeadHosts(): Unit = { | ||
| logTrace("Checking for hosts with no recent heartbeats in HeartbeatReceiver.") | ||
| val now = System.currentTimeMillis() | ||
| val minSeenTime = now - executorTimeout | ||
| for ((executorId, lastSeenMs) <- executorLastSeen) { | ||
| if (lastSeenMs < minSeenTime) { | ||
|
There was a problem hiding this comment. I think it's easier to read if it's the following instead: |
||
| logWarning(s"Removing executor $executorId with no recent heartbeats: " + | ||
| s"${now - lastSeenMs} ms exceeds timeout $executorTimeout ms") | ||
| scheduler.executorLost(executorId, SlaveLost()) | ||
|
There was a problem hiding this comment. Please add an error message in |
||
| if (sc.supportKillExecutor) { | ||
| sc.killExecutor(executorId) | ||
| } | ||
| executorLastSeen.remove(executorId) | ||
|
There was a problem hiding this comment. What happens if a heartbeat from the executor gets delivered after we kill / remove it? There was a problem hiding this comment. Because the akka connection is still alive, we can kill executor by send kill message to applicationMaster. There was a problem hiding this comment. I believe this code will correctly expire that executor as a dead host after a timeout. |
||
| } | ||
| } | ||
| } | ||
|
|
||
| override def postStop(): Unit = { | ||
| if (timeoutCheckingTask != null) { | ||
| timeoutCheckingTask.cancel() | ||
| } | ||
| super.postStop() | ||
| } | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
you'll need to update this comment if you rename the configs