[SPARK-6980] [CORE] Akka timeout exceptions indicate which conf controls them (RPC Layer) #6205
- Loading branch information
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -18,7 +18,10 @@ | |
| package org.apache.spark.rpc | ||
|
|
||
| import java.net.URI | ||
| import java.util.concurrent.TimeoutException | ||
|
|
||
| import scala.concurrent.duration.FiniteDuration | ||
| import scala.concurrent.duration._ | ||
| import scala.concurrent.{Await, Future} | ||
| import scala.language.postfixOps | ||
|
|
||
|
|
@@ -94,7 +97,7 @@ private[spark] abstract class RpcEnv(conf: SparkConf) { | |
| * Retrieve the [[RpcEndpointRef]] represented by `uri`. This is a blocking action. | ||
| */ | ||
| def setupEndpointRefByURI(uri: String): RpcEndpointRef = { | ||
| Await.result(asyncSetupEndpointRefByURI(uri), defaultLookupTimeout.duration) | ||
| } | ||
|
|
||
| /** | ||
|
|
@@ -182,3 +185,68 @@ private[spark] object RpcAddress { | |
| RpcAddress(host, port) | ||
| } | ||
| } | ||
|
|
||
|
|
||
| /** | ||
| * Associates a timeout with a configuration property so that a TimeoutException can be | ||
| * traced back to the controlling property. | ||
| * @param timeout timeout duration in seconds | ||
| * @param description description to be displayed in a timeout exception | ||
| */ | ||
| private[spark] class RpcTimeout(timeout: FiniteDuration, description: String) { | ||
|
|
||
| /** Get the timeout duration */ | ||
| def duration: FiniteDuration = timeout | ||
|
Member
Author
There was a problem hiding this comment. Is this necessary or should we just use
Contributor
There was a problem hiding this comment. yeah I think you should just change the constructor
Member
Author
There was a problem hiding this comment. Sure, that sounds good. One other minor question @squito , the constructor has the parameter class RpcTimeout(val duration: FiniteDuration, val timeoutProp: String)what do you think?
Contributor
There was a problem hiding this comment. yes, I agree that would be more consistent, good point |
||
|
|
||
| /** Get the message associated with this timeout */ | ||
| def message: String = description | ||
|
|
||
| /** Amends the standard message of TimeoutException to include the description */ | ||
| def amend(te: TimeoutException): TimeoutException = { | ||
| new TimeoutException(te.getMessage() + " " + description) | ||
| } | ||
|
|
||
| /** Wait on a future result to catch and amend a TimeoutException */ | ||
| def awaitResult[T](future: Future[T]): T = { | ||
| try { | ||
| Await.result(future, duration) | ||
| } | ||
| catch { | ||
| case te: TimeoutException => | ||
| throw amend(te) | ||
| } | ||
| } | ||
|
|
||
| // TODO(bryanc) wrap Await.ready also | ||
| } | ||
|
|
||
| object RpcTimeout { | ||
|
|
||
|
|
||
| private[this] val messagePrefix = "This timeout is controlled by " | ||
|
Contributor
There was a problem hiding this comment. instead of putting The reason I'm asking is because in the tests, where you directly construct an
Member
Author
There was a problem hiding this comment. That's a good point about the message in the tests. Way back when I first made the unit test for I don't think this is the case anymore and maybe we should use
Contributor
There was a problem hiding this comment. I don't think this is related to whether or not we get the time as seconds or millis. I'm thinking that we still leave the diff --git a/core/src/main/scala/org/apache/spark/rpc/RpcEnv.scala b/core/src/main/scala/org/apache/spark/rpc/RpcEnv.scala
index 5149cf1..e0de397 100644
--- a/core/src/main/scala/org/apache/spark/rpc/RpcEnv.scala
+++ b/core/src/main/scala/org/apache/spark/rpc/RpcEnv.scala
@@ -197,19 +197,16 @@ private[rpc] class RpcTimeoutException(message: String, cause: TimeoutException)
* Associates a timeout with a description so that a when a TimeoutException occurs, additional
* context about the timeout can be amended to the exception message.
* @param timeout timeout duration in seconds
- * @param description description to be displayed in a timeout exception
+ * @param conf the configuration parameter that controls this timeout
*/
-private[spark] class RpcTimeout(timeout: FiniteDuration, description: String) {
+private[spark] class RpcTimeout(timeout: FiniteDuration, val conf: String) {
/** Get the timeout duration */
def duration: FiniteDuration = timeout
- /** Get the message associated with this timeout */
- def message: String = description
-
/** Amends the standard message of TimeoutException to include the description */
private def createRpcTimeoutException(te: TimeoutException): RpcTimeoutException = {
- new RpcTimeoutException(te.getMessage() + " " + description, te)
+ new RpcTimeoutException(te.getMessage() + ". This timeout is controlled by " + conf, te)
}(and corresponding changes to the I do kinda wish that the values were read w/
Contributor
There was a problem hiding this comment. in fact we could also mark the constructor as
Member
Author
There was a problem hiding this comment. I see what you're saying. I kept the Quick Scala question: would it be better to make the constructor use |
||
|
|
||
| /** | ||
| * Lookup the timeout property in the configuration and create | ||
| * a RpcTimeout with the property key in the description. | ||
| * @param conf configuration properties containing the timeout | ||
| * @param timeoutProp property key for the timeout in seconds | ||
| * @throws NoSuchElementException if property is not set | ||
| */ | ||
| def apply(conf: SparkConf, timeoutProp: String): RpcTimeout = { | ||
| val timeout = { conf.getTimeAsSeconds(timeoutProp) seconds } | ||
| new RpcTimeout(timeout, messagePrefix + timeoutProp) | ||
| } | ||
|
|
||
| /** | ||
| * Lookup the timeout property in the configuration and create | ||
| * a RpcTimeout with the property key in the description. | ||
| * Uses the given default value if property is not set | ||
| * @param conf configuration properties containing the timeout | ||
| * @param timeoutProp property key for the timeout in seconds | ||
| * @param defaultValue default timeout value in seconds if property not found | ||
| */ | ||
| def apply(conf: SparkConf, timeoutProp: String, defaultValue: String): RpcTimeout = { | ||
| val timeout = { conf.getTimeAsSeconds(timeoutProp, defaultValue) seconds } | ||
|
Member
Author
There was a problem hiding this comment. Should this be
Contributor
There was a problem hiding this comment. as mentioned above, I don't think so -- lets stick w/ the current behavior of using |
||
| new RpcTimeout(timeout, messagePrefix + timeoutProp) | ||
| } | ||
|
|
||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -20,7 +20,6 @@ package org.apache.spark.rpc.akka | |
| import java.util.concurrent.ConcurrentHashMap | ||
|
|
||
| import scala.concurrent.Future | ||
| import scala.language.postfixOps | ||
| import scala.reflect.ClassTag | ||
| import scala.util.control.NonFatal | ||
|
|
@@ -212,7 +211,7 @@ private[spark] class AkkaRpcEnv private[akka] ( | |
|
|
||
| override def asyncSetupEndpointRefByURI(uri: String): Future[RpcEndpointRef] = { | ||
| import actorSystem.dispatcher | ||
| actorSystem.actorSelection(uri).resolveOne(defaultLookupTimeout.duration). | ||
|
Contributor
There was a problem hiding this comment. though this doesn't follow the same pattern of There was a problem hiding this comment. I feel there are two ways :
In my opinion 2nd one is the best solution as the previous one will require modifications where ever the usages are.
Member
Author
There was a problem hiding this comment. The timeout is a little trickier for Futures. From what I understand, creating the future is non-blocking, so we can't just call I think we might be able to use the
Contributor
There was a problem hiding this comment. what you are describing sounds reasonable. We certainly don't want to change the return type or await on the result here. You could potentially fix the doubly-appended msg by throwing a subclass of (Honestly I thiink it will also be fine to not stress too much about this case, it may not be worth it.)
Member
Author
There was a problem hiding this comment. I'm going to do some research on this and see what our options are while still keeping the same return type. |
||
| map(new AkkaRpcEndpointRef(defaultAddress, _, conf)) | ||
| } | ||
|
|
||
|
|
@@ -293,9 +292,9 @@ private[akka] class AkkaRpcEndpointRef( | |
| actorRef ! AkkaMessage(message, false) | ||
| } | ||
|
|
||
| override def ask[T: ClassTag](message: Any, timeout: RpcTimeout): Future[T] = { | ||
| import scala.concurrent.ExecutionContext.Implicits.global | ||
| actorRef.ask(AkkaMessage(message, true))(timeout.duration).flatMap { | ||
|
Contributor
There was a problem hiding this comment. same thing here about catching the timeout |
||
| case msg @ AkkaMessage(message, reply) => | ||
| if (reply) { | ||
| logError(s"Receive $msg but the sender cannot reply") | ||
|
|
||
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.
any reason not to use
timeout.awaitResulthere, so we get the better exception msg?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.
Nope - I must have missed that one, it's fixed now.