[SPARK-6602][Core] Update Master, Worker, Client, AppClient and related classes to use RpcEndpoint #5392

zsxwing · 2015-04-07T14:39:06Z

This PR updates the rest Actors in core to RpcEndpoint.

Because there is no ActorSelection in RpcEnv, I changes the logic of registerWithMaster in Worker and AppClient to avoid blocking the message loop. These changes need to be reviewed carefully.

…pcEndpoint

zsxwing · 2015-04-07T14:41:18Z

cc @rxin

SparkQA · 2015-04-07T16:09:51Z

Test build #29794 has finished for PR 5392 at commit b776817.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class Heartbeat(workerId: String, worker: RpcEndpointRef) extends DeployMessage
- case class RegisteredWorker(master: RpcEndpointRef, masterWebUiUrl: String) extends DeployMessage
- case class RegisterApplication(appDescription: ApplicationDescription, driver: RpcEndpointRef)
- case class RegisteredApplication(appId: String, master: RpcEndpointRef) extends DeployMessage
- case class MasterChanged(master: RpcEndpointRef, masterWebUiUrl: String)
This patch does not change any dependencies.

rxin · 2015-04-07T18:18:14Z

core/src/main/scala/org/apache/spark/deploy/client/AppClient.scala

+    with Logging {
+
+    var master: Option[RpcEndpointRef] = None
+    var alreadyDisconnected = false // To avoid calling listener.disconnected() multiple times


can the above two vars be private?

Added private.

vanzin · 2015-04-07T23:02:56Z

So I took a quick look, but I'm not familiar enough with the original code to provide good, informed feedback. The original protocol seems really weird to me, and I guess that leaks into the new code too. But fixing the protocol is probably out of the scope of this change...

…duledExecutor

zsxwing · 2015-04-08T13:43:21Z

I used the following codes to handle errors for the new codes in Client but kept the old codes untouched.

logError(e.getMessage, e)
System.exit(SparkExitCode.UNCAUGHT_EXCEPTION)

SparkQA · 2015-04-08T14:27:30Z

Test build #29860 has finished for PR 5392 at commit 4d1633b.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class Heartbeat(workerId: String, worker: RpcEndpointRef) extends DeployMessage
- case class RegisteredWorker(master: RpcEndpointRef, masterWebUiUrl: String) extends DeployMessage
- case class RegisterApplication(appDescription: ApplicationDescription, driver: RpcEndpointRef)
- case class RegisteredApplication(appId: String, master: RpcEndpointRef) extends DeployMessage
- case class MasterChanged(master: RpcEndpointRef, masterWebUiUrl: String)
This patch does not change any dependencies.

SparkQA · 2015-04-08T14:29:38Z

Test build #29861 has finished for PR 5392 at commit e8ad0a5.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class Heartbeat(workerId: String, worker: RpcEndpointRef) extends DeployMessage
- case class RegisteredWorker(master: RpcEndpointRef, masterWebUiUrl: String) extends DeployMessage
- case class RegisterApplication(appDescription: ApplicationDescription, driver: RpcEndpointRef)
- case class RegisteredApplication(appId: String, master: RpcEndpointRef) extends DeployMessage
- case class MasterChanged(master: RpcEndpointRef, masterWebUiUrl: String)
This patch does not change any dependencies.

SparkQA · 2015-04-08T16:53:50Z

Test build #29867 has finished for PR 5392 at commit 7fdee0e.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class Heartbeat(workerId: String, worker: RpcEndpointRef) extends DeployMessage
- case class RegisteredWorker(master: RpcEndpointRef, masterWebUiUrl: String) extends DeployMessage
- case class RegisterApplication(appDescription: ApplicationDescription, driver: RpcEndpointRef)
- case class RegisteredApplication(appId: String, master: RpcEndpointRef) extends DeployMessage
- case class MasterChanged(master: RpcEndpointRef, masterWebUiUrl: String)
This patch does not change any dependencies.

Conflicts: core/src/main/scala/org/apache/spark/deploy/LocalSparkCluster.scala core/src/main/scala/org/apache/spark/deploy/client/TestClient.scala core/src/test/scala/org/apache/spark/deploy/rest/StandaloneRestSubmitSuite.scala

SparkQA · 2015-04-14T07:39:27Z

Test build #30230 has finished for PR 5392 at commit 6637e3c.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.
This patch does not change any dependencies.

zsxwing · 2015-04-14T07:44:09Z

@vanzin I agree that this PR should focus on following the previous protocol instead of fixing issues of the protocol. Do you have other comments?

SparkQA · 2015-04-14T08:00:48Z

Test build #30233 has finished for PR 5392 at commit fadbb9e.

This patch fails MiMa tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class Heartbeat(workerId: String, worker: RpcEndpointRef) extends DeployMessage
- case class RegisteredWorker(master: RpcEndpointRef, masterWebUiUrl: String) extends DeployMessage
- case class RegisterApplication(appDescription: ApplicationDescription, driver: RpcEndpointRef)
- case class RegisteredApplication(appId: String, master: RpcEndpointRef) extends DeployMessage
- case class MasterChanged(master: RpcEndpointRef, masterWebUiUrl: String)
This patch does not change any dependencies.

zsxwing · 2015-04-14T08:32:09Z

retest this please.

SparkQA · 2015-04-14T10:07:37Z

Test build #30236 has finished for PR 5392 at commit fadbb9e.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class Heartbeat(workerId: String, worker: RpcEndpointRef) extends DeployMessage
- case class RegisteredWorker(master: RpcEndpointRef, masterWebUiUrl: String) extends DeployMessage
- case class RegisterApplication(appDescription: ApplicationDescription, driver: RpcEndpointRef)
- case class RegisteredApplication(appId: String, master: RpcEndpointRef) extends DeployMessage
- case class MasterChanged(master: RpcEndpointRef, masterWebUiUrl: String)
This patch does not change any dependencies.

vanzin · 2015-04-14T16:43:11Z

@zsxwing not really, I'm not familiar enough with the code to comment. Maybe @rxin or @andrewor14?

aarondav · 2015-06-29T21:20:15Z

core/src/main/scala/org/apache/spark/deploy/client/AppClient.scala

+    private var master: Option[RpcEndpointRef] = None
+    // To avoid calling listener.disconnected() multiple times
+    private var alreadyDisconnected = false
+    @volatile private var alreadyDead = false // To avoid calling listener.dead() multiple times


It's not generally true that ThreadSafeRpcEndpoints require their mutable state to be volatile, right? Perhaps this is just being modified from a separate thread pool?

It may be updated in a separate thread pool or the message loop of ClientEndpoint, so it's volatile.

aarondav · 2015-06-29T22:07:55Z

The core logic all looks good to me, just had some nits.

SparkQA · 2015-06-30T11:17:34Z

Test build #36121 has finished for PR 5392 at commit 2de7bed.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class Heartbeat(workerId: String, worker: RpcEndpointRef) extends DeployMessage
- case class RegisteredWorker(master: RpcEndpointRef, masterWebUiUrl: String) extends DeployMessage
- case class RegisterApplication(appDescription: ApplicationDescription, driver: RpcEndpointRef)
- case class RegisteredApplication(appId: String, master: RpcEndpointRef) extends DeployMessage
- case class SubmitDriverResponse(
- case class KillDriverResponse(
- case class MasterChanged(master: RpcEndpointRef, masterWebUiUrl: String)

zsxwing · 2015-06-30T15:47:17Z

retest this please

SparkQA · 2015-06-30T16:04:40Z

Test build #36147 has finished for PR 5392 at commit 2de7bed.

This patch fails MiMa tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class Heartbeat(workerId: String, worker: RpcEndpointRef) extends DeployMessage
- case class RegisteredWorker(master: RpcEndpointRef, masterWebUiUrl: String) extends DeployMessage
- case class RegisterApplication(appDescription: ApplicationDescription, driver: RpcEndpointRef)
- case class RegisteredApplication(appId: String, master: RpcEndpointRef) extends DeployMessage
- case class SubmitDriverResponse(
- case class KillDriverResponse(
- case class MasterChanged(master: RpcEndpointRef, masterWebUiUrl: String)

zsxwing · 2015-06-30T16:06:37Z

retest this please

SparkQA · 2015-06-30T18:21:57Z

Test build #36148 has finished for PR 5392 at commit 2de7bed.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class Heartbeat(workerId: String, worker: RpcEndpointRef) extends DeployMessage
- case class RegisteredWorker(master: RpcEndpointRef, masterWebUiUrl: String) extends DeployMessage
- case class RegisterApplication(appDescription: ApplicationDescription, driver: RpcEndpointRef)
- case class RegisteredApplication(appId: String, master: RpcEndpointRef) extends DeployMessage
- case class SubmitDriverResponse(
- case class KillDriverResponse(
- case class MasterChanged(master: RpcEndpointRef, masterWebUiUrl: String)

aarondav · 2015-06-30T21:45:37Z

core/src/main/scala/org/apache/spark/deploy/master/Master.scala

@@ -504,6 +518,7 @@ private[master] class Master(
  }

  private def completeRecovery() {
+    // TODO Why synchronized


This was due to an earlier state in the code when this method could be invoked from a different thread. It can be safely removed now.

aarondav · 2015-06-30T21:46:44Z

Just one comment thanks to your proactive TODOifying :) LGTM, feel free to merge after.

rxin · 2015-07-01T00:39:40Z

Alright I'm going to merge this. @zsxwing please submit a separate PR to address the TODO.

A follow-up pr to address #5392 (comment) Author: zsxwing <zsxwing@gmail.com> Closes #7141 from zsxwing/pr5392-follow-up and squashes the following commits: fcf7b50 [zsxwing] Remove unnecessary synchronized

Update Master, Worker, Client, AppClient and related classes to use R…

b776817

…pcEndpoint

rxin reviewed Apr 7, 2015
View reviewed changes

zsxwing added 3 commits April 8, 2015 21:38

Add Utils.newDaemonSingleThreadExecutor and newDaemonSingleThreadSche…

fbf3194

…duledExecutor

Log error and use SparkExitCode.UNCAUGHT_EXCEPTION exit code

6b2a104

Fix the code style

e8ad0a5

Fix the return type to ExecutorService and ScheduledExecutorService

7fdee0e

Fix the code style

fadbb9e

aarondav reviewed Jun 29, 2015
View reviewed changes

zsxwing added 2 commits June 30, 2015 16:56

Address comments

f12d943

Merge branch 'master' into rpc-rewrite-part3

2de7bed

aarondav reviewed Jun 30, 2015
View reviewed changes

asfgit closed this in 3bee0f1 Jul 1, 2015

zsxwing mentioned this pull request Jul 1, 2015

[SPARK-6602][Core]Remove unnecessary synchronized #7141

Closed

andrewor14 mentioned this pull request Aug 13, 2015

[SPARK-9877][Core] Fix StandaloneRestServer NPE when submitting application #8127

Closed

zsxwing deleted the rpc-rewrite-part3 branch December 23, 2015 23:12

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-6602][Core] Update Master, Worker, Client, AppClient and related classes to use RpcEndpoint #5392

[SPARK-6602][Core] Update Master, Worker, Client, AppClient and related classes to use RpcEndpoint #5392

zsxwing commented Apr 7, 2015

zsxwing commented Apr 7, 2015

SparkQA commented Apr 7, 2015

rxin Apr 7, 2015

zsxwing Apr 8, 2015

vanzin commented Apr 7, 2015

zsxwing commented Apr 8, 2015

SparkQA commented Apr 8, 2015

SparkQA commented Apr 8, 2015

SparkQA commented Apr 8, 2015

SparkQA commented Apr 14, 2015

zsxwing commented Apr 14, 2015

SparkQA commented Apr 14, 2015

zsxwing commented Apr 14, 2015

SparkQA commented Apr 14, 2015

vanzin commented Apr 14, 2015

aarondav Jun 29, 2015

zsxwing Jun 30, 2015

aarondav commented Jun 29, 2015

SparkQA commented Jun 30, 2015

zsxwing commented Jun 30, 2015

SparkQA commented Jun 30, 2015

zsxwing commented Jun 30, 2015

SparkQA commented Jun 30, 2015

aarondav Jun 30, 2015

aarondav commented Jun 30, 2015

rxin commented Jul 1, 2015

[SPARK-6602][Core] Update Master, Worker, Client, AppClient and related classes to use RpcEndpoint #5392

[SPARK-6602][Core] Update Master, Worker, Client, AppClient and related classes to use RpcEndpoint #5392

Conversation

zsxwing commented Apr 7, 2015

zsxwing commented Apr 7, 2015

SparkQA commented Apr 7, 2015

rxin Apr 7, 2015

Choose a reason for hiding this comment

zsxwing Apr 8, 2015

Choose a reason for hiding this comment

vanzin commented Apr 7, 2015

zsxwing commented Apr 8, 2015

SparkQA commented Apr 8, 2015

SparkQA commented Apr 8, 2015

SparkQA commented Apr 8, 2015

SparkQA commented Apr 14, 2015

zsxwing commented Apr 14, 2015

SparkQA commented Apr 14, 2015

zsxwing commented Apr 14, 2015

SparkQA commented Apr 14, 2015

vanzin commented Apr 14, 2015

aarondav Jun 29, 2015

Choose a reason for hiding this comment

zsxwing Jun 30, 2015

Choose a reason for hiding this comment

aarondav commented Jun 29, 2015

SparkQA commented Jun 30, 2015

zsxwing commented Jun 30, 2015

SparkQA commented Jun 30, 2015

zsxwing commented Jun 30, 2015

SparkQA commented Jun 30, 2015

aarondav Jun 30, 2015

Choose a reason for hiding this comment

aarondav commented Jun 30, 2015

rxin commented Jul 1, 2015