[SPARK-2468] Netty based block server / client module #1907

rxin · 2014-08-12T18:24:14Z

This is a rewrite of the original Netty module that was added about 1.5 years ago. The old code was turned off by default and didn't really work because it lacked a frame decoder (only worked with very very small blocks).

For this pull request, I tried to make the changes non-instrusive to the rest of Spark. I only added an init and shutdown to BlockManager/DiskBlockManager, and a bunch of comments to help me understand the existing code base.

Compared with the old Netty module, this one features:

It appears to work :)
SPARK-2941: option to specicy nio vs oio vs epoll for channel/transport. By default nio is used. (Not using Epoll yet because I have found some bugs with its implementation)
SPARK-2943: options to specify send buf and receive buf for users who want to do hyper tuning
SPARK-2942: io errors are reported from server to client (the protocol uses negative length to indicate error)
SPARK-2940: fetching multiple blocks in a single request to reduce syscalls
SPARK-2959: clients share a single thread pool
SPARK-2990: use PooledByteBufAllocator to reduce GC (basically a Netty managed pool of buffers with jmalloc)
SPARK-2625: added fetchWaitTime metric and fixed thread-safety issue in metrics update.
SPARK-2367: bump Netty version to 4.0.21.Final to address an Epoll bug (https://groups.google.com/forum/#!topic/netty/O7m-HxCJpCA)

Compared with the existing communication manager, this one features:

IMO it is substantially easier to understand
zero-copy send for the server for on-disk blocks
one-copy receive (due to a frame decoder)
don't quote me on this, but I think a lot less sys calls
SPARK-2990: use PooledByteBufAllocator to reduce GC (basically a Netty managed pool of buffers with jmalloc)
SPARK-2941: option to specicy nio vs oio vs epoll for channel/transport. By default nio is used. (Not using Epoll yet because I have found some bugs with its implementation)
SPARK-2943: options to specify send buf and receive buf for users who want to do hyper tuning

TODOs before it can fully replace the existing ConnectionManager, if that ever happens (most of them should probably be done in separate PRs since this needs to be turned on explicitly)

Basic test cases
More unit/integration tests for failures
Performance analysis
Support client connection reuse so we don't need to keep opening new connections (not sure how useful this would be)
Support putting blocks in addition to fetching blocks (i.e. two way transfer)
Support serving non-disk blocks
Support SASL authentication

For a more comprehensive list, see https://issues.apache.org/jira/browse/SPARK-2468

Thanks to @coderplay for peer coding with me on a Sunday.

SparkQA · 2014-08-12T18:29:48Z

QA tests have started for PR 1907. This patch merges cleanly.
View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/18377/consoleFull

SparkQA · 2014-08-12T18:30:28Z

QA results for PR 1907:
- This patch FAILED unit tests.
- This patch merges cleanly
- This patch adds no public classes

For more information see test ouptut:
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/18377/consoleFull

shivaram · 2014-08-12T18:34:11Z

@rxin Very nice ! Do you have any benchmarks of how fast things are ? In terms of say % of network bandwidth we can use ?

rxin · 2014-08-12T18:34:32Z

core/src/main/scala/org/apache/spark/network/netty/server.scala

+          }
+        }
+      }
+      val region = new DefaultFileRegion(fileChannel, offset, blockLength)


Note to self: make sure DefaultFileRegion can be released properly in error conditions.

SparkQA · 2014-08-12T18:34:45Z

QA tests have started for PR 1907. This patch merges cleanly.
View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/18378/consoleFull

rxin · 2014-08-12T18:35:12Z

That's part of the TODO :)

SparkQA · 2014-08-12T18:35:24Z

QA results for PR 1907:
- This patch FAILED unit tests.
- This patch merges cleanly
- This patch adds no public classes

For more information see test ouptut:
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/18378/consoleFull

SparkQA · 2014-08-12T18:44:49Z

QA tests have started for PR 1907. This patch merges cleanly.
View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/18380/consoleFull

SparkQA · 2014-08-12T18:45:37Z

