Spark source code ::Spark multithreading :: nettyrpCENv.ask

private[netty] def askT: ClassTag: Future[T] = {

// < / Any > < / Any > < / Any > < / Any > < / Any > < / Any > < / Any > < / Any > < / Any > < / Any > < / Any > < / Any > < / Any > < / Any > < / Any > < / Any > < / Any > Unit = { if (! promise.tryFailure(e)) { e match { case e : RpcEnvStoppedException => logDebug (s"Ignored failure: $e") case _ => logWarning(s"Ignored failure: $e")}}}} /* The onSuccess declared here will be populated into the onSuccess of the rpcResponseAllback. This listener is the one used in Figure 9. When we get a response from the Server side, note that we are not getting a response of type RPCFailure. Else if (message instanceof RpcResponse) {branch */ def onSuccess(Reply: Any): UnitReply match {case rpcFailure (e) => onFailure(e) case rpCreply => /* Call here promise trySuccess also for operation, [download] PayPal (https://www.gendan5.com/wallet/PayPal.html), here is what is said above, in order to a promise laid a future, */ if (! promise.trySuccess(rpcReply)) { logWarning(s"Ignored message: $reply") } } try { if (remoteAddr == address) { val p = Promise[Any]() p.future.onComplete { case Success(response) => onSuccess(response) case Failure(e) => onFailure(e) }(ThreadUtils.sameThread) dispatcher.postLocalMessage(message, p) } else { val rpcMessage = RpcOutboxMessage(message.serialize(this), onFailure, (client, response) => **onSuccess**(deserialize[Any](client, response))) postToOutbox(message.receiver, RpcMessage) / * if the callback is Failure, there will be executed * / promise in the future. Failed. Foreach {case _ : TimeoutException => rpcMessage.onTimeout() case _ => }(ThreadUtils.sameThread) } val timeoutCancelable = timeoutScheduler.schedule(new Runnable { override def run(): Unit = { onFailure(new TimeoutException(s"Cannot receive any reply from ${remoteAddr} " + s"in ${timeout.duration}")) } }, the timeout. Duration. ToNanos, TimeUnit NANOSECONDS) / * execute when the promise of the future, Will call here the onComplete method * / promise. The future. The onComplete {v = > timeoutCancelable. Cancel (true)} (ThreadUtils. SameThread)} the catch {case NonFatal(e) => onFailure(e)} /* AddMessageIfTimeout (addMessageIfTimeout) If it is an RPCTimeoutException, throw the ex as an RPCTimeoutException */ promise.future.mapTo[T].recover(timeout.addMessageIfTimeout)(ThreadUtils.sameThread)

}