Skip to content

[SPARK-10997] [core] Add "client mode" to netty rpc env. #9210

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 5 commits into from
Closed

[SPARK-10997] [core] Add "client mode" to netty rpc env. #9210

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
48f3c55
[SPARK-10997] [core] Add "client mode" to netty rpc env.
Oct 13, 2015
da4d824
Feedback.
Oct 22, 2015
11d9092
Feedback.
Oct 23, 2015
418b472
Merge branch 'master' into SPARK-10997
Oct 23, 2015
acce31f
Restore a comment.
Oct 24, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions core/src/main/scala/org/apache/spark/SparkEnv.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -256,7 +256,8 @@ object SparkEnv extends Logging {

// Create the ActorSystem for Akka and get the port it binds to.
val actorSystemName = if (isDriver) driverActorSystemName else executorActorSystemName
val rpcEnv = RpcEnv.create(actorSystemName, hostname, port, conf, securityManager)
val rpcEnv = RpcEnv.create(actorSystemName, hostname, port, conf, securityManager,
clientMode = !isDriver)
val actorSystem: ActorSystem =
if (rpcEnv.isInstanceOf[AkkaRpcEnv]) {
rpcEnv.asInstanceOf[AkkaRpcEnv].actorSystem
Expand All @@ -266,9 +267,11 @@ object SparkEnv extends Logging {
}

// Figure out which port Akka actually bound to in case the original port is 0 or occupied.
// In the non-driver case, the RPC env's address may be null since it may not be listening
// for incoming connections.
if (isDriver) {
conf.set("spark.driver.port", rpcEnv.address.port.toString)
} else {
} else if (rpcEnv.address != null) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can you add some comment on what this is checking and when this can happen

conf.set("spark.executor.port", rpcEnv.address.port.toString)
}

Expand Down
20 changes: 9 additions & 11 deletions core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,8 +45,6 @@ private[spark] class CoarseGrainedExecutorBackend(
env: SparkEnv)
extends ThreadSafeRpcEndpoint with ExecutorBackend with Logging {

Utils.checkHostPort(hostPort, "Expected hostport")

var executor: Executor = null
@volatile var driver: Option[RpcEndpointRef] = None

Expand Down Expand Up @@ -80,9 +78,8 @@ private[spark] class CoarseGrainedExecutorBackend(
}

override def receive: PartialFunction[Any, Unit] = {
case RegisteredExecutor =>
case RegisteredExecutor(hostname) =>
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

note: I'm having to change this because the client doesn't know which hostname the client connection will be created on when the register message is sent. So the server replies it back to the client, so that both agree about what's the actual address being used.

logInfo("Successfully registered with driver")
val (hostname, _) = Utils.parseHostPort(hostPort)
executor = new Executor(executorId, hostname, env, userClassPath, isLocal = false)

case RegisterExecutorFailed(message) =>
Expand Down Expand Up @@ -163,7 +160,8 @@ private[spark] object CoarseGrainedExecutorBackend extends Logging {
hostname,
port,
executorConf,
new SecurityManager(executorConf))
new SecurityManager(executorConf),
clientMode = true)
val driver = fetcher.setupEndpointRefByURI(driverUrl)
val props = driver.askWithRetry[Seq[(String, String)]](RetrieveSparkProps) ++
Seq[(String, String)](("spark.app.id", appId))
Expand All @@ -188,12 +186,12 @@ private[spark] object CoarseGrainedExecutorBackend extends Logging {
val env = SparkEnv.createExecutorEnv(
driverConf, executorId, hostname, port, cores, isLocal = false)

// SparkEnv sets spark.driver.port so it shouldn't be 0 anymore.
val boundPort = env.conf.getInt("spark.executor.port", 0)
assert(boundPort != 0)

// Start the CoarseGrainedExecutorBackend endpoint.
val sparkHostPort = hostname + ":" + boundPort
// SparkEnv will set spark.executor.port if the rpc env is listening for incoming
// connections (e.g., if it's using akka). Otherwise, the executor is running in
// client mode only, and does not accept incoming connections.
val sparkHostPort = env.conf.getOption("spark.executor.port").map { port =>
hostname + ":" + port
}.orNull
env.rpcEnv.setupEndpoint("Executor", new CoarseGrainedExecutorBackend(
env.rpcEnv, driverUrl, executorId, sparkHostPort, cores, userClassPath, env))
workerUrl.foreach { url =>
Expand Down
8 changes: 5 additions & 3 deletions core/src/main/scala/org/apache/spark/rpc/RpcEnv.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,9 +43,10 @@ private[spark] object RpcEnv {
host: String,
port: Int,
conf: SparkConf,
securityManager: SecurityManager): RpcEnv = {
securityManager: SecurityManager,
clientMode: Boolean = false): RpcEnv = {
// Using Reflection to create the RpcEnv to avoid to depend on Akka directly
val config = RpcEnvConfig(conf, name, host, port, securityManager)
val config = RpcEnvConfig(conf, name, host, port, securityManager, clientMode)
getRpcEnvFactory(conf).create(config)
}
}
Expand Down Expand Up @@ -139,4 +140,5 @@ private[spark] case class RpcEnvConfig(
name: String,
host: String,
port: Int,
securityManager: SecurityManager)
securityManager: SecurityManager,
clientMode: Boolean)
2 changes: 1 addition & 1 deletion core/src/main/scala/org/apache/spark/rpc/netty/Dispatcher.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,7 @@ private[netty] class Dispatcher(nettyEnv: NettyRpcEnv) extends Logging {
private var stopped = false

def registerRpcEndpoint(name: String, endpoint: RpcEndpoint): NettyRpcEndpointRef = {
val addr = new RpcEndpointAddress(nettyEnv.address.host, nettyEnv.address.port, name)
val addr = RpcEndpointAddress(nettyEnv.address, name)
val endpointRef = new NettyRpcEndpointRef(nettyEnv.conf, addr, nettyEnv)
synchronized {
if (stopped) {
Expand Down
Loading