[SPARK-11140] [core] Transfer files using network lib when using NettyRpcEnv. #9530

vanzin · 2015-11-06T23:17:57Z

This change abstracts the code that serves jars / files to executors so that
each RpcEnv can have its own implementation; the akka version uses the existing
HTTP-based file serving mechanism, while the netty versions uses the new
stream support added to the network lib, which makes file transfers benefit
from the easier security configuration of the network library, and should also
reduce overhead overall.

The change includes a small fix to TransportChannelHandler so that it propagates
user events to downstream handlers.

…yRpcEnv. This change abstracts the code that serves jars / files to executors so that each RpcEnv can have its own implementation; the akka version uses the existing HTTP-based file serving mechanism, while the netty versions uses the new stream support added to the network lib, which makes file transfers benefit from the easier security configuration of the network library, and should also reduce overhead overall. The change includes a small fix to TransportChannelHandler so that it propagates user events to downstream handlers.

vanzin · 2015-11-06T23:28:37Z

/cc @rxin @zsxwing

vanzin · 2015-11-07T21:25:00Z

retest this please

SparkQA · 2015-11-07T23:22:35Z

Test build #45292 has finished for PR 9530 at commit 3f8209f.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds no public classes.

vanzin · 2015-11-08T01:01:36Z

retest this please

SparkQA · 2015-11-08T02:47:08Z

Test build #45293 has finished for PR 9530 at commit 3f8209f.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2015-11-09T19:14:28Z

Test build #45383 has finished for PR 9530 at commit 87b7a91.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):\n * public class JavaAFTSurvivalRegressionExample\n

zsxwing · 2015-11-09T19:30:41Z

retest this please

This reuses more of the existing code in Utils, and also makes some subsequent changes built on top of this API possible, at the expense of some efficiency (using input streams instead of channels). If desired the Utils class can later be changed to use channels to regain the lost efficiency.

SparkQA · 2015-11-09T21:27:05Z

Test build #45402 has finished for PR 9530 at commit 87b7a91.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

vanzin · 2015-11-09T22:29:18Z

retest this please

zsxwing · 2015-11-09T23:27:00Z