QA results for PR 1907:
- This patch FAILED unit tests.
- This patch merges cleanly
- This patch adds no public classes

For more information see test ouptut:
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/18380/consoleFull

dcapwell · 2014-08-12T18:47:44Z

core/src/main/scala/org/apache/spark/network/netty/ReferenceCountedBuffer.scala

+ * The buffer's life cycle is NOT managed by the JVM, and thus requiring explicit declaration of
+ * reference by the retain method and release method.
+ */
+class ReferenceCountedBuffer(underlying: ByteBuf) {


Could make this class a value class. Based off this patch, the runtime wouldn't have ReferenceCountedBuffer objects created.

mubarak · 2014-08-12T20:06:42Z

@rxin
Sorry for this question and not related to code review at all. Any thoughts on using Akka IO package than Netty? Some pointers at https://groups.google.com/forum/#!msg/spray-user/7l1RbqxRuic/RwS3u4BRCLAJ

rxin · 2014-08-12T20:16:22Z

I'm not sure what akka.io would bring here. The main benefit seems to be more native Scala APIs. In this case I am not sure if we care since this is a low level module. Akka's abstraction is fairly high level, which can make stuff harder to debug for a low level module.

Netty is presumably more robust (due to the much larger user base). I'm also not sure whether akka-io supports using transferTo directly under the hood and managed buffers to reduce GC (my quick Google search suggests that it doesn't).

SparkQA · 2014-08-12T22:54:50Z

QA tests have started for PR 1907. This patch merges cleanly.
View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/18391/consoleFull

SparkQA · 2014-08-13T00:17:14Z

QA results for PR 1907:
- This patch FAILED unit tests.
- This patch merges cleanly
- This patch adds no public classes

For more information see test ouptut:
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/18391/consoleFull

SparkQA · 2014-08-13T02:45:03Z

QA tests have started for PR 1907. This patch merges cleanly.
View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/18405/consoleFull

SparkQA · 2014-08-13T03:04:59Z

QA tests have started for PR 1907. This patch merges cleanly.
View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/18406/consoleFull

SparkQA · 2014-08-13T03:11:38Z

QA results for PR 1907:
- This patch FAILED unit tests.
- This patch merges cleanly
- This patch adds no public classes

For more information see test ouptut:
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/18405/consoleFull

SparkQA · 2014-08-13T04:11:14Z

QA results for PR 1907:
- This patch FAILED unit tests.
- This patch merges cleanly
- This patch adds no public classes

For more information see test ouptut:
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/18406/consoleFull

SparkQA · 2014-08-13T04:19:53Z

QA tests have started for PR 1907. This patch merges cleanly.
View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/18408/consoleFull

SparkQA · 2014-08-13T04:49:50Z

QA tests have started for PR 1907. This patch merges cleanly.
View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/18410/consoleFull

SparkQA · 2014-08-14T03:12:55Z

QA results for PR 1907:
- This patch PASSES unit tests.
- This patch merges cleanly
- This patch adds no public classes

For more information see test ouptut:
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/18512/consoleFull

shivaram · 2014-08-14T18:14:40Z

One more thing -- I just remembered that a recent PR (#1632) was trying to handle cases where the server never replies to blockId requests. Can this happen with the Netty code path ? Or is there some ack-timeout built into netty ?

mubarak · 2014-08-14T19:43:03Z

I think client can make use of ReadTimeoutHandler

rxin · 2014-08-14T20:56:07Z

That's a good one to add. But again I wouldn't want to put everything into a single PR ...

shivaram · 2014-08-14T21:37:11Z

Yeah, you can probably make a JIRA for it or something like that. I just remembered that this is an issue, so I put it here :)

rxin · 2014-08-14T21:41:01Z

Yup. Created https://issues.apache.org/jira/browse/SPARK-2468

This should be pretty easy to do.

pwendell · 2014-08-15T02:00:02Z

I reviewed the changes that are relevant to spark core and they LGTM

rxin · 2014-08-15T02:01:17Z

