[jira] [Comment Edited] (SPARK-5124) Standardize internal RPC interface
[ https://issues.apache.org/jira/browse/SPARK-5124?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14348215#comment-14348215 ] Shixiong Zhu edited comment on SPARK-5124 at 3/5/15 6:09 AM: - I made some changes to the above APIs. 1. I added a new trait RpcResponse because RpcResponseCallback can not be used directly. Both `receive` and `receiveAndReply` methods can receive messages if the sender is an RpcEndpoint, so if the receiver wants to reply the sender, it needs an approach to specify which methods it wants to send. Here is the interface of RpcResponse: {quote} trait RpcResponse { def reply(response: Any): Unit def replyWithSender(response: Any, sender: RpcEndpointRef): Unit def fail(e: Throwable): Unit } {quote} Calling `reply` will send the message to RpcEndpoint.receive, and calling `replyWithSender` will send the message to RpcEndpoint.replyWithSender. 2.Insteading of adding a trait like ThreadSafeRpcEndpoint, I added a new method `setupThreadSafeEndpoint` in RpcEnv, so that RpcEnv can decide how to implement the thread-safe semantics internally. For `setupEndpoint`, RpcEnv won't guarantee sending messages thread-safely. [~vanzin] For the local Endpoint idea, I think we need a local message dispatcher to send/receive messages between different Endpoint. It can not be implemented just in the Endpoint. We will also need EndpointRef to refer to an Endpoint, and a generic trait for the local message dispatcher and RpcEnv. It looks a bit complex. What do you think? Please help review my PR: https://github.com/apache/spark/pull/4588 was (Author: zsxwing): I made some changes to the above APIs. 1. I added a new trait RpcResponse because RpcResponseCallback can not be used directly. Both `receive` and `receiveAndReply` methods can receive messages if the sender is an RpcEndpoint, so if the receiver wants to reply the sender, it needs an approach to specify which methods it wants to send. Here is the interface of RpcResponse: {quote} private[spark] trait RpcResponse { def reply(response: Any): Unit def replyWithSender(response: Any, sender: RpcEndpointRef): Unit def fail(e: Throwable): Unit } {quote} Calling `reply` will send the message to RpcEndpoint.receive, and calling `replyWithSender` will send the message to RpcEndpoint.replyWithSender. 2.Insteading of adding a trait like ThreadSafeRpcEndpoint, I added a new method `setupThreadSafeEndpoint` in RpcEnv, so that RpcEnv can decide how to implement the thread-safe semantics internally. For `setupEndpoint`, RpcEnv won't guarantee sending messages thread-safely. [~vanzin] For the local Endpoint idea, I think we need a local message dispatcher to send/receive messages between different Endpoint. It can not be implemented just in the Endpoint. We will also need EndpointRef to refer to an Endpoint, and a generic trait for the local message dispatcher and RpcEnv. It looks a bit complex. What do you think? Please help review my PR: https://github.com/apache/spark/pull/4588 > Standardize internal RPC interface > -- > > Key: SPARK-5124 > URL: https://issues.apache.org/jira/browse/SPARK-5124 > Project: Spark > Issue Type: Sub-task > Components: Spark Core >Reporter: Reynold Xin >Assignee: Shixiong Zhu > Attachments: Pluggable RPC - draft 1.pdf, Pluggable RPC - draft 2.pdf > > > In Spark we use Akka as the RPC layer. It would be great if we can > standardize the internal RPC interface to facilitate testing. This will also > provide the foundation to try other RPC implementations in the future. -- This message was sent by Atlassian JIRA (v6.3.4#6332) - To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org
[jira] [Comment Edited] (SPARK-5124) Standardize internal RPC interface
[ https://issues.apache.org/jira/browse/SPARK-5124?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14348215#comment-14348215 ] Shixiong Zhu edited comment on SPARK-5124 at 3/5/15 6:10 AM: - I made some changes to the above APIs. 1. I added a new trait RpcResponse because RpcResponseCallback can not be used directly. Both `receive` and `receiveAndReply` methods can receive messages if the sender is an RpcEndpoint, so if the receiver wants to reply the sender, it needs an approach to specify which methods it wants to send. Here is the interface of RpcResponse: {code} private[spark] trait RpcResponse { def reply(response: Any): Unit def replyWithSender(response: Any, sender: RpcEndpointRef): Unit def fail(e: Throwable): Unit } {code} Calling `reply` will send the message to RpcEndpoint.receive, and calling `replyWithSender` will send the message to RpcEndpoint.replyWithSender. 2.Insteading of adding a trait like ThreadSafeRpcEndpoint, I added a new method `setupThreadSafeEndpoint` in RpcEnv, so that RpcEnv can decide how to implement the thread-safe semantics internally. For `setupEndpoint`, RpcEnv won't guarantee sending messages thread-safely. [~vanzin] For the local Endpoint idea, I think we need a local message dispatcher to send/receive messages between different Endpoint. It can not be implemented just in the Endpoint. We will also need EndpointRef to refer to an Endpoint, and a generic trait for the local message dispatcher and RpcEnv. It looks a bit complex. What do you think? Please help review my PR: https://github.com/apache/spark/pull/4588 was (Author: zsxwing): I made some changes to the above APIs. 1. I added a new trait RpcResponse because RpcResponseCallback can not be used directly. Both `receive` and `receiveAndReply` methods can receive messages if the sender is an RpcEndpoint, so if the receiver wants to reply the sender, it needs an approach to specify which methods it wants to send. Here is the interface of RpcResponse: {quote} trait RpcResponse { def reply(response: Any): Unit def replyWithSender(response: Any, sender: RpcEndpointRef): Unit def fail(e: Throwable): Unit } {quote} Calling `reply` will send the message to RpcEndpoint.receive, and calling `replyWithSender` will send the message to RpcEndpoint.replyWithSender. 2.Insteading of adding a trait like ThreadSafeRpcEndpoint, I added a new method `setupThreadSafeEndpoint` in RpcEnv, so that RpcEnv can decide how to implement the thread-safe semantics internally. For `setupEndpoint`, RpcEnv won't guarantee sending messages thread-safely. [~vanzin] For the local Endpoint idea, I think we need a local message dispatcher to send/receive messages between different Endpoint. It can not be implemented just in the Endpoint. We will also need EndpointRef to refer to an Endpoint, and a generic trait for the local message dispatcher and RpcEnv. It looks a bit complex. What do you think? Please help review my PR: https://github.com/apache/spark/pull/4588 > Standardize internal RPC interface > -- > > Key: SPARK-5124 > URL: https://issues.apache.org/jira/browse/SPARK-5124 > Project: Spark > Issue Type: Sub-task > Components: Spark Core >Reporter: Reynold Xin >Assignee: Shixiong Zhu > Attachments: Pluggable RPC - draft 1.pdf, Pluggable RPC - draft 2.pdf > > > In Spark we use Akka as the RPC layer. It would be great if we can > standardize the internal RPC interface to facilitate testing. This will also > provide the foundation to try other RPC implementations in the future. -- This message was sent by Atlassian JIRA (v6.3.4#6332) - To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org
[jira] [Comment Edited] (SPARK-5124) Standardize internal RPC interface
[ https://issues.apache.org/jira/browse/SPARK-5124?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14270771#comment-14270771 ] Reynold Xin edited comment on SPARK-5124 at 1/9/15 8:49 AM: Thanks for doing this. I haven't looked through all the changes yet, but here are some quick comments: 1. For DAGScheduler, we are probably OK to just use an event loop instead of an actor. Just put some messages into a queue, and have a loop that processes the queue. Otherwise we are making every call to DAGScheduler going through a socket and that can severely impact scheduler throughput. (Although I haven't looked closely at your change so maybe you are doing a different thing here) 2. Is it really that expensive to listen to network events that warrant a separate NetworkRpcEndpoint? 3. Can you comment on the following two? * Thread-safe contract when processing messages (similar to Akka). * A simple fault tolerance(if a RpcEndpoint is crashed, restart it or stop it). Personally I think those are OK to leave it up to the RPC user (e.g. BlockManagerMasterActor should be able to decide if it needs to lock or not, and an RPCEndPoint should never have uncaught exceptions). was (Author: rxin): Thanks for doing this. I haven't looked through all the changes yet, but here are some quick comments: 1. For DAGScheduler, we are probably OK to just use an event loop instead of an actor. Just put some messages into a queue, and have a loop that processes the queue. Otherwise we are making every call to DAGScheduler going through a socket and that can severely impact scheduler throughput. (Although I haven't looked closely at your change so maybe you are doing a different thing here) 2. Is it really that expensive to listen to network events that warrant a separate NetworkRpcEndpoint? > Standardize internal RPC interface > -- > > Key: SPARK-5124 > URL: https://issues.apache.org/jira/browse/SPARK-5124 > Project: Spark > Issue Type: Improvement > Components: Spark Core >Reporter: Reynold Xin >Assignee: Shixiong Zhu > Attachments: Pluggable RPC - draft 1.pdf > > > In Spark we use Akka as the RPC layer. It would be great if we can > standardize the internal RPC interface to facilitate testing. This will also > provide the foundation to try other RPC implementations in the future. -- This message was sent by Atlassian JIRA (v6.3.4#6332) - To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org
[jira] [Comment Edited] (SPARK-5124) Standardize internal RPC interface
[ https://issues.apache.org/jira/browse/SPARK-5124?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14334166#comment-14334166 ] Marcelo Vanzin edited comment on SPARK-5124 at 2/24/15 1:08 AM: Hi [~zsxwing], I briefly looked over the design and parts of the implementation, and I have some comments. Overall this looks OK, even if it kinda does look a lot like akka's actor system, but I don't think there would be a problem implementing it over a different RPC library. One thing that I found very confusing is {{RpcEndpoint}} vs. {{NetworkRpcEndpoint}}. The "R" in RPC means "remote", which sort of implies networking. I think Reynold touched on this when he talked about using an event loop for "local actors". Perhaps a more flexible approach would be something like: - A generic {{Endpoint}} that defines the {{receive()}} method and other generic interfaces - {{RpcEnv}} takes an {{Endpoint}} and a name to register endpoints for remote connections. Things like "onConnected" and "onDisassociated" become messages passed to {{receive()}}, kinda like in akka today. This means that there's no need for a specialized {{RpcEndpoint}} interface. - "local" Endpoints could be exposed directly, without the need for an {{RpcEnv}}. For example, as a getter in {{SparkEnv}}. You could have some wrapper to expose {{send()}} and {{ask()}} so that the client interface looks similar for remote and local endpoints. - The default {{Endpoint}} has no thread-safety guarantees. You can wrap an {{Endpoint}} in an {{EventLoop}} if you want messages to be handled using a queue, or synchronize your {{receive()}} method (although that can block the dispatcher thread, which could be bad). But this would easily allow actors to process multiple messages concurrently if desired. What do you think? was (Author: vanzin): Hi [~zsxwing], I briefly looked over the design and parts of the implementation, and I have some comments. Overall this looks OK, even if it kinda does look a lot like akka's actor system, but I don't think there would be a problem implementing it over a different RPC library. One thing that I found very confusing is {{RpcEndpoint}} vs. {{NetworkRpcEndpoint}}. The "R" in RPC means "remote", which sort of implies networking. I think Reynold touched on this when he talked about using an event loop for "local actors". Perhaps a more flexible approach would be something like: - A generic {{Endpoint}} that defines the {{receive()}} method and other generic interfaces - {{RpcEnv}} takes an {{Endpoint}} and a name to register endpoints for remote connections. Things like "onConnected" and "onDisassociated" become messages passed to {{receive()}}, kinda like in akka today. This means that there's no need for a specialized {{RpcEndpoint}} interface. - "local" Endpoints could be exposed directly, without the need for an {{RpcEnv}}. For example, as a getter in {{SparkEnv}}. You could have some wrapper to expose {{send()}} and {{ask()}} so that the client interface looks similar for remote and local endpoints. - The default {{Endpoint}} has no thread-safety guarantees. You can wrap an {{Endpoint}} in an {{EventLoop}} if you want messages to be handled using a queue, or synchronize your {{receive()}} method (although that can block the dispatcher thread, which could be bad). But this would easily allow actors to process multiple messages concurrently is desired. What do you think? > Standardize internal RPC interface > -- > > Key: SPARK-5124 > URL: https://issues.apache.org/jira/browse/SPARK-5124 > Project: Spark > Issue Type: Sub-task > Components: Spark Core >Reporter: Reynold Xin >Assignee: Shixiong Zhu > Attachments: Pluggable RPC - draft 1.pdf, Pluggable RPC - draft 2.pdf > > > In Spark we use Akka as the RPC layer. It would be great if we can > standardize the internal RPC interface to facilitate testing. This will also > provide the foundation to try other RPC implementations in the future. -- This message was sent by Atlassian JIRA (v6.3.4#6332) - To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org
[jira] [Comment Edited] (SPARK-5124) Standardize internal RPC interface
[ https://issues.apache.org/jira/browse/SPARK-5124?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14335252#comment-14335252 ] Marcelo Vanzin edited comment on SPARK-5124 at 2/24/15 7:03 PM: Hi @rxin, Yeah, the current {{receive()}} gives you the message and the sender, but it doesn't give you a way to explicitly reply to a message. That means two things: - It's not clear whether calling something like {{sender.send()}} would be thread-safe. For example, in akka it would because akka processes one message at a time for each actor, so you're sending replies in order. But that forces you to implement the same semantics on any new RPC layer, and that is one akka feature thing that I'd prefer not to inherit (as you say, let each endpoint decide whether it needs it). - It precludes anynchronously handling that message for much of the same reasons. e.g., let's say some incoming message requires a long computation before sending the reply, and you use a thread pool for that. With {{sender.send()}}, you'd need to build correlation logic into your application. If you have something like {{message.reply()}} instead, you don't need to - the RPC layer handles correlations and timing out if a reply doesn't arrive. Granted, I have no idea how you'd implement that second bullet in akka, and perhaps for that reason I don't think Spark currently uses that pattern anywhere. But it sounds like a useful feature to have. was (Author: vanzin): Hi @rxin, Yeah, the current {{receive()}} gives you the message and the sender, but it doesn't give you a way to explicit reply to a message. That means two things: - It's not clear whether calling something like {{sender.send()}} would be thread-safe. For example, in akka it would because akka processes one message at a time for each actor, so you're sending replies in order. But that forces you to implement the same semantics on any new RPC layer, and that is one akka feature thing that I'd prefer not to inherit (as you say, let each endpoint decide whether it needs it). - It precludes anynchronously handling that message for much of the same reasons. e.g., let's say some incoming message requires a long computation before sending the reply, and you use a thread pool for that. With {{sender.send()}}, you'd need to build correlation logic into your application. If you have something like {{message.reply()}} instead, you don't need to - the RPC layer handles correlations and timing out if a reply doesn't arrive. Granted, I have no idea how you'd implement that second bullet in akka, and perhaps for that reason I don't think Spark currently uses that pattern anywhere. But it sounds like a useful feature to have. > Standardize internal RPC interface > -- > > Key: SPARK-5124 > URL: https://issues.apache.org/jira/browse/SPARK-5124 > Project: Spark > Issue Type: Sub-task > Components: Spark Core >Reporter: Reynold Xin >Assignee: Shixiong Zhu > Attachments: Pluggable RPC - draft 1.pdf, Pluggable RPC - draft 2.pdf > > > In Spark we use Akka as the RPC layer. It would be great if we can > standardize the internal RPC interface to facilitate testing. This will also > provide the foundation to try other RPC implementations in the future. -- This message was sent by Atlassian JIRA (v6.3.4#6332) - To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org
[jira] [Comment Edited] (SPARK-5124) Standardize internal RPC interface
[ https://issues.apache.org/jira/browse/SPARK-5124?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14337028#comment-14337028 ] Reynold Xin edited comment on SPARK-5124 at 2/25/15 7:29 PM: - I went and looked at the various use cases of Akka in the code base. Quite a few of them actually also use askWithReply. There are really two categories of message deliveries: one that the sender expects a reply, and one that doesn't. As a result, I think the following interface would make more sense: {code} // receive side: def receive(msg: RpcMessage) def receiveAndReply(msg: RpcMessage) = sys.err("...") // send side: def send(message: Object): Unit def sendWithReply[T](message: Object): Future[T] {code} There is an obvious pairing here. Messages sent via "send" goes to "receive", without requiring an ack (although if we use the transport layer, an implicit ack can be sent). Messages sent via "sendWithReply" goes to "receiveAndReply", and the caller needs to explicitly handle the reply. In most cases, an rpc endpoint only needs to receive messages without reply, and thus can override just receive. By making obvious distinctions between the two, I think we mitigate the risk of an end point not properly responding to a message, leading to memory leaks. was (Author: rxin): I went and looked at the various use cases of Akka in the code base. Quite a few of them actually also use askWithReply. There are really two categories of message deliveries: one that the sender expects a reply, and one that doesn't. As a result, I think the following interface would make more sense: {code} // receive side: def receive(msg: RpcMessage) def receiveAndReply(msg: RpcMessage) = sys.err("...") // send side: def send(message: Object): Unit def sendWithReply[T](message: Object): Future[T] {code} There is an obvious pairing here. Messages sent via "send" goes to "receive", without requiring an ack (although if we use the transport layer, an implicit ack can be sent). Messages sent via "sendWithReply" goes to "receiveAndReply", and the caller needs to explicitly handle the reply. In most cases, an rpc endpoint only needs to receive messages without reply, and thus can override just receive. > Standardize internal RPC interface > -- > > Key: SPARK-5124 > URL: https://issues.apache.org/jira/browse/SPARK-5124 > Project: Spark > Issue Type: Sub-task > Components: Spark Core >Reporter: Reynold Xin >Assignee: Shixiong Zhu > Attachments: Pluggable RPC - draft 1.pdf, Pluggable RPC - draft 2.pdf > > > In Spark we use Akka as the RPC layer. It would be great if we can > standardize the internal RPC interface to facilitate testing. This will also > provide the foundation to try other RPC implementations in the future. -- This message was sent by Atlassian JIRA (v6.3.4#6332) - To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org