Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/16706#discussion_r98061665
--- Diff: core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala
---
@@ -501,34 +498,105 @@ private[netty] class NettyRpcEndpointRef(
out.defaultWriteObject()
}
- override def name: String = _name
+ override def name: String = endpointAddress.name
override def ask[T: ClassTag](message: Any, timeout: RpcTimeout):
Future[T] = {
- nettyEnv.ask(RequestMessage(nettyEnv.address, this, message), timeout)
+ nettyEnv.ask(new RequestMessage(nettyEnv.address, this, message),
timeout)
}
override def send(message: Any): Unit = {
require(message != null, "Message is null")
- nettyEnv.send(RequestMessage(nettyEnv.address, this, message))
+ nettyEnv.send(new RequestMessage(nettyEnv.address, this, message))
}
- override def toString: String = s"NettyRpcEndpointRef(${_address})"
-
- def toURI: URI = new URI(_address.toString)
+ override def toString: String =
s"NettyRpcEndpointRef(${endpointAddress})"
final override def equals(that: Any): Boolean = that match {
- case other: NettyRpcEndpointRef => _address == other._address
+ case other: NettyRpcEndpointRef => endpointAddress ==
other.endpointAddress
case _ => false
}
- final override def hashCode(): Int = if (_address == null) 0 else
_address.hashCode()
+ final override def hashCode(): Int =
+ if (endpointAddress == null) 0 else endpointAddress.hashCode()
}
/**
* The message that is sent from the sender to the receiver.
+ *
+ * @param senderAddress the sender address. It's `null` if this message is
from a client
+ * `NettyRpcEnv`.
+ * @param receiver the receiver of this message.
+ * @param content the message content.
*/
-private[netty] case class RequestMessage(
- senderAddress: RpcAddress, receiver: NettyRpcEndpointRef, content: Any)
+private[netty] class RequestMessage(
+ @Nullable val senderAddress: RpcAddress,
+ val receiver: NettyRpcEndpointRef, val content: Any) {
+
+ /** Manually serialize [[RequestMessage]] to minimize the size of bytes.
*/
+ def serialize(nettyEnv: NettyRpcEnv): ByteBuffer = {
--- End diff --
It's different. If I just implement `writeObject` and call Java
serialization APIs to write `RequestMessage`, it will write the full class name
`RequestMessage` and a serialization id which are not needed by all RPC
messages.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
