Github user vanzin commented on a diff in the pull request:
https://github.com/apache/spark/pull/16706#discussion_r98121685
--- Diff: core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala
---
@@ -501,34 +498,99 @@ 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(
+ val senderAddress: RpcAddress,
+ val receiver: NettyRpcEndpointRef, val content: Any) {
+
+ /** Manually serialize [[RequestMessage]] to minimize the size of bytes.
*/
+ def serialize(nettyEnv: NettyRpcEnv): ByteBuffer = {
+ val bos = new ByteBufferOutputStream()
+ val out = new DataOutputStream(bos)
+ try {
+ writeRpcAddress(out, senderAddress)
+ writeRpcAddress(out, receiver.address)
+ out.writeUTF(receiver.name)
+ val contentBytes = nettyEnv.serialize(content)
--- End diff --
Hmmm... could you use `JavaSerializerInstance.serializeStream` here instead?
You avoid: extra object allocations in `serialize`, two copies of the
serialized content in memory, and the extra copy operation below in `out.write`.
You could also use `ObjectOutputStream` directly (it implements
`DataOutput`) but that makes it difficult to use Kryo later.
---
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]