core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala


  override def exceptionCaught(cause: Throwable, client: TransportClient): Unit = {
    val addr = client.getChannel.remoteAddress().asInstanceOf[InetSocketAddress]
    if (addr != null) {
-      val clientAddr = RpcAddress(addr.getHostName, addr.getPort)


Why reverted it back to the old codes?

Sorry, I don't follow. The line was moved inside the new conditions, nothing's really changing.

You added a new if if (clients.containsKey(client)) {. There was a similar logic here: https://github.com/apache/spark/pull/9210/files#diff-0c89b4a60c30a7cd2224bb64d93da942L477 but you removed it. Now you are adding it back.

It's explained in the following comment:
https://github.com/apache/spark/pull/9530/files#diff-0c89b4a60c30a7cd2224bb64d93da942R85

Got it. Thanks!

zsxwing · 2015-11-09T23:54:08Z

LGTM. I noticed that Utils.fetchFile cannot handle special chars (e.g., %, ) in file name. But since the http server has the same issue, it should not block this PR.

vanzin · 2015-11-10T00:01:45Z

core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala

+
+  private class TimeoutHandler(client: TransportClient) extends ChannelInboundHandlerAdapter {
+
+    override def userEventTriggered(ctx: ChannelHandlerContext, evt: Object): Unit = {


So, this is a little bit racy, since the timeout might trigger when another thread is preparing to download a file. I'll fix this and update the PR.

zsxwing · 2015-11-10T01:34:25Z

core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala

+    source
+  }
+
+  private def fileDownloadClient(host: String, port: Int): TransportClient = synchronized {


So if downloading from different hosts (different tasks in the same Executor), they need to connect one by one. Right?

Downloading from different hosts means different clients. But different tasks will still all download from the same host (the driver).

Oh, right. All tasks will download from the same driver.

Just found another case: fetchFiles is also used in DriverRunner in the Worker.

That's fine. What's your worry here?

On second thought I think that's fine since downloading are still in parallel.

SparkQA · 2015-11-10T03:07:17Z

Test build #45446 has finished for PR 9530 at commit a222c03.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

zsxwing · 2015-11-10T19:47:04Z

core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala

+          val socketAddr = client.getChannel().remoteAddress().asInstanceOf[InetSocketAddress]
+          val address = RpcAddress(socketAddr.getHostName(), socketAddr.getPort())
+          ctx.close()
+          NettyRpcEnv.this.synchronized {


The lock order here is in reverse order of fileDownloadClient. A potential dead-lock issue?

Good catch, fixed.

SparkQA · 2015-11-10T21:38:03Z

Test build #45540 has finished for PR 9530 at commit a89e665.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2015-11-11T00:20:31Z

Test build #45552 has finished for PR 9530 at commit 11f61a8.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2015-11-11T01:24:25Z

Test build #45547 has finished for PR 9530 at commit 71ac0cf.

This patch fails from timeout after a configured wait of 250m.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2015-11-14T00:51:13Z

Test build #45904 has finished for PR 9530 at commit 908a051.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):\n * case class AppendColumns[T, U](\n

vanzin · 2015-11-17T01:30:41Z

retest this please

SparkQA · 2015-11-17T03:39:40Z

Test build #46041 has finished for PR 9530 at commit 908a051.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds no public classes.

rxin · 2015-11-18T03:00:10Z

Alright I looked a bit closer at this. Since it is not mission critical and it is touching some mission critical part, I think it's best to not to merge this in 1.6.

Let's merge this once 1.6 has an rc?

vanzin · 2015-11-18T03:20:42Z

What's the problem with merging to master? Isn't that why we have branch-1.6?

I have changes built on top of this that I'm waiting to send for review.

rxin · 2015-11-18T03:39:49Z

I'm mostly thinking it'd make it harder if we found bugs that need to fix.

vanzin · 2015-11-18T04:00:21Z

Do you mean bugs in this code? Or in 1.6? If in this code, master will eventually go through stabilization also, and at that point if there are bugs, they'll surface. Whether this code goes in now or later doesn't change much in that regard.

rxin · 2015-11-18T07:32:29Z

There are still some network related code that's going into 1.6 (one is the config naming thing). I'm concerned they might conflict. Maybe you can just submit more prs basing on previous ones for reviewing?

We should be pretty close to cutting a rc anyway.

vanzin · 2015-11-18T18:05:34Z

Maybe you can just submit more prs basing on previous ones

I don't know of any clean way of doing that in github; they'd end up including all the code in this PR, which is rather ugly / sub-optimal.

I can wait for the config change (I want to push it soon anyway), but I'm not super worried about conflicts. Those are generally not hard to resolve.

SparkQA · 2015-11-18T22:37:55Z

Test build #46261 has finished for PR 9530 at commit d10fa97.

This patch fails to build.
This patch merges cleanly.
This patch adds the following public classes (experimental):\n * class ALS(override val uid: String) extends Estimator[ALSModel] with ALSParams with Writable\n

SparkQA · 2015-11-19T01:22:12Z

Test build #46265 has finished for PR 9530 at commit 7cc83e7.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):\n * class ALS(override val uid: String) extends Estimator[ALSModel] with ALSParams with Writable\n

This makes the code friendlier to SPARK-11097 (if and when it's implemented), and also avoids custom timeout handling code by reusing features of the transport library. Need to test on a real cluster, though...

SparkQA · 2015-11-19T05:08:08Z

Test build #46289 has finished for PR 9530 at commit cfd01bd.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

But use the rpc config as the default. Also now properly tested on a real cluster, and verified idle connections are closed.

SparkQA · 2015-11-19T22:13:09Z

Test build #46349 has finished for PR 9530 at commit 0e8e4bb.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds no public classes.

vanzin · 2015-11-19T22:14:37Z

sigh... retest this please.

SparkQA · 2015-11-20T02:38:50Z

Test build #46361 has finished for PR 9530 at commit 0e8e4bb.

This patch fails from timeout after a configured wait of 250m.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2015-11-20T05:10:12Z

Test build #46388 has finished for PR 9530 at commit c5a218d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

zsxwing · 2015-11-20T21:02:34Z

LGTM. But let's just merge this one to master since it's too risky to add a new feature to 1.6 now.

vanzin · 2015-11-23T18:49:47Z

retest this please

SparkQA · 2015-11-23T20:53:44Z

Test build #46542 has finished for PR 9530 at commit c5a218d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

vanzin · 2015-11-23T21:53:56Z

Merging to master.

…RpcEnv. This change abstracts the code that serves jars / files to executors so that each RpcEnv can have its own implementation; the akka version uses the existing HTTP-based file serving mechanism, while the netty versions uses the new stream support added to the network lib, which makes file transfers benefit from the easier security configuration of the network library, and should also reduce overhead overall. The change includes a small fix to TransportChannelHandler so that it propagates user events to downstream handlers. Author: Marcelo Vanzin <vanzin@cloudera.com> Closes #9530 from vanzin/SPARK-11140.

JoshRosen · 2015-11-24T07:51:24Z

The following two tests have failed for the past three builds in the Hadoop 2.4 Maven Jenkins job:

org.apache.spark.deploy.yarn.YarnClusterSuite.run Spark in yarn-client mode
org.apache.spark.deploy.yarn.YarnClusterSuite.user class path first in client mode

According to Jenkins, these failures first started in https://amplab.cs.berkeley.edu/jenkins/job/Spark-Master-Maven-with-YARN/HADOOP_PROFILE=hadoop-2.4,label=spark-test/4223/

Given the set of patches in that changeset, my hunch is that this might have somehow been caused by either #9853 or #9530 (this patch). Branch 1.6 only includes #9853 and seems to be passing these tests, so I therefore believe that this patch may be the culprit (since it's only present in master and not 1.6). @vanzin, could you help me look into this?

…RpcEnv. This change abstracts the code that serves jars / files to executors so that each RpcEnv can have its own implementation; the akka version uses the existing HTTP-based file serving mechanism, while the netty versions uses the new stream support added to the network lib, which makes file transfers benefit from the easier security configuration of the network library, and should also reduce overhead overall. The change includes a small fix to TransportChannelHandler so that it propagates user events to downstream handlers. Author: Marcelo Vanzin <vanzin@cloudera.com> Closes apache#9530 from vanzin/SPARK-11140. (cherry picked from commit c2467da)

squito · 2015-11-25T17:11:46Z

Hi @JoshRosen , I was just looking into this. With some local testing, I could confirm that this patch did lead to those failures, but that it is fixed by #9923 (in particular the handling of an empty stream in TransportResponseHandler)

squito · 2015-11-25T17:31:35Z

(or #9941, which is just the bug fixes w/out the other changes ...)

Merge branch 'master' into SPARK-11140

87b7a91

zsxwing reviewed Nov 9, 2015
View reviewed changes

vanzin reviewed Nov 10, 2015
View reviewed changes

Handle race in timeout handling.

a222c03

zsxwing reviewed Nov 10, 2015
View reviewed changes

Merge branch 'master' into SPARK-11140

a89e665

zsxwing reviewed Nov 10, 2015
View reviewed changes

Fix locking.

71ac0cf

Doc updates.

11f61a8

Doc consistency fix.

fcdc498

Merge branch 'master' into SPARK-11140

908a051

Merge branch 'master' into SPARK-11140

d10fa97

Fix borked merge.

7cc83e7

Use separate client factory for file downloads.

cfd01bd

This makes the code friendlier to SPARK-11097 (if and when it's implemented), and also avoids custom timeout handling code by reusing features of the transport library. Need to test on a real cluster, though...

Allow separate configs for all network opts in file download client.

0e8e4bb

But use the rpc config as the default. Also now properly tested on a real cluster, and verified idle connections are closed.

Merge branch 'master' into SPARK-11140

c5a218d

zsxwing mentioned this pull request Nov 20, 2015

[SPARK-11762] [network] Account for active streams when couting outstanding requests. #9747

Closed

vanzin closed this Nov 23, 2015

vanzin deleted the SPARK-11140 branch November 23, 2015 21:55


		private class TimeoutHandler(client: TransportClient) extends ChannelInboundHandlerAdapter {

		override def userEventTriggered(ctx: ChannelHandlerContext, evt: Object): Unit = {

[SPARK-11140] [core] Transfer files using network lib when using NettyRpcEnv. #9530

[SPARK-11140] [core] Transfer files using network lib when using NettyRpcEnv. #9530

Uh oh!

Conversation

vanzin commented Nov 6, 2015

Uh oh!

vanzin commented Nov 6, 2015

Uh oh!

vanzin commented Nov 7, 2015

Uh oh!

SparkQA commented Nov 7, 2015

Uh oh!

vanzin commented Nov 8, 2015

Uh oh!

SparkQA commented Nov 8, 2015

Uh oh!

SparkQA commented Nov 9, 2015

Uh oh!

zsxwing commented Nov 9, 2015

Uh oh!

SparkQA commented Nov 9, 2015

Uh oh!

vanzin commented Nov 9, 2015

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zsxwing commented Nov 9, 2015

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Nov 10, 2015

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Nov 10, 2015

Uh oh!

SparkQA commented Nov 11, 2015

Uh oh!

SparkQA commented Nov 11, 2015

Uh oh!

SparkQA commented Nov 14, 2015

Uh oh!

vanzin commented Nov 17, 2015

Uh oh!

SparkQA commented Nov 17, 2015

Uh oh!

rxin commented Nov 18, 2015

Uh oh!

vanzin commented Nov 18, 2015

Uh oh!

rxin commented Nov 18, 2015

Uh oh!

vanzin commented Nov 18, 2015

Uh oh!

rxin commented Nov 18, 2015

Uh oh!

vanzin commented Nov 18, 2015

Uh oh!