Thanks for looking at this. Merging it in master & branch-1.1. This is off by default so the danger is very small. We will continue improving it and with changes like SPARK-3019 it will make it easier to use.

loveconan1988 · 2014-08-15T02:04:20Z

I have know it,thanks.

------------------ 原始邮件 ------------------
发件人: "Reynold Xin";notifications@github.com;
发送时间: 2014年8月15日(星期五) 上午10:01
收件人: "apache/spark"spark@noreply.github.com;

主题: Re: [spark] [SPARK-2468] Netty based block server / client module(#1907)

Thanks for looking at this. Merging it in master & branch-1.1. This is off by default so the danger is very small. We will continue improving it and with changes like SPARK-3019 it will make it easier to use.

—
Reply to this email directly or view it on GitHub.

loveconan1988 · 2014-08-15T02:05:34Z

I have know it,thanks.

------------------ 原始邮件 ------------------
发件人: "asfgit";notifications@github.com;
发送时间: 2014年8月15日(星期五) 上午10:03
收件人: "apache/spark"spark@noreply.github.com;

主题: Re: [spark] [SPARK-2468] Netty based block server / client module(#1907)

Closed #1907 via 3a8b68b.

—
Reply to this email directly or view it on GitHub.

@coderplay

This is a rewrite of the original Netty module that was added about 1.5 years ago. The old code was turned off by default and didn't really work because it lacked a frame decoder (only worked with very very small blocks). For this pull request, I tried to make the changes non-instrusive to the rest of Spark. I only added an init and shutdown to BlockManager/DiskBlockManager, and a bunch of comments to help me understand the existing code base. Compared with the old Netty module, this one features: - It appears to work :) - SPARK-2941: option to specicy nio vs oio vs epoll for channel/transport. By default nio is used. (Not using Epoll yet because I have found some bugs with its implementation) - SPARK-2943: options to specify send buf and receive buf for users who want to do hyper tuning - SPARK-2942: io errors are reported from server to client (the protocol uses negative length to indicate error) - SPARK-2940: fetching multiple blocks in a single request to reduce syscalls - SPARK-2959: clients share a single thread pool - SPARK-2990: use PooledByteBufAllocator to reduce GC (basically a Netty managed pool of buffers with jmalloc) - SPARK-2625: added fetchWaitTime metric and fixed thread-safety issue in metrics update. - SPARK-2367: bump Netty version to 4.0.21.Final to address an Epoll bug (https://groups.google.com/forum/#!topic/netty/O7m-HxCJpCA) Compared with the existing communication manager, this one features: - IMO it is substantially easier to understand - zero-copy send for the server for on-disk blocks - one-copy receive (due to a frame decoder) - don't quote me on this, but I think a lot less sys calls - SPARK-2990: use PooledByteBufAllocator to reduce GC (basically a Netty managed pool of buffers with jmalloc) - SPARK-2941: option to specicy nio vs oio vs epoll for channel/transport. By default nio is used. (Not using Epoll yet because I have found some bugs with its implementation) - SPARK-2943: options to specify send buf and receive buf for users who want to do hyper tuning TODOs before it can fully replace the existing ConnectionManager, if that ever happens (most of them should probably be done in separate PRs since this needs to be turned on explicitly) - [x] Basic test cases - [ ] More unit/integration tests for failures - [ ] Performance analysis - [ ] Support client connection reuse so we don't need to keep opening new connections (not sure how useful this would be) - [ ] Support putting blocks in addition to fetching blocks (i.e. two way transfer) - [x] Support serving non-disk blocks - [ ] Support SASL authentication For a more comprehensive list, see https://issues.apache.org/jira/browse/SPARK-2468 Thanks to @coderplay for peer coding with me on a Sunday. Author: Reynold Xin <rxin@apache.org> Closes #1907 from rxin/netty and squashes the following commits: f921421 [Reynold Xin] Upgrade Netty to 4.0.22.Final to fix another Epoll bug. 4b174ca [Reynold Xin] Shivaram's code review comment. 4a3dfe7 [Reynold Xin] Switched to nio for default (instead of epoll on Linux). 56bfb9d [Reynold Xin] Bump Netty version to 4.0.21.Final for some bug fixes. b443a4b [Reynold Xin] Added debug message to help debug Jenkins failures. 57fc4d7 [Reynold Xin] Added test cases for BlockHeaderEncoder and BlockFetchingClientHandlerSuite. 22623e9 [Reynold Xin] Added exception handling and test case for BlockServerHandler and BlockFetchingClientHandler. 6550dd7 [Reynold Xin] Fixed block mgr init bug. 60c2edf [Reynold Xin] Beefed up server/client integration tests. 38d88d5 [Reynold Xin] Added missing test files. 6ce3f3c [Reynold Xin] Added some basic test cases. 47f7ce0 [Reynold Xin] Created server and client packages and moved files there. b16f412 [Reynold Xin] Added commit count. f13022d [Reynold Xin] Remove unused clone() in BlockFetcherIterator. c57d68c [Reynold Xin] Added back missing files. 842dfa7 [Reynold Xin] Made everything work with proper reference counting. 3fae001 [Reynold Xin] Connected the new netty network module with rest of Spark. 1a8f6d4 [Reynold Xin] Completed protocol documentation. 2951478 [Reynold Xin] New Netty implementation. cc7843d [Reynold Xin] Basic skeleton. (cherry picked from commit 3a8b68b) Signed-off-by: Reynold Xin <rxin@apache.org>

shane-huang · 2014-08-17T14:44:46Z

core/src/main/scala/org/apache/spark/storage/BlockManager.scala

@@ -86,13 +89,25 @@ private[spark] class BlockManager(
    new TachyonStore(this, tachyonBlockManager)
  }

+  private val useNetty = conf.getBoolean("spark.shuffle.use.netty", false)
+


since useNetty is now a member val, maybe we can change the lines in getMultiple() to use this val instead of the original boolean too. Check out below lines:
...
getMultiple(){
...
val iter =
if (conf.getBoolean("spark.shuffle.use.netty", false)) {
new BlockFetcherIterator.NettyBlockFetcherIterator(this, blocksByAddress, serializer,
readMetrics)
} else {

shane-huang · 2014-08-17T15:04:57Z

Thanks for re-implementing this. Really nice to see this working :)

Previous pull request (#1907) was reverted. This brings it back. Still looking into the hang. Author: Reynold Xin <rxin@apache.org> Closes #1971 from rxin/netty1 and squashes the following commits: b0be96f [Reynold Xin] Added test to make sure outstandingRequests are cleaned after firing the events. 4c6d0ee [Reynold Xin] Pass callbacks cleanly. 603dce7 [Reynold Xin] Upgrade Netty to 4.0.23 to fix the DefaultFileRegion bug. 88be1d4 [Reynold Xin] Downgrade to 4.0.21 to work around a bug in writing DefaultFileRegion. 002626a [Reynold Xin] Remove netty-test-file.txt. db6e6e0 [Reynold Xin] Revert "Revert "[SPARK-2468] Netty based block server / client module"" (cherry picked from commit 8b9dc99) Signed-off-by: Reynold Xin <rxin@apache.org>

Previous pull request (#1907) was reverted. This brings it back. Still looking into the hang. Author: Reynold Xin <rxin@apache.org> Closes #1971 from rxin/netty1 and squashes the following commits: b0be96f [Reynold Xin] Added test to make sure outstandingRequests are cleaned after firing the events. 4c6d0ee [Reynold Xin] Pass callbacks cleanly. 603dce7 [Reynold Xin] Upgrade Netty to 4.0.23 to fix the DefaultFileRegion bug. 88be1d4 [Reynold Xin] Downgrade to 4.0.21 to work around a bug in writing DefaultFileRegion. 002626a [Reynold Xin] Remove netty-test-file.txt. db6e6e0 [Reynold Xin] Revert "Revert "[SPARK-2468] Netty based block server / client module""

@coderplay

This is a rewrite of the original Netty module that was added about 1.5 years ago. The old code was turned off by default and didn't really work because it lacked a frame decoder (only worked with very very small blocks). For this pull request, I tried to make the changes non-instrusive to the rest of Spark. I only added an init and shutdown to BlockManager/DiskBlockManager, and a bunch of comments to help me understand the existing code base. Compared with the old Netty module, this one features: - It appears to work :) - SPARK-2941: option to specicy nio vs oio vs epoll for channel/transport. By default nio is used. (Not using Epoll yet because I have found some bugs with its implementation) - SPARK-2943: options to specify send buf and receive buf for users who want to do hyper tuning - SPARK-2942: io errors are reported from server to client (the protocol uses negative length to indicate error) - SPARK-2940: fetching multiple blocks in a single request to reduce syscalls - SPARK-2959: clients share a single thread pool - SPARK-2990: use PooledByteBufAllocator to reduce GC (basically a Netty managed pool of buffers with jmalloc) - SPARK-2625: added fetchWaitTime metric and fixed thread-safety issue in metrics update. - SPARK-2367: bump Netty version to 4.0.21.Final to address an Epoll bug (https://groups.google.com/forum/#!topic/netty/O7m-HxCJpCA) Compared with the existing communication manager, this one features: - IMO it is substantially easier to understand - zero-copy send for the server for on-disk blocks - one-copy receive (due to a frame decoder) - don't quote me on this, but I think a lot less sys calls - SPARK-2990: use PooledByteBufAllocator to reduce GC (basically a Netty managed pool of buffers with jmalloc) - SPARK-2941: option to specicy nio vs oio vs epoll for channel/transport. By default nio is used. (Not using Epoll yet because I have found some bugs with its implementation) - SPARK-2943: options to specify send buf and receive buf for users who want to do hyper tuning TODOs before it can fully replace the existing ConnectionManager, if that ever happens (most of them should probably be done in separate PRs since this needs to be turned on explicitly) - [x] Basic test cases - [ ] More unit/integration tests for failures - [ ] Performance analysis - [ ] Support client connection reuse so we don't need to keep opening new connections (not sure how useful this would be) - [ ] Support putting blocks in addition to fetching blocks (i.e. two way transfer) - [x] Support serving non-disk blocks - [ ] Support SASL authentication For a more comprehensive list, see https://issues.apache.org/jira/browse/SPARK-2468 Thanks to @coderplay for peer coding with me on a Sunday. Author: Reynold Xin <rxin@apache.org> Closes apache#1907 from rxin/netty and squashes the following commits: f921421 [Reynold Xin] Upgrade Netty to 4.0.22.Final to fix another Epoll bug. 4b174ca [Reynold Xin] Shivaram's code review comment. 4a3dfe7 [Reynold Xin] Switched to nio for default (instead of epoll on Linux). 56bfb9d [Reynold Xin] Bump Netty version to 4.0.21.Final for some bug fixes. b443a4b [Reynold Xin] Added debug message to help debug Jenkins failures. 57fc4d7 [Reynold Xin] Added test cases for BlockHeaderEncoder and BlockFetchingClientHandlerSuite. 22623e9 [Reynold Xin] Added exception handling and test case for BlockServerHandler and BlockFetchingClientHandler. 6550dd7 [Reynold Xin] Fixed block mgr init bug. 60c2edf [Reynold Xin] Beefed up server/client integration tests. 38d88d5 [Reynold Xin] Added missing test files. 6ce3f3c [Reynold Xin] Added some basic test cases. 47f7ce0 [Reynold Xin] Created server and client packages and moved files there. b16f412 [Reynold Xin] Added commit count. f13022d [Reynold Xin] Remove unused clone() in BlockFetcherIterator. c57d68c [Reynold Xin] Added back missing files. 842dfa7 [Reynold Xin] Made everything work with proper reference counting. 3fae001 [Reynold Xin] Connected the new netty network module with rest of Spark. 1a8f6d4 [Reynold Xin] Completed protocol documentation. 2951478 [Reynold Xin] New Netty implementation. cc7843d [Reynold Xin] Basic skeleton.

Previous pull request (apache#1907) was reverted. This brings it back. Still looking into the hang. Author: Reynold Xin <rxin@apache.org> Closes apache#1971 from rxin/netty1 and squashes the following commits: b0be96f [Reynold Xin] Added test to make sure outstandingRequests are cleaned after firing the events. 4c6d0ee [Reynold Xin] Pass callbacks cleanly. 603dce7 [Reynold Xin] Upgrade Netty to 4.0.23 to fix the DefaultFileRegion bug. 88be1d4 [Reynold Xin] Downgrade to 4.0.21 to work around a bug in writing DefaultFileRegion. 002626a [Reynold Xin] Remove netty-test-file.txt. db6e6e0 [Reynold Xin] Revert "Revert "[SPARK-2468] Netty based block server / client module""

…pache#1907) Iceberg releases are now from `apple-1.3.0.x` branch, but Spark CI is still using `apple-1.3.x`. We should update to the former.

rxin mentioned this pull request Aug 12, 2014

[TACHYON-62] Netty Alluxio/alluxio#333

Merged

rxin reviewed Aug 12, 2014
View reviewed changes

dcapwell reviewed Aug 12, 2014
View reviewed changes

rxin added 3 commits August 12, 2014 22:39

Basic skeleton.

cc7843d

New Netty implementation.

2951478

Completed protocol documentation.

1a8f6d4

Upgrade Netty to 4.0.22.Final to fix another Epoll bug.

f921421

asfgit closed this in 3a8b68b Aug 15, 2014

rxin mentioned this pull request Aug 15, 2014

[SPARK-2468] Netty based block server / client module #1971

Closed

shane-huang reviewed Aug 17, 2014
View reviewed changes

[SPARK-2468] Netty based block server / client module #1907

[SPARK-2468] Netty based block server / client module #1907

Uh oh!

Conversation

rxin commented Aug 12, 2014

Uh oh!

SparkQA commented Aug 12, 2014

Uh oh!

SparkQA commented Aug 12, 2014

Uh oh!

shivaram commented Aug 12, 2014

Uh oh!

rxin Aug 12, 2014

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Aug 12, 2014

Uh oh!

rxin commented Aug 12, 2014

Uh oh!

SparkQA commented Aug 12, 2014

Uh oh!

SparkQA commented Aug 12, 2014

Uh oh!

SparkQA commented Aug 12, 2014

Uh oh!

dcapwell Aug 12, 2014

Choose a reason for hiding this comment

Uh oh!

rxin Aug 12, 2014

Choose a reason for hiding this comment

Uh oh!

mubarak commented Aug 12, 2014

Uh oh!

rxin commented Aug 12, 2014

Uh oh!

SparkQA commented Aug 12, 2014

Uh oh!

SparkQA commented Aug 13, 2014

Uh oh!

SparkQA commented Aug 13, 2014

Uh oh!

SparkQA commented Aug 13, 2014

Uh oh!

SparkQA commented Aug 13, 2014

Uh oh!

SparkQA commented Aug 13, 2014

Uh oh!

SparkQA commented Aug 13, 2014

Uh oh!

SparkQA commented Aug 13, 2014

Uh oh!

SparkQA commented Aug 14, 2014

Uh oh!

shivaram commented Aug 14, 2014

Uh oh!

mubarak commented Aug 14, 2014

Uh oh!

rxin commented Aug 14, 2014

Uh oh!

shivaram commented Aug 14, 2014

Uh oh!

rxin commented Aug 14, 2014

Uh oh!

pwendell commented Aug 15, 2014

Uh oh!

rxin commented Aug 15, 2014

Uh oh!

loveconan1988 commented Aug 15, 2014

Uh oh!

loveconan1988 commented Aug 15, 2014

Uh oh!

shane-huang Aug 17, 2014

Choose a reason for hiding this comment

Uh oh!

shane-huang commented Aug 17, 2014

Uh oh!

Uh oh!